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

from __future__ import annotations

from pathlib import Path
from typing import Optional

import pandas as pd

from dragon_branch_configs import alpha_first_glued_refined_hot_cap_config, alpha_first_selective_veto_config
from dragon_strategy import DragonRuleEngine

START_DATE = "2016-01-01"
END_DATE = "2025-12-31"


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


def _load_indicator_snapshot(base_dir: Path) -> pd.DataFrame:
    df = _load_csv(base_dir, "dragon_indicator_snapshot.csv")
    df["date"] = pd.to_datetime(df["date"])
    return df.sort_values("date").reset_index(drop=True)


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 _pct(value: Optional[float]) -> str:
    if value is None or pd.isna(value):
        return "n/a"
    if value == float("inf"):
        return "inf"
    return f"{float(value):.2%}"


def _holding_bucket(days: int) -> str:
    if days <= 5:
        return "00-05d"
    if days <= 10:
        return "06-10d"
    if days <= 20:
        return "11-20d"
    if days <= 40:
        return "21-40d"
    return "41d+"


def _build_trade_quality(trades: pd.DataFrame, indicators: pd.DataFrame) -> pd.DataFrame:
    trades = trades.copy()
    trades["holding_bucket"] = trades["holding_days"].astype(int).map(_holding_bucket)

    pos_lookup = {dt.date().isoformat(): idx for idx, dt in enumerate(indicators["date"])}
    indicator_by_date = indicators.set_index(indicators["date"].dt.date)

    buy_a1: list[float] = []
    buy_b1: list[float] = []
    buy_c1: list[float] = []
    mfe_list: list[float] = []
    mae_list: list[float] = []
    entry_forward_list: list[float] = []
    exit_followthrough_list: list[float] = []

    for _, trade in trades.iterrows():
        buy_date = pd.Timestamp(trade["buy_date"]).date()
        entry_price = float(trade["buy_price"])
        exit_price = float(trade["sell_price"])
        buy_row = indicator_by_date.loc[buy_date]
        buy_a1.append(float(buy_row["a1"]))
        buy_b1.append(float(buy_row["b1"]))
        buy_c1.append(float(buy_row["c1"]))

        buy_idx = pos_lookup[trade["buy_date"]]
        sell_idx = pos_lookup[trade["sell_date"]]
        window = indicators[
            (indicators["date"] >= pd.Timestamp(trade["buy_date"])) & (indicators["date"] <= pd.Timestamp(trade["sell_date"]))
        ]
        mfe_list.append(float(window["high"].max()) / entry_price - 1.0)
        mae_list.append(float(window["low"].min()) / entry_price - 1.0)

        buy_future = indicators.iloc[buy_idx + 1 : buy_idx + 6]
        sell_future = indicators.iloc[sell_idx + 1 : sell_idx + 6]
        entry_forward_list.append(float("nan") if buy_future.empty else float(buy_future["close"].iloc[-1]) / entry_price - 1.0)
        exit_followthrough_list.append(float("nan") if sell_future.empty else float(sell_future["low"].min()) / exit_price - 1.0)

    trades["buy_a1"] = buy_a1
    trades["buy_b1"] = buy_b1
    trades["buy_c1"] = buy_c1
    trades["mfe_pct"] = mfe_list
    trades["mae_pct"] = mae_list
    trades["entry_forward_5d_pct"] = entry_forward_list
    trades["exit_followthrough_5d_pct"] = exit_followthrough_list
    return trades


