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

from __future__ import annotations

from pathlib import Path

import pandas as pd


def _load_csv(base_dir: Path, name: str) -> pd.DataFrame:
    return pd.read_csv(base_dir / name, encoding="utf-8-sig")


def _profit_factor(series: pd.Series) -> float:
    gross_profit = series[series > 0].sum()
    gross_loss = -series[series < 0].sum()
    if gross_loss == 0:
        return float("inf") if gross_profit > 0 else 0.0
    return float(gross_profit / gross_loss)


def _summarize(df: pd.DataFrame, group_cols: list[str]) -> pd.DataFrame:
    rows: list[dict[str, object]] = []
    for key, group in df.groupby(group_cols, dropna=False):
        if not isinstance(key, tuple):
            key = (key,)
        row = {col: val for col, val in zip(group_cols, key)}
        row["trades"] = int(len(group))
        row["loss_trades"] = int((group["return_pct"] < 0).sum())
        row["win_rate"] = float((group["return_pct"] > 0).mean())
        row["avg_return"] = float(group["return_pct"].mean())
        row["avg_mfe"] = float(group["mfe_pct"].mean())
        row["avg_mae"] = float(group["mae_pct"].mean())
        row["avg_buy_plus_3d_return"] = float(group["buy_plus_3d_return"].mean())
        row["avg_sell_plus_3d_followthrough"] = float(group["sell_plus_3d_followthrough"].mean())
        row["profit_factor"] = _profit_factor(group["return_pct"])
        row["drag_score"] = float(len(group) * abs(min(float(group["return_pct"].mean()), 0.0)))
        rows.append(row)
    return pd.DataFrame(rows)


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    audit = _load_csv(base_dir, "dragon_short_holding_audit.csv")

    group_specs = [
        ("entry_family", ["holding_bucket", "entry_family"]),
        ("buy_reason", ["holding_bucket", "buy_reason"]),
        ("sell_reason", ["holding_bucket", "sell_reason"]),
        ("market_state", ["holding_bucket", "market_state_layer"]),
        ("entry_gate", ["holding_bucket", "entry_qualification_layer"]),
        ("path_combo", ["holding_bucket", "entry_family", "sell_reason"]),
    ]

    frames = []
    for group_type, cols in group_specs:
        frames.append(_summarize(audit, cols).assign(group_type=group_type))
    pressure = pd.concat(frames, ignore_index=True, sort=False)
    pressure = pressure.sort_values(["drag_score", "trades"], ascending=[False, False])
    pressure.to_csv(base_dir / "dragon_short_holding_family_pressure.csv", index=False, encoding="utf-8-sig")

    top_entry = pressure[pressure["group_type"] == "entry_family"].sort_values("drag_score", ascending=False).head(8)
    top_buy = pressure[pressure["group_type"] == "buy_reason"].sort_values("drag_score", ascending=False).head(8)
    top_path = pressure[pressure["group_type"] == "path_combo"].sort_values("drag_score", ascending=False).head(8)

    lines = [
        "# Dragon Short Holding Family Review",
        "",
        "- Scope: `alpha_first_selective_veto` short trades only.",
        "- Drag score definition: `trades * abs(min(avg_return, 0))`.",
        "",
        "## Top Entry-Family Drag",
    ]
    for _, row in top_entry.iterrows():
        lines.append(
            f"- `{row['holding_bucket']} / {row['entry_family']}`: trades `{int(row['trades'])}`, loss_trades `{int(row['loss_trades'])}`, "
            f"avg_return `{row['avg_return']:.2%}`, drag_score `{row['drag_score']:.4f}`"
        )

    lines.extend(["", "## Top Buy-Reason Drag"])
    for _, row in top_buy.iterrows():
        lines.append(
            f"- `{row['holding_bucket']} / {row['buy_reason']}`: trades `{int(row['trades'])}`, loss_trades `{int(row['loss_trades'])}`, "
            f"avg_return `{row['avg_return']:.2%}`, avg_buy_plus_3d `{row['avg_buy_plus_3d_return']:.2%}`, drag_score `{row['drag_score']:.4f}`"
        )

    lines.extend(["", "## Top Path Drag"])
    for _, row in top_path.iterrows():
        lines.append(
            f"- `{row['holding_bucket']} / {row['entry_family']} -> {row['sell_reason']}`: trades `{int(row['trades'])}`, "
            f"avg_return `{row['avg_return']:.2%}`, avg_sell_plus_3d `{row['avg_sell_plus_3d_followthrough']:.2%}`, drag_score `{row['drag_score']:.4f}`"
        )

    lead_entry = top_entry.iloc[0] if not top_entry.empty else None
    lead_path = top_path.iloc[0] if not top_path.empty else None
    lines.extend(["", "## Quant Judgment"])
    if lead_entry is not None:
        lines.append(
            f"- Lead short-holding drag family: `{lead_entry['holding_bucket']} / {lead_entry['entry_family']}` with drag_score `{lead_entry['drag_score']:.4f}`."
        )
    if lead_path is not None:
        lines.append(
            f"- Lead path-level drag: `{lead_path['holding_bucket']} / {lead_path['entry_family']} -> {lead_path['sell_reason']}`."
        )
    lines.append("- The next experiment pack should attack the highest drag family first and decide whether the issue is bad entry selection or premature exit handling.")

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


if __name__ == "__main__":
    main()