feat: add playbook persistence with DB schema and CRUD store (issue #82)

- Add playbooks table to src/db.py with UNIQUE(date, market) constraint
- PlaybookStore: save/load/delete, status management, match_count tracking,
  list_recent with market filter, stats without full deserialization
- DayPlaybook JSON serialization via Pydantic model_dump_json/model_validate_json
- 23 tests, 100% coverage on playbook_store.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

docs/agent-constraints.md

@@ -0,0 +1,45 @@
+# Agent Constraints
+
+This document records **persistent behavioral constraints** for agents working on this repository.
+It is distinct from `docs/requirements-log.md`, which records **project/product requirements**.
+
+## Scope
+
+- Applies to all AI agents and automation that modify this repo.
+- Supplements (does not replace) `docs/agents.md` and `docs/workflow.md`.
+
+## Persistent Rules
+
+1. **Workflow enforcement**
+   - Follow `docs/workflow.md` for all changes.
+   - Create a Gitea issue before any code or documentation change.
+   - Work on a feature branch `feature/issue-{N}-{short-description}` and open a PR.
+   - Never commit directly to `main`.
+
+2. **Document-first routing**
+   - When performing work, consult relevant `docs/` files *before* making changes.
+   - Route decisions to the documented policy whenever applicable.
+   - If guidance conflicts, prefer the stricter/safety-first rule and note it in the PR.
+
+3. **Docs with code**
+   - Any code change must be accompanied by relevant documentation updates.
+   - If no doc update is needed, state the reason explicitly in the PR.
+
+4. **Session-persistent user constraints**
+   - If the user requests that a behavior should persist across sessions, record it here
+     (or in a dedicated policy doc) and reference it when working.
+   - Keep entries short and concrete, with dates.
+
+## Change Control
+
+- Changes to this file follow the same workflow as code changes.
+- Keep the history chronological and minimize rewording of existing entries.
+
+## History
+
+### 2026-02-08
+
+- Always enforce Gitea workflow: issue -> feature branch -> PR before changes.
+- When work requires guidance, consult the relevant `docs/` policies first.
+- Any code change must be accompanied by relevant documentation updates.
+- Persist user constraints across sessions by recording them in this document.

src/db.py

@@ -91,6 +91,27 @@ def init_db(db_path: str) -> sqlite3.Connection:
         """
     )
 
+    # Playbook storage for pre-market strategy persistence
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS playbooks (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            date TEXT NOT NULL,
+            market TEXT NOT NULL,
+            status TEXT NOT NULL DEFAULT 'pending',
+            playbook_json TEXT NOT NULL,
+            generated_at TEXT NOT NULL,
+            token_count INTEGER DEFAULT 0,
+            scenario_count INTEGER DEFAULT 0,
+            match_count INTEGER DEFAULT 0,
+            UNIQUE(date, market)
+        )
+        """
+    )
+
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_playbooks_date ON playbooks(date)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_playbooks_market ON playbooks(market)")
+
     # Create indices for efficient context queries
     conn.execute("CREATE INDEX IF NOT EXISTS idx_contexts_layer ON contexts(layer)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_contexts_timeframe ON contexts(timeframe)")

src/strategy/playbook_store.py

