yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add paper trading system: broker integration, state management, reporter

Browse Source 
New modules for live/mock broker interface, SQLite session state,
auto-trading engine, and backtest result reporting.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
main
I Luk Kim 5 months ago
parent 479eb76308
commit 9b92ab6589
7 changed files with 2257 additions and 0 deletions
Show Stats Download Patch File Download Diff File

1
apps/paper_trader/__init__.py
Unescape Escape View File

@ -0,0 +1 @@
"""Paper trading system with Alpaca API."""

360
apps/paper_trader/alpaca_broker.py
Unescape Escape View File

@ -0,0 +1,360 @@
"""Alpaca Paper Trading API wrapper.
Switch from paper to live by setting paper=False (or ALPACA_PAPER=false).
Requires: pip install alpaca-py
"""
from __future__ import annotations
import datetime as dt
import os
from dataclasses import dataclass
from typing import Any
@dataclass
class AccountInfo:
equity: float
cash: float
buying_power: float
long_market_value: float
unrealized_pl: float
portfolio_value: float
@dataclass
class Order:
id: str
symbol: str
qty: int
side: str # "buy" or "sell"
status: str
filled_avg_price: float | None
filled_qty: int
@dataclass
class Position:
symbol: str
qty: int
avg_entry_price: float
current_price: float
unrealized_pl: float
market_value: float
@dataclass
class Bar:
date: str
open: float
high: float
low: float
close: float
volume: float
@dataclass
class PortfolioHistory:
timestamps: list[int]
equity: list[float]
profit_loss: list[float]
profit_loss_pct: list[float]
class AlpacaBroker:
"""Thin wrapper around alpaca-py TradingClient for paper trading."""
def __init__(
self,
api_key: str | None = None,
secret_key: str | None = None,
paper: bool = True,
) -> None:
self._api_key = api_key or os.environ.get("ALPACA_API_KEY", "")
self._secret_key = secret_key or os.environ.get("ALPACA_SECRET_KEY", "")
self._paper = paper
if not self._api_key or not self._secret_key:
raise ValueError(
"Alpaca credentials missing. Set ALPACA_API_KEY and ALPACA_SECRET_KEY env vars."
)
try:
from alpaca.trading.client import TradingClient
from alpaca.data.historical import StockHistoricalDataClient
except ImportError as exc:
raise ImportError("alpaca-py not installed. Run: pip install alpaca-py") from exc
self._trading = TradingClient(
api_key=self._api_key,
secret_key=self._secret_key,
paper=self._paper,
)
self._data = StockHistoricalDataClient(
api_key=self._api_key,
secret_key=self._secret_key,
)
# ------------------------------------------------------------------ #
# Account
# ------------------------------------------------------------------ #
def get_account(self) -> AccountInfo:
acct = self._trading.get_account()
equity = float(acct.equity or 0)
last_equity = float(acct.last_equity or equity)
return AccountInfo(
equity=equity,
cash=float(acct.cash or 0),
buying_power=float(acct.buying_power or 0),
long_market_value=float(acct.long_market_value or 0),
unrealized_pl=equity - last_equity,
portfolio_value=float(acct.portfolio_value or equity),
)
# ------------------------------------------------------------------ #
# Orders
# ------------------------------------------------------------------ #
def submit_market_buy(self, symbol: str, qty: int) -> Order:
from alpaca.trading.requests import MarketOrderRequest
from alpaca.trading.enums import OrderSide, TimeInForce
req = MarketOrderRequest(
symbol=symbol,
qty=qty,
side=OrderSide.BUY,
time_in_force=TimeInForce.DAY,
)
order = self._trading.submit_order(req)
return self._to_order(order)
def submit_moc_buy(self, symbol: str, qty: int) -> Order:
"""Submit a Market-on-Close buy order (fills at today's closing price)."""
from alpaca.trading.requests import MarketOrderRequest
from alpaca.trading.enums import OrderSide, TimeInForce
req = MarketOrderRequest(
symbol=symbol,
qty=qty,
side=OrderSide.BUY,
time_in_force=TimeInForce.CLS,
)
order = self._trading.submit_order(req)
return self._to_order(order)
def submit_market_sell(self, symbol: str, qty: int) -> Order:
from alpaca.trading.requests import MarketOrderRequest
from alpaca.trading.enums import OrderSide, TimeInForce
req = MarketOrderRequest(
symbol=symbol,
qty=qty,
side=OrderSide.SELL,
time_in_force=TimeInForce.DAY,
)
order = self._trading.submit_order(req)
return self._to_order(order)
def get_order(self, order_id: str) -> Order:
order = self._trading.get_order_by_id(order_id)
return self._to_order(order)
def list_orders(self, status: str = "open") -> list[Order]:
from alpaca.trading.requests import GetOrdersRequest
from alpaca.trading.enums import QueryOrderStatus
status_map = {
"open": QueryOrderStatus.OPEN,
"closed": QueryOrderStatus.CLOSED,
"all": QueryOrderStatus.ALL,
}
req = GetOrdersRequest(status=status_map.get(status, QueryOrderStatus.OPEN))
orders = self._trading.get_orders(req)
return [self._to_order(o) for o in orders]
def cancel_order(self, order_id: str) -> None:
self._trading.cancel_order_by_id(order_id)
# ------------------------------------------------------------------ #
# Positions
# ------------------------------------------------------------------ #
def list_positions(self) -> list[Position]:
positions = self._trading.get_all_positions()
return [self._to_position(p) for p in positions]
def get_position(self, symbol: str) -> Position | None:
try:
pos = self._trading.get_open_position(symbol)
return self._to_position(pos)
except Exception:
return None
def close_position(self, symbol: str, qty: int | None = None, **kwargs) -> Order:
"""Close a position. Pass qty for partial close."""
from alpaca.trading.requests import ClosePositionRequest
if qty is not None:
req = ClosePositionRequest(qty=str(qty))
order = self._trading.close_position(symbol, close_options=req)
else:
order = self._trading.close_position(symbol)
return self._to_order(order)
def close_all_positions(self) -> list[Order]:
responses = self._trading.close_all_positions(cancel_orders=True)
if not responses:
return []
result = []
for resp in responses:
try:
# close_all_positions returns ClosePositionResponse; body is the actual Order
order_obj = getattr(resp, "body", resp)
result.append(self._to_order(order_obj))
except Exception:
pass
return result
# ------------------------------------------------------------------ #
# Price data
# ------------------------------------------------------------------ #
def get_bars(
self,
symbols: list[str],
start: dt.date,
end: dt.date,
) -> dict[str, list[Bar]]:
"""Fetch daily OHLCV bars for a list of symbols in [start, end]."""
if not symbols:
return {}
from alpaca.data.requests import StockBarsRequest
from alpaca.data.timeframe import TimeFrame
req = StockBarsRequest(
symbol_or_symbols=symbols,
timeframe=TimeFrame.Day,
start=dt.datetime.combine(start, dt.time.min),
end=dt.datetime.combine(end, dt.time.max),
feed="iex",
)
response = self._data.get_stock_bars(req)
result: dict[str, list[Bar]] = {}
for sym in symbols:
try:
bars_data = response[sym]
except (KeyError, TypeError):
bars_data = []
result[sym] = [
Bar(
date=b.timestamp.date().isoformat() if hasattr(b.timestamp, "date") else str(b.timestamp)[:10],
open=float(b.open),
high=float(b.high),
low=float(b.low),
close=float(b.close),
volume=float(b.volume),
)
for b in bars_data
]
return result
def get_bars_as_dict(
self,
symbols: list[str],
start: dt.date,
end: dt.date,
) -> dict[str, dict[dt.date, dict[str, Any]]]:
"""Return bars indexed by symbol → date → OHLCV dict (matches backtest format)."""
raw = self.get_bars(symbols, start, end)
result: dict[str, dict[dt.date, dict[str, Any]]] = {}
for sym, bars in raw.items():
date_map: dict[dt.date, dict[str, Any]] = {}
for bar in bars:
d = dt.date.fromisoformat(bar.date)
date_map[d] = {
"date": d,
"open": bar.open,
"high": bar.high,
"low": bar.low,
"close": bar.close,
"volume": bar.volume,
}
result[sym] = date_map
return result
def get_latest_bars(self, symbols: list[str]) -> dict[str, Bar]:
"""Fetch the latest bar for each symbol."""
if not symbols:
return {}
from alpaca.data.requests import StockLatestBarRequest
req = StockLatestBarRequest(symbol_or_symbols=symbols, feed="iex")
response = self._data.get_stock_latest_bar(req)
result: dict[str, Bar] = {}
for sym in symbols:
b = response.get(sym)
if b is not None:
result[sym] = Bar(
date=b.timestamp.date().isoformat() if hasattr(b.timestamp, "date") else str(b.timestamp)[:10],
open=float(b.open),
high=float(b.high),
low=float(b.low),
close=float(b.close),
volume=float(b.volume),
)
return result
# ------------------------------------------------------------------ #
# Portfolio history
# ------------------------------------------------------------------ #
def get_portfolio_history(self, period: str = "1M") -> PortfolioHistory:
from alpaca.trading.requests import GetPortfolioHistoryRequest
req = GetPortfolioHistoryRequest(period=period, timeframe="1D")
hist = self._trading.get_portfolio_history(req)
return PortfolioHistory(
timestamps=list(hist.timestamp or []),
equity=[float(v) for v in (hist.equity or [])],
profit_loss=[float(v) for v in (hist.profit_loss or [])],
profit_loss_pct=[float(v) for v in (hist.profit_loss_pct or [])],
)
# ------------------------------------------------------------------ #
# Helpers
# ------------------------------------------------------------------ #
@staticmethod
def _to_order(order: Any) -> Order:
return Order(
id=str(order.id),
symbol=str(order.symbol),
qty=int(float(order.qty or 0)),
side=str(order.side.value if hasattr(order.side, "value") else order.side),
status=str(order.status.value if hasattr(order.status, "value") else order.status),
filled_avg_price=float(order.filled_avg_price) if order.filled_avg_price else None,
filled_qty=int(float(order.filled_qty or 0)),
)
@staticmethod
def _to_position(pos: Any) -> Position:
return Position(
symbol=str(pos.symbol),
qty=int(float(pos.qty or 0)),
avg_entry_price=float(pos.avg_entry_price or 0),
current_price=float(pos.current_price or 0),
unrealized_pl=float(pos.unrealized_pl or 0),
market_value=float(pos.market_value or 0),
)
@classmethod
def from_env(cls) -> "AlpacaBroker":
"""Create from environment variables."""
paper = os.environ.get("ALPACA_PAPER", "true").lower() != "false"
return cls(
api_key=os.environ.get("ALPACA_API_KEY"),
secret_key=os.environ.get("ALPACA_SECRET_KEY"),
paper=paper,
)

