arbitrade/scripts/backtest_sweep.py

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