@@ -0,0 +1,184 @@
+"""Playbook persistence layer — CRUD for DayPlaybook in SQLite.
+
+Stores and retrieves market-specific daily playbooks with JSON serialization.
+Designed for the pre-market strategy system (one playbook per market per day).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import sqlite3
+from datetime import date
+
+from src.strategy.models import DayPlaybook, PlaybookStatus
+
+logger = logging.getLogger(__name__)
+
+
+class PlaybookStore:
+    """CRUD operations for DayPlaybook persistence."""
+
+    def __init__(self, conn: sqlite3.Connection) -> None:
+        self._conn = conn
+
+    def save(self, playbook: DayPlaybook) -> int:
+        """Save or replace a playbook for a given date+market.
+
+        Uses INSERT OR REPLACE to enforce UNIQUE(date, market).
+
+        Returns:
+            The row id of the inserted/replaced record.
+        """
+        playbook_json = playbook.model_dump_json()
+        cursor = self._conn.execute(
+            """
+            INSERT OR REPLACE INTO playbooks
+                (date, market, status, playbook_json, generated_at,
+                 token_count, scenario_count, match_count)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                playbook.date.isoformat(),
+                playbook.market,
+                PlaybookStatus.READY.value,
+                playbook_json,
+                playbook.generated_at,
+                playbook.token_count,
+                playbook.scenario_count,
+                0,
+            ),
+        )
+        self._conn.commit()
+        row_id = cursor.lastrowid or 0
+        logger.info(
+            "Saved playbook for %s/%s (%d stocks, %d scenarios)",
+            playbook.date, playbook.market,
+            playbook.stock_count, playbook.scenario_count,
+        )
+        return row_id
+
+    def load(self, target_date: date, market: str) -> DayPlaybook | None:
+        """Load a playbook for a specific date and market.
+
+        Returns:
+            DayPlaybook if found, None otherwise.
+        """
+        row = self._conn.execute(
+            "SELECT playbook_json FROM playbooks WHERE date = ? AND market = ?",
+            (target_date.isoformat(), market),
+        ).fetchone()
+        if row is None:
+            return None
+        return DayPlaybook.model_validate_json(row[0])
+
+    def get_status(self, target_date: date, market: str) -> PlaybookStatus | None:
+        """Get the status of a playbook without deserializing the full JSON."""
+        row = self._conn.execute(
+            "SELECT status FROM playbooks WHERE date = ? AND market = ?",
+            (target_date.isoformat(), market),
+        ).fetchone()
+        if row is None:
+            return None
+        return PlaybookStatus(row[0])
+
+    def update_status(self, target_date: date, market: str, status: PlaybookStatus) -> bool:
+        """Update the status of a playbook.
+
+        Returns:
+            True if a row was updated, False if not found.
+        """
+        cursor = self._conn.execute(
+            "UPDATE playbooks SET status = ? WHERE date = ? AND market = ?",
+            (status.value, target_date.isoformat(), market),
+        )
+        self._conn.commit()
+        return cursor.rowcount > 0
+
+    def increment_match_count(self, target_date: date, market: str) -> bool:
+        """Increment the match_count for tracking scenario hits during the day.
+
+        Returns:
+            True if a row was updated, False if not found.
+        """
+        cursor = self._conn.execute(
+            "UPDATE playbooks SET match_count = match_count + 1 WHERE date = ? AND market = ?",
+            (target_date.isoformat(), market),
+        )
+        self._conn.commit()
+        return cursor.rowcount > 0
+
+    def get_stats(self, target_date: date, market: str) -> dict | None:
+        """Get playbook stats without full deserialization.
+
+        Returns:
+            Dict with status, token_count, scenario_count, match_count, or None.
+        """
+        row = self._conn.execute(
+            """
+            SELECT status, token_count, scenario_count, match_count, generated_at
+            FROM playbooks WHERE date = ? AND market = ?
+            """,
+            (target_date.isoformat(), market),
+        ).fetchone()
+        if row is None:
+            return None
+        return {
+            "status": row[0],
+            "token_count": row[1],
+            "scenario_count": row[2],
+            "match_count": row[3],
+            "generated_at": row[4],
+        }
+
+    def list_recent(self, market: str | None = None, limit: int = 7) -> list[dict]:
+        """List recent playbooks with summary info.
+
+        Args:
+            market: Filter by market code. None for all markets.
+            limit: Max number of results.
+
+        Returns:
+            List of dicts with date, market, status, scenario_count, match_count.
+        """
+        if market is not None:
+            rows = self._conn.execute(
+                """
+                SELECT date, market, status, scenario_count, match_count
+                FROM playbooks WHERE market = ?
+                ORDER BY date DESC LIMIT ?
+                """,
+                (market, limit),
+            ).fetchall()
+        else:
+            rows = self._conn.execute(
+                """
+                SELECT date, market, status, scenario_count, match_count
+                FROM playbooks
+                ORDER BY date DESC LIMIT ?
+                """,
+                (limit,),
+            ).fetchall()
+        return [
+            {
+                "date": row[0],
+                "market": row[1],
+                "status": row[2],
+                "scenario_count": row[3],
+                "match_count": row[4],
+            }
+            for row in rows
+        ]
+
+    def delete(self, target_date: date, market: str) -> bool:
+        """Delete a playbook.
+
+        Returns:
+            True if a row was deleted, False if not found.
+        """
+        cursor = self._conn.execute(
+            "DELETE FROM playbooks WHERE date = ? AND market = ?",
+            (target_date.isoformat(), market),
+        )
+        self._conn.commit()
+        return cursor.rowcount > 0

src/strategy/scenario_engine.py

