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

from __future__ import annotations

from pathlib import Path

import pandas as pd

from dragon_rule_catalog import classify_entry_reason, classify_exit_reason
from dragon_shared import END_DATE, START_DATE, format_num, format_pct, profit_factor


DEFAULT_BRANCH = "alpha_first_glued_refined_hot_cap"
DETAIL_OUTPUT = "dragon_rule_layer_attribution.csv"
SUMMARY_OUTPUT = "dragon_rule_layer_attribution_summary.csv"
REPORT_OUTPUT = "dragon_rule_layer_attribution.md"


def _aggregate(df: pd.DataFrame, layer_col: str, family_col: str, label: str) -> pd.DataFrame:
    if df.empty:
        return pd.DataFrame(columns=["view", "layer", "family", "trades", "win_rate", "avg_return", "median_return", "profit_factor"])

    records: list[dict[str, object]] = []
    grouped = df.groupby([layer_col, family_col], dropna=False)
    for (layer_value, family_value), group in grouped:
        returns = group["return_pct"].astype(float)
        row: dict[str, object] = {
            "view": label,
            "layer": str(layer_value),
            "family": str(family_value),
        }
        row.update(
            {
                "trades": int(len(group)),
                "win_rate": float((returns > 0).mean()),
                "avg_return": float(returns.mean()),
                "median_return": float(returns.median()),
                "profit_factor": profit_factor(returns),
            }
        )
        records.append(row)
    return pd.DataFrame(records).sort_values(["trades", "avg_return"], ascending=[False, False])


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    details_path = base_dir / "dragon_historical_trade_details.csv"
    details = pd.read_csv(details_path, encoding="utf-8-sig")

    details = details[details["branch"] == DEFAULT_BRANCH].copy()
    details = details[
        (details["buy_date"] >= START_DATE)
        & (details["buy_date"] <= END_DATE)
        & (details["sell_date"] >= START_DATE)
        & (details["sell_date"] <= END_DATE)
    ].copy()

    buy_meta = details["buy_reason"].map(classify_entry_reason)
    sell_meta = details["sell_reason"].map(classify_exit_reason)

    details["buy_layer"] = [meta.layer.value for meta in buy_meta]
    details["buy_family"] = [meta.family.value for meta in buy_meta]
    details["buy_code"] = [meta.code for meta in buy_meta]
    details["sell_layer"] = [meta.layer.value for meta in sell_meta]
    details["sell_family"] = [meta.family.value for meta in sell_meta]
    details["sell_code"] = [meta.code for meta in sell_meta]

    detail_cols = [
        "branch",
        "buy_date",
        "buy_reason",
        "buy_layer",
        "buy_family",
        "buy_code",
        "sell_date",
        "sell_reason",
        "sell_layer",
        "sell_family",
        "sell_code",
        "holding_days",
        "return_pct",
    ]
    detail_df = details[detail_cols].copy()
    detail_df.to_csv(base_dir / DETAIL_OUTPUT, index=False, encoding="utf-8-sig")

    buy_summary = _aggregate(detail_df, "buy_layer", "buy_family", label="entry_layer_family")
    sell_summary = _aggregate(detail_df, "sell_layer", "sell_family", label="exit_layer_family")
    summary_df = pd.concat([buy_summary, sell_summary], ignore_index=True)
    summary_df.to_csv(base_dir / SUMMARY_OUTPUT, index=False, encoding="utf-8-sig")

    unknown_entry = detail_df[detail_df["buy_layer"] == "unknown"]["buy_reason"].value_counts()
    unknown_exit = detail_df[detail_df["sell_layer"] == "unknown"]["sell_reason"].value_counts()

    lines: list[str] = [
        "# Dragon Rule Layer Attribution",
        "",
        f"- branch: `{DEFAULT_BRANCH}`",
        f"- evaluation_window: `{START_DATE}` to `{END_DATE}`",
        f"- trades: `{len(detail_df)}`",
        "",
        "## Entry Layer Summary",
    ]

    if buy_summary.empty:
        lines.append("- no data")
    else:
        for _, row in buy_summary.iterrows():
            lines.append(
                "- "
                f"`{row['layer']}/{row['family']}` "
                f"trades `{int(row['trades'])}` "
                f"win_rate `{format_pct(float(row['win_rate']))}` "
                f"avg_return `{format_pct(float(row['avg_return']))}` "
                f"profit_factor `{format_num(float(row['profit_factor']))}`"
            )

    lines.extend(["", "## Exit Layer Summary"])
    if sell_summary.empty:
        lines.append("- no data")
    else:
        for _, row in sell_summary.iterrows():
            lines.append(
                "- "
                f"`{row['layer']}/{row['family']}` "
                f"trades `{int(row['trades'])}` "
                f"win_rate `{format_pct(float(row['win_rate']))}` "
                f"avg_return `{format_pct(float(row['avg_return']))}` "
                f"profit_factor `{format_num(float(row['profit_factor']))}`"
            )

    lines.extend(["", "## Unknown Mapping Audit"])
    if unknown_entry.empty and unknown_exit.empty:
        lines.append("- no unknown reason mapping")
    else:
        if not unknown_entry.empty:
            for reason, count in unknown_entry.items():
                lines.append(f"- unknown entry reason `{reason}`: `{int(count)}`")
        if not unknown_exit.empty:
            for reason, count in unknown_exit.items():
                lines.append(f"- unknown exit reason `{reason}`: `{int(count)}`")

    lines.extend(
        [
            "",
            "## Artifacts",
            f"- `{DETAIL_OUTPUT}`",
            f"- `{SUMMARY_OUTPUT}`",
        ]
    )

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


if __name__ == "__main__":
    main()