def _run_branch(indicators: pd.DataFrame, config) -> pd.DataFrame:
    indexed = indicators.set_index("date", drop=False)
    engine = DragonRuleEngine(config=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()
    return _build_trade_quality(trades.copy(), indicators)


def _trade_key(df: pd.DataFrame) -> set[tuple[str, str, str, str]]:
    return set(zip(df["buy_date"], df["sell_date"], df["buy_reason"], df["sell_reason"]))


def _veto_bucket(c1: float, b1: float) -> str:
    if 23 <= c1 < 28 and b1 <= 0.02:
        return "low_weak_range"
    if 40 <= c1 < 75 and b1 >= 0.10:
        return "hot_positive_b1_cap75"
    return "other"


def _recommendation(row: pd.Series) -> tuple[str, str]:
    ret = float(row["return_pct"])
    mfe = float(row["mfe_pct"])
    holding = int(row["holding_days"])
    replacement_ret = row.get("replacement_return_pct")

    if ret < 0 and holding <= 10 and float(row["exit_followthrough_5d_pct"]) <= 0:
        return "KEEP_REMOVAL", "Removed trade is a short loser and price still weakens after exit."
    if ret < 0 and mfe <= 0.02:
        return "KEEP_REMOVAL", "Removed trade never developed enough profit room to defend its inclusion."
    if ret > 0:
        if ret <= 0.01 and pd.notna(replacement_ret) and float(replacement_ret) <= ret:
            return "OBSERVE_REMOVAL", "Removed trade is only a micro-winner and the replacement path is not stronger."
        return "OVER_REMOVAL", "Removed trade keeps meaningful alpha and should not be discarded without a better replacement."
    return "KEEP_REMOVAL", "Removed trade remains a weak short-holding sample under the alpha-first objective."


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    indicators = _load_indicator_snapshot(base_dir)
    workbook_events = _load_csv(base_dir, "true_trade_events.csv")
    alpha_trades = _run_branch(indicators, alpha_first_selective_veto_config())
    refined_trades = _run_branch(indicators, alpha_first_glued_refined_hot_cap_config())

    workbook_buy = set(workbook_events[(workbook_events["layer"] == "real_trade") & (workbook_events["side"] == "BUY")]["date"])
    workbook_sell = set(workbook_events[(workbook_events["layer"] == "real_trade") & (workbook_events["side"] == "SELL")]["date"])

    removed = pd.DataFrame(
        sorted(_trade_key(alpha_trades) - _trade_key(refined_trades)),
        columns=["buy_date", "sell_date", "buy_reason", "sell_reason"],
    )

    rows: list[dict[str, object]] = []
    for _, removed_row in removed.iterrows():
        trade = alpha_trades[
            (alpha_trades["buy_date"] == removed_row["buy_date"])
            & (alpha_trades["sell_date"] == removed_row["sell_date"])
            & (alpha_trades["buy_reason"] == removed_row["buy_reason"])
            & (alpha_trades["sell_reason"] == removed_row["sell_reason"])
        ].iloc[0]
        sell_dt = pd.Timestamp(trade["sell_date"])
        replacement = refined_trades[
            (pd.to_datetime(refined_trades["buy_date"]) > sell_dt)
            & (pd.to_datetime(refined_trades["buy_date"]) <= sell_dt + pd.Timedelta(days=10))
        ].sort_values("buy_date")
        replacement_row = replacement.iloc[0] if not replacement.empty else None

        recommendation, reason = _recommendation(
            pd.Series(
                {
                    **trade.to_dict(),
                    "replacement_return_pct": None if replacement_row is None else float(replacement_row["return_pct"]),
                }
            )
        )

        rows.append(
            {
                "buy_date": trade["buy_date"],
                "sell_date": trade["sell_date"],
                "buy_reason": trade["buy_reason"],
                "sell_reason": trade["sell_reason"],
                "veto_bucket": _veto_bucket(float(trade["buy_c1"]), float(trade["buy_b1"])),
                "holding_bucket": trade["holding_bucket"],
                "holding_days": int(trade["holding_days"]),
                "return_pct": float(trade["return_pct"]),
                "mfe_pct": float(trade["mfe_pct"]),
                "mae_pct": float(trade["mae_pct"]),
                "entry_forward_5d_pct": float(trade["entry_forward_5d_pct"]),
                "exit_followthrough_5d_pct": float(trade["exit_followthrough_5d_pct"]),
                "buy_a1": float(trade["buy_a1"]),
                "buy_b1": float(trade["buy_b1"]),
                "buy_c1": float(trade["buy_c1"]),
                "buy_aligned_with_workbook": trade["buy_date"] in workbook_buy,
                "sell_aligned_with_workbook": trade["sell_date"] in workbook_sell,
                "replacement_buy_date": "" if replacement_row is None else str(replacement_row["buy_date"]),
                "replacement_sell_date": "" if replacement_row is None else str(replacement_row["sell_date"]),
                "replacement_buy_reason": "" if replacement_row is None else str(replacement_row["buy_reason"]),
                "replacement_sell_reason": "" if replacement_row is None else str(replacement_row["sell_reason"]),
                "replacement_return_pct": float("nan") if replacement_row is None else float(replacement_row["return_pct"]),
                "replacement_gap_days": float("nan")
                if replacement_row is None
                else int((pd.Timestamp(replacement_row["buy_date"]) - sell_dt).days),
                "recommendation": recommendation,
                "recommendation_reason": reason,
            }
        )

    attribution = pd.DataFrame(rows).sort_values(["veto_bucket", "buy_date"]).reset_index(drop=True)
    attribution.to_csv(base_dir / "dragon_glued_refined_removed_trade_attribution.csv", index=False, encoding="utf-8-sig")

    pf = _profit_factor(attribution["return_pct"])
    pf_text = "inf" if pf == float("inf") else f"{pf:.2f}"
    lines = [
        "# Dragon Glued Refined Removed-Trade Review",
        "",
        "## Snapshot",
        f"- removed trades vs current alpha-first: `{len(attribution)}`",
        f"- avg_return of removed set: `{_pct(float(attribution['return_pct'].mean()))}`",
        f"- win_rate of removed set: `{_pct(float((attribution['return_pct'] > 0).mean()))}`",
        f"- profit_factor of removed set: `{pf_text}`",
        "",
        "## Recommendation Mix",
        f"- KEEP_REMOVAL: `{int((attribution['recommendation'] == 'KEEP_REMOVAL').sum())}`",
        f"- OBSERVE_REMOVAL: `{int((attribution['recommendation'] == 'OBSERVE_REMOVAL').sum())}`",
        f"- OVER_REMOVAL: `{int((attribution['recommendation'] == 'OVER_REMOVAL').sum())}`",
        "",
        "## Bucket View",
    ]

    for bucket, group in attribution.groupby("veto_bucket", dropna=False):
        lines.append(
            f"- `{bucket}`: trades `{len(group)}`, avg_return `{_pct(float(group['return_pct'].mean()))}`, "
            f"win_rate `{_pct(float((group['return_pct'] > 0).mean()))}`, avg_mfe `{_pct(float(group['mfe_pct'].mean()))}`, "
            f"avg_mae `{_pct(float(group['mae_pct'].mean()))}`"
        )

    lines.extend(
        [
            "",
            "## Quant Judgment",
            "- The refined branch mostly removes weak short-holding glued trades rather than medium-quality alpha trades.",
            "- If this review remains dominated by KEEP_REMOVAL and contains no meaningful OVER_REMOVAL bucket, the branch is structurally explainable rather than a black-box overfit.",
            "",
            "## Detailed Cards",
        ]
    )

    for _, row in attribution.iterrows():
        replacement = "none"
        if isinstance(row["replacement_buy_date"], str) and row["replacement_buy_date"]:
            replacement = (
                f"{row['replacement_buy_date']} -> {row['replacement_sell_date']} / "
                f"{row['replacement_buy_reason']} -> {row['replacement_sell_reason']} / "
                f"{_pct(row['replacement_return_pct'])}"
            )
        lines.extend(
            [
                f"### {row['buy_date']} -> {row['sell_date']}",
                f"- Bucket: `{row['veto_bucket']}` | holding `{row['holding_bucket']}`",
                f"- Trade: `{row['buy_reason']} -> {row['sell_reason']}` | return `{_pct(row['return_pct'])}` | holding `{int(row['holding_days'])}` days",
                f"- MFE / MAE: `{_pct(row['mfe_pct'])}` / `{_pct(row['mae_pct'])}`",
                f"- Entry 5d / Exit followthrough 5d: `{_pct(row['entry_forward_5d_pct'])}` / `{_pct(row['exit_followthrough_5d_pct'])}`",
                f"- Entry indicators: `a1={float(row['buy_a1']):.4f}` `b1={float(row['buy_b1']):.4f}` `c1={float(row['buy_c1']):.2f}`",
                f"- Workbook aligned: buy `{bool(row['buy_aligned_with_workbook'])}` / sell `{bool(row['sell_aligned_with_workbook'])}`",
                f"- Replacement path within 10d: `{replacement}`",
                f"- Recommendation: `{row['recommendation']}` | {row['recommendation_reason']}",
                "",
            ]
        )

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


if __name__ == "__main__":
    main()