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
|