Compare commits

..

30 Commits

Author SHA1 Message Date
agentson
c4e31be27a feat: L7 real-time context write with market-scoped keys (issue #85)
Some checks failed
CI / test (pull_request) Has been cancelled
- Add L7_REALTIME writes in trading_cycle() for volatility, price, rsi, volume_ratio
- Normalize key format to {metric}_{market}_{stock_code} across scanner and main
- Fix existing key mismatch between scanner writes and main reads
- Remove unused MarketScanner dead code

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-10 04:21:52 +09:00
9d9ade14eb Merge pull request 'docs: add plan-implementation consistency check to code review checklist (#114)' (#115) from feature/issue-114-review-plan-consistency into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #115
2026-02-10 04:16:30 +09:00
agentson
9a8936ab34 docs: add plan-implementation consistency check to code review checklist (#114)
Some checks failed
CI / test (pull_request) Has been cancelled
리뷰 시 플랜과 구현의 일치 여부를 필수로 확인하는 규칙 추가.
- workflow.md에 Code Review Checklist 섹션 신설
- requirements-log.md에 사용자 요구사항 기록

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-10 04:15:51 +09:00
c5831966ed Merge pull request 'fix: derive all aggregation timeframes from trade timestamp (#112)' (#113) from fix/test-failures into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #113
2026-02-10 00:42:39 +09:00
agentson
f03cc6039b fix: derive all aggregation timeframes from trade timestamp (#112)
Some checks failed
CI / test (pull_request) Has been cancelled
run_all_aggregations() previously used datetime.now(UTC) for weekly
through annual layers while using the trade date only for daily,
causing data misalignment on backfill. Now all layers consistently
use the latest trade timestamp. Also adds "Z" suffix handling for
fromisoformat() compatibility and strengthens test assertions to
verify L4-L2 layer values end-to-end.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-10 00:40:28 +09:00
9171e54652 Merge pull request 'feat: integrate scenario engine and playbook into main trading loop (issue #84)' (#110) from feature/issue-84-main-integration into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #110
2026-02-09 23:18:24 +09:00
agentson
d64e072f06 fix: PR review — DB reload, market-local date, market-scoped scan_candidates
Some checks failed
CI / test (pull_request) Has been cancelled
Address PR #110 review findings:

1. High — Realtime mode now loads playbook from DB before calling Gemini,
   preventing duplicate API calls on process restart (4/day budget).
2. Medium — Pass market-local date (via market.timezone) to
   generate_playbook() and _empty_playbook() instead of date.today().
3. Medium — scan_candidates restructured from {stock_code: candidate}
   to {market_code: {stock_code: candidate}} to prevent KR/US symbol
   collision.

New test: test_scan_candidates_market_scoped verifies cross-market
isolation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-09 23:00:06 +09:00
agentson
b2312fbe01 fix: resolve lint issues in main.py and test_main.py
Some checks failed
CI / test (pull_request) Has been cancelled
Remove unused imports (sys, ScenarioMatch, asyncio, StockPlaybook),
fix import ordering, and split long lines for ruff compliance.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 22:28:31 +09:00
agentson
98c4a2413c feat: integrate scenario engine and playbook into main trading loop (issue #84)
Replace brain.decide() with scenario_engine.evaluate() in trading_cycle
and brain.decide_batch() with per-stock scenario evaluation in
run_daily_session. Initialize PreMarketPlanner, ScenarioEngine, and
PlaybookStore in run(). Add pre-market playbook generation on market
open (1 Gemini call per market per day), market_data enrichment from
scanner metrics (rsi, volume_ratio), portfolio_data for global rules,
scenario match notifications, and playbook lifecycle management.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 22:24:19 +09:00
6fba7c7ae8 Merge pull request 'feat: implement pre-market planner with Gemini integration (issue #83)' (#109) from feature/issue-83-pre-market-planner into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #109
2026-02-08 22:07:36 +09:00
agentson
be695a5d7c fix: address PR review — inject today param, remove unused imports, fix lint
Some checks failed
CI / test (pull_request) Has been cancelled
Review findings addressed:
- Finding 1 (ImportError): false positive — ContextLayer is re-exported from
  src.context.store, import works correctly at runtime
- Finding 2 (timezone): generate_playbook() and build_cross_market_context()
  now accept optional today parameter for market-local date injection
- Finding 3 (lint): removed unused imports (UTC, datetime, PlaybookStatus),
  fixed line-too-long in prompt template
- Tests simplified: replaced date patching with direct today= parameter

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 21:57:39 +09:00
agentson
6471e66d89 fix: correct Settings field name in planner tests (KIS_ACCOUNT_NO)
Some checks failed
CI / test (pull_request) Has been cancelled
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 21:36:42 +09:00
agentson
149039a904 feat: implement pre-market planner with Gemini integration (issue #83)
PreMarketPlanner generates DayPlaybook via single Gemini API call per market:
- Structured JSON prompt with scan candidates + strategic context
- Cross-market context (KR reads US scorecard, US reads KR scorecard)
- Robust JSON parser with markdown fence stripping
- Unknown stock filtering (only scanner candidates allowed)
- MAX_SCENARIOS_PER_STOCK enforcement
- Defensive playbook on failure (HOLD + stop-loss)
- Empty playbook when no candidates (safe, no trades)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 21:35:57 +09:00
815d675529 Merge pull request 'feat: add Telegram playbook notifications (issue #81)' (#108) from feature/issue-81-telegram-playbook-notify into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #108
2026-02-08 21:27:46 +09:00
agentson
e8634b93c3 feat: add Telegram playbook notifications (issue #81)
Some checks failed
CI / test (pull_request) Has been cancelled
- notify_playbook_generated(): market, stock/scenario count, token usage (MEDIUM)
- notify_scenario_matched(): stock, action, condition, confidence (HIGH)
- notify_playbook_failed(): market, reason with 200-char truncation (HIGH)
- 6 new tests: 3 format + 3 priority validations

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 21:25:16 +09:00
f20736fd2a Merge pull request 'feat: add playbook persistence with DB schema and CRUD store (issue #82)' (#107) from feature/issue-82-playbook-persistence into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #107
2026-02-08 21:07:13 +09:00
agentson
7f2f96a819 feat: add playbook persistence with DB schema and CRUD store (issue #82)
Some checks failed
CI / test (pull_request) Has been cancelled
- 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>
2026-02-08 21:00:04 +09:00
aaa74894dd Merge pull request 'feat: implement local scenario engine for playbook execution (issue #80)' (#102) from feature/issue-80-scenario-engine into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #102
2026-02-08 20:47:34 +09:00
agentson
e711d6702a fix: deduplicate missing-key warnings and normalize match_details
Some checks failed
CI / test (pull_request) Has been cancelled
Addresses second round of PR #102 review:
- _warn_missing_key(): logs each missing key only once per engine instance
  to prevent log spam in high-frequency trading loops
- _build_match_details(): uses _safe_float() normalized values instead of
  raw market_data to ensure consistent float types in logging/analysis
- Test: verify warning fires exactly once across repeated calls
- Test: verify match_details contains normalized float values

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 20:41:20 +09:00
agentson
d2fc829380 fix: add safe type casting and missing-key warnings in ScenarioEngine
Some checks failed
CI / test (pull_request) Has been cancelled
Addresses PR #102 review findings:
- _safe_float() prevents TypeError from str/Decimal/invalid market_data values
- Warning logs when condition references a key missing from market_data
- 5 new tests: string, percent string, Decimal, mixed invalid types, log check

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 16:23:54 +09:00
de27b1af10 Merge pull request 'Require rebase after creating feature branch' (#106) from feature/issue-105-branch-rebase into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #106
2026-02-08 16:04:57 +09:00
agentson
7370220497 Require rebase after creating feature branch
Some checks failed
CI / test (pull_request) Has been cancelled
2026-02-08 16:03:41 +09:00
b01dacf328 Merge pull request 'docs: add persistent agent constraints document (issue #100)' (#103) from feature/issue-100-agent-constraints into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #103
2026-02-08 15:12:19 +09:00
agentson
1210c17989 docs: add persistent agent constraints document (issue #100)
Some checks failed
CI / test (pull_request) Has been cancelled
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 15:10:49 +09:00
agentson
9599b188e8 feat: implement local scenario engine for playbook execution (issue #80)
Some checks failed
CI / test (pull_request) Has been cancelled
ScenarioEngine evaluates pre-defined playbook scenarios against real-time
market data with sub-100ms execution (zero API calls). Supports condition
AND-matching, global portfolio rules, and first-match-wins priority.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 02:23:53 +09:00
c43660a58c Merge pull request 'feat: add strategy/playbook Pydantic models (issue #79)' (#99) from feature/issue-79-strategy-models into main
Some checks failed
CI / test (push) Has been cancelled
2026-02-08 02:19:48 +09:00
agentson
7fd48c7764 feat: add strategy/playbook Pydantic models (issue #79)
Some checks failed
CI / test (pull_request) Has been cancelled
Define data contracts for the proactive strategy system:
- StockCondition: AND-combined condition fields (RSI, volume, price)
- StockScenario: condition-action rules with stop loss/take profit
- StockPlaybook: per-stock scenario collection
- GlobalRule: portfolio-level rules (e.g. REDUCE_ALL on loss limit)
- DayPlaybook: complete daily playbook per market with validation
- CrossMarketContext: cross-market awareness (KR↔US)
- ScenarioAction, MarketOutlook, PlaybookStatus enums

33 tests covering validation, serialization, edge cases.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 02:06:16 +09:00
a105bb7c1a Merge pull request 'feat: add pre-market planner config and remove static watchlists (issue #78)' (#98) from feature/issue-78-config-watchlist-removal into main
Some checks failed
CI / test (push) Has been cancelled
2026-02-08 02:04:23 +09:00
agentson
1a34a74232 feat: add pre-market planner config and remove static watchlists (issue #78)
Some checks failed
CI / test (pull_request) Has been cancelled
- Add pre-market planner settings: PRE_MARKET_MINUTES, MAX_SCENARIOS_PER_STOCK,
  PLANNER_TIMEOUT_SECONDS, DEFENSIVE_PLAYBOOK_ON_FAILURE, RESCAN_INTERVAL_SECONDS
- Change ENABLED_MARKETS default from KR to KR,US
- Remove static WATCHLISTS and STOCK_UNIVERSE dictionaries from main.py
- Replace watchlist-based trading with dynamic scanner-only stock discovery
- SmartVolatilityScanner is now the sole source of trading candidates
- Add active_stocks dict for scanner-discovered stocks per market
- Add smart_scanner parameter to run_daily_session()

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-08 01:58:09 +09:00
a82a167915 Merge pull request 'feat: implement Smart Volatility Scanner (issue #76)' (#77) from feature/issue-76-smart-volatility-scanner into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #77
2026-02-06 07:43:54 +09:00
22 changed files with 3885 additions and 166 deletions

45
docs/agent-constraints.md Normal file
View File

@@ -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.

View File

@@ -64,3 +64,25 @@
**참고:** Realtime 모드 전용. Daily 모드는 배치 효율성을 위해 정적 watchlist 사용. **참고:** Realtime 모드 전용. Daily 모드는 배치 효율성을 위해 정적 watchlist 사용.
**이슈/PR:** #76, #77 **이슈/PR:** #76, #77
---
## 2026-02-10
### 코드 리뷰 시 플랜-구현 일치 검증 규칙
**배경:**
- 코드 리뷰 시 플랜(EnterPlanMode에서 승인된 계획)과 실제 구현이 일치하는지 확인하는 절차가 없었음
- 플랜과 다른 구현이 리뷰 없이 통과될 위험
**요구사항:**
1. 모든 PR 리뷰에서 플랜-구현 일치 여부를 필수 체크
2. 플랜에 없는 변경은 정당한 사유 필요
3. 플랜 항목이 누락되면 PR 설명에 사유 기록
4. 스코프가 플랜과 일치하는지 확인
**구현 결과:**
- `docs/workflow.md`에 Code Review Checklist 섹션 추가
- Plan Consistency (필수), Safety & Constraints, Quality, Workflow 4개 카테고리
**이슈/PR:** #114

View File

@@ -6,6 +6,7 @@
1. **Create Gitea Issue First** — All features, bug fixes, and policy changes require a Gitea issue before any code is written 1. **Create Gitea Issue First** — All features, bug fixes, and policy changes require a Gitea issue before any code is written
2. **Create Feature Branch** — Branch from `main` using format `feature/issue-{N}-{short-description}` 2. **Create Feature Branch** — Branch from `main` using format `feature/issue-{N}-{short-description}`
- After creating the branch, run `git pull origin main` and rebase to ensure the branch is up to date
3. **Implement Changes** — Write code, tests, and documentation on the feature branch 3. **Implement Changes** — Write code, tests, and documentation on the feature branch
4. **Create Pull Request** — Submit PR to `main` branch referencing the issue number 4. **Create Pull Request** — Submit PR to `main` branch referencing the issue number
5. **Review & Merge** — After approval, merge via PR (squash or merge commit) 5. **Review & Merge** — After approval, merge via PR (squash or merge commit)
@@ -73,3 +74,37 @@ task_tool(
``` ```
Use `run_in_background=True` for independent tasks that don't block subsequent work. Use `run_in_background=True` for independent tasks that don't block subsequent work.
## Code Review Checklist
**CRITICAL: Every PR review MUST verify plan-implementation consistency.**
Before approving any PR, the reviewer (human or agent) must check ALL of the following:
### 1. Plan Consistency (MANDATORY)
- [ ] **Implementation matches the approved plan** — Compare the actual code changes against the plan created during `EnterPlanMode`. Every item in the plan must be addressed.
- [ ] **No unplanned changes** — If the implementation includes changes not in the plan, they must be explicitly justified.
- [ ] **No plan items omitted** — If any planned item was skipped, the reason must be documented in the PR description.
- [ ] **Scope matches** — The PR does not exceed or fall short of the planned scope.
### 2. Safety & Constraints
- [ ] `src/core/risk_manager.py` is unchanged (READ-ONLY)
- [ ] Circuit breaker threshold not weakened (only stricter allowed)
- [ ] Fat-finger protection (30% max order) still enforced
- [ ] Confidence < 80 still forces HOLD
- [ ] No hardcoded API keys or secrets
### 3. Quality
- [ ] All new/modified code has corresponding tests
- [ ] Test coverage >= 80%
- [ ] `ruff check src/ tests/` passes (no lint errors)
- [ ] No `assert` statements removed from tests
### 4. Workflow
- [ ] PR references the Gitea issue number
- [ ] Feature branch follows naming convention (`feature/issue-N-description`)
- [ ] Commit messages are clear and descriptive

View File

@@ -108,7 +108,7 @@ class MarketScanner:
self.context_store.set_context( self.context_store.set_context(
ContextLayer.L7_REALTIME, ContextLayer.L7_REALTIME,
timeframe, timeframe,
f"{market.code}_{stock_code}_volatility", f"volatility_{market.code}_{stock_code}",
{ {
"price": metrics.current_price, "price": metrics.current_price,
"atr": metrics.atr, "atr": metrics.atr,
@@ -179,7 +179,7 @@ class MarketScanner:
self.context_store.set_context( self.context_store.set_context(
ContextLayer.L7_REALTIME, ContextLayer.L7_REALTIME,
timeframe, timeframe,
f"{market.code}_scan_result", f"scan_result_{market.code}",
{ {
"total_scanned": len(valid_metrics), "total_scanned": len(valid_metrics),
"top_movers": [m.stock_code for m in top_movers], "top_movers": [m.stock_code for m in top_movers],

View File

@@ -55,8 +55,15 @@ class Settings(BaseSettings):
DAILY_SESSIONS: int = Field(default=4, ge=1, le=10) DAILY_SESSIONS: int = Field(default=4, ge=1, le=10)
SESSION_INTERVAL_HOURS: int = Field(default=6, ge=1, le=24) SESSION_INTERVAL_HOURS: int = Field(default=6, ge=1, le=24)
# Pre-Market Planner
PRE_MARKET_MINUTES: int = Field(default=30, ge=10, le=120)
MAX_SCENARIOS_PER_STOCK: int = Field(default=5, ge=1, le=10)
PLANNER_TIMEOUT_SECONDS: int = Field(default=60, ge=10, le=300)
DEFENSIVE_PLAYBOOK_ON_FAILURE: bool = True
RESCAN_INTERVAL_SECONDS: int = Field(default=300, ge=60, le=900)
# Market selection (comma-separated market codes) # Market selection (comma-separated market codes)
ENABLED_MARKETS: str = "KR" ENABLED_MARKETS: str = "KR,US"
# Backup and Disaster Recovery (optional) # Backup and Disaster Recovery (optional)
BACKUP_ENABLED: bool = True BACKUP_ENABLED: bool = True

View File

@@ -230,21 +230,44 @@ class ContextAggregator:
) )
def run_all_aggregations(self) -> None: def run_all_aggregations(self) -> None:
"""Run all aggregations from L7 to L1 (bottom-up).""" """Run all aggregations from L7 to L1 (bottom-up).
All timeframes are derived from the latest trade timestamp so that
past data re-aggregation produces consistent results across layers.
"""
cursor = self.conn.execute("SELECT MAX(timestamp) FROM trades")
row = cursor.fetchone()
if not row or row[0] is None:
return
ts_raw = row[0]
if ts_raw.endswith("Z"):
ts_raw = ts_raw.replace("Z", "+00:00")
latest_ts = datetime.fromisoformat(ts_raw)
trade_date = latest_ts.date()
date_str = trade_date.isoformat()
iso_year, iso_week, _ = trade_date.isocalendar()
week_str = f"{iso_year}-W{iso_week:02d}"
month_str = f"{trade_date.year}-{trade_date.month:02d}"
quarter = (trade_date.month - 1) // 3 + 1
quarter_str = f"{trade_date.year}-Q{quarter}"
year_str = str(trade_date.year)
# L7 (trades) → L6 (daily) # L7 (trades) → L6 (daily)
self.aggregate_daily_from_trades() self.aggregate_daily_from_trades(date_str)
# L6 (daily) → L5 (weekly) # L6 (daily) → L5 (weekly)
self.aggregate_weekly_from_daily() self.aggregate_weekly_from_daily(week_str)
# L5 (weekly) → L4 (monthly) # L5 (weekly) → L4 (monthly)
self.aggregate_monthly_from_weekly() self.aggregate_monthly_from_weekly(month_str)
# L4 (monthly) → L3 (quarterly) # L4 (monthly) → L3 (quarterly)
self.aggregate_quarterly_from_monthly() self.aggregate_quarterly_from_monthly(quarter_str)
# L3 (quarterly) → L2 (annual) # L3 (quarterly) → L2 (annual)
self.aggregate_annual_from_quarterly() self.aggregate_annual_from_quarterly(year_str)
# L2 (annual) → L1 (legacy) # L2 (annual) → L1 (legacy)
self.aggregate_legacy_from_annual() self.aggregate_legacy_from_annual()

View File

@@ -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 # 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_layer ON contexts(layer)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_contexts_timeframe ON contexts(timeframe)") conn.execute("CREATE INDEX IF NOT EXISTS idx_contexts_timeframe ON contexts(timeframe)")

View File

@@ -10,14 +10,13 @@ import argparse
import asyncio import asyncio
import logging import logging
import signal import signal
import sys
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import Any from typing import Any
from src.analysis.scanner import MarketScanner
from src.analysis.smart_scanner import ScanCandidate, SmartVolatilityScanner from src.analysis.smart_scanner import ScanCandidate, SmartVolatilityScanner
from src.analysis.volatility import VolatilityAnalyzer from src.analysis.volatility import VolatilityAnalyzer
from src.brain.gemini_client import GeminiClient from src.brain.context_selector import ContextSelector
from src.brain.gemini_client import GeminiClient, TradeDecision
from src.broker.kis_api import KISBroker from src.broker.kis_api import KISBroker
from src.broker.overseas import OverseasBroker from src.broker.overseas import OverseasBroker
from src.config import Settings from src.config import Settings
@@ -31,6 +30,10 @@ from src.logging.decision_logger import DecisionLogger
from src.logging_config import setup_logging from src.logging_config import setup_logging
from src.markets.schedule import MarketInfo, get_next_market_open, get_open_markets from src.markets.schedule import MarketInfo, get_next_market_open, get_open_markets
from src.notifications.telegram_client import TelegramClient, TelegramCommandHandler from src.notifications.telegram_client import TelegramClient, TelegramCommandHandler
from src.strategy.models import DayPlaybook
from src.strategy.playbook_store import PlaybookStore
from src.strategy.pre_market_planner import PreMarketPlanner
from src.strategy.scenario_engine import ScenarioEngine
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -63,14 +66,6 @@ def safe_float(value: str | float | None, default: float = 0.0) -> float:
return default return default
# Target stock codes to monitor per market
WATCHLISTS = {
"KR": ["005930", "000660", "035420"], # Samsung, SK Hynix, NAVER
"US_NASDAQ": ["AAPL", "MSFT", "GOOGL"], # Example US stocks
"US_NYSE": ["JPM", "BAC"], # Example NYSE stocks
"JP": ["7203", "6758"], # Toyota, Sony
}
TRADE_INTERVAL_SECONDS = 60 TRADE_INTERVAL_SECONDS = 60
SCAN_INTERVAL_SECONDS = 60 # Scan markets every 60 seconds SCAN_INTERVAL_SECONDS = 60 # Scan markets every 60 seconds
MAX_CONNECTION_RETRIES = 3 MAX_CONNECTION_RETRIES = 3
@@ -79,20 +74,12 @@ MAX_CONNECTION_RETRIES = 3
DAILY_TRADE_SESSIONS = 4 # Number of trading sessions per day DAILY_TRADE_SESSIONS = 4 # Number of trading sessions per day
TRADE_SESSION_INTERVAL_HOURS = 6 # Hours between sessions TRADE_SESSION_INTERVAL_HOURS = 6 # Hours between sessions
# Full stock universe per market (for scanning)
# In production, this would be loaded from a database or API
STOCK_UNIVERSE = {
"KR": ["005930", "000660", "035420", "051910", "005380", "005490"],
"US_NASDAQ": ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "TSLA"],
"US_NYSE": ["JPM", "BAC", "XOM", "JNJ", "V"],
"JP": ["7203", "6758", "9984", "6861"],
}
async def trading_cycle( async def trading_cycle(
broker: KISBroker, broker: KISBroker,
overseas_broker: OverseasBroker, overseas_broker: OverseasBroker,
brain: GeminiClient, scenario_engine: ScenarioEngine,
playbook: DayPlaybook,
risk: RiskManager, risk: RiskManager,
db_conn: Any, db_conn: Any,
decision_logger: DecisionLogger, decision_logger: DecisionLogger,
@@ -101,7 +88,7 @@ async def trading_cycle(
telegram: TelegramClient, telegram: TelegramClient,
market: MarketInfo, market: MarketInfo,
stock_code: str, stock_code: str,
scan_candidates: dict[str, ScanCandidate], scan_candidates: dict[str, dict[str, ScanCandidate]],
) -> None: ) -> None:
"""Execute one trading cycle for a single stock.""" """Execute one trading cycle for a single stock."""
cycle_start_time = asyncio.get_event_loop().time() cycle_start_time = asyncio.get_event_loop().time()
@@ -152,13 +139,59 @@ async def trading_cycle(
else 0.0 else 0.0
) )
market_data = { market_data: dict[str, Any] = {
"stock_code": stock_code, "stock_code": stock_code,
"market_name": market.name, "market_name": market.name,
"current_price": current_price, "current_price": current_price,
"foreigner_net": foreigner_net, "foreigner_net": foreigner_net,
} }
# Enrich market_data with scanner metrics for scenario engine
market_candidates = scan_candidates.get(market.code, {})
candidate = market_candidates.get(stock_code)
if candidate:
market_data["rsi"] = candidate.rsi
market_data["volume_ratio"] = candidate.volume_ratio
# 1.3. Record L7 real-time context (market-scoped keys)
timeframe = datetime.now(UTC).isoformat()
context_store.set_context(
ContextLayer.L7_REALTIME,
timeframe,
f"volatility_{market.code}_{stock_code}",
{
"momentum_score": 50.0,
"volume_surge": 1.0,
"price_change_1m": 0.0,
},
)
context_store.set_context(
ContextLayer.L7_REALTIME,
timeframe,
f"price_{market.code}_{stock_code}",
{"current_price": current_price},
)
if candidate:
context_store.set_context(
ContextLayer.L7_REALTIME,
timeframe,
f"rsi_{market.code}_{stock_code}",
{"rsi": candidate.rsi},
)
context_store.set_context(
ContextLayer.L7_REALTIME,
timeframe,
f"volume_ratio_{market.code}_{stock_code}",
{"volume_ratio": candidate.volume_ratio},
)
# Build portfolio data for global rule evaluation
portfolio_data = {
"portfolio_pnl_pct": pnl_pct,
"total_cash": total_cash,
"total_eval": total_eval,
}
# 1.5. Get volatility metrics from context store (L7_REALTIME) # 1.5. Get volatility metrics from context store (L7_REALTIME)
latest_timeframe = context_store.get_latest_timeframe(ContextLayer.L7_REALTIME) latest_timeframe = context_store.get_latest_timeframe(ContextLayer.L7_REALTIME)
volatility_score = 50.0 # Default normal volatility volatility_score = 50.0 # Default normal volatility
@@ -169,7 +202,7 @@ async def trading_cycle(
volatility_data = context_store.get_context( volatility_data = context_store.get_context(
ContextLayer.L7_REALTIME, ContextLayer.L7_REALTIME,
latest_timeframe, latest_timeframe,
f"volatility_{stock_code}", f"volatility_{market.code}_{stock_code}",
) )
if volatility_data: if volatility_data:
volatility_score = volatility_data.get("momentum_score", 50.0) volatility_score = volatility_data.get("momentum_score", 50.0)
@@ -195,8 +228,13 @@ async def trading_cycle(
volume_surge, volume_surge,
) )
# 2. Ask the brain for a decision # 2. Evaluate scenario (local, no API call)
decision = await brain.decide(market_data) match = scenario_engine.evaluate(playbook, stock_code, market_data, portfolio_data)
decision = TradeDecision(
action=match.action.value,
confidence=match.confidence,
rationale=match.rationale,
)
logger.info( logger.info(
"Decision for %s (%s): %s (confidence=%d)", "Decision for %s (%s): %s (confidence=%d)",
stock_code, stock_code,
@@ -205,6 +243,19 @@ async def trading_cycle(
decision.confidence, decision.confidence,
) )
# 2.1. Notify scenario match
if match.matched_scenario is not None:
try:
condition_parts = [f"{k}={v}" for k, v in match.match_details.items()]
await telegram.notify_scenario_matched(
stock_code=stock_code,
action=decision.action,
condition_summary=", ".join(condition_parts) if condition_parts else "matched",
confidence=float(decision.confidence),
)
except Exception as exc:
logger.warning("Scenario matched notification failed: %s", exc)
# 2.5. Log decision with context snapshot # 2.5. Log decision with context snapshot
context_snapshot = { context_snapshot = {
"L1": { "L1": {
@@ -217,7 +268,7 @@ async def trading_cycle(
"purchase_total": purchase_total, "purchase_total": purchase_total,
"pnl_pct": pnl_pct, "pnl_pct": pnl_pct,
}, },
# L3-L7 will be populated when context tree is implemented "scenario_match": match.match_details,
} }
input_data = { input_data = {
"current_price": current_price, "current_price": current_price,
@@ -296,8 +347,8 @@ async def trading_cycle(
# 6. Log trade with selection context # 6. Log trade with selection context
selection_context = None selection_context = None
if stock_code in scan_candidates: if stock_code in market_candidates:
candidate = scan_candidates[stock_code] candidate = market_candidates[stock_code]
selection_context = { selection_context = {
"rsi": candidate.rsi, "rsi": candidate.rsi,
"volume_ratio": candidate.volume_ratio, "volume_ratio": candidate.volume_ratio,
@@ -341,7 +392,9 @@ async def trading_cycle(
async def run_daily_session( async def run_daily_session(
broker: KISBroker, broker: KISBroker,
overseas_broker: OverseasBroker, overseas_broker: OverseasBroker,
brain: GeminiClient, scenario_engine: ScenarioEngine,
playbook_store: PlaybookStore,
pre_market_planner: PreMarketPlanner,
risk: RiskManager, risk: RiskManager,
db_conn: Any, db_conn: Any,
decision_logger: DecisionLogger, decision_logger: DecisionLogger,
@@ -349,13 +402,12 @@ async def run_daily_session(
criticality_assessor: CriticalityAssessor, criticality_assessor: CriticalityAssessor,
telegram: TelegramClient, telegram: TelegramClient,
settings: Settings, settings: Settings,
smart_scanner: SmartVolatilityScanner | None = None,
) -> None: ) -> None:
"""Execute one daily trading session. """Execute one daily trading session.
Designed for API efficiency with Gemini Free tier: V2 proactive strategy: 1 Gemini call for playbook generation,
- Batch decision making (1 API call per market) then local scenario evaluation per stock (0 API calls).
- Runs N times per day at fixed intervals
- Minimizes API usage while maintaining trading capability
""" """
# Get currently open markets # Get currently open markets
open_markets = get_open_markets(settings.enabled_market_list) open_markets = get_open_markets(settings.enabled_market_list)
@@ -368,21 +420,66 @@ async def run_daily_session(
# Process each open market # Process each open market
for market in open_markets: for market in open_markets:
# Get watchlist for this market # Use market-local date for playbook keying
watchlist = WATCHLISTS.get(market.code, []) market_today = datetime.now(market.timezone).date()
if not watchlist:
logger.debug("No watchlist for market %s", market.code) # Dynamic stock discovery via scanner (no static watchlists)
candidates_list: list[ScanCandidate] = []
try:
candidates_list = await smart_scanner.scan() if smart_scanner else []
except Exception as exc:
logger.error("Smart Scanner failed for %s: %s", market.name, exc)
if not candidates_list:
logger.info("No scanner candidates for market %s — skipping", market.code)
continue continue
watchlist = [c.stock_code for c in candidates_list]
candidate_map = {c.stock_code: c for c in candidates_list}
logger.info("Processing market: %s (%d stocks)", market.name, len(watchlist)) logger.info("Processing market: %s (%d stocks)", market.name, len(watchlist))
# Collect market data for all stocks in the watchlist # Generate or load playbook (1 Gemini API call per market per day)
playbook = playbook_store.load(market_today, market.code)
if playbook is None:
try:
playbook = await pre_market_planner.generate_playbook(
market=market.code,
candidates=candidates_list,
today=market_today,
)
playbook_store.save(playbook)
try:
await telegram.notify_playbook_generated(
market=market.code,
stock_count=playbook.stock_count,
scenario_count=playbook.scenario_count,
token_count=playbook.token_count,
)
except Exception as exc:
logger.warning("Playbook notification failed: %s", exc)
logger.info(
"Generated playbook for %s: %d stocks, %d scenarios",
market.code, playbook.stock_count, playbook.scenario_count,
)
except Exception as exc:
logger.error("Playbook generation failed for %s: %s", market.code, exc)
try:
await telegram.notify_playbook_failed(
market=market.code, reason=str(exc)[:200],
)
except Exception as notify_exc:
logger.warning("Playbook failed notification error: %s", notify_exc)
playbook = PreMarketPlanner._empty_playbook(market_today, market.code)
# Collect market data for all stocks from scanner
stocks_data = [] stocks_data = []
for stock_code in watchlist: for stock_code in watchlist:
try: try:
if market.is_domestic: if market.is_domestic:
orderbook = await broker.get_orderbook(stock_code) orderbook = await broker.get_orderbook(stock_code)
current_price = safe_float(orderbook.get("output1", {}).get("stck_prpr", "0")) current_price = safe_float(
orderbook.get("output1", {}).get("stck_prpr", "0")
)
foreigner_net = safe_float( foreigner_net = safe_float(
orderbook.get("output1", {}).get("frgn_ntby_qty", "0") orderbook.get("output1", {}).get("frgn_ntby_qty", "0")
) )
@@ -390,17 +487,23 @@ async def run_daily_session(
price_data = await overseas_broker.get_overseas_price( price_data = await overseas_broker.get_overseas_price(
market.exchange_code, stock_code market.exchange_code, stock_code
) )
current_price = safe_float(price_data.get("output", {}).get("last", "0")) current_price = safe_float(
price_data.get("output", {}).get("last", "0")
)
foreigner_net = 0.0 foreigner_net = 0.0
stocks_data.append( stock_data: dict[str, Any] = {
{ "stock_code": stock_code,
"stock_code": stock_code, "market_name": market.name,
"market_name": market.name, "current_price": current_price,
"current_price": current_price, "foreigner_net": foreigner_net,
"foreigner_net": foreigner_net, }
} # Enrich with scanner metrics
) cand = candidate_map.get(stock_code)
if cand:
stock_data["rsi"] = cand.rsi
stock_data["volume_ratio"] = cand.volume_ratio
stocks_data.append(stock_data)
except Exception as exc: except Exception as exc:
logger.error("Failed to fetch data for %s: %s", stock_code, exc) logger.error("Failed to fetch data for %s: %s", stock_code, exc)
continue continue
@@ -409,17 +512,19 @@ async def run_daily_session(
logger.warning("No valid stock data for market %s", market.code) logger.warning("No valid stock data for market %s", market.code)
continue continue
# Get batch decisions (1 API call for all stocks in this market)
logger.info("Requesting batch decision for %d stocks in %s", len(stocks_data), market.name)
decisions = await brain.decide_batch(stocks_data)
# Get balance data once for the market # Get balance data once for the market
if market.is_domestic: if market.is_domestic:
balance_data = await broker.get_balance() balance_data = await broker.get_balance()
output2 = balance_data.get("output2", [{}]) output2 = balance_data.get("output2", [{}])
total_eval = safe_float(output2[0].get("tot_evlu_amt", "0")) if output2 else 0 total_eval = safe_float(
total_cash = safe_float(output2[0].get("dnca_tot_amt", "0")) if output2 else 0 output2[0].get("tot_evlu_amt", "0")
purchase_total = safe_float(output2[0].get("pchs_amt_smtl_amt", "0")) if output2 else 0 ) if output2 else 0
total_cash = safe_float(
output2[0].get("dnca_tot_amt", "0")
) if output2 else 0
purchase_total = safe_float(
output2[0].get("pchs_amt_smtl_amt", "0")
) if output2 else 0
else: else:
balance_data = await overseas_broker.get_overseas_balance(market.exchange_code) balance_data = await overseas_broker.get_overseas_balance(market.exchange_code)
output2 = balance_data.get("output2", [{}]) output2 = balance_data.get("output2", [{}])
@@ -432,21 +537,37 @@ async def run_daily_session(
total_eval = safe_float(balance_info.get("frcr_evlu_tota", "0") or "0") total_eval = safe_float(balance_info.get("frcr_evlu_tota", "0") or "0")
total_cash = safe_float(balance_info.get("frcr_dncl_amt_2", "0") or "0") total_cash = safe_float(balance_info.get("frcr_dncl_amt_2", "0") or "0")
purchase_total = safe_float(balance_info.get("frcr_buy_amt_smtl", "0") or "0") purchase_total = safe_float(
balance_info.get("frcr_buy_amt_smtl", "0") or "0"
)
# Calculate daily P&L % # Calculate daily P&L %
pnl_pct = ( pnl_pct = (
((total_eval - purchase_total) / purchase_total * 100) if purchase_total > 0 else 0.0 ((total_eval - purchase_total) / purchase_total * 100)
if purchase_total > 0
else 0.0
) )
portfolio_data = {
"portfolio_pnl_pct": pnl_pct,
"total_cash": total_cash,
"total_eval": total_eval,
}
# Execute decisions for each stock # Evaluate scenarios for each stock (local, no API calls)
logger.info(
"Evaluating %d stocks against playbook for %s",
len(stocks_data), market.name,
)
for stock_data in stocks_data: for stock_data in stocks_data:
stock_code = stock_data["stock_code"] stock_code = stock_data["stock_code"]
decision = decisions.get(stock_code) match = scenario_engine.evaluate(
playbook, stock_code, stock_data, portfolio_data,
if not decision: )
logger.warning("No decision for %s — skipping", stock_code) decision = TradeDecision(
continue action=match.action.value,
confidence=match.confidence,
rationale=match.rationale,
)
logger.info( logger.info(
"Decision for %s (%s): %s (confidence=%d)", "Decision for %s (%s): %s (confidence=%d)",
@@ -468,6 +589,7 @@ async def run_daily_session(
"purchase_total": purchase_total, "purchase_total": purchase_total,
"pnl_pct": pnl_pct, "pnl_pct": pnl_pct,
}, },
"scenario_match": match.match_details,
} }
input_data = { input_data = {
"current_price": stock_data["current_price"], "current_price": stock_data["current_price"],
@@ -519,7 +641,9 @@ async def run_daily_session(
threshold=exc.threshold, threshold=exc.threshold,
) )
except Exception as notify_exc: except Exception as notify_exc:
logger.warning("Circuit breaker notification failed: %s", notify_exc) logger.warning(
"Circuit breaker notification failed: %s", notify_exc
)
raise raise
# Send order # Send order
@@ -554,7 +678,9 @@ async def run_daily_session(
except Exception as exc: except Exception as exc:
logger.warning("Telegram notification failed: %s", exc) logger.warning("Telegram notification failed: %s", exc)
except Exception as exc: except Exception as exc:
logger.error("Order execution failed for %s: %s", stock_code, exc) logger.error(
"Order execution failed for %s: %s", stock_code, exc
)
continue continue
# Log trade # Log trade
@@ -581,6 +707,20 @@ async def run(settings: Settings) -> None:
decision_logger = DecisionLogger(db_conn) decision_logger = DecisionLogger(db_conn)
context_store = ContextStore(db_conn) context_store = ContextStore(db_conn)
# V2 proactive strategy components
context_selector = ContextSelector(context_store)
scenario_engine = ScenarioEngine()
playbook_store = PlaybookStore(db_conn)
pre_market_planner = PreMarketPlanner(
gemini_client=brain,
context_store=context_store,
context_selector=context_selector,
settings=settings,
)
# Track playbooks per market (in-memory cache)
playbooks: dict[str, DayPlaybook] = {}
# Initialize Telegram notifications # Initialize Telegram notifications
telegram = TelegramClient( telegram = TelegramClient(
bot_token=settings.TELEGRAM_BOT_TOKEN, bot_token=settings.TELEGRAM_BOT_TOKEN,
@@ -726,15 +866,6 @@ async def run(settings: Settings) -> None:
# Initialize volatility hunter # Initialize volatility hunter
volatility_analyzer = VolatilityAnalyzer(min_volume_surge=2.0, min_price_change=1.0) volatility_analyzer = VolatilityAnalyzer(min_volume_surge=2.0, min_price_change=1.0)
market_scanner = MarketScanner(
broker=broker,
overseas_broker=overseas_broker,
volatility_analyzer=volatility_analyzer,
context_store=context_store,
top_n=5,
max_concurrent_scans=1, # Fully serialized to avoid EGW00201
)
# Initialize smart scanner (Python-first, AI-last pipeline) # Initialize smart scanner (Python-first, AI-last pipeline)
smart_scanner = SmartVolatilityScanner( smart_scanner = SmartVolatilityScanner(
broker=broker, broker=broker,
@@ -742,8 +873,11 @@ async def run(settings: Settings) -> None:
settings=settings, settings=settings,
) )
# Track scan candidates for selection context logging # Track scan candidates per market for selection context logging
scan_candidates: dict[str, ScanCandidate] = {} # stock_code -> candidate scan_candidates: dict[str, dict[str, ScanCandidate]] = {} # market -> {stock_code -> candidate}
# Active stocks per market (dynamically discovered by scanner)
active_stocks: dict[str, list[str]] = {} # market_code -> [stock_codes]
# Initialize latency control system # Initialize latency control system
criticality_assessor = CriticalityAssessor( criticality_assessor = CriticalityAssessor(
@@ -809,7 +943,9 @@ async def run(settings: Settings) -> None:
await run_daily_session( await run_daily_session(
broker, broker,
overseas_broker, overseas_broker,
brain, scenario_engine,
playbook_store,
pre_market_planner,
risk, risk,
db_conn, db_conn,
decision_logger, decision_logger,
@@ -817,6 +953,7 @@ async def run(settings: Settings) -> None:
criticality_assessor, criticality_assessor,
telegram, telegram,
settings, settings,
smart_scanner=smart_scanner,
) )
except CircuitBreakerTripped: except CircuitBreakerTripped:
logger.critical("Circuit breaker tripped — shutting down") logger.critical("Circuit breaker tripped — shutting down")
@@ -856,6 +993,8 @@ async def run(settings: Settings) -> None:
except Exception as exc: except Exception as exc:
logger.warning("Market close notification failed: %s", exc) logger.warning("Market close notification failed: %s", exc)
_market_states[market_code] = False _market_states[market_code] = False
# Clear playbook for closed market (new one generated next open)
playbooks.pop(market_code, None)
# No markets open — wait until next market opens # No markets open — wait until next market opens
try: try:
@@ -890,67 +1029,129 @@ async def run(settings: Settings) -> None:
logger.warning("Market open notification failed: %s", exc) logger.warning("Market open notification failed: %s", exc)
_market_states[market.code] = True _market_states[market.code] = True
# Smart Scanner: Python-first filtering (RSI + volume) before AI # Smart Scanner: dynamic stock discovery (no static watchlists)
now_timestamp = asyncio.get_event_loop().time() now_timestamp = asyncio.get_event_loop().time()
last_scan = last_scan_time.get(market.code, 0.0) last_scan = last_scan_time.get(market.code, 0.0)
if now_timestamp - last_scan >= SCAN_INTERVAL_SECONDS: rescan_interval = settings.RESCAN_INTERVAL_SECONDS
if now_timestamp - last_scan >= rescan_interval:
try: try:
logger.info("Smart Scanner: Scanning %s market", market.name) logger.info("Smart Scanner: Scanning %s market", market.name)
# Run smart scan with fallback to static universe candidates = await smart_scanner.scan()
fallback_universe = STOCK_UNIVERSE.get(market.code, [])
candidates = await smart_scanner.scan(fallback_stocks=fallback_universe)
if candidates: if candidates:
# Update watchlist with qualified candidates # Use scanner results directly as trading candidates
qualified_codes = smart_scanner.get_stock_codes(candidates) active_stocks[market.code] = smart_scanner.get_stock_codes(
candidates
)
# Merge with existing watchlist (keep some continuity) # Store candidates per market for selection context logging
current_watchlist = WATCHLISTS.get(market.code, []) scan_candidates[market.code] = {
# Keep up to 2 from existing, add new qualified c.stock_code: c for c in candidates
merged = qualified_codes + [ }
c for c in current_watchlist if c not in qualified_codes
][:2]
WATCHLISTS[market.code] = merged[:5] # Cap at 5
# Store candidates for later selection context logging
for candidate in candidates:
scan_candidates[candidate.stock_code] = candidate
logger.info( logger.info(
"Smart Scanner: Found %d qualified candidates for %s: %s", "Smart Scanner: Found %d candidates for %s: %s",
len(candidates), len(candidates),
market.name, market.name,
[f"{c.stock_code}(RSI={c.rsi:.0f})" for c in candidates], [f"{c.stock_code}(RSI={c.rsi:.0f})" for c in candidates],
) )
# Get market-local date for playbook keying
market_today = datetime.now(
market.timezone
).date()
# Load or generate playbook (1 Gemini call per market per day)
if market.code not in playbooks:
# Try DB first (survives process restart)
stored_pb = playbook_store.load(market_today, market.code)
if stored_pb is not None:
playbooks[market.code] = stored_pb
logger.info(
"Loaded existing playbook for %s from DB"
" (%d stocks, %d scenarios)",
market.code,
stored_pb.stock_count,
stored_pb.scenario_count,
)
else:
try:
pb = await pre_market_planner.generate_playbook(
market=market.code,
candidates=candidates,
today=market_today,
)
playbook_store.save(pb)
playbooks[market.code] = pb
try:
await telegram.notify_playbook_generated(
market=market.code,
stock_count=pb.stock_count,
scenario_count=pb.scenario_count,
token_count=pb.token_count,
)
except Exception as exc:
logger.warning(
"Playbook notification failed: %s", exc
)
except Exception as exc:
logger.error(
"Playbook generation failed for %s: %s",
market.code, exc,
)
try:
await telegram.notify_playbook_failed(
market=market.code,
reason=str(exc)[:200],
)
except Exception:
pass
playbooks[market.code] = (
PreMarketPlanner._empty_playbook(
market_today, market.code
)
)
else: else:
logger.info("Smart Scanner: No qualified candidates for %s", market.name) logger.info(
"Smart Scanner: No candidates for %s — no trades", market.name
)
active_stocks[market.code] = []
last_scan_time[market.code] = now_timestamp last_scan_time[market.code] = now_timestamp
except Exception as exc: except Exception as exc:
logger.error("Smart Scanner failed for %s: %s", market.name, exc) logger.error("Smart Scanner failed for %s: %s", market.name, exc)
# Get watchlist for this market # Get active stocks from scanner (dynamic, no static fallback)
watchlist = WATCHLISTS.get(market.code, []) stock_codes = active_stocks.get(market.code, [])
if not watchlist: if not stock_codes:
logger.debug("No watchlist for market %s", market.code) logger.debug("No active stocks for market %s", market.code)
continue continue
logger.info("Processing market: %s (%d stocks)", market.name, len(watchlist)) logger.info("Processing market: %s (%d stocks)", market.name, len(stock_codes))
# Process each stock in the watchlist # Process each stock from scanner results
for stock_code in watchlist: for stock_code in stock_codes:
if shutdown.is_set(): if shutdown.is_set():
break break
# Get playbook for this market
market_playbook = playbooks.get(
market.code,
PreMarketPlanner._empty_playbook(
datetime.now(market.timezone).date(), market.code
),
)
# Retry logic for connection errors # Retry logic for connection errors
for attempt in range(1, MAX_CONNECTION_RETRIES + 1): for attempt in range(1, MAX_CONNECTION_RETRIES + 1):
try: try:
await trading_cycle( await trading_cycle(
broker, broker,
overseas_broker, overseas_broker,
brain, scenario_engine,
market_playbook,
risk, risk,
db_conn, db_conn,
decision_logger, decision_logger,
@@ -999,7 +1200,8 @@ async def run(settings: Settings) -> None:
metrics = await priority_queue.get_metrics() metrics = await priority_queue.get_metrics()
if metrics.total_enqueued > 0: if metrics.total_enqueued > 0:
logger.info( logger.info(
"Priority queue metrics: enqueued=%d, dequeued=%d, size=%d, timeouts=%d, errors=%d", "Priority queue metrics: enqueued=%d, dequeued=%d,"
" size=%d, timeouts=%d, errors=%d",
metrics.total_enqueued, metrics.total_enqueued,
metrics.total_dequeued, metrics.total_dequeued,
metrics.current_size, metrics.current_size,

View File

@@ -304,6 +304,77 @@ class TelegramClient:
NotificationMessage(priority=NotificationPriority.MEDIUM, message=message) NotificationMessage(priority=NotificationPriority.MEDIUM, message=message)
) )
async def notify_playbook_generated(
self,
market: str,
stock_count: int,
scenario_count: int,
token_count: int,
) -> None:
"""
Notify that a daily playbook was generated.
Args:
market: Market code (e.g., "KR", "US")
stock_count: Number of stocks in the playbook
scenario_count: Total number of scenarios
token_count: Gemini token usage for the playbook
"""
message = (
f"<b>Playbook Generated</b>\n"
f"Market: {market}\n"
f"Stocks: {stock_count}\n"
f"Scenarios: {scenario_count}\n"
f"Tokens: {token_count}"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.MEDIUM, message=message)
)
async def notify_scenario_matched(
self,
stock_code: str,
action: str,
condition_summary: str,
confidence: float,
) -> None:
"""
Notify that a scenario matched for a stock.
Args:
stock_code: Stock ticker symbol
action: Scenario action (BUY/SELL/HOLD/REDUCE_ALL)
condition_summary: Short summary of the matched condition
confidence: Scenario confidence (0-100)
"""
message = (
f"<b>Scenario Matched</b>\n"
f"Symbol: <code>{stock_code}</code>\n"
f"Action: {action}\n"
f"Condition: {condition_summary}\n"
f"Confidence: {confidence:.0f}%"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.HIGH, message=message)
)
async def notify_playbook_failed(self, market: str, reason: str) -> None:
"""
Notify that playbook generation failed.
Args:
market: Market code (e.g., "KR", "US")
reason: Failure reason summary
"""
message = (
f"<b>Playbook Failed</b>\n"
f"Market: {market}\n"
f"Reason: {reason[:200]}"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.HIGH, message=message)
)
async def notify_system_shutdown(self, reason: str) -> None: async def notify_system_shutdown(self, reason: str) -> None:
""" """
Notify system shutdown. Notify system shutdown.

0
src/strategy/__init__.py Normal file
View File

164
src/strategy/models.py Normal file
View File

@@ -0,0 +1,164 @@
"""Pydantic models for pre-market scenario planning.
Defines the data contracts for the proactive strategy system:
- AI generates DayPlaybook before market open (structured JSON scenarios)
- Local ScenarioEngine matches conditions during market hours (no API calls)
"""
from __future__ import annotations
from datetime import UTC, date, datetime
from enum import Enum
from pydantic import BaseModel, Field, field_validator
class ScenarioAction(str, Enum):
"""Actions that can be taken by scenarios."""
BUY = "BUY"
SELL = "SELL"
HOLD = "HOLD"
REDUCE_ALL = "REDUCE_ALL"
class MarketOutlook(str, Enum):
"""AI's assessment of market direction."""
BULLISH = "bullish"
NEUTRAL_TO_BULLISH = "neutral_to_bullish"
NEUTRAL = "neutral"
NEUTRAL_TO_BEARISH = "neutral_to_bearish"
BEARISH = "bearish"
class PlaybookStatus(str, Enum):
"""Lifecycle status of a playbook."""
PENDING = "pending"
READY = "ready"
FAILED = "failed"
EXPIRED = "expired"
class StockCondition(BaseModel):
"""Condition fields for scenario matching (all optional, AND-combined).
The ScenarioEngine evaluates all non-None fields as AND conditions.
A condition matches only if ALL specified fields are satisfied.
"""
rsi_below: float | None = None
rsi_above: float | None = None
volume_ratio_above: float | None = None
volume_ratio_below: float | None = None
price_above: float | None = None
price_below: float | None = None
price_change_pct_above: float | None = None
price_change_pct_below: float | None = None
def has_any_condition(self) -> bool:
"""Check if at least one condition field is set."""
return any(
v is not None
for v in (
self.rsi_below,
self.rsi_above,
self.volume_ratio_above,
self.volume_ratio_below,
self.price_above,
self.price_below,
self.price_change_pct_above,
self.price_change_pct_below,
)
)
class StockScenario(BaseModel):
"""A single condition-action rule for one stock."""
condition: StockCondition
action: ScenarioAction
confidence: int = Field(ge=0, le=100)
allocation_pct: float = Field(ge=0, le=100, default=10.0)
stop_loss_pct: float = Field(le=0, default=-2.0)
take_profit_pct: float = Field(ge=0, default=3.0)
rationale: str = ""
class StockPlaybook(BaseModel):
"""All scenarios for a single stock (ordered by priority)."""
stock_code: str
stock_name: str = ""
scenarios: list[StockScenario] = Field(min_length=1)
class GlobalRule(BaseModel):
"""Portfolio-level rule (checked before stock-level scenarios)."""
condition: str # e.g. "portfolio_pnl_pct < -2.0"
action: ScenarioAction
rationale: str = ""
class CrossMarketContext(BaseModel):
"""Summary of another market's state for cross-market awareness."""
market: str # e.g. "US" or "KR"
date: str
total_pnl: float = 0.0
win_rate: float = 0.0
index_change_pct: float = 0.0 # e.g. KOSPI or S&P500 change
key_events: list[str] = Field(default_factory=list)
lessons: list[str] = Field(default_factory=list)
class DayPlaybook(BaseModel):
"""Complete playbook for a single trading day in a single market.
Generated by PreMarketPlanner (1 Gemini call per market per day).
Consumed by ScenarioEngine during market hours (0 API calls).
"""
date: date
market: str # "KR" or "US"
market_outlook: MarketOutlook = MarketOutlook.NEUTRAL
generated_at: str = "" # ISO timestamp
gemini_model: str = ""
token_count: int = 0
global_rules: list[GlobalRule] = Field(default_factory=list)
stock_playbooks: list[StockPlaybook] = Field(default_factory=list)
default_action: ScenarioAction = ScenarioAction.HOLD
context_summary: dict = Field(default_factory=dict)
cross_market: CrossMarketContext | None = None
@field_validator("stock_playbooks")
@classmethod
def validate_unique_stocks(cls, v: list[StockPlaybook]) -> list[StockPlaybook]:
codes = [pb.stock_code for pb in v]
if len(codes) != len(set(codes)):
raise ValueError("Duplicate stock codes in playbook")
return v
def get_stock_playbook(self, stock_code: str) -> StockPlaybook | None:
"""Find the playbook for a specific stock."""
for pb in self.stock_playbooks:
if pb.stock_code == stock_code:
return pb
return None
@property
def scenario_count(self) -> int:
"""Total number of scenarios across all stocks."""
return sum(len(pb.scenarios) for pb in self.stock_playbooks)
@property
def stock_count(self) -> int:
"""Number of stocks with scenarios."""
return len(self.stock_playbooks)
def model_post_init(self, __context: object) -> None:
"""Set generated_at if not provided."""
if not self.generated_at:
self.generated_at = datetime.now(UTC).isoformat()

View File

@@ -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

View File

@@ -0,0 +1,419 @@
"""Pre-market planner — generates DayPlaybook via Gemini before market open.
One Gemini API call per market per day. Candidates come from SmartVolatilityScanner.
On failure, returns a defensive playbook (all HOLD, no trades).
"""
from __future__ import annotations
import json
import logging
from datetime import date
from typing import Any
from src.analysis.smart_scanner import ScanCandidate
from src.brain.context_selector import ContextSelector, DecisionType
from src.brain.gemini_client import GeminiClient
from src.config import Settings
from src.context.store import ContextLayer, ContextStore
from src.strategy.models import (
CrossMarketContext,
DayPlaybook,
GlobalRule,
MarketOutlook,
ScenarioAction,
StockCondition,
StockPlaybook,
StockScenario,
)
logger = logging.getLogger(__name__)
# Mapping from string to MarketOutlook enum
_OUTLOOK_MAP: dict[str, MarketOutlook] = {
"bullish": MarketOutlook.BULLISH,
"neutral_to_bullish": MarketOutlook.NEUTRAL_TO_BULLISH,
"neutral": MarketOutlook.NEUTRAL,
"neutral_to_bearish": MarketOutlook.NEUTRAL_TO_BEARISH,
"bearish": MarketOutlook.BEARISH,
}
_ACTION_MAP: dict[str, ScenarioAction] = {
"BUY": ScenarioAction.BUY,
"SELL": ScenarioAction.SELL,
"HOLD": ScenarioAction.HOLD,
"REDUCE_ALL": ScenarioAction.REDUCE_ALL,
}
class PreMarketPlanner:
"""Generates a DayPlaybook by calling Gemini once before market open.
Flow:
1. Collect strategic context (L5-L7) + cross-market context
2. Build a structured prompt with scan candidates
3. Call Gemini for JSON scenario generation
4. Parse and validate response into DayPlaybook
5. On failure → defensive playbook (HOLD everything)
"""
def __init__(
self,
gemini_client: GeminiClient,
context_store: ContextStore,
context_selector: ContextSelector,
settings: Settings,
) -> None:
self._gemini = gemini_client
self._context_store = context_store
self._context_selector = context_selector
self._settings = settings
async def generate_playbook(
self,
market: str,
candidates: list[ScanCandidate],
today: date | None = None,
) -> DayPlaybook:
"""Generate a DayPlaybook for a market using Gemini.
Args:
market: Market code ("KR" or "US")
candidates: Stock candidates from SmartVolatilityScanner
today: Override date (defaults to date.today()). Use market-local date.
Returns:
DayPlaybook with scenarios. Empty/defensive if no candidates or failure.
"""
if today is None:
today = date.today()
if not candidates:
logger.info("No candidates for %s — returning empty playbook", market)
return self._empty_playbook(today, market)
try:
# 1. Gather context
context_data = self._gather_context()
cross_market = self.build_cross_market_context(market, today)
# 2. Build prompt
prompt = self._build_prompt(market, candidates, context_data, cross_market)
# 3. Call Gemini
market_data = {
"stock_code": "PLANNER",
"current_price": 0,
"prompt_override": prompt,
}
decision = await self._gemini.decide(market_data)
# 4. Parse response
playbook = self._parse_response(
decision.rationale, today, market, candidates, cross_market
)
playbook_with_tokens = playbook.model_copy(
update={"token_count": decision.token_count}
)
logger.info(
"Generated playbook for %s: %d stocks, %d scenarios, %d tokens",
market,
playbook_with_tokens.stock_count,
playbook_with_tokens.scenario_count,
playbook_with_tokens.token_count,
)
return playbook_with_tokens
except Exception:
logger.exception("Playbook generation failed for %s", market)
if self._settings.DEFENSIVE_PLAYBOOK_ON_FAILURE:
return self._defensive_playbook(today, market, candidates)
return self._empty_playbook(today, market)
def build_cross_market_context(
self, target_market: str, today: date | None = None,
) -> CrossMarketContext | None:
"""Build cross-market context from the other market's L6 data.
KR planner → reads US scorecard from previous night.
US planner → reads KR scorecard from today.
Args:
target_market: The market being planned ("KR" or "US")
today: Override date (defaults to date.today()). Use market-local date.
"""
other_market = "US" if target_market == "KR" else "KR"
if today is None:
today = date.today()
timeframe = today.isoformat()
scorecard_key = f"scorecard_{other_market}"
scorecard_data = self._context_store.get_context(
ContextLayer.L6_DAILY, timeframe, scorecard_key
)
if scorecard_data is None:
logger.debug("No cross-market scorecard found for %s", other_market)
return None
if isinstance(scorecard_data, str):
try:
scorecard_data = json.loads(scorecard_data)
except (json.JSONDecodeError, TypeError):
return None
if not isinstance(scorecard_data, dict):
return None
return CrossMarketContext(
market=other_market,
date=timeframe,
total_pnl=float(scorecard_data.get("total_pnl", 0.0)),
win_rate=float(scorecard_data.get("win_rate", 0.0)),
index_change_pct=float(scorecard_data.get("index_change_pct", 0.0)),
key_events=scorecard_data.get("key_events", []),
lessons=scorecard_data.get("lessons", []),
)
def _gather_context(self) -> dict[str, Any]:
"""Gather strategic context using ContextSelector."""
layers = self._context_selector.select_layers(
decision_type=DecisionType.STRATEGIC,
include_realtime=True,
)
return self._context_selector.get_context_data(layers, max_items_per_layer=10)
def _build_prompt(
self,
market: str,
candidates: list[ScanCandidate],
context_data: dict[str, Any],
cross_market: CrossMarketContext | None,
) -> str:
"""Build a structured prompt for Gemini to generate scenario JSON."""
max_scenarios = self._settings.MAX_SCENARIOS_PER_STOCK
candidates_text = "\n".join(
f" - {c.stock_code} ({c.name}): price={c.price}, "
f"RSI={c.rsi:.1f}, volume_ratio={c.volume_ratio:.1f}, "
f"signal={c.signal}, score={c.score:.1f}"
for c in candidates
)
cross_market_text = ""
if cross_market:
cross_market_text = (
f"\n## Other Market ({cross_market.market}) Summary\n"
f"- P&L: {cross_market.total_pnl:+.2f}%\n"
f"- Win Rate: {cross_market.win_rate:.0f}%\n"
f"- Index Change: {cross_market.index_change_pct:+.2f}%\n"
)
if cross_market.lessons:
cross_market_text += f"- Lessons: {'; '.join(cross_market.lessons[:3])}\n"
context_text = ""
if context_data:
context_text = "\n## Strategic Context\n"
for layer_name, layer_data in context_data.items():
if layer_data:
context_text += f"### {layer_name}\n"
for key, value in list(layer_data.items())[:5]:
context_text += f" - {key}: {value}\n"
return (
f"You are a pre-market trading strategist for the {market} market.\n"
f"Generate structured trading scenarios for today.\n\n"
f"## Candidates (from volatility scanner)\n{candidates_text}\n"
f"{cross_market_text}"
f"{context_text}\n"
f"## Instructions\n"
f"Return a JSON object with this exact structure:\n"
f'{{\n'
f' "market_outlook": "bullish|neutral_to_bullish|neutral'
f'|neutral_to_bearish|bearish",\n'
f' "global_rules": [\n'
f' {{"condition": "portfolio_pnl_pct < -2.0",'
f' "action": "REDUCE_ALL", "rationale": "..."}}\n'
f' ],\n'
f' "stocks": [\n'
f' {{\n'
f' "stock_code": "...",\n'
f' "scenarios": [\n'
f' {{\n'
f' "condition": {{"rsi_below": 30, "volume_ratio_above": 2.0}},\n'
f' "action": "BUY|SELL|HOLD",\n'
f' "confidence": 85,\n'
f' "allocation_pct": 10.0,\n'
f' "stop_loss_pct": -2.0,\n'
f' "take_profit_pct": 3.0,\n'
f' "rationale": "..."\n'
f' }}\n'
f' ]\n'
f' }}\n'
f' ]\n'
f'}}\n\n'
f"Rules:\n"
f"- Max {max_scenarios} scenarios per stock\n"
f"- Only use stocks from the candidates list\n"
f"- Confidence 0-100 (80+ for actionable trades)\n"
f"- stop_loss_pct must be <= 0, take_profit_pct must be >= 0\n"
f"- Return ONLY the JSON, no markdown fences or explanation\n"
)
def _parse_response(
self,
response_text: str,
today: date,
market: str,
candidates: list[ScanCandidate],
cross_market: CrossMarketContext | None,
) -> DayPlaybook:
"""Parse Gemini's JSON response into a validated DayPlaybook."""
cleaned = self._extract_json(response_text)
data = json.loads(cleaned)
valid_codes = {c.stock_code for c in candidates}
# Parse market outlook
outlook_str = data.get("market_outlook", "neutral")
market_outlook = _OUTLOOK_MAP.get(outlook_str, MarketOutlook.NEUTRAL)
# Parse global rules
global_rules = []
for rule_data in data.get("global_rules", []):
action_str = rule_data.get("action", "HOLD")
action = _ACTION_MAP.get(action_str, ScenarioAction.HOLD)
global_rules.append(
GlobalRule(
condition=rule_data.get("condition", ""),
action=action,
rationale=rule_data.get("rationale", ""),
)
)
# Parse stock playbooks
stock_playbooks = []
max_scenarios = self._settings.MAX_SCENARIOS_PER_STOCK
for stock_data in data.get("stocks", []):
code = stock_data.get("stock_code", "")
if code not in valid_codes:
logger.warning("Gemini returned unknown stock %s — skipping", code)
continue
scenarios = []
for sc_data in stock_data.get("scenarios", [])[:max_scenarios]:
scenario = self._parse_scenario(sc_data)
if scenario:
scenarios.append(scenario)
if scenarios:
stock_playbooks.append(
StockPlaybook(
stock_code=code,
scenarios=scenarios,
)
)
return DayPlaybook(
date=today,
market=market,
market_outlook=market_outlook,
global_rules=global_rules,
stock_playbooks=stock_playbooks,
cross_market=cross_market,
)
def _parse_scenario(self, sc_data: dict) -> StockScenario | None:
"""Parse a single scenario from JSON data. Returns None if invalid."""
try:
cond_data = sc_data.get("condition", {})
condition = StockCondition(
rsi_below=cond_data.get("rsi_below"),
rsi_above=cond_data.get("rsi_above"),
volume_ratio_above=cond_data.get("volume_ratio_above"),
volume_ratio_below=cond_data.get("volume_ratio_below"),
price_above=cond_data.get("price_above"),
price_below=cond_data.get("price_below"),
price_change_pct_above=cond_data.get("price_change_pct_above"),
price_change_pct_below=cond_data.get("price_change_pct_below"),
)
if not condition.has_any_condition():
logger.warning("Scenario has no conditions — skipping")
return None
action_str = sc_data.get("action", "HOLD")
action = _ACTION_MAP.get(action_str, ScenarioAction.HOLD)
return StockScenario(
condition=condition,
action=action,
confidence=int(sc_data.get("confidence", 50)),
allocation_pct=float(sc_data.get("allocation_pct", 10.0)),
stop_loss_pct=float(sc_data.get("stop_loss_pct", -2.0)),
take_profit_pct=float(sc_data.get("take_profit_pct", 3.0)),
rationale=sc_data.get("rationale", ""),
)
except (ValueError, TypeError) as e:
logger.warning("Failed to parse scenario: %s", e)
return None
@staticmethod
def _extract_json(text: str) -> str:
"""Extract JSON from response, stripping markdown fences if present."""
stripped = text.strip()
if stripped.startswith("```"):
# Remove first line (```json or ```) and last line (```)
lines = stripped.split("\n")
lines = lines[1:] # Remove opening fence
if lines and lines[-1].strip() == "```":
lines = lines[:-1]
stripped = "\n".join(lines)
return stripped.strip()
@staticmethod
def _empty_playbook(today: date, market: str) -> DayPlaybook:
"""Return an empty playbook (no stocks, no scenarios)."""
return DayPlaybook(
date=today,
market=market,
market_outlook=MarketOutlook.NEUTRAL,
stock_playbooks=[],
)
@staticmethod
def _defensive_playbook(
today: date,
market: str,
candidates: list[ScanCandidate],
) -> DayPlaybook:
"""Return a defensive playbook — HOLD everything with stop-loss ready."""
stock_playbooks = [
StockPlaybook(
stock_code=c.stock_code,
scenarios=[
StockScenario(
condition=StockCondition(price_change_pct_below=-3.0),
action=ScenarioAction.SELL,
confidence=90,
stop_loss_pct=-3.0,
rationale="Defensive stop-loss (planner failure)",
),
],
)
for c in candidates
]
return DayPlaybook(
date=today,
market=market,
market_outlook=MarketOutlook.NEUTRAL_TO_BEARISH,
default_action=ScenarioAction.HOLD,
stock_playbooks=stock_playbooks,
global_rules=[
GlobalRule(
condition="portfolio_pnl_pct < -2.0",
action=ScenarioAction.REDUCE_ALL,
rationale="Defensive: reduce on loss threshold",
),
],
)

View File

@@ -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

View File

@@ -300,9 +300,17 @@ class TestContextAggregator:
# Verify data exists in each layer # Verify data exists in each layer
store = aggregator.store store = aggregator.store
assert store.get_context(ContextLayer.L6_DAILY, date, "total_pnl") == 1000.0 assert store.get_context(ContextLayer.L6_DAILY, date, "total_pnl") == 1000.0
current_week = datetime.now(UTC).strftime("%Y-W%V") from datetime import date as date_cls
assert store.get_context(ContextLayer.L5_WEEKLY, current_week, "weekly_pnl") is not None trade_date = date_cls.fromisoformat(date)
# Further layers depend on time alignment, just verify no crashes iso_year, iso_week, _ = trade_date.isocalendar()
trade_week = f"{iso_year}-W{iso_week:02d}"
assert store.get_context(ContextLayer.L5_WEEKLY, trade_week, "weekly_pnl") is not None
trade_month = f"{trade_date.year}-{trade_date.month:02d}"
trade_quarter = f"{trade_date.year}-Q{(trade_date.month - 1) // 3 + 1}"
trade_year = str(trade_date.year)
assert store.get_context(ContextLayer.L4_MONTHLY, trade_month, "monthly_pnl") == 1000.0
assert store.get_context(ContextLayer.L3_QUARTERLY, trade_quarter, "quarterly_pnl") == 1000.0
assert store.get_context(ContextLayer.L2_ANNUAL, trade_year, "annual_pnl") == 1000.0
class TestLayerMetadata: class TestLayerMetadata:

View File

@@ -1,12 +1,47 @@
"""Tests for main trading loop telegram integration.""" """Tests for main trading loop integration."""
import asyncio from datetime import date
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import ANY, AsyncMock, MagicMock, patch
import pytest import pytest
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected
from src.context.layer import ContextLayer
from src.main import safe_float, trading_cycle from src.main import safe_float, trading_cycle
from src.strategy.models import (
DayPlaybook,
ScenarioAction,
StockCondition,
StockScenario,
)
from src.strategy.scenario_engine import ScenarioEngine, ScenarioMatch
def _make_playbook(market: str = "KR") -> DayPlaybook:
"""Create a minimal empty playbook for testing."""
return DayPlaybook(date=date(2026, 2, 8), market=market)
def _make_buy_match(stock_code: str = "005930") -> ScenarioMatch:
"""Create a ScenarioMatch that returns BUY."""
return ScenarioMatch(
stock_code=stock_code,
matched_scenario=None,
action=ScenarioAction.BUY,
confidence=85,
rationale="Test buy",
)
def _make_hold_match(stock_code: str = "005930") -> ScenarioMatch:
"""Create a ScenarioMatch that returns HOLD."""
return ScenarioMatch(
stock_code=stock_code,
matched_scenario=None,
action=ScenarioAction.HOLD,
confidence=0,
rationale="No scenario conditions met",
)
class TestSafeFloat: class TestSafeFloat:
@@ -81,15 +116,16 @@ class TestTradingCycleTelegramIntegration:
return broker return broker
@pytest.fixture @pytest.fixture
def mock_brain(self) -> MagicMock: def mock_scenario_engine(self) -> MagicMock:
"""Create mock brain that decides to buy.""" """Create mock scenario engine that returns BUY."""
brain = MagicMock() engine = MagicMock(spec=ScenarioEngine)
decision = MagicMock() engine.evaluate = MagicMock(return_value=_make_buy_match())
decision.action = "BUY" return engine
decision.confidence = 85
decision.rationale = "Test buy" @pytest.fixture
brain.decide = AsyncMock(return_value=decision) def mock_playbook(self) -> DayPlaybook:
return brain """Create a minimal day playbook."""
return _make_playbook()
@pytest.fixture @pytest.fixture
def mock_risk(self) -> MagicMock: def mock_risk(self) -> MagicMock:
@@ -134,6 +170,7 @@ class TestTradingCycleTelegramIntegration:
telegram.notify_trade_execution = AsyncMock() telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock() telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock() telegram.notify_circuit_breaker = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
return telegram return telegram
@pytest.fixture @pytest.fixture
@@ -151,7 +188,8 @@ class TestTradingCycleTelegramIntegration:
self, self,
mock_broker: MagicMock, mock_broker: MagicMock,
mock_overseas_broker: MagicMock, mock_overseas_broker: MagicMock,
mock_brain: MagicMock, mock_scenario_engine: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -165,7 +203,8 @@ class TestTradingCycleTelegramIntegration:
await trading_cycle( await trading_cycle(
broker=mock_broker, broker=mock_broker,
overseas_broker=mock_overseas_broker, overseas_broker=mock_overseas_broker,
brain=mock_brain, scenario_engine=mock_scenario_engine,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -190,7 +229,8 @@ class TestTradingCycleTelegramIntegration:
self, self,
mock_broker: MagicMock, mock_broker: MagicMock,
mock_overseas_broker: MagicMock, mock_overseas_broker: MagicMock,
mock_brain: MagicMock, mock_scenario_engine: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -208,7 +248,8 @@ class TestTradingCycleTelegramIntegration:
await trading_cycle( await trading_cycle(
broker=mock_broker, broker=mock_broker,
overseas_broker=mock_overseas_broker, overseas_broker=mock_overseas_broker,
brain=mock_brain, scenario_engine=mock_scenario_engine,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -228,7 +269,8 @@ class TestTradingCycleTelegramIntegration:
self, self,
mock_broker: MagicMock, mock_broker: MagicMock,
mock_overseas_broker: MagicMock, mock_overseas_broker: MagicMock,
mock_brain: MagicMock, mock_scenario_engine: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -250,7 +292,8 @@ class TestTradingCycleTelegramIntegration:
await trading_cycle( await trading_cycle(
broker=mock_broker, broker=mock_broker,
overseas_broker=mock_overseas_broker, overseas_broker=mock_overseas_broker,
brain=mock_brain, scenario_engine=mock_scenario_engine,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -275,7 +318,8 @@ class TestTradingCycleTelegramIntegration:
self, self,
mock_broker: MagicMock, mock_broker: MagicMock,
mock_overseas_broker: MagicMock, mock_overseas_broker: MagicMock,
mock_brain: MagicMock, mock_scenario_engine: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -299,7 +343,8 @@ class TestTradingCycleTelegramIntegration:
await trading_cycle( await trading_cycle(
broker=mock_broker, broker=mock_broker,
overseas_broker=mock_overseas_broker, overseas_broker=mock_overseas_broker,
brain=mock_brain, scenario_engine=mock_scenario_engine,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -319,7 +364,8 @@ class TestTradingCycleTelegramIntegration:
self, self,
mock_broker: MagicMock, mock_broker: MagicMock,
mock_overseas_broker: MagicMock, mock_overseas_broker: MagicMock,
mock_brain: MagicMock, mock_scenario_engine: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -329,18 +375,15 @@ class TestTradingCycleTelegramIntegration:
mock_market: MagicMock, mock_market: MagicMock,
) -> None: ) -> None:
"""Test no trade notification sent when decision is HOLD.""" """Test no trade notification sent when decision is HOLD."""
# Change brain decision to HOLD # Scenario engine returns HOLD
decision = MagicMock() mock_scenario_engine.evaluate = MagicMock(return_value=_make_hold_match())
decision.action = "HOLD"
decision.confidence = 50
decision.rationale = "Insufficient signal"
mock_brain.decide = AsyncMock(return_value=decision)
with patch("src.main.log_trade"): with patch("src.main.log_trade"):
await trading_cycle( await trading_cycle(
broker=mock_broker, broker=mock_broker,
overseas_broker=mock_overseas_broker, overseas_broker=mock_overseas_broker,
brain=mock_brain, scenario_engine=mock_scenario_engine,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -472,15 +515,16 @@ class TestOverseasBalanceParsing:
return market return market
@pytest.fixture @pytest.fixture
def mock_brain_hold(self) -> MagicMock: def mock_scenario_engine_hold(self) -> MagicMock:
"""Create mock brain that always holds.""" """Create mock scenario engine that always returns HOLD."""
brain = MagicMock() engine = MagicMock(spec=ScenarioEngine)
decision = MagicMock() engine.evaluate = MagicMock(return_value=_make_hold_match("AAPL"))
decision.action = "HOLD" return engine
decision.confidence = 50
decision.rationale = "Testing balance parsing" @pytest.fixture
brain.decide = AsyncMock(return_value=decision) def mock_playbook(self) -> DayPlaybook:
return brain """Create a minimal playbook."""
return _make_playbook("US")
@pytest.fixture @pytest.fixture
def mock_risk(self) -> MagicMock: def mock_risk(self) -> MagicMock:
@@ -517,14 +561,17 @@ class TestOverseasBalanceParsing:
@pytest.fixture @pytest.fixture
def mock_telegram(self) -> MagicMock: def mock_telegram(self) -> MagicMock:
"""Create mock telegram client.""" """Create mock telegram client."""
return MagicMock() telegram = MagicMock()
telegram.notify_scenario_matched = AsyncMock()
return telegram
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_overseas_balance_list_format( async def test_overseas_balance_list_format(
self, self,
mock_domestic_broker: MagicMock, mock_domestic_broker: MagicMock,
mock_overseas_broker_with_list: MagicMock, mock_overseas_broker_with_list: MagicMock,
mock_brain_hold: MagicMock, mock_scenario_engine_hold: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -539,7 +586,8 @@ class TestOverseasBalanceParsing:
await trading_cycle( await trading_cycle(
broker=mock_domestic_broker, broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_list, overseas_broker=mock_overseas_broker_with_list,
brain=mock_brain_hold, scenario_engine=mock_scenario_engine_hold,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -559,7 +607,8 @@ class TestOverseasBalanceParsing:
self, self,
mock_domestic_broker: MagicMock, mock_domestic_broker: MagicMock,
mock_overseas_broker_with_dict: MagicMock, mock_overseas_broker_with_dict: MagicMock,
mock_brain_hold: MagicMock, mock_scenario_engine_hold: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -574,7 +623,8 @@ class TestOverseasBalanceParsing:
await trading_cycle( await trading_cycle(
broker=mock_domestic_broker, broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_dict, overseas_broker=mock_overseas_broker_with_dict,
brain=mock_brain_hold, scenario_engine=mock_scenario_engine_hold,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -594,7 +644,8 @@ class TestOverseasBalanceParsing:
self, self,
mock_domestic_broker: MagicMock, mock_domestic_broker: MagicMock,
mock_overseas_broker_with_empty: MagicMock, mock_overseas_broker_with_empty: MagicMock,
mock_brain_hold: MagicMock, mock_scenario_engine_hold: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -609,7 +660,8 @@ class TestOverseasBalanceParsing:
await trading_cycle( await trading_cycle(
broker=mock_domestic_broker, broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_empty, overseas_broker=mock_overseas_broker_with_empty,
brain=mock_brain_hold, scenario_engine=mock_scenario_engine_hold,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -629,7 +681,8 @@ class TestOverseasBalanceParsing:
self, self,
mock_domestic_broker: MagicMock, mock_domestic_broker: MagicMock,
mock_overseas_broker_with_empty_price: MagicMock, mock_overseas_broker_with_empty_price: MagicMock,
mock_brain_hold: MagicMock, mock_scenario_engine_hold: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock, mock_risk: MagicMock,
mock_db: MagicMock, mock_db: MagicMock,
mock_decision_logger: MagicMock, mock_decision_logger: MagicMock,
@@ -644,7 +697,8 @@ class TestOverseasBalanceParsing:
await trading_cycle( await trading_cycle(
broker=mock_domestic_broker, broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_empty_price, overseas_broker=mock_overseas_broker_with_empty_price,
brain=mock_brain_hold, scenario_engine=mock_scenario_engine_hold,
playbook=mock_playbook,
risk=mock_risk, risk=mock_risk,
db_conn=mock_db, db_conn=mock_db,
decision_logger=mock_decision_logger, decision_logger=mock_decision_logger,
@@ -658,3 +712,404 @@ class TestOverseasBalanceParsing:
# Verify price API was called # Verify price API was called
mock_overseas_broker_with_empty_price.get_overseas_price.assert_called_once() mock_overseas_broker_with_empty_price.get_overseas_price.assert_called_once()
class TestScenarioEngineIntegration:
"""Test scenario engine integration in trading_cycle."""
@pytest.fixture
def mock_broker(self) -> MagicMock:
"""Create mock broker with standard domestic data."""
broker = MagicMock()
broker.get_orderbook = AsyncMock(
return_value={
"output1": {"stck_prpr": "50000", "frgn_ntby_qty": "100"}
}
)
broker.get_balance = AsyncMock(
return_value={
"output2": [
{
"tot_evlu_amt": "10000000",
"dnca_tot_amt": "5000000",
"pchs_amt_smtl_amt": "9500000",
}
]
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
return broker
@pytest.fixture
def mock_market(self) -> MagicMock:
"""Create mock KR market."""
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
return market
@pytest.fixture
def mock_telegram(self) -> MagicMock:
"""Create mock telegram with all notification methods."""
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
return telegram
@pytest.mark.asyncio
async def test_scenario_engine_called_with_enriched_market_data(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test scenario engine receives market_data enriched with scanner metrics."""
from src.analysis.smart_scanner import ScanCandidate
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
playbook = _make_playbook()
candidate = ScanCandidate(
stock_code="005930", name="Samsung", price=50000,
volume=1000000, volume_ratio=3.5, rsi=25.0,
signal="oversold", score=85.0,
)
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=playbook,
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=MagicMock(),
context_store=MagicMock(get_latest_timeframe=MagicMock(return_value=None)),
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
scan_candidates={"KR": {"005930": candidate}},
)
# Verify evaluate was called
engine.evaluate.assert_called_once()
call_args = engine.evaluate.call_args
market_data = call_args[0][2] # 3rd positional arg
portfolio_data = call_args[0][3] # 4th positional arg
# Scanner data should be enriched into market_data
assert market_data["rsi"] == 25.0
assert market_data["volume_ratio"] == 3.5
assert market_data["current_price"] == 50000.0
# Portfolio data should include pnl
assert "portfolio_pnl_pct" in portfolio_data
assert "total_cash" in portfolio_data
@pytest.mark.asyncio
async def test_trading_cycle_sets_l7_context_keys(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test L7 context is written with market-scoped keys."""
from src.analysis.smart_scanner import ScanCandidate
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
playbook = _make_playbook()
context_store = MagicMock(get_latest_timeframe=MagicMock(return_value=None))
candidate = ScanCandidate(
stock_code="005930", name="Samsung", price=50000,
volume=1000000, volume_ratio=3.5, rsi=25.0,
signal="oversold", score=85.0,
)
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=playbook,
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=MagicMock(),
context_store=context_store,
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
scan_candidates={"KR": {"005930": candidate}},
)
context_store.set_context.assert_any_call(
ContextLayer.L7_REALTIME,
ANY,
"volatility_KR_005930",
{"momentum_score": 50.0, "volume_surge": 1.0, "price_change_1m": 0.0},
)
context_store.set_context.assert_any_call(
ContextLayer.L7_REALTIME,
ANY,
"price_KR_005930",
{"current_price": 50000.0},
)
context_store.set_context.assert_any_call(
ContextLayer.L7_REALTIME,
ANY,
"rsi_KR_005930",
{"rsi": 25.0},
)
context_store.set_context.assert_any_call(
ContextLayer.L7_REALTIME,
ANY,
"volume_ratio_KR_005930",
{"volume_ratio": 3.5},
)
@pytest.mark.asyncio
async def test_scan_candidates_market_scoped(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test scan_candidates uses market-scoped lookup, ignoring other markets."""
from src.analysis.smart_scanner import ScanCandidate
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
# Candidate stored under US market — should NOT be found for KR market
us_candidate = ScanCandidate(
stock_code="005930", name="Overlap", price=100,
volume=500000, volume_ratio=5.0, rsi=15.0,
signal="oversold", score=90.0,
)
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=_make_playbook(),
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=MagicMock(),
context_store=MagicMock(get_latest_timeframe=MagicMock(return_value=None)),
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market, # KR market
stock_code="005930",
scan_candidates={"US": {"005930": us_candidate}}, # Wrong market
)
# Should NOT have rsi/volume_ratio because candidate is under US, not KR
market_data = engine.evaluate.call_args[0][2]
assert "rsi" not in market_data
assert "volume_ratio" not in market_data
@pytest.mark.asyncio
async def test_scenario_engine_called_without_scanner_data(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test scenario engine works when stock has no scan candidate."""
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
playbook = _make_playbook()
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=playbook,
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=MagicMock(),
context_store=MagicMock(get_latest_timeframe=MagicMock(return_value=None)),
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
scan_candidates={}, # No scanner data
)
# Should still work, just without rsi/volume_ratio
engine.evaluate.assert_called_once()
market_data = engine.evaluate.call_args[0][2]
assert "rsi" not in market_data
assert "volume_ratio" not in market_data
assert market_data["current_price"] == 50000.0
@pytest.mark.asyncio
async def test_scenario_matched_notification_sent(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test telegram notification sent when a scenario matches."""
# Create a match with matched_scenario (not None)
scenario = StockScenario(
condition=StockCondition(rsi_below=30),
action=ScenarioAction.BUY,
confidence=88,
rationale="RSI oversold bounce",
)
match = ScenarioMatch(
stock_code="005930",
matched_scenario=scenario,
action=ScenarioAction.BUY,
confidence=88,
rationale="RSI oversold bounce",
match_details={"rsi": 25.0},
)
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=match)
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=_make_playbook(),
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=MagicMock(),
context_store=MagicMock(get_latest_timeframe=MagicMock(return_value=None)),
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
scan_candidates={},
)
# Scenario matched notification should be sent
mock_telegram.notify_scenario_matched.assert_called_once()
call_kwargs = mock_telegram.notify_scenario_matched.call_args.kwargs
assert call_kwargs["stock_code"] == "005930"
assert call_kwargs["action"] == "BUY"
assert "rsi=25.0" in call_kwargs["condition_summary"]
@pytest.mark.asyncio
async def test_no_scenario_matched_notification_on_default_hold(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test no scenario notification when default HOLD is returned."""
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=_make_playbook(),
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=MagicMock(),
context_store=MagicMock(get_latest_timeframe=MagicMock(return_value=None)),
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
scan_candidates={},
)
# No scenario matched notification for default HOLD
mock_telegram.notify_scenario_matched.assert_not_called()
@pytest.mark.asyncio
async def test_decision_logger_receives_scenario_match_details(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test decision logger context includes scenario match details."""
match = ScenarioMatch(
stock_code="005930",
matched_scenario=None,
action=ScenarioAction.HOLD,
confidence=0,
rationale="No match",
match_details={"rsi": 45.0, "volume_ratio": 1.2},
)
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=match)
decision_logger = MagicMock()
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=_make_playbook(),
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=decision_logger,
context_store=MagicMock(get_latest_timeframe=MagicMock(return_value=None)),
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
scan_candidates={},
)
decision_logger.log_decision.assert_called_once()
call_kwargs = decision_logger.log_decision.call_args.kwargs
assert "scenario_match" in call_kwargs["context_snapshot"]
assert call_kwargs["context_snapshot"]["scenario_match"]["rsi"] == 45.0
@pytest.mark.asyncio
async def test_reduce_all_does_not_execute_order(
self, mock_broker: MagicMock, mock_market: MagicMock, mock_telegram: MagicMock,
) -> None:
"""Test REDUCE_ALL action does not trigger order execution."""
match = ScenarioMatch(
stock_code="005930",
matched_scenario=None,
action=ScenarioAction.REDUCE_ALL,
confidence=100,
rationale="Global rule: portfolio loss > 2%",
)
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=match)
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=_make_playbook(),
risk=MagicMock(),
db_conn=MagicMock(),
decision_logger=MagicMock(),
context_store=MagicMock(get_latest_timeframe=MagicMock(return_value=None)),
criticality_assessor=MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
),
telegram=mock_telegram,
market=mock_market,
stock_code="005930",
scan_candidates={},
)
# REDUCE_ALL is not BUY or SELL — no order sent
mock_broker.send_order.assert_not_called()
mock_telegram.notify_trade_execution.assert_not_called()

View File

@@ -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

View File

@@ -0,0 +1,552 @@
"""Tests for PreMarketPlanner — Gemini prompt builder + response parser."""
from __future__ import annotations
import json
from datetime import date
from unittest.mock import AsyncMock, MagicMock
import pytest
from src.analysis.smart_scanner import ScanCandidate
from src.brain.gemini_client import TradeDecision
from src.config import Settings
from src.context.store import ContextLayer
from src.strategy.models import (
CrossMarketContext,
DayPlaybook,
MarketOutlook,
PlaybookStatus,
ScenarioAction,
)
from src.strategy.pre_market_planner import PreMarketPlanner
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _candidate(
code: str = "005930",
name: str = "Samsung",
price: float = 71000,
rsi: float = 28.5,
volume_ratio: float = 3.2,
signal: str = "oversold",
score: float = 82.0,
) -> ScanCandidate:
return ScanCandidate(
stock_code=code,
name=name,
price=price,
volume=1_500_000,
volume_ratio=volume_ratio,
rsi=rsi,
signal=signal,
score=score,
)
def _gemini_response_json(
outlook: str = "neutral_to_bullish",
stocks: list[dict] | None = None,
global_rules: list[dict] | None = None,
) -> str:
"""Build a valid Gemini JSON response."""
if stocks is None:
stocks = [
{
"stock_code": "005930",
"scenarios": [
{
"condition": {"rsi_below": 30, "volume_ratio_above": 2.5},
"action": "BUY",
"confidence": 85,
"allocation_pct": 15.0,
"stop_loss_pct": -2.0,
"take_profit_pct": 4.0,
"rationale": "Oversold bounce with high volume",
}
],
}
]
if global_rules is None:
global_rules = [
{
"condition": "portfolio_pnl_pct < -2.0",
"action": "REDUCE_ALL",
"rationale": "Near circuit breaker",
}
]
return json.dumps(
{"market_outlook": outlook, "global_rules": global_rules, "stocks": stocks}
)
def _make_planner(
gemini_response: str = "",
token_count: int = 200,
context_data: dict | None = None,
scorecard_data: dict | None = None,
) -> PreMarketPlanner:
"""Create a PreMarketPlanner with mocked dependencies."""
if not gemini_response:
gemini_response = _gemini_response_json()
# Mock GeminiClient
gemini = AsyncMock()
gemini.decide = AsyncMock(
return_value=TradeDecision(
action="HOLD",
confidence=0,
rationale=gemini_response,
token_count=token_count,
)
)
# Mock ContextStore
store = MagicMock()
store.get_context = MagicMock(return_value=scorecard_data)
# Mock ContextSelector
selector = MagicMock()
selector.select_layers = MagicMock(return_value=[ContextLayer.L7_REALTIME, ContextLayer.L6_DAILY])
selector.get_context_data = MagicMock(return_value=context_data or {})
settings = Settings(
KIS_APP_KEY="test",
KIS_APP_SECRET="test",
KIS_ACCOUNT_NO="12345678-01",
GEMINI_API_KEY="test",
)
return PreMarketPlanner(gemini, store, selector, settings)
# ---------------------------------------------------------------------------
# generate_playbook
# ---------------------------------------------------------------------------
class TestGeneratePlaybook:
@pytest.mark.asyncio
async def test_basic_generation(self) -> None:
planner = _make_planner()
candidates = [_candidate()]
pb = await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
assert isinstance(pb, DayPlaybook)
assert pb.market == "KR"
assert pb.stock_count == 1
assert pb.scenario_count == 1
assert pb.market_outlook == MarketOutlook.NEUTRAL_TO_BULLISH
assert pb.token_count == 200
@pytest.mark.asyncio
async def test_empty_candidates_returns_empty_playbook(self) -> None:
planner = _make_planner()
pb = await planner.generate_playbook("KR", [], today=date(2026, 2, 8))
assert pb.stock_count == 0
assert pb.scenario_count == 0
assert pb.market_outlook == MarketOutlook.NEUTRAL
@pytest.mark.asyncio
async def test_gemini_failure_returns_defensive(self) -> None:
planner = _make_planner()
planner._gemini.decide = AsyncMock(side_effect=RuntimeError("API timeout"))
candidates = [_candidate()]
pb = await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
assert pb.default_action == ScenarioAction.HOLD
assert pb.market_outlook == MarketOutlook.NEUTRAL_TO_BEARISH
assert pb.stock_count == 1
# Defensive playbook has stop-loss scenarios
assert pb.stock_playbooks[0].scenarios[0].action == ScenarioAction.SELL
@pytest.mark.asyncio
async def test_gemini_failure_empty_when_defensive_disabled(self) -> None:
planner = _make_planner()
planner._settings.DEFENSIVE_PLAYBOOK_ON_FAILURE = False
planner._gemini.decide = AsyncMock(side_effect=RuntimeError("fail"))
candidates = [_candidate()]
pb = await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
assert pb.stock_count == 0
@pytest.mark.asyncio
async def test_multiple_candidates(self) -> None:
stocks = [
{
"stock_code": "005930",
"scenarios": [
{
"condition": {"rsi_below": 30},
"action": "BUY",
"confidence": 85,
"rationale": "Oversold",
}
],
},
{
"stock_code": "AAPL",
"scenarios": [
{
"condition": {"rsi_above": 75},
"action": "SELL",
"confidence": 80,
"rationale": "Overbought",
}
],
},
]
planner = _make_planner(gemini_response=_gemini_response_json(stocks=stocks))
candidates = [_candidate(), _candidate(code="AAPL", name="Apple")]
pb = await planner.generate_playbook("US", candidates, today=date(2026, 2, 8))
assert pb.stock_count == 2
codes = [sp.stock_code for sp in pb.stock_playbooks]
assert "005930" in codes
assert "AAPL" in codes
@pytest.mark.asyncio
async def test_unknown_stock_in_response_skipped(self) -> None:
stocks = [
{
"stock_code": "005930",
"scenarios": [{"condition": {"rsi_below": 30}, "action": "BUY", "confidence": 85, "rationale": "ok"}],
},
{
"stock_code": "UNKNOWN",
"scenarios": [{"condition": {"rsi_below": 20}, "action": "BUY", "confidence": 90, "rationale": "bad"}],
},
]
planner = _make_planner(gemini_response=_gemini_response_json(stocks=stocks))
candidates = [_candidate()] # Only 005930
pb = await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
assert pb.stock_count == 1
assert pb.stock_playbooks[0].stock_code == "005930"
@pytest.mark.asyncio
async def test_global_rules_parsed(self) -> None:
planner = _make_planner()
candidates = [_candidate()]
pb = await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
assert len(pb.global_rules) == 1
assert pb.global_rules[0].action == ScenarioAction.REDUCE_ALL
@pytest.mark.asyncio
async def test_token_count_from_decision(self) -> None:
planner = _make_planner(token_count=450)
candidates = [_candidate()]
pb = await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
assert pb.token_count == 450
# ---------------------------------------------------------------------------
# _parse_response
# ---------------------------------------------------------------------------
class TestParseResponse:
def test_parse_full_response(self) -> None:
planner = _make_planner()
response = _gemini_response_json(outlook="bearish")
candidates = [_candidate()]
pb = planner._parse_response(response, date(2026, 2, 8), "KR", candidates, None)
assert pb.market_outlook == MarketOutlook.BEARISH
assert pb.stock_count == 1
assert pb.stock_playbooks[0].scenarios[0].confidence == 85
def test_parse_with_markdown_fences(self) -> None:
planner = _make_planner()
response = f"```json\n{_gemini_response_json()}\n```"
candidates = [_candidate()]
pb = planner._parse_response(response, date(2026, 2, 8), "KR", candidates, None)
assert pb.stock_count == 1
def test_parse_unknown_outlook_defaults_neutral(self) -> None:
planner = _make_planner()
response = _gemini_response_json(outlook="super_bullish")
candidates = [_candidate()]
pb = planner._parse_response(response, date(2026, 2, 8), "KR", candidates, None)
assert pb.market_outlook == MarketOutlook.NEUTRAL
def test_parse_scenario_with_all_condition_fields(self) -> None:
planner = _make_planner()
stocks = [
{
"stock_code": "005930",
"scenarios": [
{
"condition": {
"rsi_below": 25,
"volume_ratio_above": 3.0,
"price_change_pct_below": -2.0,
},
"action": "BUY",
"confidence": 92,
"allocation_pct": 20.0,
"stop_loss_pct": -3.0,
"take_profit_pct": 5.0,
"rationale": "Multi-condition entry",
}
],
}
]
response = _gemini_response_json(stocks=stocks)
candidates = [_candidate()]
pb = planner._parse_response(response, date(2026, 2, 8), "KR", candidates, None)
sc = pb.stock_playbooks[0].scenarios[0]
assert sc.condition.rsi_below == 25
assert sc.condition.volume_ratio_above == 3.0
assert sc.condition.price_change_pct_below == -2.0
assert sc.allocation_pct == 20.0
assert sc.stop_loss_pct == -3.0
assert sc.take_profit_pct == 5.0
def test_parse_empty_condition_scenario_skipped(self) -> None:
planner = _make_planner()
stocks = [
{
"stock_code": "005930",
"scenarios": [
{
"condition": {},
"action": "BUY",
"confidence": 85,
"rationale": "No conditions",
},
{
"condition": {"rsi_below": 30},
"action": "BUY",
"confidence": 80,
"rationale": "Valid",
},
],
}
]
response = _gemini_response_json(stocks=stocks)
candidates = [_candidate()]
pb = planner._parse_response(response, date(2026, 2, 8), "KR", candidates, None)
# Empty condition scenario skipped, valid one kept
assert pb.stock_count == 1
assert pb.stock_playbooks[0].scenarios[0].confidence == 80
def test_parse_max_scenarios_enforced(self) -> None:
planner = _make_planner()
# Settings default MAX_SCENARIOS_PER_STOCK = 5
scenarios = [
{
"condition": {"rsi_below": 20 + i},
"action": "BUY",
"confidence": 80 + i,
"rationale": f"Scenario {i}",
}
for i in range(8) # 8 scenarios, should be capped to 5
]
stocks = [{"stock_code": "005930", "scenarios": scenarios}]
response = _gemini_response_json(stocks=stocks)
candidates = [_candidate()]
pb = planner._parse_response(response, date(2026, 2, 8), "KR", candidates, None)
assert len(pb.stock_playbooks[0].scenarios) == 5
def test_parse_invalid_json_raises(self) -> None:
planner = _make_planner()
candidates = [_candidate()]
with pytest.raises(json.JSONDecodeError):
planner._parse_response("not json at all", date(2026, 2, 8), "KR", candidates, None)
def test_parse_cross_market_preserved(self) -> None:
planner = _make_planner()
response = _gemini_response_json()
candidates = [_candidate()]
cross = CrossMarketContext(market="US", date="2026-02-07", total_pnl=1.5, win_rate=60)
pb = planner._parse_response(response, date(2026, 2, 8), "KR", candidates, cross)
assert pb.cross_market is not None
assert pb.cross_market.market == "US"
assert pb.cross_market.total_pnl == 1.5
# ---------------------------------------------------------------------------
# build_cross_market_context
# ---------------------------------------------------------------------------
class TestBuildCrossMarketContext:
def test_kr_reads_us_scorecard(self) -> None:
scorecard = {"total_pnl": 2.5, "win_rate": 65, "index_change_pct": 0.8, "lessons": ["Stay patient"]}
planner = _make_planner(scorecard_data=scorecard)
ctx = planner.build_cross_market_context("KR", today=date(2026, 2, 8))
assert ctx is not None
assert ctx.market == "US"
assert ctx.total_pnl == 2.5
assert ctx.win_rate == 65
assert "Stay patient" in ctx.lessons
# Verify it queried scorecard_US
planner._context_store.get_context.assert_called_once_with(
ContextLayer.L6_DAILY, "2026-02-08", "scorecard_US"
)
def test_us_reads_kr_scorecard(self) -> None:
scorecard = {"total_pnl": -1.0, "win_rate": 40, "index_change_pct": -0.5}
planner = _make_planner(scorecard_data=scorecard)
ctx = planner.build_cross_market_context("US", today=date(2026, 2, 8))
assert ctx is not None
assert ctx.market == "KR"
assert ctx.total_pnl == -1.0
planner._context_store.get_context.assert_called_once_with(
ContextLayer.L6_DAILY, "2026-02-08", "scorecard_KR"
)
def test_no_scorecard_returns_none(self) -> None:
planner = _make_planner(scorecard_data=None)
ctx = planner.build_cross_market_context("KR", today=date(2026, 2, 8))
assert ctx is None
def test_invalid_scorecard_returns_none(self) -> None:
planner = _make_planner(scorecard_data="not a dict and not json")
ctx = planner.build_cross_market_context("KR", today=date(2026, 2, 8))
assert ctx is None
# ---------------------------------------------------------------------------
# _build_prompt
# ---------------------------------------------------------------------------
class TestBuildPrompt:
def test_prompt_contains_candidates(self) -> None:
planner = _make_planner()
candidates = [_candidate(code="005930", name="Samsung")]
prompt = planner._build_prompt("KR", candidates, {}, None)
assert "005930" in prompt
assert "Samsung" in prompt
assert "RSI=" in prompt
assert "volume_ratio=" in prompt
def test_prompt_contains_cross_market(self) -> None:
planner = _make_planner()
cross = CrossMarketContext(
market="US", date="2026-02-07", total_pnl=1.5,
win_rate=60, index_change_pct=0.8, lessons=["Cut losses early"],
)
prompt = planner._build_prompt("KR", [_candidate()], {}, cross)
assert "Other Market (US)" in prompt
assert "+1.50%" in prompt
assert "Cut losses early" in prompt
def test_prompt_contains_context_data(self) -> None:
planner = _make_planner()
context = {"L6_DAILY": {"win_rate": 0.65, "total_pnl": 2.5}}
prompt = planner._build_prompt("KR", [_candidate()], context, None)
assert "Strategic Context" in prompt
assert "L6_DAILY" in prompt
assert "win_rate" in prompt
def test_prompt_contains_max_scenarios(self) -> None:
planner = _make_planner()
prompt = planner._build_prompt("KR", [_candidate()], {}, None)
assert f"Max {planner._settings.MAX_SCENARIOS_PER_STOCK} scenarios" in prompt
def test_prompt_market_name(self) -> None:
planner = _make_planner()
prompt = planner._build_prompt("US", [_candidate()], {}, None)
assert "US market" in prompt
# ---------------------------------------------------------------------------
# _extract_json
# ---------------------------------------------------------------------------
class TestExtractJson:
def test_plain_json(self) -> None:
assert PreMarketPlanner._extract_json('{"a": 1}') == '{"a": 1}'
def test_with_json_fence(self) -> None:
text = '```json\n{"a": 1}\n```'
assert PreMarketPlanner._extract_json(text) == '{"a": 1}'
def test_with_plain_fence(self) -> None:
text = '```\n{"a": 1}\n```'
assert PreMarketPlanner._extract_json(text) == '{"a": 1}'
def test_with_whitespace(self) -> None:
text = ' \n {"a": 1} \n '
assert PreMarketPlanner._extract_json(text) == '{"a": 1}'
# ---------------------------------------------------------------------------
# Defensive playbook
# ---------------------------------------------------------------------------
class TestDefensivePlaybook:
def test_defensive_has_stop_loss(self) -> None:
candidates = [_candidate(code="005930"), _candidate(code="AAPL")]
pb = PreMarketPlanner._defensive_playbook(date(2026, 2, 8), "KR", candidates)
assert pb.default_action == ScenarioAction.HOLD
assert pb.market_outlook == MarketOutlook.NEUTRAL_TO_BEARISH
assert pb.stock_count == 2
for sp in pb.stock_playbooks:
assert sp.scenarios[0].action == ScenarioAction.SELL
assert sp.scenarios[0].stop_loss_pct == -3.0
def test_defensive_has_global_rule(self) -> None:
pb = PreMarketPlanner._defensive_playbook(date(2026, 2, 8), "KR", [_candidate()])
assert len(pb.global_rules) == 1
assert pb.global_rules[0].action == ScenarioAction.REDUCE_ALL
def test_empty_playbook(self) -> None:
pb = PreMarketPlanner._empty_playbook(date(2026, 2, 8), "US")
assert pb.stock_count == 0
assert pb.market == "US"
assert pb.market_outlook == MarketOutlook.NEUTRAL

View File

@@ -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)

View File

@@ -0,0 +1,366 @@
"""Tests for strategy/playbook Pydantic models."""
from __future__ import annotations
from datetime import date
import pytest
from pydantic import ValidationError
from src.strategy.models import (
CrossMarketContext,
DayPlaybook,
GlobalRule,
MarketOutlook,
PlaybookStatus,
ScenarioAction,
StockCondition,
StockPlaybook,
StockScenario,
)
# ---------------------------------------------------------------------------
# StockCondition
# ---------------------------------------------------------------------------
class TestStockCondition:
def test_empty_condition(self) -> None:
cond = StockCondition()
assert not cond.has_any_condition()
def test_single_field(self) -> None:
cond = StockCondition(rsi_below=30.0)
assert cond.has_any_condition()
def test_multiple_fields(self) -> None:
cond = StockCondition(rsi_below=25.0, volume_ratio_above=3.0)
assert cond.has_any_condition()
def test_all_fields(self) -> None:
cond = StockCondition(
rsi_below=30,
rsi_above=10,
volume_ratio_above=2.0,
volume_ratio_below=10.0,
price_above=1000,
price_below=50000,
price_change_pct_above=-5.0,
price_change_pct_below=5.0,
)
assert cond.has_any_condition()
# ---------------------------------------------------------------------------
# StockScenario
# ---------------------------------------------------------------------------
class TestStockScenario:
def test_valid_scenario(self) -> None:
s = StockScenario(
condition=StockCondition(rsi_below=25.0),
action=ScenarioAction.BUY,
confidence=85,
allocation_pct=15.0,
stop_loss_pct=-2.0,
take_profit_pct=3.0,
rationale="Oversold bounce expected",
)
assert s.action == ScenarioAction.BUY
assert s.confidence == 85
def test_confidence_too_high(self) -> None:
with pytest.raises(ValidationError):
StockScenario(
condition=StockCondition(),
action=ScenarioAction.BUY,
confidence=101,
)
def test_confidence_too_low(self) -> None:
with pytest.raises(ValidationError):
StockScenario(
condition=StockCondition(),
action=ScenarioAction.BUY,
confidence=-1,
)
def test_allocation_too_high(self) -> None:
with pytest.raises(ValidationError):
StockScenario(
condition=StockCondition(),
action=ScenarioAction.BUY,
confidence=80,
allocation_pct=101.0,
)
def test_stop_loss_must_be_negative(self) -> None:
with pytest.raises(ValidationError):
StockScenario(
condition=StockCondition(),
action=ScenarioAction.BUY,
confidence=80,
stop_loss_pct=1.0,
)
def test_take_profit_must_be_positive(self) -> None:
with pytest.raises(ValidationError):
StockScenario(
condition=StockCondition(),
action=ScenarioAction.BUY,
confidence=80,
take_profit_pct=-1.0,
)
def test_defaults(self) -> None:
s = StockScenario(
condition=StockCondition(),
action=ScenarioAction.HOLD,
confidence=50,
)
assert s.allocation_pct == 10.0
assert s.stop_loss_pct == -2.0
assert s.take_profit_pct == 3.0
assert s.rationale == ""
# ---------------------------------------------------------------------------
# StockPlaybook
# ---------------------------------------------------------------------------
class TestStockPlaybook:
def test_valid_playbook(self) -> None:
pb = StockPlaybook(
stock_code="005930",
stock_name="Samsung Electronics",
scenarios=[
StockScenario(
condition=StockCondition(rsi_below=25.0),
action=ScenarioAction.BUY,
confidence=85,
),
],
)
assert pb.stock_code == "005930"
assert len(pb.scenarios) == 1
def test_empty_scenarios_rejected(self) -> None:
with pytest.raises(ValidationError):
StockPlaybook(
stock_code="005930",
scenarios=[],
)
def test_multiple_scenarios(self) -> None:
pb = StockPlaybook(
stock_code="AAPL",
scenarios=[
StockScenario(
condition=StockCondition(rsi_below=25.0),
action=ScenarioAction.BUY,
confidence=85,
),
StockScenario(
condition=StockCondition(rsi_above=75.0),
action=ScenarioAction.SELL,
confidence=80,
),
],
)
assert len(pb.scenarios) == 2
# ---------------------------------------------------------------------------
# GlobalRule
# ---------------------------------------------------------------------------
class TestGlobalRule:
def test_valid_rule(self) -> None:
rule = GlobalRule(
condition="portfolio_pnl_pct < -2.0",
action=ScenarioAction.REDUCE_ALL,
rationale="Risk limit approaching",
)
assert rule.action == ScenarioAction.REDUCE_ALL
def test_hold_rule(self) -> None:
rule = GlobalRule(
condition="volatility_index > 30",
action=ScenarioAction.HOLD,
)
assert rule.rationale == ""
# ---------------------------------------------------------------------------
# CrossMarketContext
# ---------------------------------------------------------------------------
class TestCrossMarketContext:
def test_valid_context(self) -> None:
ctx = CrossMarketContext(
market="US",
date="2026-02-07",
total_pnl=-1.5,
win_rate=40.0,
index_change_pct=-2.3,
key_events=["Fed rate decision"],
lessons=["Avoid tech sector on rate hike days"],
)
assert ctx.market == "US"
assert len(ctx.key_events) == 1
def test_defaults(self) -> None:
ctx = CrossMarketContext(market="KR", date="2026-02-07")
assert ctx.total_pnl == 0.0
assert ctx.key_events == []
assert ctx.lessons == []
# ---------------------------------------------------------------------------
# DayPlaybook
# ---------------------------------------------------------------------------
def _make_scenario(rsi_below: float = 25.0) -> StockScenario:
return StockScenario(
condition=StockCondition(rsi_below=rsi_below),
action=ScenarioAction.BUY,
confidence=85,
)
def _make_playbook(**kwargs) -> DayPlaybook:
defaults = {
"date": date(2026, 2, 7),
"market": "KR",
"stock_playbooks": [
StockPlaybook(stock_code="005930", scenarios=[_make_scenario()]),
],
}
defaults.update(kwargs)
return DayPlaybook(**defaults)
class TestDayPlaybook:
def test_valid_playbook(self) -> None:
pb = _make_playbook()
assert pb.market == "KR"
assert pb.date == date(2026, 2, 7)
assert pb.default_action == ScenarioAction.HOLD
assert pb.scenario_count == 1
assert pb.stock_count == 1
def test_generated_at_auto_set(self) -> None:
pb = _make_playbook()
assert pb.generated_at != ""
def test_explicit_generated_at(self) -> None:
pb = _make_playbook(generated_at="2026-02-07T08:30:00")
assert pb.generated_at == "2026-02-07T08:30:00"
def test_duplicate_stocks_rejected(self) -> None:
with pytest.raises(ValidationError):
DayPlaybook(
date=date(2026, 2, 7),
market="KR",
stock_playbooks=[
StockPlaybook(stock_code="005930", scenarios=[_make_scenario()]),
StockPlaybook(stock_code="005930", scenarios=[_make_scenario(30)]),
],
)
def test_empty_stock_playbooks_allowed(self) -> None:
pb = DayPlaybook(
date=date(2026, 2, 7),
market="KR",
stock_playbooks=[],
)
assert pb.stock_count == 0
assert pb.scenario_count == 0
def test_get_stock_playbook_found(self) -> None:
pb = _make_playbook()
result = pb.get_stock_playbook("005930")
assert result is not None
assert result.stock_code == "005930"
def test_get_stock_playbook_not_found(self) -> None:
pb = _make_playbook()
result = pb.get_stock_playbook("AAPL")
assert result is None
def test_with_global_rules(self) -> None:
pb = _make_playbook(
global_rules=[
GlobalRule(
condition="portfolio_pnl_pct < -2.0",
action=ScenarioAction.REDUCE_ALL,
),
],
)
assert len(pb.global_rules) == 1
def test_with_cross_market_context(self) -> None:
ctx = CrossMarketContext(market="US", date="2026-02-07", total_pnl=-1.5)
pb = _make_playbook(cross_market=ctx)
assert pb.cross_market is not None
assert pb.cross_market.market == "US"
def test_market_outlook(self) -> None:
pb = _make_playbook(market_outlook=MarketOutlook.BEARISH)
assert pb.market_outlook == MarketOutlook.BEARISH
def test_multiple_stocks_multiple_scenarios(self) -> None:
pb = DayPlaybook(
date=date(2026, 2, 7),
market="US",
stock_playbooks=[
StockPlaybook(
stock_code="AAPL",
scenarios=[_make_scenario(), _make_scenario(30)],
),
StockPlaybook(
stock_code="MSFT",
scenarios=[_make_scenario()],
),
],
)
assert pb.stock_count == 2
assert pb.scenario_count == 3
def test_serialization_roundtrip(self) -> None:
pb = _make_playbook(
market_outlook=MarketOutlook.BULLISH,
cross_market=CrossMarketContext(market="US", date="2026-02-07"),
)
json_str = pb.model_dump_json()
restored = DayPlaybook.model_validate_json(json_str)
assert restored.market == pb.market
assert restored.date == pb.date
assert restored.scenario_count == pb.scenario_count
assert restored.cross_market is not None
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class TestEnums:
def test_scenario_action_values(self) -> None:
assert ScenarioAction.BUY.value == "BUY"
assert ScenarioAction.SELL.value == "SELL"
assert ScenarioAction.HOLD.value == "HOLD"
assert ScenarioAction.REDUCE_ALL.value == "REDUCE_ALL"
def test_market_outlook_values(self) -> None:
assert len(MarketOutlook) == 5
def test_playbook_status_values(self) -> None:
assert PlaybookStatus.READY.value == "ready"
assert PlaybookStatus.EXPIRED.value == "expired"

View File

@@ -160,6 +160,83 @@ class TestNotificationSending:
assert "250.50" in payload["text"] assert "250.50" in payload["text"]
assert "92%" in payload["text"] assert "92%" in payload["text"]
@pytest.mark.asyncio
async def test_playbook_generated_format(self) -> None:
"""Playbook generated notification has expected fields."""
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True
)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await client.notify_playbook_generated(
market="KR",
stock_count=4,
scenario_count=12,
token_count=980,
)
payload = mock_post.call_args.kwargs["json"]
assert "Playbook Generated" in payload["text"]
assert "Market: KR" in payload["text"]
assert "Stocks: 4" in payload["text"]
assert "Scenarios: 12" in payload["text"]
assert "Tokens: 980" in payload["text"]
@pytest.mark.asyncio
async def test_scenario_matched_format(self) -> None:
"""Scenario matched notification has expected fields."""
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True
)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await client.notify_scenario_matched(
stock_code="AAPL",
action="BUY",
condition_summary="RSI < 30, volume_ratio > 2.0",
confidence=88.2,
)
payload = mock_post.call_args.kwargs["json"]
assert "Scenario Matched" in payload["text"]
assert "AAPL" in payload["text"]
assert "Action: BUY" in payload["text"]
assert "RSI < 30" in payload["text"]
assert "88%" in payload["text"]
@pytest.mark.asyncio
async def test_playbook_failed_format(self) -> None:
"""Playbook failed notification has expected fields."""
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True
)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await client.notify_playbook_failed(
market="US",
reason="Gemini timeout",
)
payload = mock_post.call_args.kwargs["json"]
assert "Playbook Failed" in payload["text"]
assert "Market: US" in payload["text"]
assert "Gemini timeout" in payload["text"]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_circuit_breaker_priority(self) -> None: async def test_circuit_breaker_priority(self) -> None:
"""Circuit breaker uses CRITICAL priority.""" """Circuit breaker uses CRITICAL priority."""
@@ -309,6 +386,73 @@ class TestMessagePriorities:
payload = mock_post.call_args.kwargs["json"] payload = mock_post.call_args.kwargs["json"]
assert NotificationPriority.CRITICAL.emoji in payload["text"] assert NotificationPriority.CRITICAL.emoji in payload["text"]
@pytest.mark.asyncio
async def test_playbook_generated_priority(self) -> None:
"""Playbook generated uses MEDIUM priority emoji."""
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True
)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await client.notify_playbook_generated(
market="KR",
stock_count=2,
scenario_count=4,
token_count=123,
)
payload = mock_post.call_args.kwargs["json"]
assert NotificationPriority.MEDIUM.emoji in payload["text"]
@pytest.mark.asyncio
async def test_playbook_failed_priority(self) -> None:
"""Playbook failed uses HIGH priority emoji."""
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True
)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await client.notify_playbook_failed(
market="KR",
reason="Invalid JSON",
)
payload = mock_post.call_args.kwargs["json"]
assert NotificationPriority.HIGH.emoji in payload["text"]
@pytest.mark.asyncio
async def test_scenario_matched_priority(self) -> None:
"""Scenario matched uses HIGH priority emoji."""
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True
)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await client.notify_scenario_matched(
stock_code="AAPL",
action="BUY",
condition_summary="RSI < 30",
confidence=80.0,
)
payload = mock_post.call_args.kwargs["json"]
assert NotificationPriority.HIGH.emoji in payload["text"]
class TestClientCleanup: class TestClientCleanup:
"""Test client cleanup behavior.""" """Test client cleanup behavior."""

View File

@@ -412,7 +412,7 @@ class TestMarketScanner:
scan_result = context_store.get_context( scan_result = context_store.get_context(
ContextLayer.L7_REALTIME, ContextLayer.L7_REALTIME,
latest_timeframe, latest_timeframe,
"KR_scan_result", "scan_result_KR",
) )
assert scan_result is not None assert scan_result is not None
assert scan_result["total_scanned"] == 3 assert scan_result["total_scanned"] == 3