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

from __future__ import annotations

from pathlib import Path

import pandas as pd

from dragon_branch_configs import (
    alpha_first_glued_refined_hot_cap_config,
    alpha_first_selective_veto_config,
    workbook_preserving_config,
)
from dragon_shared import END_DATE, START_DATE, evaluation_years, profit_factor
from dragon_strategy import DragonRuleEngine


def _load_indicator_snapshot(base_dir: Path) -> pd.DataFrame:
    df = pd.read_csv(base_dir / "dragon_indicator_snapshot.csv", encoding="utf-8-sig")
    df["date"] = pd.to_datetime(df["date"])
    return df.set_index("date", drop=False)


def _holding_bucket(days: int) -> str:
    if days <= 5:
        return "00-05d"
    if days <= 10:
        return "06-10d"
    if days <= 20:
        return "11-20d"
    if days <= 40:
        return "21-40d"
    return "41d+"


def _run_branch(indicator_df: pd.DataFrame, config) -> pd.DataFrame:
    engine = DragonRuleEngine(config=config)
    _, trades = engine.run(indicator_df)
    trades = trades[
        (trades["buy_date"] >= START_DATE)
        & (trades["buy_date"] <= END_DATE)
        & (trades["sell_date"] >= START_DATE)
        & (trades["sell_date"] <= END_DATE)
    ].copy()
    trades["holding_bucket"] = trades["holding_days"].astype(int).map(_holding_bucket)
    return trades


def _apply_costs(trades: pd.DataFrame, per_side_bps: float) -> pd.DataFrame:
    out = trades.copy()
    cost = per_side_bps / 10000.0
    out["net_return_pct"] = (
        (out["sell_price"].astype(float) * (1.0 - cost)) / (out["buy_price"].astype(float) * (1.0 + cost))
    ) - 1.0
    return out


def _summarize(branch: str, per_side_bps: float, trades: pd.DataFrame) -> dict[str, object]:
    returns = trades["net_return_pct"].astype(float)
    compounded = float((1.0 + returns).prod() - 1.0) if not trades.empty else float("nan")
    years = evaluation_years(START_DATE, END_DATE)
    cagr = float((1.0 + compounded) ** (1.0 / years) - 1.0) if not trades.empty else float("nan")
    return {
        "branch": branch,
        "per_side_bps": per_side_bps,
        "trades": int(len(trades)),
        "win_rate": float((returns > 0).mean()) if not trades.empty else float("nan"),
        "avg_return": float(returns.mean()) if not trades.empty else float("nan"),
        "median_return": float(returns.median()) if not trades.empty else float("nan"),
        "profit_factor": profit_factor(returns) if not trades.empty else float("nan"),
        "compounded_return": compounded,
        "cagr": cagr,
        "short_00_05d_avg_return": float(trades[trades["holding_bucket"] == "00-05d"]["net_return_pct"].mean()),
        "short_06_10d_avg_return": float(trades[trades["holding_bucket"] == "06-10d"]["net_return_pct"].mean()),
    }


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

    branches = {
        "workbook_preserving": workbook_preserving_config(),
        "alpha_first_selective_veto": alpha_first_selective_veto_config(),
        "alpha_first_glued_refined_hot_cap": alpha_first_glued_refined_hot_cap_config(),
    }
    cost_levels = [0.0, 5.0, 10.0, 20.0]

    gross_trades = {name: _run_branch(indicator_df, cfg) for name, cfg in branches.items()}

    rows: list[dict[str, object]] = []
    for branch, trades in gross_trades.items():
        for bps in cost_levels:
            net_trades = _apply_costs(trades, bps)
            rows.append(_summarize(branch, bps, net_trades))

    result = pd.DataFrame(rows)
    result.to_csv(base_dir / "dragon_cost_stress_test.csv", index=False, encoding="utf-8-sig")

    lines = [
        "# Dragon Cost Stress Test",
        "",
        f"- Evaluation window: `{START_DATE}` to `{END_DATE}`.",
        "- Cost convention: symmetric per-side cost on entry and exit.",
        "",
        "## Summary",
    ]

    for branch in branches:
        subset = result[result["branch"] == branch].sort_values("per_side_bps")
        lines.append(f"### {branch}")
        for _, row in subset.iterrows():
            lines.append(
                f"- `{int(row['per_side_bps'])} bps/side`: CAGR `{row['cagr']:.2%}`, compounded `{row['compounded_return']:.2%}`, "
                f"avg_return `{row['avg_return']:.2%}`, PF `{row['profit_factor']:.2f}`, "
                f"`00-05d` `{row['short_00_05d_avg_return']:.2%}`, `06-10d` `{row['short_06_10d_avg_return']:.2%}`"
            )
        lines.append("")

    zero = result[result["per_side_bps"] == 0.0].set_index("branch")
    twenty = result[result["per_side_bps"] == 20.0].set_index("branch")
    lines.extend(
        [
            "## Quant Judgment",
            f"- At `20 bps/side`, current alpha branch CAGR = `{twenty.loc['alpha_first_selective_veto', 'cagr']:.2%}`.",
            f"- At `20 bps/side`, refined candidate CAGR = `{twenty.loc['alpha_first_glued_refined_hot_cap', 'cagr']:.2%}`.",
            f"- CAGR delta refined minus current alpha at `0 bps/side` = `{(zero.loc['alpha_first_glued_refined_hot_cap', 'cagr'] - zero.loc['alpha_first_selective_veto', 'cagr']):.2%}`.",
            f"- CAGR delta refined minus current alpha at `20 bps/side` = `{(twenty.loc['alpha_first_glued_refined_hot_cap', 'cagr'] - twenty.loc['alpha_first_selective_veto', 'cagr']):.2%}`.",
            "- If the refined branch remains ahead under cost pressure, its edge is not just a no-cost backtest artifact.",
        ]
    )

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


if __name__ == "__main__":
    main()