cyb50-quant/research/dragon/v2/dragon_strategy_overview.py

from __future__ import annotations

from pathlib import Path

import pandas as pd

from dragon_branch_configs import (
    alpha_first_glued_refined_hot_cap_config,
    alpha_first_selective_veto_config,
    workbook_preserving_config,
)
from dragon_shared import END_DATE, START_DATE, evaluation_years, format_num as _format_num, format_pct as _format_pct, profit_factor
from dragon_strategy import DragonRuleEngine


def _load_indicator_snapshot(base_dir: Path) -> pd.DataFrame:
    df = pd.read_csv(base_dir / "dragon_indicator_snapshot.csv", encoding="utf-8-sig")
    df["date"] = pd.to_datetime(df["date"])
    return df.set_index("date", drop=False)


def _event_overlap(events: pd.DataFrame, workbook: pd.DataFrame, side: str) -> tuple[int, int]:
    wb = set(workbook[(workbook["side"] == side) & (workbook["layer"] == "real_trade")]["date"])
    st = set(events[(events["side"] == side) & (events["layer"] == "real_trade")]["date"])
    return len(wb & st), len(st)


def _run_branch(name: str, config, indicator_df: pd.DataFrame, workbook_events: pd.DataFrame) -> dict[str, object]:
    engine = DragonRuleEngine(config=config)
    events, trades = engine.run(indicator_df)
    events = events[(events["date"] >= START_DATE) & (events["date"] <= END_DATE)].copy()
    trades = trades[
        (trades["buy_date"] >= START_DATE)
        & (trades["buy_date"] <= END_DATE)
        & (trades["sell_date"] >= START_DATE)
        & (trades["sell_date"] <= END_DATE)
    ].copy()

    returns = trades["return_pct"].astype(float)
    compounded_return = float((1.0 + returns).prod() - 1.0) if not trades.empty else float("nan")
    years = evaluation_years(START_DATE, END_DATE)
    cagr = float((1.0 + compounded_return) ** (1.0 / years) - 1.0) if pd.notna(compounded_return) else float("nan")

    buy_overlap, buy_strategy_total = _event_overlap(events, workbook_events, "BUY")
    sell_overlap, sell_strategy_total = _event_overlap(events, workbook_events, "SELL")

    short_00_05d = float(trades[trades["holding_days"] <= 5]["return_pct"].mean())
    short_06_10d = float(trades[(trades["holding_days"] > 5) & (trades["holding_days"] <= 10)]["return_pct"].mean())

    return {
        "branch": name,
        "trades": int(len(trades)),
        "win_rate": float((returns > 0).mean()) if not trades.empty else float("nan"),
        "avg_return": float(returns.mean()) if not trades.empty else float("nan"),
        "median_return": float(returns.median()) if not trades.empty else float("nan"),
        "profit_factor": profit_factor(returns) if not trades.empty else float("nan"),
        "compounded_return": compounded_return,
        "cagr": cagr,
        "real_buy_overlap": int(buy_overlap),
        "real_sell_overlap": int(sell_overlap),
        "real_buy_strategy_total": int(buy_strategy_total),
        "real_sell_strategy_total": int(sell_strategy_total),
        "short_00_05d_avg_return": short_00_05d,
        "short_06_10d_avg_return": short_06_10d,
    }


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    indicator_df = _load_indicator_snapshot(base_dir)
    workbook_events = pd.read_csv(base_dir / "true_trade_events.csv", encoding="utf-8-sig")

    rows = [
        _run_branch("workbook_preserving", workbook_preserving_config(), indicator_df, workbook_events),
        _run_branch("alpha_first_selective_veto", alpha_first_selective_veto_config(), indicator_df, workbook_events),
        _run_branch("alpha_first_glued_refined_hot_cap", alpha_first_glued_refined_hot_cap_config(), indicator_df, workbook_events),
    ]
    df = pd.DataFrame(rows)
    df.to_csv(base_dir / "dragon_strategy_overview.csv", index=False, encoding="utf-8-sig")

    role_map = {
        "workbook_preserving": "official reconstruction baseline",
        "alpha_first_selective_veto": "current formal alpha branch",
        "alpha_first_glued_refined_hot_cap": "leading high-alpha candidate",
    }
    style_map = {
        "workbook_preserving": "most like workbook",
        "alpha_first_selective_veto": "balanced",
        "alpha_first_glued_refined_hot_cap": "most aggressive",
    }
    suit_map = {
        "workbook_preserving": "适合优先保留原表结构",
        "alpha_first_selective_veto": "适合兼顾原表和收益质量",
        "alpha_first_glued_refined_hot_cap": "适合更偏实战 alpha",
    }

    lines = [
        "# Dragon Strategy Overview",
        "",
        f"- Evaluation window: `{START_DATE}` to `{END_DATE}`.",
        "- Return metrics use compounded trade returns without extra slippage/fee adjustments.",
        "",
        "## Headline Table",
    ]
    for _, row in df.iterrows():
        branch = row["branch"]
        lines.extend(
            [
                f"### {branch}",
                f"- Role: `{role_map[branch]}`",
                f"- Style: `{style_map[branch]}`",
                f"- Trades: `{int(row['trades'])}`",
                f"- Win rate: `{_format_pct(float(row['win_rate']))}`",
                f"- Avg / Median trade: `{_format_pct(float(row['avg_return']))}` / `{_format_pct(float(row['median_return']))}`",
                f"- Profit factor: `{_format_num(float(row['profit_factor']))}`",
                f"- Compounded return: `{_format_pct(float(row['compounded_return']))}`",
                f"- CAGR: `{_format_pct(float(row['cagr']))}`",
                f"- Real BUY / SELL overlap: `{int(row['real_buy_overlap'])}/{int(row['real_sell_overlap'])}`",
                f"- Short `00-05d` / `06-10d`: `{_format_pct(float(row['short_00_05d_avg_return']))}` / `{_format_pct(float(row['short_06_10d_avg_return']))}`",
                f"- Suitable for: {suit_map[branch]}",
                "",
            ]
        )

    workbook = df[df["branch"] == "workbook_preserving"].iloc[0]
    alpha = df[df["branch"] == "alpha_first_selective_veto"].iloc[0]
    refined = df[df["branch"] == "alpha_first_glued_refined_hot_cap"].iloc[0]

    lines.extend(
        [
            "## Quick Read",
            f"- If you want the version most like the workbook, use `workbook_preserving`: CAGR `{_format_pct(float(workbook['cagr']))}`, overlap `{int(workbook['real_buy_overlap'])}/{int(workbook['real_sell_overlap'])}`.",
            f"- If you want the balanced version, use `alpha_first_selective_veto`: CAGR `{_format_pct(float(alpha['cagr']))}`, profit factor `{_format_num(float(alpha['profit_factor']))}`, overlap `{int(alpha['real_buy_overlap'])}/{int(alpha['real_sell_overlap'])}`.",
            f"- If you want the strongest alpha candidate, use `alpha_first_glued_refined_hot_cap`: CAGR `{_format_pct(float(refined['cagr']))}`, profit factor `{_format_num(float(refined['profit_factor']))}`, overlap `{int(refined['real_buy_overlap'])}/{int(refined['real_sell_overlap'])}`.",
            "",
            "## Quant Take",
            f"- `alpha_first_selective_veto` vs workbook: CAGR `+{(float(alpha['cagr']) - float(workbook['cagr'])):.2%}`, profit factor `+{(float(alpha['profit_factor']) - float(workbook['profit_factor'])):.2f}`, BUY/SELL overlap delta `{int(alpha['real_buy_overlap'] - workbook['real_buy_overlap'])}/{int(alpha['real_sell_overlap'] - workbook['real_sell_overlap'])}`.",
            f"- `alpha_first_glued_refined_hot_cap` vs current alpha: CAGR `+{(float(refined['cagr']) - float(alpha['cagr'])):.2%}`, profit factor `+{(float(refined['profit_factor']) - float(alpha['profit_factor'])):.2f}`, BUY/SELL overlap delta `{int(refined['real_buy_overlap'] - alpha['real_buy_overlap'])}/{int(refined['real_sell_overlap'] - alpha['real_sell_overlap'])}`.",
            "- Operationally, the refined branch is the best-performing candidate, but the current formal alpha branch remains the governance default because it gives up fewer workbook-aligned dates.",
        ]
    )

    (base_dir / "dragon_strategy_overview.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()