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

from __future__ import annotations

from pathlib import Path

import pandas as pd

from dragon_indicators import DragonIndicatorConfig, DragonIndicatorEngine
from dragon_strategy import DragonRuleEngine
from dragon_strategy_config import StrategyConfig
from dragon_workbook import DragonWorkbook


def _find_workbook(base_dir: Path) -> Path:
    matches = sorted(base_dir.glob("*.xlsx"))
    if not matches:
        raise FileNotFoundError(f"No workbook found in {base_dir}")
    return matches[0]


def _load_workbook_events(workbook_path: Path) -> pd.DataFrame:
    workbook = DragonWorkbook(workbook_path)
    return pd.DataFrame(
        [{"date": e.date.isoformat(), "side": e.side, "layer": e.layer} for e in workbook.split_layers()]
    )


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


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 _run(label: str, config: StrategyConfig, workbook_events: pd.DataFrame, indicator_df: pd.DataFrame, first_date: str, last_date: str) -> dict[str, object]:
    events, trades = DragonRuleEngine(config=config).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 = _overlap(workbook_events, events, "BUY", "real_trade")
    sell_overlap, sell_missing, sell_extra = _overlap(workbook_events, events, "SELL", "real_trade")

    deep_trades = trades[trades["buy_reason"].astype(str).str.startswith("deep_oversold_rebound_buy")]

    return {
        "experiment": label,
        "trades": int(len(trades)),
        "avg_return": float(trades["return_pct"].mean()),
        "profit_factor": _profit_factor(trades["return_pct"]),
        "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_oversold_trade_count": int(len(deep_trades)),
        "deep_oversold_avg_return": float(deep_trades["return_pct"].mean()) if not deep_trades.empty else float("nan"),
    }


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

    engine = DragonIndicatorEngine(DragonIndicatorConfig(start_date="2015-01-01", end_date="2026-01-31"))
    indicator_df = engine.compute(engine.fetch_daily_data())

    baseline = StrategyConfig()
    experiments = [
        ("baseline", baseline),
        (
            "block_positive_b1_rebound",
            baseline.with_updates(deep_oversold_block_positive_b1_rebound=True),
        ),
        (
            "block_shallow_false_start_without_ql",
            baseline.with_updates(deep_oversold_block_shallow_false_start_without_ql=True),
        ),
        (
            "block_both_remaining_weak_subtypes",
            baseline.with_updates(
                deep_oversold_block_positive_b1_rebound=True,
                deep_oversold_block_shallow_false_start_without_ql=True,
            ),
        ),
    ]

    rows = [_run(label, cfg, workbook_events, indicator_df, first_date, last_date) for label, cfg in experiments]
    df = pd.DataFrame(rows)
    base_row = df[df["experiment"] == "baseline"].iloc[0]
    for col in ["trades", "avg_return", "profit_factor", "real_buy_overlap", "real_sell_overlap", "deep_oversold_trade_count", "deep_oversold_avg_return"]:
        df[f"delta_{col}"] = df[col] - base_row[col]

    df.to_csv(base_dir / "dragon_deep_oversold_experiments.csv", index=False, encoding="utf-8-sig")

    lines = [
        "# Dragon Deep Oversold Experiments",
        "",
        f"- Baseline deep-oversold trade count: `{int(base_row['deep_oversold_trade_count'])}`",
        f"- Baseline real BUY / SELL overlap: `{int(base_row['real_buy_overlap'])}` / `{int(base_row['real_sell_overlap'])}`",
        "",
        "## Experiment Summary",
    ]
    for _, row in df.iterrows():
        lines.append(
            f"- `{row['experiment']}`: deep trades `{int(row['deep_oversold_trade_count'])}`, "
            f"delta_avg_return `{row['delta_avg_return']:.2%}`, real BUY `{int(row['real_buy_overlap'])}`, real SELL `{int(row['real_sell_overlap'])}`"
        )

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


if __name__ == "__main__":
    main()