#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ database/telemetry.py — TelemetryDB | SQLite Persistence Engine (Track B) Captures the Centaur continuous feedback loop: AI score -> Execution (Trade_Opened) -> Trade Result (Trade_Closed) Consumes SDP Trade_Opened / Trade_Closed envelopes from the router and persists them in a single `trades` table (ticket is the natural key). Thread-safety: One shared sqlite3 connection with check_same_thread=False plus a threading.Lock serializing every execute/commit. WAL journal mode keeps the single writer fast and reads non-blocking. Duplicate Trade_Opened (primary-key conflict) and closes without an open record are handled gracefully and logged — they never raise. """ import logging import sqlite3 import threading from typing import Any, Dict, Optional logger = logging.getLogger(__name__) # -------------------------------------------------------------------------- # Schema (DDL) — single robust `trades` table # -------------------------------------------------------------------------- _DDL = """ CREATE TABLE IF NOT EXISTS trades ( ticket INTEGER PRIMARY KEY, -- MT5 position/order ticket symbol TEXT NOT NULL, direction TEXT NOT NULL, -- "buy" | "sell" lot REAL NOT NULL, entry_price REAL NOT NULL, ai_score REAL NOT NULL DEFAULT 0.0, -- advisory score at open open_time TEXT, -- ISO-8601 UTC close_time TEXT, -- ISO-8601 UTC (set on close) profit REAL, -- final PnL in deposit currency r_multiple REAL -- final R-multiple ); """ class TelemetryDB: """Thread-safe SQLite telemetry store for the feedback loop.""" def __init__(self, db_path: str = "centaur_telemetry.db") -> None: self.path = db_path self._lock = threading.Lock() # check_same_thread=False: the router shares ONE connection across # per-client threads; the lock serializes all access to it. self._conn = sqlite3.connect(db_path, check_same_thread=False) with self._lock: self._conn.execute("PRAGMA journal_mode=WAL;") # concurrent readers self._conn.execute("PRAGMA synchronous=NORMAL;") # safe + fast with WAL self._conn.execute(_DDL) self._conn.commit() logger.info("TelemetryDB: connected to %s (WAL on, table 'trades' ready).", db_path) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def log_trade_opened(self, msg: Dict[str, Any]) -> bool: """Insert a record from an SDP Trade_Opened envelope. INSERT OR IGNORE keeps the FIRST record per ticket: a duplicate open (e.g., a replay) must never clobber a later close. """ p = msg.get("payload") or {} ticket = self._extract_ticket(p) if ticket is None: return False try: lot = float(p.get("lot", 0.0) or 0.0) entry = float(p.get("entry_price", 0.0) or 0.0) score = float(msg.get("algorithmic_confidence_score", 0.0) or 0.0) except (TypeError, ValueError): lot = entry = score = 0.0 # non-numeric telemetry -> neutral defaults try: with self._lock: cur = self._conn.execute( "INSERT OR IGNORE INTO trades " "(ticket, symbol, direction, lot, entry_price, ai_score, open_time) " "VALUES (?, ?, ?, ?, ?, ?, ?)", ( ticket, str(msg.get("symbol", "?")), str(p.get("direction", "?")), lot, entry, score, str(msg.get("timestamp", "")), ), ) self._conn.commit() if cur.rowcount == 0: logger.warning("TelemetryDB: ticket %s already open — duplicate ignored.", ticket) return False logger.info("TelemetryDB: opened ticket=%s %s %s lot=%s ai_score=%.2f", ticket, msg.get("symbol", "?"), p.get("direction", "?"), lot, score) return True except sqlite3.Error as exc: # IntegrityError included (PK conflicts) logger.error("TelemetryDB: log_trade_opened failed (ticket=%s): %s", ticket, exc) return False def log_trade_closed(self, msg: Dict[str, Any]) -> bool: """Update the existing record with the final outcome (profit, R, close time).""" p = msg.get("payload") or {} ticket = self._extract_ticket(p) if ticket is None: return False profit = p.get("profit") r_mult = p.get("r_multiple") if profit is None or r_mult is None: logger.warning("TelemetryDB: Trade_Closed ticket=%s missing " "profit/r_multiple — ignored.", ticket) return False try: profit_f = float(profit) r_f = float(r_mult) except (TypeError, ValueError): logger.warning("TelemetryDB: non-numeric outcome for ticket=%s — ignored.", ticket) return False try: with self._lock: cur = self._conn.execute( "UPDATE trades SET profit = ?, r_multiple = ?, close_time = ? " "WHERE ticket = ?", (profit_f, r_f, str(msg.get("timestamp", "")), ticket), ) self._conn.commit() if cur.rowcount == 0: logger.warning("TelemetryDB: no open record for ticket=%s — close ignored.", ticket) return False logger.info("TelemetryDB: closed ticket=%s profit=%.2f r_multiple=%.2f", ticket, profit_f, r_f) return True except sqlite3.Error as exc: # IntegrityError included logger.error("TelemetryDB: log_trade_closed failed (ticket=%s): %s", ticket, exc) return False def fetch_trade(self, ticket: int) -> Optional[Dict[str, Any]]: """Read one record back (used by tests and future calibration jobs).""" with self._lock: row = self._conn.execute( "SELECT ticket, symbol, direction, lot, entry_price, ai_score, " " open_time, close_time, profit, r_multiple " "FROM trades WHERE ticket = ?", (int(ticket),), ).fetchone() if row is None: return None return { "ticket": row[0], "symbol": row[1], "direction": row[2], "lot": row[3], "entry_price": row[4], "ai_score": row[5], "open_time": row[6], "close_time": row[7], "profit": row[8], "r_multiple": row[9], } def close(self) -> None: """Release the connection (call on shutdown).""" with self._lock: try: self._conn.close() except sqlite3.Error as exc: logger.warning("TelemetryDB: close failed: %s", exc) # ------------------------------------------------------------------ # Internals # ------------------------------------------------------------------ @staticmethod def _extract_ticket(payload: Dict[str, Any]) -> Optional[int]: """Return a valid int ticket or None (logged) for malformed frames.""" ticket = payload.get("ticket") if ticket is None: logger.warning("TelemetryDB: frame without 'ticket' — ignored.") return None try: return int(ticket) except (TypeError, ValueError): logger.warning("TelemetryDB: invalid ticket %r — ignored.", ticket) return None