cyb50-quant/research/dragon/v2/dragon_layered_pnl_attribution.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
from dragon_rc1_golden_baseline import _load_indicator_snapshot
from dragon_shared import END_DATE, START_DATE, format_num, format_pct, profit_factor
from dragon_strategy import DragonRuleEngine


def _summary(frame: pd.DataFrame, group_cols: list[str]) -> pd.DataFrame:
    grouped = frame.groupby(group_cols, dropna=False)
    rows: list[dict[str, object]] = []
    for keys, group in grouped:
        if not isinstance(keys, tuple):
            keys = (keys,)
        returns = group["return_pct"].astype(float)
        row = {group_cols[i]: keys[i] for i in range(len(group_cols))}
        row.update(
            {
                "trades": int(len(group)),
                "win_rate": float((returns > 0).mean()),
                "avg_return": float(returns.mean()),
                "median_return": float(returns.median()),
                "compounded_return": float((1.0 + returns).prod() - 1.0),
                "profit_factor": profit_factor(returns),
            }
        )
        rows.append(row)
    return pd.DataFrame(rows).sort_values("trades", ascending=False).reset_index(drop=True)


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    indexed, source = _load_indicator_snapshot(base_dir)

    engine = DragonRuleEngine(config=alpha_first_glued_refined_hot_cap_config())
    _, trades = engine.run(indexed)
    trades = trades[
        (trades["buy_date"] >= START_DATE)
        & (trades["buy_date"] <= END_DATE)
        & (trades["sell_date"] >= START_DATE)
        & (trades["sell_date"] <= END_DATE)
    ].copy()

    layer_summary = _summary(trades, ["buy_reason_layer", "sell_reason_layer"])
    family_summary = _summary(trades, ["buy_reason_family", "sell_reason_family"])
    entry_summary = _summary(trades, ["buy_reason_layer", "buy_reason_family"])
    exit_summary = _summary(trades, ["sell_reason_layer", "sell_reason_family"])

    layer_summary.to_csv(base_dir / "dragon_layered_pnl_attribution.csv", index=False, encoding="utf-8-sig")
    family_summary.to_csv(base_dir / "dragon_layered_family_pnl_attribution.csv", index=False, encoding="utf-8-sig")
    entry_summary.to_csv(base_dir / "dragon_layered_entry_pnl_attribution.csv", index=False, encoding="utf-8-sig")
    exit_summary.to_csv(base_dir / "dragon_layered_exit_pnl_attribution.csv", index=False, encoding="utf-8-sig")

    lines: list[str] = [
        "# Dragon Layered PnL Attribution",
        "",
        f"- window: `{START_DATE} -> {END_DATE}`",
        "- branch: `alpha_first_glued_refined_hot_cap` (RC1)",
        f"- indicator source: `{source}`",
        f"- trades: `{int(len(trades))}`",
        "",
        "## Entry-Layer x Exit-Layer",
    ]
    for _, row in layer_summary.iterrows():
        lines.append(
            "- "
            f"{row['buy_reason_layer']} -> {row['sell_reason_layer']}: "
            f"trades `{int(row['trades'])}`, "
            f"win_rate `{format_pct(float(row['win_rate']))}`, "
            f"avg_return `{format_pct(float(row['avg_return']))}`, "
            f"PF `{format_num(float(row['profit_factor']))}`"
        )

    lines.extend(
        [
            "",
            "## Top Entry Families",
        ]
    )
    for _, row in entry_summary.head(10).iterrows():
        lines.append(
            "- "
            f"{row['buy_reason_layer']} / {row['buy_reason_family']}: "
            f"trades `{int(row['trades'])}`, "
            f"avg_return `{format_pct(float(row['avg_return']))}`, "
            f"PF `{format_num(float(row['profit_factor']))}`"
        )

    lines.extend(
        [
            "",
            "## Top Exit Families",
        ]
    )
    for _, row in exit_summary.head(10).iterrows():
        lines.append(
            "- "
            f"{row['sell_reason_layer']} / {row['sell_reason_family']}: "
            f"trades `{int(row['trades'])}`, "
            f"avg_return `{format_pct(float(row['avg_return']))}`, "
            f"PF `{format_num(float(row['profit_factor']))}`"
        )

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


if __name__ == "__main__":
    main()