481
apps/paper_trader/auto.py
Unescape Escape View File

@ -0,0 +1,481 @@
"""Paper trading auto daemon — runs on Phoenix (MST) timezone.
Watches the ET market schedule and automatically executes:
07:00 ET Pre-market pipeline (filing_poller → label_generator, both conventions)
09:35 ET run-open (exits + after-close entries at market open)
15:45 ET run-close (same-day MOC entries before market close)
16:30 ET Post-close pipeline (pending label regen + after-close labels)
Usage:
python -m apps.paper_trader.auto --session v504_live
python -m apps.paper_trader.auto # all active sessions
python -m apps.paper_trader.auto --dry-run # print without executing
"""
from __future__ import annotations
import argparse
import datetime as dt
import json
import os
import subprocess
import sys
import time as _time
from dataclasses import dataclass, field
from zoneinfo import ZoneInfo
from rich import box as rbox
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
_TZ_ET = ZoneInfo("America/New_York")
_TZ_PHX = ZoneInfo("America/Phoenix") # MST, UTC-7, no DST
_console = Console(width=120)
_DEFAULT_DB = os.environ.get("PAPER_TRADER_DB", "paper_trading.db")
# How often (seconds) to reprint the countdown while waiting
_STATUS_INTERVAL = 600 # 10 min
# ------------------------------------------------------------------ #
# Schedule definition
# ------------------------------------------------------------------ #
@dataclass
class ScheduledEvent:
name: str
et_hour: int
et_min: int
description: str
kind: str # pipeline_pre | run_open | run_close | pipeline_post
SCHEDULE: list[ScheduledEvent] = [
ScheduledEvent("pipeline_pre", 7, 0,
"Pre-market pipeline (poller → parser → label_gen both conventions)",
"pipeline_pre"),
ScheduledEvent("run_open", 9, 35,
"run-open — exits + after-close entries at market open",
"run_open"),
ScheduledEvent("run_close", 15, 45,
"run-close — same-day MOC entries (Alpaca cutoff 3:45 PM ET)",
"run_close"),
ScheduledEvent("pipeline_post", 16, 30,
"Post-close pipeline — pending label regen + after-close labels",
"pipeline_post"),
]
# Pipeline commands (in execution order)
_PIPELINE_CMDS = [
["python", "-m", "apps.pipeline.filing_poller.main"],
["python", "-m", "apps.pipeline.filing_fetcher.main"],
["python", "-m", "apps.pipeline.event_parser.main"],
["python", "-m", "apps.pipeline.feature_builder.main"],
["python", "-m", "apps.pipeline.label_generator.main"],
["python", "-m", "apps.pipeline.label_generator.main",
"--entry-convention", "reaction_close"],
]
_POST_PIPELINE_CMDS = [
["python", "-m", "apps.pipeline.filing_poller.main"],
["python", "-m", "apps.pipeline.filing_fetcher.main"],
["python", "-m", "apps.pipeline.event_parser.main"],
["python", "-m", "apps.pipeline.feature_builder.main"],
# reaction_close first (regenerates pending labels with final close price)
["python", "-m", "apps.pipeline.label_generator.main",
"--entry-convention", "reaction_close"],
# then next_open labels for today's after-close events
["python", "-m", "apps.pipeline.label_generator.main"],
]
# ------------------------------------------------------------------ #
# Time helpers
# ------------------------------------------------------------------ #
def _now_et() -> dt.datetime:
return dt.datetime.now(tz=_TZ_ET)
def _now_phx() -> dt.datetime:
return dt.datetime.now(tz=_TZ_PHX)
def _to_phx_str(et_dt: dt.datetime) -> str:
return et_dt.astimezone(_TZ_PHX).strftime("%I:%M %p")
def _fmt_countdown(seconds: float) -> str:
if seconds <= 0:
return "now"
h = int(seconds // 3600)
m = int((seconds % 3600) // 60)
s = int(seconds % 60)
if h > 0:
return f"{h}h {m:02d}m"
if m > 0:
return f"{m}m {s:02d}s"
return f"{s}s"
def _et_dt_for(date: dt.date, event: ScheduledEvent) -> dt.datetime:
return dt.datetime(date.year, date.month, date.day,
event.et_hour, event.et_min, tzinfo=_TZ_ET)
def _is_trading_day(date: dt.date) -> bool:
try:
from libs.common.time_utils import is_trading_day
return is_trading_day(date)
except Exception:
# Fallback: Mon–Fri excluding obvious US holidays
return date.weekday() < 5
def _next_trading_day(from_date: dt.date) -> dt.date:
check = from_date + dt.timedelta(days=1)
for _ in range(14):
if _is_trading_day(check):
return check
check += dt.timedelta(days=1)
raise RuntimeError("No trading day found within 14 days")
def _prev_trading_day(from_date: dt.date) -> dt.date:
check = from_date - dt.timedelta(days=1)
for _ in range(14):
if _is_trading_day(check):
return check
check -= dt.timedelta(days=1)
raise RuntimeError("No previous trading day found within 14 days")
# ------------------------------------------------------------------ #
# Catch-up: run missed pipeline steps on startup
# ------------------------------------------------------------------ #
def _run_catchup(now_et: dt.datetime, dry_run: bool) -> None:
"""자동으로 놓친 파이프라인 단계를 시작 시 실행.
파이프라인은 멱등성이 있어 재실행 안전 (기존 레코드는 skip).
트레이딩 커맨드(run-open/run-close)는 타이밍이 중요해 catch-up 제외.
"""
today = now_et.date()
catchup_items: list[tuple[str, str, list[list[str]]]] = []
# (label, reason, cmds)
# 1. 직전 거래일의 post-close pipeline (16:30 ET)이 안 돌았을 가능성
# → 오늘 장마감 후 이벤트 라벨링이 안 됨 → 다음 run-open 준비 안 됨
prev_td = _prev_trading_day(today)
prev_post_close_et = _et_dt_for(prev_td, SCHEDULE[3]) # pipeline_post = 16:30
if prev_post_close_et < now_et:
catchup_items.append((
f"Post-close pipeline ({prev_td})",
f"직전 거래일 16:30 ET 파이프라인 — 오늘 run-open 후보 준비",
_POST_PIPELINE_CMDS,
))
# 2. 오늘이 거래일이고 pre-market pipeline (07:00 ET) 시간이 지났으면
if _is_trading_day(today):
pre_market_et = _et_dt_for(today, SCHEDULE[0]) # pipeline_pre = 07:00
if pre_market_et < now_et:
catchup_items.append((
f"Pre-market pipeline ({today})",
f"오늘 07:00 ET 파이프라인 — 새벽 공시 라벨링",
_PIPELINE_CMDS,
))
if not catchup_items:
return
_console.print("\n[bold yellow]━━━ Catch-up: 놓친 파이프라인 실행 ━━━[/]")
for label, reason, cmds in catchup_items:
_console.print(f"\n[yellow]▶ {label}[/] [dim]{reason}[/]")
_run_pipeline(cmds, dry_run)
_console.print("[bold yellow]━━━ Catch-up 완료 ━━━[/]\n")
# ------------------------------------------------------------------ #
# Subprocess helpers
# ------------------------------------------------------------------ #
def _run_cmd(cmd: list[str], dry_run: bool, pipeline: bool = False) -> bool:
label = " ".join(cmd[2:] if cmd[:2] == ["python", "-m"] else cmd)
if dry_run:
_console.print(f" [dim][DRY] {label}[/]")
return True
_console.print(f" [dim]▶ {label}[/]", end="")
try:
result = subprocess.run(
cmd, check=False,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
)
ok = result.returncode == 0
if pipeline:
# Pipeline commands: parse JSON logs, show only summary + errors
summary_parts: list[str] = []
error_lines: list[str] = []
for line in (result.stdout or "").splitlines():
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
level = rec.get("level", "")
event = rec.get("event", "")
if level in ("error", "critical"):
error_lines.append(f"[red] {event}: {rec}[/]")
elif event.endswith("_done"):
# Extract key counts from the summary event
parts = [f"{k}={v}" for k, v in rec.items()
if k not in ("event", "level", "timestamp", "job_run_id")]
summary_parts.append(f"{event}({', '.join(parts)})")
except (json.JSONDecodeError, ValueError):
pass # httpx HTTP Request lines and other non-JSON: silently skip
_console.print(f" {'[green]OK[/]' if ok else '[red]FAILED[/]'}")
if summary_parts:
_console.print(f" [dim]{' | '.join(summary_parts)}[/]")
for err in error_lines:
_console.print(err)
else:
# Non-pipeline commands (paper run-open, run-close): pass through output
_console.print(f" {'[green]OK[/]' if ok else '[red]FAILED[/]'}")
if result.stdout:
_console.print(result.stdout.rstrip())
if not ok:
_console.print(f" [red] ↳ FAILED (exit {result.returncode})[/]")
return ok
except Exception as exc:
_console.print(f"\n [red] ↳ ERROR: {exc}[/]")
return False
def _run_pipeline(cmds: list[list[str]], dry_run: bool) -> None:
for cmd in cmds:
_run_cmd(cmd, dry_run, pipeline=True)
def _run_paper(command: str, sessions: list[str], db: str, dry_run: bool) -> None:
for session in sessions:
cmd = ["python", "-m", "apps.paper_trader.cli", command,
"--session", session, "--db", db]
_run_cmd(cmd, dry_run)
# ------------------------------------------------------------------ #
# Display
# ------------------------------------------------------------------ #
def _print_header(sessions: list[str]) -> None:
now_phx = _now_phx()
now_et = _now_et()
_console.print()
_console.print(Panel(
f"[bold cyan]fithia2 paper auto[/] | "
f"PHX [bold]{now_phx.strftime('%I:%M %p MST')}[/] "
f"ET [bold]{now_et.strftime('%I:%M %p %Z')}[/]\n"
f"[dim]Sessions: {', '.join(sessions)}[/]",
border_style="cyan",
padding=(0, 2),
))
def _print_schedule(now_et: dt.datetime, completed: set[str]) -> None:
today = now_et.date()
is_td = _is_trading_day(today)
tbl = Table(box=rbox.SIMPLE, show_header=True, header_style="bold yellow",
padding=(0, 2), expand=False)
tbl.add_column("ET", style="bold", no_wrap=True)
tbl.add_column("PHX", style="dim", no_wrap=True)
tbl.add_column("Action")
tbl.add_column("", no_wrap=True)
for ev in SCHEDULE:
et_dt = _et_dt_for(today, ev)
phx_str = _to_phx_str(et_dt)
et_str = et_dt.strftime("%I:%M %p")
if ev.name in completed:
marker = "[green]✓ done[/]"
elif et_dt <= now_et:
marker = "[dim]skipped[/]"
else:
secs = (et_dt - now_et).total_seconds()
marker = f"[dim]in {_fmt_countdown(secs)}[/]"
tbl.add_row(et_str, phx_str, ev.description, marker)
day_str = f"[bold]{today.strftime('%a %Y-%m-%d')}[/]"
td_str = "[green]Trading Day[/]" if is_td else "[red]Non-Trading Day[/]"
_console.print(f"\nSchedule {day_str} {td_str}")
_console.print(tbl)
def _log(msg: str) -> None:
ts = _now_phx().strftime("%H:%M PHX")
_console.print(f"[dim]{ts}[/] {msg}")
# ------------------------------------------------------------------ #
# Main daemon loop
# ------------------------------------------------------------------ #
def _get_active_sessions(db: str) -> list[str]:
try:
from apps.paper_trader.state import StateManager
return [s.session_name for s in StateManager(db).list_sessions()
if s.status == "active"]
except Exception:
return []
def run_auto(sessions: list[str], db: str, dry_run: bool) -> None:
resolved = sessions or _get_active_sessions(db)
if not resolved:
_console.print(f"[red]No active sessions in '{db}'. Use --session NAME or create a session first.[/]")
sys.exit(1)
_console.print(f"\n[bold cyan]fithia2 paper auto[/] sessions: [bold]{', '.join(resolved)}[/]")
if dry_run:
_console.print("[yellow]DRY RUN — commands will not execute[/]")
_console.print("Ctrl+C to stop.\n")
# 시작 시 놓친 파이프라인 자동 catch-up
_run_catchup(_now_et(), dry_run)
completed: set[str] = set()
last_schedule_date: dt.date | None = None
last_status_print: float = 0.0
try:
while True:
now_et = _now_et()
today = now_et.date()
# New day → reset
if last_schedule_date != today:
completed.clear()
last_schedule_date = today
last_status_print = 0.0
_print_header(resolved)
if not _is_trading_day(today):
next_td = _next_trading_day(today)
_log(f"Non-trading day. Next trading day: [bold]{next_td}[/]")
else:
# Mark events that already passed when starting mid-day as skipped
for ev in SCHEDULE:
if _et_dt_for(today, ev) <= now_et:
completed.add(ev.name)
_log(f"[dim]Skipping past event: {ev.description}[/]")
_print_schedule(now_et, completed)
if not _is_trading_day(today):
_time.sleep(1800) # 30 min; loop will recheck
continue
# Find next pending event
pending = [ev for ev in SCHEDULE if ev.name not in completed]
if not pending:
# All done today → sleep until tomorrow's first event
next_td = _next_trading_day(today)
first = SCHEDULE[0]
wake_et = _et_dt_for(next_td, first)
wait = (wake_et - now_et).total_seconds()
_log(f"[green]All done for today.[/] Sleeping until [bold]{wake_et.strftime('%I:%M %p ET')} "
f"({_to_phx_str(wake_et)} PHX)[/] on {next_td} "
f"— {_fmt_countdown(wait)}")
_time.sleep(min(wait, 3600))
continue
next_ev = pending[0]
next_et = _et_dt_for(today, next_ev)
wait = (next_et - now_et).total_seconds()
if wait > 90:
# Periodic status line every _STATUS_INTERVAL seconds
now_mono = _time.monotonic()
if now_mono - last_status_print >= _STATUS_INTERVAL:
_log(f"Next: [bold]{next_ev.description}[/] "
f"at {next_et.strftime('%I:%M %p ET')} ({_to_phx_str(next_et)} PHX) "
f"— {_fmt_countdown(wait)}")
last_status_print = now_mono
_time.sleep(min(wait - 60, _STATUS_INTERVAL))
continue
# ≤ 90s away — wait out remainder
if wait > 0:
_log(f"[yellow]Firing in {_fmt_countdown(wait)}: {next_ev.description}[/]")
_time.sleep(wait)
# ── Execute ──────────────────────────────────────────────
now_phx_str = _now_phx().strftime("%H:%M PHX")
_console.print(f"\n{'─'*60}")
_console.print(f"[bold green]{now_phx_str} ▶ {next_ev.description}[/]")
_console.print(f"{'─'*60}")
if next_ev.kind == "pipeline_pre":
_run_pipeline(_PIPELINE_CMDS, dry_run)
elif next_ev.kind == "run_open":
_run_paper("run-open", resolved, db, dry_run)
elif next_ev.kind == "run_close":
_run_paper("run-close", resolved, db, dry_run)
elif next_ev.kind == "pipeline_post":
_run_pipeline(_POST_PIPELINE_CMDS, dry_run)
completed.add(next_ev.name)
last_status_print = 0.0 # reprint schedule next status line
done_str = _now_phx().strftime("%H:%M PHX")
_console.print(f"[green]✓ Done ({done_str})[/]")
_print_schedule(_now_et(), completed)
except KeyboardInterrupt:
_console.print("\n[yellow]Auto daemon stopped.[/]")
# ------------------------------------------------------------------ #
# Entry point
# ------------------------------------------------------------------ #
def main() -> None:
parser = argparse.ArgumentParser(
prog="fithia2-paper-auto",
description="Paper trading auto daemon (Phoenix/MST timezone)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
ET schedule (runs every trading day):
07:00 Pre-market pipeline (filing_poller → label_generator, both conventions)
09:35 run-open (exits + after-close entries at market open)
15:45 run-close (same-day MOC entries, Alpaca cutoff 3:45 PM ET)
16:30 Post-close pipeline (pending label regen + after-close labels)
Examples:
python -m apps.paper_trader.auto --session v504_live
python -m apps.paper_trader.auto # all active sessions
python -m apps.paper_trader.auto --dry-run --session v504_live
""",
)
parser.add_argument(
"--session", "-s", nargs="*", dest="session", default=[],
metavar="NAME",
help="Session name(s). Default: all active sessions.",
)
parser.add_argument(
"--db", default=_DEFAULT_DB, metavar="PATH",
help=f"SQLite DB path (default: {_DEFAULT_DB} or $PAPER_TRADER_DB)",
)
parser.add_argument(
"--dry-run", action="store_true",
help="Show what would run without executing.",
)
args = parser.parse_args()
run_auto(sessions=args.session, db=args.db, dry_run=args.dry_run)
if __name__ == "__main__":
main()

322
apps/paper_trader/mock_broker.py
Unescape Escape View File

@ -0,0 +1,322 @@
"""MockBroker: simulates Alpaca Paper Trading API using Oracle historical bar data.
Used by simulate_2week.py and the `fithia2 paper backtest` command.
"""
from __future__ import annotations
import asyncio
import datetime as dt
from typing import Any
from apps.paper_trader.alpaca_broker import AccountInfo, Bar, Order, Position
class MockBroker:
"""Simulates Alpaca Paper Trading API using Oracle historical bar data.
Must call set_sim_context(date, phase) before each phase ('open' or 'close').
"""
def __init__(self, initial_equity: float, bars_cache: dict[str, dict[dt.date, dict]]) -> None:
self._initial_equity = initial_equity
self._cash = float(initial_equity)
self._positions: dict[str, dict] = {} # symbol -> {qty, avg_price, current_price}
self._bars = bars_cache # pre-fetched: symbol -> {date -> {open,high,low,close,volume}}
self._sim_date: dt.date | None = None
self._sim_phase: str = "close" # 'open' or 'close'
self._order_counter = 0
self._trade_log: list[dict] = []
# ── Context ────────────────────────────────────────────────────────────────
def set_sim_context(self, date: dt.date, phase: str) -> None:
"""Set current simulation date + phase before each engine call."""
self._sim_date = date
self._sim_phase = phase
# Refresh position current prices using the appropriate price field for the phase
price_field = "open" if phase == "open" else "close"
for sym, pos in self._positions.items():
bar = self._get_bar_for_date(sym, date)
if bar:
pos["current_price"] = bar.get(price_field) or bar["close"]
# ── Account ────────────────────────────────────────────────────────────────
def get_account(self) -> AccountInfo:
market_value = sum(p["qty"] * p["current_price"] for p in self._positions.values())
equity = self._cash + market_value
unrealized_pl = sum(
(p["current_price"] - p["avg_price"]) * p["qty"]
for p in self._positions.values()
)
return AccountInfo(
equity=equity,
cash=self._cash,
buying_power=self._cash,
long_market_value=market_value,
unrealized_pl=unrealized_pl,
portfolio_value=equity,
)
# ── Orders ─────────────────────────────────────────────────────────────────
def submit_market_buy(self, symbol: str, qty: int) -> Order:
"""Fill at today's open price."""
price = self._fill_price(symbol, "open")
if price is None or price <= 0:
price = self._fill_price(symbol, "close") or 0.0
return self._fill_buy(symbol, qty, price, "market")
def submit_moc_buy(self, symbol: str, qty: int) -> Order:
"""Fill at today's close price."""
price = self._fill_price(symbol, "close")
if price is None or price <= 0:
price = self._fill_price(symbol, "open") or 0.0
return self._fill_buy(symbol, qty, price, "moc")
def submit_market_sell(self, symbol: str, qty: int) -> Order:
return self.close_position(symbol, qty)
def get_order(self, order_id: str) -> Order:
for t in self._trade_log:
if t["order_id"] == order_id:
return Order(
id=order_id, symbol=t["symbol"], qty=t["qty"],
side=t["side"], status="filled",
filled_avg_price=t["price"], filled_qty=t["qty"],
)
raise ValueError(f"Order {order_id} not found")
def list_orders(self, status: str = "open") -> list[Order]:
return []
def cancel_order(self, order_id: str) -> None:
pass
# ── Positions ──────────────────────────────────────────────────────────────
def list_positions(self) -> list[Position]:
return [
Position(
symbol=sym,
qty=p["qty"],
avg_entry_price=p["avg_price"],
current_price=p["current_price"],
unrealized_pl=(p["current_price"] - p["avg_price"]) * p["qty"],
market_value=p["qty"] * p["current_price"],
)
for sym, p in self._positions.items()
if p["qty"] > 0
]
def get_position(self, symbol: str) -> Position | None:
p = self._positions.get(symbol)
if p is None or p["qty"] <= 0:
return None
return Position(
symbol=symbol,
qty=p["qty"],
avg_entry_price=p["avg_price"],
current_price=p["current_price"],
unrealized_pl=(p["current_price"] - p["avg_price"]) * p["qty"],
market_value=p["qty"] * p["current_price"],
)
def close_position(self, symbol: str, qty: int | None = None, *, fill_price: float | None = None) -> Order:
p = self._positions.get(symbol)
if p is None or p["qty"] <= 0:
raise ValueError(f"No open position in {symbol}")
close_qty = qty if qty is not None else p["qty"]
price = fill_price if fill_price is not None else (self._fill_price(symbol, self._sim_phase) or p["avg_price"])
self._cash += price * close_qty
if close_qty >= p["qty"]:
del self._positions[symbol]
else:
p["qty"] -= close_qty
self._order_counter += 1
oid = f"mock_sell_{self._order_counter}"
self._trade_log.append({
"order_id": oid, "symbol": symbol, "qty": close_qty,
"side": "sell", "price": price,
})
return Order(
id=oid, symbol=symbol, qty=close_qty, side="sell",
status="filled", filled_avg_price=price, filled_qty=close_qty,
)
def close_all_positions(self) -> list[Order]:
orders = []
for sym in list(self._positions.keys()):
try:
orders.append(self.close_position(sym))
except Exception:
pass
return orders
# ── Price data ─────────────────────────────────────────────────────────────
def get_bars_as_dict(
self,
symbols: list[str],
start: dt.date,
end: dt.date,
) -> dict[str, dict[dt.date, dict[str, Any]]]:
"""Return bars from pre-fetched cache filtered to [start, end]."""
result: dict[str, dict[dt.date, dict]] = {}
for sym in symbols:
sym_bars = self._bars.get(sym, {})
result[sym] = {d: b for d, b in sym_bars.items() if start <= d <= end}
return result
def get_latest_bars(self, symbols: list[str]) -> dict[str, Bar]:
result: dict[str, Bar] = {}
for sym in symbols:
sym_bars = self._bars.get(sym, {})
if not sym_bars:
continue
latest = max(sym_bars.keys())
b = sym_bars[latest]
result[sym] = Bar(
date=latest.isoformat(),
open=b["open"], high=b["high"], low=b["low"],
close=b["close"], volume=b.get("volume", 0),
)
return result
# ── Helpers ────────────────────────────────────────────────────────────────
def _fill_price(self, symbol: str, field: str) -> float | None:
bar = self._get_bar_for_date(symbol, self._sim_date)
if bar:
return bar.get(field)
sym_bars = self._bars.get(symbol, {})
if sym_bars:
available = [d for d in sym_bars if d <= (self._sim_date or dt.date.today())]
if available:
return sym_bars[max(available)].get(field)
return None
def _get_bar_for_date(self, symbol: str, date: dt.date | None) -> dict | None:
if date is None:
return None
sym_bars = self._bars.get(symbol, {})
if date in sym_bars:
return sym_bars[date]
available = [d for d in sym_bars if d <= date]
if available:
return sym_bars[max(available)]
return None
def _fill_buy(self, symbol: str, qty: int, price: float, order_type: str) -> Order:
if price <= 0:
raise ValueError(f"Cannot fill {symbol}: no price available for {self._sim_date}")
cost = price * qty
self._cash -= cost
if symbol in self._positions:
p = self._positions[symbol]
total_qty = p["qty"] + qty
p["avg_price"] = (p["avg_price"] * p["qty"] + price * qty) / total_qty
p["qty"] = total_qty
p["current_price"] = price
else:
self._positions[symbol] = {
"qty": qty,
"avg_price": price,
"current_price": price,
}
self._order_counter += 1
oid = f"mock_buy_{self._order_counter}"
self._trade_log.append({
"order_id": oid, "symbol": symbol, "qty": qty,
"side": "buy", "price": price,
})
return Order(
id=oid, symbol=symbol, qty=qty, side="buy",
status="filled", filled_avg_price=price, filled_qty=qty,
)
# ─────────────────────────────────────────────────────────────────────────────
# Bar pre-fetcher
# ─────────────────────────────────────────────────────────────────────────────
async def prefetch_bars(
symbols: list[str],
start: dt.date,
end: dt.date,
oracle_url: str,
console=None,
) -> dict[str, dict[dt.date, dict]]:
"""Fetch OHLCV bars from Oracle for given symbols and date range."""
from libs.oracle_client import OracleClient, PriceService
cache: dict[str, dict[dt.date, dict]] = {}
sem = asyncio.Semaphore(10)
async def _fetch_one(sym: str) -> None:
async with sem:
try:
async with OracleClient(base_url=oracle_url) as client:
svc = PriceService(client)
resp = await svc.get_daily_bars(
sym,
start=start.isoformat(),
end=end.isoformat(),
)
date_map: dict[dt.date, dict] = {}
for b in resp.bars:
d = dt.date.fromisoformat(b.date[:10])
date_map[d] = {
"date": d,
"open": float(b.open),
"high": float(b.high),
"low": float(b.low),
"close": float(b.close),
"volume": float(b.volume),
}
cache[sym] = date_map
except Exception as exc:
msg = f" Bar fetch failed for {sym}: {exc}"
if console is not None:
console.print(f" [yellow]{msg}[/]")
cache[sym] = {}
await asyncio.gather(*(_fetch_one(sym) for sym in symbols))
return cache
async def get_event_symbols_from_db(
db_dsn: str,
start_date: dt.date,
end_date: dt.date,
console=None,
) -> list[str]:
"""Query pipeline DB for symbols appearing in events during the date range."""
try:
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from libs.db.models import Event, EventLabel, SymbolMaster
engine = create_async_engine(db_dsn, echo=False, connect_args={"timeout": 5})
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async with async_session() as session:
stmt = (
select(SymbolMaster.ticker)
.join(Event, Event.symbol_id == SymbolMaster.symbol_id)
.join(EventLabel, Event.event_id == EventLabel.event_id)
.where(EventLabel.entry_date >= start_date)
.where(EventLabel.entry_date <= end_date)
.where(EventLabel.label_status.in_(["ok", "truncated", "pending"]))
.distinct()
)
rows = (await session.execute(stmt)).all()
await engine.dispose()
return [r[0] for r in rows if r[0]]
except Exception as exc:
msg = f"DB symbol fetch failed: {type(exc).__name__}: {exc}" if str(exc) else f"DB symbol fetch failed: {type(exc).__name__}"
if console is not None:
console.print(f" [red]{msg}[/]")
return []

105
apps/paper_trader/models.py
Unescape Escape View File

@ -0,0 +1,105 @@
"""SQLite table definitions for the paper trading system."""
from __future__ import annotations
import sqlite3
from pathlib import Path
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
session_name TEXT NOT NULL,
config_path TEXT NOT NULL,
initial_equity REAL NOT NULL,
created_at TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'active'
);
CREATE TABLE IF NOT EXISTS strategy_states (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL REFERENCES sessions(session_id),
symbol TEXT NOT NULL,
event_id TEXT NOT NULL,
engine_id TEXT NOT NULL DEFAULT 'default',
order_id TEXT,
entry_date TEXT NOT NULL,
stop_price REAL NOT NULL,
target_price REAL NOT NULL,
current_stop REAL NOT NULL,
peak_price REAL NOT NULL,
days_held INTEGER NOT NULL DEFAULT 0,
trade_direction TEXT NOT NULL DEFAULT 'long',
candidate_json TEXT NOT NULL,
plan_json TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'open',
UNIQUE(session_id, symbol, status)
);
CREATE TABLE IF NOT EXISTS session_state (
session_id TEXT PRIMARY KEY REFERENCES sessions(session_id),
consecutive_losses INTEGER NOT NULL DEFAULT 0,
cooldown_remaining INTEGER NOT NULL DEFAULT 0,
kill_switch_triggered INTEGER NOT NULL DEFAULT 0,
daily_new_risk_used REAL NOT NULL DEFAULT 0.0,
last_processed_date TEXT
);
CREATE TABLE IF NOT EXISTS processed_events (
session_id TEXT NOT NULL,
event_id TEXT NOT NULL,
processed_date TEXT NOT NULL,
action TEXT NOT NULL,
skip_reason TEXT,
PRIMARY KEY (session_id, event_id)
);
CREATE TABLE IF NOT EXISTS trades (
trade_id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
symbol TEXT NOT NULL,
entry_date TEXT,
exit_date TEXT,
entry_price REAL,
exit_price REAL,
exit_reason TEXT,
shares INTEGER,
net_pnl REAL,
r_multiple REAL,
holding_days INTEGER
);
CREATE TABLE IF NOT EXISTS daily_snapshots (
session_id TEXT NOT NULL,
date TEXT NOT NULL,
equity REAL NOT NULL,
cash REAL NOT NULL,
market_value REAL NOT NULL,
daily_pnl REAL,
total_pnl REAL,
drawdown_pct REAL,
open_position_count INTEGER,
PRIMARY KEY (session_id, date)
);
CREATE TABLE IF NOT EXISTS processed_dates (
session_id TEXT NOT NULL,
date TEXT NOT NULL,
PRIMARY KEY (session_id, date)
);
CREATE TABLE IF NOT EXISTS processed_phases (
session_id TEXT NOT NULL,
date TEXT NOT NULL,
phase TEXT NOT NULL,
PRIMARY KEY (session_id, date, phase)
);
"""
def create_schema(db_path: str | Path) -> None:
"""Initialize the SQLite database with the paper trading schema."""
conn = sqlite3.connect(str(db_path))
try:
conn.executescript(SCHEMA_SQL)
conn.commit()
finally:
conn.close()

576
apps/paper_trader/reporter.py
Unescape Escape View File

@ -0,0 +1,576 @@
"""Rich console output for paper trading status, positions, trades, and equity."""
from __future__ import annotations
import datetime as dt
from typing import Any
from rich import box
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from apps.paper_trader.alpaca_broker import AlpacaBroker, Position
from apps.paper_trader.state import SessionRow, StateManager, StrategyStateRow
_console = Console(width=140)
def _pnl_color(val: float) -> str:
return "green" if val >= 0 else "red"
def _fmt_pct(val: float) -> str:
sign = "+" if val >= 0 else ""
return f"{sign}{val:.2f}%"
def _fmt_pnl(val: float) -> str:
sign = "+" if val >= 0 else ""
return f"{sign}${val:,.2f}"
# ------------------------------------------------------------------ #
# Session status
# ------------------------------------------------------------------ #
def print_status(
session: SessionRow,
broker: AlpacaBroker,
state: StateManager,
) -> None:
account = broker.get_account()
session_st = state.get_session_state(session.session_id)
snapshots = state.list_snapshots(session.session_id)
initial_equity = session.initial_equity
# Use latest snapshot equity if available, otherwise fall back to initial
if snapshots:
current_equity = snapshots[-1]["equity"]
else:
current_equity = initial_equity
total_pnl = current_equity - initial_equity
total_pnl_pct = total_pnl / initial_equity * 100 if initial_equity else 0.0
peak_equity = state.get_peak_equity(session.session_id, initial_equity)
drawdown_pct = max(0.0, (peak_equity - current_equity) / peak_equity * 100) if peak_equity > 0 else 0.0
positions = broker.list_positions()
kill_status = "[bold red]ON[/]" if session_st.kill_switch_triggered else "[green]OFF[/]"
cooldown_str = f"{session_st.cooldown_remaining}d" if session_st.cooldown_remaining > 0 else "none"
pnl_style = _pnl_color(total_pnl)
_console.print()
_console.print(Panel(
f"[bold cyan]Session:[/] {session.session_name} "
f"[dim]|[/] [bold]Config:[/] {session.config_path} "
f"[dim]|[/] [bold]Status:[/] {session.status}",
border_style="cyan",
padding=(0, 1),
))
tbl = Table(box=box.SIMPLE, show_header=False, padding=(0, 2))
tbl.add_column("Key", style="bold yellow", no_wrap=True)
tbl.add_column("Value", no_wrap=True)
tbl.add_row("Equity", f"[bold]${current_equity:,.2f}[/] [{pnl_style}]{_fmt_pnl(total_pnl)} ({_fmt_pct(total_pnl_pct)})[/]")
tbl.add_row("Cash", f"${account.cash:,.2f}")
tbl.add_row("Market Value", f"${account.long_market_value:,.2f}")
tbl.add_row("Open Positions", str(len(positions)))
tbl.add_row("Drawdown", f"[{'red' if drawdown_pct > 5 else 'green'}]{drawdown_pct:.2f}%[/]")
tbl.add_row("Consecutive Losses", str(session_st.consecutive_losses))
tbl.add_row("Cooldown", cooldown_str)
tbl.add_row("Kill Switch", kill_status)
tbl.add_row("Total Trades", str(len(state.list_trades(session.session_id))))
if snapshots:
tbl.add_row("Last Processed", snapshots[-1]["date"])
_console.print(tbl)
_console.print()
# ------------------------------------------------------------------ #
# Positions
# ------------------------------------------------------------------ #
def print_positions(
session: SessionRow,
broker: AlpacaBroker,
state: StateManager,
) -> None:
positions = broker.list_positions()
strategy_states = {
ss.symbol: ss
for ss in state.get_open_strategy_states(session.session_id)
}
if not positions:
_console.print("[dim]No open positions.[/]")
return
tbl = Table(
box=box.SIMPLE_HEAD,
show_header=True,
header_style="bold yellow",
padding=(0, 1),
title=f"[bold cyan]Open Positions[/] — {session.session_name}",
title_justify="left",
)
tbl.add_column("Symbol", style="bold", no_wrap=True)
tbl.add_column("Qty", justify="right")
tbl.add_column("Entry", justify="right")
tbl.add_column("Current", justify="right")
tbl.add_column("P&L $", justify="right")
tbl.add_column("P&L %", justify="right")
tbl.add_column("Days", justify="right")
tbl.add_column("Stop", justify="right")
tbl.add_column("Target", justify="right")
for pos in sorted(positions, key=lambda p: p.symbol):
ss = strategy_states.get(pos.symbol)
stop_str = f"${ss.current_stop:.2f}" if ss else "-"
target_str = f"${ss.target_price:.2f}" if ss else "-"
days_str = str(ss.days_held) if ss else "-"
pnl = pos.unrealized_pl
pnl_pct = pnl / (pos.avg_entry_price * pos.qty) * 100 if pos.avg_entry_price and pos.qty else 0.0
pnl_color = _pnl_color(pnl)
tbl.add_row(
pos.symbol,
str(pos.qty),
f"${pos.avg_entry_price:.2f}",
f"${pos.current_price:.2f}",
f"[{pnl_color}]{_fmt_pnl(pnl)}[/]",
f"[{pnl_color}]{_fmt_pct(pnl_pct)}[/]",
days_str,
stop_str,
target_str,
)
_console.print()
_console.print(tbl)
_console.print()
# ------------------------------------------------------------------ #
# Trades
# ------------------------------------------------------------------ #
def print_trades(
session: SessionRow,
state: StateManager,
last: int | None = None,
) -> None:
trades = state.list_trades(session.session_id, limit=last)
if not trades:
_console.print("[dim]No trades recorded yet.[/]")
return
tbl = Table(
box=box.SIMPLE_HEAD,
show_header=True,
header_style="bold yellow",
padding=(0, 1),
title=f"[bold cyan]Trades[/] — {session.session_name}",
title_justify="left",
)
tbl.add_column("Symbol", style="bold", no_wrap=True)
tbl.add_column("Entry Date", no_wrap=True)
tbl.add_column("Exit Date", no_wrap=True)
tbl.add_column("Entry $", justify="right")
tbl.add_column("Exit $", justify="right")
tbl.add_column("Shares", justify="right")
tbl.add_column("Net P&L", justify="right")
tbl.add_column("R", justify="right")
tbl.add_column("Days", justify="right")
tbl.add_column("Reason", style="dim")
for t in trades:
pnl = t.get("net_pnl") or 0.0
r = t.get("r_multiple") or 0.0
pnl_color = _pnl_color(pnl)
r_color = _pnl_color(r)
tbl.add_row(
str(t.get("symbol", "")),
str(t.get("entry_date", "-")),
str(t.get("exit_date", "-")),
f"${t.get('entry_price') or 0:.2f}",
f"${t.get('exit_price') or 0:.2f}",
str(t.get("shares", "-")),
f"[{pnl_color}]{_fmt_pnl(pnl)}[/]",
f"[{r_color}]{r:+.2f}R[/]",
str(t.get("holding_days", "-")),
str(t.get("exit_reason", "-")),
)
_console.print()
_console.print(tbl)
_console.print()
# ------------------------------------------------------------------ #
# Equity curve (ASCII sparkline)
# ------------------------------------------------------------------ #
def print_equity(
session: SessionRow,
state: StateManager,
) -> None:
snapshots = state.list_snapshots(session.session_id)
if not snapshots:
_console.print("[dim]No equity history yet.[/]")
return
tbl = Table(
box=box.SIMPLE_HEAD,
show_header=True,
header_style="bold yellow",
padding=(0, 1),
title=f"[bold cyan]Equity Curve[/] — {session.session_name}",
title_justify="left",
)
tbl.add_column("Date", no_wrap=True)
tbl.add_column("Equity", justify="right")
tbl.add_column("Daily P&L", justify="right")
tbl.add_column("Total P&L", justify="right")
tbl.add_column("Drawdown", justify="right")
tbl.add_column("Positions", justify="right")
for snap in snapshots:
equity = snap.get("equity", 0.0)
daily_pnl = snap.get("daily_pnl") or 0.0
total_pnl = snap.get("total_pnl") or 0.0
dd = snap.get("drawdown_pct") or 0.0
n_pos = snap.get("open_position_count")
daily_color = _pnl_color(daily_pnl)
total_color = _pnl_color(total_pnl)
dd_color = "red" if dd > 5 else "green"
tbl.add_row(
str(snap.get("date", "-")),
f"${equity:,.2f}",
f"[{daily_color}]{_fmt_pnl(daily_pnl)}[/]",
f"[{total_color}]{_fmt_pnl(total_pnl)}[/]",
f"[{dd_color}]{dd:.2f}%[/]",
str(n_pos) if n_pos is not None else "-",
)
_console.print()
_console.print(tbl)
_console.print()
# ------------------------------------------------------------------ #
# Sessions list
# ------------------------------------------------------------------ #
def print_sessions(sessions: list[SessionRow]) -> None:
if not sessions:
_console.print("[dim]No sessions found.[/]")
return
tbl = Table(
box=box.SIMPLE_HEAD,
show_header=True,
header_style="bold yellow",
padding=(0, 1),
title="[bold cyan]Paper Trading Sessions[/]",
title_justify="left",
)
tbl.add_column("ID", style="dim", no_wrap=True)
tbl.add_column("Name", style="bold", no_wrap=True)
tbl.add_column("Config", no_wrap=True)
tbl.add_column("Capital", justify="right")
tbl.add_column("Status", no_wrap=True)
tbl.add_column("Created", no_wrap=True)
for s in sessions:
status_color = "green" if s.status == "active" else "dim"
tbl.add_row(
s.session_id,
s.session_name,
s.config_path,
f"${s.initial_equity:,.0f}",
f"[{status_color}]{s.status}[/]",
s.created_at[:10],
)
_console.print()
_console.print(tbl)
_console.print()
# ------------------------------------------------------------------ #
# Run summary
# ------------------------------------------------------------------ #
def _print_overlay_detail(r: dict) -> None:
"""Print overlay regime allocation + per-book summary."""
name = r["session_name"]
# Regime day counts
regime_counts = r.get("regime_day_counts", {})
if regime_counts:
regime_tbl = Table(
box=box.SIMPLE_HEAD, show_header=True, header_style="bold yellow",
padding=(0, 1), title=f"[bold magenta]Regime Days — {name}[/]", title_justify="left",
)
regime_tbl.add_column("Regime", style="bold")
regime_tbl.add_column("Days", justify="right")
regime_tbl.add_column("Allocation", no_wrap=True)
allocations = r.get("allocations", {})
for regime, count in sorted(regime_counts.items()):
alloc = allocations.get(regime, {})
alloc_str = " ".join(f"{k}={v:.0%}" for k, v in alloc.items())
regime_tbl.add_row(regime, str(count), alloc_str)
_console.print(regime_tbl)
# Per-book summary
book_results = r.get("book_results", [])
if book_results:
book_tbl = Table(
box=box.SIMPLE_HEAD, show_header=True, header_style="bold yellow",
padding=(0, 1), title=f"[bold magenta]Books — {name}[/]", title_justify="left",
)
book_tbl.add_column("Book", style="bold")
book_tbl.add_column("Return", justify="right")
book_tbl.add_column("MaxDD", justify="right")
book_tbl.add_column("Trades", justify="right")
book_tbl.add_column("WinRate", justify="right")
book_tbl.add_column("Sharpe", justify="right")
for br in book_results:
bs = br["result"]["summary"]
ret_color = "green" if bs["return_pct"] >= 0 else "red"
book_tbl.add_row(
br["label"],
f"[{ret_color}]{bs['return_pct']:+.2f}%[/{ret_color}]",
f"[red]-{bs['max_dd_pct']:.2f}%[/]",
str(bs["trade_count"]),
f"{bs['win_rate']:.0f}%",
f"{bs['sharpe']:+.2f}",
)
_console.print(book_tbl)
def print_backtest_results(results: list[dict], output_dir: str | None = None, show_trades: bool = True) -> None:
"""Print equity curve comparison, summary table, and per-strategy trade logs."""
import csv
import os
if not results:
_console.print("[dim]No backtest results.[/]")
return
# ── Summary table ──────────────────────────────────────────────────────
sum_tbl = Table(
box=box.SIMPLE_HEAD,
show_header=True,
header_style="bold yellow",
padding=(0, 1),
title="[bold cyan]Summary[/]",
title_justify="left",
)
sum_tbl.add_column("Strategy", style="bold", no_wrap=True)
sum_tbl.add_column("Return", justify="right")
sum_tbl.add_column("MaxDD", justify="right")
sum_tbl.add_column("Trades", justify="right")
sum_tbl.add_column("WinRate", justify="right")
sum_tbl.add_column("Sharpe", justify="right")
for r in results:
s = r["summary"]
ret_color = "green" if s["return_pct"] >= 0 else "red"
name = r["session_name"]
if r.get("is_overlay"):
name = f"{name} [overlay]"
sum_tbl.add_row(
name,
f"[{ret_color}]{s['return_pct']:+.2f}%[/{ret_color}]",
f"[red]-{s['max_dd_pct']:.2f}%[/]",
str(s["trade_count"]),
f"{s['win_rate']:.0f}%" if s.get("win_rate") else "-",
f"{s['sharpe']:+.2f}",
)
_console.print(sum_tbl)
# ── Overlay detail sections ───────────────────────────────────────────
for r in results:
if not r.get("is_overlay"):
continue
_print_overlay_detail(r)
# ── Per-strategy trade logs ────────────────────────────────────────────
if not show_trades:
if output_dir:
import csv as _csv
os.makedirs(output_dir, exist_ok=True)
for r in results:
name = r["session_name"]
eq_path = os.path.join(output_dir, f"{name}_equity.csv")
with open(eq_path, "w", newline="") as f:
w = _csv.DictWriter(f, fieldnames=["date", "equity"])
w.writeheader()
for row in r["equity_curve"]:
w.writerow({"date": row["date"].isoformat(), "equity": row["equity"]})
trades_path = os.path.join(output_dir, f"{name}_trades.csv")
if r.get("trades"):
with open(trades_path, "w", newline="") as f:
fieldnames = ["symbol", "event_type", "score", "entry_date", "exit_date", "shares", "entry_price", "exit_price", "pnl", "reason"]
w = _csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
w.writeheader()
w.writerows(r["trades"])
_console.print(f"\n[dim]Results saved to {output_dir}[/]")
return
for r in results:
trades = r.get("trades", [])
if not trades:
_console.print(f" [dim]{r['session_name']}: no completed trades[/]")
continue
tbl = Table(
box=box.SIMPLE_HEAD,
show_header=True,
header_style="bold yellow",
padding=(0, 1),
title=f"[bold cyan]Trades — {r['session_name']}[/]",
title_justify="left",
)
tbl.add_column("Symbol", style="bold", no_wrap=True)
tbl.add_column("EventType", no_wrap=True)
tbl.add_column("Score", justify="right")
tbl.add_column("Entry", no_wrap=True)
tbl.add_column("Exit", no_wrap=True)
tbl.add_column("Days", justify="right")
tbl.add_column("Shares", justify="right")
tbl.add_column("EntryPx", justify="right")
tbl.add_column("ExitPx", justify="right")
tbl.add_column("P&L", justify="right")
tbl.add_column("Reason", style="dim")
for t in sorted(trades, key=lambda x: x.get("entry_date", "")):
pnl = t.get("pnl", 0.0)
color = _pnl_color(pnl)
entry_d = t.get("entry_date", "")
exit_d = t.get("exit_date", "")
try:
import datetime as _dt
holding_days = (_dt.date.fromisoformat(str(exit_d)) - _dt.date.fromisoformat(str(entry_d))).days if entry_d and exit_d else "-"
except Exception:
holding_days = "-"
score = t.get("score")
score_str = f"{score:.2f}" if score is not None else "-"
entry_px = t.get("entry_price")
exit_px = t.get("exit_price")
entry_px_str = f"${entry_px:.2f}" if entry_px is not None else "-"
exit_px_str = f"${exit_px:.2f}" if exit_px is not None else "-"
tbl.add_row(
t.get("symbol", ""),
t.get("event_type", "-"),
score_str,
str(entry_d) if entry_d else "-",
str(exit_d) if exit_d else "-",
str(holding_days),
str(t.get("shares", "-")),
entry_px_str,
exit_px_str,
f"[{color}]{_fmt_pnl(pnl)}[/{color}]",
t.get("reason", "-"),
)
_console.print(tbl)
# ── Optional CSV save ──────────────────────────────────────────────────
if output_dir:
os.makedirs(output_dir, exist_ok=True)
for r in results:
name = r["session_name"]
eq_path = os.path.join(output_dir, f"{name}_equity.csv")
with open(eq_path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=["date", "equity"])
w.writeheader()
for row in r["equity_curve"]:
w.writerow({"date": row["date"].isoformat(), "equity": row["equity"]})
trades_path = os.path.join(output_dir, f"{name}_trades.csv")
if r.get("trades"):
with open(trades_path, "w", newline="") as f:
fieldnames = ["symbol", "event_type", "score", "entry_date", "exit_date", "shares", "entry_price", "exit_price", "pnl", "reason"]
w = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
w.writeheader()
w.writerows(r["trades"])
_console.print(f"\n[dim]Results saved to {output_dir}[/]")
def print_run_summary(summary: dict) -> None:
date = summary.get("date", "-")
status = summary.get("status", "-")
if status in ("already_processed", "non_trading_day"):
_console.print(f"[dim]{date}: {status}[/]")
return
_console.print(f"\nProcessing [bold]{date}[/]...")
exits = summary.get("exits", [])
entries = summary.get("entries", [])
rejected = summary.get("rejected", [])
candidates_detected = summary.get("candidates_detected", None)
account = summary.get("account", {})
# Exits
if exits:
_console.print(" [bold]EXITS:[/]")
for e in exits:
pnl = e.get("pnl", 0.0)
r = e.get("r_multiple", 0.0)
color = _pnl_color(pnl)
_console.print(
f" [{color}]{e['symbol']}[/] — {e['reason']} "
f"P&L: [{color}]{_fmt_pnl(pnl)}[/] R: [{color}]{r:+.2f}R[/]"
)
else:
_console.print(" [dim]EXITS: none[/]")
# Candidates detected
if candidates_detected is not None:
if candidates_detected == 0:
_console.print(" [dim]CANDIDATES: 0 events in DB for this date[/]")
else:
_console.print(f" [dim]CANDIDATES: {candidates_detected} events detected from DB[/]")
# Entries
if entries:
_console.print(" [bold]ENTRIES:[/]")
for e in entries:
_console.print(
f" [green]{e['symbol']}[/] ({e['event_type']}, score={e['score']:.2f}) "
f"-> BUY {e['shares']} shares stop=${e['stop']:.2f} target=${e['target']:.2f}"
)
else:
_console.print(" [dim]ENTRIES: none[/]")
# Rejected (only if non-zero)
if rejected:
_console.print(f" [dim]REJECTED: {len(rejected)} candidates[/]")
# Account summary
if account:
pnl = account.get("total_pnl", 0.0)
color = _pnl_color(pnl)
_console.print(
f"\n Equity: [bold]${account.get('equity', 0):,.2f}[/] "
f"Total P&L: [{color}]{_fmt_pnl(pnl)}[/] "
f"Drawdown: {account.get('drawdown_pct', 0):.2f}%"
)
_console.print()

412
apps/paper_trader/state.py
Unescape Escape View File

@ -0,0 +1,412 @@
"""SQLite state management for paper trading sessions."""
from __future__ import annotations
import datetime as dt
import sqlite3
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from apps.paper_trader.models import create_schema
@dataclass
class SessionRow:
session_id: str
session_name: str
config_path: str
initial_equity: float
created_at: str
status: str
@dataclass
class StrategyStateRow:
session_id: str
symbol: str
event_id: str
engine_id: str
entry_date: str
stop_price: float
target_price: float
current_stop: float
peak_price: float
days_held: int
trade_direction: str
candidate_json: str
plan_json: str
status: str
id: int | None = None
order_id: str | None = None
@dataclass
class SessionStateRow:
session_id: str
consecutive_losses: int = 0
cooldown_remaining: int = 0
kill_switch_triggered: bool = False
daily_new_risk_used: float = 0.0
last_processed_date: str | None = None
@dataclass
class DailySnapshotRow:
session_id: str
date: str
equity: float
cash: float
market_value: float
daily_pnl: float | None = None
total_pnl: float | None = None
drawdown_pct: float | None = None
open_position_count: int | None = None
class StateManager:
"""SQLite CRUD for paper trading state."""
def __init__(self, db_path: str | Path) -> None:
self.db_path = Path(db_path)
create_schema(self.db_path)
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(str(self.db_path))
conn.row_factory = sqlite3.Row
return conn
# ------------------------------------------------------------------ #
# Sessions
# ------------------------------------------------------------------ #
def create_session(
self,
session_name: str,
config_path: str,
initial_equity: float,
) -> str:
session_id = str(uuid.uuid4())[:8]
created_at = dt.datetime.now(tz=dt.timezone.utc).isoformat()
with self._connect() as conn:
conn.execute(
"INSERT INTO sessions (session_id, session_name, config_path, initial_equity, created_at, status) "
"VALUES (?, ?, ?, ?, ?, 'active')",
(session_id, session_name, config_path, initial_equity, created_at),
)
conn.execute(
"INSERT INTO session_state (session_id) VALUES (?)",
(session_id,),
)
return session_id
def get_session(self, session_name_or_id: str) -> SessionRow | None:
with self._connect() as conn:
row = conn.execute(
"SELECT * FROM sessions WHERE session_id = ? OR session_name = ? LIMIT 1",
(session_name_or_id, session_name_or_id),
).fetchone()
if row is None:
return None
return SessionRow(**dict(row))
def list_sessions(self) -> list[SessionRow]:
with self._connect() as conn:
rows = conn.execute(
"SELECT * FROM sessions ORDER BY created_at"
).fetchall()
return [SessionRow(**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:
"""Delete a session and all related data."""
with self._connect() as conn:
for table in (
"processed_phases", "processed_dates", "daily_snapshots",
"trades", "processed_events", "strategy_states", "session_state",
):
conn.execute(f"DELETE FROM {table} WHERE session_id = ?", (session_id,))
conn.execute("DELETE FROM sessions WHERE session_id = ?", (session_id,))
# ------------------------------------------------------------------ #
# Strategy states
# ------------------------------------------------------------------ #
def get_open_strategy_states(self, session_id: str) -> list[StrategyStateRow]:
with self._connect() as conn:
rows = conn.execute(
"SELECT * FROM strategy_states WHERE session_id = ? AND status IN ('open', 'partial')",
(session_id,),
).fetchall()
return [StrategyStateRow(**dict(r)) for r in rows]
def get_strategy_state_by_symbol(
self, session_id: str, symbol: str
) -> StrategyStateRow | None:
with self._connect() as conn:
row = conn.execute(
"SELECT * FROM strategy_states WHERE session_id = ? AND symbol = ? AND status IN ('open', 'partial') LIMIT 1",
(session_id, symbol),
).fetchone()
if row is None:
return None
return StrategyStateRow(**dict(row))
def save_strategy_state(self, session_id: str, state: StrategyStateRow) -> None:
with self._connect() as conn:
conn.execute(
"""INSERT INTO strategy_states
(session_id, symbol, event_id, engine_id, order_id, entry_date,
stop_price, target_price, current_stop, peak_price, days_held,
trade_direction, candidate_json, plan_json, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(session_id, symbol, status) DO UPDATE SET
current_stop=excluded.current_stop,
peak_price=excluded.peak_price,
days_held=excluded.days_held,
order_id=excluded.order_id""",
(
session_id, state.symbol, state.event_id, state.engine_id,
state.order_id, state.entry_date, state.stop_price, state.target_price,
state.current_stop, state.peak_price, state.days_held,
state.trade_direction, state.candidate_json, state.plan_json, state.status,
),
)
def update_strategy_state(
self,
session_id: str,
symbol: str,
*,
days_held: int | None = None,
current_stop: float | None = None,
peak_price: float | None = None,
status: str | None = None,
) -> None:
updates: list[str] = []
values: list[Any] = []
if days_held is not None:
updates.append("days_held = ?")
values.append(days_held)
if current_stop is not None:
updates.append("current_stop = ?")
values.append(current_stop)
if peak_price is not None:
updates.append("peak_price = ?")
values.append(peak_price)
if status is not None:
updates.append("status = ?")
values.append(status)
if not updates:
return
values.extend([session_id, symbol])
with self._connect() as conn:
conn.execute(
f"UPDATE strategy_states SET {', '.join(updates)} "
"WHERE session_id = ? AND symbol = ? AND status IN ('open', 'partial')",
values,
)
def close_strategy_state(self, session_id: str, symbol: str) -> None:
with self._connect() as conn:
conn.execute(
"DELETE FROM strategy_states "
"WHERE session_id = ? AND symbol = ? AND status IN ('open', 'partial')",
(session_id, symbol),
)
# ------------------------------------------------------------------ #
# Session-level state
# ------------------------------------------------------------------ #
def get_session_state(self, session_id: str) -> SessionStateRow:
with self._connect() as conn:
row = conn.execute(
"SELECT * FROM session_state WHERE session_id = ?",
(session_id,),
).fetchone()
if row is None:
return SessionStateRow(session_id=session_id)
d = dict(row)
d["kill_switch_triggered"] = bool(d["kill_switch_triggered"])
return SessionStateRow(**d)
def update_session_state(self, state: SessionStateRow) -> None:
with self._connect() as conn:
conn.execute(
"""INSERT INTO session_state
(session_id, consecutive_losses, cooldown_remaining,
kill_switch_triggered, daily_new_risk_used, last_processed_date)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(session_id) DO UPDATE SET
consecutive_losses=excluded.consecutive_losses,
cooldown_remaining=excluded.cooldown_remaining,
kill_switch_triggered=excluded.kill_switch_triggered,
daily_new_risk_used=excluded.daily_new_risk_used,
last_processed_date=excluded.last_processed_date""",
(
state.session_id,
state.consecutive_losses,
state.cooldown_remaining,
int(state.kill_switch_triggered),
state.daily_new_risk_used,
state.last_processed_date,
),
)
# ------------------------------------------------------------------ #
# Processed events
# ------------------------------------------------------------------ #
def has_processed_event(self, session_id: str, event_id: str) -> bool:
with self._connect() as conn:
row = conn.execute(
"SELECT 1 FROM processed_events WHERE session_id = ? AND event_id = ?",
(session_id, event_id),
).fetchone()
return row is not None
def record_processed_event(
self,
session_id: str,
event_id: str,
processed_date: str,
action: str,
skip_reason: str | None = None,
) -> None:
with self._connect() as conn:
conn.execute(
"INSERT OR IGNORE INTO processed_events "
"(session_id, event_id, processed_date, action, skip_reason) "
"VALUES (?, ?, ?, ?, ?)",
(session_id, event_id, processed_date, action, skip_reason),
)
# ------------------------------------------------------------------ #
# Trades
# ------------------------------------------------------------------ #
def record_trade(
self,
session_id: str,
symbol: str,
entry_date: str | None,
exit_date: str,
entry_price: float | None,
exit_price: float,
exit_reason: str,
shares: int,
net_pnl: float,
r_multiple: float,
holding_days: int,
) -> str:
trade_id = str(uuid.uuid4())
with self._connect() as conn:
conn.execute(
"INSERT INTO trades (trade_id, session_id, symbol, entry_date, exit_date, "
"entry_price, exit_price, exit_reason, shares, net_pnl, r_multiple, holding_days) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
trade_id, session_id, symbol, entry_date, exit_date,
entry_price, exit_price, exit_reason, shares,
net_pnl, r_multiple, holding_days,
),
)
return trade_id
def list_trades(self, session_id: str, limit: int | None = None) -> list[dict]:
with self._connect() as conn:
if limit:
rows = conn.execute(
"SELECT * FROM trades WHERE session_id = ? ORDER BY exit_date DESC LIMIT ?",
(session_id, limit),
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM trades WHERE session_id = ? ORDER BY exit_date",
(session_id,),
).fetchall()
return [dict(r) for r in rows]
# ------------------------------------------------------------------ #
# Daily snapshots
# ------------------------------------------------------------------ #
def save_daily_snapshot(self, row: DailySnapshotRow) -> None:
with self._connect() as conn:
conn.execute(
"""INSERT INTO daily_snapshots
(session_id, date, equity, cash, market_value, daily_pnl,
total_pnl, drawdown_pct, open_position_count)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(session_id, date) DO UPDATE SET
equity=excluded.equity,
cash=excluded.cash,
market_value=excluded.market_value,
daily_pnl=excluded.daily_pnl,
total_pnl=excluded.total_pnl,
drawdown_pct=excluded.drawdown_pct,
open_position_count=excluded.open_position_count""",
(
row.session_id, row.date, row.equity, row.cash, row.market_value,
row.daily_pnl, row.total_pnl, row.drawdown_pct, row.open_position_count,
),
)
def list_snapshots(self, session_id: str) -> list[dict]:
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_peak_equity(self, session_id: str, initial_equity: float) -> float:
with self._connect() as conn:
row = conn.execute(
"SELECT MAX(equity) FROM daily_snapshots WHERE session_id = ?",
(session_id,),
).fetchone()
if row and row[0] is not None:
return max(float(row[0]), initial_equity)
return initial_equity
# ------------------------------------------------------------------ #
# Idempotency: processed dates
# ------------------------------------------------------------------ #
def is_phase_processed(self, session_id: str, date: dt.date, phase: str) -> bool:
with self._connect() as conn:
row = conn.execute(
"SELECT 1 FROM processed_phases WHERE session_id = ? AND date = ? AND phase = ?",
(session_id, date.isoformat(), phase),
).fetchone()
return row is not None
def mark_phase_processed(self, session_id: str, date: dt.date, phase: str) -> None:
with self._connect() as conn:
conn.execute(
"INSERT OR IGNORE INTO processed_phases (session_id, date, phase) VALUES (?, ?, ?)",
(session_id, date.isoformat(), phase),
)
def is_date_processed(self, session_id: str, date: dt.date) -> bool:
with self._connect() as conn:
row = conn.execute(
"SELECT 1 FROM processed_dates WHERE session_id = ? AND date = ?",
(session_id, date.isoformat()),
).fetchone()
return row is not None
def mark_date_processed(self, session_id: str, date: dt.date) -> None:
with self._connect() as conn:
conn.execute(
"INSERT OR IGNORE INTO processed_dates (session_id, date) VALUES (?, ?)",
(session_id, date.isoformat()),
)
Write Preview
Loading…
Cancel
Save