diff --git a/apps/paper_trader/__init__.py b/apps/paper_trader/__init__.py
new file mode 100644
index 0000000..83d80c3
--- /dev/null
+++ b/apps/paper_trader/__init__.py
@@ -0,0 +1 @@
+"""Paper trading system with Alpaca API."""
diff --git a/apps/paper_trader/alpaca_broker.py b/apps/paper_trader/alpaca_broker.py
new file mode 100644
index 0000000..abcf61c
--- /dev/null
+++ b/apps/paper_trader/alpaca_broker.py
@@ -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,
+        )
diff --git a/apps/paper_trader/auto.py b/apps/paper_trader/auto.py
new file mode 100644
index 0000000..20ab7df
--- /dev/null
+++ b/apps/paper_trader/auto.py
@@ -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()
diff --git a/apps/paper_trader/mock_broker.py b/apps/paper_trader/mock_broker.py
new file mode 100644
index 0000000..d7129f1
--- /dev/null
+++ b/apps/paper_trader/mock_broker.py
@@ -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 []
diff --git a/apps/paper_trader/models.py b/apps/paper_trader/models.py
new file mode 100644
index 0000000..734f51e
--- /dev/null
+++ b/apps/paper_trader/models.py
@@ -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()
diff --git a/apps/paper_trader/reporter.py b/apps/paper_trader/reporter.py
new file mode 100644
index 0000000..121c6ef
--- /dev/null
+++ b/apps/paper_trader/reporter.py
@@ -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()
diff --git a/apps/paper_trader/state.py b/apps/paper_trader/state.py
new file mode 100644
index 0000000..e0d8f48
--- /dev/null
+++ b/apps/paper_trader/state.py
@@ -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()),
+            )