okx-codex-trader/scripts/validate_eth_bb_squeeze_t_gate_robustness.py

from __future__ import annotations

import argparse
import sys
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from scripts.search_eth_bb_squeeze_t_gates import (
    BAR,
    BTC_SYMBOL,
    COSTS,
    ETH_SYMBOL,
    OUTPUT_DIR,
    PRIMARY_COST,
    Variant,
    _align_pair,
    _format_ts,
    _load_candles,
    cost_equity_frame,
    equity_metrics,
    markdown_table,
    run_variant,
    worst_month,
)


HORIZONS = (
    ("10y", 10.0 * 365),
    ("180d", 180.0),
    ("90d", 90.0),
)


def build_candidates() -> list[Variant]:
    return [
        Variant(96, 960, 0.25, 0.01, 0.035, "both", "btc-up", 0.006, 0.25, 24, 48, "btc_against", 0.0, 1, 0.006, 0.25, 0.008),
        Variant(96, 960, 0.25, 0.01, 0.035, "both", "btc-up", 0.006, 0.25, 24, 96, "btc_against", 0.0, 1, 0.006, 0.25, 0.008),
        Variant(96, 960, 0.25, 0.01, 0.035, "both", "btc-up", 0.006, 0.25, 24, 192, "btc_against", 0.0, 1, 0.006, 0.25, 0.008),
        Variant(96, 960, 0.25, 0.01, 0.035, "both", "btc-up", 0.006, 0.25, 24, 48, "btc_against", 0.001, 1, 0.006, 0.25, 0.008),
    ]


def slice_days(eth: list, btc: list, days: float) -> tuple[list, list]:
    cutoff = eth[-1].ts - int(days * 86_400_000)
    start = next(index for index, candle in enumerate(eth) if candle.ts >= cutoff)
    return eth[start:], btc[start:]


def evaluate_candidates(eth: list, btc: list) -> pd.DataFrame:
    primary_cost = dict(COSTS)[PRIMARY_COST]
    rows: list[dict[str, object]] = []
    for variant in build_candidates():
        for horizon, days in HORIZONS:
            horizon_eth, horizon_btc = slice_days(eth, btc, days)
            result, gate_stats = run_variant(horizon_eth, horizon_btc, variant)
            if not result.equity_curve:
                continue
            frame = cost_equity_frame(result, primary_cost)
            metrics = equity_metrics(frame, horizon_eth[0].ts, horizon_eth[-1].ts)
            month, month_return = worst_month(frame)
            rows.append(
                {
                    "horizon": horizon,
                    "name": variant.name,
                    "reentry_bars": variant.reentry_bars,
                    "middle_exit_buffer_pct": variant.middle_exit_buffer_pct,
                    "middle_exit_confirm_bars": variant.middle_exit_confirm_bars,
                    "first_candle": _format_ts(horizon_eth[0].ts),
                    "last_candle": _format_ts(horizon_eth[-1].ts),
                    "trades": result.trade_count,
                    "reentry_entries": gate_stats["reentry_entries"],
                    "net_total_return": metrics["net_total_return"],
                    "net_annualized_return": metrics["net_annualized_return"],
                    "net_max_drawdown": metrics["net_max_drawdown"],
                    "net_calmar": metrics["net_calmar"],
                    "worst_month": month,
                    "worst_month_return": month_return,
                }
            )
    return pd.DataFrame(rows)


def candidate_score(frame: pd.DataFrame) -> pd.DataFrame:
    pivot = frame.pivot(index="name", columns="horizon", values=["net_total_return", "net_max_drawdown", "net_calmar", "trades"])
    rows: list[dict[str, object]] = []
    for name in pivot.index:
        row = {"name": name}
        source = frame[frame["name"] == name].iloc[0]
        row["reentry_bars"] = int(source["reentry_bars"])
        row["middle_exit_buffer_pct"] = float(source["middle_exit_buffer_pct"])
        row["middle_exit_confirm_bars"] = int(source["middle_exit_confirm_bars"])
        for horizon, _days in HORIZONS:
            row[f"{horizon}_return"] = float(pivot.loc[name, ("net_total_return", horizon)])
            row[f"{horizon}_mdd"] = float(pivot.loc[name, ("net_max_drawdown", horizon)])
            row[f"{horizon}_calmar"] = float(pivot.loc[name, ("net_calmar", horizon)])
            row[f"{horizon}_trades"] = int(pivot.loc[name, ("trades", horizon)])
        row["positive_recent_windows"] = int(row["180d_return"] > 0.0) + int(row["90d_return"] > 0.0)
        row["min_recent_return"] = min(row["180d_return"], row["90d_return"])
        row["max_recent_mdd"] = max(row["180d_mdd"], row["90d_mdd"])
        rows.append(row)
    return pd.DataFrame(rows).sort_values(
        ["positive_recent_windows", "min_recent_return", "max_recent_mdd", "10y_calmar"],
        ascending=[False, False, True, False],
    )


