You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
308 lines
13 KiB
Python
308 lines
13 KiB
Python
"""SQLite state management for ORB paper trading."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from apps.orb_trader.models import (
|
|
CREATE_TABLES_SQL,
|
|
ORBCandidateRow,
|
|
ORBDailySnapshotRow,
|
|
ORBDailyStateRow,
|
|
ORBPositionRow,
|
|
ORBSessionRow,
|
|
ORBTradeRow,
|
|
)
|
|
|
|
_DEFAULT_DB = "data/paper/orb.db"
|
|
|
|
|
|
class ORBStateManager:
|
|
"""SQLite CRUD layer for ORB paper trading state."""
|
|
|
|
def __init__(self, db_path: str | None = None) -> None:
|
|
self._db_path = db_path or os.environ.get("ORB_TRADER_DB", _DEFAULT_DB)
|
|
Path(self._db_path).parent.mkdir(parents=True, exist_ok=True)
|
|
self._init_db()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(self._db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
return conn
|
|
|
|
def _init_db(self) -> None:
|
|
with self._connect() as conn:
|
|
conn.executescript(CREATE_TABLES_SQL)
|
|
|
|
# ── Sessions ──────────────────────────────────────────────────────────────
|
|
|
|
def create_session(
|
|
self,
|
|
name: str,
|
|
config_path: str,
|
|
initial_equity: float,
|
|
) -> str:
|
|
session_id = str(uuid.uuid4())[:8]
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO sessions (session_id, session_name, config_path, initial_equity, created_at)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(session_id, name, config_path, initial_equity, now),
|
|
)
|
|
return session_id
|
|
|
|
def get_session(self, name_or_id: str) -> ORBSessionRow | None:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM sessions WHERE session_id=? OR session_name=?",
|
|
(name_or_id, name_or_id),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return ORBSessionRow(**dict(row))
|
|
|
|
def list_sessions(self) -> list[ORBSessionRow]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM sessions ORDER BY created_at DESC"
|
|
).fetchall()
|
|
return [ORBSessionRow(**dict(r)) for r in rows]
|
|
|
|
def set_session_status(self, session_id: str, status: str) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"UPDATE sessions SET status=? WHERE session_id=?",
|
|
(status, session_id),
|
|
)
|
|
|
|
def delete_session(self, session_id: str) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute("DELETE FROM orb_candidates WHERE session_id=?", (session_id,))
|
|
conn.execute("DELETE FROM orb_positions WHERE session_id=?", (session_id,))
|
|
conn.execute("DELETE FROM trades WHERE session_id=?", (session_id,))
|
|
conn.execute("DELETE FROM daily_snapshots WHERE session_id=?", (session_id,))
|
|
conn.execute("DELETE FROM daily_state WHERE session_id=?", (session_id,))
|
|
conn.execute("DELETE FROM sessions WHERE session_id=?", (session_id,))
|
|
|
|
# ── Positions ─────────────────────────────────────────────────────────────
|
|
|
|
def save_position(self, pos: ORBPositionRow) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""INSERT OR REPLACE INTO orb_positions
|
|
(session_id, date, ticker, direction, entry_price, entry_time,
|
|
shares, orb_high, orb_low, atr_at_entry, stop_distance, current_stop,
|
|
peak_price, trailing_active, rvol, composite_score, order_id, status)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
pos.session_id, pos.date, pos.ticker, pos.direction,
|
|
pos.entry_price, pos.entry_time, pos.shares,
|
|
pos.orb_high, pos.orb_low, pos.atr_at_entry,
|
|
pos.stop_distance, pos.current_stop, pos.peak_price,
|
|
int(pos.trailing_active), pos.rvol, pos.composite_score,
|
|
pos.order_id, pos.status,
|
|
),
|
|
)
|
|
|
|
def get_open_positions(self, session_id: str, date: str) -> list[ORBPositionRow]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM orb_positions WHERE session_id=? AND date=? AND status='open'",
|
|
(session_id, date),
|
|
).fetchall()
|
|
return [_row_to_position(r) for r in rows]
|
|
|
|
def get_all_open_positions(self, session_id: str) -> list[ORBPositionRow]:
|
|
"""Get all open positions across all dates (for reconciliation)."""
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM orb_positions WHERE session_id=? AND status='open'",
|
|
(session_id,),
|
|
).fetchall()
|
|
return [_row_to_position(r) for r in rows]
|
|
|
|
def update_position_stop(
|
|
self,
|
|
session_id: str,
|
|
date: str,
|
|
ticker: str,
|
|
current_stop: float,
|
|
peak_price: float,
|
|
trailing_active: bool,
|
|
) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""UPDATE orb_positions
|
|
SET current_stop=?, peak_price=?, trailing_active=?
|
|
WHERE session_id=? AND date=? AND ticker=? AND status='open'""",
|
|
(current_stop, peak_price, int(trailing_active), session_id, date, ticker),
|
|
)
|
|
|
|
def close_position_record(self, session_id: str, date: str, ticker: str) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""UPDATE orb_positions SET status='closed'
|
|
WHERE session_id=? AND date=? AND ticker=?""",
|
|
(session_id, date, ticker),
|
|
)
|
|
|
|
# ── Trades ────────────────────────────────────────────────────────────────
|
|
|
|
def save_trade(self, trade: ORBTradeRow) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""INSERT OR REPLACE INTO trades
|
|
(trade_id, session_id, date, ticker, direction, entry_price, exit_price,
|
|
entry_time, exit_time, shares, pnl, r_multiple, exit_reason,
|
|
atr_at_entry, rvol, composite_score)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
trade.trade_id, trade.session_id, trade.date, trade.ticker,
|
|
trade.direction, trade.entry_price, trade.exit_price,
|
|
trade.entry_time, trade.exit_time, trade.shares,
|
|
trade.pnl, trade.r_multiple, trade.exit_reason,
|
|
trade.atr_at_entry, trade.rvol, trade.composite_score,
|
|
),
|
|
)
|
|
|
|
def list_trades(
|
|
self,
|
|
session_id: str,
|
|
limit: int | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
if limit:
|
|
rows = conn.execute(
|
|
"SELECT * FROM trades WHERE session_id=? ORDER BY date DESC, exit_time DESC LIMIT ?",
|
|
(session_id, limit),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT * FROM trades WHERE session_id=? ORDER BY date DESC, exit_time DESC",
|
|
(session_id,),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
# ── Daily Snapshots ───────────────────────────────────────────────────────
|
|
|
|
def save_daily_snapshot(self, snap: ORBDailySnapshotRow) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""INSERT OR REPLACE INTO daily_snapshots
|
|
(session_id, date, equity, daily_pnl, total_pnl, trades_taken, stops_hit, drawdown_pct)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
snap.session_id, snap.date, snap.equity, snap.daily_pnl,
|
|
snap.total_pnl, snap.trades_taken, snap.stops_hit, snap.drawdown_pct,
|
|
),
|
|
)
|
|
|
|
def list_snapshots(self, session_id: str) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM daily_snapshots WHERE session_id=? ORDER BY date",
|
|
(session_id,),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def get_equity(self, session_id: str) -> float | None:
|
|
"""Return the equity from the latest daily snapshot, or None if no snapshots exist."""
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT equity FROM daily_snapshots WHERE session_id=? ORDER BY date DESC LIMIT 1",
|
|
(session_id,),
|
|
).fetchone()
|
|
return float(row[0]) if row else None
|
|
|
|
def get_peak_equity(self, session_id: str, initial_equity: float) -> float:
|
|
"""Return the peak equity across all snapshots (for drawdown calculation)."""
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT MAX(equity) FROM daily_snapshots WHERE session_id=?",
|
|
(session_id,),
|
|
).fetchone()
|
|
return float(row[0]) if row and row[0] is not None else initial_equity
|
|
|
|
# ── Daily State ───────────────────────────────────────────────────────────
|
|
|
|
def get_daily_state(self, session_id: str, date: str) -> ORBDailyStateRow:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM daily_state WHERE session_id=? AND date=?",
|
|
(session_id, date),
|
|
).fetchone()
|
|
if row is None:
|
|
return ORBDailyStateRow(session_id=session_id, date=date)
|
|
return ORBDailyStateRow(
|
|
session_id=row["session_id"],
|
|
date=row["date"],
|
|
cumulative_loss=row["cumulative_loss"],
|
|
stops_hit=row["stops_hit"],
|
|
kill_switch=bool(row["kill_switch"]),
|
|
phase=row["phase"],
|
|
)
|
|
|
|
def update_daily_state(self, session_id: str, date: str, **kwargs: Any) -> None:
|
|
state = self.get_daily_state(session_id, date)
|
|
for k, v in kwargs.items():
|
|
setattr(state, k, v)
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""INSERT OR REPLACE INTO daily_state
|
|
(session_id, date, cumulative_loss, stops_hit, kill_switch, phase)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
state.session_id, state.date, state.cumulative_loss,
|
|
state.stops_hit, int(state.kill_switch), state.phase,
|
|
),
|
|
)
|
|
|
|
# ── Candidates ────────────────────────────────────────────────────────────
|
|
|
|
def save_candidate(self, cand: ORBCandidateRow) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""INSERT OR REPLACE INTO orb_candidates
|
|
(session_id, date, ticker, direction, orb_high, orb_low, breakout_level,
|
|
atr, rvol, gap_pct, composite_score, status)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
cand.session_id, cand.date, cand.ticker, cand.direction,
|
|
cand.orb_high, cand.orb_low, cand.breakout_level,
|
|
cand.atr, cand.rvol, cand.gap_pct, cand.composite_score, cand.status,
|
|
),
|
|
)
|
|
|
|
def update_candidate_status(
|
|
self, session_id: str, date: str, ticker: str, status: str
|
|
) -> None:
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""UPDATE orb_candidates SET status=?
|
|
WHERE session_id=? AND date=? AND ticker=?""",
|
|
(status, session_id, date, ticker),
|
|
)
|
|
|
|
def list_candidates(self, session_id: str, date: str) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM orb_candidates WHERE session_id=? AND date=? ORDER BY composite_score DESC",
|
|
(session_id, date),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _row_to_position(row: sqlite3.Row) -> ORBPositionRow:
|
|
d = dict(row)
|
|
d["trailing_active"] = bool(d.get("trailing_active", 0))
|
|
return ORBPositionRow(**d)
|