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

from __future__ import annotations

from itertools import product
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_shared import END_DATE, START_DATE, profit_factor
from dragon_strategy import DragonRuleEngine


def _load_indicator_snapshot(base_dir: Path) -> pd.DataFrame:
    df = pd.read_csv(base_dir / "dragon_indicator_snapshot.csv", encoding="utf-8-sig")
    df["date"] = pd.to_datetime(df["date"])
    return df.set_index("date", drop=False)


def _load_true_trade_events(base_dir: Path) -> pd.DataFrame:
    return pd.read_csv(base_dir / "true_trade_events.csv", encoding="utf-8-sig")


def _event_match(strategy_events: pd.DataFrame, workbook_events: pd.DataFrame, side: str) -> tuple[int, int]:
    wb = set(workbook_events[(workbook_events["side"] == side) & (workbook_events["layer"] == "real_trade")]["date"])
    st = set(strategy_events[(strategy_events["side"] == side) & (strategy_events["layer"] == "real_trade")]["date"])
    return len(wb & st), len(st - wb)


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 _run(label: str, indicator_df: pd.DataFrame, workbook_events: pd.DataFrame, config) -> dict[str, object]:
    engine = DragonRuleEngine(config=config)
    events, trades = engine.run(indicator_df)
    events = events[(events["date"] >= START_DATE) & (events["date"] <= END_DATE)].copy()
    trades = trades[
        (trades["buy_date"] >= START_DATE)
        & (trades["buy_date"] <= END_DATE)
        & (trades["sell_date"] >= START_DATE)
        & (trades["sell_date"] <= END_DATE)
    ].copy()
    trades["holding_bucket"] = trades["holding_days"].astype(int).map(_holding_bucket)

    buy_overlap, buy_extra = _event_match(events, workbook_events, "BUY")
    sell_overlap, sell_extra = _event_match(events, workbook_events, "SELL")

    return {
        "label": label,
        "hot_c1_max": getattr(config, "glued_selective_hot_c1_max"),
        "hot_b1_min": getattr(config, "glued_selective_hot_b1_min"),
        "low_c1_min": getattr(config, "glued_selective_low_c1_min"),
        "low_b1_max": getattr(config, "glued_selective_low_b1_max"),
        "trades": int(len(trades)),
        "win_rate": float((trades["return_pct"] > 0).mean()) if not trades.empty else float("nan"),
        "avg_return": float(trades["return_pct"].mean()) if not trades.empty else float("nan"),
        "profit_factor": profit_factor(trades["return_pct"]) if not trades.empty else float("nan"),
        "real_buy_overlap": int(buy_overlap),
        "real_buy_extra": int(buy_extra),
        "real_sell_overlap": int(sell_overlap),
        "real_sell_extra": int(sell_extra),
        "short_00_05d_avg_return": float(trades[trades["holding_bucket"] == "00-05d"]["return_pct"].mean()),
        "short_06_10d_avg_return": float(trades[trades["holding_bucket"] == "06-10d"]["return_pct"].mean()),
    }


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    indicator_df = _load_indicator_snapshot(base_dir)
    workbook_events = _load_true_trade_events(base_dir)
    baseline = alpha_first_selective_veto_config()
    candidate = alpha_first_glued_refined_hot_cap_config()

    rows = [
        _run("current_alpha_control", indicator_df, workbook_events, baseline),
        _run("refined_candidate_baseline", indicator_df, workbook_events, candidate),
    ]

    for hot_c1_max, hot_b1_min, low_c1_min, low_b1_max in product([72.0, 75.0, 78.0], [0.09, 0.10, 0.11], [22.0, 23.0, 24.0], [0.01, 0.02, 0.03]):
        cfg = baseline.with_updates(
            glued_selective_hot_c1_min=40.0,
            glued_selective_hot_c1_max=hot_c1_max,
            glued_selective_hot_b1_min=hot_b1_min,
            glued_selective_low_c1_min=low_c1_min,
            glued_selective_low_c1_max=28.0,
            glued_selective_low_b1_max=low_b1_max,
        )
        label = f"hcap{int(hot_c1_max)}_hb1_{hot_b1_min:.2f}_lmin{int(low_c1_min)}_lb1_{low_b1_max:.2f}"
        rows.append(_run(label, indicator_df, workbook_events, cfg))

    result = pd.DataFrame(rows)
    result.to_csv(base_dir / "dragon_glued_refined_sensitivity.csv", index=False, encoding="utf-8-sig")

    candidate_row = result[result["label"] == "refined_candidate_baseline"].iloc[0]
    neighborhood = result[~result["label"].isin(["current_alpha_control", "refined_candidate_baseline"])].copy()
    neighborhood["delta_avg_return"] = neighborhood["avg_return"] - candidate_row["avg_return"]
    neighborhood["delta_profit_factor"] = neighborhood["profit_factor"] - candidate_row["profit_factor"]
    neighborhood["delta_real_buy_overlap"] = neighborhood["real_buy_overlap"] - candidate_row["real_buy_overlap"]
    neighborhood["delta_real_sell_overlap"] = neighborhood["real_sell_overlap"] - candidate_row["real_sell_overlap"]

    summary = pd.DataFrame(
        [
            {
                "scope": "candidate_neighborhood",
                "cases": int(len(neighborhood)),
                "avg_return_min": float(neighborhood["avg_return"].min()),
                "avg_return_max": float(neighborhood["avg_return"].max()),
                "profit_factor_min": float(neighborhood["profit_factor"].min()),
                "profit_factor_max": float(neighborhood["profit_factor"].max()),
                "real_buy_overlap_min": int(neighborhood["real_buy_overlap"].min()),
                "real_sell_overlap_min": int(neighborhood["real_sell_overlap"].min()),
                "count_better_avg_return": int((neighborhood["avg_return"] >= candidate_row["avg_return"]).sum()),
                "count_better_profit_factor": int((neighborhood["profit_factor"] >= candidate_row["profit_factor"]).sum()),
            }
        ]
    )
    summary.to_csv(base_dir / "dragon_glued_refined_sensitivity_summary.csv", index=False, encoding="utf-8-sig")

    top_avg = neighborhood.sort_values(["avg_return", "profit_factor"], ascending=[False, False]).head(10)
    robust = neighborhood[
        (neighborhood["avg_return"] >= candidate_row["avg_return"] - 0.0015)
        & (neighborhood["profit_factor"] >= candidate_row["profit_factor"] - 0.20)
        & (neighborhood["real_buy_overlap"] >= candidate_row["real_buy_overlap"] - 1)
        & (neighborhood["real_sell_overlap"] >= candidate_row["real_sell_overlap"] - 1)
    ].copy()

    lines = [
        "# Dragon Glued Refined Sensitivity",
        "",
        "- Scope: local neighborhood around the refined glued candidate, not a broad black-box search.",
        "",
        "## Candidate Baseline",
        f"- avg_return `{candidate_row['avg_return']:.2%}`, profit_factor `{candidate_row['profit_factor']:.2f}`, real BUY / SELL `{int(candidate_row['real_buy_overlap'])}/{int(candidate_row['real_sell_overlap'])}`",
        "",
        "## Neighborhood Summary",
        f"- tested cases: `{int(len(neighborhood))}`",
        f"- avg_return range: `{neighborhood['avg_return'].min():.2%}` to `{neighborhood['avg_return'].max():.2%}`",
        f"- profit_factor range: `{neighborhood['profit_factor'].min():.2f}` to `{neighborhood['profit_factor'].max():.2f}`",
        f"- overlap floor: BUY `{int(neighborhood['real_buy_overlap'].min())}`, SELL `{int(neighborhood['real_sell_overlap'].min())}`",
        f"- cases with avg_return >= candidate: `{int((neighborhood['avg_return'] >= candidate_row['avg_return']).sum())}`",
        f"- cases with profit_factor >= candidate: `{int((neighborhood['profit_factor'] >= candidate_row['profit_factor']).sum())}`",
        f"- robust-nearby cases: `{int(len(robust))}`",
        "",
        "## Top Local Variants",
    ]

    for _, row in top_avg.iterrows():
        lines.append(
            f"- `{row['label']}`: avg_return `{row['avg_return']:.2%}`, profit_factor `{row['profit_factor']:.2f}`, "
            f"real BUY / SELL `{int(row['real_buy_overlap'])}/{int(row['real_sell_overlap'])}`"
        )

    lines.extend(
        [
            "",
            "## Quant Judgment",
            "- If the neighborhood remains strong around the candidate, the branch is locally stable rather than dependent on a single knife-edge threshold.",
            "- If only one exact setting dominates while nearby cases collapse, the branch still carries threshold fragility risk.",
        ]
    )

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


if __name__ == "__main__":
    main()