cyb50-quant/research/dragon/v2/dragon_deep_oversold_confirmation_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"])
    df = df.set_index("date", drop=False)
    return df


def _load_workbook_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:confirmed_positive_b1_rebound",
                "deep_oversold_rebound_buy:confirmed_shallow_false_start",
            ]
        )
    ].copy()
    changed_events = events[
        (events["layer"] == "real_trade")
        & (
            events["reason"].str.startswith("deep_oversold_rebound_buy")
            | events["reason"].eq("predictive_error_reentry_buy")
        )
    ].copy()
    changed_events["experiment"] = label

    row = {
        "experiment": label,
        "confirm_window_bars": config.deep_oversold_confirm_window_bars,
        "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_confirmed_trade_count": int(
            deep_trades["buy_reason"].str.contains(":confirmed_", regex=False).sum()
        ),
        "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, changed_events


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

    baseline_config = StrategyConfig()
    experiments = [
        ("baseline", baseline_config),
        (
            "confirm_weak_with_ql_1bar",
            baseline_config.with_updates(
                deep_oversold_confirm_weak_with_ql=True,
                deep_oversold_confirm_window_bars=1,
            ),
        ),
        (
            "confirm_weak_with_ql_2bar",
            baseline_config.with_updates(
                deep_oversold_confirm_weak_with_ql=True,
                deep_oversold_confirm_window_bars=2,
            ),
        ),
        (
            "confirm_weak_with_ql_3bar",
            baseline_config.with_updates(
                deep_oversold_confirm_weak_with_ql=True,
                deep_oversold_confirm_window_bars=3,
            ),
        ),
    ]

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

    result_df = pd.DataFrame(rows)
    baseline = 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_confirmed_trade_count",
        "deep_avg_return",
        "deep_weak_avg_return",
    ]:
        result_df[f"delta_{col}"] = result_df[col] - baseline[col]

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

    lines = [
        "# Dragon Deep Oversold Confirmation Experiments",
        "",
        "- Goal: test an alpha-first branch for weak deep-oversold subtypes by replacing immediate entry with QL confirmation inside a short waiting window.",
        "- Scope: only `positive_b1_rebound` and `shallow_false_start`; default baseline behavior remains unchanged.",
        "",
        "## 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'])}`, confirmed deep trades `{int(row['deep_confirmed_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 alpha-first branch in this pack: `{row['experiment']}` with avg_return `{row['avg_return']:.2%}` and profit_factor `{row['profit_factor']:.2f}`.",
                "- This pack is expected to lose workbook date alignment because entries are delayed by confirmation; treat it as a research branch, not a baseline patch.",
                "- If the best branch improves weak-subtype trade quality materially, the next step should be to audit shifted dates trade-by-trade rather than porting it directly into the baseline.",
            ]
        )

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


if __name__ == "__main__":
    main()