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())