@@ -0,0 +1,270 @@
+"""Local scenario engine for playbook execution.
+
+Matches real-time market conditions against pre-defined scenarios
+without any API calls. Designed for sub-100ms execution.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+from src.strategy.models import (
+    DayPlaybook,
+    GlobalRule,
+    ScenarioAction,
+    StockCondition,
+    StockScenario,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ScenarioMatch:
+    """Result of matching market conditions against scenarios."""
+
+    stock_code: str
+    matched_scenario: StockScenario | None
+    action: ScenarioAction
+    confidence: int
+    rationale: str
+    global_rule_triggered: GlobalRule | None = None
+    match_details: dict[str, Any] = field(default_factory=dict)
+
+
+class ScenarioEngine:
+    """Evaluates playbook scenarios against real-time market data.
+
+    No API calls — pure Python condition matching.
+
+    Expected market_data keys: "rsi", "volume_ratio", "current_price", "price_change_pct".
+    Callers must normalize data source keys to match this contract.
+    """
+
+    def __init__(self) -> None:
+        self._warned_keys: set[str] = set()
+
+    @staticmethod
+    def _safe_float(value: Any) -> float | None:
+        """Safely cast a value to float. Returns None on failure."""
+        if value is None:
+            return None
+        try:
+            return float(value)
+        except (ValueError, TypeError):
+            return None
+
+    def _warn_missing_key(self, key: str) -> None:
+        """Log a missing-key warning once per key per engine instance."""
+        if key not in self._warned_keys:
+            self._warned_keys.add(key)
+            logger.warning("Condition requires '%s' but key missing from market_data", key)
+
+    def evaluate(
+        self,
+        playbook: DayPlaybook,
+        stock_code: str,
+        market_data: dict[str, Any],
+        portfolio_data: dict[str, Any],
+    ) -> ScenarioMatch:
+        """Match market conditions to scenarios and return a decision.
+
+        Algorithm:
+        1. Check global rules first (portfolio-level circuit breakers)
+        2. Find the StockPlaybook for the given stock_code
+        3. Iterate scenarios in order (first match wins)
+        4. If no match, return playbook.default_action (HOLD)
+
+        Args:
+            playbook: Today's DayPlaybook for this market
+            stock_code: Stock ticker to evaluate
+            market_data: Real-time market data (price, rsi, volume_ratio, etc.)
+            portfolio_data: Portfolio state (pnl_pct, total_cash, etc.)
+
+        Returns:
+            ScenarioMatch with the decision
+        """
+        # 1. Check global rules
+        triggered_rule = self.check_global_rules(playbook, portfolio_data)
+        if triggered_rule is not None:
+            logger.info(
+                "Global rule triggered for %s: %s -> %s",
+                stock_code,
+                triggered_rule.condition,
+                triggered_rule.action.value,
+            )
+            return ScenarioMatch(
+                stock_code=stock_code,
+                matched_scenario=None,
+                action=triggered_rule.action,
+                confidence=100,
+                rationale=f"Global rule: {triggered_rule.rationale or triggered_rule.condition}",
+                global_rule_triggered=triggered_rule,
+            )
+
+        # 2. Find stock playbook
+        stock_pb = playbook.get_stock_playbook(stock_code)
+        if stock_pb is None:
+            logger.debug("No playbook for %s — defaulting to %s", stock_code, playbook.default_action)
+            return ScenarioMatch(
+                stock_code=stock_code,
+                matched_scenario=None,
+                action=playbook.default_action,
+                confidence=0,
+                rationale=f"No scenarios defined for {stock_code}",
+            )
+
+        # 3. Iterate scenarios (first match wins)
+        for scenario in stock_pb.scenarios:
+            if self.evaluate_condition(scenario.condition, market_data):
+                logger.info(
+                    "Scenario matched for %s: %s (confidence=%d)",
+                    stock_code,
+                    scenario.action.value,
+                    scenario.confidence,
+                )
+                return ScenarioMatch(
+                    stock_code=stock_code,
+                    matched_scenario=scenario,
+                    action=scenario.action,
+                    confidence=scenario.confidence,
+                    rationale=scenario.rationale,
+                    match_details=self._build_match_details(scenario.condition, market_data),
+                )
+
+        # 4. No match — default action
+        logger.debug("No scenario matched for %s — defaulting to %s", stock_code, playbook.default_action)
+        return ScenarioMatch(
+            stock_code=stock_code,
+            matched_scenario=None,
+            action=playbook.default_action,
+            confidence=0,
+            rationale="No scenario conditions met — holding position",
+        )
+
+    def check_global_rules(
+        self,
+        playbook: DayPlaybook,
+        portfolio_data: dict[str, Any],
+    ) -> GlobalRule | None:
+        """Check portfolio-level rules. Returns first triggered rule or None."""
+        for rule in playbook.global_rules:
+            if self._evaluate_global_condition(rule.condition, portfolio_data):
+                return rule
+        return None
+
+    def evaluate_condition(
+        self,
+        condition: StockCondition,
+        market_data: dict[str, Any],
+    ) -> bool:
+        """Evaluate all non-None fields in condition as AND.
+
+        Returns True only if ALL specified conditions are met.
+        Empty condition (no fields set) returns False for safety.
+        """
+        if not condition.has_any_condition():
+            return False
+
+        checks: list[bool] = []
+
+        rsi = self._safe_float(market_data.get("rsi"))
+        if condition.rsi_below is not None or condition.rsi_above is not None:
+            if "rsi" not in market_data:
+                self._warn_missing_key("rsi")
+        if condition.rsi_below is not None:
+            checks.append(rsi is not None and rsi < condition.rsi_below)
+        if condition.rsi_above is not None:
+            checks.append(rsi is not None and rsi > condition.rsi_above)
+
+        volume_ratio = self._safe_float(market_data.get("volume_ratio"))
+        if condition.volume_ratio_above is not None or condition.volume_ratio_below is not None:
+            if "volume_ratio" not in market_data:
+                self._warn_missing_key("volume_ratio")
+        if condition.volume_ratio_above is not None:
+            checks.append(volume_ratio is not None and volume_ratio > condition.volume_ratio_above)
+        if condition.volume_ratio_below is not None:
+            checks.append(volume_ratio is not None and volume_ratio < condition.volume_ratio_below)
+
+        price = self._safe_float(market_data.get("current_price"))
+        if condition.price_above is not None or condition.price_below is not None:
+            if "current_price" not in market_data:
+                self._warn_missing_key("current_price")
+        if condition.price_above is not None:
+            checks.append(price is not None and price > condition.price_above)
+        if condition.price_below is not None:
+            checks.append(price is not None and price < condition.price_below)
+
+        price_change_pct = self._safe_float(market_data.get("price_change_pct"))
+        if condition.price_change_pct_above is not None or condition.price_change_pct_below is not None:
+            if "price_change_pct" not in market_data:
+                self._warn_missing_key("price_change_pct")
+        if condition.price_change_pct_above is not None:
+            checks.append(price_change_pct is not None and price_change_pct > condition.price_change_pct_above)
+        if condition.price_change_pct_below is not None:
+            checks.append(price_change_pct is not None and price_change_pct < condition.price_change_pct_below)
+
+        return len(checks) > 0 and all(checks)
+
+    def _evaluate_global_condition(
+        self,
+        condition_str: str,
+        portfolio_data: dict[str, Any],
+    ) -> bool:
+        """Evaluate a simple global condition string against portfolio data.
+
+        Supports: "field < value", "field > value", "field <= value", "field >= value"
+        """
+        parts = condition_str.strip().split()
+        if len(parts) != 3:
+            logger.warning("Invalid global condition format: %s", condition_str)
+            return False
+
+        field_name, operator, value_str = parts
+        try:
+            threshold = float(value_str)
+        except ValueError:
+            logger.warning("Invalid threshold in condition: %s", condition_str)
+            return False
+
+        actual = portfolio_data.get(field_name)
+        if actual is None:
+            return False
+
+        try:
+            actual_val = float(actual)
+        except (ValueError, TypeError):
+            return False
+
+        if operator == "<":
+            return actual_val < threshold
+        elif operator == ">":
+            return actual_val > threshold
+        elif operator == "<=":
+            return actual_val <= threshold
+        elif operator == ">=":
+            return actual_val >= threshold
+        else:
+            logger.warning("Unknown operator in condition: %s", operator)
+            return False
+
+    def _build_match_details(
+        self,
+        condition: StockCondition,
+        market_data: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Build a summary of which conditions matched and their normalized values."""
+        details: dict[str, Any] = {}
+
+        if condition.rsi_below is not None or condition.rsi_above is not None:
+            details["rsi"] = self._safe_float(market_data.get("rsi"))
+        if condition.volume_ratio_above is not None or condition.volume_ratio_below is not None:
+            details["volume_ratio"] = self._safe_float(market_data.get("volume_ratio"))
+        if condition.price_above is not None or condition.price_below is not None:
+            details["current_price"] = self._safe_float(market_data.get("current_price"))
+        if condition.price_change_pct_above is not None or condition.price_change_pct_below is not None:
+            details["price_change_pct"] = self._safe_float(market_data.get("price_change_pct"))
+
+        return details

