from __future__ import annotations import json import sqlite3 from contextlib import contextmanager from pathlib import Path from typing import Any, Iterator from crypto_spot_bot.models import Position, Signal, Trade, utc_now class Storage: def __init__(self, path: str | Path): self.path = Path(path) self.path.parent.mkdir(parents=True, exist_ok=True) self.init_schema() @contextmanager def connect(self) -> Iterator[sqlite3.Connection]: conn = sqlite3.connect(self.path) conn.row_factory = sqlite3.Row try: yield conn conn.commit() finally: conn.close() def init_schema(self) -> None: with self.connect() as conn: conn.executescript( """ CREATE TABLE IF NOT EXISTS positions ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, qty REAL NOT NULL, entry_price REAL NOT NULL, notional_usdt REAL NOT NULL, entry_fee_usdt REAL NOT NULL DEFAULT 0, stop_loss REAL NOT NULL, take_profit REAL NOT NULL, highest_price REAL NOT NULL, opened_at TEXT NOT NULL, entry_reason TEXT NOT NULL DEFAULT '', entry_confidence REAL NOT NULL DEFAULT 0, entry_pattern TEXT NOT NULL DEFAULT '', entry_diagnostics_json TEXT NOT NULL DEFAULT '{}', status TEXT NOT NULL DEFAULT 'OPEN' ); CREATE TABLE IF NOT EXISTS trades ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, side TEXT NOT NULL, qty REAL NOT NULL, entry_price REAL, exit_price REAL, gross_pnl REAL NOT NULL DEFAULT 0, fee_usdt REAL NOT NULL DEFAULT 0, net_pnl REAL NOT NULL DEFAULT 0, reason TEXT NOT NULL DEFAULT '', entry_pattern TEXT NOT NULL DEFAULT '', entry_confidence REAL NOT NULL DEFAULT 0, entry_diagnostics_json TEXT NOT NULL DEFAULT '{}', opened_at TEXT, closed_at TEXT ); CREATE TABLE IF NOT EXISTS signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, action TEXT NOT NULL, confidence REAL NOT NULL, reason TEXT NOT NULL, diagnostics_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS equity ( id INTEGER PRIMARY KEY AUTOINCREMENT, equity REAL NOT NULL, cash REAL NOT NULL, exposure REAL NOT NULL, drawdown REAL NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, level TEXT NOT NULL, message TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS runtime ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS llm_advice ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, model TEXT NOT NULL, prompt_json TEXT NOT NULL DEFAULT '{}', response_text TEXT NOT NULL DEFAULT '', advice_json TEXT NOT NULL DEFAULT '{}', error TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL ); """ ) columns = { row["name"] for row in conn.execute("PRAGMA table_info(positions)").fetchall() } if "entry_fee_usdt" not in columns: conn.execute( "ALTER TABLE positions ADD COLUMN entry_fee_usdt REAL NOT NULL DEFAULT 0" ) for column, definition in { "entry_reason": "TEXT NOT NULL DEFAULT ''", "entry_confidence": "REAL NOT NULL DEFAULT 0", "entry_pattern": "TEXT NOT NULL DEFAULT ''", "entry_diagnostics_json": "TEXT NOT NULL DEFAULT '{}'", }.items(): if column not in columns: conn.execute(f"ALTER TABLE positions ADD COLUMN {column} {definition}") trade_columns = { row["name"] for row in conn.execute("PRAGMA table_info(trades)").fetchall() } for column, definition in { "entry_pattern": "TEXT NOT NULL DEFAULT ''", "entry_confidence": "REAL NOT NULL DEFAULT 0", "entry_diagnostics_json": "TEXT NOT NULL DEFAULT '{}'", }.items(): if column not in trade_columns: conn.execute(f"ALTER TABLE trades ADD COLUMN {column} {definition}") def insert_position(self, position: Position) -> int: with self.connect() as conn: cur = conn.execute( """ INSERT INTO positions ( symbol, qty, entry_price, notional_usdt, entry_fee_usdt, stop_loss, take_profit, highest_price, opened_at, entry_reason, entry_confidence, entry_pattern, entry_diagnostics_json, status ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'OPEN') """, ( position.symbol, position.qty, position.entry_price, position.notional_usdt, position.entry_fee_usdt, position.stop_loss, position.take_profit, position.highest_price, position.opened_at.isoformat(), position.entry_reason, position.entry_confidence, position.entry_pattern, json.dumps(position.entry_diagnostics, ensure_ascii=False), ), ) return int(cur.lastrowid) def close_position(self, position_id: int) -> None: with self.connect() as conn: conn.execute("UPDATE positions SET status='CLOSED' WHERE id=?", (position_id,)) def update_position_highest(self, position_id: int, highest_price: float) -> None: with self.connect() as conn: conn.execute( "UPDATE positions SET highest_price=? WHERE id=? AND status='OPEN'", (highest_price, position_id), ) def open_positions(self) -> list[Position]: with self.connect() as conn: rows = conn.execute( "SELECT * FROM positions WHERE status='OPEN' ORDER BY opened_at" ).fetchall() return [ Position( id=int(row["id"]), symbol=row["symbol"], qty=float(row["qty"]), entry_price=float(row["entry_price"]), notional_usdt=float(row["notional_usdt"]), entry_fee_usdt=float(row["entry_fee_usdt"]), stop_loss=float(row["stop_loss"]), take_profit=float(row["take_profit"]), highest_price=float(row["highest_price"]), opened_at=_parse_datetime(row["opened_at"]), entry_reason=row["entry_reason"], entry_confidence=float(row["entry_confidence"]), entry_pattern=row["entry_pattern"], entry_diagnostics=_json_or_default(row["entry_diagnostics_json"], {}), ) for row in rows ] def insert_trade(self, trade: Trade) -> int: with self.connect() as conn: cur = conn.execute( """ INSERT INTO trades ( symbol, side, qty, entry_price, exit_price, gross_pnl, fee_usdt, net_pnl, reason, entry_pattern, entry_confidence, entry_diagnostics_json, opened_at, closed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( trade.symbol, trade.side, trade.qty, trade.entry_price, trade.exit_price, trade.gross_pnl, trade.fee_usdt, trade.net_pnl, trade.reason, trade.entry_pattern, trade.entry_confidence, json.dumps(trade.entry_diagnostics, ensure_ascii=False), trade.opened_at.isoformat() if trade.opened_at else None, trade.closed_at.isoformat() if trade.closed_at else None, ), ) return int(cur.lastrowid) def recent_trades(self, limit: int = 50) -> list[dict[str, Any]]: with self.connect() as conn: rows = conn.execute("SELECT * FROM trades ORDER BY id DESC LIMIT ?", (limit,)).fetchall() return [dict(row) for row in rows] def closed_trades(self, limit: int = 200) -> list[dict[str, Any]]: with self.connect() as conn: rows = conn.execute( """ SELECT * FROM trades WHERE side='SELL' AND closed_at IS NOT NULL ORDER BY id DESC LIMIT ? """, (limit,), ).fetchall() return [dict(row) for row in rows] def insert_signal(self, signal: Signal) -> None: with self.connect() as conn: conn.execute( """ INSERT INTO signals (symbol, action, confidence, reason, diagnostics_json, created_at) VALUES (?, ?, ?, ?, ?, ?) """, ( signal.symbol, signal.action, signal.confidence, signal.reason, json.dumps(signal.diagnostics, ensure_ascii=False), signal.created_at.isoformat(), ), ) def recent_signals(self, limit: int = 80) -> list[dict[str, Any]]: with self.connect() as conn: rows = conn.execute("SELECT * FROM signals ORDER BY id DESC LIMIT ?", (limit,)).fetchall() return [dict(row) for row in rows] def insert_equity(self, equity: float, cash: float, exposure: float, drawdown: float) -> None: with self.connect() as conn: conn.execute( "INSERT INTO equity (equity, cash, exposure, drawdown, created_at) VALUES (?, ?, ?, ?, ?)", (equity, cash, exposure, drawdown, utc_now().isoformat()), ) def latest_equity(self) -> dict[str, Any] | None: with self.connect() as conn: row = conn.execute("SELECT * FROM equity ORDER BY id DESC LIMIT 1").fetchone() return dict(row) if row else None def event(self, message: str, level: str = "INFO") -> None: with self.connect() as conn: conn.execute( "INSERT INTO events (level, message, created_at) VALUES (?, ?, ?)", (level, message, utc_now().isoformat()), ) def recent_events(self, limit: int = 80) -> list[dict[str, Any]]: with self.connect() as conn: rows = conn.execute("SELECT * FROM events ORDER BY id DESC LIMIT ?", (limit,)).fetchall() return [dict(row) for row in rows] def insert_llm_advice( self, *, symbol: str, model: str, prompt_json: dict[str, Any], response_text: str, advice_json: dict[str, Any], error: str = "", ) -> None: with self.connect() as conn: conn.execute( """ INSERT INTO llm_advice ( symbol, model, prompt_json, response_text, advice_json, error, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( symbol, model, json.dumps(prompt_json, ensure_ascii=False), response_text, json.dumps(advice_json, ensure_ascii=False), error, utc_now().isoformat(), ), ) def recent_llm_advice(self, limit: int = 80) -> list[dict[str, Any]]: with self.connect() as conn: rows = conn.execute("SELECT * FROM llm_advice ORDER BY id DESC LIMIT ?", (limit,)).fetchall() items: list[dict[str, Any]] = [] for row in rows: item = dict(row) item["prompt"] = _json_or_default(item.pop("prompt_json", "{}"), {}) item["advice"] = _json_or_default(item.pop("advice_json", "{}"), {}) items.append(item) return items def set_runtime(self, key: str, value: Any) -> None: with self.connect() as conn: conn.execute( """ INSERT INTO runtime (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at """, (key, json.dumps(value, ensure_ascii=False), utc_now().isoformat()), ) def get_runtime(self, key: str, default: Any = None) -> Any: with self.connect() as conn: row = conn.execute("SELECT value FROM runtime WHERE key=?", (key,)).fetchone() if not row: return default try: return json.loads(row["value"]) except json.JSONDecodeError: return default def clear_all(self) -> None: with self.connect() as conn: for table in ("positions", "trades", "signals", "equity", "events", "runtime", "llm_advice"): conn.execute(f"DELETE FROM {table}") def _json_or_default(value: str, default: Any) -> Any: try: return json.loads(value) except json.JSONDecodeError: return default def _parse_datetime(value: str): from datetime import datetime return datetime.fromisoformat(value)