"""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 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}" ) return [ AssetConfig(symbol="BTC-PERPETUAL", strategy_file=btc_files[0]), AssetConfig(symbol="ETH-PERPETUAL", strategy_file=eth_files[0]), ] 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 = now.replace(minute=0, second=0, microsecond=0) start = end - timedelta(hours=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()