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

from __future__ import annotations

import hashlib
import json
from dataclasses import asdict
from datetime import datetime
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
from dragon_strategy import DragonRuleEngine


RC_VERSION = "RC1"
RC_BRANCH = "alpha_first_glued_refined_hot_cap"
TRADES_OUTPUT = "dragon_rc1_golden_trades.csv"
EVENTS_OUTPUT = "dragon_rc1_golden_events.csv"
MANIFEST_OUTPUT = "dragon_rc1_golden_manifest.json"
EVENTS_CORE_COLUMNS = [
    "date",
    "side",
    "layer",
    "reason",
    "close",
    "a1",
    "b1",
    "c1",
    "kdj_buy",
    "kdj_sell",
    "ql_buy",
    "ql_sell",
]
TRADES_CORE_COLUMNS = [
    "buy_date",
    "buy_price",
    "buy_reason",
    "sell_date",
    "sell_price",
    "sell_reason",
    "holding_days",
    "return_pct",
]


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as fp:
        for chunk in iter(lambda: fp.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _df_sha256(df: pd.DataFrame) -> str:
    digest = hashlib.sha256()
    payload = df.to_csv(index=False, encoding="utf-8", float_format="%.15g")
    digest.update(payload.encode("utf-8"))
    return digest.hexdigest()


def _load_indicator_snapshot(base_dir: Path) -> tuple[pd.DataFrame, str]:
    full_snapshot = base_dir / "dragon_indicator_snapshot_full.csv"
    if full_snapshot.exists():
        df = pd.read_csv(full_snapshot, encoding="utf-8-sig")
        source = full_snapshot.name
    else:
        fallback_snapshot = base_dir / "dragon_indicator_snapshot.csv"
        if fallback_snapshot.exists():
            df = pd.read_csv(fallback_snapshot, encoding="utf-8-sig")
            source = fallback_snapshot.name
        else:
            engine = DragonIndicatorEngine(DragonIndicatorConfig(start_date="2015-01-01", end_date=None))
            df = (
                engine.compute(engine.fetch_daily_data())
                .reset_index(drop=False)
                .rename(columns={"index": "date"})
            )
            source = "fetched_live_history"
    df["date"] = pd.to_datetime(df["date"])
    indexed = df.sort_values("date").set_index("date")
    return indexed, source


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    indexed, source = _load_indicator_snapshot(base_dir)

    config = alpha_first_glued_refined_hot_cap_config()
    engine = DragonRuleEngine(config=config)
    events, trades = engine.run(indexed)

    events = events[
        (events["date"] >= START_DATE)
        & (events["date"] <= END_DATE)
    ].copy()
    trades = trades[
        (trades["buy_date"] >= START_DATE)
        & (trades["buy_date"] <= END_DATE)
        & (trades["sell_date"] >= START_DATE)
        & (trades["sell_date"] <= END_DATE)
    ].copy()

    events.insert(0, "branch", RC_BRANCH)
    trades.insert(0, "branch", RC_BRANCH)

    events.sort_values(["date", "side", "layer", "reason"], inplace=True)
    trades.sort_values(["buy_date", "sell_date", "buy_reason", "sell_reason"], inplace=True)

    events_path = base_dir / EVENTS_OUTPUT
    trades_path = base_dir / TRADES_OUTPUT
    manifest_path = base_dir / MANIFEST_OUTPUT

    events.to_csv(events_path, index=False, encoding="utf-8-sig")
    trades.to_csv(trades_path, index=False, encoding="utf-8-sig")

    returns = trades["return_pct"].astype(float) if not trades.empty else pd.Series(dtype=float)
    summary = {
        "trade_count": int(len(trades)),
        "event_count": int(len(events)),
        "win_rate": float((returns > 0).mean()) if not returns.empty else float("nan"),
        "avg_return": float(returns.mean()) if not returns.empty else float("nan"),
        "median_return": float(returns.median()) if not returns.empty else float("nan"),
    }

    manifest = {
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "release_version": RC_VERSION,
        "branch": RC_BRANCH,
        "evaluation_window": {"start": START_DATE, "end": END_DATE},
        "indicator_source": source,
        "artifacts": {
            "trades": {
                "path": TRADES_OUTPUT,
                "sha256": _sha256(trades_path),
                "columns": list(trades.columns),
                "core_columns": TRADES_CORE_COLUMNS,
                "core_sha256": _df_sha256(trades[TRADES_CORE_COLUMNS]),
            },
            "events": {
                "path": EVENTS_OUTPUT,
                "sha256": _sha256(events_path),
                "columns": list(events.columns),
                "core_columns": EVENTS_CORE_COLUMNS,
                "core_sha256": _df_sha256(events[EVENTS_CORE_COLUMNS]),
            },
        },
        "summary": summary,
        "config_snapshot": {
            **asdict(config),
            "disabled_rules": sorted(config.disabled_rules),
        },
    }

    manifest_path.write_text(
        json.dumps(manifest, indent=2, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()