"""Paper-trading runner Phase 3 — forward-test virtuale BTC + ETH. Loop infinito (o limitato via --max-ticks) che ogni ``--poll-seconds``: 1. fetch OHLCV 1h ultime ~500 barre via Cerbero 2. per ogni strategia: compile + esegui ultimo bar 3. apply segnale al portfolio multi-asset 4. snapshot equity in DB I bar 1h chiudono al minuto :00. Il loop riconosce un "nuovo bar chiuso" confrontando l'ultimo timestamp del DataFrame con quello dell'iterazione precedente. Tick consecutivi su stesso bar = hold (no double-trade). Esempio: uv run python scripts/run_paper_trading.py \ --name phase3-papertrade-001 \ --initial-capital 1000 \ --max-ticks 336 # 2 settimane * 24 ore """ from __future__ import annotations import argparse import importlib.resources import time from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path from multi_swarm_core.cerbero.client import CerberoClient from multi_swarm_core.config import load_settings from multi_swarm_core.data.cerbero_ohlcv import CerberoOHLCVLoader, OHLCVRequest from strategy_crypto.backend import PaperExecutor, PaperRepository, Portfolio PROJECT_ROOT = Path(__file__).resolve().parent.parent # Mapping timeframe stringa Cerbero -> minuti del bar. Le strategie tradano # sul "bar appena chiuso", quindi end deve essere snappato al boundary del # loro timeframe (NON sempre al top dell'ora) per evitare la regressione in # cui ETH 5m veniva valutato una volta sola ogni 60 min. _TIMEFRAME_MINUTES: dict[str, int] = { "1m": 1, "5m": 5, "15m": 15, "30m": 30, "1h": 60, "4h": 240, "1d": 1440, } def _align_end_to_timeframe(now: datetime, timeframe: str) -> datetime: """Snap ``now`` al boundary del bar timeframe (UTC, naive seconds). Es.: now=14:37:42, tf="5m" -> 14:35:00 now=14:37:42, tf="1h" -> 14:00:00 now=14:00:00, tf="1h" -> 14:00:00 """ bar_min = _TIMEFRAME_MINUTES[timeframe] aligned = now.replace(second=0, microsecond=0) if bar_min >= 1440: return aligned.replace(hour=0, minute=0) total_min = aligned.hour * 60 + aligned.minute snapped = (total_min // bar_min) * bar_min return aligned.replace(hour=snapped // 60, minute=snapped % 60) def _default_strategies_dir() -> Path: """Cartella JSON shippata col package strategy_crypto.""" return Path(str(importlib.resources.files("strategy_crypto") / "strategies")) @dataclass(frozen=True) class AssetConfig: symbol: str # es. "BTC-PERPETUAL" strategy_file: Path exchange: str = "deribit" timeframe: str = "1h" def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="Paper-trading runner Phase 3") p.add_argument("--name", default="phase3-papertrade-001") p.add_argument("--initial-capital", type=float, default=1000.0) p.add_argument("--fees-bp", type=float, default=5.0) p.add_argument("--poll-seconds", type=int, default=300, help="Polling interval (5min default)") p.add_argument("--max-ticks", type=int, default=0, help="0 = infinito; per smoke test usa 1") p.add_argument("--lookback-bars", type=int, default=500, help="Quante bar fetchare per indicatori") p.add_argument( "--strategies-dir", default=str(_default_strategies_dir()), help="Cartella contenente btc_*.json e eth_*.json (default: package strategy_crypto/strategies)", ) return p.parse_args() def load_assets(strategies_dir: Path) -> list[AssetConfig]: btc_files = sorted(strategies_dir.glob("btc_*.json")) eth_files = sorted(strategies_dir.glob("eth_*.json")) if not btc_files or not eth_files: raise FileNotFoundError( f"Expected btc_*.json and eth_*.json in {strategies_dir}" ) # ETH winner c04dff7086 e' tunato su 5m: a 1h la strategia perde (cum_ret -33% 7y). # BTC winner 238e4812 e' tunato su 1h: tick native = paper tick. return [ AssetConfig(symbol="BTC-PERPETUAL", strategy_file=btc_files[0], timeframe="1h"), AssetConfig(symbol="ETH-PERPETUAL", strategy_file=eth_files[0], timeframe="5m"), ] def main() -> None: args = parse_args() settings = load_settings() token = ( settings.cerbero_mainnet_token.get_secret_value() if settings.cerbero_mainnet_token else settings.cerbero_testnet_token.get_secret_value() ) cerbero = CerberoClient( base_url=settings.cerbero_base_url, token=token, bot_tag=settings.cerbero_bot_tag, ) loader = CerberoOHLCVLoader(client=cerbero, cache_dir=settings.series_dir) assets = load_assets(Path(args.strategies_dir)) executors: list[PaperExecutor] = [ PaperExecutor(strategy_json_path=a.strategy_file, symbol=a.symbol) for a in assets ] print(f"Loaded {len(assets)} strategies:") for a, ex in zip(assets, executors, strict=True): print(f" {a.symbol}: {a.strategy_file.name} -> {len(ex._strategy.rules)} rules") portfolio = Portfolio( initial_capital=args.initial_capital, fees_bp=args.fees_bp, n_sleeves=len(assets), ) repo = PaperRepository(settings.strategy_crypto_db_path) repo.init_schema() config = { "assets": [ {"symbol": a.symbol, "strategy": a.strategy_file.name, "exchange": a.exchange} for a in assets ], "fees_bp": args.fees_bp, "poll_seconds": args.poll_seconds, "lookback_bars": args.lookback_bars, } run_id = repo.create_run( name=args.name, initial_capital=args.initial_capital, config=config ) print(f"Paper run started: {run_id} ({args.name})") print(f" initial_capital=${args.initial_capital:.2f}, sleeve=${portfolio.sleeve_capital:.2f}") tick_count = 0 last_bars_seen: dict[str, datetime] = {} try: while args.max_ticks == 0 or tick_count < args.max_ticks: now = datetime.now(UTC) last_prices: dict[str, float] = {} for asset, executor in zip(assets, executors, strict=True): # fetch OHLCV most recent lookback bars: end snappato al timeframe # dell'asset, non sempre all'ora (altrimenti ETH 5m veniva valutato # solo ogni 60 min, regressione vs backtest tunato 5m). bar_min = _TIMEFRAME_MINUTES[asset.timeframe] end = _align_end_to_timeframe(now, asset.timeframe) start = end - timedelta(minutes=bar_min * (args.lookback_bars + 1)) req = OHLCVRequest( symbol=asset.symbol, timeframe=asset.timeframe, start=start, end=end, exchange=asset.exchange, ) # bypass cache for live data try: ohlcv = loader._fetch(req) # noqa: SLF001 except Exception as e: # noqa: BLE001 print(f"[{now.isoformat()}] {asset.symbol} fetch FAIL: {e}") continue if len(ohlcv) < 10: print(f"[{now.isoformat()}] {asset.symbol} too few bars ({len(ohlcv)})") continue bar_ts = ohlcv.index[-1] last_bar_dt = bar_ts.to_pydatetime() if hasattr(bar_ts, "to_pydatetime") else bar_ts # skip se barra gia' processata in questo tick if last_bars_seen.get(asset.symbol) == last_bar_dt: last_prices[asset.symbol] = float(ohlcv["close"].iloc[-1]) continue last_bars_seen[asset.symbol] = last_bar_dt result = executor.execute_tick(portfolio, ohlcv, now) repo.save_tick(run_id, result) last_prices[asset.symbol] = result.close_price if result.action_taken != "hold": pnl_str = ( f"pnl=${result.trade.net_pnl:+.2f}" if result.trade else "" ) print( f"[{now.isoformat()}] {asset.symbol} bar={last_bar_dt} " f"close={result.close_price:.2f} signal={result.signal.value} " f"action={result.action_taken} {pnl_str}" ) repo.sync_open_positions(run_id, portfolio) eq, pos_val = portfolio.equity(last_prices) repo.save_equity_snapshot(run_id, now, eq, portfolio.cash, pos_val) tick_count += 1 print( f"[{now.isoformat()}] tick={tick_count} " f"equity=${eq:.2f} cash=${portfolio.cash:.2f} pos_val=${pos_val:.2f} " f"open={list(portfolio.positions.keys())}" ) if args.max_ticks > 0 and tick_count >= args.max_ticks: break time.sleep(args.poll_seconds) repo.stop_run(run_id, status="completed") except KeyboardInterrupt: print("\nInterrupted by user") repo.stop_run(run_id, status="interrupted") except Exception as e: # noqa: BLE001 print(f"Run failed: {e}") repo.stop_run(run_id, status="failed") raise print(f"Paper run {run_id} stopped after {tick_count} ticks") print(f"Final equity: ${portfolio.equity({})[0]:.2f}") print(f"Trades closed: {len(portfolio.closed_trades)}") if __name__ == "__main__": main()