ftdt-quant-lab/strategies/persistence.py

"""
FTDT Quant Lab — PostgreSQL Persistence Layer.

Tables:
  strategies_snap  — per-tick strategy state (PnL, position, trades)
  trade_log        — every fill with PnL attribution
  equity_history   — per-strategy equity curve
  fill_tracker     — seen_fills persistence (prevents cross-restart blocking)
"""

import os, json, time
import psycopg2
from datetime import datetime

DB = os.getenv("FTDT_DB", "dbname=ftdt_quant user=ftdt password=ftdt_quant_2024 host=localhost")

def get_conn():
    return psycopg2.connect(DB)

def init_db():
    """Create tables if they don't exist."""
    conn = get_conn()
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS strategies_snap (
            id SERIAL PRIMARY KEY,
            ts TIMESTAMPTZ DEFAULT NOW(),
            name TEXT NOT NULL,
            pnl DOUBLE PRECISION DEFAULT 0,
            pnl_pct DOUBLE PRECISION DEFAULT 0,
            position DOUBLE PRECISION DEFAULT 0,
            trades_today INTEGER DEFAULT 0,
            wins INTEGER DEFAULT 0,
            win_rate DOUBLE PRECISION DEFAULT 0,
            equity DOUBLE PRECISION DEFAULT 100,
            status TEXT DEFAULT 'idle',
            instrument TEXT DEFAULT ''
        );
        CREATE INDEX IF NOT EXISTS idx_strat_name_ts ON strategies_snap(name, ts);
        
        CREATE TABLE IF NOT EXISTS trade_log (
            id SERIAL PRIMARY KEY,
            ts TIMESTAMPTZ DEFAULT NOW(),
            strategy TEXT NOT NULL,
            side TEXT,
            size DOUBLE PRECISION,
            price DOUBLE PRECISION,
            pnl DOUBLE PRECISION DEFAULT 0,
            fee DOUBLE PRECISION DEFAULT 0,
            fill_tid BIGINT,
            reason TEXT DEFAULT ''
        );
        CREATE INDEX IF NOT EXISTS idx_trade_strat_ts ON trade_log(strategy, ts);
        
        CREATE TABLE IF NOT EXISTS equity_history (
            id SERIAL PRIMARY KEY,
            ts TIMESTAMPTZ DEFAULT NOW(),
            strategy TEXT NOT NULL,
            equity DOUBLE PRECISION
        );
        CREATE INDEX IF NOT EXISTS idx_equity_strat_ts ON equity_history(strategy, ts);
        
        CREATE TABLE IF NOT EXISTS fill_tracker (
            tid BIGINT PRIMARY KEY,
            seen_at TIMESTAMPTZ DEFAULT NOW()
        );
    """)
    conn.commit()
    cur.close()
    conn.close()
    return True

def save_strategies(strategies: dict):
    """Save current strategy states to PG."""
    conn = get_conn()
    cur = conn.cursor()
    now = datetime.utcnow()
    for name, s in strategies.items():
        cur.execute(
            "INSERT INTO strategies_snap (ts, name, pnl, pnl_pct, position, trades_today, wins, win_rate, equity, status, instrument) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
            (now, name,
             s.get("pnl", 0), s.get("pnl_pct", 0), s.get("position", 0),
             s.get("trades_today", 0), s.get("wins", 0), s.get("win_rate", 0),
             s.get("allocation", 100) + s.get("pnl", 0),
             s.get("status", "idle"), s.get("instrument", ""))
        )
    conn.commit()
    cur.close()
    conn.close()

def save_trade(strategy: str, side: str, size: float, price: float, pnl: float, fee: float, tid: int, reason: str = ""):
    """Save a single trade fill to PG."""
    conn = get_conn()
    cur = conn.cursor()
    cur.execute(
        "INSERT INTO trade_log (ts, strategy, side, size, price, pnl, fee, fill_tid, reason) "
        "VALUES (NOW(), %s, %s, %s, %s, %s, %s, %s, %s)",
        (strategy, side, size, price, pnl, fee, tid, reason)
    )
    conn.commit()
    cur.close()
    conn.close()

def save_equity(strategy: str, equity: float):
    """Save equity point for a strategy."""
    conn = get_conn()
    cur = conn.cursor()
    cur.execute(
        "INSERT INTO equity_history (ts, strategy, equity) VALUES (NOW(), %s, %s)",
        (strategy, equity)
    )
    conn.commit()
    cur.close()
    conn.close()

# ═══════════ Fill Tracker (seen_fills) ═══════════

def load_fill_tracker() -> set:
    """Load seen_fills from PG — avoids reloading ALL history from API on restart."""
    seen = set()
    try:
        conn = get_conn()
        cur = conn.cursor()
        cur.execute("SELECT tid FROM fill_tracker")
        for row in cur.fetchall():
            seen.add(row[0])
        cur.close()
        conn.close()
    except Exception:
        pass
    return seen

def save_fill_tids(tids: set):
    """Batch save new fill TIDs to PG."""
    if not tids:
        return
    conn = get_conn()
    cur = conn.cursor()
    for tid in tids:
        try:
            cur.execute(
                "INSERT INTO fill_tracker (tid) VALUES (%s) ON CONFLICT (tid) DO NOTHING",
                (tid,)
            )
        except Exception:
            pass
    conn.commit()
    cur.close()
    conn.close()

# ═══════════ Query Helpers ═══════════

def get_trades(strategy: str = None, limit: int = 200):
    conn = get_conn()
    cur = conn.cursor()
    if strategy:
        cur.execute("SELECT * FROM trade_log WHERE strategy=%s ORDER BY ts DESC LIMIT %s", (strategy, limit))
    else:
        cur.execute("SELECT * FROM trade_log ORDER BY ts DESC LIMIT %s", (limit,))
    rows = cur.fetchall()
    cur.close()
    conn.close()
    return rows

def get_equity(strategy: str, limit: int = 500):
    conn = get_conn()
    cur = conn.cursor()
    cur.execute("SELECT ts, equity FROM equity_history WHERE strategy=%s ORDER BY ts ASC LIMIT %s", (strategy, limit))
    rows = cur.fetchall()
    cur.close()
    conn.close()
    return [(str(r[0]), r[1]) for r in rows]