729 lines
28 KiB
Python
729 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import time
|
|
from contextlib import contextmanager
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
from typing import Any, Iterator
|
|
|
|
from crypto_spot_bot.models import Position, Signal, Trade, utc_now
|
|
|
|
|
|
MAX_SIGNAL_DIAGNOSTICS_BYTES = 4 * 1024
|
|
PRUNE_BATCH_SIZE = 5000
|
|
MAX_RUNTIME_ROWS = {
|
|
"signals": 50_000,
|
|
"equity": 100_000,
|
|
"events": 20_000,
|
|
"llm_advice": 20_000,
|
|
}
|
|
_STORED_FORECAST_KEYS = {
|
|
"enabled",
|
|
"usable",
|
|
"model",
|
|
"volatility_model",
|
|
"expected_return_percent",
|
|
"expected_price",
|
|
"volatility_percent",
|
|
"probability_up",
|
|
"confidence_adjustment",
|
|
"block_entry",
|
|
"validation_mae_percent",
|
|
"baseline_mae_percent",
|
|
"skill",
|
|
"horizon",
|
|
"reason",
|
|
"expected_gross_return_percent",
|
|
"quantile_10_percent",
|
|
"quantile_50_percent",
|
|
"quantile_90_percent",
|
|
"conservative_return_percent",
|
|
"target_transform",
|
|
"horizon_forecasts",
|
|
"candidates",
|
|
"quality_gate_passed",
|
|
"model_created_at",
|
|
"model_age_hours",
|
|
"model_fresh",
|
|
}
|
|
|
|
|
|
class Storage:
|
|
def __init__(self, path: str | Path):
|
|
self.path = Path(path)
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._last_hold_signal: dict[tuple[str, str], float] = {}
|
|
self.init_schema()
|
|
|
|
@contextmanager
|
|
def connect(self) -> Iterator[sqlite3.Connection]:
|
|
conn = sqlite3.connect(self.path)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA busy_timeout=5000")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
try:
|
|
yield conn
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def init_schema(self) -> None:
|
|
with self.connect() as conn:
|
|
# New runtime databases reclaim deleted telemetry pages incrementally.
|
|
# Existing databases keep their current mode until compacted once.
|
|
conn.execute("PRAGMA auto_vacuum=INCREMENTAL")
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
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 '{}',
|
|
protective_order_id TEXT NOT NULL DEFAULT '',
|
|
protective_order_link_id TEXT NOT NULL DEFAULT '',
|
|
mode TEXT NOT NULL DEFAULT 'paper',
|
|
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,
|
|
mode TEXT NOT NULL DEFAULT 'paper'
|
|
);
|
|
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,
|
|
mode TEXT NOT NULL DEFAULT 'paper'
|
|
);
|
|
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
|
|
);
|
|
CREATE TABLE IF NOT EXISTS orders (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
client_order_id TEXT NOT NULL UNIQUE,
|
|
exchange_order_id TEXT NOT NULL DEFAULT '',
|
|
symbol TEXT NOT NULL,
|
|
side TEXT NOT NULL,
|
|
order_kind TEXT NOT NULL DEFAULT 'MARKET',
|
|
status TEXT NOT NULL,
|
|
requested_qty REAL NOT NULL DEFAULT 0,
|
|
requested_notional REAL NOT NULL DEFAULT 0,
|
|
executed_qty REAL NOT NULL DEFAULT 0,
|
|
executed_value REAL NOT NULL DEFAULT 0,
|
|
fee_usdt REAL NOT NULL DEFAULT 0,
|
|
raw_json TEXT NOT NULL DEFAULT '{}',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_positions_status_opened
|
|
ON positions(status, opened_at);
|
|
CREATE INDEX IF NOT EXISTS idx_trades_closed
|
|
ON trades(side, closed_at, id DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_signals_symbol_created
|
|
ON signals(symbol, created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_equity_created
|
|
ON equity(created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_events_created
|
|
ON events(created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_orders_status_updated
|
|
ON orders(status, updated_at DESC);
|
|
"""
|
|
)
|
|
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 '{}'",
|
|
"protective_order_id": "TEXT NOT NULL DEFAULT ''",
|
|
"protective_order_link_id": "TEXT NOT NULL DEFAULT ''",
|
|
"mode": "TEXT NOT NULL DEFAULT 'paper'",
|
|
}.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 '{}'",
|
|
"mode": "TEXT NOT NULL DEFAULT 'paper'",
|
|
}.items():
|
|
if column not in trade_columns:
|
|
conn.execute(f"ALTER TABLE trades ADD COLUMN {column} {definition}")
|
|
equity_columns = {
|
|
row["name"]
|
|
for row in conn.execute("PRAGMA table_info(equity)").fetchall()
|
|
}
|
|
if "mode" not in equity_columns:
|
|
conn.execute("ALTER TABLE equity ADD COLUMN mode TEXT NOT NULL DEFAULT 'paper'")
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_equity_mode_created ON equity(mode, created_at DESC)"
|
|
)
|
|
|
|
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,
|
|
protective_order_id, protective_order_link_id, mode, 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),
|
|
position.protective_order_id,
|
|
position.protective_order_link_id,
|
|
position.mode,
|
|
),
|
|
)
|
|
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 update_position_protective_order(
|
|
self,
|
|
position_id: int,
|
|
order_id: str,
|
|
order_link_id: str,
|
|
) -> None:
|
|
with self.connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
UPDATE positions
|
|
SET protective_order_id=?, protective_order_link_id=?
|
|
WHERE id=? AND status='OPEN'
|
|
""",
|
|
(order_id, order_link_id, position_id),
|
|
)
|
|
|
|
def update_position_after_partial_sell(
|
|
self,
|
|
position_id: int,
|
|
*,
|
|
qty: float,
|
|
notional_usdt: float,
|
|
entry_fee_usdt: float,
|
|
) -> None:
|
|
with self.connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
UPDATE positions
|
|
SET qty=?, notional_usdt=?, entry_fee_usdt=?,
|
|
protective_order_id='', protective_order_link_id=''
|
|
WHERE id=? AND status='OPEN'
|
|
""",
|
|
(qty, notional_usdt, entry_fee_usdt, position_id),
|
|
)
|
|
|
|
def open_positions(self, mode: str | None = None) -> list[Position]:
|
|
with self.connect() as conn:
|
|
if mode:
|
|
rows = conn.execute(
|
|
"SELECT * FROM positions WHERE status='OPEN' AND mode=? ORDER BY opened_at",
|
|
(mode,),
|
|
).fetchall()
|
|
else:
|
|
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"], {}),
|
|
protective_order_id=row["protective_order_id"],
|
|
protective_order_link_id=row["protective_order_link_id"],
|
|
mode=row["mode"],
|
|
)
|
|
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
|
|
, mode
|
|
) 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,
|
|
trade.mode,
|
|
),
|
|
)
|
|
return int(cur.lastrowid)
|
|
|
|
def recent_trades(self, limit: int = 50, mode: str | None = None) -> list[dict[str, Any]]:
|
|
with self.connect() as conn:
|
|
if mode:
|
|
rows = conn.execute(
|
|
"SELECT * FROM trades WHERE mode=? ORDER BY id DESC LIMIT ?",
|
|
(mode, limit),
|
|
).fetchall()
|
|
else:
|
|
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, mode: str | None = None) -> list[dict[str, Any]]:
|
|
with self.connect() as conn:
|
|
query = """
|
|
SELECT * FROM trades
|
|
WHERE side='SELL' AND closed_at IS NOT NULL
|
|
"""
|
|
params: tuple[Any, ...]
|
|
if mode:
|
|
query += " AND mode=?"
|
|
params = (mode, limit)
|
|
else:
|
|
params = (limit,)
|
|
query += " ORDER BY id DESC LIMIT ?"
|
|
rows = conn.execute(query, params).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def closed_trade_summary(self, mode: str | None = None) -> dict[str, Any]:
|
|
with self.connect() as conn:
|
|
query = """
|
|
SELECT
|
|
COUNT(*) AS trades,
|
|
COALESCE(SUM(net_pnl), 0) AS net_pnl,
|
|
COALESCE(SUM(gross_pnl), 0) AS gross_pnl,
|
|
COALESCE(SUM(fee_usdt), 0) AS fee_usdt,
|
|
COALESCE(SUM(CASE WHEN net_pnl > 0 THEN 1 ELSE 0 END), 0) AS wins,
|
|
COALESCE(SUM(CASE WHEN net_pnl < 0 THEN 1 ELSE 0 END), 0) AS losses
|
|
FROM trades
|
|
WHERE side='SELL' AND closed_at IS NOT NULL
|
|
"""
|
|
params: tuple[Any, ...] = ()
|
|
if mode:
|
|
query += " AND mode=?"
|
|
params = (mode,)
|
|
row = conn.execute(query, params).fetchone()
|
|
trades = int(row["trades"] if row else 0)
|
|
wins = int(row["wins"] if row else 0)
|
|
losses = int(row["losses"] if row else 0)
|
|
return {
|
|
"trades": trades,
|
|
"net_pnl": round(float(row["net_pnl"] if row else 0.0), 6),
|
|
"gross_pnl": round(float(row["gross_pnl"] if row else 0.0), 6),
|
|
"fee_usdt": round(float(row["fee_usdt"] if row else 0.0), 6),
|
|
"wins": wins,
|
|
"losses": losses,
|
|
"win_rate": round(wins / trades, 4) if trades else 0.0,
|
|
}
|
|
|
|
def insert_signal(self, signal: Signal, hold_sample_seconds: int = 0) -> bool:
|
|
if signal.action == "HOLD" and hold_sample_seconds > 0:
|
|
fingerprint = f"{signal.action}\0{signal.reason}"
|
|
now = time.monotonic()
|
|
sample_key = (signal.symbol, fingerprint)
|
|
previous = self._last_hold_signal.get(sample_key)
|
|
if previous is not None and now - previous < hold_sample_seconds:
|
|
return False
|
|
self._last_hold_signal[sample_key] = now
|
|
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,
|
|
_signal_diagnostics_json(signal.diagnostics),
|
|
signal.created_at.isoformat(),
|
|
),
|
|
)
|
|
return True
|
|
|
|
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,
|
|
mode: str = "paper",
|
|
) -> None:
|
|
with self.connect() as conn:
|
|
conn.execute(
|
|
"INSERT INTO equity (equity, cash, exposure, drawdown, created_at, mode) VALUES (?, ?, ?, ?, ?, ?)",
|
|
(equity, cash, exposure, drawdown, utc_now().isoformat(), mode),
|
|
)
|
|
|
|
def latest_equity(self, mode: str | None = None) -> dict[str, Any] | None:
|
|
with self.connect() as conn:
|
|
if mode:
|
|
row = conn.execute(
|
|
"SELECT * FROM equity WHERE mode=? ORDER BY id DESC LIMIT 1",
|
|
(mode,),
|
|
).fetchone()
|
|
else:
|
|
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 upsert_order(
|
|
self,
|
|
*,
|
|
client_order_id: str,
|
|
exchange_order_id: str = "",
|
|
symbol: str,
|
|
side: str,
|
|
order_kind: str,
|
|
status: str,
|
|
requested_qty: float = 0.0,
|
|
requested_notional: float = 0.0,
|
|
executed_qty: float = 0.0,
|
|
executed_value: float = 0.0,
|
|
fee_usdt: float = 0.0,
|
|
raw: dict[str, Any] | None = None,
|
|
) -> None:
|
|
now = utc_now().isoformat()
|
|
with self.connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO orders (
|
|
client_order_id, exchange_order_id, symbol, side, order_kind,
|
|
status, requested_qty, requested_notional, executed_qty,
|
|
executed_value, fee_usdt, raw_json, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(client_order_id) DO UPDATE SET
|
|
exchange_order_id=excluded.exchange_order_id,
|
|
status=excluded.status,
|
|
executed_qty=excluded.executed_qty,
|
|
executed_value=excluded.executed_value,
|
|
fee_usdt=excluded.fee_usdt,
|
|
raw_json=excluded.raw_json,
|
|
updated_at=excluded.updated_at
|
|
""",
|
|
(
|
|
client_order_id,
|
|
exchange_order_id,
|
|
symbol,
|
|
side,
|
|
order_kind,
|
|
status,
|
|
requested_qty,
|
|
requested_notional,
|
|
executed_qty,
|
|
executed_value,
|
|
fee_usdt,
|
|
json.dumps(raw or {}, ensure_ascii=False),
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
|
|
def recent_orders(self, limit: int = 100) -> list[dict[str, Any]]:
|
|
with self.connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM orders ORDER BY id DESC LIMIT ?",
|
|
(max(1, min(limit, 500)),),
|
|
).fetchall()
|
|
items = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
item["raw"] = _json_or_default(item.pop("raw_json", "{}"), {})
|
|
items.append(item)
|
|
return items
|
|
|
|
def pending_orders(self) -> list[dict[str, Any]]:
|
|
terminal = ("Filled", "Cancelled", "Rejected", "PartiallyFilledCanceled", "Deactivated")
|
|
placeholders = ",".join("?" for _ in terminal)
|
|
with self.connect() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT * FROM orders WHERE status NOT IN ({placeholders}) ORDER BY id",
|
|
terminal,
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def prune(self, retention_days: int) -> dict[str, int]:
|
|
if retention_days <= 0:
|
|
return {}
|
|
cutoff = (utc_now() - timedelta(days=retention_days)).isoformat()
|
|
deleted: dict[str, int] = {}
|
|
for table in ("signals", "equity", "events", "llm_advice"):
|
|
with self.connect() as conn:
|
|
max_id_row = conn.execute(f"SELECT MAX(id) AS value FROM {table}").fetchone()
|
|
max_id = int(max_id_row["value"] or 0) if max_id_row else 0
|
|
cap_boundary = max(0, max_id - MAX_RUNTIME_ROWS[table])
|
|
removed = 0
|
|
if cap_boundary > 0:
|
|
cursor = conn.execute(
|
|
f"""
|
|
DELETE FROM {table}
|
|
WHERE id IN (
|
|
SELECT id FROM {table}
|
|
WHERE id <= ?
|
|
ORDER BY id
|
|
LIMIT ?
|
|
)
|
|
""",
|
|
(cap_boundary, PRUNE_BATCH_SIZE),
|
|
)
|
|
removed = max(0, int(cursor.rowcount))
|
|
remaining = max(0, PRUNE_BATCH_SIZE - removed)
|
|
if remaining:
|
|
cursor = conn.execute(
|
|
f"""
|
|
DELETE FROM {table}
|
|
WHERE id IN (
|
|
SELECT id FROM {table}
|
|
WHERE created_at < ?
|
|
ORDER BY id
|
|
LIMIT ?
|
|
)
|
|
""",
|
|
(cutoff, remaining),
|
|
)
|
|
removed += max(0, int(cursor.rowcount))
|
|
deleted[table] = removed
|
|
conn.execute("PRAGMA incremental_vacuum(512)")
|
|
return deleted
|
|
|
|
def clear_all(self) -> None:
|
|
with self.connect() as conn:
|
|
for table in ("positions", "trades", "signals", "equity", "events", "runtime", "llm_advice", "orders"):
|
|
conn.execute(f"DELETE FROM {table}")
|
|
|
|
|
|
def _signal_diagnostics_json(diagnostics: dict[str, Any]) -> str:
|
|
compact = dict(diagnostics)
|
|
forecast = compact.get("forecast")
|
|
if isinstance(forecast, dict):
|
|
compact["forecast"] = {
|
|
key: value for key, value in forecast.items() if key in _STORED_FORECAST_KEYS
|
|
}
|
|
encoded = json.dumps(compact, ensure_ascii=False, separators=(",", ":"))
|
|
size = len(encoded.encode("utf-8"))
|
|
if size <= MAX_SIGNAL_DIAGNOSTICS_BYTES:
|
|
return encoded
|
|
|
|
fallback = {
|
|
"truncated": True,
|
|
"original_size_bytes": size,
|
|
"strategy_mode": compact.get("strategy_mode"),
|
|
"trade_mode": compact.get("trade_mode"),
|
|
"checks": compact.get("checks", {}),
|
|
"forecast": compact.get("forecast", {}),
|
|
}
|
|
encoded = json.dumps(fallback, ensure_ascii=False, separators=(",", ":"))
|
|
if len(encoded.encode("utf-8")) <= MAX_SIGNAL_DIAGNOSTICS_BYTES:
|
|
return encoded
|
|
return json.dumps(
|
|
{
|
|
"truncated": True,
|
|
"original_size_bytes": size,
|
|
"strategy_mode": compact.get("strategy_mode"),
|
|
"trade_mode": compact.get("trade_mode"),
|
|
},
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
)
|
|
|
|
|
|
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)
|