from __future__ import annotations import argparse import itertools import json import time from dataclasses import dataclass from pathlib import Path from statistics import mean from typing import Any, Mapping, Sequence import requests from config import BotConfig, RuntimePaths from market_universe import select_scan_coins from position_sizing import calculate_position_size from strategy import PositionState, exit_long_position, should_enter_by_strategy_family from strategy_registry import StrategyPreset, load_strategy_presets, preset_to_config @dataclass(frozen=True) class Candle: ts: int | float open: float high: float low: float close: float volume: float = 0.0 @dataclass(frozen=True) class ReplayTrade: strategy_id: str coin: str entry_ts: float exit_ts: float entry_price: float exit_price: float size: float pnl_usd: float exit_reason: str minutes_in_trade: int @dataclass(frozen=True) class ReplayResult: strategy_id: str closed_trades: int win_rate: float avg_pnl_usd: float total_pnl_usd: float max_drawdown_usd: float profit_factor: float score: float trades: list[ReplayTrade] @dataclass(frozen=True) class BreakdownStats: closed_trades: int win_rate: float avg_pnl_usd: float total_pnl_usd: float @dataclass(frozen=True) class TradeBreakdowns: by_coin: dict[str, BreakdownStats] by_exit_reason: dict[str, BreakdownStats] @dataclass(frozen=True) class WalkForwardResult: strategy_id: str aggregate: ReplayResult segments: list[ReplayResult] winning_segments: int losing_segments: int stability_score: float breakdowns: TradeBreakdowns @dataclass(frozen=True) class CoinSliceReport: excluded_coins: tuple[str, ...] included_coins: tuple[str, ...] results: list[WalkForwardResult] @dataclass(frozen=True) class ParameterSweepResult: strategy_id: str parameters: dict[str, int | float | str | bool] result: WalkForwardResult def candle_from_hl(row: Mapping[str, Any]) -> Candle: return Candle( ts=float(row.get("t", 0)) / 1000.0, open=float(row["o"]), high=float(row["h"]), low=float(row["l"]), close=float(row["c"]), volume=float(row.get("v", 0.0) or 0.0), ) def _atr(candles: Sequence[Candle], fallback_price: float) -> float: if len(candles) < 2: return fallback_price * 0.02 ranges = [] for prev, cur in zip(candles, candles[1:]): ranges.append(max(cur.high - cur.low, abs(cur.high - prev.close), abs(cur.low - prev.close))) if not ranges: return fallback_price * 0.02 return mean(ranges[-14:]) def _max_drawdown(pnls: Sequence[float]) -> float: equity = 0.0 peak = 0.0 max_dd = 0.0 for pnl in pnls: equity += pnl peak = max(peak, equity) max_dd = min(max_dd, equity - peak) return round(max_dd, 8) def _profit_factor(pnls: Sequence[float]) -> float: gains = sum(p for p in pnls if p > 0) losses = abs(sum(p for p in pnls if p < 0)) if losses == 0: return round(gains, 8) if gains else 0.0 return round(gains / losses, 8) def _score(pnls: Sequence[float], closed: int) -> float: if not pnls: return -2.0 avg_pnl = mean(pnls) win_rate = sum(1 for p in pnls if p > 0) / closed pf = min(_profit_factor(pnls), 5.0) dd_penalty = abs(_max_drawdown(pnls)) * 0.35 sample_penalty = 1.0 if closed < 5 else 0.0 return round(avg_pnl + win_rate * 0.5 + pf * 0.1 - dd_penalty - sample_penalty, 8) def run_replay_for_preset(preset: StrategyPreset, candles_by_coin: Mapping[str, Sequence[Candle]], *, volumes: Mapping[str, float]) -> ReplayResult: cfg = preset_to_config(preset, BotConfig.from_file()) scan_coins = set(select_scan_coins(volumes, cfg)) open_positions: dict[str, tuple[float, float, float, PositionState]] = {} trades: list[ReplayTrade] = [] cooldowns: dict[str, float] = {} timeline: list[tuple[float, str, Candle, Sequence[Candle], int]] = [] for coin, candles in candles_by_coin.items(): normalized = list(candles) for idx, candle in enumerate(normalized): timeline.append((float(candle.ts), coin.upper(), candle, normalized, idx)) timeline.sort(key=lambda item: item[0]) price_history: dict[str, list[tuple[float, float]]] = {} for ts, coin, candle, coin_candles, idx in timeline: if coin not in scan_coins: continue price_history.setdefault(coin, []).append((ts, candle.close)) if cfg.strategy_family in {"volatility_squeeze_breakout", "confirmed_squeeze_breakout"}: price_history[coin] = price_history[coin][-max(2, cfg.squeeze_lookback_ticks + 1):] elif cfg.strategy_family in {"trend_pullback_sma_vwap", "bollinger_rsi_mean_reversion", "multi_day_trend_investment", "relative_strength_rotation"}: price_history[coin] = price_history[coin][-1200:] else: cutoff = ts - cfg.crash_window_minutes * 60 price_history[coin] = [row for row in price_history[coin] if row[0] >= cutoff] if coin in open_positions: entry_price, size, leverage, state = open_positions[coin] decision = exit_long_position(entry_px=entry_price, current_px=candle.close, leverage=leverage, state=state, now_ts=ts, cfg=cfg) if decision.close: pnl = (candle.close - entry_price) * size trades.append(ReplayTrade( strategy_id=preset.strategy_id, coin=coin, entry_ts=state.entry_ts, exit_ts=ts, entry_price=entry_price, exit_price=candle.close, size=size, pnl_usd=round(pnl, 8), exit_reason=decision.reason, minutes_in_trade=int((ts - state.entry_ts) // 60), )) cooldowns[coin] = ts open_positions.pop(coin, None) continue if len(open_positions) >= cfg.max_total_trades or len(price_history[coin]) < 2: continue baseline = coin_candles[max(0, idx - 5)].open if coin_candles else 0.0 entry = should_enter_by_strategy_family( coin=coin, history=price_history[coin], current_price=candle.close, volume_24h=volumes.get(coin, 0.0), baseline_price=baseline, open_positions=set(open_positions), cooldowns=cooldowns, now_ts=ts, cfg=cfg, market_histories=price_history, ) if entry.enter: atr = _atr(coin_candles[max(0, idx - 15): idx + 1], candle.close) leverage = float(cfg.default_leverage) atr_stop_pct = (cfg.atr_sl_multiplier * atr / candle.close) * 100.0 if candle.close > 0 else cfg.max_hard_stop_pct stop_distance_pct = max(0.01, min(cfg.max_hard_stop_pct, atr_stop_pct)) sizing = calculate_position_size( coin=coin, current_price=candle.close, leverage=leverage, stop_distance_pct=stop_distance_pct, cfg=cfg, ) size = sizing.size open_positions[coin] = ( candle.close, size, leverage, PositionState(entry_ts=ts, high_px=candle.close, atr_sl_px=candle.close - cfg.atr_sl_multiplier * atr), ) last_by_coin = {coin.upper(): list(candles)[-1] for coin, candles in candles_by_coin.items() if candles} for coin, (entry_price, size, _leverage, state) in list(open_positions.items()): last = last_by_coin.get(coin) if not last: continue pnl = (last.close - entry_price) * size trades.append(ReplayTrade( strategy_id=preset.strategy_id, coin=coin, entry_ts=state.entry_ts, exit_ts=float(last.ts), entry_price=entry_price, exit_price=last.close, size=size, pnl_usd=round(pnl, 8), exit_reason="end_of_replay", minutes_in_trade=int((float(last.ts) - state.entry_ts) // 60), )) pnls = [trade.pnl_usd for trade in trades] closed = len(trades) return ReplayResult( strategy_id=preset.strategy_id, closed_trades=closed, win_rate=round(sum(1 for p in pnls if p > 0) / closed, 4) if closed else 0.0, avg_pnl_usd=round(mean(pnls), 8) if pnls else 0.0, total_pnl_usd=round(sum(pnls), 8), max_drawdown_usd=_max_drawdown(pnls), profit_factor=_profit_factor(pnls), score=_score(pnls, closed), trades=trades, ) def _stats_for_trades(trades: Sequence[ReplayTrade]) -> BreakdownStats: pnls = [trade.pnl_usd for trade in trades] closed = len(pnls) return BreakdownStats( closed_trades=closed, win_rate=round(sum(1 for p in pnls if p > 0) / closed, 4) if closed else 0.0, avg_pnl_usd=round(mean(pnls), 8) if pnls else 0.0, total_pnl_usd=round(sum(pnls), 8), ) def aggregate_trade_breakdowns(trades: Sequence[ReplayTrade]) -> TradeBreakdowns: by_coin_raw: dict[str, list[ReplayTrade]] = {} by_exit_raw: dict[str, list[ReplayTrade]] = {} for trade in trades: by_coin_raw.setdefault(trade.coin, []).append(trade) by_exit_raw.setdefault(trade.exit_reason, []).append(trade) return TradeBreakdowns( by_coin={coin: _stats_for_trades(rows) for coin, rows in sorted(by_coin_raw.items())}, by_exit_reason={reason: _stats_for_trades(rows) for reason, rows in sorted(by_exit_raw.items())}, ) def _slice_candles_by_segment(candles_by_coin: Mapping[str, Sequence[Candle]], segment_index: int, segments: int) -> dict[str, list[Candle]]: sliced: dict[str, list[Candle]] = {} for coin, rows in candles_by_coin.items(): normalized = list(rows) if not normalized: continue start = len(normalized) * segment_index // segments end = len(normalized) * (segment_index + 1) // segments part = normalized[start:end] if part: sliced[coin] = part return sliced def run_walk_forward_tournament(presets: Sequence[StrategyPreset], candles_by_coin: Mapping[str, Sequence[Candle]], *, volumes: Mapping[str, float], segments: int = 3) -> list[WalkForwardResult]: segments = max(1, segments) results: list[WalkForwardResult] = [] for preset in presets: aggregate = run_replay_for_preset(preset, candles_by_coin, volumes=volumes) segment_results = [ run_replay_for_preset(preset, _slice_candles_by_segment(candles_by_coin, index, segments), volumes=volumes) for index in range(segments) ] winning = sum(1 for item in segment_results if item.total_pnl_usd > 0) losing = sum(1 for item in segment_results if item.total_pnl_usd < 0) stability = round(winning / len(segment_results), 4) if segment_results else 0.0 results.append(WalkForwardResult( strategy_id=preset.strategy_id, aggregate=aggregate, segments=segment_results, winning_segments=winning, losing_segments=losing, stability_score=stability, breakdowns=aggregate_trade_breakdowns(aggregate.trades), )) return sorted(results, key=lambda r: (r.aggregate.score, r.stability_score, r.aggregate.total_pnl_usd), reverse=True) def run_replay_tournament(presets: Sequence[StrategyPreset], candles_by_coin: Mapping[str, Sequence[Candle]], *, volumes: Mapping[str, float]) -> list[ReplayResult]: results = [run_replay_for_preset(preset, candles_by_coin, volumes=volumes) for preset in presets] return sorted(results, key=lambda r: (r.score, r.total_pnl_usd, r.closed_trades), reverse=True) def run_coin_slice_replay(presets: Sequence[StrategyPreset], candles_by_coin: Mapping[str, Sequence[Candle]], *, volumes: Mapping[str, float], exclude_coins: Sequence[str] = (), include_coins: Sequence[str] = (), segments: int = 3) -> CoinSliceReport: excluded = tuple(sorted({coin.upper() for coin in exclude_coins})) allowed = {coin.upper() for coin in include_coins} sliced: dict[str, Sequence[Candle]] = {} for coin, rows in candles_by_coin.items(): symbol = coin.upper() if symbol in excluded: continue if allowed and symbol not in allowed: continue sliced[symbol] = rows sliced_volumes = {coin: volumes.get(coin, 0.0) for coin in sliced} return CoinSliceReport( excluded_coins=excluded, included_coins=tuple(sorted(sliced)), results=run_walk_forward_tournament(presets, sliced, volumes=sliced_volumes, segments=segments), ) def _sweep_strategy_id(base_id: str, params: Mapping[str, int | float | str | bool]) -> str: bits = [] for key, value in sorted(params.items()): safe_value = str(value).replace(".", "p").replace("-", "m") bits.append(f"{key}_{safe_value}") return base_id + "__" + "__".join(bits) def run_parameter_sweep(base_preset: StrategyPreset, candles_by_coin: Mapping[str, Sequence[Candle]], *, volumes: Mapping[str, float], grid: Mapping[str, Sequence[int | float | str | bool]], segments: int = 3) -> list[ParameterSweepResult]: keys = list(grid) variants: list[ParameterSweepResult] = [] for values in itertools.product(*(grid[key] for key in keys)): params = dict(zip(keys, values)) merged = {**base_preset.parameters, **params} strategy_id = _sweep_strategy_id(base_preset.strategy_id, params) preset = StrategyPreset(strategy_id=strategy_id, label=f"{base_preset.label} sweep", source="parameter_sweep", parameters=merged) result = run_walk_forward_tournament([preset], candles_by_coin, volumes=volumes, segments=segments)[0] variants.append(ParameterSweepResult(strategy_id=strategy_id, parameters=params, result=result)) return sorted(variants, key=lambda item: (item.result.aggregate.score, item.result.stability_score, item.result.aggregate.total_pnl_usd), reverse=True) def render_parameter_sweep_report(results: Sequence[ParameterSweepResult], *, top_n: int = 10, min_samples: int = 5) -> str: lines = ["Parameter Sweep"] for idx, item in enumerate(results[:top_n], start=1): aggregate = item.result.aggregate warning = " observation-only" if aggregate.closed_trades < min_samples else "" lines.append( f"{idx}. {item.strategy_id}: score={aggregate.score:+.4f}, stability={item.result.stability_score:.2f}, " f"n={aggregate.closed_trades}, win={aggregate.win_rate:.2f}, total={aggregate.total_pnl_usd:+.4f}, " f"max_dd={aggregate.max_drawdown_usd:+.4f}, pf={aggregate.profit_factor:.2f}, params={item.parameters}{warning}" ) return "\n".join(lines) def render_walk_forward_report(results: Sequence[WalkForwardResult], *, min_samples: int = 5) -> str: lines = ["Historical Replay Tournament", "Mode: walk-forward"] for idx, result in enumerate(results, start=1): aggregate = result.aggregate warning = " observation-only" if aggregate.closed_trades < min_samples else "" segment_totals = ", ".join(f"{seg.total_pnl_usd:+.2f}" for seg in result.segments) lines.append( f"{idx}. {result.strategy_id}: score={aggregate.score:+.4f}, stability={result.stability_score:.2f}, " f"n={aggregate.closed_trades}, win={aggregate.win_rate:.2f}, avg={aggregate.avg_pnl_usd:+.4f}, " f"total={aggregate.total_pnl_usd:+.4f}, max_dd={aggregate.max_drawdown_usd:+.4f}, " f"pf={aggregate.profit_factor:.2f}, segments=[{segment_totals}]{warning}" ) if result.breakdowns.by_coin: coin_bits = [] for coin, stats in sorted(result.breakdowns.by_coin.items(), key=lambda item: item[1].total_pnl_usd, reverse=True)[:5]: coin_bits.append(f"{coin}:n={stats.closed_trades},total={stats.total_pnl_usd:+.2f},win={stats.win_rate:.2f}") lines.append(" coins: " + "; ".join(coin_bits)) if result.breakdowns.by_exit_reason: reason_bits = [] for reason, stats in sorted(result.breakdowns.by_exit_reason.items(), key=lambda item: item[1].closed_trades, reverse=True)[:4]: reason_bits.append(f"{reason}:n={stats.closed_trades},total={stats.total_pnl_usd:+.2f}") lines.append(" exits: " + "; ".join(reason_bits)) return "\n".join(lines) def render_replay_report(results: Sequence[ReplayResult], *, min_samples: int = 5) -> str: lines = ["Historical Replay Tournament"] for idx, result in enumerate(results, start=1): warning = " observation-only" if result.closed_trades < min_samples else "" lines.append( f"{idx}. {result.strategy_id}: score={result.score:+.4f}, n={result.closed_trades}, " f"win={result.win_rate:.2f}, avg={result.avg_pnl_usd:+.4f}, total={result.total_pnl_usd:+.4f}, " f"max_dd={result.max_drawdown_usd:+.4f}, pf={result.profit_factor:.2f}{warning}" ) return "\n".join(lines) def fetch_hyperliquid_candles( coin: str, *, interval: str, hours_back: int, cache_dir: Path, tls_verify: bool = True, request_post=requests.post, sleep_fn=time.sleep, max_attempts: int = 4, allow_stale_cache: bool = True, fresh_cache_ttl_seconds: int = 1800, ) -> list[Candle]: cache_dir.mkdir(parents=True, exist_ok=True) cache_file = cache_dir / f"{coin}_{interval}_{hours_back}h.json" if cache_file.exists() and time.time() - cache_file.stat().st_mtime < fresh_cache_ttl_seconds: raw = json.loads(cache_file.read_text(encoding="utf-8")) return [candle_from_hl(row) for row in raw] end_ms = int(time.time() * 1000) start_ms = end_ms - hours_back * 3600 * 1000 payload = {"type": "candleSnapshot", "req": {"coin": coin, "interval": interval, "startTime": start_ms, "endTime": end_ms}} last_exc: Exception | None = None for attempt in range(1, max(1, max_attempts) + 1): try: response = request_post("https://api.hyperliquid.xyz/info", json=payload, verify=tls_verify, timeout=20) response.raise_for_status() raw = response.json() cache_file.write_text(json.dumps(raw), encoding="utf-8") return [candle_from_hl(row) for row in raw] except Exception as exc: last_exc = exc if attempt < max(1, max_attempts): sleep_fn(float(2 ** (attempt - 1))) if allow_stale_cache and cache_file.exists(): raw = json.loads(cache_file.read_text(encoding="utf-8")) return [candle_from_hl(row) for row in raw] if last_exc is not None: raise last_exc return [] def main(argv: list[str] | None = None) -> int: cfg = BotConfig.from_file() paths = RuntimePaths.from_config(cfg) parser = argparse.ArgumentParser(description="Run side-effect-free historical replay tournament across strategy presets.") parser.add_argument("--coins", nargs="*", default=list(cfg.allowed_coins[:8]) or ["BTC", "ETH", "SOL", "HYPE"]) parser.add_argument("--strategies", nargs="*", default=["v61_tight_survival", "v60_survival", "v59_1_frequency_boost", "candidate_v62_anti_chop_context", "candidate_v62_scalp_fast_rebound", "candidate_v62_deep_capitulation"]) parser.add_argument("--hours-back", type=int, default=48) parser.add_argument("--interval", default="15m") parser.add_argument("--min-samples", type=int, default=5) parser.add_argument("--segments", type=int, default=3, help="Walk-forward segment count; use 1 for aggregate-only style.") args = parser.parse_args(argv) cache_dir = paths.runtime_dir / "candle_cache" candles: dict[str, list[Candle]] = {} volumes: dict[str, float] = {} for coin in args.coins: try: rows = fetch_hyperliquid_candles(coin.upper(), interval=args.interval, hours_back=args.hours_back, cache_dir=cache_dir, tls_verify=cfg.tls_verify) except Exception as exc: print(f"WARN {coin}: candles unavailable: {exc}") continue if rows: candles[coin.upper()] = rows volumes[coin.upper()] = max(sum(c.volume for c in rows), cfg.min_volume_24h) presets_by_id = load_strategy_presets() presets = [presets_by_id[sid] for sid in args.strategies if sid in presets_by_id] results = run_walk_forward_tournament(presets, candles, volumes=volumes, segments=args.segments) report = render_walk_forward_report(results, min_samples=args.min_samples) report_path = paths.reports_dir / "historical_replay_report.txt" report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(report, encoding="utf-8") print(report) print(f"Report: {report_path}") return 0 if __name__ == "__main__": raise SystemExit(main())