from __future__ import annotations import json import sqlite3 import threading import uuid from contextlib import closing from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any def _decimal(value: object) -> Decimal: try: return Decimal(str(value)) except (InvalidOperation, TypeError, ValueError): return Decimal("0") def _text(value: Decimal, places: int = 8) -> str: return format(value, f".{places}f").rstrip("0").rstrip(".") or "0" class PaperTradeStore: """Forward-only paper trades. This class has no Binance client dependency.""" def __init__(self, database_file: Path) -> None: self._database_file = database_file self._lock = threading.Lock() def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self._database_file, timeout=15) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute("PRAGMA busy_timeout = 15000") connection.execute("PRAGMA journal_mode = WAL") self._ensure_schema(connection) return connection @staticmethod def _ensure_schema(connection: sqlite3.Connection) -> None: connection.executescript( """ CREATE TABLE IF NOT EXISTS paper_trades ( id TEXT PRIMARY KEY, signal_id TEXT NOT NULL UNIQUE, symbol TEXT NOT NULL, status TEXT NOT NULL, opened_at TEXT NOT NULL, closed_at TEXT, entry_price TEXT NOT NULL, effective_entry_price TEXT NOT NULL, last_price TEXT NOT NULL, quantity TEXT NOT NULL, notional_usdt TEXT NOT NULL, stop_loss TEXT NOT NULL, current_stop TEXT NOT NULL, target_1 TEXT NOT NULL, target_2 TEXT NOT NULL, target_1_hit_at TEXT, remaining_fraction TEXT NOT NULL, realized_gross_pnl_usdt TEXT NOT NULL, fees_usdt TEXT NOT NULL, net_pnl_usdt TEXT, return_percent TEXT, max_favorable_excursion_percent TEXT NOT NULL, max_adverse_excursion_percent TEXT NOT NULL, exit_reason TEXT, exit_price TEXT ); CREATE INDEX IF NOT EXISTS idx_paper_trades_status ON paper_trades(status); CREATE INDEX IF NOT EXISTS idx_paper_trades_opened ON paper_trades(opened_at DESC); CREATE TABLE IF NOT EXISTS paper_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, paper_trade_id TEXT NOT NULL, event_type TEXT NOT NULL, created_at TEXT NOT NULL, price TEXT NOT NULL, payload_json TEXT NOT NULL, FOREIGN KEY(paper_trade_id) REFERENCES paper_trades(id) ); """ ) @staticmethod def _public_trade(row: sqlite3.Row) -> dict[str, Any]: return {key: row[key] for key in row.keys()} @staticmethod def _event( connection: sqlite3.Connection, trade_id: str, event_type: str, now_text: str, price: Decimal, payload: dict[str, Any] | None = None, ) -> dict[str, Any]: event = { "paper_trade_id": trade_id, "event": event_type, "created_at": now_text, "price": _text(price), **(payload or {}), } connection.execute( "INSERT INTO paper_events(paper_trade_id, event_type, created_at, price, payload_json) VALUES (?, ?, ?, ?, ?)", (trade_id, event_type, now_text, _text(price), json.dumps(event, ensure_ascii=False)), ) return event def process( self, analyses: list[dict[str, Any]], signal_state: dict[str, Any], *, now: datetime | None = None, enabled: bool = True, notional_usdt: Decimal = Decimal("100"), fee_bps: Decimal = Decimal("10"), slippage_bps: Decimal = Decimal("5"), move_stop_to_break_even: bool = True, ) -> dict[str, Any]: current_time = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) now_text = current_time.isoformat() if not enabled: return {"enabled": False, "events": [], "open_trades": [], "summary": {}} fee_rate = max(Decimal("0"), fee_bps) / Decimal("10000") slippage_rate = max(Decimal("0"), slippage_bps) / Decimal("10000") safe_notional = max(Decimal("1"), notional_usdt) analyses_by_symbol = { str(item.get("symbol", "")): item for item in analyses if item.get("symbol") } active_signals = { str(item.get("id", "")): item for item in signal_state.get("active_signals", []) } signal_events = signal_state.get("events", []) lifecycle_events = { str(item.get("signal_id", "")): str(item.get("event", "")) for item in signal_events } emitted: list[dict[str, Any]] = [] with self._lock, closing(self._connect()) as connection: for event in signal_events: if event.get("event") != "NEW_STRONG_SIGNAL": continue signal_id = str(event.get("signal_id", "")) signal = active_signals.get(signal_id) if not signal: continue existing = connection.execute( "SELECT id FROM paper_trades WHERE signal_id = ?", (signal_id,) ).fetchone() if existing is not None: continue entry = _decimal(signal.get("current_price")) if entry <= 0: continue effective_entry = entry * (Decimal("1") + slippage_rate) quantity = safe_notional / effective_entry entry_fee = safe_notional * fee_rate trade_id = uuid.uuid4().hex connection.execute( """ INSERT INTO paper_trades( id, signal_id, symbol, status, opened_at, entry_price, effective_entry_price, last_price, quantity, notional_usdt, stop_loss, current_stop, target_1, target_2, remaining_fraction, realized_gross_pnl_usdt, fees_usdt, max_favorable_excursion_percent, max_adverse_excursion_percent ) VALUES (?, ?, ?, 'OPEN', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '1', '0', ?, '0', '0') """, ( trade_id, signal_id, signal.get("symbol"), now_text, _text(entry), _text(effective_entry), _text(entry), _text(quantity), _text(safe_notional, 2), signal.get("stop_loss"), signal.get("stop_loss"), signal.get("target_1"), signal.get("target_2"), _text(entry_fee, 8), ), ) emitted.append( self._event(connection, trade_id, "PAPER_TRADE_OPENED", now_text, entry) ) open_rows = connection.execute( "SELECT * FROM paper_trades WHERE status = 'OPEN' ORDER BY opened_at" ).fetchall() for original_row in open_rows: row = original_row symbol = str(row["symbol"]) analysis = analyses_by_symbol.get(symbol, {}) price = _decimal(analysis.get("current_price")) or _decimal(row["last_price"]) if price <= 0: continue entry = _decimal(row["entry_price"]) effective_entry = _decimal(row["effective_entry_price"]) quantity = _decimal(row["quantity"]) favorable = ((price - entry) / entry) * Decimal("100") if entry > 0 else Decimal("0") mfe = max(_decimal(row["max_favorable_excursion_percent"]), favorable) mae = min(_decimal(row["max_adverse_excursion_percent"]), favorable) connection.execute( "UPDATE paper_trades SET last_price = ?, max_favorable_excursion_percent = ?, max_adverse_excursion_percent = ? WHERE id = ?", (_text(price), _text(mfe, 4), _text(mae, 4), row["id"]), ) target_1 = _decimal(row["target_1"]) target_2 = _decimal(row["target_2"]) current_stop = _decimal(row["current_stop"]) remaining = _decimal(row["remaining_fraction"]) trade_notional = _decimal(row["notional_usdt"]) gross = _decimal(row["realized_gross_pnl_usdt"]) fees = _decimal(row["fees_usdt"]) target_1_hit = row["target_1_hit_at"] is not None if not target_1_hit and price >= target_1: exit_price = target_1 * (Decimal("1") - slippage_rate) fraction = min(Decimal("0.5"), remaining) proceeds = exit_price * quantity * fraction gross += (exit_price - effective_entry) * quantity * fraction fees += proceeds * fee_rate remaining -= fraction current_stop = entry if move_stop_to_break_even else current_stop connection.execute( """ UPDATE paper_trades SET target_1_hit_at = ?, remaining_fraction = ?, realized_gross_pnl_usdt = ?, fees_usdt = ?, current_stop = ? WHERE id = ? """, ( now_text, _text(remaining, 4), _text(gross, 8), _text(fees, 8), _text(current_stop), row["id"], ), ) emitted.append( self._event(connection, row["id"], "PAPER_TARGET_1", now_text, target_1) ) target_1_hit = True close_reason: str | None = None close_price = price if target_1_hit and price >= target_2: close_reason = "TARGET_2" close_price = target_2 elif price <= current_stop: close_reason = "STOP_AFTER_TARGET_1" if target_1_hit else "STOP_LOSS" close_price = price else: lifecycle = lifecycle_events.get(str(row["signal_id"])) if lifecycle == "SIGNAL_INVALIDATED": close_reason = "SIGNAL_INVALIDATED" elif lifecycle == "SIGNAL_EXPIRED": close_reason = "SIGNAL_EXPIRED" if close_reason is not None and remaining > 0: effective_exit = close_price * (Decimal("1") - slippage_rate) proceeds = effective_exit * quantity * remaining gross += (effective_exit - effective_entry) * quantity * remaining fees += proceeds * fee_rate net = gross - fees return_percent = (net / trade_notional) * Decimal("100") if trade_notional > 0 else Decimal("0") status = "CLOSED_" + close_reason connection.execute( """ UPDATE paper_trades SET status = ?, closed_at = ?, remaining_fraction = '0', realized_gross_pnl_usdt = ?, fees_usdt = ?, net_pnl_usdt = ?, return_percent = ?, exit_reason = ?, exit_price = ?, last_price = ? WHERE id = ? """, ( status, now_text, _text(gross, 8), _text(fees, 8), _text(net, 8), _text(return_percent, 4), close_reason, _text(close_price), _text(price), row["id"], ), ) emitted.append( self._event( connection, row["id"], "PAPER_TRADE_CLOSED", now_text, close_price, {"reason": close_reason, "net_pnl_usdt": _text(net, 8)}, ) ) open_trades = connection.execute( "SELECT * FROM paper_trades WHERE status = 'OPEN' ORDER BY opened_at DESC" ).fetchall() closed_trades = connection.execute( "SELECT * FROM paper_trades WHERE status != 'OPEN' ORDER BY closed_at DESC" ).fetchall() closed_count = len(closed_trades) wins = sum(1 for row in closed_trades if _decimal(row["net_pnl_usdt"]) > 0) total_net = sum((_decimal(row["net_pnl_usdt"]) for row in closed_trades), Decimal("0")) average_return = ( sum((_decimal(row["return_percent"]) for row in closed_trades), Decimal("0")) / Decimal(closed_count) if closed_count else Decimal("0") ) connection.commit() return { "enabled": True, "events": emitted, "open_trades": [self._public_trade(row) for row in open_trades], "recent_closed_trades": [self._public_trade(row) for row in closed_trades[:100]], "summary": { "open_count": len(open_trades), "closed_count": closed_count, "wins": wins, "win_rate_percent": format((Decimal(wins) / Decimal(closed_count)) * Decimal("100"), ".2f") if closed_count else "0.00", "total_net_pnl_usdt": _text(total_net, 4), "average_return_percent": _text(average_return, 4), }, "assumptions": { "notional_usdt": _text(safe_notional, 2), "fee_bps_each_side": _text(fee_bps, 2), "slippage_bps_each_side": _text(slippage_bps, 2), "target_1_fraction": "0.50", "move_stop_to_break_even_after_target_1": move_stop_to_break_even, "price_sampling": "BACKGROUND_SCAN_CURRENT_PRICE", }, } def recent(self, limit: int = 100) -> list[dict[str, Any]]: safe_limit = max(1, min(int(limit), 500)) with self._lock, closing(self._connect()) as connection: rows = connection.execute( "SELECT * FROM paper_trades ORDER BY opened_at DESC LIMIT ?", (safe_limit,) ).fetchall() return [self._public_trade(row) for row in rows]