From 0b8943c926fbc0bb18597aeacee822b85b1fa456 Mon Sep 17 00:00:00 2001 From: ramseshk Date: Thu, 6 Aug 2026 03:12:10 +0000 Subject: [PATCH] PostgreSQL persistence layer + seen_fills fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New: strategies/persistence.py Tables: strategies_snap, trade_log, equity_history, fill_tracker Auto-creates on first use, batches inserts per tick Fix: seen_fills loads from PG (not 2000 API fills) Before: every restart loaded all 2000 fills from API into seen_fills, blocking new fills with matching TIDs for ~20min After: only loads last 100 from API + full history from PG. New fills saved to PG immediately - survives restarts. Live node integration: - write_metrics() → save_strategies() every tick - On fill → save_trade() to trade_log - On fill → TID saved to fill_tracker for cross-restart dedup --- live/node.py | 29 ++++++- strategies/persistence.py | 173 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 200 insertions(+), 2 deletions(-) create mode 100644 strategies/persistence.py diff --git a/live/node.py b/live/node.py index 31da664..49ad1d7 100644 --- a/live/node.py +++ b/live/node.py @@ -45,6 +45,7 @@ trades_log: list[dict] = [] equity_history: list[dict] = [] strategy_equity: dict[str, list] = {} seen_fills: set[int] = set() +_fill_persist_queue: set[int] = set() # New fills to save to PG btc_prices: deque = deque(maxlen=60) eth_prices: deque = deque(maxlen=60) active_cloids: dict = {} # Track active order IDs per strategy @@ -92,6 +93,14 @@ def get_orderbook(coin): except: return 0,0,0 def write_metrics(addr): + try: + from strategies.persistence import save_strategies, save_fill_tids + save_strategies(STRATEGIES) + if _fill_persist_queue: + save_fill_tids(_fill_persist_queue) + _fill_persist_queue.clear() + except Exception: + pass total_pnl = sum(s["pnl"] for s in STRATEGIES.values()) total_pnl_pct = (total_pnl/TOTAL_EQUITY)*100 if TOTAL_EQUITY>0 else 0 for s in STRATEGIES.values(): @@ -274,8 +283,19 @@ async def main(): log.info(f"Cleared {len(open_ords)} stale orders") existing = get_fills(addr) - for f in existing: seen_fills.add(f.get("tid",0)) - log.info(f"Tracking {len(seen_fills)} existing fills") + # Load seen_fills from PG persistence (not API — prevents blocking new fills) + try: + from strategies.persistence import load_fill_tracker + persisted = load_fill_tracker() + seen_fills.update(persisted) + if persisted: + log.info(f"Loaded {len(persisted)} fill TIDs from PG") + except Exception as e: + log.warning(f"PG persistence not available: {e}") + # Fallback: load recent fills from API + for f in existing[-500:]: # Only last 500 fills (not all 2000) + seen_fills.add(f.get("tid",0)) + log.info(f"Tracking {len(seen_fills)} fills ({len(persisted) if 'persisted' in dir() else 0} from PG, {min(len(existing),500)} from API)") for s in STRATEGIES.values(): s["status"]="running" for name in STRATEGIES: strategy_equity[name]=[] @@ -298,6 +318,7 @@ async def main(): tid=f.get("tid",0) if tid in seen_fills: continue seen_fills.add(tid) + _fill_persist_queue.add(tid) # Queue for PG persistence side=f.get("side",""); sz=float(f.get("sz",0)); px=float(f.get("px",0)) closed_pnl=float(f.get("closedPnl",0)); fee=float(f.get("fee","0")) @@ -316,6 +337,10 @@ async def main(): STRATEGIES[strat]["pnl_pct"]=STRATEGIES[strat]["pnl"]/STRATEGIES[strat]["allocation"]*100 strategy_equity[strat].append({"t":time.time(),"v":STRATEGIES[strat]["allocation"]+STRATEGIES[strat]["pnl"]}) trades_log.append({"time":datetime.now().strftime("%H:%M:%S"),"strategy":strat,"side":"BUY" if side=="B" else "SELL","size":sz,"price":px,"pnl":round(net,4),"fee":round(abs(fee),4)}) + try: + from strategies.persistence import save_trade + save_trade(strat, side, sz, px, closed_pnl, float(fee), tid, reason or "") + except Exception: pass new_fills+=1 # Signals every 5 ticks diff --git a/strategies/persistence.py b/strategies/persistence.py new file mode 100644 index 0000000..42f99d0 --- /dev/null +++ b/strategies/persistence.py @@ -0,0 +1,173 @@ +""" +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]