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

from __future__ import annotations

import json
from dataclasses import asdict
from pathlib import Path

import pandas as pd

from dragon_strategy_config import StrategyConfig


def _load_csv(base_dir: Path, name: str) -> pd.DataFrame:
    return pd.read_csv(base_dir / name, 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 _format_pct(value: float) -> str:
    if pd.isna(value):
        return "NA"
    if value == float("inf"):
        return "inf"
    return f"{value:.2%}"


def _format_num(value: float) -> str:
    if pd.isna(value):
        return "NA"
    if value == float("inf"):
        return "inf"
    return f"{value:.2f}"


def _baseline_snapshot(config: StrategyConfig) -> dict[str, object]:
    snapshot = asdict(config)
    snapshot["disabled_rules"] = sorted(config.disabled_rules)
    return snapshot


def main() -> None:
    base_dir = Path(__file__).resolve().parent

    trades = _load_csv(base_dir, "dragon_strategy_trades.csv")
    fit = (base_dir / "dragon_strategy_fit.md").read_text(encoding="utf-8")
    entry_contrib = _load_csv(base_dir, "dragon_rule_contribution_entry.csv")
    ablation = _load_csv(base_dir, "dragon_rule_ablation.csv")
    sensitivity = _load_csv(base_dir, "dragon_threshold_sensitivity_summary.csv")
    walk_forward = _load_csv(base_dir, "dragon_walk_forward_summary.csv")
    family_stability = _load_csv(base_dir, "dragon_walk_forward_family_stability.csv")

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

    baseline_ablation = ablation[ablation["experiment"] == "baseline"].iloc[0]
    returns = trades["return_pct"].astype(float)
    overall_profit_factor = _profit_factor(returns)

    core_alpha_names = ["glued_buy", "early_crash_probe_buy", "oversold_recovery_buy"]
    structural_support_names = ["dual_gold_resonance_buy", "deep_oversold_rebound_buy:classic_oversold"]
    active_research_names = [
        "deep_oversold_rebound_buy:positive_b1_rebound",
        "deep_oversold_rebound_buy:shallow_false_start",
        "deep_oversold_rebound_buy:mixed_oversold",
        "deep_oversold_rebound_buy:deep_capitulation",
        "post_washout_kdj_reentry_buy",
        "oversold_reversal_after_ql_buy",
        "post_sell_rebound_buy",
    ]

    core_alpha = entry_contrib[entry_contrib["buy_reason"].isin(core_alpha_names)].copy()
    structural_support = entry_contrib[entry_contrib["buy_reason"].isin(structural_support_names)].copy()
    weak_research = entry_contrib[entry_contrib["buy_reason"].isin(active_research_names)].copy()
    weak_research = weak_research.sort_values(["avg_return", "trades"], ascending=[True, False])

    fragile = sensitivity[sensitivity["stable_real_alignment"] == False].sort_values("avg_return_range", ascending=False)
    robust = sensitivity[sensitivity["stable_real_alignment"] == True].sort_values("avg_return_range")

    anchored = walk_forward[walk_forward["scheme"] == "anchored_expanding"].copy()
    rolling = walk_forward[walk_forward["scheme"] == "rolling_3y"].copy()
    anchored_positive = int((anchored["test_avg_return"] > 0).sum()) if not anchored.empty else 0
    anchored_total = int(len(anchored))
    rolling_positive = int((rolling["test_avg_return"] > 0).sum()) if not rolling.empty else 0
    rolling_total = int(len(rolling))

    stable_families = family_stability[
        (family_stability["avg_yearly_avg_return"] > 0)
        & (family_stability["positive_years"] >= family_stability["negative_years"])
    ].sort_values(["avg_yearly_avg_return", "total_trades"], ascending=[False, False])
    unstable_families = family_stability[
        (family_stability["avg_yearly_avg_return"] < 0)
        | (family_stability["negative_years"] > family_stability["positive_years"])
    ].sort_values(["avg_yearly_avg_return", "min_yearly_avg_return"])

    lines = [
        "# Dragon Formal Research Baseline",
        "",
        "## Scope",
        "- Universe: `399673` only.",
        "- Objective: preserve workbook real-trade alignment while upgrading the strategy into a researchable, testable, parameter-aware baseline.",
        "- Current baseline type: `workbook-preserving baseline`.",
        "",
        "## Locked Baseline Metrics",
        f"- real BUY overlap: `{int(baseline_ablation['real_buy_overlap'])}/106`",
        f"- real SELL overlap: `{int(baseline_ablation['real_sell_overlap'])}/105`",
        f"- aux BUY overlap: `{int(baseline_ablation['aux_buy_overlap'])}/1`",
        f"- aux SELL overlap: `{int(baseline_ablation['aux_sell_overlap'])}/21`",
        f"- strategy trades: `{int(baseline_ablation['trades'])}`",
        f"- win_rate: `{_format_pct(float(baseline_ablation['win_rate']))}`",
        f"- avg_return: `{_format_pct(float(baseline_ablation['avg_return']))}`",
        f"- median_return: `{_format_pct(float(baseline_ablation['median_return']))}`",
        f"- profit_factor: `{_format_num(overall_profit_factor)}`",
        "",
        "## Baseline Config Snapshot",
        "- Snapshot file: `dragon_baseline_config_snapshot.json`.",
        "- Rule switches default to the current aligned strategy baseline; any future research branch should fork from this snapshot rather than editing against memory.",
        "",
        "## Core Alpha Families",
    ]
    for _, row in core_alpha.iterrows():
        lines.append(
            f"- `{row['buy_reason']}`: trades `{int(row['trades'])}`, avg_return `{_format_pct(float(row['avg_return']))}`, "
            f"win_rate `{_format_pct(float(row['win_rate']))}`"
        )

    lines.extend(["", "## Structural Support Families"])
    for _, row in structural_support.iterrows():
        lines.append(
            f"- `{row['buy_reason']}`: trades `{int(row['trades'])}`, avg_return `{_format_pct(float(row['avg_return']))}`, "
            f"win_rate `{_format_pct(float(row['win_rate']))}`"
        )

    lines.extend(
        [
            "",
            "## Frozen Bridge Rules",
            "- `predictive_b1_break_exit`: bridge-style split-chain exit; loosening worsens results, tightening breaks workbook alignment.",
            "- `predictive_error_reentry_buy`: part of the same bridge chain; should be evaluated together with the predictive-break exit, not as an isolated entry.",
            "- Any internal hold gates added only to preserve workbook-aligned split paths should remain frozen unless the objective explicitly changes away from workbook preservation.",
            "",
            "## Redundant Or Label-Only Families",
            "- `non_glued_positive_expansion_buy`: now absorbed by `dual_gold_resonance_buy` on the same in-sample dates; treat as redundant label, not independent alpha.",
            "- Auxiliary same-side post-exit sell compression: keep as hygiene logic, not as a primary optimization frontier.",
            "",
            "## Active Research Families",
        ]
    )
    for _, row in weak_research.iterrows():
        lines.append(
            f"- `{row['buy_reason']}`: trades `{int(row['trades'])}`, avg_return `{_format_pct(float(row['avg_return']))}`, "
            f"win_rate `{_format_pct(float(row['win_rate']))}`"
        )

    lines.extend(
        [
            "",
            "## Threshold Classification",
            "- Fragile parameters: change them only inside explicit experiment branches and always rerun full alignment diagnostics.",
        ]
    )
    for _, row in fragile.iterrows():
        lines.append(
            f"- `{row['parameter']}`: avg_return_range `{_format_pct(float(row['avg_return_range']))}`, "
            f"min real BUY `{int(row['real_buy_overlap_min'])}`, min real SELL `{int(row['real_sell_overlap_min'])}`"
        )

    lines.append("- Relatively robust parameters: acceptable first candidates for future controlled sweeps.")
    for _, row in robust.head(4).iterrows():
        lines.append(
            f"- `{row['parameter']}`: avg_return_range `{_format_pct(float(row['avg_return_range']))}`, "
            f"profit_factor_range `{_format_num(float(row['profit_factor_range']))}`"
        )

    lines.extend(
        [
            "",
            "## Temporal Stability",
            f"- Anchored expanding windows: positive out-of-sample years `{anchored_positive}/{anchored_total}`.",
            f"- Rolling 3Y windows: positive out-of-sample years `{rolling_positive}/{rolling_total}`.",
            "- This validation holds the strategy fixed; it is a time-stability audit, not a refit-based optimizer.",
            "- Strong family persistence candidates:",
        ]
    )
    for _, row in stable_families.head(5).iterrows():
        lines.append(
            f"- `{row['entry_family']}`: years_active `{int(row['years_active'])}`, positive_years `{int(row['positive_years'])}`, "
            f"negative_years `{int(row['negative_years'])}`, avg_yearly_avg_return `{_format_pct(float(row['avg_yearly_avg_return']))}`"
        )

    lines.append("- Weak family persistence candidates:")
    for _, row in unstable_families.head(5).iterrows():
        lines.append(
            f"- `{row['entry_family']}`: years_active `{int(row['years_active'])}`, positive_years `{int(row['positive_years'])}`, "
            f"negative_years `{int(row['negative_years'])}`, avg_yearly_avg_return `{_format_pct(float(row['avg_yearly_avg_return']))}`"
        )

    lines.extend(
        [
            "",
            "## Operating Rules For Future Research",
            "- Do not trade off `106/106` and `105/105` alignment silently. Any alignment loss must be treated as a branch with an explicit objective change.",
            "- Do not blind-tune predictive-break thresholds. That family is frozen under the current baseline objective.",
            "- Do not optimize the auxiliary layer first. The main leverage is now in weak entry-family redesign and short-holding loss control.",
            "- New ideas should first be tested as local attribution experiments, then full-sample reruns, then temporal-stability checks.",
            "",
            "## Next Research Track",
            "- Track A: redesign remaining `deep_oversold_rebound_buy` weak subtypes with delayed confirmation or fallback routing, not blunt deletion.",
            "- Track B: explicitly target short holding buckets `00-05d` and `06-10d`, which remain the main quality drag.",
            "- Track C: separate a future `alpha-first` research branch from this workbook-preserving baseline if the goal later changes from reconstruction to pure performance.",
        ]
    )

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

    # Keep the baseline fit markdown referenced by this script as an explicit dependency.
    if "real_trade BUY: workbook `106`" not in fit:
        raise RuntimeError("Unexpected baseline fit file contents; baseline report expects the aligned workbook-preserving version.")


if __name__ == "__main__":
    main()