cyb50-quant/research/dragon/v2/dragon_refined_alpha_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, alpha_first_selective_veto_config
from dragon_strategy import DragonRuleEngine

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


BUY_REASON_STATE = {
    "deep_oversold_rebound_buy": "low_oversold_regime",
    "oversold_recovery_buy": "low_oversold_regime",
    "oversold_reversal_after_ql_buy": "rebound_after_sell_regime",
    "post_sell_rebound_buy": "rebound_after_sell_regime",
    "post_washout_kdj_reentry_buy": "rebound_after_sell_regime",
    "predictive_error_reentry_buy": "rebound_after_sell_regime",
    "hot_exit_reentry_buy": "rebound_after_sell_regime",
    "early_crash_probe_buy": "crash_probe_regime",
    "dual_gold_resonance_buy": "low_oversold_regime",
    "glued_buy": "mid_regime",
    "non_glued_positive_expansion_buy": "high_regime",
}


SELL_REASON_MANAGEMENT = {
    "crash_protection_exit": "predictive_risk_exit",
    "predictive_b1_break_exit": "predictive_risk_exit",
    "prewarning_reduction_exit": "prewarning_exit",
    "high_regime_momentum_break": "prewarning_exit",
    "high_regime_confirmed_exit:kdj_sell": "confirmed_trend_exit",
    "ql_high_zone_take_profit": "high_regime_take_profit",
    "ql_mid_zone_take_profit": "high_regime_take_profit",
    "medium_hot_take_profit": "high_regime_take_profit",
    "high_zone_post_ql_fade_exit": "ql_followthrough_exit",
    "post_ql_decay_exit": "ql_followthrough_exit",
    "post_dual_sell_decay_exit": "ql_followthrough_exit",
    "knife_take_profit_1": "first_take_profit",
    "knife_take_profit_2_glued": "first_take_profit",
    "knife_take_profit_2_wait_ql_s": "first_take_profit",
    "early_positive_take_profit": "first_take_profit",
    "oversold_rebound_take_profit": "first_take_profit",
    "glued_exit:kdj_sell": "confirmed_trend_exit",
    "small_positive_a1_declining:ql_sell": "confirmed_trend_exit",
    "negative_a1_no_b1_recovery:kdj_sell": "negative_a1_exit",
    "negative_a1_no_b1_recovery:ql_sell": "negative_a1_exit",
    "negative_a1_b1_not_strong:kdj_sell": "negative_a1_exit",
    "low_zone_dual_gold_exit:kdj_sell": "negative_a1_exit",
    "hard_exit:kdj_sell": "hard_risk_exit",
    "hard_exit:ql_sell": "hard_risk_exit",
    "early_failed_rebound_exit": "predictive_risk_exit",
}


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 _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 _format_pct(value: float) -> str:
    if pd.isna(value):
        return "NA"
    if value == float("inf"):
        return "inf"
    return f"{value:.2%}"


def _format_num(value: float) -> str:
    if pd.isna(value):
        return "NA"
    if value == float("inf"):
        return "inf"
    return f"{value:.2f}"


def _entry_family(reason: str) -> str:
    return str(reason).split(":", 1)[0]


def _entry_variant(reason: str) -> str:
    parts = str(reason).split(":", 1)
    return "base" if len(parts) == 1 else parts[1]


def _infer_state_layer(buy_reason: str, buy_c1: float) -> str:
    state = BUY_REASON_STATE.get(_entry_family(buy_reason))
    if state == "mid_regime":
        if buy_c1 < 20:
            return "low_oversold_regime"
        if buy_c1 >= 70:
            return "high_regime"
    return state or "mid_regime"


def _infer_exit_management_layer(sell_reason: str) -> str:
    return SELL_REASON_MANAGEMENT.get(sell_reason, "default_exit_management")


