okx-codex-trader/scripts/evaluate_eth_crash_follow_overlay.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_bearish_price_proxy import Spec, joined_frames, load_frame, resample, run_spec
from scripts.search_long_short_fusion import INITIAL_EQUITY, component_returns, markdown_table, metrics


OUTPUT_DIR = Path("reports/long-short-fusion")
PREFIX = "eth-crash-follow-overlay"
BASELINE_EQUITY = Path("reports/eth-exploration/eth-focused-portfolio-conservative-equity.csv")
BASELINE_PORTFOLIO = "all_legs-risk-3-c0124-eth_btc_rsi_filter+btc_lead_eth_lag_15m+eth_robust_twap"
HORIZONS = (
    ("full", None),
    ("3y", pd.DateOffset(years=3)),
    ("1y", pd.DateOffset(years=1)),
    ("6m", pd.DateOffset(months=6)),
    ("3m", pd.DateOffset(months=3)),
)
CRASH_FOLLOW = Spec("crash_follow", "1H", 20, 120, 8, 0.035, 0.02, 0.06, 96, "btc_riskoff")
WEIGHTS = (0.05, 0.10, 0.15, 0.20, 0.25, 0.30)


def baseline_equity(path: Path, portfolio: str) -> pd.Series:
    frame = pd.read_csv(path)
    selected = frame[
        (frame["portfolio"] == portfolio)
        & (frame["cost_model"] == "maker_taker")
        & (frame["scope"] == "all_legs")
    ].copy()
    selected["date"] = pd.to_datetime(selected["date"], utc=True)
    series = selected.sort_values("date").set_index("date")["equity"].astype(float)
    series.name = portfolio
    return series


def crash_follow_returns(spec: Spec) -> pd.Series:
    eth = load_frame("ETH-USDT-SWAP")
    btc = load_frame("BTC-USDT-SWAP")
    frame = joined_frames(resample(eth, spec.bar), resample(btc, spec.bar))
    equity, _ = run_spec(spec, frame)
    returns = component_returns(equity)
    returns.name = spec.name
    return returns


def horizon_metrics(series: pd.Series) -> dict[str, dict[str, object]]:
    end = series.index[-1]
    rows = {}
    for label, offset in HORIZONS:
        scoped = series if offset is None else series[series.index >= end - offset]
        if len(scoped) < 2:
            scoped = series
        rows[label] = {
            "start": scoped.index[0].strftime("%Y-%m-%d"),
            "end": scoped.index[-1].strftime("%Y-%m-%d"),
            **metrics(scoped),
        }
    return rows


def overlay_equity(base: pd.Series, proxy_returns: pd.Series, weight: float) -> pd.Series:
    aligned = pd.DataFrame({"base": component_returns(base), "proxy": proxy_returns}).dropna()
    combined = aligned["base"] + aligned["proxy"] * weight
    equity = INITIAL_EQUITY * (1.0 + combined).cumprod()
    equity.name = f"{base.name}+{CRASH_FOLLOW.name}@{weight:.2f}"
    return equity


def result_rows(base: pd.Series, proxy_returns: pd.Series) -> pd.DataFrame:
    base_rows = horizon_metrics(base)
    rows = []
    for weight in WEIGHTS:
        overlaid = overlay_equity(base, proxy_returns, weight)
        overlaid_rows = horizon_metrics(overlaid)
        for horizon, row in overlaid_rows.items():
            baseline = base_rows[horizon]
            rows.append(
                {
                    "overlay_weight": weight,
                    "horizon": horizon,
                    "start": row["start"],
                    "end": row["end"],
                    "total_return": row["total_return"],
                    "annualized_return": row["annualized_return"],
                    "max_drawdown": row["max_drawdown"],
                    "calmar": row["calmar"],
                    "baseline_total_return": baseline["total_return"],
                    "baseline_max_drawdown": baseline["max_drawdown"],
                    "delta_total_return": row["total_return"] - baseline["total_return"],
                    "delta_max_drawdown": row["max_drawdown"] - baseline["max_drawdown"],
                }
            )
    return pd.DataFrame(rows)


def report_text(command: str, baseline_name: str, proxy_name: str, paths: list[Path], results: pd.DataFrame) -> str:
    full = results[results["horizon"] == "full"].copy()
    best = full.sort_values(["calmar", "delta_max_drawdown", "total_return"], ascending=[False, True, False]).iloc[0]
    recent = results[results["horizon"].isin(["1y", "6m", "3m"])]
    recent_ok = recent.groupby("overlay_weight")["total_return"].apply(lambda values: bool((values > 0.0).all()))
    include = bool(
        recent_ok.get(float(best["overlay_weight"]), False)
        and best["delta_total_return"] > 0.0
        and best["delta_max_drawdown"] <= 0.005
    )
    verdict = (
        f"Include crash_follow as a small overlay dimension in fusion search, capped at {best['overlay_weight']:.2f} in the first pass. Higher tested weights add return but expand full/3y drawdown too much for this conservative baseline."
        if include
        else "Do not add crash_follow to fusion search now; the overlay does not improve the baseline cleanly across recent windows."
    )
    keep = [
        "overlay_weight",
        "horizon",
        "total_return",
        "annualized_return",
        "max_drawdown",
        "calmar",
        "delta_total_return",
        "delta_max_drawdown",
    ]
    return "\n".join(
        [
            "# ETH Crash-Follow Overlay Evaluation",
            "",
            f"Run command: `{command}`",
            "",
            "Output files:",
            *[f"- `{path}`" for path in paths],
            "",
            f"Baseline: `{baseline_name}` from `reports/eth-exploration/eth-focused-portfolio-conservative-equity.csv`.",
            f"Overlay proxy: `{proxy_name}` from the ETH bearish price-proxy all-window-positive candidate.",
            "",
            "Method: daily baseline return plus `overlay_weight * crash_follow_daily_return`; local candle data only; no live path touched.",
            "",
            "## Results",
            "",
            markdown_table(results[keep]),
            "",
            "## Verdict",
            "",
            verdict,
            "",
        ]
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--baseline-equity", type=Path, default=BASELINE_EQUITY)
    parser.add_argument("--baseline-portfolio", default=BASELINE_PORTFOLIO)
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    args = parser.parse_args()

    args.output_dir.mkdir(parents=True, exist_ok=True)
    base_equity = baseline_equity(args.baseline_equity, args.baseline_portfolio)
    proxy_returns = crash_follow_returns(CRASH_FOLLOW)
    results = result_rows(base_equity, proxy_returns)

    results_path = args.output_dir / f"{PREFIX}.csv"
    report_path = args.output_dir / f"{PREFIX}.md"
    results.to_csv(results_path, index=False)
    report_path.write_text(
        report_text(
            "rtk .venv/bin/python scripts/evaluate_eth_crash_follow_overlay.py",
            args.baseline_portfolio,
            CRASH_FOLLOW.name,
            [results_path, report_path],
            results,
        ),
        encoding="utf-8",
    )
    print(report_path)
    print(results.to_string(index=False))
    return 0


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