MoFin/prompt_manager/analytics.py

"""策略→提示词版本分析引擎

核心功能：将策略评估结果按提示词版本聚合，计算每个版本的准确率。
"""

import json
from pathlib import Path
from datetime import datetime
from typing import Optional

from .tracking import load_associations, get_associations_for_prompt_version
from .registry import get_prompt, get_version_history

PROJECT_DIR = Path("/home/hmo/projects/MoFin")
DECISIONS_PATH = PROJECT_DIR / "data" / "decisions.json"
ACCURACY_PATH = PROJECT_DIR / "data" / "accuracy_stats.json"
EVAL_PATH = PROJECT_DIR / "data" / "evaluation.json"


def _load_json(path, default=None):
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {} if default is None else default


def analyze_prompt_version_effectiveness() -> dict:
    """按提示词版本聚合策略评估结果

    关联 decisions.json 中的 evaluation 字段和 associations.json 中的版本记录，
    计算出每个提示词版本的：
    - 生成的策略总数
    - 达到止盈数（成功）
    - 跌破止损数（失败）
    - 待验证数
    - 盈亏比平均值
    """
    # 加载数据
    decisions = _load_json(DECISIONS_PATH, {"decisions": []})
    associations = load_associations().get("associations", [])

    # 建立 code → 最新关联的映射
    code_to_pv = {}  # code -> {prompt_id, version}
    for a in associations:
        code = a.get("code")
        if code and code not in code_to_pv:
            code_to_pv[code] = {
                "prompt_id": a.get("prompt_id"),
                "version": a.get("prompt_version"),
            }

    # 按 prompt_id@version 分组统计
    version_stats = {}

    for d in decisions.get("decisions", []):
        code = d.get("code")
        if not code:
            continue

        pv = code_to_pv.get(code)
        if not pv:
            continue

        key = f"{pv['prompt_id']}@{pv['version']}"
        if key not in version_stats:
            version_stats[key] = {
                "prompt_id": pv["prompt_id"],
                "version": pv["version"],
                "total": 0,
                "take_profit_hit": 0,
                "stop_loss_hit": 0,
                "in_entry_zone": 0,
                "pending": 0,
                "avg_rr": 0.0,
                "rr_sum": 0.0,
                "rr_count": 0,
                "stocks": [],
            }

        vs = version_stats[key]
        vs["total"] += 1
        vs["stocks"].append(code)

        # 从 evaluation 中读取状态
        evals = d.get("evaluation", [])
        for ev in evals:
            if isinstance(ev, dict) and ev.get("phase") == 1:
                theo = ev.get("theoretical", {})
                status = theo.get("status", "")
                if status == "take_profit_hit":
                    vs["take_profit_hit"] += 1
                elif status == "stop_loss_hit":
                    vs["stop_loss_hit"] += 1
                elif status == "in_entry_zone":
                    vs["in_entry_zone"] += 1
                else:
                    vs["pending"] += 1

        # 盈亏比
        rr = d.get("rr_ratio")
        if rr is not None and isinstance(rr, (int, float)):
            vs["rr_sum"] += rr
            vs["rr_count"] += 1

    # 计算平均盈亏比和成功率
    for key, vs in version_stats.items():
        if vs["rr_count"] > 0:
            vs["avg_rr"] = round(vs["rr_sum"] / vs["rr_count"], 2)

        total_outcome = vs["take_profit_hit"] + vs["stop_loss_hit"]
        if total_outcome > 0:
            vs["success_rate"] = round(vs["take_profit_hit"] / total_outcome * 100, 1)
        else:
            vs["success_rate"] = None

        del vs["rr_sum"]

    return version_stats


def get_prompt_version_comparison() -> dict:
    """生成版本对比报告"""
    version_stats = analyze_prompt_version_effectiveness()

    # 补充每个版本的标签信息
    for key, vs in version_stats.items():
        prompt = get_prompt(vs["prompt_id"])
        if prompt:
            for v in prompt.versions:
                if v.version == vs["version"]:
                    vs["label"] = v.label
                    vs["changelog"] = v.changelog
                    vs["tags"] = v.tags
                    break
        if "label" not in vs:
            vs["label"] = vs["version"]

    return version_stats


def generate_report() -> str:
    """生成版本有效性报告文本"""
    comparison = get_prompt_version_comparison()

    lines = []
    lines.append("📊 提示词版本有效性分析 | " + datetime.now().strftime("%Y-%m-%d"))
    lines.append("")

    if not comparison:
        lines.append("暂无数据 — 请先运行策略生成和评估后重试")
        return "\n".join(lines)

    # 按 prompt_id 分组显示
    by_prompt = {}
    for key, vs in comparison.items():
        pid = vs["prompt_id"]
        by_prompt.setdefault(pid, []).append(vs)

    for pid, versions in sorted(by_prompt.items()):
        prompt = get_prompt(pid)
        lines.append(f"\n## {prompt.name if prompt else pid}")
        lines.append(f"  {prompt.description if prompt else ''}")
        lines.append("")

        # 按版本号排序
        versions.sort(key=lambda x: x["version"])

        header = f"  {'版本':<10} {'标签':<20} {'策略数':<8} {'止盈':<8} {'止损':<8} {'成功率':<10} {'平均R/R':<10}"
        lines.append(header)
        lines.append("  " + "-" * len(header))

        for vs in versions:
            sr = f"{vs['success_rate']}%" if vs['success_rate'] is not None else "-"
            label = vs.get("label", "")[:18]
            lines.append(
                f"  {vs['version']:<10} {label:<20} "
                f"{vs['total']:<8} {vs['take_profit_hit']:<8} "
                f"{vs['stop_loss_hit']:<8} {sr:<10} "
                f"{vs['avg_rr']:<10}"
            )

    lines.append("")
    lines.append("---")
    lines.append("注：数据来自 decisions.json evaluation + associations.json")
    return "\n".join(lines)