def _entry_role(reason: str) -> str:
    family = _entry_family(reason)
    if family in {"glued_buy", "early_crash_probe_buy", "oversold_recovery_buy"}:
        return "core_alpha_family"
    if reason in {"deep_oversold_rebound_buy:classic_oversold", "dual_gold_resonance_buy"}:
        return "support_alpha_family"
    if family in {"predictive_error_reentry_buy", "hot_exit_reentry_buy"}:
        return "bridge_reentry_family"
    if family == "post_washout_kdj_reentry_buy":
        return "workbook_restart_family"
    if family in {"post_sell_rebound_buy", "oversold_reversal_after_ql_buy"}:
        return "secondary_research_family"
    if family == "deep_oversold_rebound_buy":
        return "weak_research_family"
    return "other_family"


def _group_stats(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)}
        returns = group["return_pct"].astype(float)
        row.update(
            {
                "trades": int(len(group)),
                "trade_share": float(len(group) / len(df)),
                "win_rate": float((returns > 0).mean()),
                "avg_return": float(returns.mean()),
                "median_return": float(returns.median()),
                "sum_return_pct": float(returns.sum()),
                "profit_factor": _profit_factor(returns),
                "avg_holding_days": float(group["holding_days"].mean()),
                "avg_mfe_pct": float(group["mfe_pct"].mean()),
                "avg_mae_pct": float(group["mae_pct"].mean()),
                "avg_giveback_from_peak_pct": float(group["giveback_from_peak_pct"].mean()),
                "avg_entry_forward_5d_pct": float(group["entry_forward_5d_pct"].mean()),
                "avg_exit_followthrough_5d_pct": float(group["exit_followthrough_5d_pct"].mean()),
                "avg_exit_rebound_5d_pct": float(group["exit_rebound_5d_pct"].mean()),
            }
        )
        rows.append(row)
    return pd.DataFrame(rows)


def _top_value(group: pd.DataFrame, col: str) -> str:
    counts = group[col].value_counts(dropna=False)
    if counts.empty:
        return ""
    return str(counts.index[0])


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 _recheck_verdict(row: pd.Series) -> tuple[str, str]:
    ret = float(row["return_pct"])
    mfe = float(row["mfe_pct"])
    holding = int(row["holding_days"])
    forward = float(row["entry_forward_5d_pct"])
    follow = float(row["exit_followthrough_5d_pct"])
    replacement_ret = row["replacement_return_pct"]

    if ret < 0 and holding <= 10 and follow <= 0:
        return "KEEP_REMOVAL", "short loser and price still weakened after the exit"
    if ret < 0 and mfe <= 0.02:
        return "KEEP_REMOVAL", "trade never developed enough profit room to defend inclusion"
    if ret < 0 and forward <= 0:
        return "KEEP_REMOVAL", "entry had no useful short-term follow-through and remained weak"
    if ret > 0 and ret <= 0.01 and pd.notna(replacement_ret) and float(replacement_ret) <= ret:
        return "OBSERVE_REMOVAL", "micro-winner but replacement path is not clearly worse"
    if ret > 0:
        return "OVER_REMOVAL", "removed trade kept meaningful alpha and should not be deleted silently"
    return "KEEP_REMOVAL", "removed trade remains weak under the alpha-first objective"


