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

from __future__ import annotations

from pathlib import Path

import pandas as pd

from dragon_strategy import DragonRuleEngine
from dragon_strategy_config import StrategyConfig


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 _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 _event_match(strategy_events: pd.DataFrame, workbook_events: pd.DataFrame, side: str) -> tuple[int, 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(wb - st), len(st - wb)


def _run_experiment(
    label: str,
    config: StrategyConfig,
    indicator_df: pd.DataFrame,
    workbook_events: pd.DataFrame,
    first_date: str,
    last_date: str,
) -> tuple[dict[str, object], pd.DataFrame]:
    engine = DragonRuleEngine(config=config)
    events, trades = engine.run(indicator_df)
    events = events[(events["date"] >= first_date) & (events["date"] <= last_date)].copy()
    trades = trades[
        (trades["buy_date"] >= first_date)
        & (trades["buy_date"] <= last_date)
        & (trades["sell_date"] >= first_date)
        & (trades["sell_date"] <= last_date)
    ].copy()

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

    deep_trades = trades[trades["buy_reason"].str.startswith("deep_oversold_rebound_buy")].copy()
    weak_trades = deep_trades[
        deep_trades["buy_reason"].isin(
            [
                "deep_oversold_rebound_buy:positive_b1_rebound",
                "deep_oversold_rebound_buy:shallow_false_start",
                "deep_oversold_rebound_buy:mixed_oversold",
            ]
        )
    ].copy()

    event_slice = events[
        (events["layer"] == "real_trade")
        & events["reason"].str.startswith("deep_oversold_rebound_buy")
    ].copy()
    event_slice["experiment"] = label

    row = {
        "experiment": label,
        "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"),
        "median_return": float(trades["return_pct"].median()) 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_missing": int(buy_missing),
        "real_buy_extra": int(buy_extra),
        "real_sell_overlap": int(sell_overlap),
        "real_sell_missing": int(sell_missing),
        "real_sell_extra": int(sell_extra),
        "deep_trade_count": int(len(deep_trades)),
        "deep_weak_trade_count": int(len(weak_trades)),
        "deep_avg_return": float(deep_trades["return_pct"].mean()) if not deep_trades.empty else float("nan"),
        "deep_weak_avg_return": float(weak_trades["return_pct"].mean()) if not weak_trades.empty else float("nan"),
    }
    return row, event_slice


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    indicator_df = _load_indicator_snapshot(base_dir)
    workbook_events = _load_true_trade_events(base_dir)
    first_date = workbook_events["date"].min()
    last_date = workbook_events["date"].max()

    baseline = StrategyConfig()
    experiments = [
        ("baseline", baseline),
        (
            "selective_veto_positive_c1_lt_15_3",
            baseline.with_updates(deep_oversold_selective_positive_b1_c1_max=15.3),
        ),
        (
            "selective_veto_shallow_jan_style",
            baseline.with_updates(
                deep_oversold_selective_shallow_c1_min=12.0,
                deep_oversold_selective_shallow_b1_min=-0.025,
            ),
        ),
        (
            "selective_veto_positive_and_shallow",
            baseline.with_updates(
                deep_oversold_selective_positive_b1_c1_max=15.3,
                deep_oversold_selective_shallow_c1_min=12.0,
                deep_oversold_selective_shallow_b1_min=-0.025,
            ),
        ),
        (
            "selective_veto_plus_mixed_c1_lt_10_2_no_ql",
            baseline.with_updates(
                deep_oversold_selective_positive_b1_c1_max=15.3,
                deep_oversold_selective_shallow_c1_min=12.0,
                deep_oversold_selective_shallow_b1_min=-0.025,
                deep_oversold_selective_mixed_c1_max=10.2,
                deep_oversold_selective_mixed_require_no_ql=True,
            ),
        ),
        (
            "block_all_remaining_weak_subtypes",
            baseline.with_updates(
                deep_oversold_block_positive_b1_rebound=True,
                deep_oversold_block_shallow_false_start_without_ql=True,
            ),
        ),
    ]

    rows: list[dict[str, object]] = []
    event_frames: list[pd.DataFrame] = []
    for label, config in experiments:
        row, event_slice = _run_experiment(label, config, indicator_df, workbook_events, first_date, last_date)
        rows.append(row)
        event_frames.append(event_slice)

    result_df = pd.DataFrame(rows)
    baseline_row = result_df[result_df["experiment"] == "baseline"].iloc[0]
    for col in [
        "trades",
        "win_rate",
        "avg_return",
        "median_return",
        "profit_factor",
        "real_buy_overlap",
        "real_sell_overlap",
        "deep_trade_count",
        "deep_weak_trade_count",
        "deep_avg_return",
        "deep_weak_avg_return",
    ]:
        result_df[f"delta_{col}"] = result_df[col] - baseline_row[col]

    event_df = pd.concat(event_frames, ignore_index=True)
    result_df.to_csv(base_dir / "dragon_deep_oversold_selective_veto_experiments.csv", index=False, encoding="utf-8-sig")
    event_df.to_csv(base_dir / "dragon_deep_oversold_selective_veto_event_changes.csv", index=False, encoding="utf-8-sig")

    lines = [
        "# Dragon Deep Oversold Selective Veto Experiments",
        "",
        "- Goal: test whether weak deep-oversold alpha can be improved by vetoing only the most pathological local patterns, instead of blocking whole subtypes.",
        "- Baseline objective is not preserved here; this is an alpha-first research pack.",
        "",
        "## Summary",
    ]
    for _, row in result_df.iterrows():
        lines.append(
            f"- `{row['experiment']}`: trades `{int(row['trades'])}`, avg_return `{row['avg_return']:.2%}`, "
            f"profit_factor `{row['profit_factor']:.2f}`, real BUY `{int(row['real_buy_overlap'])}`, real SELL `{int(row['real_sell_overlap'])}`, "
            f"deep weak trades `{int(row['deep_weak_trade_count'])}`"
        )

    lines.extend(["", "## Delta Vs Baseline"])
    for _, row in result_df[result_df["experiment"] != "baseline"].iterrows():
        lines.append(
            f"- `{row['experiment']}`: delta_avg_return `{row['delta_avg_return']:.2%}`, "
            f"delta_profit_factor `{row['delta_profit_factor']:.2f}`, "
            f"delta_deep_weak_avg_return `{row['delta_deep_weak_avg_return']:.2%}`, "
            f"real BUY `{int(row['real_buy_overlap'])}`, real SELL `{int(row['real_sell_overlap'])}`"
        )

    best = result_df[result_df["experiment"] != "baseline"].sort_values(
        ["avg_return", "profit_factor"], ascending=[False, False]
    ).head(1)
    if not best.empty:
        row = best.iloc[0]
        lines.extend(
            [
                "",
                "## Quant Judgment",
                f"- Best branch in this pack: `{row['experiment']}` with avg_return `{row['avg_return']:.2%}` and profit_factor `{row['profit_factor']:.2f}`.",
                "- Compare this result to `block_all_remaining_weak_subtypes` to see whether narrow veto meaningfully preserves useful edge while still removing obvious losers.",
                "- If a narrow veto matches most of the broad-block benefit with smaller date loss, it is the better alpha-first redesign candidate.",
            ]
        )

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


if __name__ == "__main__":
    main()