tests/test_playbook_store.py

@@ -0,0 +1,289 @@
+"""Tests for playbook persistence (PlaybookStore + DB schema)."""
+
+from __future__ import annotations
+
+from datetime import date
+
+import pytest
+
+from src.db import init_db
+from src.strategy.models import (
+    DayPlaybook,
+    GlobalRule,
+    MarketOutlook,
+    PlaybookStatus,
+    ScenarioAction,
+    StockCondition,
+    StockPlaybook,
+    StockScenario,
+)
+from src.strategy.playbook_store import PlaybookStore
+
+
+@pytest.fixture
+def conn():
+    """Create an in-memory DB with schema."""
+    connection = init_db(":memory:")
+    yield connection
+    connection.close()
+
+
+@pytest.fixture
+def store(conn) -> PlaybookStore:
+    return PlaybookStore(conn)
+
+
+def _make_playbook(
+    target_date: date = date(2026, 2, 8),
+    market: str = "KR",
+    outlook: MarketOutlook = MarketOutlook.NEUTRAL,
+    stock_codes: list[str] | None = None,
+) -> DayPlaybook:
+    """Create a test playbook with sensible defaults."""
+    if stock_codes is None:
+        stock_codes = ["005930"]
+    return DayPlaybook(
+        date=target_date,
+        market=market,
+        market_outlook=outlook,
+        token_count=150,
+        stock_playbooks=[
+            StockPlaybook(
+                stock_code=code,
+                scenarios=[
+                    StockScenario(
+                        condition=StockCondition(rsi_below=30.0),
+                        action=ScenarioAction.BUY,
+                        confidence=85,
+                        rationale=f"Oversold bounce for {code}",
+                    ),
+                ],
+            )
+            for code in stock_codes
+        ],
+        global_rules=[
+            GlobalRule(
+                condition="portfolio_pnl_pct < -2.0",
+                action=ScenarioAction.REDUCE_ALL,
+                rationale="Near circuit breaker",
+            ),
+        ],
+    )
+
+
+# ---------------------------------------------------------------------------
+# Schema
+# ---------------------------------------------------------------------------
+
+
+class TestSchema:
+    def test_playbooks_table_exists(self, conn) -> None:
+        row = conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' AND name='playbooks'"
+        ).fetchone()
+        assert row is not None
+
+    def test_unique_constraint(self, store: PlaybookStore) -> None:
+        pb = _make_playbook()
+        store.save(pb)
+        # Saving again for same date+market should replace, not error
+        pb2 = _make_playbook(stock_codes=["005930", "000660"])
+        store.save(pb2)
+        loaded = store.load(date(2026, 2, 8), "KR")
+        assert loaded is not None
+        assert loaded.stock_count == 2
+
+
+# ---------------------------------------------------------------------------
+# Save / Load
+# ---------------------------------------------------------------------------
+
+
+class TestSaveLoad:
+    def test_save_and_load(self, store: PlaybookStore) -> None:
+        pb = _make_playbook()
+        row_id = store.save(pb)
+        assert row_id > 0
+
+        loaded = store.load(date(2026, 2, 8), "KR")
+        assert loaded is not None
+        assert loaded.date == date(2026, 2, 8)
+        assert loaded.market == "KR"
+        assert loaded.stock_count == 1
+        assert loaded.scenario_count == 1
+
+    def test_load_not_found(self, store: PlaybookStore) -> None:
+        result = store.load(date(2026, 1, 1), "KR")
+        assert result is None
+
+    def test_save_preserves_all_fields(self, store: PlaybookStore) -> None:
+        pb = _make_playbook(
+            outlook=MarketOutlook.BULLISH,
+            stock_codes=["005930", "AAPL"],
+        )
+        store.save(pb)
+        loaded = store.load(date(2026, 2, 8), "KR")
+        assert loaded is not None
+        assert loaded.market_outlook == MarketOutlook.BULLISH
+        assert loaded.stock_count == 2
+        assert loaded.global_rules[0].action == ScenarioAction.REDUCE_ALL
+        assert loaded.token_count == 150
+
+    def test_save_different_markets(self, store: PlaybookStore) -> None:
+        kr = _make_playbook(market="KR")
+        us = _make_playbook(market="US", stock_codes=["AAPL"])
+        store.save(kr)
+        store.save(us)
+
+        kr_loaded = store.load(date(2026, 2, 8), "KR")
+        us_loaded = store.load(date(2026, 2, 8), "US")
+        assert kr_loaded is not None
+        assert us_loaded is not None
+        assert kr_loaded.market == "KR"
+        assert us_loaded.market == "US"
+        assert kr_loaded.stock_playbooks[0].stock_code == "005930"
+        assert us_loaded.stock_playbooks[0].stock_code == "AAPL"
+
+    def test_save_different_dates(self, store: PlaybookStore) -> None:
+        d1 = _make_playbook(target_date=date(2026, 2, 7))
+        d2 = _make_playbook(target_date=date(2026, 2, 8))
+        store.save(d1)
+        store.save(d2)
+
+        assert store.load(date(2026, 2, 7), "KR") is not None
+        assert store.load(date(2026, 2, 8), "KR") is not None
+
+    def test_replace_updates_data(self, store: PlaybookStore) -> None:
+        pb1 = _make_playbook(outlook=MarketOutlook.BEARISH)
+        store.save(pb1)
+
+        pb2 = _make_playbook(outlook=MarketOutlook.BULLISH)
+        store.save(pb2)
+
+        loaded = store.load(date(2026, 2, 8), "KR")
+        assert loaded is not None
+        assert loaded.market_outlook == MarketOutlook.BULLISH
+
+
+# ---------------------------------------------------------------------------
+# Status
+# ---------------------------------------------------------------------------
+
+
+class TestStatus:
+    def test_get_status(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook())
+        status = store.get_status(date(2026, 2, 8), "KR")
+        assert status == PlaybookStatus.READY
+
+    def test_get_status_not_found(self, store: PlaybookStore) -> None:
+        assert store.get_status(date(2026, 1, 1), "KR") is None
+
+    def test_update_status(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook())
+        updated = store.update_status(date(2026, 2, 8), "KR", PlaybookStatus.EXPIRED)
+        assert updated is True
+
+        status = store.get_status(date(2026, 2, 8), "KR")
+        assert status == PlaybookStatus.EXPIRED
+
+    def test_update_status_not_found(self, store: PlaybookStore) -> None:
+        updated = store.update_status(date(2026, 1, 1), "KR", PlaybookStatus.FAILED)
+        assert updated is False
+
+
+# ---------------------------------------------------------------------------
+# Match count
+# ---------------------------------------------------------------------------
+
+
+class TestMatchCount:
+    def test_increment_match_count(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook())
+        store.increment_match_count(date(2026, 2, 8), "KR")
+        store.increment_match_count(date(2026, 2, 8), "KR")
+
+        stats = store.get_stats(date(2026, 2, 8), "KR")
+        assert stats is not None
+        assert stats["match_count"] == 2
+
+    def test_increment_not_found(self, store: PlaybookStore) -> None:
+        result = store.increment_match_count(date(2026, 1, 1), "KR")
+        assert result is False
+
+
+# ---------------------------------------------------------------------------
+# Stats
+# ---------------------------------------------------------------------------
+
+
+class TestStats:
+    def test_get_stats(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook())
+        stats = store.get_stats(date(2026, 2, 8), "KR")
+        assert stats is not None
+        assert stats["status"] == "ready"
+        assert stats["token_count"] == 150
+        assert stats["scenario_count"] == 1
+        assert stats["match_count"] == 0
+        assert stats["generated_at"] != ""
+
+    def test_get_stats_not_found(self, store: PlaybookStore) -> None:
+        assert store.get_stats(date(2026, 1, 1), "KR") is None
+
+
+# ---------------------------------------------------------------------------
+# List recent
+# ---------------------------------------------------------------------------
+
+
+class TestListRecent:
+    def test_list_recent(self, store: PlaybookStore) -> None:
+        for day in range(5, 10):
+            store.save(_make_playbook(target_date=date(2026, 2, day)))
+        results = store.list_recent(market="KR", limit=3)
+        assert len(results) == 3
+        # Most recent first
+        assert results[0]["date"] == "2026-02-09"
+        assert results[2]["date"] == "2026-02-07"
+
+    def test_list_recent_all_markets(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook(market="KR"))
+        store.save(_make_playbook(market="US", stock_codes=["AAPL"]))
+        results = store.list_recent(market=None, limit=10)
+        assert len(results) == 2
+
+    def test_list_recent_empty(self, store: PlaybookStore) -> None:
+        results = store.list_recent(market="KR")
+        assert results == []
+
+    def test_list_recent_filter_by_market(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook(market="KR"))
+        store.save(_make_playbook(market="US", stock_codes=["AAPL"]))
+        kr_only = store.list_recent(market="KR")
+        assert len(kr_only) == 1
+        assert kr_only[0]["market"] == "KR"
+
+
+# ---------------------------------------------------------------------------
+# Delete
+# ---------------------------------------------------------------------------
+
+
+class TestDelete:
+    def test_delete(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook())
+        deleted = store.delete(date(2026, 2, 8), "KR")
+        assert deleted is True
+        assert store.load(date(2026, 2, 8), "KR") is None
+
+    def test_delete_not_found(self, store: PlaybookStore) -> None:
+        deleted = store.delete(date(2026, 1, 1), "KR")
+        assert deleted is False
+
+    def test_delete_one_market_keeps_other(self, store: PlaybookStore) -> None:
+        store.save(_make_playbook(market="KR"))
+        store.save(_make_playbook(market="US", stock_codes=["AAPL"]))
+        store.delete(date(2026, 2, 8), "KR")
+        assert store.load(date(2026, 2, 8), "KR") is None
+        assert store.load(date(2026, 2, 8), "US") is not None