def _build_trade_quality(trades: pd.DataFrame, indicators: pd.DataFrame) -> pd.DataFrame:
    trades = trades.copy()
    trades["buy_dt"] = pd.to_datetime(trades["buy_date"])
    trades["sell_dt"] = pd.to_datetime(trades["sell_date"])
    trades["sell_year"] = trades["sell_dt"].dt.year.astype(int)
    trades["holding_bucket"] = trades["holding_days"].astype(int).map(_holding_bucket)

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

    buy_a1: list[float] = []
    buy_b1: list[float] = []
    buy_c1: list[float] = []
    sell_a1: list[float] = []
    sell_b1: list[float] = []
    sell_c1: list[float] = []
    mfe_list: list[float] = []
    mae_list: list[float] = []
    giveback_list: list[float] = []
    entry_forward_list: list[float] = []
    exit_followthrough_list: list[float] = []
    exit_rebound_list: list[float] = []

    for _, trade in trades.iterrows():
        buy_date = pd.Timestamp(trade["buy_date"]).date()
        sell_date = pd.Timestamp(trade["sell_date"]).date()
        buy_row = indicator_by_date.loc[buy_date]
        sell_row = indicator_by_date.loc[sell_date]

        buy_a1.append(float(buy_row["a1"]))
        buy_b1.append(float(buy_row["b1"]))
        buy_c1.append(float(buy_row["c1"]))
        sell_a1.append(float(sell_row["a1"]))
        sell_b1.append(float(sell_row["b1"]))
        sell_c1.append(float(sell_row["c1"]))

        entry_price = float(trade["buy_price"])
        exit_price = float(trade["sell_price"])
        buy_idx = pos_lookup[trade["buy_date"]]
        sell_idx = pos_lookup[trade["sell_date"]]

        window = indicators[(indicators["date"] >= trade["buy_dt"]) & (indicators["date"] <= trade["sell_dt"])]
        max_high = float(window["high"].max())
        min_low = float(window["low"].min())
        mfe_list.append(max_high / entry_price - 1.0)
        mae_list.append(min_low / entry_price - 1.0)
        giveback_list.append(exit_price / max_high - 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)
        exit_rebound_list.append(float("nan") if sell_future.empty else float(sell_future["high"].max()) / exit_price - 1.0)

    trades["buy_a1"] = buy_a1
    trades["buy_b1"] = buy_b1
    trades["buy_c1"] = buy_c1
    trades["sell_a1"] = sell_a1
    trades["sell_b1"] = sell_b1
    trades["sell_c1"] = sell_c1
    trades["mfe_pct"] = mfe_list
    trades["mae_pct"] = mae_list
    trades["giveback_from_peak_pct"] = giveback_list
    trades["entry_forward_5d_pct"] = entry_forward_list
    trades["exit_followthrough_5d_pct"] = exit_followthrough_list
    trades["exit_rebound_5d_pct"] = exit_rebound_list
    trades["entry_family"] = trades["buy_reason"].map(_entry_family)
    trades["entry_variant"] = trades["buy_reason"].map(_entry_variant)
    trades["entry_role"] = trades["buy_reason"].map(_entry_role)
    trades["market_state_layer"] = trades.apply(lambda row: _infer_state_layer(str(row["buy_reason"]), float(row["buy_c1"])), axis=1)
    trades["exit_management_layer"] = trades["sell_reason"].map(_infer_exit_management_layer)
    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, 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 _branch_snapshot(df: pd.DataFrame) -> dict[str, float]:
    returns = df["return_pct"].astype(float)
    return {
        "trades": float(len(df)),
        "win_rate": float((returns > 0).mean()),
        "avg_return": float(returns.mean()),
        "profit_factor": _profit_factor(returns),
        "avg_mfe": float(df["mfe_pct"].mean()),
        "avg_mae": float(df["mae_pct"].mean()),
    }


def _family_decomposition(refined: pd.DataFrame) -> pd.DataFrame:
    level_frames: list[pd.DataFrame] = []
    for level_name, group_cols in [
        ("entry_family", ["entry_family", "entry_role"]),
        ("entry_reason", ["buy_reason", "entry_role", "market_state_layer"]),
    ]:
        frame = _group_stats(refined, group_cols)
        if "entry_family" in frame.columns:
            frame["group_name"] = frame["entry_family"]
        else:
            frame["group_name"] = frame["buy_reason"]
        frame["decomposition_level"] = level_name
        top_exit = []
        for _, row in frame.iterrows():
            if level_name == "entry_family":
                group = refined[refined["entry_family"] == row["group_name"]]
            else:
                group = refined[refined["buy_reason"] == row["group_name"]]
            top_exit.append(_top_value(group, "sell_reason"))
        frame["top_exit_reason"] = top_exit
        level_frames.append(frame)

    result = pd.concat(level_frames, ignore_index=True, sort=False)
    result = result.sort_values(["decomposition_level", "sum_return_pct", "trades"], ascending=[True, False, False]).reset_index(drop=True)
    result["contribution_rank"] = result.groupby("decomposition_level")["sum_return_pct"].rank(method="dense", ascending=False).astype(int)
    cols = [
        "decomposition_level",
        "group_name",
        "entry_role",
        "market_state_layer",
        "buy_reason",
        "entry_family",
        "top_exit_reason",
        "trades",
        "trade_share",
        "win_rate",
        "avg_return",
        "median_return",
        "sum_return_pct",
        "profit_factor",
        "avg_holding_days",
        "avg_mfe_pct",
        "avg_mae_pct",
        "avg_giveback_from_peak_pct",
        "avg_entry_forward_5d_pct",
        "avg_exit_followthrough_5d_pct",
        "avg_exit_rebound_5d_pct",
        "contribution_rank",
    ]
    return result[[col for col in cols if col in result.columns]].copy()


