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

from __future__ import annotations

import json
from dataclasses import asdict
from pathlib import Path

import pandas as pd

from dragon_branch_configs import alpha_first_glued_refined_hot_cap_config
from dragon_indicators import DragonIndicatorConfig, DragonIndicatorEngine
from dragon_shared import END_DATE, START_DATE, evaluation_years, format_num as _format_num, format_pct as _format_pct, profit_factor
from dragon_strategy import DragonRuleEngine
RC_VERSION = "RC1"
RC_BRANCH = "alpha_first_glued_refined_hot_cap"


def _max_drawdown(returns: pd.Series) -> tuple[float, int]:
    equity = (1.0 + returns.astype(float)).cumprod()
    peak = equity.cummax()
    dd = equity / peak - 1.0
    max_dd = float(dd.min()) if not dd.empty else float("nan")

    duration = 0
    max_duration = 0
    for value in dd:
        if value < 0:
            duration += 1
            max_duration = max(max_duration, duration)
        else:
            duration = 0
    return max_dd, max_duration


def _load_indicator_snapshot(base_dir: Path) -> pd.DataFrame:
    path = base_dir / "dragon_indicator_snapshot.csv"
    if path.exists():
        df = pd.read_csv(path, encoding="utf-8-sig")
        df["date"] = pd.to_datetime(df["date"])
        return df.sort_values("date").reset_index(drop=True)

    engine = DragonIndicatorEngine(DragonIndicatorConfig(start_date="2015-01-01", end_date="2026-01-31"))
    df = engine.compute(engine.fetch_daily_data()).reset_index(drop=False).rename(columns={"index": "date"})
    df.to_csv(path, index=False, encoding="utf-8-sig")
    return df


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    indicators = _load_indicator_snapshot(base_dir)
    indexed = indicators.set_index("date", drop=False)

    config = alpha_first_glued_refined_hot_cap_config()
    engine = DragonRuleEngine(config=config)
    events, trades = engine.run(indexed)
    trades = trades[
        (trades["buy_date"] >= START_DATE)
        & (trades["buy_date"] <= END_DATE)
        & (trades["sell_date"] >= START_DATE)
        & (trades["sell_date"] <= END_DATE)
    ].copy()
    returns = trades["return_pct"].astype(float)
    compounded = float((1.0 + returns).prod() - 1.0)
    years = evaluation_years(START_DATE, END_DATE)
    cagr = float((1.0 + compounded) ** (1.0 / years) - 1.0)
    max_dd, dd_duration = _max_drawdown(returns)

    snapshot = {
        "release_version": RC_VERSION,
        "branch_name": RC_BRANCH,
        "evaluation_window": {"start": START_DATE, "end": END_DATE},
        "trade_count": int(len(trades)),
        "win_rate": float((returns > 0).mean()),
        "avg_return": float(returns.mean()),
        "median_return": float(returns.median()),
        "profit_factor": profit_factor(returns),
        "compounded_return": compounded,
        "cagr": cagr,
        "max_drawdown": max_dd,
        "drawdown_duration_trades": dd_duration,
        "config": {
            **asdict(config),
            "disabled_rules": sorted(config.disabled_rules),
        },
    }

    (base_dir / "dragon_rc1_config_snapshot.json").write_text(
        json.dumps(snapshot, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )

    lines = [
        "# Dragon RC1 Release",
        "",
        f"- Release version: `{RC_VERSION}`",
        f"- Strategy branch: `{RC_BRANCH}`",
        f"- Freeze date: `{pd.Timestamp.now().date().isoformat()}`",
        f"- Evaluation window: `{START_DATE}` to `{END_DATE}`",
        "",
        "## Freeze Intent",
        "- `RC1` is the first frozen release candidate of the refined alpha branch.",
        "- Its purpose is not to continue blind optimization, but to serve as the tracked strategy candidate for daily signal production and monitoring.",
        "",
        "## Frozen Headline",
        f"- trades: `{int(len(trades))}`",
        f"- win_rate: `{_format_pct(float((returns > 0).mean()))}`",
        f"- avg_return: `{_format_pct(float(returns.mean()))}`",
        f"- median_return: `{_format_pct(float(returns.median()))}`",
        f"- profit_factor: `{_format_num(profit_factor(returns))}`",
        f"- compounded_return: `{_format_pct(compounded)}`",
        f"- CAGR: `{_format_pct(cagr)}`",
        f"- max_drawdown: `{_format_pct(max_dd)}`",
        f"- drawdown_duration_trades: `{dd_duration}`",
        "",
        "## Operating Rule",
        "- Treat `RC1` as the forward default branch.",
        "- Do not change frozen core parameters inside `RC1` directly.",
        "- Any future core-threshold change must create a new named branch and go through attribution plus execution-aware stress again.",
        "",
        "## Artifacts",
        "- `dragon_rc1_config_snapshot.json`",
        "- `dragon_formal_strategy_governance.md`",
        "- `dragon_parameter_governance.md`",
        "- `dragon_formal_strategy_memo.md`",
    ]
    (base_dir / "dragon_rc1_release.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()