def write_report(*, detail: pd.DataFrame, score: pd.DataFrame, output_dir: Path) -> str:
    best = score.iloc[0]
    recent = detail[detail["horizon"].isin(["180d", "90d"])].copy()
    recent = recent.sort_values(["horizon", "net_total_return"], ascending=[True, False])
    lines = [
        "# ETH BB squeeze T-gated robustness validation",
        "",
        "Scope: adjacent variants around the current best T-gated BB squeeze candidate. No live executor changes.",
        f"Cost model: `{PRIMARY_COST}`.",
        "",
        "Output files:",
        f"- `{output_dir / 'eth-bb-squeeze-t-gate-robustness-detail.csv'}`",
        f"- `{output_dir / 'eth-bb-squeeze-t-gate-robustness-score.csv'}`",
        f"- `{output_dir / 'eth-bb-squeeze-t-gate-robustness-report.md'}`",
        "",
        "Candidate score:",
        markdown_table(
            score[
                [
                    "reentry_bars",
                    "middle_exit_buffer_pct",
                    "middle_exit_confirm_bars",
                    "10y_return",
                    "10y_calmar",
                    "180d_return",
                    "180d_mdd",
                    "90d_return",
                    "90d_mdd",
                    "positive_recent_windows",
                ]
            ]
        ),
        "",
        "Recent horizon detail:",
        markdown_table(
            recent[
                [
                    "horizon",
                    "reentry_bars",
                    "middle_exit_buffer_pct",
                    "trades",
                    "reentry_entries",
                    "net_total_return",
                    "net_max_drawdown",
                    "net_calmar",
                    "worst_month",
                    "worst_month_return",
                ]
            ]
        ),
        "",
        "Verdict:",
        (
            f"- Most stable adjacent candidate: reentry_bars={int(best['reentry_bars'])}, "
            f"middle_exit_buffer_pct={best['middle_exit_buffer_pct']:.3g}, "
            f"middle_exit_confirm_bars={int(best['middle_exit_confirm_bars'])}."
        ),
        (
            f"- Recent performance check: 180d {best['180d_return']:.2%}, "
            f"90d {best['90d_return']:.2%}; max recent MDD {best['max_recent_mdd']:.2%}."
        ),
        "- Small live executor readiness should require positive 90d and 180d behavior across adjacent variants, not only the 10y top row.",
    ]
    return "\n".join(lines) + "\n"


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--bar", default=BAR)
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    args = parser.parse_args()

    eth = _load_candles(ETH_SYMBOL, args.bar)
    btc = _load_candles(BTC_SYMBOL, args.bar)
    eth, btc = _align_pair(eth, btc)
    eth, btc = slice_days(eth, btc, HORIZONS[0][1])

    detail = evaluate_candidates(eth, btc)
    score = candidate_score(detail)

    args.output_dir.mkdir(parents=True, exist_ok=True)
    detail_path = args.output_dir / "eth-bb-squeeze-t-gate-robustness-detail.csv"
    score_path = args.output_dir / "eth-bb-squeeze-t-gate-robustness-score.csv"
    report_path = args.output_dir / "eth-bb-squeeze-t-gate-robustness-report.md"
    detail.to_csv(detail_path, index=False)
    score.to_csv(score_path, index=False)
    report_path.write_text(write_report(detail=detail, score=score, output_dir=args.output_dir), encoding="utf-8")
    print(score.to_string(index=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())