Compare commits

...

30 Commits

Author SHA1 Message Date
agentson
86c94cff62 feat: cross-market date fix and strategic context selector (issue #88)
Some checks failed
CI / test (pull_request) Has been cancelled
KR planner now reads US scorecard from previous day (timezone-aware),
and generate_playbook uses STRATEGIC context selection.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-14 23:20:24 +09:00
692cb61991 Merge pull request 'feat: main.py에 일일 리뷰 연결 (issue #93)' (#122) from feature/issue-93-daily-review-integration into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #122
2026-02-14 23:15:26 +09:00
agentson
392422992b feat: integrate DailyReviewer into market close flow (issue #93)
Some checks failed
CI / test (pull_request) Has been cancelled
Extract _handle_market_close() helper that runs EOD aggregation,
generates scorecard with optional AI lessons, and sends Telegram summary.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-14 23:13:57 +09:00
cc637a9738 Merge pull request 'feat: Daily Reviewer - 시장별 성적표 생성 (issue #91)' (#121) from feature/issue-91-daily-reviewer into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #121
2026-02-14 23:08:05 +09:00
agentson
8c27473fed feat: DailyReviewer for market-scoped scorecards and AI lessons (issue #91)
Some checks failed
CI / test (pull_request) Has been cancelled
Generate per-market daily scorecards from decision_logs and trades,
optional Gemini-powered lessons, and store results in L6 context.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-14 23:07:12 +09:00
bde54c7487 Merge pull request 'feat: Decision outcome 업데이트 (issue #92)' (#120) from feature/issue-92-decision-outcome into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #120
2026-02-14 22:41:29 +09:00
agentson
a14f944fcc feat: link decision outcomes to trades via decision_id (issue #92)
Some checks failed
CI / test (pull_request) Has been cancelled
Add decision_id column to trades table, capture log_decision() return
value, and update original BUY decision outcome on SELL execution.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-14 21:36:57 +09:00
56f7405baa Merge pull request 'feat: 컨텍스트 집계 스케줄러 (issue #87)' (#119) from feature/issue-87-context-scheduler into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #119
2026-02-10 04:28:42 +09:00
agentson
e3b1ecc572 feat: context aggregation scheduler (issue #87)
Some checks failed
CI / test (pull_request) Has been cancelled
- Add ContextScheduler with run_if_due() for periodic rollups
- Weekly (Sunday), monthly (last day), quarterly, annual, legacy schedules
- Daily cleanup of expired contexts via ContextStore
- Dedup guard: each task runs at most once per day

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-10 04:26:51 +09:00
8acf72b22c Merge pull request 'feat: DailyScorecard 모델 정의 (issue #90)' (#118) from feature/issue-90-scorecard-model into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #118
2026-02-10 04:26:21 +09:00
agentson
c95102a0bd feat: DailyScorecard model for per-market performance review (issue #90)
Some checks failed
CI / test (pull_request) Has been cancelled
- Add DailyScorecard dataclass with market-scoped fields
- Fields: date, market, decisions, pnl, win_rate, scenario_match_rate, lessons, cross_market_note
- Export from src/evolution/__init__.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-10 04:25:37 +09:00
0685d62f9c Merge pull request 'feat: EOD 집계 시장 필터 추가 (issue #86)' (#117) from feature/issue-86-eod-market-filter into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #117
2026-02-10 04:24:58 +09:00
agentson
78021d4695 feat: EOD aggregation with market filter (issue #86)
Some checks failed
CI / test (pull_request) Has been cancelled
- Add market parameter to aggregate_daily_from_trades() for per-market L6 aggregation
- Store market-scoped keys (total_pnl_KR, win_rate_US, etc.) in L6/L5/L4 layers
- Hook aggregate_daily_from_trades() into market close detection in run()
- Update tests for market-scoped context keys

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-10 04:23:49 +09:00
3cdd10783b Merge pull request 'feat: L7 실시간 컨텍스트 시장별 기록 (issue #85)' (#116) from feature/issue-85-l7-context-write into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #116
2026-02-10 04:22:57 +09:00
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
21 changed files with 3467 additions and 192 deletions

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

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

@@ -5,6 +5,7 @@ The context tree implements Pillar 2: hierarchical memory management across
""" """
from src.context.layer import ContextLayer from src.context.layer import ContextLayer
from src.context.scheduler import ContextScheduler
from src.context.store import ContextStore from src.context.store import ContextStore
__all__ = ["ContextLayer", "ContextStore"] __all__ = ["ContextLayer", "ContextScheduler", "ContextStore"]

View File

@@ -18,16 +18,33 @@ class ContextAggregator:
self.conn = conn self.conn = conn
self.store = ContextStore(conn) self.store = ContextStore(conn)
def aggregate_daily_from_trades(self, date: str | None = None) -> None: def aggregate_daily_from_trades(
self, date: str | None = None, market: str | None = None
) -> None:
"""Aggregate L6 (daily) context from trades table. """Aggregate L6 (daily) context from trades table.
Args: Args:
date: Date in YYYY-MM-DD format. If None, uses today. date: Date in YYYY-MM-DD format. If None, uses today.
market: Market code filter (e.g., "KR", "US"). If None, aggregates all markets.
""" """
if date is None: if date is None:
date = datetime.now(UTC).date().isoformat() date = datetime.now(UTC).date().isoformat()
# Calculate daily metrics from trades if market is None:
cursor = self.conn.execute(
"""
SELECT DISTINCT market
FROM trades
WHERE DATE(timestamp) = ?
""",
(date,),
)
markets = [row[0] for row in cursor.fetchall() if row[0]]
else:
markets = [market]
for market_code in markets:
# Calculate daily metrics from trades for the market
cursor = self.conn.execute( cursor = self.conn.execute(
""" """
SELECT SELECT
@@ -41,29 +58,43 @@ class ContextAggregator:
SUM(CASE WHEN pnl > 0 THEN 1 ELSE 0 END) as wins, SUM(CASE WHEN pnl > 0 THEN 1 ELSE 0 END) as wins,
SUM(CASE WHEN pnl < 0 THEN 1 ELSE 0 END) as losses SUM(CASE WHEN pnl < 0 THEN 1 ELSE 0 END) as losses
FROM trades FROM trades
WHERE DATE(timestamp) = ? WHERE DATE(timestamp) = ? AND market = ?
""", """,
(date,), (date, market_code),
) )
row = cursor.fetchone() row = cursor.fetchone()
if row and row[0] > 0: # At least one trade if row and row[0] > 0: # At least one trade
trade_count, buys, sells, holds, avg_conf, total_pnl, stocks, wins, losses = row trade_count, buys, sells, holds, avg_conf, total_pnl, stocks, wins, losses = row
# Store daily metrics in L6 key_suffix = f"_{market_code}"
self.store.set_context(ContextLayer.L6_DAILY, date, "trade_count", trade_count)
self.store.set_context(ContextLayer.L6_DAILY, date, "buys", buys) # Store daily metrics in L6 with market suffix
self.store.set_context(ContextLayer.L6_DAILY, date, "sells", sells)
self.store.set_context(ContextLayer.L6_DAILY, date, "holds", holds)
self.store.set_context( self.store.set_context(
ContextLayer.L6_DAILY, date, "avg_confidence", round(avg_conf, 2) ContextLayer.L6_DAILY, date, f"trade_count{key_suffix}", trade_count
)
self.store.set_context(ContextLayer.L6_DAILY, date, f"buys{key_suffix}", buys)
self.store.set_context(ContextLayer.L6_DAILY, date, f"sells{key_suffix}", sells)
self.store.set_context(ContextLayer.L6_DAILY, date, f"holds{key_suffix}", holds)
self.store.set_context(
ContextLayer.L6_DAILY,
date,
f"avg_confidence{key_suffix}",
round(avg_conf, 2),
) )
self.store.set_context( self.store.set_context(
ContextLayer.L6_DAILY, date, "total_pnl", round(total_pnl, 2) ContextLayer.L6_DAILY,
date,
f"total_pnl{key_suffix}",
round(total_pnl, 2),
)
self.store.set_context(
ContextLayer.L6_DAILY, date, f"unique_stocks{key_suffix}", stocks
) )
self.store.set_context(ContextLayer.L6_DAILY, date, "unique_stocks", stocks)
win_rate = round(wins / max(wins + losses, 1) * 100, 2) win_rate = round(wins / max(wins + losses, 1) * 100, 2)
self.store.set_context(ContextLayer.L6_DAILY, date, "win_rate", win_rate) self.store.set_context(
ContextLayer.L6_DAILY, date, f"win_rate{key_suffix}", win_rate
)
def aggregate_weekly_from_daily(self, week: str | None = None) -> None: def aggregate_weekly_from_daily(self, week: str | None = None) -> None:
"""Aggregate L5 (weekly) context from L6 (daily). """Aggregate L5 (weekly) context from L6 (daily).
@@ -92,14 +123,25 @@ class ContextAggregator:
daily_data[row[0]].append(json.loads(row[1])) daily_data[row[0]].append(json.loads(row[1]))
if daily_data: if daily_data:
# Sum all PnL values # Sum all PnL values (market-specific if suffixed)
if "total_pnl" in daily_data: if "total_pnl" in daily_data:
total_pnl = sum(daily_data["total_pnl"]) total_pnl = sum(daily_data["total_pnl"])
self.store.set_context( self.store.set_context(
ContextLayer.L5_WEEKLY, week, "weekly_pnl", round(total_pnl, 2) ContextLayer.L5_WEEKLY, week, "weekly_pnl", round(total_pnl, 2)
) )
# Average all confidence values for key, values in daily_data.items():
if key.startswith("total_pnl_"):
market_code = key.split("total_pnl_", 1)[1]
total_pnl = sum(values)
self.store.set_context(
ContextLayer.L5_WEEKLY,
week,
f"weekly_pnl_{market_code}",
round(total_pnl, 2),
)
# Average all confidence values (market-specific if suffixed)
if "avg_confidence" in daily_data: if "avg_confidence" in daily_data:
conf_values = daily_data["avg_confidence"] conf_values = daily_data["avg_confidence"]
avg_conf = sum(conf_values) / len(conf_values) avg_conf = sum(conf_values) / len(conf_values)
@@ -107,6 +149,17 @@ class ContextAggregator:
ContextLayer.L5_WEEKLY, week, "avg_confidence", round(avg_conf, 2) ContextLayer.L5_WEEKLY, week, "avg_confidence", round(avg_conf, 2)
) )
for key, values in daily_data.items():
if key.startswith("avg_confidence_"):
market_code = key.split("avg_confidence_", 1)[1]
avg_conf = sum(values) / len(values)
self.store.set_context(
ContextLayer.L5_WEEKLY,
week,
f"avg_confidence_{market_code}",
round(avg_conf, 2),
)
def aggregate_monthly_from_weekly(self, month: str | None = None) -> None: def aggregate_monthly_from_weekly(self, month: str | None = None) -> None:
"""Aggregate L4 (monthly) context from L5 (weekly). """Aggregate L4 (monthly) context from L5 (weekly).
@@ -135,8 +188,16 @@ class ContextAggregator:
if weekly_data: if weekly_data:
# Sum all weekly PnL values # Sum all weekly PnL values
total_pnl_values: list[float] = []
if "weekly_pnl" in weekly_data: if "weekly_pnl" in weekly_data:
total_pnl = sum(weekly_data["weekly_pnl"]) total_pnl_values.extend(weekly_data["weekly_pnl"])
for key, values in weekly_data.items():
if key.startswith("weekly_pnl_"):
total_pnl_values.extend(values)
if total_pnl_values:
total_pnl = sum(total_pnl_values)
self.store.set_context( self.store.set_context(
ContextLayer.L4_MONTHLY, month, "monthly_pnl", round(total_pnl, 2) ContextLayer.L4_MONTHLY, month, "monthly_pnl", round(total_pnl, 2)
) )
@@ -230,21 +291,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()

135
src/context/scheduler.py Normal file
View File

@@ -0,0 +1,135 @@
"""Context aggregation scheduler for periodic rollups and cleanup."""
from __future__ import annotations
import sqlite3
from calendar import monthrange
from dataclasses import dataclass
from datetime import UTC, datetime
from src.context.aggregator import ContextAggregator
from src.context.store import ContextStore
@dataclass(frozen=True)
class ScheduleResult:
"""Represents which scheduled tasks ran."""
weekly: bool = False
monthly: bool = False
quarterly: bool = False
annual: bool = False
legacy: bool = False
cleanup: bool = False
class ContextScheduler:
"""Run periodic context aggregations and cleanup when due."""
def __init__(
self,
conn: sqlite3.Connection | None = None,
aggregator: ContextAggregator | None = None,
store: ContextStore | None = None,
) -> None:
if aggregator is None:
if conn is None:
raise ValueError("conn is required when aggregator is not provided")
aggregator = ContextAggregator(conn)
self.aggregator = aggregator
if store is None:
store = getattr(aggregator, "store", None)
if store is None:
if conn is None:
raise ValueError("conn is required when store is not provided")
store = ContextStore(conn)
self.store = store
self._last_run: dict[str, str] = {}
def run_if_due(self, now: datetime | None = None) -> ScheduleResult:
"""Run scheduled aggregations if their schedule is due.
Args:
now: Current datetime (UTC). If None, uses current time.
Returns:
ScheduleResult indicating which tasks ran.
"""
if now is None:
now = datetime.now(UTC)
today = now.date().isoformat()
result = ScheduleResult()
if self._should_run("cleanup", today):
self.store.cleanup_expired_contexts()
result = self._with(result, cleanup=True)
if self._is_sunday(now) and self._should_run("weekly", today):
week = now.strftime("%Y-W%V")
self.aggregator.aggregate_weekly_from_daily(week)
result = self._with(result, weekly=True)
if self._is_last_day_of_month(now) and self._should_run("monthly", today):
month = now.strftime("%Y-%m")
self.aggregator.aggregate_monthly_from_weekly(month)
result = self._with(result, monthly=True)
if self._is_last_day_of_quarter(now) and self._should_run("quarterly", today):
quarter = self._current_quarter(now)
self.aggregator.aggregate_quarterly_from_monthly(quarter)
result = self._with(result, quarterly=True)
if self._is_last_day_of_year(now) and self._should_run("annual", today):
year = str(now.year)
self.aggregator.aggregate_annual_from_quarterly(year)
result = self._with(result, annual=True)
# Legacy rollup runs after annual aggregation.
self.aggregator.aggregate_legacy_from_annual()
result = self._with(result, legacy=True)
return result
def _should_run(self, key: str, date_str: str) -> bool:
if self._last_run.get(key) == date_str:
return False
self._last_run[key] = date_str
return True
@staticmethod
def _is_sunday(now: datetime) -> bool:
return now.weekday() == 6
@staticmethod
def _is_last_day_of_month(now: datetime) -> bool:
last_day = monthrange(now.year, now.month)[1]
return now.day == last_day
@classmethod
def _is_last_day_of_quarter(cls, now: datetime) -> bool:
if now.month not in (3, 6, 9, 12):
return False
return cls._is_last_day_of_month(now)
@staticmethod
def _is_last_day_of_year(now: datetime) -> bool:
return now.month == 12 and now.day == 31
@staticmethod
def _current_quarter(now: datetime) -> str:
quarter = (now.month - 1) // 3 + 1
return f"{now.year}-Q{quarter}"
@staticmethod
def _with(result: ScheduleResult, **kwargs: bool) -> ScheduleResult:
return ScheduleResult(
weekly=kwargs.get("weekly", result.weekly),
monthly=kwargs.get("monthly", result.monthly),
quarterly=kwargs.get("quarterly", result.quarterly),
annual=kwargs.get("annual", result.annual),
legacy=kwargs.get("legacy", result.legacy),
cleanup=kwargs.get("cleanup", result.cleanup),
)

View File

@@ -6,6 +6,7 @@ import json
import sqlite3 import sqlite3
from datetime import UTC, datetime from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from typing import Any
def init_db(db_path: str) -> sqlite3.Connection: def init_db(db_path: str) -> sqlite3.Connection:
@@ -26,7 +27,8 @@ def init_db(db_path: str) -> sqlite3.Connection:
price REAL, price REAL,
pnl REAL DEFAULT 0.0, pnl REAL DEFAULT 0.0,
market TEXT DEFAULT 'KR', market TEXT DEFAULT 'KR',
exchange_code TEXT DEFAULT 'KRX' exchange_code TEXT DEFAULT 'KRX',
decision_id TEXT
) )
""" """
) )
@@ -41,6 +43,8 @@ def init_db(db_path: str) -> sqlite3.Connection:
conn.execute("ALTER TABLE trades ADD COLUMN exchange_code TEXT DEFAULT 'KRX'") conn.execute("ALTER TABLE trades ADD COLUMN exchange_code TEXT DEFAULT 'KRX'")
if "selection_context" not in columns: if "selection_context" not in columns:
conn.execute("ALTER TABLE trades ADD COLUMN selection_context TEXT") conn.execute("ALTER TABLE trades ADD COLUMN selection_context TEXT")
if "decision_id" not in columns:
conn.execute("ALTER TABLE trades ADD COLUMN decision_id TEXT")
# Context tree tables for multi-layered memory management # Context tree tables for multi-layered memory management
conn.execute( conn.execute(
@@ -143,6 +147,7 @@ def log_trade(
market: str = "KR", market: str = "KR",
exchange_code: str = "KRX", exchange_code: str = "KRX",
selection_context: dict[str, any] | None = None, selection_context: dict[str, any] | None = None,
decision_id: str | None = None,
) -> None: ) -> None:
"""Insert a trade record into the database. """Insert a trade record into the database.
@@ -166,9 +171,9 @@ def log_trade(
""" """
INSERT INTO trades ( INSERT INTO trades (
timestamp, stock_code, action, confidence, rationale, timestamp, stock_code, action, confidence, rationale,
quantity, price, pnl, market, exchange_code, selection_context quantity, price, pnl, market, exchange_code, selection_context, decision_id
) )
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", """,
( (
datetime.now(UTC).isoformat(), datetime.now(UTC).isoformat(),
@@ -182,6 +187,30 @@ def log_trade(
market, market,
exchange_code, exchange_code,
context_json, context_json,
decision_id,
), ),
) )
conn.commit() conn.commit()
def get_latest_buy_trade(
conn: sqlite3.Connection, stock_code: str, market: str
) -> dict[str, Any] | None:
"""Fetch the most recent BUY trade for a stock and market."""
cursor = conn.execute(
"""
SELECT decision_id, price, quantity
FROM trades
WHERE stock_code = ?
AND market = ?
AND action = 'BUY'
AND decision_id IS NOT NULL
ORDER BY timestamp DESC
LIMIT 1
""",
(stock_code, market),
)
row = cursor.fetchone()
if not row:
return None
return {"decision_id": row[0], "price": row[1], "quantity": row[2]}

View File

@@ -1,12 +1,14 @@
"""Evolution engine for self-improving trading strategies.""" """Evolution engine for self-improving trading strategies."""
from src.evolution.ab_test import ABTester, ABTestResult, StrategyPerformance from src.evolution.ab_test import ABTester, ABTestResult, StrategyPerformance
from src.evolution.daily_review import DailyReviewer
from src.evolution.optimizer import EvolutionOptimizer from src.evolution.optimizer import EvolutionOptimizer
from src.evolution.performance_tracker import ( from src.evolution.performance_tracker import (
PerformanceDashboard, PerformanceDashboard,
PerformanceTracker, PerformanceTracker,
StrategyMetrics, StrategyMetrics,
) )
from src.evolution.scorecard import DailyScorecard
__all__ = [ __all__ = [
"EvolutionOptimizer", "EvolutionOptimizer",
@@ -16,4 +18,6 @@ __all__ = [
"PerformanceTracker", "PerformanceTracker",
"PerformanceDashboard", "PerformanceDashboard",
"StrategyMetrics", "StrategyMetrics",
"DailyScorecard",
"DailyReviewer",
] ]

View File

@@ -0,0 +1,196 @@
"""Daily review generator for market-scoped end-of-day scorecards."""
from __future__ import annotations
import json
import logging
import re
import sqlite3
from dataclasses import asdict
from src.brain.gemini_client import GeminiClient
from src.context.layer import ContextLayer
from src.context.store import ContextStore
from src.evolution.scorecard import DailyScorecard
logger = logging.getLogger(__name__)
class DailyReviewer:
"""Builds daily scorecards and optional AI-generated lessons."""
def __init__(
self,
conn: sqlite3.Connection,
context_store: ContextStore,
gemini_client: GeminiClient | None = None,
) -> None:
self._conn = conn
self._context_store = context_store
self._gemini = gemini_client
def generate_scorecard(self, date: str, market: str) -> DailyScorecard:
"""Generate a market-scoped scorecard from decision logs and trades."""
decision_rows = self._conn.execute(
"""
SELECT action, confidence, context_snapshot
FROM decision_logs
WHERE DATE(timestamp) = ? AND market = ?
""",
(date, market),
).fetchall()
total_decisions = len(decision_rows)
buys = sum(1 for row in decision_rows if row[0] == "BUY")
sells = sum(1 for row in decision_rows if row[0] == "SELL")
holds = sum(1 for row in decision_rows if row[0] == "HOLD")
avg_confidence = (
round(sum(int(row[1]) for row in decision_rows) / total_decisions, 2)
if total_decisions > 0
else 0.0
)
matched = 0
for row in decision_rows:
try:
snapshot = json.loads(row[2]) if row[2] else {}
except json.JSONDecodeError:
snapshot = {}
scenario_match = snapshot.get("scenario_match", {})
if isinstance(scenario_match, dict) and scenario_match:
matched += 1
scenario_match_rate = (
round((matched / total_decisions) * 100, 2)
if total_decisions
else 0.0
)
trade_stats = self._conn.execute(
"""
SELECT
COALESCE(SUM(pnl), 0.0),
SUM(CASE WHEN pnl > 0 THEN 1 ELSE 0 END),
SUM(CASE WHEN pnl < 0 THEN 1 ELSE 0 END)
FROM trades
WHERE DATE(timestamp) = ? AND market = ?
""",
(date, market),
).fetchone()
total_pnl = round(float(trade_stats[0] or 0.0), 2) if trade_stats else 0.0
wins = int(trade_stats[1] or 0) if trade_stats else 0
losses = int(trade_stats[2] or 0) if trade_stats else 0
win_rate = round((wins / (wins + losses)) * 100, 2) if (wins + losses) > 0 else 0.0
top_winners = [
row[0]
for row in self._conn.execute(
"""
SELECT stock_code, SUM(pnl) AS stock_pnl
FROM trades
WHERE DATE(timestamp) = ? AND market = ?
GROUP BY stock_code
HAVING stock_pnl > 0
ORDER BY stock_pnl DESC
LIMIT 3
""",
(date, market),
).fetchall()
]
top_losers = [
row[0]
for row in self._conn.execute(
"""
SELECT stock_code, SUM(pnl) AS stock_pnl
FROM trades
WHERE DATE(timestamp) = ? AND market = ?
GROUP BY stock_code
HAVING stock_pnl < 0
ORDER BY stock_pnl ASC
LIMIT 3
""",
(date, market),
).fetchall()
]
return DailyScorecard(
date=date,
market=market,
total_decisions=total_decisions,
buys=buys,
sells=sells,
holds=holds,
total_pnl=total_pnl,
win_rate=win_rate,
avg_confidence=avg_confidence,
scenario_match_rate=scenario_match_rate,
top_winners=top_winners,
top_losers=top_losers,
lessons=[],
cross_market_note="",
)
async def generate_lessons(self, scorecard: DailyScorecard) -> list[str]:
"""Generate concise lessons from scorecard metrics using Gemini."""
if self._gemini is None:
return []
prompt = (
"You are a trading performance reviewer.\n"
"Return ONLY a JSON array of 1-3 short lessons in English.\n"
f"Market: {scorecard.market}\n"
f"Date: {scorecard.date}\n"
f"Total decisions: {scorecard.total_decisions}\n"
f"Buys/Sells/Holds: {scorecard.buys}/{scorecard.sells}/{scorecard.holds}\n"
f"Total PnL: {scorecard.total_pnl}\n"
f"Win rate: {scorecard.win_rate}%\n"
f"Average confidence: {scorecard.avg_confidence}\n"
f"Scenario match rate: {scorecard.scenario_match_rate}%\n"
f"Top winners: {', '.join(scorecard.top_winners) or 'N/A'}\n"
f"Top losers: {', '.join(scorecard.top_losers) or 'N/A'}\n"
)
try:
decision = await self._gemini.decide(
{
"stock_code": "REVIEW",
"market_name": scorecard.market,
"current_price": 0,
"prompt_override": prompt,
}
)
return self._parse_lessons(decision.rationale)
except Exception as exc:
logger.warning("Failed to generate daily lessons: %s", exc)
return []
def store_scorecard_in_context(self, scorecard: DailyScorecard) -> None:
"""Store scorecard in L6 using market-scoped key."""
self._context_store.set_context(
ContextLayer.L6_DAILY,
scorecard.date,
f"scorecard_{scorecard.market}",
asdict(scorecard),
)
def _parse_lessons(self, raw_text: str) -> list[str]:
"""Parse lessons from JSON array response or fallback text."""
raw_text = raw_text.strip()
try:
parsed = json.loads(raw_text)
if isinstance(parsed, list):
return [str(item).strip() for item in parsed if str(item).strip()][:3]
except json.JSONDecodeError:
pass
match = re.search(r"\[.*\]", raw_text, re.DOTALL)
if match:
try:
parsed = json.loads(match.group(0))
if isinstance(parsed, list):
return [str(item).strip() for item in parsed if str(item).strip()][:3]
except json.JSONDecodeError:
pass
lines = [line.strip("-* \t") for line in raw_text.splitlines() if line.strip()]
return lines[:3]

View File

@@ -0,0 +1,25 @@
"""Daily scorecard model for end-of-day performance review."""
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass
class DailyScorecard:
"""Structured daily performance snapshot for a single market."""
date: str
market: str
total_decisions: int
buys: int
sells: int
holds: int
total_pnl: float
win_rate: float
avg_confidence: float
scenario_match_rate: float
top_winners: list[str] = field(default_factory=list)
top_losers: list[str] = field(default_factory=list)
lessons: list[str] = field(default_factory=list)
cross_market_note: str = ""

View File

@@ -10,27 +10,32 @@ 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
from src.context.aggregator import ContextAggregator
from src.context.layer import ContextLayer from src.context.layer import ContextLayer
from src.context.store import ContextStore from src.context.store import ContextStore
from src.core.criticality import CriticalityAssessor from src.core.criticality import CriticalityAssessor
from src.core.priority_queue import PriorityTaskQueue from src.core.priority_queue import PriorityTaskQueue
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager
from src.db import init_db, log_trade from src.db import get_latest_buy_trade, init_db, log_trade
from src.evolution.daily_review import DailyReviewer
from src.logging.decision_logger import DecisionLogger 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__)
@@ -75,7 +80,8 @@ TRADE_SESSION_INTERVAL_HOURS = 6 # Hours between sessions
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,
@@ -84,7 +90,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()
@@ -135,13 +141,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
@@ -152,7 +204,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)
@@ -178,8 +230,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,
@@ -188,6 +245,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": {
@@ -200,7 +270,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,
@@ -210,7 +280,7 @@ async def trading_cycle(
"pnl_pct": pnl_pct, "pnl_pct": pnl_pct,
} }
decision_logger.log_decision( decision_id = decision_logger.log_decision(
stock_code=stock_code, stock_code=stock_code,
market=market.code, market=market.code,
exchange_code=market.exchange_code, exchange_code=market.exchange_code,
@@ -222,6 +292,9 @@ async def trading_cycle(
) )
# 3. Execute if actionable # 3. Execute if actionable
quantity = 0
trade_price = current_price
trade_pnl = 0.0
if decision.action in ("BUY", "SELL"): if decision.action in ("BUY", "SELL"):
# Determine order size (simplified: 1 lot) # Determine order size (simplified: 1 lot)
quantity = 1 quantity = 1
@@ -277,10 +350,22 @@ async def trading_cycle(
except Exception as exc: except Exception as exc:
logger.warning("Telegram notification failed: %s", exc) logger.warning("Telegram notification failed: %s", exc)
if decision.action == "SELL":
buy_trade = get_latest_buy_trade(db_conn, stock_code, market.code)
if buy_trade and buy_trade.get("price") is not None:
buy_price = float(buy_trade["price"])
buy_qty = int(buy_trade.get("quantity") or 1)
trade_pnl = (trade_price - buy_price) * buy_qty
decision_logger.update_outcome(
decision_id=buy_trade["decision_id"],
pnl=trade_pnl,
accuracy=1 if trade_pnl > 0 else 0,
)
# 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,
@@ -294,9 +379,13 @@ async def trading_cycle(
action=decision.action, action=decision.action,
confidence=decision.confidence, confidence=decision.confidence,
rationale=decision.rationale, rationale=decision.rationale,
quantity=quantity,
price=trade_price,
pnl=trade_pnl,
market=market.code, market=market.code,
exchange_code=market.exchange_code, exchange_code=market.exchange_code,
selection_context=selection_context, selection_context=selection_context,
decision_id=decision_id,
) )
# 7. Latency monitoring # 7. Latency monitoring
@@ -324,7 +413,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,
@@ -336,10 +427,8 @@ async def run_daily_session(
) -> 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)
@@ -352,27 +441,66 @@ async def run_daily_session(
# Process each open market # Process each open market
for market in open_markets: for market in open_markets:
# Use market-local date for playbook keying
market_today = datetime.now(market.timezone).date()
# Dynamic stock discovery via scanner (no static watchlists) # Dynamic stock discovery via scanner (no static watchlists)
candidates_list: list[ScanCandidate] = []
try: try:
candidates = await smart_scanner.scan() candidates_list = await smart_scanner.scan() if smart_scanner else []
watchlist = [c.stock_code for c in candidates] if candidates else []
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)
watchlist = []
if not watchlist: if not candidates_list:
logger.info("No scanner candidates for market %s — skipping", market.code) 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))
# 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 # 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")
) )
@@ -380,17 +508,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
@@ -399,17 +533,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", [{}])
@@ -422,21 +558,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)",
@@ -458,6 +610,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"],
@@ -467,7 +620,7 @@ async def run_daily_session(
"pnl_pct": pnl_pct, "pnl_pct": pnl_pct,
} }
decision_logger.log_decision( decision_id = decision_logger.log_decision(
stock_code=stock_code, stock_code=stock_code,
market=market.code, market=market.code,
exchange_code=market.exchange_code, exchange_code=market.exchange_code,
@@ -479,6 +632,9 @@ async def run_daily_session(
) )
# Execute if actionable # Execute if actionable
quantity = 0
trade_price = stock_data["current_price"]
trade_pnl = 0.0
if decision.action in ("BUY", "SELL"): if decision.action in ("BUY", "SELL"):
quantity = 1 quantity = 1
order_amount = stock_data["current_price"] * quantity order_amount = stock_data["current_price"] * quantity
@@ -509,7 +665,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
@@ -544,9 +702,23 @@ 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
if decision.action == "SELL":
buy_trade = get_latest_buy_trade(db_conn, stock_code, market.code)
if buy_trade and buy_trade.get("price") is not None:
buy_price = float(buy_trade["price"])
buy_qty = int(buy_trade.get("quantity") or 1)
trade_pnl = (trade_price - buy_price) * buy_qty
decision_logger.update_outcome(
decision_id=buy_trade["decision_id"],
pnl=trade_pnl,
accuracy=1 if trade_pnl > 0 else 0,
)
# Log trade # Log trade
log_trade( log_trade(
conn=db_conn, conn=db_conn,
@@ -554,13 +726,52 @@ async def run_daily_session(
action=decision.action, action=decision.action,
confidence=decision.confidence, confidence=decision.confidence,
rationale=decision.rationale, rationale=decision.rationale,
quantity=quantity,
price=trade_price,
pnl=trade_pnl,
market=market.code, market=market.code,
exchange_code=market.exchange_code, exchange_code=market.exchange_code,
decision_id=decision_id,
) )
logger.info("Daily trading session completed") logger.info("Daily trading session completed")
async def _handle_market_close(
market_code: str,
market_name: str,
market_timezone: Any,
telegram: TelegramClient,
context_aggregator: ContextAggregator,
daily_reviewer: DailyReviewer,
) -> None:
"""Handle market-close tasks: notify, aggregate, review, and store context."""
await telegram.notify_market_close(market_name, 0.0)
market_date = datetime.now(market_timezone).date().isoformat()
context_aggregator.aggregate_daily_from_trades(
date=market_date,
market=market_code,
)
scorecard = daily_reviewer.generate_scorecard(market_date, market_code)
daily_reviewer.store_scorecard_in_context(scorecard)
lessons = await daily_reviewer.generate_lessons(scorecard)
if lessons:
scorecard.lessons = lessons
daily_reviewer.store_scorecard_in_context(scorecard)
await telegram.send_message(
f"<b>Daily Review ({market_code})</b>\n"
f"Date: {scorecard.date}\n"
f"Decisions: {scorecard.total_decisions}\n"
f"P&L: {scorecard.total_pnl:+.2f}\n"
f"Win Rate: {scorecard.win_rate:.2f}%\n"
f"Lessons: {', '.join(scorecard.lessons) if scorecard.lessons else 'N/A'}"
)
async def run(settings: Settings) -> None: async def run(settings: Settings) -> None:
"""Main async loop — iterate over open markets on a timer.""" """Main async loop — iterate over open markets on a timer."""
broker = KISBroker(settings) broker = KISBroker(settings)
@@ -570,6 +781,22 @@ async def run(settings: Settings) -> None:
db_conn = init_db(settings.DB_PATH) db_conn = init_db(settings.DB_PATH)
decision_logger = DecisionLogger(db_conn) decision_logger = DecisionLogger(db_conn)
context_store = ContextStore(db_conn) context_store = ContextStore(db_conn)
context_aggregator = ContextAggregator(db_conn)
# V2 proactive strategy components
context_selector = ContextSelector(context_store)
scenario_engine = ScenarioEngine()
playbook_store = PlaybookStore(db_conn)
daily_reviewer = DailyReviewer(db_conn, context_store, gemini_client=brain)
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(
@@ -716,15 +943,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,
@@ -732,8 +950,8 @@ 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 per market (dynamically discovered by scanner)
active_stocks: dict[str, list[str]] = {} # market_code -> [stock_codes] active_stocks: dict[str, list[str]] = {} # market_code -> [stock_codes]
@@ -802,7 +1020,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,
@@ -846,10 +1066,19 @@ async def run(settings: Settings) -> None:
market_info = MARKETS.get(market_code) market_info = MARKETS.get(market_code)
if market_info: if market_info:
await telegram.notify_market_close(market_info.name, 0.0) await _handle_market_close(
market_code=market_code,
market_name=market_info.name,
market_timezone=market_info.timezone,
telegram=telegram,
context_aggregator=context_aggregator,
daily_reviewer=daily_reviewer,
)
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:
@@ -887,7 +1116,8 @@ async def run(settings: Settings) -> None:
# Smart Scanner: dynamic stock discovery (no static watchlists) # 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)
@@ -899,9 +1129,10 @@ async def run(settings: Settings) -> None:
candidates candidates
) )
# Store candidates for selection context logging # Store candidates per market for selection context logging
for candidate in candidates: scan_candidates[market.code] = {
scan_candidates[candidate.stock_code] = candidate c.stock_code: c for c in candidates
}
logger.info( logger.info(
"Smart Scanner: Found %d candidates for %s: %s", "Smart Scanner: Found %d candidates for %s: %s",
@@ -909,6 +1140,62 @@ async def run(settings: Settings) -> None:
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( logger.info(
"Smart Scanner: No candidates for %s — no trades", market.name "Smart Scanner: No candidates for %s — no trades", market.name
@@ -933,13 +1220,22 @@ async def run(settings: Settings) -> None:
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,
@@ -988,7 +1284,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.

View File

@@ -0,0 +1,420 @@
"""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, timedelta
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_date = today - timedelta(days=1) if target_market == "KR" else today
timeframe = timeframe_date.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

@@ -161,7 +161,7 @@ class TestContextAggregator:
self, aggregator: ContextAggregator, db_conn: sqlite3.Connection self, aggregator: ContextAggregator, db_conn: sqlite3.Connection
) -> None: ) -> None:
"""Test aggregating daily metrics from trades.""" """Test aggregating daily metrics from trades."""
date = "2026-02-04" date = datetime.now(UTC).date().isoformat()
# Create sample trades # Create sample trades
log_trade(db_conn, "005930", "BUY", 85, "Good signal", quantity=10, price=70000, pnl=500) log_trade(db_conn, "005930", "BUY", 85, "Good signal", quantity=10, price=70000, pnl=500)
@@ -175,36 +175,44 @@ class TestContextAggregator:
db_conn.commit() db_conn.commit()
# Aggregate # Aggregate
aggregator.aggregate_daily_from_trades(date) aggregator.aggregate_daily_from_trades(date, market="KR")
# Verify L6 contexts # Verify L6 contexts
store = aggregator.store store = aggregator.store
assert store.get_context(ContextLayer.L6_DAILY, date, "trade_count") == 3 assert store.get_context(ContextLayer.L6_DAILY, date, "trade_count_KR") == 3
assert store.get_context(ContextLayer.L6_DAILY, date, "buys") == 1 assert store.get_context(ContextLayer.L6_DAILY, date, "buys_KR") == 1
assert store.get_context(ContextLayer.L6_DAILY, date, "sells") == 1 assert store.get_context(ContextLayer.L6_DAILY, date, "sells_KR") == 1
assert store.get_context(ContextLayer.L6_DAILY, date, "holds") == 1 assert store.get_context(ContextLayer.L6_DAILY, date, "holds_KR") == 1
assert store.get_context(ContextLayer.L6_DAILY, date, "total_pnl") == 2000.0 assert store.get_context(ContextLayer.L6_DAILY, date, "total_pnl_KR") == 2000.0
assert store.get_context(ContextLayer.L6_DAILY, date, "unique_stocks") == 3 assert store.get_context(ContextLayer.L6_DAILY, date, "unique_stocks_KR") == 3
# 2 wins, 0 losses # 2 wins, 0 losses
assert store.get_context(ContextLayer.L6_DAILY, date, "win_rate") == 100.0 assert store.get_context(ContextLayer.L6_DAILY, date, "win_rate_KR") == 100.0
def test_aggregate_weekly_from_daily(self, aggregator: ContextAggregator) -> None: def test_aggregate_weekly_from_daily(self, aggregator: ContextAggregator) -> None:
"""Test aggregating weekly metrics from daily.""" """Test aggregating weekly metrics from daily."""
week = "2026-W06" week = "2026-W06"
# Set daily contexts # Set daily contexts
aggregator.store.set_context(ContextLayer.L6_DAILY, "2026-02-02", "total_pnl", 100.0) aggregator.store.set_context(
aggregator.store.set_context(ContextLayer.L6_DAILY, "2026-02-03", "total_pnl", 200.0) ContextLayer.L6_DAILY, "2026-02-02", "total_pnl_KR", 100.0
aggregator.store.set_context(ContextLayer.L6_DAILY, "2026-02-02", "avg_confidence", 80.0) )
aggregator.store.set_context(ContextLayer.L6_DAILY, "2026-02-03", "avg_confidence", 85.0) aggregator.store.set_context(
ContextLayer.L6_DAILY, "2026-02-03", "total_pnl_KR", 200.0
)
aggregator.store.set_context(
ContextLayer.L6_DAILY, "2026-02-02", "avg_confidence_KR", 80.0
)
aggregator.store.set_context(
ContextLayer.L6_DAILY, "2026-02-03", "avg_confidence_KR", 85.0
)
# Aggregate # Aggregate
aggregator.aggregate_weekly_from_daily(week) aggregator.aggregate_weekly_from_daily(week)
# Verify L5 contexts # Verify L5 contexts
store = aggregator.store store = aggregator.store
weekly_pnl = store.get_context(ContextLayer.L5_WEEKLY, week, "weekly_pnl") weekly_pnl = store.get_context(ContextLayer.L5_WEEKLY, week, "weekly_pnl_KR")
avg_conf = store.get_context(ContextLayer.L5_WEEKLY, week, "avg_confidence") avg_conf = store.get_context(ContextLayer.L5_WEEKLY, week, "avg_confidence_KR")
assert weekly_pnl == 300.0 assert weekly_pnl == 300.0
assert avg_conf == 82.5 assert avg_conf == 82.5
@@ -214,9 +222,15 @@ class TestContextAggregator:
month = "2026-02" month = "2026-02"
# Set weekly contexts # Set weekly contexts
aggregator.store.set_context(ContextLayer.L5_WEEKLY, "2026-W05", "weekly_pnl", 100.0) aggregator.store.set_context(
aggregator.store.set_context(ContextLayer.L5_WEEKLY, "2026-W06", "weekly_pnl", 200.0) ContextLayer.L5_WEEKLY, "2026-W05", "weekly_pnl_KR", 100.0
aggregator.store.set_context(ContextLayer.L5_WEEKLY, "2026-W07", "weekly_pnl", 150.0) )
aggregator.store.set_context(
ContextLayer.L5_WEEKLY, "2026-W06", "weekly_pnl_KR", 200.0
)
aggregator.store.set_context(
ContextLayer.L5_WEEKLY, "2026-W07", "weekly_pnl_KR", 150.0
)
# Aggregate # Aggregate
aggregator.aggregate_monthly_from_weekly(month) aggregator.aggregate_monthly_from_weekly(month)
@@ -285,7 +299,7 @@ class TestContextAggregator:
self, aggregator: ContextAggregator, db_conn: sqlite3.Connection self, aggregator: ContextAggregator, db_conn: sqlite3.Connection
) -> None: ) -> None:
"""Test running all aggregations from L7 to L1.""" """Test running all aggregations from L7 to L1."""
date = "2026-02-04" date = datetime.now(UTC).date().isoformat()
# Create sample trades # Create sample trades
log_trade(db_conn, "005930", "BUY", 85, "Good signal", quantity=10, price=70000, pnl=1000) log_trade(db_conn, "005930", "BUY", 85, "Good signal", quantity=10, price=70000, pnl=1000)
@@ -299,10 +313,18 @@ 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_KR") == 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_KR") 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

@@ -0,0 +1,104 @@
"""Tests for ContextScheduler."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
from src.context.scheduler import ContextScheduler
@dataclass
class StubAggregator:
"""Stub aggregator that records calls."""
weekly_calls: list[str]
monthly_calls: list[str]
quarterly_calls: list[str]
annual_calls: list[str]
legacy_calls: int
def aggregate_weekly_from_daily(self, week: str) -> None:
self.weekly_calls.append(week)
def aggregate_monthly_from_weekly(self, month: str) -> None:
self.monthly_calls.append(month)
def aggregate_quarterly_from_monthly(self, quarter: str) -> None:
self.quarterly_calls.append(quarter)
def aggregate_annual_from_quarterly(self, year: str) -> None:
self.annual_calls.append(year)
def aggregate_legacy_from_annual(self) -> None:
self.legacy_calls += 1
@dataclass
class StubStore:
"""Stub store that records cleanup calls."""
cleanup_calls: int = 0
def cleanup_expired_contexts(self) -> None:
self.cleanup_calls += 1
def make_scheduler() -> tuple[ContextScheduler, StubAggregator, StubStore]:
aggregator = StubAggregator([], [], [], [], 0)
store = StubStore()
scheduler = ContextScheduler(aggregator=aggregator, store=store)
return scheduler, aggregator, store
def test_run_if_due_weekly() -> None:
scheduler, aggregator, store = make_scheduler()
now = datetime(2026, 2, 8, 10, 0, tzinfo=UTC) # Sunday
result = scheduler.run_if_due(now)
assert result.weekly is True
assert aggregator.weekly_calls == ["2026-W06"]
assert store.cleanup_calls == 1
def test_run_if_due_monthly() -> None:
scheduler, aggregator, _store = make_scheduler()
now = datetime(2026, 2, 28, 12, 0, tzinfo=UTC) # Last day of month
result = scheduler.run_if_due(now)
assert result.monthly is True
assert aggregator.monthly_calls == ["2026-02"]
def test_run_if_due_quarterly() -> None:
scheduler, aggregator, _store = make_scheduler()
now = datetime(2026, 3, 31, 12, 0, tzinfo=UTC) # Last day of Q1
result = scheduler.run_if_due(now)
assert result.quarterly is True
assert aggregator.quarterly_calls == ["2026-Q1"]
def test_run_if_due_annual_and_legacy() -> None:
scheduler, aggregator, _store = make_scheduler()
now = datetime(2026, 12, 31, 12, 0, tzinfo=UTC)
result = scheduler.run_if_due(now)
assert result.annual is True
assert result.legacy is True
assert aggregator.annual_calls == ["2026"]
assert aggregator.legacy_calls == 1
def test_cleanup_runs_once_per_day() -> None:
scheduler, _aggregator, store = make_scheduler()
now = datetime(2026, 2, 9, 9, 0, tzinfo=UTC)
scheduler.run_if_due(now)
scheduler.run_if_due(now)
assert store.cleanup_calls == 1

383
tests/test_daily_review.py Normal file
View File

@@ -0,0 +1,383 @@
"""Tests for DailyReviewer."""
from __future__ import annotations
import json
import sqlite3
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from src.context.layer import ContextLayer
from src.context.store import ContextStore
from src.db import init_db, log_trade
from src.evolution.daily_review import DailyReviewer
from src.evolution.scorecard import DailyScorecard
from src.logging.decision_logger import DecisionLogger
@pytest.fixture
def db_conn() -> sqlite3.Connection:
return init_db(":memory:")
@pytest.fixture
def context_store(db_conn: sqlite3.Connection) -> ContextStore:
return ContextStore(db_conn)
def _log_decision(
logger: DecisionLogger,
*,
stock_code: str,
market: str,
action: str,
confidence: int,
scenario_match: dict[str, float] | None = None,
) -> str:
return logger.log_decision(
stock_code=stock_code,
market=market,
exchange_code="KRX" if market == "KR" else "NASDAQ",
action=action,
confidence=confidence,
rationale="test",
context_snapshot={"scenario_match": scenario_match or {}},
input_data={"stock_code": stock_code},
)
def test_generate_scorecard_market_scoped(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
reviewer = DailyReviewer(db_conn, context_store)
logger = DecisionLogger(db_conn)
buy_id = _log_decision(
logger,
stock_code="005930",
market="KR",
action="BUY",
confidence=90,
scenario_match={"rsi": 29.0},
)
_log_decision(
logger,
stock_code="000660",
market="KR",
action="HOLD",
confidence=60,
)
_log_decision(
logger,
stock_code="AAPL",
market="US",
action="SELL",
confidence=80,
scenario_match={"volume_ratio": 2.1},
)
log_trade(
db_conn,
"005930",
"BUY",
90,
"buy",
quantity=1,
price=100.0,
pnl=10.0,
market="KR",
exchange_code="KRX",
decision_id=buy_id,
)
log_trade(
db_conn,
"000660",
"HOLD",
60,
"hold",
quantity=0,
price=0.0,
pnl=0.0,
market="KR",
exchange_code="KRX",
)
log_trade(
db_conn,
"AAPL",
"SELL",
80,
"sell",
quantity=1,
price=200.0,
pnl=-5.0,
market="US",
exchange_code="NASDAQ",
)
scorecard = reviewer.generate_scorecard("2026-02-14", "KR")
assert scorecard.market == "KR"
assert scorecard.total_decisions == 2
assert scorecard.buys == 1
assert scorecard.sells == 0
assert scorecard.holds == 1
assert scorecard.total_pnl == 10.0
assert scorecard.win_rate == 100.0
assert scorecard.avg_confidence == 75.0
assert scorecard.scenario_match_rate == 50.0
def test_generate_scorecard_top_winners_and_losers(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
reviewer = DailyReviewer(db_conn, context_store)
logger = DecisionLogger(db_conn)
for code, pnl in [("005930", 30.0), ("000660", 10.0), ("035420", -15.0), ("051910", -5.0)]:
decision_id = _log_decision(
logger,
stock_code=code,
market="KR",
action="BUY" if pnl >= 0 else "SELL",
confidence=80,
scenario_match={"rsi": 30.0},
)
log_trade(
db_conn,
code,
"BUY" if pnl >= 0 else "SELL",
80,
"test",
quantity=1,
price=100.0,
pnl=pnl,
market="KR",
exchange_code="KRX",
decision_id=decision_id,
)
scorecard = reviewer.generate_scorecard("2026-02-14", "KR")
assert scorecard.top_winners == ["005930", "000660"]
assert scorecard.top_losers == ["035420", "051910"]
def test_generate_scorecard_empty_day(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
reviewer = DailyReviewer(db_conn, context_store)
scorecard = reviewer.generate_scorecard("2026-02-14", "KR")
assert scorecard.total_decisions == 0
assert scorecard.total_pnl == 0.0
assert scorecard.win_rate == 0.0
assert scorecard.avg_confidence == 0.0
assert scorecard.scenario_match_rate == 0.0
assert scorecard.top_winners == []
assert scorecard.top_losers == []
@pytest.mark.asyncio
async def test_generate_lessons_without_gemini_returns_empty(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
reviewer = DailyReviewer(db_conn, context_store, gemini_client=None)
lessons = await reviewer.generate_lessons(
DailyScorecard(
date="2026-02-14",
market="KR",
total_decisions=1,
buys=1,
sells=0,
holds=0,
total_pnl=5.0,
win_rate=100.0,
avg_confidence=90.0,
scenario_match_rate=100.0,
)
)
assert lessons == []
@pytest.mark.asyncio
async def test_generate_lessons_parses_json_array(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
mock_gemini = MagicMock()
mock_gemini.decide = AsyncMock(
return_value=SimpleNamespace(rationale='["Cut losers earlier", "Reduce midday churn"]')
)
reviewer = DailyReviewer(db_conn, context_store, gemini_client=mock_gemini)
lessons = await reviewer.generate_lessons(
DailyScorecard(
date="2026-02-14",
market="KR",
total_decisions=3,
buys=1,
sells=1,
holds=1,
total_pnl=-2.5,
win_rate=50.0,
avg_confidence=70.0,
scenario_match_rate=66.7,
)
)
assert lessons == ["Cut losers earlier", "Reduce midday churn"]
@pytest.mark.asyncio
async def test_generate_lessons_fallback_to_lines(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
mock_gemini = MagicMock()
mock_gemini.decide = AsyncMock(
return_value=SimpleNamespace(rationale="- Keep risk tighter\n- Increase selectivity")
)
reviewer = DailyReviewer(db_conn, context_store, gemini_client=mock_gemini)
lessons = await reviewer.generate_lessons(
DailyScorecard(
date="2026-02-14",
market="US",
total_decisions=2,
buys=1,
sells=1,
holds=0,
total_pnl=1.0,
win_rate=50.0,
avg_confidence=75.0,
scenario_match_rate=100.0,
)
)
assert lessons == ["Keep risk tighter", "Increase selectivity"]
@pytest.mark.asyncio
async def test_generate_lessons_handles_gemini_error(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
mock_gemini = MagicMock()
mock_gemini.decide = AsyncMock(side_effect=RuntimeError("boom"))
reviewer = DailyReviewer(db_conn, context_store, gemini_client=mock_gemini)
lessons = await reviewer.generate_lessons(
DailyScorecard(
date="2026-02-14",
market="US",
total_decisions=0,
buys=0,
sells=0,
holds=0,
total_pnl=0.0,
win_rate=0.0,
avg_confidence=0.0,
scenario_match_rate=0.0,
)
)
assert lessons == []
def test_store_scorecard_in_context(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
reviewer = DailyReviewer(db_conn, context_store)
scorecard = DailyScorecard(
date="2026-02-14",
market="KR",
total_decisions=5,
buys=2,
sells=1,
holds=2,
total_pnl=15.0,
win_rate=66.67,
avg_confidence=82.0,
scenario_match_rate=80.0,
lessons=["Keep position sizing stable"],
cross_market_note="US risk-off",
)
reviewer.store_scorecard_in_context(scorecard)
stored = context_store.get_context(
ContextLayer.L6_DAILY,
"2026-02-14",
"scorecard_KR",
)
assert stored is not None
assert stored["market"] == "KR"
assert stored["total_pnl"] == 15.0
assert stored["lessons"] == ["Keep position sizing stable"]
def test_store_scorecard_key_is_market_scoped(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
reviewer = DailyReviewer(db_conn, context_store)
kr = DailyScorecard(
date="2026-02-14",
market="KR",
total_decisions=1,
buys=1,
sells=0,
holds=0,
total_pnl=1.0,
win_rate=100.0,
avg_confidence=90.0,
scenario_match_rate=100.0,
)
us = DailyScorecard(
date="2026-02-14",
market="US",
total_decisions=1,
buys=0,
sells=1,
holds=0,
total_pnl=-1.0,
win_rate=0.0,
avg_confidence=70.0,
scenario_match_rate=100.0,
)
reviewer.store_scorecard_in_context(kr)
reviewer.store_scorecard_in_context(us)
kr_ctx = context_store.get_context(ContextLayer.L6_DAILY, "2026-02-14", "scorecard_KR")
us_ctx = context_store.get_context(ContextLayer.L6_DAILY, "2026-02-14", "scorecard_US")
assert kr_ctx["market"] == "KR"
assert us_ctx["market"] == "US"
assert kr_ctx["total_pnl"] == 1.0
assert us_ctx["total_pnl"] == -1.0
def test_generate_scorecard_handles_invalid_context_snapshot(
db_conn: sqlite3.Connection, context_store: ContextStore,
) -> None:
reviewer = DailyReviewer(db_conn, context_store)
db_conn.execute(
"""
INSERT INTO decision_logs (
decision_id, timestamp, stock_code, market, exchange_code,
action, confidence, rationale, context_snapshot, input_data
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"d1",
"2026-02-14T09:00:00+00:00",
"005930",
"KR",
"KRX",
"HOLD",
50,
"test",
"{invalid_json",
json.dumps({}),
),
)
db_conn.commit()
scorecard = reviewer.generate_scorecard("2026-02-14", "KR")
assert scorecard.total_decisions == 1
assert scorecard.scenario_match_rate == 0.0

View File

@@ -1,12 +1,61 @@
"""Tests for main trading loop telegram integration.""" """Tests for main trading loop integration."""
import asyncio from datetime import UTC, date
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import ANY, AsyncMock, MagicMock, patch
import pytest import pytest
from src.context.layer import ContextLayer
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected
from src.main import safe_float, trading_cycle from src.db import init_db, log_trade
from src.evolution.scorecard import DailyScorecard
from src.logging.decision_logger import DecisionLogger
from src.main import _handle_market_close, 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",
)
def _make_sell_match(stock_code: str = "005930") -> ScenarioMatch:
"""Create a ScenarioMatch that returns SELL."""
return ScenarioMatch(
stock_code=stock_code,
matched_scenario=None,
action=ScenarioAction.SELL,
confidence=90,
rationale="Test sell",
)
class TestSafeFloat: class TestSafeFloat:
@@ -81,15 +130,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 +184,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 +202,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 +217,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 +243,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 +262,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 +283,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 +306,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 +332,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 +357,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 +378,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 +389,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 +529,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 +575,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 +600,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 +621,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 +637,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 +658,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 +674,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 +695,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 +711,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 +726,572 @@ 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()
@pytest.mark.asyncio
async def test_sell_updates_original_buy_decision_outcome() -> None:
"""SELL should update the original BUY decision outcome in decision_logs."""
db_conn = init_db(":memory:")
decision_logger = DecisionLogger(db_conn)
buy_decision_id = decision_logger.log_decision(
stock_code="005930",
market="KR",
exchange_code="KRX",
action="BUY",
confidence=85,
rationale="Initial buy",
context_snapshot={},
input_data={},
)
log_trade(
conn=db_conn,
stock_code="005930",
action="BUY",
confidence=85,
rationale="Initial buy",
quantity=1,
price=100.0,
pnl=0.0,
market="KR",
exchange_code="KRX",
decision_id=buy_decision_id,
)
broker = MagicMock()
broker.get_orderbook = AsyncMock(
return_value={"output1": {"stck_prpr": "120", "frgn_ntby_qty": "0"}}
)
broker.get_balance = AsyncMock(
return_value={
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "10000",
"pchs_amt_smtl_amt": "90000",
}
]
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
overseas_broker = MagicMock()
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_sell_match())
risk = MagicMock()
context_store = MagicMock(
get_latest_timeframe=MagicMock(return_value=None),
set_context=MagicMock(),
)
criticality_assessor = MagicMock(
assess_market_conditions=MagicMock(return_value=MagicMock(value="NORMAL")),
get_timeout=MagicMock(return_value=5.0),
)
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
await trading_cycle(
broker=broker,
overseas_broker=overseas_broker,
scenario_engine=engine,
playbook=_make_playbook(),
risk=risk,
db_conn=db_conn,
decision_logger=decision_logger,
context_store=context_store,
criticality_assessor=criticality_assessor,
telegram=telegram,
market=market,
stock_code="005930",
scan_candidates={},
)
updated_buy = decision_logger.get_decision_by_id(buy_decision_id)
assert updated_buy is not None
assert updated_buy.outcome_pnl == 20.0
assert updated_buy.outcome_accuracy == 1
@pytest.mark.asyncio
async def test_handle_market_close_runs_daily_review_flow() -> None:
"""Market close should aggregate, create scorecard, lessons, and notify."""
telegram = MagicMock()
telegram.notify_market_close = AsyncMock()
telegram.send_message = AsyncMock()
context_aggregator = MagicMock()
reviewer = MagicMock()
reviewer.generate_scorecard.return_value = DailyScorecard(
date="2026-02-14",
market="KR",
total_decisions=3,
buys=1,
sells=1,
holds=1,
total_pnl=12.5,
win_rate=50.0,
avg_confidence=75.0,
scenario_match_rate=66.7,
)
reviewer.generate_lessons = AsyncMock(return_value=["Cut losers faster"])
await _handle_market_close(
market_code="KR",
market_name="Korea",
market_timezone=UTC,
telegram=telegram,
context_aggregator=context_aggregator,
daily_reviewer=reviewer,
)
telegram.notify_market_close.assert_called_once_with("Korea", 0.0)
context_aggregator.aggregate_daily_from_trades.assert_called_once()
reviewer.generate_scorecard.assert_called_once()
assert reviewer.store_scorecard_in_context.call_count == 2
reviewer.generate_lessons.assert_called_once()
telegram.send_message.assert_called_once()
@pytest.mark.asyncio
async def test_handle_market_close_without_lessons_stores_once() -> None:
"""If no lessons are generated, scorecard should be stored once."""
telegram = MagicMock()
telegram.notify_market_close = AsyncMock()
telegram.send_message = AsyncMock()
context_aggregator = MagicMock()
reviewer = MagicMock()
reviewer.generate_scorecard.return_value = DailyScorecard(
date="2026-02-14",
market="US",
total_decisions=1,
buys=0,
sells=1,
holds=0,
total_pnl=-3.0,
win_rate=0.0,
avg_confidence=65.0,
scenario_match_rate=100.0,
)
reviewer.generate_lessons = AsyncMock(return_value=[])
await _handle_market_close(
market_code="US",
market_name="United States",
market_timezone=UTC,
telegram=telegram,
context_aggregator=context_aggregator,
daily_reviewer=reviewer,
)
assert reviewer.store_scorecard_in_context.call_count == 1

View File

@@ -0,0 +1,586 @@
"""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.context_selector import DecisionType
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,
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
@pytest.mark.asyncio
async def test_generate_playbook_uses_strategic_context_selector(self) -> None:
planner = _make_planner()
candidates = [_candidate()]
await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
planner._context_selector.select_layers.assert_called_once_with(
decision_type=DecisionType.STRATEGIC,
include_realtime=True,
)
planner._context_selector.get_context_data.assert_called_once()
# ---------------------------------------------------------------------------
# _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-07", "scorecard_US"
)
assert ctx.date == "2026-02-07"
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

81
tests/test_scorecard.py Normal file
View File

@@ -0,0 +1,81 @@
"""Tests for DailyScorecard model."""
from __future__ import annotations
from src.evolution.scorecard import DailyScorecard
def test_scorecard_initialization() -> None:
scorecard = DailyScorecard(
date="2026-02-08",
market="KR",
total_decisions=10,
buys=3,
sells=2,
holds=5,
total_pnl=1234.5,
win_rate=60.0,
avg_confidence=78.5,
scenario_match_rate=70.0,
top_winners=["005930", "000660"],
top_losers=["035420"],
lessons=["Avoid chasing breakouts"],
cross_market_note="US volatility spillover",
)
assert scorecard.market == "KR"
assert scorecard.total_decisions == 10
assert scorecard.total_pnl == 1234.5
assert scorecard.top_winners == ["005930", "000660"]
assert scorecard.lessons == ["Avoid chasing breakouts"]
assert scorecard.cross_market_note == "US volatility spillover"
def test_scorecard_defaults() -> None:
scorecard = DailyScorecard(
date="2026-02-08",
market="US",
total_decisions=0,
buys=0,
sells=0,
holds=0,
total_pnl=0.0,
win_rate=0.0,
avg_confidence=0.0,
scenario_match_rate=0.0,
)
assert scorecard.top_winners == []
assert scorecard.top_losers == []
assert scorecard.lessons == []
assert scorecard.cross_market_note == ""
def test_scorecard_list_isolation() -> None:
a = DailyScorecard(
date="2026-02-08",
market="KR",
total_decisions=1,
buys=1,
sells=0,
holds=0,
total_pnl=10.0,
win_rate=100.0,
avg_confidence=90.0,
scenario_match_rate=100.0,
)
b = DailyScorecard(
date="2026-02-08",
market="US",
total_decisions=1,
buys=0,
sells=1,
holds=0,
total_pnl=-5.0,
win_rate=0.0,
avg_confidence=60.0,
scenario_match_rate=50.0,
)
a.top_winners.append("005930")
assert b.top_winners == []

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