def _alpha_attribution(refined: pd.DataFrame) -> pd.DataFrame:
    frame = _group_stats(
        refined,
        ["market_state_layer", "entry_family", "buy_reason", "exit_management_layer", "sell_reason"],
    )
    frame["attribution_label"] = frame.apply(
        lambda row: "core_alpha_source"
        if row["sum_return_pct"] > 0.20 and row["avg_return"] > 0
        else "drag_source"
        if row["sum_return_pct"] < 0
        else "mixed_source",
        axis=1,
    )
    frame = frame.sort_values(["sum_return_pct", "trades"], ascending=[False, False]).reset_index(drop=True)
    frame["sum_return_rank"] = frame["sum_return_pct"].rank(method="dense", ascending=False).astype(int)
    return frame


def _removed_trade_recheck(alpha: pd.DataFrame, refined: pd.DataFrame, workbook_events: pd.DataFrame) -> pd.DataFrame:
    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) - _trade_key(refined)),
        columns=["buy_date", "sell_date", "buy_reason", "sell_reason"],
    )

    rows: list[dict[str, object]] = []
    for _, removed_row in removed.iterrows():
        trade = alpha[
            (alpha["buy_date"] == removed_row["buy_date"])
            & (alpha["sell_date"] == removed_row["sell_date"])
            & (alpha["buy_reason"] == removed_row["buy_reason"])
            & (alpha["sell_reason"] == removed_row["sell_reason"])
        ].iloc[0]

        sell_dt = pd.Timestamp(trade["sell_date"])
        replacement = refined[
            (pd.to_datetime(refined["buy_date"]) > sell_dt)
            & (pd.to_datetime(refined["buy_date"]) <= sell_dt + pd.Timedelta(days=10))
        ].sort_values("buy_date")
        replacement_row = replacement.iloc[0] if not replacement.empty else None
        replacement_ret = float("nan") if replacement_row is None else float(replacement_row["return_pct"])
        verdict, verdict_reason = _recheck_verdict(
            pd.Series({**trade.to_dict(), "replacement_return_pct": replacement_ret})
        )

        rows.append(
            {
                "buy_date": trade["buy_date"],
                "sell_date": trade["sell_date"],
                "buy_reason": trade["buy_reason"],
                "sell_reason": trade["sell_reason"],
                "entry_family": trade["entry_family"],
                "entry_role": trade["entry_role"],
                "market_state_layer": trade["market_state_layer"],
                "exit_management_layer": trade["exit_management_layer"],
                "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"]),
                "giveback_from_peak_pct": float(trade["giveback_from_peak_pct"]),
                "entry_forward_5d_pct": float(trade["entry_forward_5d_pct"]),
                "exit_followthrough_5d_pct": float(trade["exit_followthrough_5d_pct"]),
                "exit_rebound_5d_pct": float(trade["exit_rebound_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": replacement_ret,
                "replacement_gap_days": float("nan")
                if replacement_row is None
                else int((pd.Timestamp(replacement_row["buy_date"]) - sell_dt).days),
                "verdict": verdict,
                "verdict_reason": verdict_reason,
            }
        )

    return pd.DataFrame(rows).sort_values(["veto_bucket", "buy_date"]).reset_index(drop=True)


