195 lines
8 KiB
Python
195 lines
8 KiB
Python
 | #!/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
|