"""SQLite memory: strategy versions, every trade (with the conditions that led to it), equity, events.""" import json import sqlite3 import pandas as pd SCHEMA = """ CREATE TABLE IF NOT EXISTS strategies( version INTEGER PRIMARY KEY AUTOINCREMENT, parent INTEGER, params TEXT, status TEXT, created TEXT, retired TEXT, reason TEXT, start_equity REAL, account INTEGER, backtest TEXT, findings TEXT); CREATE TABLE IF NOT EXISTS trades( id INTEGER PRIMARY KEY AUTOINCREMENT, version INTEGER, side TEXT, status TEXT, entry_time TEXT, entry REAL, qty REAL, sl REAL, tp REAL, risk_amount REAL, confidence INTEGER, factors TEXT, fees REAL, ticket TEXT, exit_time TEXT, exit REAL, pnl REAL, exit_reason TEXT, bars_held INTEGER); CREATE TABLE IF NOT EXISTS equity(ts TEXT, version INTEGER, equity REAL); CREATE TABLE IF NOT EXISTS events(ts TEXT, level TEXT, msg TEXT); CREATE TABLE IF NOT EXISTS brain_log( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT, version INTEGER, kind TEXT, provider TEXT, model TEXT, latency_ms INTEGER, ok INTEGER, applied TEXT, summary TEXT, response TEXT, trade_id INTEGER, error TEXT); CREATE TABLE IF NOT EXISTS research(ts TEXT, brief TEXT, sources TEXT); """ class Journal: def __init__(self, path): self.db = sqlite3.connect(str(path), check_same_thread=False) self.db.row_factory = sqlite3.Row self.db.executescript(SCHEMA) self.db.commit() def _all(self, sql, args=()): return [dict(r) for r in self.db.execute(sql, args).fetchall()] # ---- versions def new_version(self, params, start_equity, parent=None, account=None, findings=None, now=None): now = str(now or pd.Timestamp.now(tz="UTC")) cur = self.db.execute( "INSERT INTO strategies(parent,params,status,created,start_equity,account,findings) VALUES(?,?,?,?,?,?,?)", (parent, json.dumps(params), "active", now, start_equity, account, json.dumps(findings or []))) self.db.commit() return cur.lastrowid def version(self, v): rows = self._all("SELECT * FROM strategies WHERE version=?", (v,)) if not rows: return None r = rows[0] r["params"] = json.loads(r["params"]) r["backtest"] = json.loads(r["backtest"]) if r["backtest"] else None r["findings"] = json.loads(r["findings"]) if r["findings"] else [] return r def active_version(self): rows = self._all("SELECT version FROM strategies WHERE status='active' ORDER BY version DESC LIMIT 1") return self.version(rows[0]["version"]) if rows else None def all_versions(self): return self._all("SELECT version,parent,status,created,retired,reason,account FROM strategies ORDER BY version") def set_start_equity(self, v, eq): self.db.execute("UPDATE strategies SET start_equity=? WHERE version=?", (eq, v)) self.db.commit() def set_backtest(self, v, summary): self.db.execute("UPDATE strategies SET backtest=? WHERE version=?", (json.dumps(summary), v)) self.db.commit() def retire(self, v, reason, now=None, status="retired"): self.db.execute("UPDATE strategies SET status=?, retired=?, reason=? WHERE version=?", (status, str(now or pd.Timestamp.now(tz="UTC")), reason, v)) self.db.commit() # ---- trades def add_trade(self, v, side, entry_time, entry, qty, sl, tp, risk_amount, confidence, factors, fees, ticket=None): cur = self.db.execute( "INSERT INTO trades(version,side,status,entry_time,entry,qty,sl,tp,risk_amount,confidence,factors,fees,ticket)" " VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)", (v, side, "open", str(entry_time), entry, qty, sl, tp, risk_amount, confidence, json.dumps(factors), fees, ticket)) self.db.commit() return cur.lastrowid def close_trade(self, tid, exit_time, exit_price, pnl, fees, reason, bars_held): self.db.execute( "UPDATE trades SET status='closed',exit_time=?,exit=?,pnl=?,fees=?,exit_reason=?,bars_held=? WHERE id=?", (str(exit_time), exit_price, pnl, fees, reason, bars_held, tid)) self.db.commit() def update_trade_sl(self, tid, new_sl): """Persist a stop-loss update (e.g. break-even move) without closing the trade.""" self.db.execute("UPDATE trades SET sl=? WHERE id=?", (new_sl, tid)) self.db.commit() def _trade(self, r): r["factors"] = json.loads(r["factors"]) if r["factors"] else {} return r def current_trade(self, v): rows = self._all("SELECT * FROM trades WHERE version=? AND status='open' ORDER BY id DESC LIMIT 1", (v,)) return self._trade(rows[0]) if rows else None def trades(self, v, status=None): sql, args = "SELECT * FROM trades WHERE version=?", [v] if status: sql += " AND status=?" args.append(status) return [self._trade(r) for r in self._all(sql + " ORDER BY id", args)] # ---- equity / events def log_equity(self, v, equity, ts=None): self.db.execute("INSERT INTO equity VALUES(?,?,?)", (str(ts or pd.Timestamp.now(tz="UTC")), v, equity)) self.db.commit() def equity_series(self, v): return [r["equity"] for r in self._all("SELECT equity FROM equity WHERE version=? ORDER BY rowid", (v,))] def event(self, level, msg, ts=None): self.db.execute("INSERT INTO events VALUES(?,?,?)", (str(ts or pd.Timestamp.now(tz="UTC")), level, msg)) self.db.commit() def events(self, limit=30, level=None): sql, args = "SELECT * FROM events", [] if level: sql += " WHERE level=?" args.append(level) return self._all(sql + " ORDER BY rowid DESC LIMIT ?", args + [limit]) def count_events(self, level, since=None): sql, args = "SELECT COUNT(*) c FROM events WHERE level=?", [level] if since: sql += " AND ts>=?" args.append(str(since)) return self._all(sql, args)[0]["c"] # ---- AI brain def log_brain(self, ts, version, kind, provider, model, latency_ms, ok, applied, summary, response, trade_id=None, error=""): cur = self.db.execute( "INSERT INTO brain_log(ts,version,kind,provider,model,latency_ms,ok,applied,summary,response,trade_id,error)" " VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", (str(ts), version, kind, provider, model, int(latency_ms or 0), 1 if ok else 0, applied, (summary or "")[:900], json.dumps(response, default=str) if response is not None else None, trade_id, (error or "")[:300])) self.db.commit() return cur.lastrowid def attach_brain_trade(self, log_id, trade_id): self.db.execute("UPDATE brain_log SET trade_id=? WHERE id=?", (trade_id, log_id)) self.db.commit() def brain_logs(self, kind=None, limit=20, version=None): sql, args = "SELECT * FROM brain_log WHERE 1=1", [] if kind: sql += " AND kind=?" args.append(kind) if version is not None: sql += " AND version=?" args.append(version) rows = self._all(sql + " ORDER BY id DESC LIMIT ?", args + [limit]) for r in rows: r["response"] = json.loads(r["response"]) if r["response"] else None return rows def brain_calls_today(self): today = str(pd.Timestamp.now(tz="UTC").date()) return self._all("SELECT COUNT(*) c FROM brain_log WHERE ts LIKE ?", (today + "%",))[0]["c"] def save_research(self, ts, brief, sources): self.db.execute("INSERT INTO research VALUES(?,?,?)", (str(ts), json.dumps(brief, default=str), json.dumps(sources, default=str))) self.db.commit() def latest_research(self): rows = self._all("SELECT * FROM research ORDER BY rowid DESC LIMIT 1") if not rows: return None return {"ts": rows[0]["ts"], "brief": json.loads(rows[0]["brief"]), "sources": json.loads(rows[0]["sources"] or "[]")} def brain_trade_outcomes(self): rows = self._all("SELECT b.response r, t.pnl pnl, t.risk_amount ra FROM brain_log b JOIN trades t ON t.id=b.trade_id " "WHERE b.kind='review' AND b.ok=1 AND t.status='closed'") return [{"decision": (json.loads(r["r"]) if r["r"] else {}).get("decision", "APPROVE"), "pnl": r["pnl"], "risk_amount": r["ra"]} for r in rows]