feat(strategy_pythagoras): port paper-trading backend (Portfolio, Executor, Repository)

src/strategy_pythagoras/strategy_pythagoras/backend/__init__.py

@@ -0,0 +1,23 @@
+"""Backend paper-trading per la strategia strategy_pythagoras.
+
+Espone le classi principali per import ergonomici in scripts/runner:
+
+    from strategy_pythagoras.backend import PaperExecutor, Portfolio, PaperRepository
+
+Per i tipi interni (TickResult, OpenPosition, Trade) importare dal sotto-modulo.
+"""
+
+from .executor import PaperExecutor, TickResult
+from .persistence import PaperRepository
+from .portfolio import OpenPosition, Portfolio
+from .schema import PAPER_SCHEMA_SQL, init_schema
+
+__all__ = [
+    "PAPER_SCHEMA_SQL",
+    "OpenPosition",
+    "PaperExecutor",
+    "PaperRepository",
+    "Portfolio",
+    "TickResult",
+    "init_schema",
+]

src/strategy_pythagoras/strategy_pythagoras/backend/executor.py

@@ -0,0 +1,97 @@
+"""PaperExecutor: applica un segnale di strategia a un Portfolio.
+
+Il flusso per ogni tick:
+
+    bar OHLCV chiuso -> compile_strategy(strategy) -> Series[Side]
+    -> last_signal = series.iloc[-1]
+    -> match con posizione attuale -> open / close / hold
+
+Niente delay 1-bar: in paper-trading il segnale viene calcolato sulla
+barra appena chiusa e applicato al prezzo close della stessa. La latenza
+reale tra tick e ordine va misurata separatamente (Phase 3 spec).
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+
+import pandas as pd  # type: ignore[import-untyped]
+
+from multi_swarm_core.backtest.orders import Side, Trade
+from multi_swarm_core.protocol.compiler import compile_strategy
+from multi_swarm_core.protocol.parser import parse_strategy
+
+from .portfolio import OpenPosition, Portfolio
+
+
+@dataclass
+class TickResult:
+    ts: datetime
+    symbol: str
+    bar_ts: datetime
+    close_price: float
+    signal: Side
+    action_taken: str  # "open_long" | "open_short" | "close" | "reverse" | "hold"
+    trade: Trade | None = None
+    new_position: OpenPosition | None = None
+
+
+class PaperExecutor:
+    def __init__(self, strategy_json_path: Path, symbol: str) -> None:
+        text = strategy_json_path.read_text()
+        # parse_strategy si aspetta JSON pulito, non fence; il file e' gia' JSON.
+        self._strategy = parse_strategy(text)
+        self._compiled = compile_strategy(self._strategy)
+        self.symbol = symbol
+        self.strategy_path = strategy_json_path
+
+    def execute_tick(
+        self,
+        portfolio: Portfolio,
+        ohlcv: pd.DataFrame,
+        now: datetime,
+    ) -> TickResult:
+        """Esegui un tick: calcola segnale su tutto ``ohlcv`` (per indicatori
+        con lookback), prendi l'ultimo, e applica al portfolio."""
+        if len(ohlcv) == 0:
+            raise ValueError("Empty OHLCV passed to execute_tick")
+        signals = self._compiled(ohlcv)
+        # ultimo bar chiuso
+        bar_ts = ohlcv.index[-1]
+        close_price = float(ohlcv["close"].iloc[-1])
+        signal = Side(signals.iloc[-1]) if signals.iloc[-1] is not None else Side.FLAT
+
+        current = portfolio.positions.get(self.symbol)
+        action = "hold"
+        trade: Trade | None = None
+        new_position: OpenPosition | None = None
+
+        if current is None and signal != Side.FLAT:
+            new_position = portfolio.open(self.symbol, signal, close_price, now)
+            action = f"open_{signal.value}"
+        elif current is not None and signal == Side.FLAT:
+            trade = portfolio.close(self.symbol, close_price, now)
+            action = "close"
+        elif current is not None and signal != current.side:
+            # reverse: chiudi e riapri opposto
+            trade = portfolio.close(self.symbol, close_price, now)
+            new_position = portfolio.open(self.symbol, signal, close_price, now)
+            action = "reverse"
+
+        return TickResult(
+            ts=now,
+            symbol=self.symbol,
+            bar_ts=bar_ts.to_pydatetime() if hasattr(bar_ts, "to_pydatetime") else bar_ts,
+            close_price=close_price,
+            signal=signal,
+            action_taken=action,
+            trade=trade,
+            new_position=new_position,
+        )
+
+    @property
+    def strategy_dict(self) -> dict:
+        return json.loads(self.strategy_path.read_text())