def _winner_structure(refined: pd.DataFrame) -> pd.DataFrame:
    winners = refined[refined["return_pct"] > 0].copy()
    if winners.empty:
        return pd.DataFrame()
    frame = _group_stats(winners, ["entry_family", "holding_bucket"])
    frame = frame.sort_values(["sum_return_pct", "trades"], ascending=[False, False]).reset_index(drop=True)
    return frame


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 = _run_branch(indicators, alpha_first_selective_veto_config())
    refined = _run_branch(indicators, alpha_first_glued_refined_hot_cap_config())

    family_decomposition = _family_decomposition(refined)
    alpha_attribution = _alpha_attribution(refined)
    removed_recheck = _removed_trade_recheck(alpha, refined, workbook_events)
    winner_structure = _winner_structure(refined)

    family_decomposition.to_csv(base_dir / "dragon_refined_family_decomposition.csv", index=False, encoding="utf-8-sig")
    alpha_attribution.to_csv(base_dir / "dragon_refined_alpha_attribution.csv", index=False, encoding="utf-8-sig")
    removed_recheck.to_csv(base_dir / "dragon_refined_removed_trade_recheck.csv", index=False, encoding="utf-8-sig")

    refined_snapshot = _branch_snapshot(refined)
    alpha_snapshot = _branch_snapshot(alpha)

    removed_pf = _profit_factor(removed_recheck["return_pct"]) if not removed_recheck.empty else float("nan")
    removed_pf_text = _format_num(removed_pf)

    top_family = family_decomposition[
        (family_decomposition["decomposition_level"] == "entry_family") & (family_decomposition["trades"] >= 3)
    ].head(5)
    weak_family = family_decomposition[
        (family_decomposition["decomposition_level"] == "entry_reason")
        & (family_decomposition["trades"] >= 1)
        & (family_decomposition["entry_role"].isin(["weak_research_family", "secondary_research_family"]))
    ].sort_values(["avg_return", "sum_return_pct"]).head(5)
    top_combo = alpha_attribution.head(8)
    drag_combo = alpha_attribution[alpha_attribution["sum_return_pct"] < 0].head(8)
    winner_top = winner_structure.head(5)

    lines = [
        "# Dragon Refined Edge Review",
        "",
        "## Scope",
        "- Target branch: `alpha_first_glued_refined_hot_cap`",
        "- Control branch: `alpha_first_selective_veto`",
        "- Evaluation window: `2016-01-01` to `2025-12-31`",
        "",
        "## Headline",
        f"- control: trades `{int(alpha_snapshot['trades'])}`, win_rate `{_format_pct(alpha_snapshot['win_rate'])}`, avg_return `{_format_pct(alpha_snapshot['avg_return'])}`, profit_factor `{_format_num(alpha_snapshot['profit_factor'])}`",
        f"- refined: trades `{int(refined_snapshot['trades'])}`, win_rate `{_format_pct(refined_snapshot['win_rate'])}`, avg_return `{_format_pct(refined_snapshot['avg_return'])}`, profit_factor `{_format_num(refined_snapshot['profit_factor'])}`",
        f"- refined minus control: trades `{int(refined_snapshot['trades'] - alpha_snapshot['trades'])}`, avg_return `{_format_pct(refined_snapshot['avg_return'] - alpha_snapshot['avg_return'])}`, profit_factor `{_format_num(refined_snapshot['profit_factor'] - alpha_snapshot['profit_factor'])}`",
        "",
        "## Main Edge Source",
        "- Refined alpha is still primarily a `glued_buy` story, but now with stricter removal of weak short-holding glued trades.",
        "- The branch is not winning by adding new complex trade paths; it is winning by deleting low-quality short trades while preserving the medium and long-holding winners.",
        "",
        "## Entry Family Decomposition",
    ]

    for _, row in top_family.iterrows():
        lines.append(
            f"- `{row['group_name']}` [{row['entry_role']}]: trades `{int(row['trades'])}`, share `{_format_pct(float(row['trade_share']))}`, "
            f"avg_return `{_format_pct(float(row['avg_return']))}`, sum_return `{_format_pct(float(row['sum_return_pct']))}`, "
            f"profit_factor `{_format_num(float(row['profit_factor']))}`, top_exit `{row['top_exit_reason']}`"
        )

    lines.extend(["", "## Weak Research Pockets"])
    for _, row in weak_family.iterrows():
        lines.append(
            f"- `{row['group_name']}`: trades `{int(row['trades'])}`, avg_return `{_format_pct(float(row['avg_return']))}`, "
            f"sum_return `{_format_pct(float(row['sum_return_pct']))}`, profit_factor `{_format_num(float(row['profit_factor']))}`"
        )

    lines.extend(["", "## Kept Winner Structure"])
    for _, row in winner_top.iterrows():
        lines.append(
            f"- `{row['entry_family']} / {row['holding_bucket']}`: winners `{int(row['trades'])}`, avg_return `{_format_pct(float(row['avg_return']))}`, "
            f"sum_return `{_format_pct(float(row['sum_return_pct']))}`"
        )

    lines.extend(["", "## Entry / Exit Interaction Attribution"])
    for _, row in top_combo.iterrows():
        lines.append(
            f"- positive `{row['market_state_layer']} / {row['buy_reason']} -> {row['sell_reason']}`: trades `{int(row['trades'])}`, "
            f"sum_return `{_format_pct(float(row['sum_return_pct']))}`, avg_return `{_format_pct(float(row['avg_return']))}`, "
            f"PF `{_format_num(float(row['profit_factor']))}`"
        )

    lines.extend(["", "## Drag Interaction Pockets"])
    for _, row in drag_combo.iterrows():
        lines.append(
            f"- drag `{row['market_state_layer']} / {row['buy_reason']} -> {row['sell_reason']}`: trades `{int(row['trades'])}`, "
            f"sum_return `{_format_pct(float(row['sum_return_pct']))}`, avg_return `{_format_pct(float(row['avg_return']))}`, "
            f"PF `{_format_num(float(row['profit_factor']))}`"
        )

    lines.extend(
        [
            "",
            "## Removed-Trade Recheck",
            f"- removed trades vs control: `{int(len(removed_recheck))}`",
            f"- removed-set avg_return `{_format_pct(float(removed_recheck['return_pct'].mean()))}`",
            f"- removed-set win_rate `{_format_pct(float((removed_recheck['return_pct'] > 0).mean()))}`",
            f"- removed-set profit_factor `{removed_pf_text}`",
            f"- KEEP_REMOVAL `{int((removed_recheck['verdict'] == 'KEEP_REMOVAL').sum())}` | OBSERVE_REMOVAL `{int((removed_recheck['verdict'] == 'OBSERVE_REMOVAL').sum())}` | OVER_REMOVAL `{int((removed_recheck['verdict'] == 'OVER_REMOVAL').sum())}`",
        ]
    )

    for bucket, group in removed_recheck.groupby("veto_bucket", dropna=False):
        lines.append(
            f"- `{bucket}`: trades `{len(group)}`, avg_return `{_format_pct(float(group['return_pct'].mean()))}`, "
            f"avg_holding `{group['holding_days'].mean():.1f}`, avg_mfe `{_format_pct(float(group['mfe_pct'].mean()))}`"
        )

    lines.extend(
        [
            "",
            "## Quant Judgment",
            "- Core alpha remains concentrated in `glued_buy`, `early_crash_probe_buy`, and the preserved medium/long holding structure.",
            "- `dual_gold_resonance_buy` and `deep_oversold_rebound_buy:classic_oversold` remain support families, not the main alpha engine.",
            "- Weak pockets still exist in secondary rebound / weak deep-oversold variants, but they are not where the refined branch gets its headline improvement.",
            "- The refined branch improves mainly by deleting low-quality short glued trades; this remains explainable and not dependent on deleting profitable samples.",
            "- The next step should therefore move to execution-aware robustness, not back to workbook-style residual tuning.",
        ]
    )

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


if __name__ == "__main__":
    main()