tests/test_scenario_engine.py

@@ -0,0 +1,442 @@
+"""Tests for the local scenario engine."""
+
+from __future__ import annotations
+
+from datetime import date
+
+import pytest
+
+from src.strategy.models import (
+    DayPlaybook,
+    GlobalRule,
+    ScenarioAction,
+    StockCondition,
+    StockPlaybook,
+    StockScenario,
+)
+from src.strategy.scenario_engine import ScenarioEngine, ScenarioMatch
+
+
+@pytest.fixture
+def engine() -> ScenarioEngine:
+    return ScenarioEngine()
+
+
+def _scenario(
+    rsi_below: float | None = None,
+    rsi_above: float | None = None,
+    volume_ratio_above: float | None = None,
+    action: ScenarioAction = ScenarioAction.BUY,
+    confidence: int = 85,
+    **kwargs,
+) -> StockScenario:
+    return StockScenario(
+        condition=StockCondition(
+            rsi_below=rsi_below,
+            rsi_above=rsi_above,
+            volume_ratio_above=volume_ratio_above,
+            **kwargs,
+        ),
+        action=action,
+        confidence=confidence,
+        rationale=f"Test scenario: {action.value}",
+    )
+
+
+def _playbook(
+    stock_code: str = "005930",
+    scenarios: list[StockScenario] | None = None,
+    global_rules: list[GlobalRule] | None = None,
+    default_action: ScenarioAction = ScenarioAction.HOLD,
+) -> DayPlaybook:
+    if scenarios is None:
+        scenarios = [_scenario(rsi_below=30.0)]
+    return DayPlaybook(
+        date=date(2026, 2, 7),
+        market="KR",
+        stock_playbooks=[StockPlaybook(stock_code=stock_code, scenarios=scenarios)],
+        global_rules=global_rules or [],
+        default_action=default_action,
+    )
+
+
+# ---------------------------------------------------------------------------
+# evaluate_condition
+# ---------------------------------------------------------------------------
+
+
+class TestEvaluateCondition:
+    def test_rsi_below_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(rsi_below=30.0)
+        assert engine.evaluate_condition(cond, {"rsi": 25.0})
+
+    def test_rsi_below_no_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(rsi_below=30.0)
+        assert not engine.evaluate_condition(cond, {"rsi": 35.0})
+
+    def test_rsi_above_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(rsi_above=70.0)
+        assert engine.evaluate_condition(cond, {"rsi": 75.0})
+
+    def test_rsi_above_no_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(rsi_above=70.0)
+        assert not engine.evaluate_condition(cond, {"rsi": 65.0})
+
+    def test_volume_ratio_above_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(volume_ratio_above=3.0)
+        assert engine.evaluate_condition(cond, {"volume_ratio": 4.5})
+
+    def test_volume_ratio_below_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(volume_ratio_below=1.0)
+        assert engine.evaluate_condition(cond, {"volume_ratio": 0.5})
+
+    def test_price_above_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(price_above=50000)
+        assert engine.evaluate_condition(cond, {"current_price": 55000})
+
+    def test_price_below_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(price_below=50000)
+        assert engine.evaluate_condition(cond, {"current_price": 45000})
+
+    def test_price_change_pct_above_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(price_change_pct_above=2.0)
+        assert engine.evaluate_condition(cond, {"price_change_pct": 3.5})
+
+    def test_price_change_pct_below_match(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(price_change_pct_below=-3.0)
+        assert engine.evaluate_condition(cond, {"price_change_pct": -4.0})
+
+    def test_multiple_conditions_and_logic(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(rsi_below=30.0, volume_ratio_above=3.0)
+        # Both met
+        assert engine.evaluate_condition(cond, {"rsi": 25.0, "volume_ratio": 4.0})
+        # Only RSI met
+        assert not engine.evaluate_condition(cond, {"rsi": 25.0, "volume_ratio": 2.0})
+        # Only volume met
+        assert not engine.evaluate_condition(cond, {"rsi": 35.0, "volume_ratio": 4.0})
+        # Neither met
+        assert not engine.evaluate_condition(cond, {"rsi": 35.0, "volume_ratio": 2.0})
+
+    def test_empty_condition_returns_false(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition()
+        assert not engine.evaluate_condition(cond, {"rsi": 25.0})
+
+    def test_missing_data_returns_false(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(rsi_below=30.0)
+        assert not engine.evaluate_condition(cond, {})
+
+    def test_none_data_returns_false(self, engine: ScenarioEngine) -> None:
+        cond = StockCondition(rsi_below=30.0)
+        assert not engine.evaluate_condition(cond, {"rsi": None})
+
+    def test_boundary_value_not_matched(self, engine: ScenarioEngine) -> None:
+        """rsi_below=30 should NOT match rsi=30 (strict less than)."""
+        cond = StockCondition(rsi_below=30.0)
+        assert not engine.evaluate_condition(cond, {"rsi": 30.0})
+
+    def test_boundary_value_above_not_matched(self, engine: ScenarioEngine) -> None:
+        """rsi_above=70 should NOT match rsi=70 (strict greater than)."""
+        cond = StockCondition(rsi_above=70.0)
+        assert not engine.evaluate_condition(cond, {"rsi": 70.0})
+
+    def test_string_value_no_exception(self, engine: ScenarioEngine) -> None:
+        """String numeric value should not raise TypeError."""
+        cond = StockCondition(rsi_below=30.0)
+        # "25" can be cast to float → should match
+        assert engine.evaluate_condition(cond, {"rsi": "25"})
+        # "35" → should not match
+        assert not engine.evaluate_condition(cond, {"rsi": "35"})
+
+    def test_percent_string_returns_false(self, engine: ScenarioEngine) -> None:
+        """Percent string like '30%' cannot be cast to float → False, no exception."""
+        cond = StockCondition(rsi_below=30.0)
+        assert not engine.evaluate_condition(cond, {"rsi": "30%"})
+
+    def test_decimal_value_no_exception(self, engine: ScenarioEngine) -> None:
+        """Decimal values should be safely handled."""
+        from decimal import Decimal
+
+        cond = StockCondition(rsi_below=30.0)
+        assert engine.evaluate_condition(cond, {"rsi": Decimal("25.0")})
+
+    def test_mixed_invalid_types_no_exception(self, engine: ScenarioEngine) -> None:
+        """Various invalid types should not raise exceptions."""
+        cond = StockCondition(
+            rsi_below=30.0, volume_ratio_above=2.0,
+            price_above=100, price_change_pct_below=-1.0,
+        )
+        data = {
+            "rsi": [25],           # list
+            "volume_ratio": "bad",  # non-numeric string
+            "current_price": {},    # dict
+            "price_change_pct": object(),  # arbitrary object
+        }
+        # Should return False (invalid types → None → False), never raise
+        assert not engine.evaluate_condition(cond, data)
+
+    def test_missing_key_logs_warning_once(self, caplog) -> None:
+        """Missing key warning should fire only once per key per engine instance."""
+        import logging
+
+        eng = ScenarioEngine()
+        cond = StockCondition(rsi_below=30.0)
+        with caplog.at_level(logging.WARNING):
+            eng.evaluate_condition(cond, {})
+            eng.evaluate_condition(cond, {})
+            eng.evaluate_condition(cond, {})
+        # Warning should appear exactly once despite 3 calls
+        assert caplog.text.count("'rsi' but key missing") == 1
+
+
+# ---------------------------------------------------------------------------
+# check_global_rules
+# ---------------------------------------------------------------------------
+
+
+class TestCheckGlobalRules:
+    def test_no_rules(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(global_rules=[])
+        result = engine.check_global_rules(pb, {"portfolio_pnl_pct": -1.0})
+        assert result is None
+
+    def test_rule_triggered(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(
+                    condition="portfolio_pnl_pct < -2.0",
+                    action=ScenarioAction.REDUCE_ALL,
+                    rationale="Near circuit breaker",
+                ),
+            ]
+        )
+        result = engine.check_global_rules(pb, {"portfolio_pnl_pct": -2.5})
+        assert result is not None
+        assert result.action == ScenarioAction.REDUCE_ALL
+
+    def test_rule_not_triggered(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(
+                    condition="portfolio_pnl_pct < -2.0",
+                    action=ScenarioAction.REDUCE_ALL,
+                ),
+            ]
+        )
+        result = engine.check_global_rules(pb, {"portfolio_pnl_pct": -1.0})
+        assert result is None
+
+    def test_first_rule_wins(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(condition="portfolio_pnl_pct < -2.0", action=ScenarioAction.REDUCE_ALL),
+                GlobalRule(condition="portfolio_pnl_pct < -1.0", action=ScenarioAction.HOLD),
+            ]
+        )
+        result = engine.check_global_rules(pb, {"portfolio_pnl_pct": -2.5})
+        assert result is not None
+        assert result.action == ScenarioAction.REDUCE_ALL
+
+    def test_greater_than_operator(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(condition="volatility_index > 30", action=ScenarioAction.HOLD),
+            ]
+        )
+        result = engine.check_global_rules(pb, {"volatility_index": 35})
+        assert result is not None
+
+    def test_missing_field_not_triggered(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(condition="unknown_field < -2.0", action=ScenarioAction.REDUCE_ALL),
+            ]
+        )
+        result = engine.check_global_rules(pb, {"portfolio_pnl_pct": -5.0})
+        assert result is None
+
+    def test_invalid_condition_format(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(condition="bad format", action=ScenarioAction.HOLD),
+            ]
+        )
+        result = engine.check_global_rules(pb, {})
+        assert result is None
+
+    def test_le_operator(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(condition="portfolio_pnl_pct <= -2.0", action=ScenarioAction.REDUCE_ALL),
+            ]
+        )
+        assert engine.check_global_rules(pb, {"portfolio_pnl_pct": -2.0}) is not None
+        assert engine.check_global_rules(pb, {"portfolio_pnl_pct": -1.9}) is None
+
+    def test_ge_operator(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            global_rules=[
+                GlobalRule(condition="volatility >= 80.0", action=ScenarioAction.HOLD),
+            ]
+        )
+        assert engine.check_global_rules(pb, {"volatility": 80.0}) is not None
+        assert engine.check_global_rules(pb, {"volatility": 79.9}) is None
+
+
+# ---------------------------------------------------------------------------
+# evaluate (full pipeline)
+# ---------------------------------------------------------------------------
+
+
+class TestEvaluate:
+    def test_scenario_match(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(scenarios=[_scenario(rsi_below=30.0)])
+        result = engine.evaluate(pb, "005930", {"rsi": 25.0}, {})
+        assert result.action == ScenarioAction.BUY
+        assert result.confidence == 85
+        assert result.matched_scenario is not None
+
+    def test_no_scenario_match_returns_default(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(scenarios=[_scenario(rsi_below=30.0)])
+        result = engine.evaluate(pb, "005930", {"rsi": 50.0}, {})
+        assert result.action == ScenarioAction.HOLD
+        assert result.confidence == 0
+        assert result.matched_scenario is None
+
+    def test_stock_not_in_playbook(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(stock_code="005930")
+        result = engine.evaluate(pb, "AAPL", {"rsi": 25.0}, {})
+        assert result.action == ScenarioAction.HOLD
+        assert result.confidence == 0
+
+    def test_global_rule_takes_priority(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            scenarios=[_scenario(rsi_below=30.0)],
+            global_rules=[
+                GlobalRule(
+                    condition="portfolio_pnl_pct < -2.0",
+                    action=ScenarioAction.REDUCE_ALL,
+                    rationale="Loss limit",
+                ),
+            ],
+        )
+        result = engine.evaluate(
+            pb,
+            "005930",
+            {"rsi": 25.0},  # Would match scenario
+            {"portfolio_pnl_pct": -2.5},  # But global rule triggers first
+        )
+        assert result.action == ScenarioAction.REDUCE_ALL
+        assert result.global_rule_triggered is not None
+        assert result.matched_scenario is None
+
+    def test_first_scenario_wins(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            scenarios=[
+                _scenario(rsi_below=30.0, action=ScenarioAction.BUY, confidence=90),
+                _scenario(rsi_below=25.0, action=ScenarioAction.BUY, confidence=95),
+            ]
+        )
+        result = engine.evaluate(pb, "005930", {"rsi": 20.0}, {})
+        # Both match, but first wins
+        assert result.confidence == 90
+
+    def test_sell_scenario(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            scenarios=[
+                _scenario(rsi_above=75.0, action=ScenarioAction.SELL, confidence=80),
+            ]
+        )
+        result = engine.evaluate(pb, "005930", {"rsi": 80.0}, {})
+        assert result.action == ScenarioAction.SELL
+
+    def test_empty_playbook(self, engine: ScenarioEngine) -> None:
+        pb = DayPlaybook(date=date(2026, 2, 7), market="KR", stock_playbooks=[])
+        result = engine.evaluate(pb, "005930", {"rsi": 25.0}, {})
+        assert result.action == ScenarioAction.HOLD
+
+    def test_match_details_populated(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(scenarios=[_scenario(rsi_below=30.0, volume_ratio_above=2.0)])
+        result = engine.evaluate(
+            pb, "005930", {"rsi": 25.0, "volume_ratio": 3.0}, {}
+        )
+        assert result.match_details.get("rsi") == 25.0
+        assert result.match_details.get("volume_ratio") == 3.0
+
+    def test_custom_default_action(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            scenarios=[_scenario(rsi_below=10.0)],  # Very unlikely to match
+            default_action=ScenarioAction.SELL,
+        )
+        result = engine.evaluate(pb, "005930", {"rsi": 50.0}, {})
+        assert result.action == ScenarioAction.SELL
+
+    def test_multiple_stocks_in_playbook(self, engine: ScenarioEngine) -> None:
+        pb = DayPlaybook(
+            date=date(2026, 2, 7),
+            market="US",
+            stock_playbooks=[
+                StockPlaybook(
+                    stock_code="AAPL",
+                    scenarios=[_scenario(rsi_below=25.0, confidence=90)],
+                ),
+                StockPlaybook(
+                    stock_code="MSFT",
+                    scenarios=[_scenario(rsi_above=75.0, action=ScenarioAction.SELL, confidence=80)],
+                ),
+            ],
+        )
+        aapl = engine.evaluate(pb, "AAPL", {"rsi": 20.0}, {})
+        assert aapl.action == ScenarioAction.BUY
+        assert aapl.confidence == 90
+
+        msft = engine.evaluate(pb, "MSFT", {"rsi": 80.0}, {})
+        assert msft.action == ScenarioAction.SELL
+
+    def test_complex_multi_condition(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(
+            scenarios=[
+                _scenario(
+                    rsi_below=30.0,
+                    volume_ratio_above=3.0,
+                    price_change_pct_below=-2.0,
+                    confidence=95,
+                ),
+            ]
+        )
+        # All conditions met
+        result = engine.evaluate(
+            pb,
+            "005930",
+            {"rsi": 22.0, "volume_ratio": 4.0, "price_change_pct": -3.0},
+            {},
+        )
+        assert result.action == ScenarioAction.BUY
+        assert result.confidence == 95
+
+        # One condition not met
+        result2 = engine.evaluate(
+            pb,
+            "005930",
+            {"rsi": 22.0, "volume_ratio": 4.0, "price_change_pct": -1.0},
+            {},
+        )
+        assert result2.action == ScenarioAction.HOLD
+
+    def test_scenario_match_returns_rationale(self, engine: ScenarioEngine) -> None:
+        pb = _playbook(scenarios=[_scenario(rsi_below=30.0)])
+        result = engine.evaluate(pb, "005930", {"rsi": 25.0}, {})
+        assert result.rationale != ""
+
+    def test_result_stock_code(self, engine: ScenarioEngine) -> None:
+        pb = _playbook()
+        result = engine.evaluate(pb, "005930", {"rsi": 25.0}, {})
+        assert result.stock_code == "005930"
+
+    def test_match_details_normalized(self, engine: ScenarioEngine) -> None:
+        """match_details should contain _safe_float normalized values, not raw."""
+        pb = _playbook(scenarios=[_scenario(rsi_below=30.0)])
+        # Pass string value — should be normalized to float in match_details
+        result = engine.evaluate(pb, "005930", {"rsi": "25.0"}, {})
+        assert result.action == ScenarioAction.BUY
+        assert result.match_details["rsi"] == 25.0
+        assert isinstance(result.match_details["rsi"], float)