from __future__ import annotations import json import sqlite3 import time from dataclasses import dataclass from decimal import Decimal from pathlib import Path from typing import Any from .models import OrderResult, Signal AUTO_BUY_ATTEMPT_STATUSES = ( "reserved", "buy_submitted", "protected", "protection_failed", "rollback_completed", "rollback_failed", ) AUTO_SELL_ATTEMPT_STATUSES = ( "reserved", "sell_submitted", "sold", "sell_failed", ) @dataclass(frozen=True) class AutoBuyReservation: run_id: int | None approved: bool reason: str @dataclass(frozen=True) class AutoSellReservation: run_id: int | None approved: bool reason: str class Storage: def __init__(self, path: str | Path) -> None: self.path = Path(path) def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self.path) connection.execute("PRAGMA foreign_keys = ON") connection.row_factory = sqlite3.Row return connection def initialize(self) -> None: if self.path.parent != Path("."): self.path.parent.mkdir(parents=True, exist_ok=True) with self._connect() as connection: connection.executescript( """ CREATE TABLE IF NOT EXISTS signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, symbol TEXT NOT NULL, signal_type TEXT NOT NULL, side TEXT, reason TEXT NOT NULL, confidence TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS risk_decisions ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, symbol TEXT NOT NULL, signal_type TEXT NOT NULL, side TEXT, approved INTEGER NOT NULL, reason TEXT NOT NULL, quote_amount TEXT ); CREATE TABLE IF NOT EXISTS order_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, symbol TEXT NOT NULL, side TEXT NOT NULL, client_order_id TEXT NOT NULL, status TEXT NOT NULL, dry_run INTEGER NOT NULL, raw_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS runtime_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, level TEXT NOT NULL, message TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS auto_buy_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, symbol TEXT, status TEXT NOT NULL, score TEXT NOT NULL, reason TEXT NOT NULL, quote_amount TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS auto_buy_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, run_id INTEGER NOT NULL, event_type TEXT NOT NULL, symbol TEXT, client_order_id TEXT, status TEXT, raw_json TEXT NOT NULL, FOREIGN KEY(run_id) REFERENCES auto_buy_runs(id) ); CREATE TABLE IF NOT EXISTS auto_sell_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, symbol TEXT, status TEXT NOT NULL, score TEXT NOT NULL, reason TEXT NOT NULL, quantity TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS auto_sell_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, run_id INTEGER NOT NULL, event_type TEXT NOT NULL, symbol TEXT, client_order_id TEXT, status TEXT, raw_json TEXT NOT NULL, FOREIGN KEY(run_id) REFERENCES auto_sell_runs(id) ); CREATE TABLE IF NOT EXISTS ai_trade_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, symbol TEXT, action TEXT NOT NULL, confidence TEXT NOT NULL, status TEXT NOT NULL, reason TEXT NOT NULL, auto_buy_run_id INTEGER, auto_sell_run_id INTEGER, raw_json TEXT NOT NULL, FOREIGN KEY(auto_buy_run_id) REFERENCES auto_buy_runs(id), FOREIGN KEY(auto_sell_run_id) REFERENCES auto_sell_runs(id) ); CREATE TABLE IF NOT EXISTS spot_trade_fills ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, trade_id INTEGER NOT NULL, order_id INTEGER NOT NULL, order_list_id INTEGER, created_at_ms INTEGER NOT NULL, side TEXT NOT NULL, price TEXT NOT NULL, quantity TEXT NOT NULL, quote_quantity TEXT NOT NULL, commission TEXT NOT NULL, commission_asset TEXT NOT NULL, is_maker INTEGER NOT NULL, is_best_match INTEGER NOT NULL, raw_json TEXT NOT NULL, UNIQUE(symbol, trade_id) ); CREATE TABLE IF NOT EXISTS futures_ai_trade_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, symbol TEXT, action TEXT NOT NULL, confidence TEXT NOT NULL, status TEXT NOT NULL, reason TEXT NOT NULL, leverage INTEGER, margin_amount TEXT, stop_loss_price TEXT, take_profit_price TEXT, close_percent INTEGER, raw_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS futures_order_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, run_id INTEGER NOT NULL, event_type TEXT NOT NULL, symbol TEXT, client_order_id TEXT, client_algo_id TEXT, status TEXT, raw_json TEXT NOT NULL, FOREIGN KEY(run_id) REFERENCES futures_ai_trade_runs(id) ); CREATE TABLE IF NOT EXISTS futures_position_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at_ms INTEGER NOT NULL, run_id INTEGER NOT NULL, symbol TEXT NOT NULL, side TEXT NOT NULL, quantity TEXT NOT NULL, entry_price TEXT NOT NULL, mark_price TEXT NOT NULL, liquidation_price TEXT NOT NULL, isolated_margin TEXT NOT NULL, unrealized_pnl TEXT NOT NULL, FOREIGN KEY(run_id) REFERENCES futures_ai_trade_runs(id) ); """ ) def record_signal(self, signal: Signal) -> int: created_at_ms = signal.created_at_ms if signal.created_at_ms is not None else _now_ms() with self._connect() as connection: cursor = connection.execute( """ INSERT INTO signals (created_at_ms, symbol, signal_type, side, reason, confidence) VALUES (?, ?, ?, ?, ?, ?) """, ( created_at_ms, signal.symbol, signal.signal_type.value, signal.side.value if signal.side else None, signal.reason, str(signal.confidence), ), ) return int(cursor.lastrowid) def record_risk_decision(self, signal: Signal, approved: bool, reason: str, quote_amount: Decimal | None) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO risk_decisions (created_at_ms, symbol, signal_type, side, approved, reason, quote_amount) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( signal.created_at_ms if signal.created_at_ms is not None else _now_ms(), signal.symbol, signal.signal_type.value, signal.side.value if signal.side else None, 1 if approved else 0, reason, str(quote_amount) if quote_amount is not None else None, ), ) return int(cursor.lastrowid) def record_order_result(self, result: OrderResult) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO order_results (created_at_ms, symbol, side, client_order_id, status, dry_run, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( _now_ms(), result.symbol, result.side.value, result.client_order_id, result.status, 1 if result.dry_run else 0, json.dumps(result.raw, sort_keys=True, default=str), ), ) return int(cursor.lastrowid) def record_auto_buy_run( self, symbol: str | None, status: str, score: Decimal, reason: str, quote_amount: Decimal, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO auto_buy_runs (created_at_ms, symbol, status, score, reason, quote_amount) VALUES (?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), symbol.upper() if symbol else None, status, str(score), reason, str(quote_amount), ), ) return int(cursor.lastrowid) def update_auto_buy_run_status(self, run_id: int, status: str, reason: str) -> None: with self._connect() as connection: connection.execute( "UPDATE auto_buy_runs SET status = ?, reason = ? WHERE id = ?", (status, reason, run_id), ) def reserve_auto_buy_attempt( self, symbol: str, score: Decimal, reason: str, quote_amount: Decimal, daily_limit: int, day_start_ms: int, recent_since_ms: int, created_at_ms: int | None = None, ) -> AutoBuyReservation: placeholders = _auto_buy_attempt_status_placeholders() with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") daily_row = connection.execute( f""" SELECT COUNT(*) AS count FROM auto_buy_runs WHERE created_at_ms >= ? AND status IN ({placeholders}) """, (day_start_ms, *AUTO_BUY_ATTEMPT_STATUSES), ).fetchone() if int(daily_row["count"]) >= daily_limit: return AutoBuyReservation(None, False, "daily auto-buy limit reached") recent_row = connection.execute( f""" SELECT 1 FROM auto_buy_runs WHERE symbol = ? AND created_at_ms >= ? AND status IN ({placeholders}) LIMIT 1 """, (symbol.upper(), recent_since_ms, *AUTO_BUY_ATTEMPT_STATUSES), ).fetchone() if recent_row is not None: return AutoBuyReservation(None, False, "symbol cooldown active") cursor = connection.execute( """ INSERT INTO auto_buy_runs (created_at_ms, symbol, status, score, reason, quote_amount) VALUES (?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), symbol.upper(), "reserved", str(score), reason, str(quote_amount), ), ) return AutoBuyReservation(int(cursor.lastrowid), True, "reserved") def record_auto_buy_event( self, run_id: int, event_type: str, symbol: str | None, client_order_id: str | None, status: str | None, raw: dict, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO auto_buy_events (created_at_ms, run_id, event_type, symbol, client_order_id, status, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), run_id, event_type, symbol.upper() if symbol else None, client_order_id, status, json.dumps(raw, sort_keys=True, default=str), ), ) return int(cursor.lastrowid) def record_auto_sell_run( self, symbol: str | None, status: str, score: Decimal, reason: str, quantity: Decimal, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO auto_sell_runs (created_at_ms, symbol, status, score, reason, quantity) VALUES (?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), symbol.upper() if symbol else None, status, str(score), reason, str(quantity), ), ) return int(cursor.lastrowid) def update_auto_sell_run_status(self, run_id: int, status: str, reason: str) -> None: with self._connect() as connection: connection.execute( "UPDATE auto_sell_runs SET status = ?, reason = ? WHERE id = ?", (status, reason, run_id), ) def reserve_auto_sell_attempt( self, symbol: str, score: Decimal, reason: str, quantity: Decimal, daily_limit: int, day_start_ms: int, recent_since_ms: int, created_at_ms: int | None = None, ) -> AutoSellReservation: placeholders = _auto_sell_attempt_status_placeholders() with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") daily_row = connection.execute( f""" SELECT COUNT(*) AS count FROM auto_sell_runs WHERE created_at_ms >= ? AND status IN ({placeholders}) """, (day_start_ms, *AUTO_SELL_ATTEMPT_STATUSES), ).fetchone() if int(daily_row["count"]) >= daily_limit: return AutoSellReservation(None, False, "daily auto-sell limit reached") recent_row = connection.execute( f""" SELECT 1 FROM auto_sell_runs WHERE symbol = ? AND created_at_ms >= ? AND status IN ({placeholders}) LIMIT 1 """, (symbol.upper(), recent_since_ms, *AUTO_SELL_ATTEMPT_STATUSES), ).fetchone() if recent_row is not None: return AutoSellReservation(None, False, "symbol cooldown active") cursor = connection.execute( """ INSERT INTO auto_sell_runs (created_at_ms, symbol, status, score, reason, quantity) VALUES (?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), symbol.upper(), "reserved", str(score), reason, str(quantity), ), ) return AutoSellReservation(int(cursor.lastrowid), True, "reserved") def record_auto_sell_event( self, run_id: int, event_type: str, symbol: str | None, client_order_id: str | None, status: str | None, raw: dict, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO auto_sell_events (created_at_ms, run_id, event_type, symbol, client_order_id, status, raw_json) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), run_id, event_type, symbol.upper() if symbol else None, client_order_id, status, json.dumps(raw, sort_keys=True, default=str), ), ) return int(cursor.lastrowid) def record_ai_trade_run( self, symbol: str | None, action: str, confidence: Decimal, status: str, reason: str, raw: dict | None = None, auto_buy_run_id: int | None = None, auto_sell_run_id: int | None = None, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO ai_trade_runs ( created_at_ms, symbol, action, confidence, status, reason, auto_buy_run_id, auto_sell_run_id, raw_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), symbol.upper() if symbol else None, action.upper(), str(confidence), status, reason, auto_buy_run_id, auto_sell_run_id, json.dumps(raw or {}, sort_keys=True, default=str), ), ) return int(cursor.lastrowid) def record_spot_trade_fill(self, symbol: str, trade: dict[str, Any]) -> bool: normalized_symbol = symbol.upper() trade_id = int(trade["id"]) order_id = int(trade["orderId"]) order_list_id = trade.get("orderListId") side = "BUY" if bool(trade["isBuyer"]) else "SELL" is_maker = 1 if bool(trade.get("isMaker")) else 0 is_best_match = 1 if bool(trade.get("isBestMatch")) else 0 with self._connect() as connection: cursor = connection.execute( """ INSERT OR IGNORE INTO spot_trade_fills ( symbol, trade_id, order_id, order_list_id, created_at_ms, side, price, quantity, quote_quantity, commission, commission_asset, is_maker, is_best_match, raw_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( normalized_symbol, trade_id, order_id, int(order_list_id) if order_list_id is not None else None, int(trade["time"]), side, str(trade["price"]), str(trade["qty"]), str(trade["quoteQty"]), str(trade["commission"]), str(trade["commissionAsset"]).upper(), is_maker, is_best_match, json.dumps(trade, sort_keys=True, default=str), ), ) return cursor.rowcount == 1 def latest_spot_trade_time_ms(self, symbol: str) -> int | None: with self._connect() as connection: row = connection.execute( "SELECT MAX(created_at_ms) AS created_at_ms FROM spot_trade_fills WHERE symbol = ?", (symbol.upper(),), ).fetchone() if row is None or row["created_at_ms"] is None: return None return int(row["created_at_ms"]) def list_spot_trade_fills(self, symbol: str | None = None, limit: int = 50) -> list[dict[str, Any]]: if limit < 0: raise ValueError("limit must be non-negative") with self._connect() as connection: if symbol is None: rows = connection.execute( "SELECT * FROM spot_trade_fills ORDER BY created_at_ms DESC, trade_id DESC LIMIT ?", (limit,), ).fetchall() else: rows = connection.execute( """ SELECT * FROM spot_trade_fills WHERE symbol = ? ORDER BY created_at_ms DESC, trade_id DESC LIMIT ? """, (symbol.upper(), limit), ).fetchall() return [dict(row) for row in rows] def update_ai_trade_run_status( self, run_id: int, status: str, reason: str, raw: dict | None = None, auto_buy_run_id: int | None = None, auto_sell_run_id: int | None = None, ) -> None: with self._connect() as connection: connection.execute( """ UPDATE ai_trade_runs SET status = ?, reason = ?, auto_buy_run_id = COALESCE(?, auto_buy_run_id), auto_sell_run_id = COALESCE(?, auto_sell_run_id), raw_json = COALESCE(?, raw_json) WHERE id = ? """, ( status, reason, auto_buy_run_id, auto_sell_run_id, json.dumps(raw, sort_keys=True, default=str) if raw is not None else None, run_id, ), ) def record_futures_ai_trade_run( self, symbol: str | None, action: str, confidence: Decimal, status: str, reason: str, raw: dict | None = None, leverage: int | None = None, margin_amount: Decimal | None = None, stop_loss_price: Decimal | None = None, take_profit_price: Decimal | None = None, close_percent: int | None = None, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO futures_ai_trade_runs ( created_at_ms, symbol, action, confidence, status, reason, leverage, margin_amount, stop_loss_price, take_profit_price, close_percent, raw_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), symbol.upper() if symbol else None, action.upper(), str(confidence), status, reason, leverage, str(margin_amount) if margin_amount is not None else None, str(stop_loss_price) if stop_loss_price is not None else None, str(take_profit_price) if take_profit_price is not None else None, close_percent, json.dumps(raw or {}, sort_keys=True, default=str), ), ) return int(cursor.lastrowid) def update_futures_ai_trade_run_status( self, run_id: int, status: str, reason: str, raw: dict | None = None, ) -> None: with self._connect() as connection: connection.execute( """ UPDATE futures_ai_trade_runs SET status = ?, reason = ?, raw_json = COALESCE(?, raw_json) WHERE id = ? """, ( status, reason, json.dumps(raw, sort_keys=True, default=str) if raw is not None else None, run_id, ), ) def record_futures_order_event( self, run_id: int, event_type: str, symbol: str | None, client_order_id: str | None, client_algo_id: str | None, status: str | None, raw: dict, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO futures_order_events ( created_at_ms, run_id, event_type, symbol, client_order_id, client_algo_id, status, raw_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), run_id, event_type, symbol.upper() if symbol else None, client_order_id, client_algo_id, status, json.dumps(raw, sort_keys=True, default=str), ), ) return int(cursor.lastrowid) def record_futures_position_snapshot( self, run_id: int, symbol: str, side: str, quantity: Decimal, entry_price: Decimal, mark_price: Decimal, liquidation_price: Decimal, isolated_margin: Decimal, unrealized_pnl: Decimal, created_at_ms: int | None = None, ) -> int: with self._connect() as connection: cursor = connection.execute( """ INSERT INTO futures_position_snapshots ( created_at_ms, run_id, symbol, side, quantity, entry_price, mark_price, liquidation_price, isolated_margin, unrealized_pnl ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( created_at_ms if created_at_ms is not None else _now_ms(), run_id, symbol.upper(), side.upper(), str(quantity), str(entry_price), str(mark_price), str(liquidation_price), str(isolated_margin), str(unrealized_pnl), ), ) return int(cursor.lastrowid) def list_orders(self, limit: int = 50) -> list[dict[str, Any]]: if limit < 0: raise ValueError("limit must be non-negative") with self._connect() as connection: rows = connection.execute( "SELECT * FROM order_results ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [dict(row) for row in rows] def list_auto_buy_runs(self, limit: int = 50) -> list[dict[str, Any]]: if limit < 0: raise ValueError("limit must be non-negative") with self._connect() as connection: rows = connection.execute( "SELECT * FROM auto_buy_runs ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [dict(row) for row in rows] def list_auto_buy_events(self, run_id: int) -> list[dict[str, Any]]: with self._connect() as connection: rows = connection.execute( "SELECT * FROM auto_buy_events WHERE run_id = ? ORDER BY id", (run_id,), ).fetchall() return [dict(row) for row in rows] def list_auto_sell_runs(self, limit: int = 50) -> list[dict[str, Any]]: if limit < 0: raise ValueError("limit must be non-negative") with self._connect() as connection: rows = connection.execute( "SELECT * FROM auto_sell_runs ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [dict(row) for row in rows] def list_auto_sell_events(self, run_id: int) -> list[dict[str, Any]]: with self._connect() as connection: rows = connection.execute( "SELECT * FROM auto_sell_events WHERE run_id = ? ORDER BY id", (run_id,), ).fetchall() return [dict(row) for row in rows] def list_ai_trade_runs(self, limit: int = 50) -> list[dict[str, Any]]: if limit < 0: raise ValueError("limit must be non-negative") with self._connect() as connection: rows = connection.execute( "SELECT * FROM ai_trade_runs ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [dict(row) for row in rows] def list_futures_ai_trade_runs(self, limit: int = 50) -> list[dict[str, Any]]: if limit < 0: raise ValueError("limit must be non-negative") with self._connect() as connection: rows = connection.execute( "SELECT * FROM futures_ai_trade_runs ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [dict(row) for row in rows] def list_futures_order_events(self, run_id: int) -> list[dict[str, Any]]: with self._connect() as connection: rows = connection.execute( "SELECT * FROM futures_order_events WHERE run_id = ? ORDER BY id", (run_id,), ).fetchall() return [dict(row) for row in rows] def list_futures_position_snapshots(self, run_id: int) -> list[dict[str, Any]]: with self._connect() as connection: rows = connection.execute( "SELECT * FROM futures_position_snapshots WHERE run_id = ? ORDER BY id", (run_id,), ).fetchall() return [dict(row) for row in rows] def has_recent_futures_close(self, symbol: str, since_ms: int) -> bool: with self._connect() as connection: row = connection.execute( """ SELECT 1 FROM futures_ai_trade_runs WHERE symbol = ? AND action = 'CLOSE' AND status IN ('closed', 'partially_closed') AND created_at_ms >= ? LIMIT 1 """, (symbol.upper(), since_ms), ).fetchone() return row is not None def count_auto_buy_attempts_since(self, since_ms: int) -> int: placeholders = _auto_buy_attempt_status_placeholders() with self._connect() as connection: row = connection.execute( f""" SELECT COUNT(*) AS count FROM auto_buy_runs WHERE created_at_ms >= ? AND status IN ({placeholders}) """, (since_ms, *AUTO_BUY_ATTEMPT_STATUSES), ).fetchone() return int(row["count"]) def has_recent_auto_buy_attempt(self, symbol: str, since_ms: int) -> bool: placeholders = _auto_buy_attempt_status_placeholders() with self._connect() as connection: row = connection.execute( f""" SELECT 1 FROM auto_buy_runs WHERE symbol = ? AND created_at_ms >= ? AND status IN ({placeholders}) LIMIT 1 """, (symbol.upper(), since_ms, *AUTO_BUY_ATTEMPT_STATUSES), ).fetchone() return row is not None def count_auto_sell_attempts_since(self, since_ms: int) -> int: placeholders = _auto_sell_attempt_status_placeholders() with self._connect() as connection: row = connection.execute( f""" SELECT COUNT(*) AS count FROM auto_sell_runs WHERE created_at_ms >= ? AND status IN ({placeholders}) """, (since_ms, *AUTO_SELL_ATTEMPT_STATUSES), ).fetchone() return int(row["count"]) def has_recent_auto_sell_attempt(self, symbol: str, since_ms: int) -> bool: placeholders = _auto_sell_attempt_status_placeholders() with self._connect() as connection: row = connection.execute( f""" SELECT 1 FROM auto_sell_runs WHERE symbol = ? AND created_at_ms >= ? AND status IN ({placeholders}) LIMIT 1 """, (symbol.upper(), since_ms, *AUTO_SELL_ATTEMPT_STATUSES), ).fetchone() return row is not None def _auto_buy_attempt_status_placeholders() -> str: return ", ".join("?" for _ in AUTO_BUY_ATTEMPT_STATUSES) def _auto_sell_attempt_status_placeholders() -> str: return ", ".join("?" for _ in AUTO_SELL_ATTEMPT_STATUSES) def _now_ms() -> int: return int(time.time() * 1000)