from __future__ import annotations import argparse import sys from dataclasses import dataclass from pathlib import Path import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts import explore_ultrashort as explore COST_SCENARIOS = ( ("maker_maker", 0.0012), ("maker_taker", 0.0021), ("taker_taker", 0.0030), ) PRIMARY_COST = "maker_taker" HORIZONS = ( ("3y", pd.DateOffset(years=3)), ("1y", pd.DateOffset(years=1)), ("6m", pd.DateOffset(months=6)), ("3m", pd.DateOffset(months=3)), ) @dataclass(frozen=True) class Strategy: family: str candidate: object pair: bool def build_strategies() -> list[Strategy]: strategies: list[Strategy] = [ Strategy("baseline_rsi2", explore.build_rsi2_long_guarded_candidate(50, 3.0, 55.0, 0.008, 48), False), Strategy("baseline_rsi2", explore.build_rsi2_long_guarded_candidate(120, 3.0, 55.0, 0.008, 48), False), Strategy( "baseline_price_twap", explore.build_rsi2_long_guarded_price_twap_candidate(50, 3.0, 55.0, 0.008, 48, (0.001, 0.003, 0.005), 2), False, ), Strategy( "baseline_price_twap", explore.build_rsi2_long_guarded_price_twap_candidate(120, 3.0, 55.0, 0.008, 48, (0.001, 0.003, 0.005), 2), False, ), ] strategies.extend( Strategy( "btc_trend_momentum_rsi2", explore.build_eth_btc_rsi_filter_candidate( eth_trend, eth_rsi, 55.0, btc_trend, btc_momentum, btc_min_momentum, ), True, ) for eth_trend in (50, 120) for eth_rsi in (3.0, 5.0) for btc_trend in (120, 240, 480) for btc_momentum in (96, 240) for btc_min_momentum in (0.0, 0.01) ) strategies.extend( Strategy( "btc_shock_guard_rsi2", explore.build_eth_btc_shock_filter_candidate( 50, 3.0, 55.0, 480, 240, btc_min_momentum, btc_shock_lookback, btc_max_realized_vol, btc_max_drawdown, ), True, ) for btc_min_momentum in (0.0, 0.01) for btc_shock_lookback in (96, 240) for btc_max_realized_vol in (0.006, 0.01) for btc_max_drawdown in (0.03, 0.05) ) strategies.extend( Strategy( "ethbtc_ratio_pullback", explore.build_eth_btc_ratio_pullback_candidate( 480, btc_momentum, btc_min_momentum, ratio_length, ratio_std, 5.0, 0.008, ), True, ) for btc_momentum in (96, 240) for btc_min_momentum in (0.0, 0.01) for ratio_length in (48, 96) for ratio_std in (1.5, 2.0) ) strategies.extend( Strategy( "btc_lead_eth_lag", explore.build_btc_lead_eth_lag_candidate( lead_lookback, btc_return_threshold, lag_gap, max_hold_bars, 0.006, take_profit_pct, ), True, ) for lead_lookback in (8, 16) for btc_return_threshold in (0.012, 0.018) for lag_gap in (0.006, 0.012) for max_hold_bars in (8, 32) for take_profit_pct in (0.012, 0.018) ) return strategies def window_rows(strategy: Strategy, eth: list[explore.Candle], btc: list[explore.Candle], window_size: int) -> list[dict[str, object]]: if strategy.pair: return explore.evaluate_pair_candidate_window_rows( candidate=strategy.candidate, eth_candles=eth, btc_candles=btc, window_size=window_size, leverage=explore.LEVERAGE, ) return explore.evaluate_candidate_window_rows( candidate=strategy.candidate, candles=eth, window_size=window_size, leverage=explore.LEVERAGE, ) def full_result(strategy: Strategy, eth: list[explore.Candle], btc: list[explore.Candle]) -> explore.SegmentResult: if strategy.pair: return strategy.candidate.run( eth_candles=eth, btc_candles=btc, leverage=explore.LEVERAGE, warmup_bars=strategy.candidate.warmup_bars, ) return strategy.candidate.run( candles=eth, leverage=explore.LEVERAGE, warmup_bars=strategy.candidate.warmup_bars, ) def append_cost_rows( *, strategy: Strategy, bar: str, eth: list[explore.Candle], rows: list[dict[str, object]], result: explore.SegmentResult, summary_rows: list[dict[str, object]], total_rows: list[dict[str, object]], horizon_rows: list[dict[str, object]], ) -> None: for cost_name, cost_value in COST_SCENARIOS: summary = explore.add_cost_metrics( pd.DataFrame([explore.summarize_window_rows(rows, strategy.candidate.name)]), cost_value, ).iloc[0].to_dict() summary_rows.append( { "family": strategy.family, "cost": cost_name, "symbol": "ETH-USDT-SWAP", "signal_symbol": "BTC-USDT-SWAP" if strategy.pair else "", "bar": bar, "actual_bars": len(eth), "first_candle": explore._format_ts(eth[0].ts), "last_candle": explore._format_ts(eth[-1].ts), **summary, } ) net_equity = explore.cost_adjusted_trade_equity_frame(result, cost_value) metrics = explore.annualized_metrics_from_equity(net_equity, eth[0].ts, eth[-1].ts) years_actual = (eth[-1].ts - eth[0].ts) / 86_400_000 / 365 gross_annualized = (1.0 + result.total_return) ** (1.0 / years_actual) - 1.0 if result.total_return > -1.0 else 0.0 total_rows.append( { "family": strategy.family, "cost": cost_name, "symbol": "ETH-USDT-SWAP", "signal_symbol": "BTC-USDT-SWAP" if strategy.pair else "", "bar": bar, "name": strategy.candidate.name, "first_candle": explore._format_ts(eth[0].ts), "last_candle": explore._format_ts(eth[-1].ts), "years": years_actual, "trades": result.trade_count, "gross_total_return": result.total_return, "gross_annualized_return": gross_annualized, "gross_max_drawdown_mark_to_market": result.max_drawdown, **metrics, } ) horizon = explore.recent_horizon_metrics_from_equity(net_equity, eth[-1].ts, HORIZONS) for horizon_row in horizon.to_dict("records"): horizon_rows.append( { "family": strategy.family, "cost": cost_name, "symbol": "ETH-USDT-SWAP", "signal_symbol": "BTC-USDT-SWAP" if strategy.pair else "", "bar": bar, "name": strategy.candidate.name, "trades": result.trade_count, **horizon_row, } ) def markdown_table(frame: pd.DataFrame) -> str: columns = list(frame.columns) rows = [columns, ["---" for _ in columns]] for record in frame.to_dict("records"): rows.append([record[column] for column in columns]) return "\n".join("| " + " | ".join(format_markdown_cell(value) for value in row) + " |" for row in rows) def format_markdown_cell(value: object) -> str: if isinstance(value, float): return f"{value:.6g}" return str(value).replace("|", "\\|") def markdown_report( *, summary: pd.DataFrame, total: pd.DataFrame, horizon: pd.DataFrame, output_files: list[Path], command: str, ) -> str: primary_summary = summary[summary["cost"] == PRIMARY_COST].copy() primary_total = total[total["cost"] == PRIMARY_COST].copy() top = primary_summary.head(10) family = ( primary_summary.groupby("family", as_index=False) .agg( best_net_ci95_low=("net_ci95_low", "max"), best_net_avg_return=("net_avg_return", "max"), best_positive_window_rate=("positive_window_rate", "max"), candidate_count=("name", "count"), ) .sort_values(["best_net_ci95_low", "best_net_avg_return"], ascending=False) ) horizon_top = ( horizon[horizon["cost"] == PRIMARY_COST] .sort_values(["horizon", "net_annualized_return"], ascending=[True, False]) .groupby("horizon", observed=True) .head(3) ) best = top.iloc[0].to_dict() if len(top) else {} lines = [ "# ETH BTC regime variants", "", f"Run command: `{command}`", "", "Output files:", *[f"- `{path}`" for path in output_files], "", "Primary sort: maker_taker cost, by net_ci95_low then net_avg_return.", "", "Top 10 candidates:", markdown_table(top[ [ "family", "name", "net_avg_return", "net_ci95_low", "positive_window_rate", "trades", "avg_trades_per_window", "max_drawdown", ] ]), "", "Family summary:", markdown_table(family), "", "Recent horizon leaders:", markdown_table(horizon_top[ [ "horizon", "family", "name", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar", ] ]), "", "Interpretation:", f"- Effective: BTC trend plus momentum gating on ETH RSI2. Best maker_taker window result is `{best.get('name', '')}` with net_ci95_low {format_markdown_cell(best.get('net_ci95_low', 0.0))} and net_avg_return {format_markdown_cell(best.get('net_avg_return', 0.0))}.", "- Effective but not incremental: loose BTC shock guards tie the best trend/momentum result, so the tested vol/drawdown caps mostly did not bind.", "- Not robust: BTC lead ETH lag has positive best net_avg_return but negative best net_ci95_low, so the average is not enough to promote it.", "- Not effective: ETHBTC ratio low pullback variants are negative on both best net_avg_return and best net_ci95_low.", "- Baseline note: ETH price-TWAP has strong recent horizon returns, but its sampled-window maker_taker net_ci95_low is deeply negative; it is not a robust regime condition in this run.", ] if len(primary_total): total_top = primary_total.sort_values(["net_calmar", "net_annualized_return"], ascending=False).head(5) lines.extend(["", "Best full-period net Calmar:", markdown_table(total_top[["family", "name", "trades", "net_total_return", "net_annualized_return", "net_max_drawdown", "net_calmar"]])]) return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--bar", default="15m") parser.add_argument("--years", type=float, default=3.25) parser.add_argument("--window-size", type=int, default=explore.WINDOW_SIZE) parser.add_argument("--output-dir", type=Path, default=Path("reports/eth-exploration")) args = parser.parse_args() requested_bars = explore.history_bars_for_years(args.bar, args.years) client = explore.OkxClient() eth = explore.get_candles_cached(client, "ETH-USDT-SWAP", args.bar, requested_bars) btc = explore.get_candles_cached(client, "BTC-USDT-SWAP", args.bar, requested_bars) eth, btc = explore.align_pair_candles(eth, btc) strategies = build_strategies() summary_rows: list[dict[str, object]] = [] total_rows: list[dict[str, object]] = [] horizon_rows: list[dict[str, object]] = [] for index, strategy in enumerate(strategies, start=1): rows = window_rows(strategy, eth, btc, args.window_size) result = full_result(strategy, eth, btc) append_cost_rows( strategy=strategy, bar=args.bar, eth=eth, rows=rows, result=result, summary_rows=summary_rows, total_rows=total_rows, horizon_rows=horizon_rows, ) print(f"done {index}/{len(strategies)} {strategy.family} {strategy.candidate.name}") summary = pd.DataFrame(summary_rows).sort_values( ["cost", "net_ci95_low", "net_avg_return"], ascending=[True, False, False], ) primary = summary[summary["cost"] == PRIMARY_COST] others = summary[summary["cost"] != PRIMARY_COST] summary = pd.concat([primary, others], ignore_index=True) total = pd.DataFrame(total_rows).sort_values(["cost", "net_calmar", "net_annualized_return"], ascending=[True, False, False]) horizon = pd.DataFrame(horizon_rows) horizon["horizon"] = pd.Categorical(horizon["horizon"], categories=["3y", "1y", "6m", "3m"], ordered=True) horizon = horizon.sort_values(["cost", "horizon", "net_annualized_return"], ascending=[True, True, False]) args.output_dir.mkdir(parents=True, exist_ok=True) summary_path = args.output_dir / "eth-btc-regime-summary.csv" total_path = args.output_dir / "eth-btc-regime-total.csv" horizon_path = args.output_dir / "eth-btc-regime-horizon.csv" top10_path = args.output_dir / "eth-btc-regime-top10.csv" report_path = args.output_dir / "eth-btc-regime-report.md" summary.to_csv(summary_path, index=False) total.to_csv(total_path, index=False) horizon.to_csv(horizon_path, index=False) summary[summary["cost"] == PRIMARY_COST].head(10).to_csv(top10_path, index=False) command = f"rtk .venv/bin/python {Path(__file__).as_posix()} --bar {args.bar} --years {args.years} --window-size {args.window_size}" report_path.write_text( markdown_report( summary=summary, total=total, horizon=horizon, output_files=[summary_path, total_path, horizon_path, top10_path, report_path], command=command, ), encoding="utf-8", ) print(summary[summary["cost"] == PRIMARY_COST].head(10).to_string(index=False)) return 0 if __name__ == "__main__": raise SystemExit(main())