feat: Add backtesting parameter sweep support and related functionality

scripts/backtest_sweep.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+import argparse
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+
+from arbitrade.backtesting import load_replay_events
+from arbitrade.backtesting.sweep import (
+    PromotionCriteria,
+    SweepResult,
+    build_parameter_grid,
+    persist_sweep_results,
+    run_parameter_search,
+)
+from arbitrade.detection.graph import CurrencyGraph, TriangularCycle
+
+
+def _parse_balances(raw: str) -> Mapping[str, float]:
+    balances: dict[str, float] = {}
+    for entry in raw.split(","):
+        stripped = entry.strip()
+        if not stripped:
+            continue
+        asset, value = stripped.split("=", 1)
+        balances[asset.strip().upper()] = float(value)
+    return balances
+
+
+def _parse_float_list(raw: str) -> list[float]:
+    values = [item.strip() for item in raw.split(",") if item.strip()]
+    if not values:
+        raise ValueError("expected at least one numeric value")
+    return [float(value) for value in values]
+
+
+def _parse_pair_universes(raw: str) -> list[tuple[str, ...]]:
+    universes: list[tuple[str, ...]] = []
+    for chunk in raw.split(";"):
+        symbols = tuple(item.strip().upper()
+                        for item in chunk.split("|") if item.strip())
+        if symbols:
+            universes.append(symbols)
+    if not universes:
+        raise ValueError("at least one pair universe must be provided")
+    return universes
+
+
+def _build_graph_from_symbols(symbols: Sequence[str]) -> dict[str, list[TriangularCycle]]:
+    graph = CurrencyGraph()
+    for symbol in symbols:
+        normalized = symbol.upper()
+        if "/" not in normalized:
+            continue
+        base, quote = normalized.split("/", 1)
+        graph.add_pair(base, quote, normalized)
+
+    cycles = graph.triangular_cycles()
+    return graph.index_cycles_by_pair(cycles)
+
+
+def _print_top_results(results: Sequence[SweepResult], *, limit: int = 5) -> None:
+    print(f"Top {min(limit, len(results))} result(s) by out-of-sample score:")
+    for index, result in enumerate(results[:limit], start=1):
+        print(
+            "- "
+            f"#{index} "
+            f"theta={result.parameters.min_profit_threshold:.6f}, "
+            f"capital={result.parameters.trade_capital:.2f}, "
+            f"pairs={','.join(result.parameters.pair_universe)}, "
+            f"staleness={result.parameters.staleness_threshold_seconds:.2f}s, "
+            f"test_score={result.test_score:.4f}, "
+            f"promotion_ready={result.promotion_ready}"
+        )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Run backtesting parameter sweep with train/test split.")
+    parser.add_argument("--events", type=Path, required=True)
+    parser.add_argument("--starting-balances", type=str, default="USD=1000.0")
+    parser.add_argument("--theta-values", type=str,
+                        default="0.0003,0.0005,0.0008")
+    parser.add_argument("--trade-capital-values",
+                        type=str, default="50,100,150")
+    parser.add_argument(
+        "--pair-universes",
+        type=str,
+        default="BTC/USD|ETH/BTC|ETH/USD",
+        help="Semicolon-separated universes, each with | delimited pairs",
+    )
+    parser.add_argument("--staleness-threshold-values",
+                        type=str, default="3,5,8")
+    parser.add_argument("--train-ratio", type=float, default=0.7)
+    parser.add_argument("--output", type=Path,
+                        default=Path("ops/backtesting/parameter_sweep_results.json"))
+
+    parser.add_argument("--min-test-realized-pnl-usd", type=float, default=0.0)
+    parser.add_argument("--min-test-win-rate", type=float, default=0.5)
+    parser.add_argument("--min-test-fill-rate", type=float, default=0.9)
+    parser.add_argument("--max-test-drawdown-usd", type=float, default=25.0)
+    parser.add_argument("--max-generalization-gap-ratio",
+                        type=float, default=0.5)
+
+    args = parser.parse_args()
+
+    events = load_replay_events(args.events)
+    symbols = sorted({event.symbol.upper() for event in events})
+    cycles_by_pair = _build_graph_from_symbols(symbols)
+    if not cycles_by_pair:
+        raise SystemExit(
+            "No triangular cycles found in supplied replay events")
+
+    grid = build_parameter_grid(
+        theta_values=_parse_float_list(args.theta_values),
+        trade_capital_values=_parse_float_list(args.trade_capital_values),
+        pair_universes=_parse_pair_universes(args.pair_universes),
+        staleness_threshold_values=_parse_float_list(
+            args.staleness_threshold_values),
+    )
+
+    artifacts = run_parameter_search(
+        events=events,
+        cycles_by_pair=cycles_by_pair,
+        parameter_grid=grid,
+        starting_balances=_parse_balances(args.starting_balances),
+        train_ratio=args.train_ratio,
+        promotion_criteria=PromotionCriteria(
+            min_test_realized_pnl_usd=args.min_test_realized_pnl_usd,
+            min_test_win_rate=args.min_test_win_rate,
+            min_test_fill_rate=args.min_test_fill_rate,
+            max_test_drawdown_usd=args.max_test_drawdown_usd,
+            max_generalization_gap_ratio=args.max_generalization_gap_ratio,
+        ),
+    )
+
+    persist_sweep_results(args.output, artifacts)
+
+    print(f"Completed sweep combinations: {len(artifacts.results)}")
+    print(f"Promotion-ready combinations: {len(artifacts.promoted)}")
+    print(f"Results written: {args.output}")
+
+    _print_top_results(artifacts.results)
+    if artifacts.promoted:
+        print("Promotion candidates (paper-trading canary):")
+        _print_top_results(artifacts.promoted)
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())