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