src/strategy_pythagoras/strategy_pythagoras/backend/persistence.py

@@ -0,0 +1,123 @@
+"""Persistenza paper-trading: scrive su un DB dedicato (state/strategy_pythagoras_paper.db)
+con le tabelle ``paper_trading_*`` definite localmente in :mod:`.schema`.
+
+Il DB e' isolato dal ``runs.db`` del core GA: nessun naming conflict con
+future strategie (state/strategy_<asset>.db), nessuna contention di lock
+fra writer GA e writer paper.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from .executor import TickResult
+from .portfolio import Portfolio
+from .schema import init_schema as _init_paper_schema
+
+
+class PaperRepository:
+    def __init__(self, db_path: Path | str):
+        self.db_path = Path(db_path)
+
+    def init_schema(self) -> None:
+        """Crea (se mancanti) le tabelle paper_trading_* su ``self.db_path``."""
+        _init_paper_schema(self.db_path)
+
+    def _conn(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path, isolation_level=None)
+        conn.row_factory = sqlite3.Row
+        conn.execute("PRAGMA foreign_keys = ON")
+        conn.execute("PRAGMA journal_mode = WAL")
+        return conn
+
+    @staticmethod
+    def _now() -> str:
+        return datetime.now(UTC).isoformat()
+
+    def create_run(self, name: str, initial_capital: float, config: dict[str, Any]) -> str:
+        rid = uuid.uuid4().hex
+        with self._conn() as conn:
+            conn.execute(
+                "INSERT INTO paper_trading_runs "
+                "(id, name, started_at, status, initial_capital, config_json) "
+                "VALUES (?,?,?,?,?,?)",
+                (rid, name, self._now(), "running", initial_capital, json.dumps(config)),
+            )
+        return rid
+
+    def stop_run(self, run_id: str, status: str = "stopped") -> None:
+        with self._conn() as conn:
+            conn.execute(
+                "UPDATE paper_trading_runs SET stopped_at=?, status=? WHERE id=?",
+                (self._now(), status, run_id),
+            )
+
+    def save_tick(self, run_id: str, tick: TickResult) -> None:
+        with self._conn() as conn:
+            conn.execute(
+                "INSERT INTO paper_trading_ticks "
+                "(paper_run_id, symbol, ts, bar_ts, close_price, signal, action_taken) "
+                "VALUES (?,?,?,?,?,?,?)",
+                (
+                    run_id,
+                    tick.symbol,
+                    tick.ts.isoformat(),
+                    tick.bar_ts.isoformat() if hasattr(tick.bar_ts, "isoformat") else str(tick.bar_ts),
+                    tick.close_price,
+                    tick.signal.value,
+                    tick.action_taken,
+                ),
+            )
+            if tick.trade is not None:
+                t = tick.trade
+                conn.execute(
+                    "INSERT INTO paper_trading_trades "
+                    "(paper_run_id, symbol, side, qty, entry_price, exit_price, "
+                    "entry_ts, exit_ts, pnl, fees) VALUES (?,?,?,?,?,?,?,?,?,?)",
+                    (
+                        run_id,
+                        tick.symbol,
+                        t.side.value,
+                        t.size,
+                        t.entry_price,
+                        t.exit_price,
+                        t.entry_ts.isoformat(),
+                        t.exit_ts.isoformat(),
+                        t.net_pnl,
+                        t.fees,
+                    ),
+                )
+
+    def save_equity_snapshot(
+        self,
+        run_id: str,
+        ts: datetime,
+        equity: float,
+        cash: float,
+        positions_value: float,
+    ) -> None:
+        with self._conn() as conn:
+            conn.execute(
+                "INSERT INTO paper_trading_equity "
+                "(paper_run_id, ts, equity, cash, positions_value) VALUES (?,?,?,?,?)",
+                (run_id, ts.isoformat(), equity, cash, positions_value),
+            )
+
+    def sync_open_positions(self, run_id: str, portfolio: Portfolio) -> None:
+        """Sostituisce snapshot posizioni aperte. Idempotente: cancella e reinserisce."""
+        with self._conn() as conn:
+            conn.execute(
+                "DELETE FROM paper_trading_positions WHERE paper_run_id=?", (run_id,)
+            )
+            for sym, pos in portfolio.positions.items():
+                conn.execute(
+                    "INSERT INTO paper_trading_positions "
+                    "(paper_run_id, symbol, side, qty, entry_price, entry_ts) "
+                    "VALUES (?,?,?,?,?,?)",
+                    (run_id, sym, pos.side.value, pos.qty, pos.entry_price, pos.entry_ts.isoformat()),
+                )

src/strategy_pythagoras/strategy_pythagoras/backend/portfolio.py

@@ -0,0 +1,104 @@
+"""Portfolio multi-asset per paper-trading.
+
+Modello semplificato: capitale unico ``cash``, allocazione equal-weight
+fra N posizioni (sleeve = 1/N del capitale iniziale per ogni simbolo).
+Niente leva, niente liquidation, fees su entry+exit (bp del notional).
+
+Una :class:`Position` rappresenta una posizione aperta su un singolo
+simbolo (long/short, qty in unita' dell'asset, prezzo di entry). La
+posizione viene chiusa con :meth:`Portfolio.close` che produce un
+:class:`Trade` realized e accredita ``cash``.
+
+Mark-to-market via :meth:`Portfolio.equity`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from multi_swarm_core.backtest.orders import Side, Trade
+
+
+@dataclass(frozen=True)
+class OpenPosition:
+    symbol: str
+    side: Side
+    qty: float
+    entry_price: float
+    entry_ts: datetime
+
+
+@dataclass
+class Portfolio:
+    initial_capital: float
+    fees_bp: float = 5.0
+    n_sleeves: int = 2  # numero strategie / asset previsti
+    cash: float = field(init=False)
+    positions: dict[str, OpenPosition] = field(default_factory=dict)
+    closed_trades: list[Trade] = field(default_factory=list)
+
+    def __post_init__(self) -> None:
+        self.cash = self.initial_capital
+
+    @property
+    def sleeve_capital(self) -> float:
+        return self.initial_capital / self.n_sleeves
+
+    def open(
+        self,
+        symbol: str,
+        side: Side,
+        price: float,
+        ts: datetime,
+    ) -> OpenPosition:
+        if symbol in self.positions:
+            raise ValueError(f"Position already open on {symbol}")
+        if side == Side.FLAT:
+            raise ValueError("Cannot open a FLAT position")
+        # sleeve fisso: alloca 1/n_sleeves del capitale iniziale, qty = notional/price.
+        notional = self.sleeve_capital
+        qty = notional / price
+        fees = notional * (self.fees_bp / 10000.0)
+        self.cash -= fees
+        pos = OpenPosition(symbol=symbol, side=side, qty=qty, entry_price=price, entry_ts=ts)
+        self.positions[symbol] = pos
+        return pos
+
+    def close(
+        self,
+        symbol: str,
+        price: float,
+        ts: datetime,
+    ) -> Trade:
+        if symbol not in self.positions:
+            raise ValueError(f"No open position on {symbol}")
+        pos = self.positions.pop(symbol)
+        trade = Trade(
+            entry_ts=pos.entry_ts,
+            exit_ts=ts,
+            side=pos.side,
+            size=pos.qty,
+            entry_price=pos.entry_price,
+            exit_price=price,
+            fees_bp=self.fees_bp,
+        )
+        # net_pnl include gia' i fees sull'intero round-trip; abbiamo gia'
+        # addebitato meta' fees all'open, ora addebitiamo il resto.
+        self.cash += trade.gross_pnl - (trade.fees / 2.0)
+        self.closed_trades.append(trade)
+        return trade
+
+    def equity(self, last_prices: dict[str, float]) -> tuple[float, float]:
+        """Ritorna (equity_totale, positions_value) marcando posizioni aperte
+        al ``last_prices[symbol]``. Posizioni senza prezzo disponibile valgono
+        notional di entry (fallback conservativo)."""
+        positions_value = 0.0
+        for sym, pos in self.positions.items():
+            price = last_prices.get(sym, pos.entry_price)
+            unreal = pos.qty * (
+                price - pos.entry_price if pos.side == Side.LONG
+                else pos.entry_price - price
+            )
+            positions_value += pos.qty * pos.entry_price + unreal
+        return self.cash + positions_value, positions_value