Multi_Swarm_Coevolutive/scripts/run_paper_trading.py

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