cyb50-quant/research/dragon/v2/dragon_backtest.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_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 _event_match_report(workbook_events: pd.DataFrame, strategy_events: pd.DataFrame, side: str, layer: str) -> dict[str, object]:
    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 {
        "side": side,
        "layer": layer,
        "workbook": len(wb),
        "strategy": len(st),
        "overlap": len(hit),
        "missing_from_strategy": sorted(wb - st),
        "extra_in_strategy": sorted(st - wb),
    }


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    workbook_path = _find_workbook(base_dir)
    workbook = DragonWorkbook(workbook_path)
    workbook_events = pd.DataFrame(
        [
            {
                "date": event.date.isoformat(),
                "side": event.side,
                "layer": event.layer,
                "signal_reason": event.signal_reason,
                "note": event.note,
            }
            for event in workbook.split_layers()
        ]
    )

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

    strategy = DragonRuleEngine()
    strategy_events, strategy_trades = strategy.run(indicator_df)
    first_workbook_date = pd.to_datetime(workbook_events["date"]).min().date().isoformat()
    last_workbook_date = pd.to_datetime(workbook_events["date"]).max().date().isoformat()
    strategy_events = strategy_events[
        (strategy_events["date"] >= first_workbook_date) & (strategy_events["date"] <= last_workbook_date)
    ].copy()
    strategy_trades = strategy_trades[
        (strategy_trades["buy_date"] >= first_workbook_date)
        & (strategy_trades["buy_date"] <= last_workbook_date)
        & (strategy_trades["sell_date"] >= first_workbook_date)
        & (strategy_trades["sell_date"] <= last_workbook_date)
    ].copy()
    strategy_events.to_csv(base_dir / "dragon_strategy_events.csv", index=False, encoding="utf-8-sig")
    strategy_trades.to_csv(base_dir / "dragon_strategy_trades.csv", index=False, encoding="utf-8-sig")

    comparisons = [
        _event_match_report(workbook_events, strategy_events, side="BUY", layer="real_trade"),
        _event_match_report(workbook_events, strategy_events, side="SELL", layer="real_trade"),
        _event_match_report(workbook_events, strategy_events, side="BUY", layer="aux_signal"),
        _event_match_report(workbook_events, strategy_events, side="SELL", layer="aux_signal"),
    ]

    lines = [
        "# Dragon Baseline Backtest",
        "",
        f"- Source workbook: `{workbook_path.name}`",
        f"- Strategy events: `{len(strategy_events)}`",
        f"- Strategy trades: `{len(strategy_trades)}`",
        "",
        "## Event Fit",
    ]
    for item in comparisons:
        lines.append(
            f"- {item['layer']} {item['side']}: workbook `{item['workbook']}`, strategy `{item['strategy']}`, overlap `{item['overlap']}`"
        )
        lines.append(f"- missing_from_strategy: `{item['missing_from_strategy'][:20]}`")
        lines.append(f"- extra_in_strategy: `{item['extra_in_strategy'][:20]}`")

    if not strategy_trades.empty:
        wins = (strategy_trades["return_pct"] > 0).sum()
        avg_ret = strategy_trades["return_pct"].mean()
        med_ret = strategy_trades["return_pct"].median()
        lines.extend(
            [
                "",
                "## Strategy Trade Stats",
                f"- trades: `{len(strategy_trades)}`",
                f"- win_rate: `{wins}/{len(strategy_trades)} = {wins / len(strategy_trades):.2%}`",
                f"- avg_return: `{avg_ret:.2%}`",
                f"- median_return: `{med_ret:.2%}`",
            ]
        )

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


if __name__ == "__main__":
    main()