Compare commits

..

2 Commits

Author SHA1 Message Date
agentson
93e31cf667 docs: restore onboarding context and clarify runtime-impact gaps
Some checks failed
CI / test (pull_request) Has been cancelled
2026-02-16 12:29:54 +09:00
agentson
cc1489fd7c docs: sync V2 status and process docs for issue #131
Some checks failed
CI / test (pull_request) Has been cancelled
2026-02-16 11:58:49 +09:00
36 changed files with 851 additions and 6655 deletions

View File

@@ -15,9 +15,6 @@ pytest -v --cov=src
# Run (paper trading)
python -m src.main --mode=paper
# Run with dashboard
python -m src.main --mode=paper --dashboard
```
## Telegram Notifications (Optional)
@@ -46,10 +43,6 @@ Get real-time alerts for trades, circuit breakers, and system events via Telegra
- Market open/close notifications
- 📝 System startup/shutdown status
### Interactive Commands
With `TELEGRAM_COMMANDS_ENABLED=true` (default), the bot supports 9 bidirectional commands: `/help`, `/status`, `/positions`, `/report`, `/scenarios`, `/review`, `/dashboard`, `/stop`, `/resume`.
**Fail-safe**: Notifications never crash the trading system. Missing credentials or API errors are logged but trading continues normally.
## Smart Volatility Scanner (Optional)
@@ -116,23 +109,17 @@ User requirements and feedback are tracked in [docs/requirements-log.md](docs/re
```
src/
├── analysis/ # Technical analysis (RSI, volatility, smart scanner)
├── backup/ # Disaster recovery (scheduler, cloud storage, health)
├── brain/ # Gemini AI decision engine (prompt optimizer, context selector)
├── broker/ # KIS API client (domestic + overseas)
├── context/ # L1-L7 hierarchical memory system
├── brain/ # Gemini AI decision engine
├── core/ # Risk manager (READ-ONLY)
├── dashboard/ # FastAPI read-only monitoring (8 API endpoints)
├── data/ # External data integration (news, market data, calendar)
├── evolution/ # Self-improvement (optimizer, daily review, scorecard)
├── logging/ # Decision logger (audit trail)
├── evolution/ # Self-improvement optimizer
├── markets/ # Market schedules and timezone handling
├── notifications/ # Telegram alerts + bidirectional commands (9 commands)
├── strategy/ # Pre-market planner, scenario engine, playbook store
├── notifications/ # Telegram real-time alerts
├── db.py # SQLite trade logging
├── main.py # Trading loop orchestrator
└── config.py # Settings (from .env)
tests/ # 551 tests across 25 files
tests/ # 343 tests across 14 files
docs/ # Extended documentation
```
@@ -144,7 +131,6 @@ ruff check src/ tests/ # Lint
mypy src/ --strict # Type check
python -m src.main --mode=paper # Paper trading
python -m src.main --mode=paper --dashboard # With dashboard
python -m src.main --mode=live # Live trading (⚠️ real money)
# Gitea workflow (requires tea CLI)

260
README.md
View File

@@ -1,234 +1,126 @@
# The Ouroboros — 자가 진화형 AI 투자 시스템
KIS(한국투자증권) API로 매매하고, Google Gemini로 판단하며, 자체 전략 코드를 TDD 기반으로 진화시키는 자율 주식 트레이딩 에이전트.
KIS API 기반 자동매매 + Gemini 기반 장전 전략 생성 + 장중 로컬 시나리오 실행 + 장후 리뷰/진화 루프를 결합한 시스템입니다.
## 아키텍처
## 현재 상태 (2026-02-16)
```
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ KIS Broker │◄───►│ Main │◄───►│ Gemini Brain│
│ (매매 실행) │ │ (거래 루프) │ │ (의사결정) │
└─────────────┘ └──────┬──────┘ └─────────────┘
┌────────────┼────────────┐
│ │ │
┌──────┴──────┐ ┌──┴───┐ ┌──────┴──────┐
│Risk Manager │ │ DB │ │ Telegram │
│ (안전장치) │ │ │ │ (알림+명령) │
└──────┬──────┘ └──────┘ └─────────────┘
┌────────┼────────┐
│ │ │
┌────┴────┐┌──┴──┐┌────┴─────┐
│Strategy ││Ctx ││Evolution │
│(플레이북)││(메모리)││ (진화) │
└─────────┘└─────┘└──────────┘
```
- V2 계획 기준 완료: **18/20**
- 부분 완료: **1/20** (`1-7` 일부 항목)
- 미완료: **1/20** (`4-1` Telegram 확장 명령어)
**v2 핵심**: "Plan Once, Execute Locally" — 장 시작 전 AI가 시나리오 플레이북을 1회 생성하고, 거래 시간에는 로컬 시나리오 매칭만 수행하여 API 비용과 지연 시간을 대폭 절감.
핵심 전환은 이미 반영되었습니다.
## 핵심 모듈
- 기존: 장중 `brain.decide()` 실시간 의존
- 현재: 장전 `DayPlaybook` 생성 + 장중 `ScenarioEngine` 로컬 매칭
| 모듈 | 위치 | 설명 |
|------|------|------|
| 설정 | `src/config.py` | Pydantic 기반 환경변수 로딩 및 타입 검증 (35+ 변수) |
| 브로커 | `src/broker/` | KIS API 비동기 래퍼 (국내 + 해외 9개 시장) |
| 두뇌 | `src/brain/` | Gemini 프롬프트 구성, JSON 파싱, 토큰 최적화 |
| 방패 | `src/core/risk_manager.py` | 서킷 브레이커 + 팻 핑거 체크 (READ-ONLY) |
| 전략 | `src/strategy/` | Pre-Market Planner, Scenario Engine, Playbook Store |
| 컨텍스트 | `src/context/` | L1-L7 계층형 메모리 시스템 |
| 분석 | `src/analysis/` | RSI, ATR, Smart Volatility Scanner |
| 알림 | `src/notifications/` | 텔레그램 양방향 (알림 + 9개 명령어) |
| 대시보드 | `src/dashboard/` | FastAPI 읽기 전용 모니터링 (8개 API) |
| 진화 | `src/evolution/` | 전략 진화 + Daily Review + Scorecard |
| 의사결정 로그 | `src/logging/` | 전체 거래 결정 감사 추적 |
| 데이터 | `src/data/` | 뉴스, 시장 데이터, 경제 캘린더 연동 |
| 백업 | `src/backup/` | 자동 백업, S3 클라우드, 무결성 검증 |
| DB | `src/db.py` | SQLite 거래 로그 (5개 테이블) |
## 핵심 구성
## 안전장치
- `src/main.py`: 시장 루프, 플레이북 생성/적용, EOD 집계, 리뷰/진화 연결
- `src/strategy/`: `models`, `pre_market_planner`, `scenario_engine`, `playbook_store`
- `src/context/`: `store`, `aggregator`, `scheduler` (L1~L7)
- `src/evolution/daily_review.py`: 시장별 scorecard/lessons 생성
- `src/dashboard/app.py`: FastAPI 관측 API
- `src/notifications/telegram_client.py`: 알림 및 명령 핸들러
| 규칙 | 내용 |
|------|------|
| 서킷 브레이커 | 일일 손실률 -3.0% 초과 시 전체 매매 중단 (`SystemExit`) |
| 팻 핑거 방지 | 주문 금액이 보유 현금의 30% 초과 시 주문 거부 |
| 신뢰도 임계값 | Gemini 신뢰도 80 미만이면 강제 HOLD |
| 레이트 리미터 | Leaky Bucket 알고리즘으로 API 호출 제한 |
| 토큰 자동 갱신 | 만료 1분 전 자동으로 Access Token 재발급 |
| 손절 모니터링 | 플레이북 시나리오 기반 실시간 포지션 보호 |
## 빠른 시작
## Quick Start
### 1. 환경 설정
```bash
cp .env.example .env
# .env 파일에 KIS API 키와 Gemini API 키 입력
```
필수 값:
- `KIS_APP_KEY`
- `KIS_APP_SECRET`
- `KIS_ACCOUNT_NO`
- `GEMINI_API_KEY`
### 2. 의존성 설치
```bash
pip install ".[dev]"
pip install -e ".[dev]"
```
### 3. 테스트 실행
### 3. 테스트
```bash
pytest -v --cov=src --cov-report=term-missing
pytest -v --cov=src
ruff check src/ tests/
mypy src/ --strict
```
### 4. 실행 (모의투자)
## 실행
### 기본 실행
```bash
# 기본 실행
python -m src.main --mode=paper
```
# 대시보드 활성화
### 대시보드 포함 실행
```bash
python -m src.main --mode=paper --dashboard
```
### 5. Docker 실행
또는 환경변수:
```bash
docker compose up -d ouroboros
DASHBOARD_ENABLED=true
DASHBOARD_HOST=127.0.0.1
DASHBOARD_PORT=8080
```
## 지원 시장
## 주요 API/기능
| 국가 | 거래소 | 코드 |
|------|--------|------|
| 🇰🇷 한국 | KRX | KR |
| 🇺🇸 미국 | NASDAQ, NYSE, AMEX | US_NASDAQ, US_NYSE, US_AMEX |
| 🇯🇵 일본 | TSE | JP |
| 🇭🇰 홍콩 | SEHK | HK |
| 🇨🇳 중국 | 상하이, 선전 | CN_SHA, CN_SZA |
| 🇻🇳 베트남 | 하노이, 호치민 | VN_HNX, VN_HSX |
- 플레이북 저장: `playbooks` 테이블 (`date + market` UNIQUE)
- 의사결정/결과 연결: `trades.decision_id` + `DecisionLogger.update_outcome()`
- 시장별 scorecard 컨텍스트: `scorecard_KR`, `scorecard_US`
- 컨텍스트 스케줄러: weekly/monthly/quarterly/annual/legacy + cleanup
- 대시보드 API:
- `/api/status`
- `/api/playbook/{date}?market=KR`
- `/api/scorecard/{date}?market=KR`
- `/api/performance?market=all`
- `/api/context/{layer}`
- `/api/decisions?market=KR`
- `/api/scenarios/active?market=US`
`ENABLED_MARKETS` 환경변수로 활성 시장 선택 (기본: `KR,US`).
## 현재 갭 (코드 기준)
## 텔레그램 (선택사항)
거래 실행, 서킷 브레이커 발동, 시스템 상태 등을 텔레그램으로 실시간 알림 받을 수 있습니다.
### 빠른 설정
1. **봇 생성**: 텔레그램에서 [@BotFather](https://t.me/BotFather) 메시지 → `/newbot` 명령
2. **채팅 ID 확인**: [@userinfobot](https://t.me/userinfobot) 메시지 → `/start` 명령
3. **환경변수 설정**: `.env` 파일에 추가
```bash
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
TELEGRAM_CHAT_ID=123456789
TELEGRAM_ENABLED=true
```
4. **테스트**: 봇과 대화 시작 (`/start` 전송) 후 에이전트 실행
**상세 문서**: [src/notifications/README.md](src/notifications/README.md)
### 알림 종류
- 🟢 거래 체결 알림 (BUY/SELL + 신뢰도)
- 🚨 서킷 브레이커 발동 (자동 거래 중단)
- ⚠️ 팻 핑거 차단 (과도한 주문 차단)
- 장 시작/종료 알림
- 📝 시스템 시작/종료 상태
### 양방향 명령어
`TELEGRAM_COMMANDS_ENABLED=true` (기본값) 설정 시 9개 대화형 명령어 지원:
| 명령어 | 설명 |
|--------|------|
| `/help` | 사용 가능한 명령어 목록 |
| `/status` | 거래 상태 (모드, 시장, P&L) |
| `/positions` | 계좌 요약 (잔고, 현금, P&L) |
| `/report` | 일일 요약 (거래 수, P&L, 승률) |
| `/scenarios` | 오늘의 플레이북 시나리오 |
| `/review` | 최근 스코어카드 (L6_DAILY) |
| `/dashboard` | 대시보드 URL 표시 |
| `/stop` | 거래 일시 정지 |
| `/resume` | 거래 재개 |
**안전장치**: 알림 실패해도 거래는 계속 진행됩니다.
- `Issue 4-1` 미구현: `/report`, `/scenarios`, `/review`, `/dashboard` Telegram 명령 미등록
- `Issue 1-7` 일부 미완:
- `price_change_pct` 정규화 어댑터 명시 구현 없음
- 영향: `price_change_pct_above/below` 조건을 사용하는 시나리오는 사실상 매칭 불가(dead path)
- HOLD 시 별도 손절 모니터링 플래그 처리 분리 미흡
- 시장 코드 정합성 이슈:
- 설정 기본값은 `ENABLED_MARKETS="KR,US"`
- 스케줄 정의는 `US_NASDAQ`, `US_NYSE` 중심
- 영향: `get_open_markets(["KR", "US"])`에서 `US` 미정의로 US 시장이 누락될 수 있음(런타임 영향)
## 테스트
551개 테스트가 25개 파일에 걸쳐 구현되어 있습니다. 최소 커버리지 80%.
로컬 수집 기준:
```
tests/test_scenario_engine.py — 시나리오 매칭 (44개)
tests/test_data_integration.py — 외부 데이터 연동 (38개)
tests/test_pre_market_planner.py — 플레이북 생성 (37개)
tests/test_main.py — 거래 루프 통합 (37개)
tests/test_token_efficiency.py — 토큰 최적화 (34개)
tests/test_strategy_models.py — 전략 모델 검증 (33개)
tests/test_telegram_commands.py — 텔레그램 명령어 (31개)
tests/test_latency_control.py — 지연시간 제어 (30개)
tests/test_telegram.py — 텔레그램 알림 (25개)
... 외 16개 파일
```bash
pytest --collect-only -q
# 538 tests collected
```
**상세**: [docs/testing.md](docs/testing.md)
권장 검증:
## 기술 스택
- **언어**: Python 3.11+ (asyncio 기반)
- **브로커**: KIS Open API (REST, 국내+해외)
- **AI**: Google Gemini Pro
- **DB**: SQLite (5개 테이블: trades, contexts, decision_logs, playbooks, context_metadata)
- **대시보드**: FastAPI + uvicorn
- **검증**: pytest + coverage (551 tests)
- **CI/CD**: GitHub Actions
- **배포**: Docker + Docker Compose
## 프로젝트 구조
```
The-Ouroboros/
├── docs/
│ ├── architecture.md # 시스템 아키텍처
│ ├── testing.md # 테스트 가이드
│ ├── commands.md # 명령어 레퍼런스
│ ├── context-tree.md # L1-L7 메모리 시스템
│ ├── workflow.md # Git 워크플로우
│ ├── agents.md # 에이전트 정책
│ ├── skills.md # 도구 목록
│ ├── disaster_recovery.md # 백업/복구
│ └── requirements-log.md # 요구사항 기록
├── src/
│ ├── analysis/ # 기술적 분석 (RSI, ATR, Smart Scanner)
│ ├── backup/ # 백업 (스케줄러, S3, 무결성 검증)
│ ├── brain/ # Gemini 의사결정 (프롬프트 최적화, 컨텍스트 선택)
│ ├── broker/ # KIS API (국내 + 해외)
│ ├── context/ # L1-L7 계층 메모리
│ ├── core/ # 리스크 관리 (READ-ONLY)
│ ├── dashboard/ # FastAPI 모니터링 대시보드
│ ├── data/ # 외부 데이터 연동
│ ├── evolution/ # 전략 진화 + Daily Review
│ ├── logging/ # 의사결정 감사 추적
│ ├── markets/ # 시장 스케줄 + 타임존
│ ├── notifications/ # 텔레그램 알림 + 명령어
│ ├── strategy/ # 플레이북 (Planner, Scenario Engine)
│ ├── config.py # Pydantic 설정
│ ├── db.py # SQLite 데이터베이스
│ └── main.py # 비동기 거래 루프
├── tests/ # 551개 테스트 (25개 파일)
├── Dockerfile # 멀티스테이지 빌드
├── docker-compose.yml # 서비스 오케스트레이션
└── pyproject.toml # 의존성 및 도구 설정
```bash
pytest -v --cov=src
ruff check src/ tests/
mypy src/ --strict
```
## 문서
- **[아키텍처](docs/architecture.md)** — 시스템 설계, 컴포넌트, 데이터 흐름
- **[테스트](docs/testing.md)** — 테스트 구조, 커버리지, 작성 가이드
- **[명령어](docs/commands.md)** — CLI, Dashboard, Telegram 명령어
- **[컨텍스트 트리](docs/context-tree.md)** — L1-L7 계층 메모리
- **[워크플로우](docs/workflow.md)** — Git 워크플로우 정책
- **[에이전트 정책](docs/agents.md)** — 안전 제약, 금지 행위
- **[백업/복구](docs/disaster_recovery.md)** — 재해 복구 절차
- **[요구사항](docs/requirements-log.md)** — 사용자 요구사항 추적
## 라이선스
이 프로젝트의 라이선스는 [LICENSE](LICENSE) 파일을 참조하세요.
- 아키텍처: `docs/architecture.md`
- 컨텍스트 트리: `docs/context-tree.md`
- 워크플로우: `docs/workflow.md`
- 요구사항 로그: `docs/requirements-log.md`
- 명령 레퍼런스: `docs/commands.md`

View File

@@ -2,642 +2,140 @@
## Overview
Self-evolving AI trading agent for global stock markets via KIS (Korea Investment & Securities) API. The main loop in `src/main.py` orchestrates components across multiple markets with two trading modes: daily (batch API calls) or realtime (per-stock decisions).
The Ouroboros V2는 `Proactive` 구조를 중심으로 동작합니다.
**v2 Proactive Playbook Architecture**: The system uses a "plan once, execute locally" approach. Pre-market, the AI generates a playbook of scenarios (one Gemini API call per market per day). During trading hours, a local scenario engine matches live market data against these pre-computed scenarios — no additional AI calls needed. This dramatically reduces API costs and latency.
- 장전: Gemini 1회 호출로 시장별 `DayPlaybook` 생성
- 장중: `ScenarioEngine`이 로컬 조건 매칭으로 의사결정
- 장후: `ContextAggregator` + `DailyReviewer`로 성과 집계/교훈 생성
## Trading Modes
`main.py`가 아래 컴포넌트를 오케스트레이션합니다.
The system supports two trading frequency modes controlled by the `TRADE_MODE` environment variable:
- `KISBroker` / `OverseasBroker`
- `PreMarketPlanner` / `ScenarioEngine` / `PlaybookStore`
- `ContextStore` / `ContextAggregator` / `ContextScheduler`
- `DailyReviewer` / `EvolutionOptimizer`
- `TelegramClient` / `TelegramCommandHandler`
### Daily Mode (default)
안전/운영 컴포넌트도 핵심입니다.
Optimized for Gemini Free tier API limits (20 calls/day):
- `RiskManager`: circuit breaker, fat-finger 검증
- `PriorityTaskQueue` + `CriticalityAssessor`: 우선순위/지연 제어
- **Batch decisions**: 1 API call per market per session
- **Fixed schedule**: 4 sessions per day at 6-hour intervals (configurable)
- **API efficiency**: Processes all stocks in a market simultaneously
- **Use case**: Free tier users, cost-conscious deployments
- **Configuration**:
```bash
TRADE_MODE=daily
DAILY_SESSIONS=4 # Sessions per day (1-10)
SESSION_INTERVAL_HOURS=6 # Hours between sessions (1-24)
```
## Market Scope
**Example**: With 2 markets (US, KR) and 4 sessions/day = 8 API calls/day (within 20 call limit)
V2 기본 설정은 `ENABLED_MARKETS="KR,US"` 입니다.
### Realtime Mode
현재 코드 기준 주의점(런타임 영향):
High-frequency trading with individual stock analysis:
- 설정은 `KR,US`를 기본값으로 사용
- 스케줄 레이어(`src/markets/schedule.py`)는 `US_NASDAQ`, `US_NYSE` 구조를 아직 유지
- `US` 코드가 스케줄에 직접 정의되지 않아 US 시장 누락 가능성이 있음
- **Per-stock decisions**: 1 API call per stock per cycle
- **60-second interval**: Continuous monitoring
- **Use case**: Production deployments with Gemini paid tier
- **Configuration**:
```bash
TRADE_MODE=realtime
```
## Decision Flow
**Note**: Realtime mode requires Gemini API subscription due to high call volume.
### 1) Pre-market
## Core Components
1. `SmartVolatilityScanner.scan()`으로 후보 종목 수집
2. `PreMarketPlanner.generate_playbook(market, candidates)` 호출
3. 결과를 `PlaybookStore.save()`로 DB 저장
4. 실패 시 empty/defensive playbook 사용
### 1. Broker (`src/broker/`)
### 2) In-market
**KISBroker** (`kis_api.py`) — Async KIS API client for domestic Korean market
1. 시장 데이터 + 스캐너 메트릭(`rsi`, `volume_ratio`) 구성
2. `ScenarioEngine.evaluate(playbook, stock_code, market_data, portfolio_data)`
3. `TradeDecision` 변환 후 주문/로그/알림 처리
4. `decision_logs``trades``decision_id`로 연결
- Automatic OAuth token refresh (valid for 24 hours)
- Leaky-bucket rate limiter (configurable RPS, default 2.0)
- POST body hash-key signing for order authentication
- Custom SSL context with disabled hostname verification for VTS (virtual trading) endpoint due to known certificate mismatch
- `fetch_market_rankings()` — Fetch volume surge rankings from KIS API
- `get_daily_prices()` — Fetch OHLCV history for technical analysis
### 3) End-of-day
**OverseasBroker** (`overseas.py`) — KIS overseas stock API wrapper
1. `ContextAggregator.aggregate_daily_from_trades(date, market)`
2. `DailyReviewer.generate_scorecard(date, market)`
3. `store_scorecard_in_context()``scorecard_{market}` 저장
4. `generate_lessons()`로 장후 교훈 생성
5. (US 종료 시) `EvolutionOptimizer.evolve()` 실행
- Reuses KISBroker infrastructure (session, token, rate limiter) via composition
- Supports 9 global markets: US (NASDAQ/NYSE/AMEX), Japan, Hong Kong, China (Shanghai/Shenzhen), Vietnam (Hanoi/HCM)
- Different API endpoints for overseas price/balance/order operations
## Risk Policy
**Market Schedule** (`src/markets/schedule.py`) — Timezone-aware market management
- `RiskManager`는 주문 전 검증을 강제합니다.
- circuit breaker: 손실 임계치 하회 시 거래 중단
- fat-finger: 주문 금액 과대 시 주문 차단
- 실패 시 알림은 보내되, 예외 처리로 루프 안정성 유지
- `MarketInfo` dataclass with timezone, trading hours, lunch breaks
- Automatic DST handling via `zoneinfo.ZoneInfo`
- `is_market_open()` checks weekends, trading hours, lunch breaks
- `get_open_markets()` returns currently active markets
- `get_next_market_open()` finds next market to open and when
- 10 global markets defined (KR, US_NASDAQ, US_NYSE, US_AMEX, JP, HK, CN_SHA, CN_SZA, VN_HNX, VN_HSX)
## Error Handling Strategy
**Overseas Ranking API Methods** (added in v0.10.x):
- `fetch_overseas_rankings()` — Fetch overseas ranking universe (fluctuation / volume)
- Ranking endpoint paths and TR_IDs are configurable via environment variables
- API 호출 실패: 재시도(지수 백오프) 후 종목/사이클 스킵
- 시나리오/플래너 실패: empty 또는 defensive playbook으로 안전 폴백
- Telegram 실패: warning 로깅 후 거래 루프 지속
- 대시보드 스레드 실패: warning 로깅 후 메인 트레이딩 루프와 분리 유지
### 2. Analysis (`src/analysis/`)
## Configuration Reference
**VolatilityAnalyzer** (`volatility.py`) — Technical indicator calculations
상세 설정은 `src/config.py`를 기준으로 합니다.
- ATR (Average True Range) for volatility measurement
- RSI (Relative Strength Index) using Wilder's smoothing method
- Price change percentages across multiple timeframes
- Volume surge ratios and price-volume divergence
- Momentum scoring (0-100 scale)
- Breakout/breakdown pattern detection
- 거래 모드: `TRADE_MODE`, `DAILY_SESSIONS`, `SESSION_INTERVAL_HOURS`
- 전략: `PRE_MARKET_MINUTES`, `MAX_SCENARIOS_PER_STOCK`, `RESCAN_INTERVAL_SECONDS`
- 시장: `ENABLED_MARKETS`
- 대시보드: `DASHBOARD_ENABLED`, `DASHBOARD_HOST`, `DASHBOARD_PORT`
- 알림: `TELEGRAM_*`
**SmartVolatilityScanner** (`smart_scanner.py`) — Python-first filtering pipeline
## Context Tree
- **Domestic (KR)**:
- **Step 1**: Fetch domestic fluctuation ranking as primary universe
- **Step 2**: Fetch domestic volume ranking for liquidity bonus
- **Step 3**: Compute volatility-first score (max of daily change% and intraday range%)
- **Step 4**: Apply liquidity bonus and return top N candidates
- **Overseas (US/JP/HK/CN/VN)**:
- **Step 1**: Fetch overseas ranking universe (fluctuation rank + volume rank bonus)
- **Step 2**: Compute volatility-first score (max of daily change% and intraday range%)
- **Step 3**: Apply liquidity bonus from volume ranking
- **Step 4**: Return top N candidates (default 3)
- **Fallback (overseas only)**: If ranking API is unavailable, uses dynamic universe
from runtime active symbols + recent traded symbols + current holdings (no static watchlist)
- **Realtime mode only**: Daily mode uses batch processing for API efficiency
레이어 전략:
**Benefits:**
- Reduces Gemini API calls from 20-30 stocks to 1-3 qualified candidates
- Fast Python-based filtering before expensive AI judgment
- Logs selection context (RSI-compatible proxy, volume_ratio, signal, score) for Evolution system
- `L7~L5`: 시장별 키
- `L4~L1`: 글로벌 통합 롤업
### 3. Brain (`src/brain/`)
구현 포인트:
**GeminiClient** (`gemini_client.py`) — AI decision engine powered by Google Gemini
- `L7` 쓰기: `volatility_{market}_{stock}`
- `L6` 집계: `total_pnl_KR`, `trade_count_US`
- `ContextScheduler.run_if_due()`:
- 주간/월간/분기/연간/legacy 집계
- 일 1회 `cleanup_expired_contexts()` 호출
- Constructs structured prompts from market data
- Parses JSON responses into `TradeDecision` objects (`action`, `confidence`, `rationale`)
- Forces HOLD when confidence < threshold (default 80)
- Falls back to safe HOLD on any parse/API error
- Handles markdown-wrapped JSON, malformed responses, invalid actions
## Data Model (핵심)
**PromptOptimizer** (`prompt_optimizer.py`) — Token efficiency optimization
### `trades`
- Reduces prompt size while preserving decision quality
- Caches optimized prompts
- `market`, `exchange_code`, `selection_context`, `decision_id` 포함
- SELL 시 `get_latest_buy_trade()`를 통해 원본 BUY `decision_id`를 찾아 결과 업데이트
**ContextSelector** (`context_selector.py`) — Relevant context selection for prompts
### `decision_logs`
- Selects appropriate context layers for current market conditions
- 의사결정 입력/컨텍스트 스냅샷 저장
- `outcome_pnl`, `outcome_accuracy` 업데이트 가능
### 4. Risk Manager (`src/core/risk_manager.py`)
### `playbooks`
**RiskManager** — Safety circuit breaker and order validation
- `UNIQUE(date, market)`
- `status`, `token_count`, `scenario_count`, `match_count` 관리
> **READ-ONLY by policy** (see [`docs/agents.md`](./agents.md))
## Dashboard
- **Circuit Breaker**: Halts all trading via `SystemExit` when daily P&L drops below -3.0%
- Threshold may only be made stricter, never relaxed
- Calculated as `(total_eval - purchase_total) / purchase_total * 100`
- **Fat-Finger Protection**: Rejects orders exceeding 30% of available cash
- Must always be enforced, cannot be disabled
`src/dashboard/app.py`의 FastAPI 앱이 SQLite를 직접 조회합니다.
### 5. Strategy (`src/strategy/`)
엔드포인트:
**Pre-Market Planner** (`pre_market_planner.py`) — AI playbook generation
- `GET /api/status`
- `GET /api/playbook/{date}?market=KR`
- `GET /api/scorecard/{date}?market=KR`
- `GET /api/performance?market=all`
- `GET /api/context/{layer}`
- `GET /api/decisions?market=KR`
- `GET /api/scenarios/active?market=US`
- Runs before market open (configurable `PRE_MARKET_MINUTES`, default 30)
- Generates scenario-based playbooks via single Gemini API call per market
- Handles timeout (`PLANNER_TIMEOUT_SECONDS`, default 60) with defensive playbook fallback
- Persists playbooks to database for audit trail
실행 통합:
**Scenario Engine** (`scenario_engine.py`) — Local scenario matching
- CLI `--dashboard`
- 또는 `DASHBOARD_ENABLED=true`
- `main.py`에서 daemon thread로 uvicorn 실행
- Matches live market data against pre-computed playbook scenarios
- No AI calls during trading hours — pure Python matching logic
- Returns matched scenarios with confidence scores
- Configurable `MAX_SCENARIOS_PER_STOCK` (default 5)
- Periodic rescan at `RESCAN_INTERVAL_SECONDS` (default 300)
## Known Gaps (2026-02-16)
**Playbook Store** (`playbook_store.py`) — Playbook persistence
- SQLite-backed storage for daily playbooks
- Date and market-based retrieval
- Status tracking (generated, active, expired)
**Models** (`models.py`) — Pydantic data models
- Scenario, Playbook, MatchResult, and related type definitions
### 6. Context System (`src/context/`)
**Context Store** (`store.py`) — L1-L7 hierarchical memory
- 7-layer context system (see [docs/context-tree.md](./context-tree.md)):
- L1: Tick-level (real-time price)
- L2: Intraday (session summary)
- L3: Daily (end-of-day)
- L4: Weekly (trend analysis)
- L5: Monthly (strategy review)
- L6: Daily Review (scorecard)
- L7: Evolution (long-term learning)
- Key-value storage with timeframe tagging
- SQLite persistence in `contexts` table
**Context Scheduler** (`scheduler.py`) — Periodic aggregation
- Scheduled summarization from lower to higher layers
- Configurable aggregation intervals
**Context Summarizer** (`summarizer.py`) — Layer summarization
- Aggregates lower-layer data into higher-layer summaries
### 7. Dashboard (`src/dashboard/`)
**FastAPI App** (`app.py`) — Read-only monitoring dashboard
- Runs as daemon thread when enabled (`--dashboard` CLI flag or `DASHBOARD_ENABLED=true`)
- Configurable host/port (`DASHBOARD_HOST`, `DASHBOARD_PORT`, default `127.0.0.1:8080`)
- Serves static HTML frontend
**8 API Endpoints:**
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/` | GET | Static HTML dashboard |
| `/api/status` | GET | Daily trading status by market |
| `/api/playbook/{date}` | GET | Playbook for specific date and market |
| `/api/scorecard/{date}` | GET | Daily scorecard from L6_DAILY context |
| `/api/performance` | GET | Trading performance metrics (by market + combined) |
| `/api/context/{layer}` | GET | Query context by layer (L1-L7) |
| `/api/decisions` | GET | Decision log entries with outcomes |
| `/api/scenarios/active` | GET | Today's matched scenarios |
### 8. Notifications (`src/notifications/telegram_client.py`)
**TelegramClient** — Real-time event notifications via Telegram Bot API
- Sends alerts for trades, circuit breakers, fat-finger rejections, system events
- Non-blocking: failures are logged but never crash trading
- Rate-limited: 1 message/second default to respect Telegram API limits
- Auto-disabled when credentials missing
**TelegramCommandHandler** — Bidirectional command interface
- Long polling from Telegram API (configurable `TELEGRAM_POLLING_INTERVAL`)
- 9 interactive commands: `/help`, `/status`, `/positions`, `/report`, `/scenarios`, `/review`, `/dashboard`, `/stop`, `/resume`
- Authorization filtering by `TELEGRAM_CHAT_ID`
- Enable/disable via `TELEGRAM_COMMANDS_ENABLED` (default: true)
**Notification Types:**
- Trade execution (BUY/SELL with confidence)
- Circuit breaker trips (critical alert)
- Fat-finger protection triggers (order rejection)
- Market open/close events
- System startup/shutdown status
- Playbook generation results
- Stop-loss monitoring alerts
### 9. Evolution (`src/evolution/`)
**StrategyOptimizer** (`optimizer.py`) — Self-improvement loop
- Analyzes high-confidence losing trades from SQLite
- Asks Gemini to generate new `BaseStrategy` subclasses
- Validates generated strategies by running full pytest suite
- Simulates PR creation for human review
- Only activates strategies that pass all tests
**DailyReview** (`daily_review.py`) — End-of-day review
- Generates comprehensive trade performance summary
- Stores results in L6_DAILY context layer
- Tracks win rate, P&L, confidence accuracy
**DailyScorecard** (`scorecard.py`) — Performance scoring
- Calculates daily metrics (trades, P&L, win rate, avg confidence)
- Enables trend tracking across days
**Stop-Loss Monitoring** — Real-time position protection
- Monitors positions against stop-loss levels from playbook scenarios
- Sends Telegram alerts when thresholds approached or breached
### 10. Decision Logger (`src/logging/decision_logger.py`)
**DecisionLogger** — Comprehensive audit trail
- Logs every trading decision with full context snapshot
- Captures input data, rationale, confidence, and outcomes
- Supports outcome tracking (P&L, accuracy) for post-analysis
- Stored in `decision_logs` table with indexed queries
- Review workflow support (reviewed flag, review notes)
### 11. Data Integration (`src/data/`)
**External Data Sources** (optional):
- `news_api.py` — News sentiment data
- `market_data.py` — Extended market data
- `economic_calendar.py` — Economic event calendar
### 12. Backup (`src/backup/`)
**Disaster Recovery** (see [docs/disaster_recovery.md](./disaster_recovery.md)):
- `scheduler.py` — Automated backup scheduling
- `exporter.py` — Data export to various formats
- `cloud_storage.py` — S3-compatible cloud backup
- `health_monitor.py` — Backup integrity verification
## Data Flow
### Playbook Mode (Daily — Primary v2 Flow)
```
┌─────────────────────────────────────────────────────────────┐
│ Pre-Market Phase (before market open) │
└─────────────────────────────────────────────────────────────┘
┌──────────────────────────────────┐
│ Pre-Market Planner │
│ - 1 Gemini API call per market │
│ - Generate scenario playbook │
│ - Store in playbooks table │
└──────────────────┬───────────────┘
┌─────────────────────────────────────────────────────────────┐
│ Trading Hours (market open → close) │
└─────────────────────────────────────────────────────────────┘
┌──────────────────────────────────┐
│ Market Schedule Check │
│ - Get open markets │
│ - Filter by enabled markets │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Scenario Engine (local) │
│ - Match live data vs playbook │
│ - No AI calls needed │
│ - Return matched scenarios │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Risk Manager: Validate Order │
│ - Check circuit breaker │
│ - Check fat-finger limit │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Broker: Execute Order │
│ - Domestic: send_order() │
│ - Overseas: send_overseas_order()│
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Decision Logger + DB │
│ - Full audit trail │
│ - Context snapshot │
│ - Telegram notification │
└──────────────────┬───────────────┘
┌─────────────────────────────────────────────────────────────┐
│ Post-Market Phase │
└─────────────────────────────────────────────────────────────┘
┌──────────────────────────────────┐
│ Daily Review + Scorecard │
│ - Performance summary │
│ - Store in L6_DAILY context │
│ - Evolution learning │
└──────────────────────────────────┘
```
### Realtime Mode (with Smart Scanner)
```
┌─────────────────────────────────────────────────────────────┐
│ Main Loop (60s cycle per market) │
└─────────────────────────────────────────────────────────────┘
┌──────────────────────────────────┐
│ Market Schedule Check │
│ - Get open markets │
│ - Filter by enabled markets │
│ - Wait if all closed │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Smart Scanner (Python-first) │
│ - Domestic: fluctuation rank │
│ + volume rank bonus │
│ + volatility-first scoring │
│ - Overseas: ranking universe │
│ + volatility-first scoring │
│ - Fallback: dynamic universe │
│ - Return top 3 qualified stocks │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ For Each Qualified Candidate │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Broker: Fetch Market Data │
│ - Domestic: orderbook + balance │
│ - Overseas: price + balance │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Brain: Get Decision (AI) │
│ - Build prompt with market data │
│ - Call Gemini API │
│ - Parse JSON response │
│ - Return TradeDecision │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Risk Manager: Validate Order │
│ - Check circuit breaker │
│ - Check fat-finger limit │
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Broker: Execute Order │
│ - Domestic: send_order() │
│ - Overseas: send_overseas_order()│
└──────────────────┬───────────────┘
┌──────────────────────────────────┐
│ Decision Logger + Notifications │
│ - Log trade to SQLite │
│ - selection_context (JSON) │
│ - Telegram notification │
└──────────────────────────────────┘
```
## Database Schema
**SQLite** (`src/db.py`) — Database: `data/trades.db`
### trades
```sql
CREATE TABLE trades (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
stock_code TEXT NOT NULL,
action TEXT NOT NULL, -- BUY | SELL | HOLD
confidence INTEGER NOT NULL, -- 0-100
rationale TEXT,
quantity INTEGER,
price REAL,
pnl REAL DEFAULT 0.0,
market TEXT DEFAULT 'KR',
exchange_code TEXT DEFAULT 'KRX',
selection_context TEXT, -- JSON: {rsi, volume_ratio, signal, score}
decision_id TEXT -- Links to decision_logs
);
```
### contexts
```sql
CREATE TABLE contexts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
layer TEXT NOT NULL, -- L1 through L7
timeframe TEXT,
key TEXT NOT NULL,
value TEXT NOT NULL, -- JSON data
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
-- Indices: idx_contexts_layer, idx_contexts_timeframe, idx_contexts_updated
```
### decision_logs
```sql
CREATE TABLE decision_logs (
decision_id TEXT PRIMARY KEY,
timestamp TEXT NOT NULL,
stock_code TEXT,
market TEXT,
exchange_code TEXT,
action TEXT,
confidence INTEGER,
rationale TEXT,
context_snapshot TEXT, -- JSON: full context at decision time
input_data TEXT, -- JSON: market data used
outcome_pnl REAL,
outcome_accuracy REAL,
reviewed INTEGER DEFAULT 0,
review_notes TEXT
);
-- Indices: idx_decision_logs_timestamp, idx_decision_logs_reviewed, idx_decision_logs_confidence
```
### playbooks
```sql
CREATE TABLE playbooks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT NOT NULL,
market TEXT NOT NULL,
status TEXT DEFAULT 'generated',
playbook_json TEXT NOT NULL, -- Full playbook with scenarios
generated_at TEXT NOT NULL,
token_count INTEGER,
scenario_count INTEGER,
match_count INTEGER DEFAULT 0
);
-- Indices: idx_playbooks_date, idx_playbooks_market
```
### context_metadata
```sql
CREATE TABLE context_metadata (
layer TEXT PRIMARY KEY,
description TEXT,
retention_days INTEGER,
aggregation_source TEXT
);
```
## Configuration
**Pydantic Settings** (`src/config.py`)
Loaded from `.env` file:
```bash
# Required
KIS_APP_KEY=your_app_key
KIS_APP_SECRET=your_app_secret
KIS_ACCOUNT_NO=XXXXXXXX-XX
GEMINI_API_KEY=your_gemini_key
# Optional — Trading Mode
MODE=paper # paper | live
TRADE_MODE=daily # daily | realtime
DAILY_SESSIONS=4 # Sessions per day (daily mode only)
SESSION_INTERVAL_HOURS=6 # Hours between sessions (daily mode only)
# Optional — Database
DB_PATH=data/trades.db
# Optional — Risk
CONFIDENCE_THRESHOLD=80
MAX_LOSS_PCT=3.0
MAX_ORDER_PCT=30.0
# Optional — Markets
ENABLED_MARKETS=KR,US # Comma-separated market codes
RATE_LIMIT_RPS=2.0 # KIS API requests per second
# Optional — Pre-Market Planner (v2)
PRE_MARKET_MINUTES=30 # Minutes before market open to generate playbook
MAX_SCENARIOS_PER_STOCK=5 # Max scenarios per stock in playbook
PLANNER_TIMEOUT_SECONDS=60 # Timeout for playbook generation
DEFENSIVE_PLAYBOOK_ON_FAILURE=true # Fallback on AI failure
RESCAN_INTERVAL_SECONDS=300 # Scenario rescan interval during trading
# Optional — Smart Scanner (realtime mode only)
RSI_OVERSOLD_THRESHOLD=30 # 0-50, oversold threshold
RSI_MOMENTUM_THRESHOLD=70 # 50-100, momentum threshold
VOL_MULTIPLIER=2.0 # Minimum volume ratio (2.0 = 200%)
SCANNER_TOP_N=3 # Max qualified candidates per scan
# Optional — Dashboard
DASHBOARD_ENABLED=false # Enable FastAPI dashboard
DASHBOARD_HOST=127.0.0.1 # Dashboard bind address
DASHBOARD_PORT=8080 # Dashboard port (1-65535)
# Optional — Telegram
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
TELEGRAM_CHAT_ID=123456789
TELEGRAM_ENABLED=true
TELEGRAM_COMMANDS_ENABLED=true # Enable bidirectional commands
TELEGRAM_POLLING_INTERVAL=1.0 # Command polling interval (seconds)
# Optional — Backup
BACKUP_ENABLED=false
BACKUP_DIR=data/backups
S3_ENDPOINT_URL=...
S3_ACCESS_KEY=...
S3_SECRET_KEY=...
S3_BUCKET_NAME=...
S3_REGION=...
# Optional — External Data
NEWS_API_KEY=...
NEWS_API_PROVIDER=...
MARKET_DATA_API_KEY=...
# Position Sizing (optional)
POSITION_SIZING_ENABLED=true
POSITION_BASE_ALLOCATION_PCT=5.0
POSITION_MIN_ALLOCATION_PCT=1.0
POSITION_MAX_ALLOCATION_PCT=10.0
POSITION_VOLATILITY_TARGET_SCORE=50.0
# Legacy/compat scanner thresholds (kept for backward compatibility)
RSI_OVERSOLD_THRESHOLD=30
RSI_MOMENTUM_THRESHOLD=70
VOL_MULTIPLIER=2.0
# Overseas Ranking API (optional override; account-dependent)
OVERSEAS_RANKING_ENABLED=true
OVERSEAS_RANKING_FLUCT_TR_ID=HHDFS76200100
OVERSEAS_RANKING_VOLUME_TR_ID=HHDFS76200200
OVERSEAS_RANKING_FLUCT_PATH=/uapi/overseas-price/v1/quotations/inquire-updown-rank
OVERSEAS_RANKING_VOLUME_PATH=/uapi/overseas-price/v1/quotations/inquire-volume-rank
```
Tests use in-memory SQLite (`DB_PATH=":memory:"`) and dummy credentials via `tests/conftest.py`.
## Error Handling
### Connection Errors (Broker API)
- Retry with exponential backoff (2^attempt seconds)
- Max 3 retries per stock
- After exhaustion, skip stock and continue with next
### API Quota Errors (Gemini)
- Return safe HOLD decision with confidence=0
- Log error but don't crash
- Agent continues trading on next cycle
### Circuit Breaker Tripped
- Immediately halt via `SystemExit`
- Log critical message
- Requires manual intervention to restart
### Market Closed
- Wait until next market opens
- Use `get_next_market_open()` to calculate wait time
- Sleep until market open time
### Telegram API Errors
- Log warning but continue trading
- Missing credentials → auto-disable notifications
- Network timeout → skip notification, no retry
- Invalid token → log error, trading unaffected
- Rate limit exceeded → queued via rate limiter
### Playbook Generation Failure
- Timeout → fall back to defensive playbook (`DEFENSIVE_PLAYBOOK_ON_FAILURE`)
- API error → use previous day's playbook if available
- No playbook → skip pre-market phase, fall back to direct AI calls
**Guarantee**: Notification and dashboard failures never interrupt trading operations.
- `Issue 4-1` Telegram 확장 명령 미구현 (`/report`, `/scenarios`, `/review`, `/dashboard`)
- `Issue 1-7` 일부 미완:
- `price_change_pct` 정규화 계층 명시 미흡
- 영향: `price_change_pct` 기반 조건은 현재 사실상 매칭되지 않음
- HOLD 시 별도 손절 모니터링 플래그 처리 미완
- US 스캐닝 확장(`fetch_overseas_rankings`) 미구현

View File

@@ -1,206 +1,82 @@
# Command Reference
## Common Command Failures
**Critical: Learn from failures. Never repeat the same failed command without modification.**
### tea CLI (Gitea Command Line Tool)
#### ❌ TTY Error - Interactive Confirmation Fails
```bash
~/bin/tea issues create --repo X --title "Y" --description "Z"
# Error: huh: could not open a new TTY: open /dev/tty: no such device or address
```
**💡 Reason:** tea tries to open `/dev/tty` for interactive confirmation prompts, which is unavailable in non-interactive environments.
**✅ Solution:** Use `YES=""` environment variable to bypass confirmation
```bash
YES="" ~/bin/tea issues create --repo jihoson/The-Ouroboros --title "Title" --description "Body"
YES="" ~/bin/tea issues edit <number> --repo jihoson/The-Ouroboros --description "Updated body"
YES="" ~/bin/tea pulls create --repo jihoson/The-Ouroboros --head feature-branch --base main --title "Title" --description "Body"
```
**📝 Notes:**
- Always set default login: `~/bin/tea login default local`
- Use `--repo jihoson/The-Ouroboros` when outside repo directory
- tea is preferred over direct Gitea API calls for consistency
#### ❌ Wrong Parameter Name
```bash
tea issues create --body "text"
# Error: flag provided but not defined: -body
```
**💡 Reason:** Parameter is `--description`, not `--body`.
**✅ Solution:** Use correct parameter name
```bash
YES="" ~/bin/tea issues create --description "text"
```
### Gitea API (Direct HTTP Calls)
#### ❌ Wrong Hostname
```bash
curl http://gitea.local:3000/api/v1/...
# Error: Could not resolve host: gitea.local
```
**💡 Reason:** Gitea instance runs on `localhost:3000`, not `gitea.local`.
**✅ Solution:** Use correct hostname (but prefer tea CLI)
```bash
curl http://localhost:3000/api/v1/repos/jihoson/The-Ouroboros/issues \
-H "Authorization: token $GITEA_TOKEN" \
-H "Content-Type: application/json" \
-d '{"title":"...", "body":"..."}'
```
**📝 Notes:**
- Prefer `tea` CLI over direct API calls
- Only use curl for operations tea doesn't support
### Git Commands
#### ❌ User Not Configured
```bash
git commit -m "message"
# Error: Author identity unknown
```
**💡 Reason:** Git user.name and user.email not set.
**✅ Solution:** Configure git user
```bash
git config user.name "agentson"
git config user.email "agentson@localhost"
```
#### ❌ Permission Denied on Push
```bash
git push origin branch
# Error: User permission denied for writing
```
**💡 Reason:** Repository access token lacks write permissions or user lacks repo write access.
**✅ Solution:**
1. Verify user has write access to repository (admin grants this)
2. Ensure git credential has correct token with `write:repository` scope
3. Check remote URL uses correct authentication
### Python/Pytest
#### ❌ Module Import Error
```bash
pytest tests/test_foo.py
# ModuleNotFoundError: No module named 'src'
```
**💡 Reason:** Package not installed in development mode.
**✅ Solution:** Install package with dev dependencies
```bash
pip install -e ".[dev]"
```
#### ❌ Async Test Hangs
```python
async def test_something(): # Hangs forever
result = await async_function()
```
**💡 Reason:** Missing pytest-asyncio or wrong configuration.
**✅ Solution:** Already configured in pyproject.toml
```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
```
No decorator needed for async tests.
## Build & Test Commands
## Core Runtime Commands
```bash
# Install all dependencies (production + dev)
pip install -e ".[dev]"
# Run full test suite with coverage (551 tests across 25 files)
pytest -v --cov=src --cov-report=term-missing
# Run a single test file
pytest tests/test_risk.py -v
# Run a single test by name
pytest tests/test_brain.py -k "test_parse_valid_json" -v
# Lint
ruff check src/ tests/
# Type check (strict mode, non-blocking in CI)
mypy src/ --strict
# Run the trading agent
# run (paper)
python -m src.main --mode=paper
# Run with dashboard enabled
# run with dashboard thread
python -m src.main --mode=paper --dashboard
# Docker
docker compose up -d ouroboros # Run agent
docker compose --profile test up test # Run tests in container
# tests
pytest -v --cov=src
# lint
ruff check src/ tests/
# type-check
mypy src/ --strict
```
## Dashboard
## Dashboard Runtime Controls
The FastAPI dashboard provides read-only monitoring of the trading system.
`Issue 4-3` 기준 반영:
### Starting the Dashboard
- CLI: `--dashboard`
- ENV: `DASHBOARD_ENABLED=true`
- Host/Port:
- `DASHBOARD_HOST` (default `127.0.0.1`)
- `DASHBOARD_PORT` (default `8080`)
## Telegram Commands (현재 구현)
`main.py` 등록 기준:
- `/help`
- `/status`
- `/positions`
- `/stop`
- `/resume`
## Telegram Commands (미구현 상태)
V2 플랜 `Issue 4-1` 항목은 아직 미구현:
- `/report [KR|US]`
- `/scenarios [KR|US]`
- `/review [KR|US]`
- `/dashboard`
## Gitea / tea Workflow Commands
이슈 선등록 후 작업 시작:
```bash
# Via CLI flag
python -m src.main --mode=paper --dashboard
# Via environment variable
DASHBOARD_ENABLED=true python -m src.main --mode=paper
YES="" ~/bin/tea issues create \
--repo jihoson/The-Ouroboros \
--title "..." \
--description "..."
```
Dashboard runs as a daemon thread on `DASHBOARD_HOST:DASHBOARD_PORT` (default: `127.0.0.1:8080`).
### API Endpoints
| Endpoint | Description |
|----------|-------------|
| `GET /` | HTML dashboard UI |
| `GET /api/status` | Daily trading status by market |
| `GET /api/playbook/{date}` | Playbook for specific date (query: `market`) |
| `GET /api/scorecard/{date}` | Daily scorecard from L6_DAILY context |
| `GET /api/performance` | Performance metrics by market and combined |
| `GET /api/context/{layer}` | Context data by layer L1-L7 (query: `timeframe`) |
| `GET /api/decisions` | Decision log entries (query: `limit`, `market`) |
| `GET /api/scenarios/active` | Today's matched scenarios |
## Telegram Commands
When `TELEGRAM_COMMANDS_ENABLED=true` (default), the bot accepts these interactive commands:
| Command | Description |
|---------|-------------|
| `/help` | List available commands |
| `/status` | Show trading status (mode, markets, P&L) |
| `/positions` | Display account summary (balance, cash, P&L) |
| `/report` | Daily summary metrics (trades, P&L, win rate) |
| `/scenarios` | Show today's playbook scenarios |
| `/review` | Display recent scorecards (L6_DAILY layer) |
| `/dashboard` | Show dashboard URL if enabled |
| `/stop` | Pause trading |
| `/resume` | Resume trading |
Commands are only processed from the authorized `TELEGRAM_CHAT_ID`.
## Environment Setup
작업은 `worktree` 기준 권장:
```bash
# Create .env file from example
cp .env.example .env
# Edit .env with your credentials
# Required: KIS_APP_KEY, KIS_APP_SECRET, KIS_ACCOUNT_NO, GEMINI_API_KEY
# Verify configuration
python -c "from src.config import Settings; print(Settings())"
git worktree add ../The-Ouroboros-issue-<N> feature/issue-<N>-<slug>
```
PR 생성:
```bash
YES="" ~/bin/tea pulls create \
--repo jihoson/The-Ouroboros \
--head feature/issue-<N>-<slug> \
--base main \
--title "..." \
--description "..."
```
## Known tea CLI Gotcha
TTY 없는 환경에서는 `tea` 확인 프롬프트가 실패할 수 있습니다.
항상 `YES=""`를 붙여 비대화식으로 실행하세요.

View File

@@ -1,243 +1,81 @@
# Context Tree: Multi-Layered Memory Management
The context tree implements **Pillar 2** of The Ouroboros: hierarchical memory management across 7 time horizons, from real-time market data to generational trading wisdom.
## Summary
## Overview
컨텍스트 트리는 L7(실시간)부터 L1(레거시)까지 계층화된 메모리 구조입니다.
Instead of a flat memory structure, The Ouroboros maintains a **7-tier context tree** where each layer represents a different time horizon and level of abstraction:
- L7~L5: 시장별 독립 데이터 중심
- L4~L1: 글로벌 포트폴리오 통합 데이터
```
L1 (Legacy) ← Cumulative wisdom across generations
L2 (Annual) ← Yearly performance metrics
L3 (Quarterly) ← Quarterly strategy adjustments
L4 (Monthly) ← Monthly portfolio rebalancing
L5 (Weekly) ← Weekly stock selection
L6 (Daily) ← Daily trade logs
L7 (Real-time) ← Live market data
```
## Layer Policy
Data flows **bottom-up**: real-time trades aggregate into daily summaries, which roll up to weekly, then monthly, quarterly, annual, and finally into permanent legacy knowledge.
### L7_REALTIME (시장+종목 스코프)
## The 7 Layers
- 주요 키 패턴:
- `volatility_{market}_{stock_code}`
- `price_{market}_{stock_code}`
- `rsi_{market}_{stock_code}`
- `volume_ratio_{market}_{stock_code}`
### L7: Real-time
**Retention**: 7 days
**Timeframe format**: `YYYY-MM-DD` (same-day)
**Content**: Current positions, live quotes, orderbook snapshots, tick-by-tick volatility
`trading_cycle()`에서 실시간으로 기록합니다.
**Use cases**:
- Immediate execution decisions
- Stop-loss triggers
- Real-time P&L tracking
### L6_DAILY (시장 스코프)
**Example keys**:
- `current_position_{stock_code}`: Current holdings
- `live_price_{stock_code}`: Latest quote
- `volatility_5m_{stock_code}`: 5-minute rolling volatility
EOD 집계 결과를 시장별 키로 저장합니다.
### L6: Daily
**Retention**: 90 days
**Timeframe format**: `YYYY-MM-DD`
**Content**: Daily trade logs, end-of-day P&L, market summaries, decision accuracy
- `trade_count_KR`, `buys_KR`, `sells_KR`, `holds_KR`
- `avg_confidence_US`, `total_pnl_US`, `win_rate_US`
- scorecard 저장 키: `scorecard_KR`, `scorecard_US`
**Use cases**:
- Daily performance review
- Identify patterns in recent trading
- Backtest strategy adjustments
### L5_WEEKLY
**Example keys**:
- `total_pnl`: Daily profit/loss
- `trade_count`: Number of trades
- `win_rate`: Percentage of profitable trades
- `avg_confidence`: Average Gemini confidence
L6 일일 데이터에서 시장별 주간 합계를 생성합니다.
### L5: Weekly
**Retention**: 1 year
**Timeframe format**: `YYYY-Www` (ISO week, e.g., `2026-W06`)
**Content**: Weekly stock selection, sector rotation, volatility regime classification
- `weekly_pnl_KR`, `weekly_pnl_US`
- `avg_confidence_KR`, `avg_confidence_US`
**Use cases**:
- Weekly strategy adjustment
- Sector momentum tracking
- Identify hot/cold markets
### L4_MONTHLY 이상
**Example keys**:
- `weekly_pnl`: Week's total P&L
- `top_performers`: Best-performing stocks
- `sector_focus`: Dominant sectors
- `avg_confidence`: Weekly average confidence
글로벌 통합 롤업입니다.
### L4: Monthly
**Retention**: 2 years
**Timeframe format**: `YYYY-MM`
**Content**: Monthly portfolio rebalancing, risk exposure analysis, drawdown recovery
- L5 → L4: `monthly_pnl`
- L4 → L3: `quarterly_pnl`
- L3 → L2: `annual_pnl`
- L2 → L1: `total_pnl`, `years_traded`, `avg_annual_pnl`
**Use cases**:
- Monthly performance reporting
- Risk exposure adjustment
- Correlation analysis
## Aggregation Flow
**Example keys**:
- `monthly_pnl`: Month's total P&L
- `sharpe_ratio`: Risk-adjusted return
- `max_drawdown`: Largest peak-to-trough decline
- `rebalancing_notes`: Manual insights
- EOD: `ContextAggregator.aggregate_daily_from_trades(date, market)`
- 주기 롤업: `ContextScheduler.run_if_due()`
### L3: Quarterly
**Retention**: 3 years
**Timeframe format**: `YYYY-Qn` (e.g., `2026-Q1`)
**Content**: Quarterly strategy pivots, market phase detection (bull/bear/sideways), macro regime changes
`ContextScheduler`는 다음을 처리합니다.
**Use cases**:
- Strategic pivots (e.g., growth → value)
- Macro regime classification
- Long-term pattern recognition
**Example keys**:
- `quarterly_pnl`: Quarter's total P&L
- `market_phase`: Bull/Bear/Sideways
- `strategy_adjustments`: Major changes made
- `lessons_learned`: Key insights
### L2: Annual
**Retention**: 10 years
**Timeframe format**: `YYYY`
**Content**: Yearly returns, Sharpe ratio, max drawdown, win rate, strategy effectiveness
**Use cases**:
- Annual performance review
- Multi-year trend analysis
- Strategy benchmarking
**Example keys**:
- `annual_pnl`: Year's total P&L
- `sharpe_ratio`: Annual risk-adjusted return
- `win_rate`: Yearly win percentage
- `best_strategy`: Most successful strategy
- `worst_mistake`: Biggest lesson learned
### L1: Legacy
**Retention**: Forever
**Timeframe format**: `LEGACY` (single timeframe)
**Content**: Cumulative trading history, core principles, generational wisdom
**Use cases**:
- Long-term philosophy
- Foundational rules
- Lessons that transcend market cycles
**Example keys**:
- `total_pnl`: All-time profit/loss
- `years_traded`: Trading longevity
- `avg_annual_pnl`: Long-term average return
- `core_principles`: Immutable trading rules
- `greatest_trades`: Hall of fame
- `never_again`: Permanent warnings
- weekly/monthly/quarterly/annual/legacy 집계
- 일 1회 `ContextStore.cleanup_expired_contexts()` 실행
- 동일 날짜 중복 실행 방지(`_last_run`)
## Usage
### Setting Context
```python
from src.context import ContextLayer, ContextStore
from src.db import init_db
from datetime import UTC, datetime
conn = init_db("data/ouroboros.db")
store = ContextStore(conn)
# Store daily P&L
store.set_context(
layer=ContextLayer.L6_DAILY,
timeframe="2026-02-04",
key="total_pnl",
value=1234.56
)
# Store weekly insight
store.set_context(
layer=ContextLayer.L5_WEEKLY,
timeframe="2026-W06",
key="top_performers",
value=["005930", "000660", "035720"] # JSON-serializable
)
# Store legacy wisdom
store.set_context(
layer=ContextLayer.L1_LEGACY,
timeframe="LEGACY",
key="core_principles",
value=[
"Cut losses fast",
"Let winners run",
"Never average down on losing positions"
]
)
```
### Retrieving Context
```python
# Get a specific value
pnl = store.get_context(ContextLayer.L6_DAILY, "2026-02-04", "total_pnl")
# Returns: 1234.56
# Get all keys for a timeframe
daily_summary = store.get_all_contexts(ContextLayer.L6_DAILY, "2026-02-04")
# Returns: {"total_pnl": 1234.56, "trade_count": 10, "win_rate": 60.0, ...}
# Get all data for a layer (any timeframe)
all_daily = store.get_all_contexts(ContextLayer.L6_DAILY)
# Returns: {"total_pnl": 1234.56, "trade_count": 10, ...} (latest timeframes first)
# Get the latest timeframe
latest = store.get_latest_timeframe(ContextLayer.L6_DAILY)
# Returns: "2026-02-04"
```
### Automatic Aggregation
The `ContextAggregator` rolls up data from lower to higher layers:
```python
from src.context.aggregator import ContextAggregator
from src.context.scheduler import ContextScheduler
aggregator = ContextAggregator(conn)
scheduler = ContextScheduler(aggregator=aggregator, store=context_store)
# Aggregate daily metrics from trades
aggregator.aggregate_daily_from_trades("2026-02-04")
# EOD market-scoped daily aggregation
aggregator.aggregate_daily_from_trades(date="2026-02-16", market="KR")
# Roll up weekly from daily
aggregator.aggregate_weekly_from_daily("2026-W06")
# Roll up all layers at once (bottom-up)
aggregator.run_all_aggregations()
# Run scheduled rollups when due
scheduler.run_if_due(now=datetime.now(UTC))
```
**Aggregation schedule** (recommended):
- **L7 → L6**: Every midnight (daily rollup)
- **L6 → L5**: Every Sunday (weekly rollup)
- **L5 → L4**: First day of each month (monthly rollup)
- **L4 → L3**: First day of quarter (quarterly rollup)
- **L3 → L2**: January 1st (annual rollup)
- **L2 → L1**: On demand (major milestones)
## Retention
### Context Cleanup
`src/context/layer.py` 기준:
Expired contexts are automatically deleted based on retention policies:
```python
# Manual cleanup
deleted = store.cleanup_expired_contexts()
# Returns: {ContextLayer.L7_REALTIME: 42, ContextLayer.L6_DAILY: 15, ...}
```
**Retention policies** (defined in `src/context/layer.py`):
- L1: Forever
- L2: 10 years
- L3: 3 years
@@ -246,93 +84,8 @@ deleted = store.cleanup_expired_contexts()
- L6: 90 days
- L7: 7 days
## Integration with Gemini Brain
## Current Notes (2026-02-16)
The context tree provides hierarchical memory for decision-making:
```python
from src.brain.gemini_client import GeminiClient
# Build prompt with multi-layer context
def build_enhanced_prompt(stock_code: str, store: ContextStore) -> str:
# L7: Real-time data
current_price = store.get_context(ContextLayer.L7_REALTIME, "2026-02-04", f"live_price_{stock_code}")
# L6: Recent daily performance
yesterday_pnl = store.get_context(ContextLayer.L6_DAILY, "2026-02-03", "total_pnl")
# L5: Weekly trend
weekly_data = store.get_all_contexts(ContextLayer.L5_WEEKLY, "2026-W06")
# L1: Core principles
principles = store.get_context(ContextLayer.L1_LEGACY, "LEGACY", "core_principles")
return f"""
Analyze {stock_code} for trading decision.
Current price: {current_price}
Yesterday's P&L: {yesterday_pnl}
This week: {weekly_data}
Core principles:
{chr(10).join(f'- {p}' for p in principles)}
Decision (BUY/SELL/HOLD):
"""
```
## Database Schema
```sql
-- Context storage
CREATE TABLE contexts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
layer TEXT NOT NULL, -- L1_LEGACY, L2_ANNUAL, ..., L7_REALTIME
timeframe TEXT NOT NULL, -- "LEGACY", "2026", "2026-Q1", "2026-02", "2026-W06", "2026-02-04"
key TEXT NOT NULL, -- "total_pnl", "win_rate", "core_principles", etc.
value TEXT NOT NULL, -- JSON-serialized value
created_at TEXT NOT NULL, -- ISO 8601 timestamp
updated_at TEXT NOT NULL, -- ISO 8601 timestamp
UNIQUE(layer, timeframe, key)
);
-- Layer metadata
CREATE TABLE context_metadata (
layer TEXT PRIMARY KEY,
description TEXT NOT NULL,
retention_days INTEGER, -- NULL = keep forever
aggregation_source TEXT -- Parent layer for rollup
);
-- Indices for fast queries
CREATE INDEX idx_contexts_layer ON contexts(layer);
CREATE INDEX idx_contexts_timeframe ON contexts(timeframe);
CREATE INDEX idx_contexts_updated ON contexts(updated_at);
```
## Best Practices
1. **Write to leaf layers only** — Never manually write to L1-L5; let aggregation populate them
2. **Aggregate regularly** — Schedule aggregation jobs to keep higher layers fresh
3. **Query specific timeframes** — Use `get_context(layer, timeframe, key)` for precise retrieval
4. **Clean up periodically** — Run `cleanup_expired_contexts()` weekly to free space
5. **Preserve L1 forever** — Legacy wisdom should never expire
6. **Use JSON-serializable values** — Store dicts, lists, strings, numbers (not custom objects)
## Testing
See `tests/test_context.py` for comprehensive test coverage (18 tests, 100% coverage on context modules).
```bash
pytest tests/test_context.py -v
```
## References
- **Implementation**: `src/context/`
- `layer.py`: Layer definitions and metadata
- `store.py`: CRUD operations
- `aggregator.py`: Bottom-up aggregation logic
- **Database**: `src/db.py` (table initialization)
- **Tests**: `tests/test_context.py`
- **Related**: Pillar 2 (Multi-layered Context Management)
- L7 쓰기와 L6 시장별 집계는 `main.py`에 연결됨
- scheduler 기반 cleanup/rollup도 연결됨
- cross-market scorecard 조회는 `PreMarketPlanner`에서 사용 중

View File

@@ -91,178 +91,43 @@
## 2026-02-16
### 문서 v2 동기화 (전체 문서 현행화)
### V2 진행상태 재정렬 + 문서 동기화
**배경:**
- v2 기능 구현 완료 후 문서가 실제 코드 상태와 크게 괴리
- 문서에는 54 tests / 4 files로 기록되었으나 실제로는 551 tests / 25 files
- v2 핵심 기능(Playbook, Scenario Engine, Dashboard, Telegram Commands, Daily Review, Context System, Backup) 문서화 누락
- V2 이슈 다수가 병렬로 진행되며 구현/문서 간 상태 불일치가 발생
- 사용자 요청으로 "현재 코드 기준 사실"에 맞춘 전면 문서 갱신 필요
**요구사항:**
1. `docs/testing.md` — 551 tests / 25 files 반영, 전체 테스트 파일 설명
2. `docs/architecture.md` — v2 컴포넌트(Strategy, Context, Dashboard, Decision Logger 등) 추가, Playbook Mode 데이터 플로우, DB 스키마 5개 테이블, v2 환경변수
3. `docs/commands.md` — Dashboard 실행 명령어, Telegram 명령어 9종 레퍼런스
4. `CLAUDE.md` — Project Structure 트리 확장, 테스트 수 업데이트, `--dashboard` 플래그
5. `docs/skills.md` — DB 파일명 `trades.db`로 통일, Dashboard 명령어 추가
6. 기존에 유효한 트러블슈팅, 코드 예제 등은 유지
**확인된 상태(코드 기준):**
- 완료: 18/20
- 부분 완료: `1-7`
- 미완료: `4-1`
**구현 결과:**
- 6개 문서 파일 업데이트
- 이전 시도(2개 커밋)는 기존 내용을 과도하게 삭제하여 폐기, main 기준으로 재작업
**핵심 반영 사항:**
1. 대시보드 실행 통합(`Issue 4-3`) 반영
- `--dashboard` 플래그
- `DASHBOARD_ENABLED`, `DASHBOARD_HOST`, `DASHBOARD_PORT`
2. 컨텍스트 스케줄러 및 시장 스코프 키 정책 반영
3. scorecard/review/evolution 연결 상태 반영
4. 미완료 갭 명시
- Telegram 확장 명령어(`4-1`) 미구현
- `1-7` 잔여 항목(키 정규화/HOLD 손절 모니터링/US 코드 정합성)
**이슈/PR:** #131, PR #134
**프로세스 요구사항 강화:**
- 모든 문서 작업도 Gitea 이슈 선등록 후 진행
- 병렬 작업 후 상태 정합성 점검 결과를 `requirements-log`에 기록
### 해외 스캐너 개선: 랭킹 연동 + 변동성 우선 선별
**이슈/브랜치:**
- Issue: #131
- Branch(worktree): `feature/issue-131-docs-v2-status-sync`
**배경:**
- `run_overnight` 실운영에서 미국장 동안 거래가 0건 지속
- 원인: 해외 시장에서도 국내 랭킹/일봉 API 경로를 사용하던 구조적 불일치
### 문서 보강 2차 (리뷰 반영)
**요구사항:**
1. 해외 시장도 랭킹 API 기반 유니버스 탐색 지
2. 단순 상승률/거래대금 상위가 아니라, **변동성이 큰 종목**을 우선 선별
3. 고정 티커 fallback 금지
**리뷰 피드백 반영:**
- README에 Quick Start(환경설정/설치/검증) 복
- architecture에 RiskManager/에러 처리/설정 레퍼런스 복원
- testing 문서에 기존 핵심 테스트 파일 설명 복원
- 시장 코드 불일치(`KR,US` vs `US_NASDAQ/US_NYSE`)를 "런타임 영향"으로 격상 명시
- `price_change_pct` 누락 영향(조건 dead path)을 명시
**구현 결과:**
- `src/broker/overseas.py`
- `fetch_overseas_rankings()` 추가 (fluctuation / volume)
- 해외 랭킹 API 경로/TR_ID를 설정값으로 오버라이드 가능하게 구현
- `src/analysis/smart_scanner.py`
- market-aware 스캔(국내/해외 분리)
- 해외: 랭킹 API 유니버스 + 변동성 우선 점수(일변동률 vs 장중 고저폭)
- 거래대금/거래량 랭킹은 유동성 보정 점수로 활용
- 랭킹 실패 시에는 동적 유니버스(active/recent/holdings)만 사용
- `src/config.py`
- `OVERSEAS_RANKING_*` 설정 추가
**효과:**
- 해외 시장에서 스캐너 후보 0개로 정지되는 상황 완화
- 종목 선정 기준이 단순 상승률 중심에서 변동성 중심으로 개선
- 고정 티커 없이도 시장 주도 변동 종목 탐지 가능
### 국내 스캐너/주문수량 정렬: 변동성 우선 + 리스크 타기팅
**배경:**
- 해외만 변동성 우선으로 동작하고, 국내는 RSI/거래량 필터 중심으로 동작해 시장 간 전략 일관성이 낮았음
- 매수 수량이 고정 1주라서 변동성 구간별 익스포저 관리가 어려웠음
**요구사항:**
1. 국내 스캐너도 변동성 우선 선별로 해외와 통일
2. 고변동 종목일수록 포지션 크기를 줄이는 수량 산식 적용
**구현 결과:**
- `src/analysis/smart_scanner.py`
- 국내: `fluctuation ranking + volume ranking bonus` 기반 점수화로 전환
- 점수는 `max(abs(change_rate), intraday_range_pct)` 중심으로 계산
- 국내 랭킹 응답 스키마 키(`price`, `change_rate`, `volume`) 파싱 보강
- `src/main.py`
- `_determine_order_quantity()` 추가
- BUY 시 변동성 점수 기반 동적 수량 산정 적용
- `trading_cycle`, `run_daily_session` 경로 모두 동일 수량 로직 사용
- `src/config.py`
- `POSITION_SIZING_*` 설정 추가
**효과:**
- 국내/해외 스캐너 기준이 변동성 중심으로 일관화
- 고변동 구간에서 자동 익스포저 축소, 저변동 구간에서 과소진입 완화
## 2026-02-18
### KIS 해외 랭킹 API 404 에러 수정
**배경:**
- KIS 해외주식 랭킹 API(`fetch_overseas_rankings`)가 모든 거래소에서 HTTP 404를 반환
- Smart Scanner가 해외 시장 후보 종목을 찾지 못해 거래가 전혀 실행되지 않음
**근본 원인:**
- TR_ID, API 경로, 거래소 코드가 모두 KIS 공식 문서와 불일치
**구현 결과:**
- `src/config.py`: TR_ID/Path 기본값을 KIS 공식 스펙으로 수정
- `src/broker/overseas.py`: 랭킹 API 전용 거래소 코드 매핑 추가 (NASD→NAS, NYSE→NYS, AMEX→AMS), 올바른 API 파라미터 사용
- `tests/test_overseas_broker.py`: 19개 단위 테스트 추가
**효과:**
- 해외 시장 랭킹 스캔이 정상 동작하여 Smart Scanner가 후보 종목 탐지 가능
### Gemini prompt_override 미적용 버그 수정
**배경:**
- `run_overnight` 실행 시 모든 시장에서 Playbook 생성 실패 (`JSONDecodeError`)
- defensive playbook으로 폴백되어 모든 종목이 HOLD 처리
**근본 원인:**
- `pre_market_planner.py``market_data["prompt_override"]`에 Playbook 전용 프롬프트를 넣어 `gemini.decide()` 호출
- `gemini_client.py``decide()` 메서드가 `prompt_override` 키를 전혀 확인하지 않고 항상 일반 트레이드 결정 프롬프트 생성
- Gemini가 Playbook JSON 대신 일반 트레이드 결정을 반환하여 파싱 실패
**구현 결과:**
- `src/brain/gemini_client.py`: `decide()` 메서드에서 `prompt_override` 우선 사용 로직 추가
- `tests/test_brain.py`: 3개 테스트 추가 (override 전달, optimization 우회, 미지정 시 기존 동작 유지)
**이슈/PR:** #143
### 미국장 거래 미실행 근본 원인 분석 및 수정 (자율 실행 세션)
**배경:**
- 사용자 요청: "미국장 열면 프로그램 돌려서 거래 한 번도 못 한 거 꼭 원인 찾아서 해결해줘"
- 프로그램을 미국장 개장(9:30 AM EST) 전부터 실행하여 실시간 로그를 분석
**발견된 근본 원인 #1: Defensive Playbook — BUY 조건 없음**
- Gemini free tier (20 RPD) 소진 → `generate_playbook()` 실패 → `_defensive_playbook()` 폴백
- Defensive playbook은 `price_change_pct_below: -3.0 → SELL` 조건만 존재, BUY 조건 없음
- ScenarioEngine이 항상 HOLD 반환 → 거래 0건
**수정 #1 (PR #146, Issue #145):**
- `src/strategy/pre_market_planner.py`: `_smart_fallback_playbook()` 메서드 추가
- 스캐너 signal 기반 BUY 조건 생성: `momentum → volume_ratio_above`, `oversold → rsi_below`
- 기존 defensive stop-loss SELL 조건 유지
- Gemini 실패 시 defensive → smart fallback으로 전환
- 테스트 10개 추가
**발견된 근본 원인 #2: 가격 API 거래소 코드 불일치 + VTS 잔고 API 오류**
실제 로그:
```
Scenario matched for MRNX: BUY (confidence=80) ✓
Decision for EWUS (NYSE American): BUY (confidence=80) ✓
Skip BUY APLZ (NYSE American): no affordable quantity (cash=0.00, price=0.00) ✗
```
- `get_overseas_price()`: `NASD`/`NYSE`/`AMEX` 전송 → API가 `NAS`/`NYS`/`AMS` 기대 → 빈 응답 → `price=0`
- `VTTS3012R` 잔고 API: "ERROR : INPUT INVALID_CHECK_ACNO" → `total_cash=0`
- 결과: `_determine_order_quantity()` 가 0 반환 → 주문 건너뜀
**수정 #2 (PR #148, Issue #147):**
- `src/broker/overseas.py`: `_PRICE_EXCHANGE_MAP = _RANKING_EXCHANGE_MAP` 추가, 가격 API에 매핑 적용
- `src/config.py`: `PAPER_OVERSEAS_CASH: float = Field(default=50000.0)` — paper 모드 시뮬레이션 잔고
- `src/main.py`: 잔고 0일 때 PAPER_OVERSEAS_CASH 폴백, 가격 0일 때 candidate.price 폴백
- 테스트 8개 추가
**효과:**
- BUY 결정 → 실제 주문 전송까지의 파이프라인이 완전히 동작
- Paper 모드에서 KIS VTS 해외 잔고 API 오류에 관계없이 시뮬레이션 거래 가능
**이슈/PR:** #145, #146, #147, #148
### 해외주식 시장가 주문 거부 수정 (Fix #3, 연속 발견)
**배경:**
- Fix #147 적용 후 주문 전송 시작 → KIS VTS가 거부: "지정가만 가능한 상품입니다"
**근본 원인:**
- `trading_cycle()`, `run_daily_session()` 양쪽에서 `send_overseas_order(price=0.0)` 하드코딩
- `price=0``ORD_DVSN="01"` (시장가) 전송 → KIS VTS 거부
- Fix #147에서 이미 `current_price`를 올바르게 계산했으나 주문 시 미사용
**구현 결과:**
- `src/main.py`: 두 곳에서 `price=0.0``price=current_price`/`price=stock_data["current_price"]`
- `tests/test_main.py`: 회귀 테스트 `test_overseas_buy_order_uses_limit_price` 추가
**최종 확인 로그:**
```
Order result: 모의투자 매수주문이 완료 되었습니다. ✓
```
**이슈/PR:** #149, #150
**의도:**
- V2 상태 반영과 기존 온보딩/운영 문서 가치를 동시에 유지

View File

@@ -34,12 +34,6 @@ python -m src.main --mode=paper
```
Runs the agent in paper-trading mode (no real orders).
### Start Trading Agent with Dashboard
```bash
python -m src.main --mode=paper --dashboard
```
Runs the agent with FastAPI dashboard on `127.0.0.1:8080` (configurable via `DASHBOARD_HOST`/`DASHBOARD_PORT`).
### Start Trading Agent (Production)
```bash
docker compose up -d ouroboros
@@ -65,7 +59,7 @@ Analyze the last 30 days of trade logs and generate performance metrics.
python -m src.evolution.optimizer --evolve
```
Triggers the evolution engine to:
1. Analyze `trades.db` for failing patterns
1. Analyze `trade_logs.db` for failing patterns
2. Ask Gemini to generate a new strategy
3. Run tests on the new strategy
4. Create a PR if tests pass
@@ -97,12 +91,12 @@ curl http://localhost:8080/health
### View Trade Logs
```bash
sqlite3 data/trades.db "SELECT * FROM trades ORDER BY timestamp DESC LIMIT 20;"
sqlite3 data/trade_logs.db "SELECT * FROM trades ORDER BY timestamp DESC LIMIT 20;"
```
### Export Trade History
```bash
sqlite3 -header -csv data/trades.db "SELECT * FROM trades;" > trades_export.csv
sqlite3 -header -csv data/trade_logs.db "SELECT * FROM trades;" > trades_export.csv
```
## Safety Checklist (Pre-Deploy)

View File

@@ -1,287 +1,63 @@
# Testing Guidelines
## Test Structure
## Current Test Baseline (2026-02-16)
**551 tests** across **25 files**. `asyncio_mode = "auto"` in pyproject.toml — async tests need no special decorator.
The `settings` fixture in `conftest.py` provides safe defaults with test credentials and in-memory DB.
### Test Files
#### Core Components
##### `tests/test_risk.py` (14 tests)
- Circuit breaker boundaries and exact threshold triggers
- Fat-finger edge cases and percentage validation
- P&L calculation edge cases
- Order validation logic
##### `tests/test_broker.py` (11 tests)
- OAuth token lifecycle
- Rate limiting enforcement
- Hash key generation
- Network error handling
- SSL context configuration
##### `tests/test_brain.py` (24 tests)
- Valid JSON parsing and markdown-wrapped JSON handling
- Malformed JSON fallback
- Missing fields handling
- Invalid action validation
- Confidence threshold enforcement
- Empty response handling
- Prompt construction for different markets
##### `tests/test_market_schedule.py` (24 tests)
- Market open/close logic
- Timezone handling (UTC, Asia/Seoul, America/New_York, etc.)
- DST (Daylight Saving Time) transitions
- Weekend handling and lunch break logic
- Multiple market filtering
- Next market open calculation
##### `tests/test_db.py` (3 tests)
- Database initialization and table creation
- Trade logging with all fields (market, exchange_code, decision_id)
- Query and retrieval operations
##### `tests/test_main.py` (37 tests)
- Trading loop orchestration
- Market iteration and stock processing
- Dashboard integration (`--dashboard` flag)
- Telegram command handler wiring
- Error handling and graceful shutdown
#### Strategy & Playbook (v2)
##### `tests/test_pre_market_planner.py` (37 tests)
- Pre-market playbook generation
- Gemini API integration for scenario creation
- Timeout handling and defensive playbook fallback
- Multi-market playbook generation
##### `tests/test_scenario_engine.py` (44 tests)
- Scenario matching against live market data
- Confidence scoring and threshold filtering
- Multiple scenario type handling
- Edge cases (no match, partial match, expired scenarios)
##### `tests/test_playbook_store.py` (23 tests)
- Playbook persistence to SQLite
- Date-based retrieval and market filtering
- Playbook status management (generated, active, expired)
- JSON serialization/deserialization
##### `tests/test_strategy_models.py` (33 tests)
- Pydantic model validation for scenarios, playbooks, decisions
- Field constraints and default values
- Serialization round-trips
#### Analysis & Scanning
##### `tests/test_volatility.py` (24 tests)
- ATR and RSI calculation accuracy
- Volume surge ratio computation
- Momentum scoring
- Breakout/breakdown pattern detection
- Market scanner watchlist management
##### `tests/test_smart_scanner.py` (13 tests)
- Python-first filtering pipeline
- RSI and volume ratio filter logic
- Candidate scoring and ranking
- Fallback to static watchlist
#### Context & Memory
##### `tests/test_context.py` (18 tests)
- L1-L7 layer storage and retrieval
- Context key-value CRUD operations
- Timeframe-based queries
- Layer metadata management
##### `tests/test_context_scheduler.py` (5 tests)
- Periodic context aggregation scheduling
- Layer summarization triggers
#### Evolution & Review
##### `tests/test_evolution.py` (24 tests)
- Strategy optimization loop
- High-confidence losing trade analysis
- Generated strategy validation
##### `tests/test_daily_review.py` (10 tests)
- End-of-day review generation
- Trade performance summarization
- Context layer (L6_DAILY) integration
##### `tests/test_scorecard.py` (3 tests)
- Daily scorecard metrics calculation
- Win rate, P&L, confidence tracking
#### Notifications & Commands
##### `tests/test_telegram.py` (25 tests)
- Message sending and formatting
- Rate limiting (leaky bucket)
- Error handling (network timeout, invalid token)
- Auto-disable on missing credentials
- Notification types (trade, circuit breaker, fat-finger, market events)
##### `tests/test_telegram_commands.py` (31 tests)
- 9 command handlers (/help, /status, /positions, /report, /scenarios, /review, /dashboard, /stop, /resume)
- Long polling and command dispatch
- Authorization filtering by chat_id
- Command response formatting
#### Dashboard
##### `tests/test_dashboard.py` (14 tests)
- FastAPI endpoint responses (8 API routes)
- Status, playbook, scorecard, performance, context, decisions, scenarios
- Query parameter handling (market, date, limit)
#### Performance & Quality
##### `tests/test_token_efficiency.py` (34 tests)
- Gemini token usage optimization
- Prompt size reduction verification
- Cache effectiveness
##### `tests/test_latency_control.py` (30 tests)
- API call latency measurement
- Rate limiter timing accuracy
- Async operation overhead
##### `tests/test_decision_logger.py` (9 tests)
- Decision audit trail completeness
- Context snapshot capture
- Outcome tracking (P&L, accuracy)
##### `tests/test_data_integration.py` (38 tests)
- External data source integration
- News API, market data, economic calendar
- Error handling for API failures
##### `tests/test_backup.py` (23 tests)
- Backup scheduler and execution
- Cloud storage (S3) upload
- Health monitoring
- Data export functionality
## Coverage Requirements
**Minimum coverage: 80%**
Check coverage:
```bash
pytest -v --cov=src --cov-report=term-missing
```
**Note:** `main.py` has lower coverage as it contains the main loop which is tested via integration/manual testing.
## Test Configuration
### `pyproject.toml`
```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
python_files = ["test_*.py"]
```
### `tests/conftest.py`
```python
@pytest.fixture
def settings() -> Settings:
"""Provide test settings with safe defaults."""
return Settings(
KIS_APP_KEY="test_key",
KIS_APP_SECRET="test_secret",
KIS_ACCOUNT_NO="12345678-01",
GEMINI_API_KEY="test_gemini_key",
MODE="paper",
DB_PATH=":memory:", # In-memory SQLite
CONFIDENCE_THRESHOLD=80,
ENABLED_MARKETS="KR",
)
```
## Writing New Tests
### Naming Convention
- Test files: `test_<module>.py`
- Test functions: `test_<feature>_<scenario>()`
- Use descriptive names that explain what is being tested
### Good Test Example
```python
async def test_send_order_with_market_price(broker, settings):
"""Market orders should use price=0 and ORD_DVSN='01'."""
# Arrange
stock_code = "005930"
order_type = "BUY"
quantity = 10
# Act
with patch.object(broker._session, 'post') as mock_post:
mock_post.return_value.__aenter__.return_value.status = 200
mock_post.return_value.__aenter__.return_value.json = AsyncMock(
return_value={"rt_cd": "0", "msg1": "OK"}
)
await broker.send_order(stock_code, order_type, quantity, price=0)
# Assert
call_args = mock_post.call_args
body = call_args.kwargs['json']
assert body['ORD_DVSN'] == '01' # Market order
assert body['ORD_UNPR'] == '0' # Price 0
```
### Test Checklist
- [ ] Test passes in isolation (`pytest tests/test_foo.py::test_bar -v`)
- [ ] Test has clear docstring explaining what it tests
- [ ] Arrange-Act-Assert structure
- [ ] Uses appropriate fixtures from conftest.py
- [ ] Mocks external dependencies (API calls, network)
- [ ] Tests edge cases and error conditions
- [ ] Doesn't rely on test execution order
## Running Tests
수집 기준:
```bash
# All tests
pytest -v
# Specific file
pytest tests/test_risk.py -v
# Specific test
pytest tests/test_brain.py::test_parse_valid_json -v
# With coverage
pytest -v --cov=src --cov-report=term-missing
# Stop on first failure
pytest -x
# Verbose output with print statements
pytest -v -s
pytest --collect-only -q
# 538 tests collected
```
## CI/CD Integration
V2 핵심 영역 테스트가 포함되어 있습니다.
Tests run automatically on:
- Every commit to feature branches
- Every PR to main
- Scheduled daily runs
- `tests/test_strategy_models.py`
- `tests/test_pre_market_planner.py`
- `tests/test_scenario_engine.py`
- `tests/test_playbook_store.py`
- `tests/test_context_scheduler.py`
- `tests/test_daily_review.py`
- `tests/test_scorecard.py`
- `tests/test_dashboard.py`
- `tests/test_main.py`
**Blocking conditions:**
- Test failures → PR blocked
- Coverage < 80% → PR blocked (warning only for main.py)
기존 핵심 영역 테스트도 유지됩니다.
**Non-blocking:**
- `mypy --strict` errors (type hints encouraged but not enforced)
- `ruff check` warnings (must be acknowledged)
- `tests/test_risk.py`: circuit breaker/fat-finger 안전장치 검증
- `tests/test_broker.py`: KIS API 호출/에러 처리/인증 흐름 검증
- `tests/test_brain.py`: Gemini 응답 파싱/신뢰도 게이트 검증
- `tests/test_market_schedule.py`: 시장 오픈/클로즈/타임존 로직 검증
## Required Checks
```bash
pytest -v --cov=src
ruff check src/ tests/
mypy src/ --strict
```
## FastAPI Note
대시보드 테스트(`tests/test_dashboard.py`)는 `fastapi`가 환경에 없으면 skip될 수 있습니다.
의도된 동작이며 CI/개발환경에서 의존성 설치 여부를 확인하세요.
## Targeted Smoke Commands
```bash
# dashboard integration
pytest -q tests/test_main.py -k "dashboard"
# planner/scenario/review paths
pytest -q tests/test_pre_market_planner.py tests/test_scenario_engine.py tests/test_daily_review.py
# context rollup/scheduler
pytest -q tests/test_context.py tests/test_context_scheduler.py
```
## Review Checklist (테스트 관점)
- 플랜 항목별 테스트 존재 여부 확인
- 시장 스코프 키(`*_KR`, `*_US`) 검증 확인
- EOD 흐름(`aggregate_daily_from_trades`, `scorecard_{market}` 저장) 검증
- decision outcome 연결(`decision_id`) 검증
- 대시보드 API market filter 검증

View File

@@ -8,8 +8,9 @@
2. **Create Feature Branch** — Branch from `main` using format `feature/issue-{N}-{short-description}`
- After creating the branch, run `git pull origin main` and rebase to ensure the branch is up to date
3. **Implement Changes** — Write code, tests, and documentation on the feature branch
4. **Create Pull Request** — Submit PR to `main` branch referencing the issue number
5. **Review & Merge** — After approval, merge via PR (squash or merge commit)
4. **Sync Status Docs** — Before PR, update `README.md` and relevant `docs/*.md` so implementation status/gaps are explicit
5. **Create Pull Request** — Submit PR to `main` branch referencing the issue number
6. **Review & Merge** — After approval, merge via PR (squash or merge commit)
**Never commit directly to `main`.** This policy applies to all changes, no exceptions.

View File

@@ -1,54 +0,0 @@
#!/usr/bin/env bash
# Morning summary for overnight run logs.
set -euo pipefail
LOG_DIR="${LOG_DIR:-data/overnight}"
if [ ! -d "$LOG_DIR" ]; then
echo "로그 디렉터리가 없습니다: $LOG_DIR"
exit 1
fi
latest_run="$(ls -1t "$LOG_DIR"/run_*.log 2>/dev/null | head -n 1 || true)"
latest_watchdog="$(ls -1t "$LOG_DIR"/watchdog_*.log 2>/dev/null | head -n 1 || true)"
if [ -z "$latest_run" ]; then
echo "run 로그가 없습니다: $LOG_DIR/run_*.log"
exit 1
fi
echo "Overnight report"
echo "- run log: $latest_run"
if [ -n "$latest_watchdog" ]; then
echo "- watchdog log: $latest_watchdog"
fi
start_line="$(head -n 1 "$latest_run" || true)"
end_line="$(tail -n 1 "$latest_run" || true)"
info_count="$(rg -c '"level": "INFO"' "$latest_run" || true)"
warn_count="$(rg -c '"level": "WARNING"' "$latest_run" || true)"
error_count="$(rg -c '"level": "ERROR"' "$latest_run" || true)"
critical_count="$(rg -c '"level": "CRITICAL"' "$latest_run" || true)"
traceback_count="$(rg -c 'Traceback' "$latest_run" || true)"
echo "- start: ${start_line:-N/A}"
echo "- end: ${end_line:-N/A}"
echo "- INFO: ${info_count:-0}"
echo "- WARNING: ${warn_count:-0}"
echo "- ERROR: ${error_count:-0}"
echo "- CRITICAL: ${critical_count:-0}"
echo "- Traceback: ${traceback_count:-0}"
if [ -n "$latest_watchdog" ]; then
watchdog_errors="$(rg -c '\[ERROR\]' "$latest_watchdog" || true)"
echo "- watchdog ERROR: ${watchdog_errors:-0}"
echo ""
echo "최근 watchdog 로그:"
tail -n 5 "$latest_watchdog" || true
fi
echo ""
echo "최근 앱 로그:"
tail -n 20 "$latest_run" || true

View File

@@ -1,87 +0,0 @@
#!/usr/bin/env bash
# Start The Ouroboros overnight with logs and watchdog.
set -euo pipefail
LOG_DIR="${LOG_DIR:-data/overnight}"
CHECK_INTERVAL="${CHECK_INTERVAL:-30}"
TMUX_AUTO="${TMUX_AUTO:-true}"
TMUX_ATTACH="${TMUX_ATTACH:-true}"
TMUX_SESSION_PREFIX="${TMUX_SESSION_PREFIX:-ouroboros_overnight}"
if [ -z "${APP_CMD:-}" ]; then
if [ -x ".venv/bin/python" ]; then
PYTHON_BIN=".venv/bin/python"
elif command -v python3 >/dev/null 2>&1; then
PYTHON_BIN="python3"
elif command -v python >/dev/null 2>&1; then
PYTHON_BIN="python"
else
echo ".venv/bin/python 또는 python3/python 실행 파일을 찾을 수 없습니다."
exit 1
fi
dashboard_port="${DASHBOARD_PORT:-8080}"
APP_CMD="DASHBOARD_PORT=$dashboard_port $PYTHON_BIN -m src.main --mode=paper --dashboard"
fi
mkdir -p "$LOG_DIR"
timestamp="$(date +"%Y%m%d_%H%M%S")"
RUN_LOG="$LOG_DIR/run_${timestamp}.log"
WATCHDOG_LOG="$LOG_DIR/watchdog_${timestamp}.log"
PID_FILE="$LOG_DIR/app.pid"
WATCHDOG_PID_FILE="$LOG_DIR/watchdog.pid"
if [ -f "$PID_FILE" ]; then
old_pid="$(cat "$PID_FILE" || true)"
if [ -n "$old_pid" ] && kill -0 "$old_pid" 2>/dev/null; then
echo "앱이 이미 실행 중입니다. pid=$old_pid"
exit 1
fi
fi
echo "[$(date -u +"%Y-%m-%dT%H:%M:%SZ")] starting: $APP_CMD" | tee -a "$RUN_LOG"
nohup bash -lc "$APP_CMD" >>"$RUN_LOG" 2>&1 &
app_pid=$!
echo "$app_pid" > "$PID_FILE"
echo "[$(date -u +"%Y-%m-%dT%H:%M:%SZ")] app pid=$app_pid" | tee -a "$RUN_LOG"
nohup env PID_FILE="$PID_FILE" LOG_FILE="$WATCHDOG_LOG" CHECK_INTERVAL="$CHECK_INTERVAL" \
bash scripts/watchdog.sh >/dev/null 2>&1 &
watchdog_pid=$!
echo "$watchdog_pid" > "$WATCHDOG_PID_FILE"
cat <<EOF
시작 완료
- app pid: $app_pid
- watchdog pid: $watchdog_pid
- app log: $RUN_LOG
- watchdog log: $WATCHDOG_LOG
실시간 확인:
tail -f "$RUN_LOG"
tail -f "$WATCHDOG_LOG"
EOF
if [ "$TMUX_AUTO" = "true" ]; then
if ! command -v tmux >/dev/null 2>&1; then
echo "tmux를 찾지 못해 자동 세션 생성은 건너뜁니다."
exit 0
fi
session_name="${TMUX_SESSION_PREFIX}_${timestamp}"
window_name="overnight"
tmux new-session -d -s "$session_name" -n "$window_name" "tail -f '$RUN_LOG'"
tmux split-window -t "${session_name}:${window_name}" -v "tail -f '$WATCHDOG_LOG'"
tmux select-layout -t "${session_name}:${window_name}" even-vertical
echo "tmux session 생성: $session_name"
echo "수동 접속: tmux attach -t $session_name"
if [ -z "${TMUX:-}" ] && [ "$TMUX_ATTACH" = "true" ]; then
tmux attach -t "$session_name"
fi
fi

View File

@@ -1,76 +0,0 @@
#!/usr/bin/env bash
# Stop The Ouroboros overnight app/watchdog/tmux session.
set -euo pipefail
LOG_DIR="${LOG_DIR:-data/overnight}"
PID_FILE="$LOG_DIR/app.pid"
WATCHDOG_PID_FILE="$LOG_DIR/watchdog.pid"
TMUX_SESSION_PREFIX="${TMUX_SESSION_PREFIX:-ouroboros_overnight}"
KILL_TIMEOUT="${KILL_TIMEOUT:-5}"
stop_pid() {
local name="$1"
local pid="$2"
if [ -z "$pid" ]; then
echo "$name PID가 비어 있습니다."
return 1
fi
if ! kill -0 "$pid" 2>/dev/null; then
echo "$name 프로세스가 이미 종료됨 (pid=$pid)"
return 0
fi
kill "$pid" 2>/dev/null || true
for _ in $(seq 1 "$KILL_TIMEOUT"); do
if ! kill -0 "$pid" 2>/dev/null; then
echo "$name 종료됨 (pid=$pid)"
return 0
fi
sleep 1
done
kill -9 "$pid" 2>/dev/null || true
if ! kill -0 "$pid" 2>/dev/null; then
echo "$name 강제 종료됨 (pid=$pid)"
return 0
fi
echo "$name 종료 실패 (pid=$pid)"
return 1
}
status=0
if [ -f "$WATCHDOG_PID_FILE" ]; then
watchdog_pid="$(cat "$WATCHDOG_PID_FILE" || true)"
stop_pid "watchdog" "$watchdog_pid" || status=1
rm -f "$WATCHDOG_PID_FILE"
else
echo "watchdog pid 파일 없음: $WATCHDOG_PID_FILE"
fi
if [ -f "$PID_FILE" ]; then
app_pid="$(cat "$PID_FILE" || true)"
stop_pid "app" "$app_pid" || status=1
rm -f "$PID_FILE"
else
echo "app pid 파일 없음: $PID_FILE"
fi
if command -v tmux >/dev/null 2>&1; then
sessions="$(tmux ls 2>/dev/null | awk -F: -v p="$TMUX_SESSION_PREFIX" '$1 ~ "^" p "_" {print $1}')"
if [ -n "$sessions" ]; then
while IFS= read -r s; do
[ -z "$s" ] && continue
tmux kill-session -t "$s" 2>/dev/null || true
echo "tmux 세션 종료: $s"
done <<< "$sessions"
else
echo "종료할 tmux 세션 없음 (prefix=${TMUX_SESSION_PREFIX}_)"
fi
fi
exit "$status"

View File

@@ -1,42 +0,0 @@
#!/usr/bin/env bash
# Simple watchdog for The Ouroboros process.
set -euo pipefail
PID_FILE="${PID_FILE:-data/overnight/app.pid}"
LOG_FILE="${LOG_FILE:-data/overnight/watchdog.log}"
CHECK_INTERVAL="${CHECK_INTERVAL:-30}"
STATUS_EVERY="${STATUS_EVERY:-10}"
mkdir -p "$(dirname "$LOG_FILE")"
log() {
printf '%s %s\n' "$(date -u +"%Y-%m-%dT%H:%M:%SZ")" "$1" | tee -a "$LOG_FILE"
}
if [ ! -f "$PID_FILE" ]; then
log "[ERROR] pid file not found: $PID_FILE"
exit 1
fi
PID="$(cat "$PID_FILE")"
if [ -z "$PID" ]; then
log "[ERROR] pid file is empty: $PID_FILE"
exit 1
fi
log "[INFO] watchdog started (pid=$PID, interval=${CHECK_INTERVAL}s)"
count=0
while true; do
if kill -0 "$PID" 2>/dev/null; then
count=$((count + 1))
if [ $((count % STATUS_EVERY)) -eq 0 ]; then
log "[INFO] process alive (pid=$PID)"
fi
else
log "[ERROR] process stopped (pid=$PID)"
exit 1
fi
sleep "$CHECK_INTERVAL"
done

View File

@@ -1,4 +1,8 @@
"""Smart Volatility Scanner with volatility-first market ranking logic."""
"""Smart Volatility Scanner with RSI and volume filters.
Fetches market rankings from KIS API and applies technical filters
to identify high-probability trading candidates.
"""
from __future__ import annotations
@@ -8,9 +12,7 @@ from typing import Any
from src.analysis.volatility import VolatilityAnalyzer
from src.broker.kis_api import KISBroker
from src.broker.overseas import OverseasBroker
from src.config import Settings
from src.markets.schedule import MarketInfo
logger = logging.getLogger(__name__)
@@ -30,19 +32,19 @@ class ScanCandidate:
class SmartVolatilityScanner:
"""Scans market rankings and applies volatility-first filters.
"""Scans market rankings and applies RSI/volume filters.
Flow:
1. Fetch fluctuation rankings as primary universe
2. Fetch volume rankings for liquidity bonus
3. Score by volatility first, liquidity second
4. Return top N qualified candidates
1. Fetch volume rankings from KIS API
2. For each ranked stock, fetch daily prices
3. Calculate RSI and volume ratio
4. Apply filters: volume > VOL_MULTIPLIER AND (RSI < 30 OR RSI > 70)
5. Return top N qualified candidates
"""
def __init__(
self,
broker: KISBroker,
overseas_broker: OverseasBroker | None,
volatility_analyzer: VolatilityAnalyzer,
settings: Settings,
) -> None:
@@ -54,7 +56,6 @@ class SmartVolatilityScanner:
settings: Application settings
"""
self.broker = broker
self.overseas_broker = overseas_broker
self.analyzer = volatility_analyzer
self.settings = settings
@@ -66,129 +67,107 @@ class SmartVolatilityScanner:
async def scan(
self,
market: MarketInfo | None = None,
fallback_stocks: list[str] | None = None,
) -> list[ScanCandidate]:
"""Execute smart scan and return qualified candidates.
Args:
market: Target market info (domestic vs overseas behavior)
fallback_stocks: Stock codes to use if ranking API fails
Returns:
List of ScanCandidate, sorted by score, up to top_n items
"""
if market and not market.is_domestic:
return await self._scan_overseas(market, fallback_stocks)
return await self._scan_domestic(fallback_stocks)
async def _scan_domestic(
self,
fallback_stocks: list[str] | None = None,
) -> list[ScanCandidate]:
"""Scan domestic market using volatility-first ranking + liquidity bonus."""
# 1) Primary universe from fluctuation ranking.
# Step 1: Fetch rankings
try:
fluct_rows = await self.broker.fetch_market_rankings(
ranking_type="fluctuation",
limit=50,
)
except ConnectionError as exc:
logger.warning("Domestic fluctuation ranking failed: %s", exc)
fluct_rows = []
# 2) Liquidity bonus from volume ranking.
try:
volume_rows = await self.broker.fetch_market_rankings(
rankings = await self.broker.fetch_market_rankings(
ranking_type="volume",
limit=50,
limit=30, # Fetch more than needed for filtering
)
logger.info("Fetched %d stocks from volume rankings", len(rankings))
except ConnectionError as exc:
logger.warning("Domestic volume ranking failed: %s", exc)
volume_rows = []
if not fluct_rows and fallback_stocks:
logger.info(
"Domestic ranking unavailable; using fallback symbols (%d)",
len(fallback_stocks),
)
fluct_rows = [
{
"stock_code": code,
"name": code,
"price": 0.0,
"volume": 0.0,
"change_rate": 0.0,
"volume_increase_rate": 0.0,
}
for code in fallback_stocks
]
if not fluct_rows:
return []
volume_rank_bonus: dict[str, float] = {}
for idx, row in enumerate(volume_rows):
code = _extract_stock_code(row)
if not code:
continue
volume_rank_bonus[code] = max(0.0, 15.0 - idx * 0.3)
logger.warning("Ranking API failed, using fallback: %s", exc)
if fallback_stocks:
# Create minimal ranking data for fallback
rankings = [
{
"stock_code": code,
"name": code,
"price": 0,
"volume": 0,
"change_rate": 0,
"volume_increase_rate": 0,
}
for code in fallback_stocks
]
else:
return []
# Step 2: Analyze each stock
candidates: list[ScanCandidate] = []
for stock in fluct_rows:
stock_code = _extract_stock_code(stock)
for stock in rankings:
stock_code = stock["stock_code"]
if not stock_code:
continue
try:
price = _extract_last_price(stock)
change_rate = _extract_change_rate_pct(stock)
volume = _extract_volume(stock)
# Fetch daily prices for RSI calculation
daily_prices = await self.broker.get_daily_prices(stock_code, days=20)
intraday_range_pct = 0.0
volume_ratio = _safe_float(stock.get("volume_increase_rate"), 0.0) / 100.0 + 1.0
# Use daily chart to refine range/volume when available.
daily_prices = await self.broker.get_daily_prices(stock_code, days=2)
if daily_prices:
latest = daily_prices[-1]
latest_close = _safe_float(latest.get("close"), default=price)
if price <= 0:
price = latest_close
latest_high = _safe_float(latest.get("high"))
latest_low = _safe_float(latest.get("low"))
if latest_close > 0 and latest_high > 0 and latest_low > 0 and latest_high >= latest_low:
intraday_range_pct = (latest_high - latest_low) / latest_close * 100.0
if volume <= 0:
volume = _safe_float(latest.get("volume"))
if len(daily_prices) >= 2:
prev_day_volume = _safe_float(daily_prices[-2].get("volume"))
if prev_day_volume > 0:
volume_ratio = max(volume_ratio, volume / prev_day_volume)
volatility_pct = max(abs(change_rate), intraday_range_pct)
if price <= 0 or volatility_pct < 0.8:
if len(daily_prices) < 15: # Need at least 14+1 for RSI
logger.debug("Insufficient price history for %s", stock_code)
continue
volatility_score = min(volatility_pct / 10.0, 1.0) * 85.0
liquidity_score = volume_rank_bonus.get(stock_code, 0.0)
score = min(100.0, volatility_score + liquidity_score)
signal = "momentum" if change_rate >= 0 else "oversold"
implied_rsi = max(0.0, min(100.0, 50.0 + (change_rate * 4.0)))
# Calculate RSI
close_prices = [p["close"] for p in daily_prices]
rsi = self.analyzer.calculate_rsi(close_prices, period=14)
candidates.append(
ScanCandidate(
stock_code=stock_code,
name=stock.get("name", stock_code),
price=price,
volume=volume,
volume_ratio=max(1.0, volume_ratio, volatility_pct / 2.0),
rsi=implied_rsi,
signal=signal,
score=score,
# Calculate volume ratio (today vs previous day avg)
if len(daily_prices) >= 2:
prev_day_volume = daily_prices[-2]["volume"]
current_volume = stock.get("volume", 0) or daily_prices[-1]["volume"]
volume_ratio = (
current_volume / prev_day_volume if prev_day_volume > 0 else 1.0
)
else:
volume_ratio = stock.get("volume_increase_rate", 0) / 100 + 1 # Fallback
# Apply filters
volume_qualified = volume_ratio >= self.vol_multiplier
rsi_oversold = rsi < self.rsi_oversold
rsi_momentum = rsi > self.rsi_momentum
if volume_qualified and (rsi_oversold or rsi_momentum):
signal = "oversold" if rsi_oversold else "momentum"
# Calculate composite score
# Higher score for: extreme RSI + high volume
rsi_extremity = abs(rsi - 50) / 50 # 0-1 scale
volume_score = min(volume_ratio / 5, 1.0) # Cap at 5x
score = (rsi_extremity * 0.6 + volume_score * 0.4) * 100
candidates.append(
ScanCandidate(
stock_code=stock_code,
name=stock.get("name", stock_code),
price=stock.get("price", daily_prices[-1]["close"]),
volume=current_volume,
volume_ratio=volume_ratio,
rsi=rsi,
signal=signal,
score=score,
)
)
logger.info(
"Qualified: %s (%s) RSI=%.1f vol=%.1fx signal=%s score=%.1f",
stock_code,
stock.get("name", ""),
rsi,
volume_ratio,
signal,
score,
)
)
except ConnectionError as exc:
logger.warning("Failed to analyze %s: %s", stock_code, exc)
@@ -197,171 +176,10 @@ class SmartVolatilityScanner:
logger.error("Unexpected error analyzing %s: %s", stock_code, exc)
continue
logger.info("Domestic ranking scan found %d candidates", len(candidates))
# Sort by score and return top N
candidates.sort(key=lambda c: c.score, reverse=True)
return candidates[: self.top_n]
async def _scan_overseas(
self,
market: MarketInfo,
fallback_stocks: list[str] | None = None,
) -> list[ScanCandidate]:
"""Scan overseas symbols using ranking API first, then fallback universe."""
if self.overseas_broker is None:
logger.warning(
"Overseas scanner unavailable for %s: overseas broker not configured",
market.name,
)
return []
candidates = await self._scan_overseas_from_rankings(market)
if not candidates:
candidates = await self._scan_overseas_from_symbols(market, fallback_stocks)
candidates.sort(key=lambda c: c.score, reverse=True)
return candidates[: self.top_n]
async def _scan_overseas_from_rankings(
self,
market: MarketInfo,
) -> list[ScanCandidate]:
"""Build overseas candidates from ranking APIs using volatility-first scoring."""
assert self.overseas_broker is not None
try:
fluct_rows = await self.overseas_broker.fetch_overseas_rankings(
exchange_code=market.exchange_code,
ranking_type="fluctuation",
limit=50,
)
except Exception as exc:
logger.warning(
"Overseas fluctuation ranking failed for %s: %s", market.code, exc
)
fluct_rows = []
if not fluct_rows:
return []
volume_rank_bonus: dict[str, float] = {}
try:
volume_rows = await self.overseas_broker.fetch_overseas_rankings(
exchange_code=market.exchange_code,
ranking_type="volume",
limit=50,
)
except Exception as exc:
logger.warning(
"Overseas volume ranking failed for %s: %s", market.code, exc
)
volume_rows = []
for idx, row in enumerate(volume_rows):
code = _extract_stock_code(row)
if not code:
continue
# Top-ranked by traded value/volume gets higher liquidity bonus.
volume_rank_bonus[code] = max(0.0, 15.0 - idx * 0.3)
candidates: list[ScanCandidate] = []
for row in fluct_rows:
stock_code = _extract_stock_code(row)
if not stock_code:
continue
price = _extract_last_price(row)
change_rate = _extract_change_rate_pct(row)
volume = _extract_volume(row)
intraday_range_pct = _extract_intraday_range_pct(row, price)
volatility_pct = max(abs(change_rate), intraday_range_pct)
# Volatility-first filter (not simple gainers/value ranking).
if price <= 0 or volatility_pct < 0.8:
continue
volatility_score = min(volatility_pct / 10.0, 1.0) * 85.0
liquidity_score = volume_rank_bonus.get(stock_code, 0.0)
score = min(100.0, volatility_score + liquidity_score)
signal = "momentum" if change_rate >= 0 else "oversold"
implied_rsi = max(0.0, min(100.0, 50.0 + (change_rate * 4.0)))
candidates.append(
ScanCandidate(
stock_code=stock_code,
name=str(row.get("name") or row.get("ovrs_item_name") or stock_code),
price=price,
volume=volume,
volume_ratio=max(1.0, volatility_pct / 2.0),
rsi=implied_rsi,
signal=signal,
score=score,
)
)
if candidates:
logger.info(
"Overseas ranking scan found %d candidates for %s",
len(candidates),
market.name,
)
return candidates
async def _scan_overseas_from_symbols(
self,
market: MarketInfo,
symbols: list[str] | None,
) -> list[ScanCandidate]:
"""Fallback overseas scan from dynamic symbol universe."""
assert self.overseas_broker is not None
if not symbols:
logger.info("Overseas scanner: no symbol universe for %s", market.name)
return []
logger.info(
"Overseas scanner: scanning %d fallback symbols for %s",
len(symbols),
market.name,
)
candidates: list[ScanCandidate] = []
for stock_code in symbols:
try:
price_data = await self.overseas_broker.get_overseas_price(
market.exchange_code, stock_code
)
output = price_data.get("output", {})
price = _extract_last_price(output)
change_rate = _extract_change_rate_pct(output)
volume = _extract_volume(output)
intraday_range_pct = _extract_intraday_range_pct(output, price)
volatility_pct = max(abs(change_rate), intraday_range_pct)
if price <= 0 or volatility_pct < 0.8:
continue
score = min(volatility_pct / 10.0, 1.0) * 100.0
signal = "momentum" if change_rate >= 0 else "oversold"
implied_rsi = max(0.0, min(100.0, 50.0 + (change_rate * 4.0)))
candidates.append(
ScanCandidate(
stock_code=stock_code,
name=stock_code,
price=price,
volume=volume,
volume_ratio=max(1.0, volatility_pct / 2.0),
rsi=implied_rsi,
signal=signal,
score=score,
)
)
except ConnectionError as exc:
logger.warning("Failed to analyze overseas %s: %s", stock_code, exc)
except Exception as exc:
logger.error("Unexpected error analyzing overseas %s: %s", stock_code, exc)
logger.info(
"Overseas symbol fallback scan found %d candidates for %s",
len(candidates),
market.name,
)
return candidates
def get_stock_codes(self, candidates: list[ScanCandidate]) -> list[str]:
"""Extract stock codes from candidates for watchlist update.
@@ -372,78 +190,3 @@ class SmartVolatilityScanner:
List of stock codes
"""
return [c.stock_code for c in candidates]
def _safe_float(value: Any, default: float = 0.0) -> float:
"""Convert arbitrary values to float safely."""
if value in (None, ""):
return default
try:
return float(value)
except (TypeError, ValueError):
return default
def _extract_stock_code(row: dict[str, Any]) -> str:
"""Extract normalized stock code from various API schemas."""
return (
str(
row.get("symb")
or row.get("ovrs_pdno")
or row.get("stock_code")
or row.get("pdno")
or ""
)
.strip()
.upper()
)
def _extract_last_price(row: dict[str, Any]) -> float:
"""Extract last/close-like price from API schema variants."""
return _safe_float(
row.get("last")
or row.get("ovrs_nmix_prpr")
or row.get("stck_prpr")
or row.get("price")
or row.get("close")
)
def _extract_change_rate_pct(row: dict[str, Any]) -> float:
"""Extract daily change rate (%) from API schema variants."""
return _safe_float(
row.get("rate")
or row.get("change_rate")
or row.get("prdy_ctrt")
or row.get("evlu_pfls_rt")
or row.get("chg_rt")
)
def _extract_volume(row: dict[str, Any]) -> float:
"""Extract volume/traded-amount proxy from schema variants."""
return _safe_float(
row.get("tvol") or row.get("acml_vol") or row.get("vol") or row.get("volume")
)
def _extract_intraday_range_pct(row: dict[str, Any], price: float) -> float:
"""Estimate intraday range percentage from high/low fields."""
if price <= 0:
return 0.0
high = _safe_float(
row.get("high")
or row.get("ovrs_hgpr")
or row.get("stck_hgpr")
or row.get("day_hgpr")
)
low = _safe_float(
row.get("low")
or row.get("ovrs_lwpr")
or row.get("stck_lwpr")
or row.get("day_lwpr")
)
if high <= 0 or low <= 0 or high < low:
return 0.0
return (high - low) / price * 100.0

View File

@@ -410,10 +410,8 @@ class GeminiClient:
cached=True,
)
# Build prompt (prompt_override takes priority for callers like pre_market_planner)
if "prompt_override" in market_data:
prompt = market_data["prompt_override"]
elif self._enable_optimization:
# Build optimized prompt
if self._enable_optimization:
prompt = self._optimizer.build_compressed_prompt(market_data)
else:
prompt = await self.build_prompt(market_data, news_sentiment)

View File

@@ -20,39 +20,6 @@ _KIS_VTS_HOST = "openapivts.koreainvestment.com"
logger = logging.getLogger(__name__)
def kr_tick_unit(price: float) -> int:
"""Return KRX tick size for the given price level.
KRX price tick rules (domestic stocks):
price < 2,000 → 1원
2,000 ≤ price < 5,000 → 5원
5,000 ≤ price < 20,000 → 10원
20,000 ≤ price < 50,000 → 50원
50,000 ≤ price < 200,000 → 100원
200,000 ≤ price < 500,000 → 500원
500,000 ≤ price → 1,000원
"""
if price < 2_000:
return 1
if price < 5_000:
return 5
if price < 20_000:
return 10
if price < 50_000:
return 50
if price < 200_000:
return 100
if price < 500_000:
return 500
return 1_000
def kr_round_down(price: float) -> int:
"""Round *down* price to the nearest KRX tick unit."""
tick = kr_tick_unit(price)
return int(price // tick * tick)
class LeakyBucket:
"""Simple leaky-bucket rate limiter for async code."""
@@ -137,14 +104,12 @@ class KISBroker:
time_since_last_attempt = now - self._last_refresh_attempt
if time_since_last_attempt < self._refresh_cooldown:
remaining = self._refresh_cooldown - time_since_last_attempt
# Do not fail fast here. If token is unavailable, upstream calls
# will all fail for up to a minute and scanning returns no trades.
logger.warning(
"Token refresh on cooldown. Waiting %.1fs before retry (KIS allows 1/minute)",
remaining,
error_msg = (
f"Token refresh on cooldown. "
f"Retry in {remaining:.1f}s (KIS allows 1/minute)"
)
await asyncio.sleep(remaining)
now = asyncio.get_event_loop().time()
logger.warning(error_msg)
raise ConnectionError(error_msg)
logger.info("Refreshing KIS access token")
self._last_refresh_attempt = now
@@ -231,55 +196,6 @@ class KISBroker:
except (TimeoutError, aiohttp.ClientError) as exc:
raise ConnectionError(f"Network error fetching orderbook: {exc}") from exc
async def get_current_price(
self, stock_code: str
) -> tuple[float, float, float]:
"""Fetch current price data for a domestic stock.
Uses the ``inquire-price`` API (FHKST01010100), which works in both
real and VTS environments and returns the actual last-traded price.
Returns:
(current_price, prdy_ctrt, frgn_ntby_qty)
- current_price: Last traded price in KRW.
- prdy_ctrt: Day change rate (%).
- frgn_ntby_qty: Foreigner net buy quantity.
"""
await self._rate_limiter.acquire()
session = self._get_session()
headers = await self._auth_headers("FHKST01010100")
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": stock_code,
}
url = f"{self._base_url}/uapi/domestic-stock/v1/quotations/inquire-price"
def _f(val: str | None) -> float:
try:
return float(val or "0")
except ValueError:
return 0.0
try:
async with session.get(url, headers=headers, params=params) as resp:
if resp.status != 200:
text = await resp.text()
raise ConnectionError(
f"get_current_price failed ({resp.status}): {text}"
)
data = await resp.json()
out = data.get("output", {})
return (
_f(out.get("stck_prpr")),
_f(out.get("prdy_ctrt")),
_f(out.get("frgn_ntby_qty")),
)
except (TimeoutError, aiohttp.ClientError) as exc:
raise ConnectionError(
f"Network error fetching current price: {exc}"
) from exc
async def get_balance(self) -> dict[str, Any]:
"""Fetch current account balance and holdings."""
await self._rate_limiter.acquire()
@@ -331,23 +247,13 @@ class KISBroker:
session = self._get_session()
tr_id = "VTTC0802U" if order_type == "BUY" else "VTTC0801U"
# KRX requires limit orders to be rounded down to the tick unit.
# ORD_DVSN: "00"=지정가, "01"=시장가
if price > 0:
ord_dvsn = "00" # 지정가
ord_price = kr_round_down(price)
else:
ord_dvsn = "01" # 시장가
ord_price = 0
body = {
"CANO": self._account_no,
"ACNT_PRDT_CD": self._product_cd,
"PDNO": stock_code,
"ORD_DVSN": ord_dvsn,
"ORD_DVSN": "01" if price > 0 else "06", # 01=지정가, 06=시장가
"ORD_QTY": str(quantity),
"ORD_UNPR": str(ord_price),
"ORD_UNPR": str(price),
}
hash_key = await self._get_hash_key(body)
@@ -396,46 +302,26 @@ class KISBroker:
await self._rate_limiter.acquire()
session = self._get_session()
if ranking_type == "volume":
# 거래량순위: FHPST01710000 / /quotations/volume-rank
tr_id = "FHPST01710000"
url = f"{self._base_url}/uapi/domestic-stock/v1/quotations/volume-rank"
params: dict[str, str] = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_COND_SCR_DIV_CODE": "20171",
"FID_INPUT_ISCD": "0000",
"FID_DIV_CLS_CODE": "0",
"FID_BLNG_CLS_CODE": "0",
"FID_TRGT_CLS_CODE": "111111111",
"FID_TRGT_EXLS_CLS_CODE": "0000000000",
"FID_INPUT_PRICE_1": "0",
"FID_INPUT_PRICE_2": "0",
"FID_VOL_CNT": "0",
"FID_INPUT_DATE_1": "",
}
else:
# 등락률순위: FHPST01700000 / /ranking/fluctuation (소문자 파라미터)
tr_id = "FHPST01700000"
url = f"{self._base_url}/uapi/domestic-stock/v1/ranking/fluctuation"
params = {
"fid_cond_mrkt_div_code": "J",
"fid_cond_scr_div_code": "20170",
"fid_input_iscd": "0000",
"fid_rank_sort_cls_code": "0000",
"fid_input_cnt_1": str(limit),
"fid_prc_cls_code": "0",
"fid_input_price_1": "0",
"fid_input_price_2": "0",
"fid_vol_cnt": "0",
"fid_trgt_cls_code": "0",
"fid_trgt_exls_cls_code": "0",
"fid_div_cls_code": "0",
"fid_rsfl_rate1": "0",
"fid_rsfl_rate2": "0",
}
# TR_ID for volume ranking
tr_id = "FHPST01710000" if ranking_type == "volume" else "FHPST01710100"
headers = await self._auth_headers(tr_id)
params = {
"FID_COND_MRKT_DIV_CODE": "J", # Stock/ETF/ETN
"FID_COND_SCR_DIV_CODE": "20001", # Volume surge
"FID_INPUT_ISCD": "0000", # All stocks
"FID_DIV_CLS_CODE": "0", # All types
"FID_BLNG_CLS_CODE": "0",
"FID_TRGT_CLS_CODE": "111111111",
"FID_TRGT_EXLS_CLS_CODE": "000000",
"FID_INPUT_PRICE_1": "0",
"FID_INPUT_PRICE_2": "0",
"FID_VOL_CNT": "0",
"FID_INPUT_DATE_1": "",
}
url = f"{self._base_url}/uapi/domestic-stock/v1/quotations/volume-rank"
try:
async with session.get(url, headers=headers, params=params) as resp:
if resp.status != 200:

View File

@@ -12,24 +12,6 @@ from src.broker.kis_api import KISBroker
logger = logging.getLogger(__name__)
# Ranking API uses different exchange codes than order/quote APIs.
_RANKING_EXCHANGE_MAP: dict[str, str] = {
"NASD": "NAS",
"NYSE": "NYS",
"AMEX": "AMS",
"SEHK": "HKS",
"SHAA": "SHS",
"SZAA": "SZS",
"HSX": "HSX",
"HNX": "HNX",
"TSE": "TSE",
}
# Price inquiry API (HHDFS00000300) uses the same short exchange codes as rankings.
# NASD → NAS, NYSE → NYS, AMEX → AMS (confirmed: AMEX returns empty, AMS returns price).
_PRICE_EXCHANGE_MAP: dict[str, str] = _RANKING_EXCHANGE_MAP
class OverseasBroker:
"""KIS Overseas Stock API wrapper that reuses KISBroker infrastructure."""
@@ -62,11 +44,9 @@ class OverseasBroker:
session = self._broker._get_session()
headers = await self._broker._auth_headers("HHDFS00000300")
# Map internal exchange codes to the short form expected by the price API.
price_excd = _PRICE_EXCHANGE_MAP.get(exchange_code, exchange_code)
params = {
"AUTH": "",
"EXCD": price_excd,
"EXCD": exchange_code,
"SYMB": stock_code,
}
url = f"{self._broker._base_url}/uapi/overseas-price/v1/quotations/price"
@@ -84,81 +64,6 @@ class OverseasBroker:
f"Network error fetching overseas price: {exc}"
) from exc
async def fetch_overseas_rankings(
self,
exchange_code: str,
ranking_type: str = "fluctuation",
limit: int = 30,
) -> list[dict[str, Any]]:
"""Fetch overseas rankings (price change or volume surge).
Ranking API specs may differ by account/product. Endpoint paths and
TR_IDs are configurable via settings and can be overridden in .env.
"""
if not self._broker._settings.OVERSEAS_RANKING_ENABLED:
return []
await self._broker._rate_limiter.acquire()
session = self._broker._get_session()
ranking_excd = _RANKING_EXCHANGE_MAP.get(exchange_code, exchange_code)
if ranking_type == "volume":
tr_id = self._broker._settings.OVERSEAS_RANKING_VOLUME_TR_ID
path = self._broker._settings.OVERSEAS_RANKING_VOLUME_PATH
params: dict[str, str] = {
"AUTH": "",
"EXCD": ranking_excd,
"MIXN": "0",
"VOL_RANG": "0",
}
else:
tr_id = self._broker._settings.OVERSEAS_RANKING_FLUCT_TR_ID
path = self._broker._settings.OVERSEAS_RANKING_FLUCT_PATH
params = {
"AUTH": "",
"EXCD": ranking_excd,
"NDAY": "0",
"GUBN": "1",
"VOL_RANG": "0",
}
headers = await self._broker._auth_headers(tr_id)
url = f"{self._broker._base_url}{path}"
try:
async with session.get(url, headers=headers, params=params) as resp:
if resp.status != 200:
text = await resp.text()
if resp.status == 404:
logger.warning(
"Overseas ranking endpoint unavailable (404) for %s/%s; "
"using symbol fallback scan",
exchange_code,
ranking_type,
)
return []
raise ConnectionError(
f"fetch_overseas_rankings failed ({resp.status}): {text}"
)
data = await resp.json()
rows = self._extract_ranking_rows(data)
if rows:
return rows[:limit]
logger.debug(
"Overseas ranking returned empty for %s/%s (keys=%s)",
exchange_code,
ranking_type,
list(data.keys()),
)
return []
except (TimeoutError, aiohttp.ClientError) as exc:
raise ConnectionError(
f"Network error fetching overseas rankings: {exc}"
) from exc
async def get_overseas_balance(self, exchange_code: str) -> dict[str, Any]:
"""
Fetch overseas account balance.
@@ -257,27 +162,14 @@ class OverseasBroker:
f"send_overseas_order failed ({resp.status}): {text}"
)
data = await resp.json()
rt_cd = data.get("rt_cd", "")
msg1 = data.get("msg1", "")
if rt_cd == "0":
logger.info(
"Overseas order submitted",
extra={
"exchange": exchange_code,
"stock_code": stock_code,
"action": order_type,
},
)
else:
logger.warning(
"Overseas order rejected (rt_cd=%s): %s [%s %s %s qty=%d]",
rt_cd,
msg1,
order_type,
stock_code,
exchange_code,
quantity,
)
logger.info(
"Overseas order submitted",
extra={
"exchange": exchange_code,
"stock_code": stock_code,
"action": order_type,
},
)
return data
except (TimeoutError, aiohttp.ClientError) as exc:
raise ConnectionError(
@@ -306,11 +198,3 @@ class OverseasBroker:
"HSX": "VND",
}
return currency_map.get(exchange_code, "USD")
def _extract_ranking_rows(self, data: dict[str, Any]) -> list[dict[str, Any]]:
"""Extract list rows from ranking response across schema variants."""
candidates = [data.get("output"), data.get("output1"), data.get("output2")]
for value in candidates:
if isinstance(value, list):
return [row for row in value if isinstance(row, dict)]
return []

View File

@@ -38,11 +38,6 @@ class Settings(BaseSettings):
RSI_MOMENTUM_THRESHOLD: int = Field(default=70, ge=50, le=100)
VOL_MULTIPLIER: float = Field(default=2.0, gt=1.0, le=10.0)
SCANNER_TOP_N: int = Field(default=3, ge=1, le=10)
POSITION_SIZING_ENABLED: bool = True
POSITION_BASE_ALLOCATION_PCT: float = Field(default=5.0, gt=0.0, le=30.0)
POSITION_MIN_ALLOCATION_PCT: float = Field(default=1.0, gt=0.0, le=20.0)
POSITION_MAX_ALLOCATION_PCT: float = Field(default=10.0, gt=0.0, le=50.0)
POSITION_VOLATILITY_TARGET_SCORE: float = Field(default=50.0, gt=0.0, le=100.0)
# Database
DB_PATH: str = "data/trade_logs.db"
@@ -55,11 +50,6 @@ class Settings(BaseSettings):
# Trading mode
MODE: str = Field(default="paper", pattern="^(paper|live)$")
# Simulated USD cash for VTS (paper) overseas trading.
# KIS VTS overseas balance API returns errors for most accounts.
# This value is used as a fallback when the balance API returns 0 in paper mode.
PAPER_OVERSEAS_CASH: float = Field(default=50000.0, ge=0.0)
# Trading frequency mode (daily = batch API calls, realtime = per-stock calls)
TRADE_MODE: str = Field(default="daily", pattern="^(daily|realtime)$")
DAILY_SESSIONS: int = Field(default=4, ge=1, le=10)
@@ -93,28 +83,6 @@ class Settings(BaseSettings):
TELEGRAM_COMMANDS_ENABLED: bool = True
TELEGRAM_POLLING_INTERVAL: float = 1.0 # seconds
# Telegram notification type filters (granular control)
# circuit_breaker is always sent regardless — safety-critical
TELEGRAM_NOTIFY_TRADES: bool = True # BUY/SELL execution alerts
TELEGRAM_NOTIFY_MARKET_OPEN_CLOSE: bool = True # Market open/close alerts
TELEGRAM_NOTIFY_FAT_FINGER: bool = True # Fat-finger rejection alerts
TELEGRAM_NOTIFY_SYSTEM_EVENTS: bool = True # System start/shutdown alerts
TELEGRAM_NOTIFY_PLAYBOOK: bool = True # Playbook generated/failed alerts
TELEGRAM_NOTIFY_SCENARIO_MATCH: bool = True # Scenario matched alerts (most frequent)
TELEGRAM_NOTIFY_ERRORS: bool = True # Error alerts
# Overseas ranking API (KIS endpoint/TR_ID may vary by account/product)
# Override these from .env if your account uses different specs.
OVERSEAS_RANKING_ENABLED: bool = True
OVERSEAS_RANKING_FLUCT_TR_ID: str = "HHDFS76290000"
OVERSEAS_RANKING_VOLUME_TR_ID: str = "HHDFS76270000"
OVERSEAS_RANKING_FLUCT_PATH: str = (
"/uapi/overseas-stock/v1/ranking/updown-rate"
)
OVERSEAS_RANKING_VOLUME_PATH: str = (
"/uapi/overseas-stock/v1/ranking/volume-surge"
)
# Dashboard (optional)
DASHBOARD_ENABLED: bool = False
DASHBOARD_HOST: str = "127.0.0.1"
@@ -133,7 +101,4 @@ class Settings(BaseSettings):
@property
def enabled_market_list(self) -> list[str]:
"""Parse ENABLED_MARKETS into list of market codes."""
from src.markets.schedule import expand_market_codes
raw = [m.strip() for m in self.ENABLED_MARKETS.split(",") if m.strip()]
return expand_market_codes(raw)
return [m.strip() for m in self.ENABLED_MARKETS.split(",") if m.strip()]

View File

@@ -26,19 +26,7 @@ def create_dashboard_app(db_path: str) -> FastAPI:
def get_status() -> dict[str, Any]:
today = datetime.now(UTC).date().isoformat()
with _connect(db_path) as conn:
market_rows = conn.execute(
"""
SELECT DISTINCT market FROM (
SELECT market FROM trades WHERE DATE(timestamp) = ?
UNION
SELECT market FROM decision_logs WHERE DATE(timestamp) = ?
UNION
SELECT market FROM playbooks WHERE date = ?
) ORDER BY market
""",
(today, today, today),
).fetchall()
markets = [row[0] for row in market_rows] if market_rows else []
markets = ["KR", "US"]
market_status: dict[str, Any] = {}
total_trades = 0
total_pnl = 0.0
@@ -259,50 +247,6 @@ def create_dashboard_app(db_path: str) -> FastAPI:
)
return {"market": market, "count": len(decisions), "decisions": decisions}
@app.get("/api/pnl/history")
def get_pnl_history(
days: int = Query(default=30, ge=1, le=365),
market: str = Query("all"),
) -> dict[str, Any]:
"""Return daily P&L history for charting."""
with _connect(db_path) as conn:
if market == "all":
rows = conn.execute(
"""
SELECT DATE(timestamp) AS date,
SUM(pnl) AS daily_pnl,
COUNT(*) AS trade_count
FROM trades
WHERE pnl IS NOT NULL
AND DATE(timestamp) >= DATE('now', ?)
GROUP BY DATE(timestamp)
ORDER BY DATE(timestamp)
""",
(f"-{days} days",),
).fetchall()
else:
rows = conn.execute(
"""
SELECT DATE(timestamp) AS date,
SUM(pnl) AS daily_pnl,
COUNT(*) AS trade_count
FROM trades
WHERE pnl IS NOT NULL
AND market = ?
AND DATE(timestamp) >= DATE('now', ?)
GROUP BY DATE(timestamp)
ORDER BY DATE(timestamp)
""",
(market, f"-{days} days"),
).fetchall()
return {
"days": days,
"market": market,
"labels": [row["date"] for row in rows],
"pnl": [round(float(row["daily_pnl"]), 2) for row in rows],
"trades": [int(row["trade_count"]) for row in rows],
}
@app.get("/api/scenarios/active")
def get_active_scenarios(
market: str = Query("US"),

View File

@@ -1,10 +1,9 @@
<!doctype html>
<html lang="ko">
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>The Ouroboros Dashboard</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.0/dist/chart.umd.min.js"></script>
<style>
:root {
--bg: #0b1724;
@@ -12,390 +11,51 @@
--fg: #e6eef7;
--muted: #9fb3c8;
--accent: #3cb371;
--red: #e05555;
--border: #28455f;
}
* { box-sizing: border-box; margin: 0; padding: 0; }
body {
margin: 0;
font-family: ui-monospace, SFMono-Regular, Menlo, monospace;
background: radial-gradient(circle at top left, #173b58, var(--bg));
color: var(--fg);
min-height: 100vh;
font-size: 13px;
}
.wrap { max-width: 1100px; margin: 0 auto; padding: 20px 16px; }
/* Header */
header {
display: flex;
align-items: center;
justify-content: space-between;
margin-bottom: 20px;
padding-bottom: 12px;
border-bottom: 1px solid var(--border);
.wrap {
max-width: 900px;
margin: 48px auto;
padding: 0 16px;
}
header h1 { font-size: 18px; color: var(--accent); letter-spacing: 0.5px; }
.header-right { display: flex; align-items: center; gap: 12px; color: var(--muted); font-size: 12px; }
.refresh-btn {
background: none; border: 1px solid var(--border); color: var(--muted);
padding: 4px 10px; border-radius: 6px; cursor: pointer; font-family: inherit;
font-size: 12px; transition: border-color 0.2s;
}
.refresh-btn:hover { border-color: var(--accent); color: var(--accent); }
/* Summary cards */
.cards { display: grid; grid-template-columns: repeat(4, 1fr); gap: 12px; margin-bottom: 20px; }
@media (max-width: 700px) { .cards { grid-template-columns: repeat(2, 1fr); } }
.card {
background: var(--panel);
border: 1px solid var(--border);
border-radius: 10px;
padding: 16px;
background: color-mix(in oklab, var(--panel), black 12%);
border: 1px solid #28455f;
border-radius: 12px;
padding: 20px;
}
.card-label { color: var(--muted); font-size: 11px; margin-bottom: 6px; text-transform: uppercase; letter-spacing: 0.5px; }
.card-value { font-size: 22px; font-weight: 700; }
.card-sub { color: var(--muted); font-size: 11px; margin-top: 4px; }
.positive { color: var(--accent); }
.negative { color: var(--red); }
.neutral { color: var(--fg); }
/* Chart panel */
.chart-panel {
background: var(--panel);
border: 1px solid var(--border);
border-radius: 10px;
padding: 16px;
margin-bottom: 20px;
h1 {
margin-top: 0;
}
.panel-header {
display: flex;
align-items: center;
justify-content: space-between;
margin-bottom: 16px;
code {
color: var(--accent);
}
.panel-title { font-size: 13px; color: var(--muted); font-weight: 600; }
.chart-container { position: relative; height: 180px; }
.chart-error { color: var(--muted); text-align: center; padding: 40px 0; font-size: 12px; }
/* Days selector */
.days-selector { display: flex; gap: 4px; }
.day-btn {
background: none; border: 1px solid var(--border); color: var(--muted);
padding: 3px 8px; border-radius: 4px; cursor: pointer; font-family: inherit; font-size: 11px;
li {
margin: 6px 0;
color: var(--muted);
}
.day-btn.active { border-color: var(--accent); color: var(--accent); background: rgba(60, 179, 113, 0.08); }
/* Decisions panel */
.decisions-panel {
background: var(--panel);
border: 1px solid var(--border);
border-radius: 10px;
padding: 16px;
}
.market-tabs { display: flex; gap: 6px; flex-wrap: wrap; }
.tab-btn {
background: none; border: 1px solid var(--border); color: var(--muted);
padding: 4px 10px; border-radius: 6px; cursor: pointer; font-family: inherit; font-size: 11px;
}
.tab-btn.active { border-color: var(--accent); color: var(--accent); background: rgba(60, 179, 113, 0.08); }
.decisions-table { width: 100%; border-collapse: collapse; margin-top: 14px; }
.decisions-table th {
text-align: left; color: var(--muted); font-size: 11px; font-weight: 600;
padding: 6px 8px; border-bottom: 1px solid var(--border); white-space: nowrap;
}
.decisions-table td {
padding: 8px 8px; border-bottom: 1px solid rgba(40, 69, 95, 0.5);
vertical-align: middle; white-space: nowrap;
}
.decisions-table tr:last-child td { border-bottom: none; }
.decisions-table tr:hover td { background: rgba(255,255,255,0.02); }
.badge {
display: inline-block; padding: 2px 7px; border-radius: 4px;
font-size: 11px; font-weight: 700; letter-spacing: 0.5px;
}
.badge-buy { background: rgba(60, 179, 113, 0.15); color: var(--accent); }
.badge-sell { background: rgba(224, 85, 85, 0.15); color: var(--red); }
.badge-hold { background: rgba(159, 179, 200, 0.12); color: var(--muted); }
.conf-bar-wrap { display: flex; align-items: center; gap: 6px; min-width: 90px; }
.conf-bar { flex: 1; height: 6px; background: rgba(255,255,255,0.08); border-radius: 3px; overflow: hidden; }
.conf-fill { height: 100%; border-radius: 3px; background: var(--accent); transition: width 0.3s; }
.conf-val { color: var(--muted); font-size: 11px; min-width: 26px; text-align: right; }
.rationale-cell { max-width: 200px; overflow: hidden; text-overflow: ellipsis; color: var(--muted); }
.empty-row td { text-align: center; color: var(--muted); padding: 24px; }
/* Spinner */
.spinner { display: inline-block; width: 12px; height: 12px; border: 2px solid var(--border); border-top-color: var(--accent); border-radius: 50%; animation: spin 0.8s linear infinite; }
@keyframes spin { to { transform: rotate(360deg); } }
</style>
</head>
<body>
<div class="wrap">
<!-- Header -->
<header>
<h1>&#x1F40D; The Ouroboros</h1>
<div class="header-right">
<span id="last-updated">--</span>
<button class="refresh-btn" onclick="refreshAll()">&#x21BA; 새로고침</button>
</div>
</header>
<!-- Summary cards -->
<div class="cards">
<div class="card">
<div class="card-label">오늘 거래</div>
<div class="card-value neutral" id="card-trades">--</div>
<div class="card-sub" id="card-trades-sub">거래 건수</div>
</div>
<div class="card">
<div class="card-label">오늘 P&amp;L</div>
<div class="card-value" id="card-pnl">--</div>
<div class="card-sub" id="card-pnl-sub">실현 손익</div>
</div>
<div class="card">
<div class="card-label">승률</div>
<div class="card-value neutral" id="card-winrate">--</div>
<div class="card-sub">전체 누적</div>
</div>
<div class="card">
<div class="card-label">누적 거래</div>
<div class="card-value neutral" id="card-total">--</div>
<div class="card-sub">전체 기간</div>
</div>
</div>
<!-- P&L Chart -->
<div class="chart-panel">
<div class="panel-header">
<span class="panel-title">P&amp;L 추이</span>
<div class="days-selector">
<button class="day-btn active" data-days="7" onclick="selectDays(this)">7일</button>
<button class="day-btn" data-days="30" onclick="selectDays(this)">30일</button>
<button class="day-btn" data-days="90" onclick="selectDays(this)">90일</button>
</div>
</div>
<div class="chart-container">
<canvas id="pnl-chart"></canvas>
<div class="chart-error" id="chart-error" style="display:none">데이터 없음</div>
</div>
</div>
<!-- Decisions log -->
<div class="decisions-panel">
<div class="panel-header">
<span class="panel-title">최근 결정 로그</span>
<div class="market-tabs" id="market-tabs">
<button class="tab-btn active" data-market="KR" onclick="selectMarket(this)">KR</button>
<button class="tab-btn" data-market="US_NASDAQ" onclick="selectMarket(this)">US_NASDAQ</button>
<button class="tab-btn" data-market="US_NYSE" onclick="selectMarket(this)">US_NYSE</button>
<button class="tab-btn" data-market="JP" onclick="selectMarket(this)">JP</button>
<button class="tab-btn" data-market="HK" onclick="selectMarket(this)">HK</button>
</div>
</div>
<table class="decisions-table">
<thead>
<tr>
<th>시각</th>
<th>종목</th>
<th>액션</th>
<th>신뢰도</th>
<th>사유</th>
</tr>
</thead>
<tbody id="decisions-body">
<tr class="empty-row"><td colspan="5"><span class="spinner"></span></td></tr>
</tbody>
</table>
<div class="card">
<h1>The Ouroboros Dashboard API</h1>
<p>Use the following endpoints:</p>
<ul>
<li><code>/api/status</code></li>
<li><code>/api/playbook/{date}?market=KR</code></li>
<li><code>/api/scorecard/{date}?market=KR</code></li>
<li><code>/api/performance?market=all</code></li>
<li><code>/api/context/{layer}</code></li>
<li><code>/api/decisions?market=KR</code></li>
<li><code>/api/scenarios/active?market=US</code></li>
</ul>
</div>
</div>
<script>
let pnlChart = null;
let currentDays = 7;
let currentMarket = 'KR';
function fmt(dt) {
try {
const d = new Date(dt);
return d.toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit', hour12: false });
} catch { return dt || '--'; }
}
function fmtPnl(v) {
if (v === null || v === undefined) return '--';
const n = parseFloat(v);
const cls = n > 0 ? 'positive' : n < 0 ? 'negative' : 'neutral';
const sign = n > 0 ? '+' : '';
return `<span class="${cls}">${sign}${n.toFixed(2)}</span>`;
}
function badge(action) {
const a = (action || '').toUpperCase();
const cls = a === 'BUY' ? 'badge-buy' : a === 'SELL' ? 'badge-sell' : 'badge-hold';
return `<span class="badge ${cls}">${a}</span>`;
}
function confBar(conf) {
const pct = Math.min(Math.max(conf || 0, 0), 100);
return `<div class="conf-bar-wrap">
<div class="conf-bar"><div class="conf-fill" style="width:${pct}%"></div></div>
<span class="conf-val">${pct}</span>
</div>`;
}
async function fetchStatus() {
try {
const r = await fetch('/api/status');
if (!r.ok) return;
const d = await r.json();
const t = d.totals || {};
document.getElementById('card-trades').textContent = t.trade_count ?? '--';
const pnlEl = document.getElementById('card-pnl');
const pnlV = t.total_pnl;
if (pnlV !== undefined) {
const n = parseFloat(pnlV);
const sign = n > 0 ? '+' : '';
pnlEl.textContent = `${sign}${n.toFixed(2)}`;
pnlEl.className = `card-value ${n > 0 ? 'positive' : n < 0 ? 'negative' : 'neutral'}`;
}
document.getElementById('card-pnl-sub').textContent = `결정 ${t.decision_count ?? 0}`;
} catch {}
}
async function fetchPerformance() {
try {
const r = await fetch('/api/performance?market=all');
if (!r.ok) return;
const d = await r.json();
const c = d.combined || {};
document.getElementById('card-winrate').textContent = c.win_rate !== undefined ? `${c.win_rate}%` : '--';
document.getElementById('card-total').textContent = c.total_trades ?? '--';
} catch {}
}
async function fetchPnlHistory(days) {
try {
const r = await fetch(`/api/pnl/history?days=${days}`);
if (!r.ok) throw new Error('fetch failed');
const d = await r.json();
renderChart(d);
} catch {
document.getElementById('chart-error').style.display = 'block';
}
}
function renderChart(data) {
const errEl = document.getElementById('chart-error');
if (!data.labels || data.labels.length === 0) {
errEl.style.display = 'block';
return;
}
errEl.style.display = 'none';
const colors = data.pnl.map(v => v >= 0 ? 'rgba(60,179,113,0.75)' : 'rgba(224,85,85,0.75)');
const borderColors = data.pnl.map(v => v >= 0 ? '#3cb371' : '#e05555');
if (pnlChart) { pnlChart.destroy(); pnlChart = null; }
const ctx = document.getElementById('pnl-chart').getContext('2d');
pnlChart = new Chart(ctx, {
type: 'bar',
data: {
labels: data.labels,
datasets: [{
label: 'Daily P&L',
data: data.pnl,
backgroundColor: colors,
borderColor: borderColors,
borderWidth: 1,
borderRadius: 3,
}]
},
options: {
responsive: true,
maintainAspectRatio: false,
plugins: {
legend: { display: false },
tooltip: {
callbacks: {
label: ctx => {
const v = ctx.parsed.y;
const sign = v >= 0 ? '+' : '';
const trades = data.trades[ctx.dataIndex];
return [`P&L: ${sign}${v.toFixed(2)}`, `거래: ${trades}`];
}
}
}
},
scales: {
x: {
ticks: { color: '#9fb3c8', font: { size: 10 }, maxRotation: 0 },
grid: { color: 'rgba(40,69,95,0.4)' }
},
y: {
ticks: { color: '#9fb3c8', font: { size: 10 } },
grid: { color: 'rgba(40,69,95,0.4)' }
}
}
}
});
}
async function fetchDecisions(market) {
const tbody = document.getElementById('decisions-body');
tbody.innerHTML = '<tr class="empty-row"><td colspan="5"><span class="spinner"></span></td></tr>';
try {
const r = await fetch(`/api/decisions?market=${market}&limit=50`);
if (!r.ok) throw new Error('fetch failed');
const d = await r.json();
if (!d.decisions || d.decisions.length === 0) {
tbody.innerHTML = '<tr class="empty-row"><td colspan="5">결정 로그 없음</td></tr>';
return;
}
tbody.innerHTML = d.decisions.map(dec => `
<tr>
<td>${fmt(dec.timestamp)}</td>
<td>${dec.stock_code || '--'}</td>
<td>${badge(dec.action)}</td>
<td>${confBar(dec.confidence)}</td>
<td class="rationale-cell" title="${(dec.rationale || '').replace(/"/g, '&quot;')}">${dec.rationale || '--'}</td>
</tr>
`).join('');
} catch {
tbody.innerHTML = '<tr class="empty-row"><td colspan="5">데이터 로드 실패</td></tr>';
}
}
function selectDays(btn) {
document.querySelectorAll('.day-btn').forEach(b => b.classList.remove('active'));
btn.classList.add('active');
currentDays = parseInt(btn.dataset.days, 10);
fetchPnlHistory(currentDays);
}
function selectMarket(btn) {
document.querySelectorAll('.tab-btn').forEach(b => b.classList.remove('active'));
btn.classList.add('active');
currentMarket = btn.dataset.market;
fetchDecisions(currentMarket);
}
async function refreshAll() {
document.getElementById('last-updated').textContent = '업데이트 중...';
await Promise.all([
fetchStatus(),
fetchPerformance(),
fetchPnlHistory(currentDays),
fetchDecisions(currentMarket),
]);
const now = new Date();
const timeStr = now.toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit', second: '2-digit', hour12: false });
document.getElementById('last-updated').textContent = `마지막 업데이트: ${timeStr}`;
}
// Initial load
refreshAll();
// Auto-refresh every 30 seconds
setInterval(refreshAll, 30000);
</script>
</body>
</html>

View File

@@ -214,42 +214,3 @@ def get_latest_buy_trade(
if not row:
return None
return {"decision_id": row[0], "price": row[1], "quantity": row[2]}
def get_open_position(
conn: sqlite3.Connection, stock_code: str, market: str
) -> dict[str, Any] | None:
"""Return open position if latest trade is BUY, else None."""
cursor = conn.execute(
"""
SELECT action, decision_id, price, quantity
FROM trades
WHERE stock_code = ?
AND market = ?
ORDER BY timestamp DESC
LIMIT 1
""",
(stock_code, market),
)
row = cursor.fetchone()
if not row or row[0] != "BUY":
return None
return {"decision_id": row[1], "price": row[2], "quantity": row[3]}
def get_recent_symbols(
conn: sqlite3.Connection, market: str, limit: int = 30
) -> list[str]:
"""Return recent unique symbols for a market, newest first."""
cursor = conn.execute(
"""
SELECT stock_code, MAX(timestamp) AS last_ts
FROM trades
WHERE market = ?
GROUP BY stock_code
ORDER BY last_ts DESC
LIMIT ?
""",
(market, limit),
)
return [row[0] for row in cursor.fetchall() if row and row[0]]

File diff suppressed because it is too large Load Diff

View File

@@ -123,23 +123,6 @@ MARKETS: dict[str, MarketInfo] = {
),
}
MARKET_SHORTHAND: dict[str, list[str]] = {
"US": ["US_NASDAQ", "US_NYSE", "US_AMEX"],
"CN": ["CN_SHA", "CN_SZA"],
"VN": ["VN_HAN", "VN_HCM"],
}
def expand_market_codes(codes: list[str]) -> list[str]:
"""Expand shorthand market codes into concrete exchange market codes."""
expanded: list[str] = []
for code in codes:
if code in MARKET_SHORTHAND:
expanded.extend(MARKET_SHORTHAND[code])
else:
expanded.append(code)
return expanded
def is_market_open(market: MarketInfo, now: datetime | None = None) -> bool:
"""

View File

@@ -4,9 +4,8 @@ import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, fields
from dataclasses import dataclass
from enum import Enum
from typing import ClassVar
import aiohttp
@@ -59,45 +58,6 @@ class LeakyBucket:
self._tokens -= 1.0
@dataclass
class NotificationFilter:
"""Granular on/off flags for each notification type.
circuit_breaker is intentionally omitted — it is always sent regardless.
"""
# Maps user-facing command keys to dataclass field names
KEYS: ClassVar[dict[str, str]] = {
"trades": "trades",
"market": "market_open_close",
"fatfinger": "fat_finger",
"system": "system_events",
"playbook": "playbook",
"scenario": "scenario_match",
"errors": "errors",
}
trades: bool = True
market_open_close: bool = True
fat_finger: bool = True
system_events: bool = True
playbook: bool = True
scenario_match: bool = True
errors: bool = True
def set_flag(self, key: str, value: bool) -> bool:
"""Set a filter flag by user-facing key. Returns False if key is unknown."""
field = self.KEYS.get(key.lower())
if field is None:
return False
setattr(self, field, value)
return True
def as_dict(self) -> dict[str, bool]:
"""Return {user_key: current_value} for display."""
return {k: getattr(self, field) for k, field in self.KEYS.items()}
@dataclass
class NotificationMessage:
"""Internal notification message structure."""
@@ -119,7 +79,6 @@ class TelegramClient:
chat_id: str | None = None,
enabled: bool = True,
rate_limit: float = DEFAULT_RATE,
notification_filter: NotificationFilter | None = None,
) -> None:
"""
Initialize Telegram client.
@@ -129,14 +88,12 @@ class TelegramClient:
chat_id: Target chat ID (user or group)
enabled: Enable/disable notifications globally
rate_limit: Maximum messages per second
notification_filter: Granular per-type on/off flags
"""
self._bot_token = bot_token
self._chat_id = chat_id
self._enabled = enabled
self._rate_limiter = LeakyBucket(rate=rate_limit)
self._session: aiohttp.ClientSession | None = None
self._filter = notification_filter if notification_filter is not None else NotificationFilter()
if not enabled:
logger.info("Telegram notifications disabled via configuration")
@@ -161,26 +118,6 @@ class TelegramClient:
if self._session is not None and not self._session.closed:
await self._session.close()
def set_notification(self, key: str, value: bool) -> bool:
"""Toggle a notification type by user-facing key at runtime.
Args:
key: User-facing key (e.g. "scenario", "market", "all")
value: True to enable, False to disable
Returns:
True if key was valid, False if unknown.
"""
if key == "all":
for k in NotificationFilter.KEYS:
self._filter.set_flag(k, value)
return True
return self._filter.set_flag(key, value)
def filter_status(self) -> dict[str, bool]:
"""Return current per-type filter state keyed by user-facing names."""
return self._filter.as_dict()
async def send_message(self, text: str, parse_mode: str = "HTML") -> bool:
"""
Send a generic text message to Telegram.
@@ -256,8 +193,6 @@ class TelegramClient:
price: Execution price
confidence: AI confidence level (0-100)
"""
if not self._filter.trades:
return
emoji = "🟢" if action == "BUY" else "🔴"
message = (
f"<b>{emoji} {action}</b>\n"
@@ -277,8 +212,6 @@ class TelegramClient:
Args:
market_name: Name of the market (e.g., "Korea", "United States")
"""
if not self._filter.market_open_close:
return
message = f"<b>Market Open</b>\n{market_name} trading session started"
await self._send_notification(
NotificationMessage(priority=NotificationPriority.LOW, message=message)
@@ -292,8 +225,6 @@ class TelegramClient:
market_name: Name of the market
pnl_pct: Final P&L percentage for the session
"""
if not self._filter.market_open_close:
return
pnl_sign = "+" if pnl_pct >= 0 else ""
pnl_emoji = "📈" if pnl_pct >= 0 else "📉"
message = (
@@ -340,8 +271,6 @@ class TelegramClient:
total_cash: Total available cash
max_pct: Maximum allowed percentage
"""
if not self._filter.fat_finger:
return
attempted_pct = (order_amount / total_cash) * 100 if total_cash > 0 else 0
message = (
f"<b>Fat-Finger Protection</b>\n"
@@ -364,8 +293,6 @@ class TelegramClient:
mode: Trading mode ("paper" or "live")
enabled_markets: List of enabled market codes
"""
if not self._filter.system_events:
return
mode_emoji = "📝" if mode == "paper" else "💰"
markets_str = ", ".join(enabled_markets)
message = (
@@ -393,8 +320,6 @@ class TelegramClient:
scenario_count: Total number of scenarios
token_count: Gemini token usage for the playbook
"""
if not self._filter.playbook:
return
message = (
f"<b>Playbook Generated</b>\n"
f"Market: {market}\n"
@@ -422,8 +347,6 @@ class TelegramClient:
condition_summary: Short summary of the matched condition
confidence: Scenario confidence (0-100)
"""
if not self._filter.scenario_match:
return
message = (
f"<b>Scenario Matched</b>\n"
f"Symbol: <code>{stock_code}</code>\n"
@@ -443,8 +366,6 @@ class TelegramClient:
market: Market code (e.g., "KR", "US")
reason: Failure reason summary
"""
if not self._filter.playbook:
return
message = (
f"<b>Playbook Failed</b>\n"
f"Market: {market}\n"
@@ -461,8 +382,6 @@ class TelegramClient:
Args:
reason: Reason for shutdown (e.g., "Normal shutdown", "Circuit breaker")
"""
if not self._filter.system_events:
return
message = f"<b>System Shutdown</b>\n{reason}"
priority = (
NotificationPriority.CRITICAL
@@ -484,8 +403,6 @@ class TelegramClient:
error_msg: Error message
context: Error context (e.g., stock code, market)
"""
if not self._filter.errors:
return
message = (
f"<b>Error: {error_type}</b>\n"
f"Context: {context}\n"
@@ -512,7 +429,6 @@ class TelegramCommandHandler:
self._client = client
self._polling_interval = polling_interval
self._commands: dict[str, Callable[[], Awaitable[None]]] = {}
self._commands_with_args: dict[str, Callable[[list[str]], Awaitable[None]]] = {}
self._last_update_id = 0
self._polling_task: asyncio.Task[None] | None = None
self._running = False
@@ -521,7 +437,7 @@ class TelegramCommandHandler:
self, command: str, handler: Callable[[], Awaitable[None]]
) -> None:
"""
Register a command handler (no arguments).
Register a command handler.
Args:
command: Command name (without leading slash, e.g., "start")
@@ -530,19 +446,6 @@ class TelegramCommandHandler:
self._commands[command] = handler
logger.debug("Registered command handler: /%s", command)
def register_command_with_args(
self, command: str, handler: Callable[[list[str]], Awaitable[None]]
) -> None:
"""
Register a command handler that receives trailing arguments.
Args:
command: Command name (without leading slash, e.g., "notify")
handler: Async function receiving list of argument tokens
"""
self._commands_with_args[command] = handler
logger.debug("Registered command handler (with args): /%s", command)
async def start_polling(self) -> None:
"""Start long polling for commands."""
if self._running:
@@ -663,14 +566,11 @@ class TelegramCommandHandler:
# Remove @botname suffix if present (for group chats)
command_name = command_parts[0].split("@")[0]
# Execute handler (args-aware handlers take priority)
args_handler = self._commands_with_args.get(command_name)
if args_handler:
logger.info("Executing command: /%s %s", command_name, command_parts[1:])
await args_handler(command_parts[1:])
elif command_name in self._commands:
# Execute handler
handler = self._commands.get(command_name)
if handler:
logger.info("Executing command: /%s", command_name)
await self._commands[command_name]()
await handler()
else:
logger.debug("Unknown command: /%s", command_name)
await self._client.send_message(

View File

@@ -1,8 +1,7 @@
"""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 smart rule-based fallback playbook that uses scanner signals
(momentum/oversold) to generate BUY conditions, avoiding the all-HOLD problem.
On failure, returns a defensive playbook (all HOLD, no trades).
"""
from __future__ import annotations
@@ -135,7 +134,7 @@ class PreMarketPlanner:
except Exception:
logger.exception("Playbook generation failed for %s", market)
if self._settings.DEFENSIVE_PLAYBOOK_ON_FAILURE:
return self._smart_fallback_playbook(today, market, candidates, self._settings)
return self._defensive_playbook(today, market, candidates)
return self._empty_playbook(today, market)
def build_cross_market_context(
@@ -471,99 +470,3 @@ class PreMarketPlanner:
),
],
)
@staticmethod
def _smart_fallback_playbook(
today: date,
market: str,
candidates: list[ScanCandidate],
settings: Settings,
) -> DayPlaybook:
"""Rule-based fallback playbook when Gemini is unavailable.
Uses scanner signals (RSI, volume_ratio) to generate meaningful BUY
conditions instead of the all-SELL defensive playbook. Candidates are
already pre-qualified by SmartVolatilityScanner, so we trust their
signals and build actionable scenarios from them.
Scenario logic per candidate:
- momentum signal: BUY when volume_ratio exceeds scanner threshold
- oversold signal: BUY when RSI is below oversold threshold
- always: SELL stop-loss at -3.0% as guard
"""
stock_playbooks = []
for c in candidates:
scenarios: list[StockScenario] = []
if c.signal == "momentum":
scenarios.append(
StockScenario(
condition=StockCondition(
volume_ratio_above=settings.VOL_MULTIPLIER,
),
action=ScenarioAction.BUY,
confidence=80,
allocation_pct=10.0,
stop_loss_pct=-3.0,
take_profit_pct=5.0,
rationale=(
f"Rule-based BUY: momentum signal, "
f"volume={c.volume_ratio:.1f}x (fallback planner)"
),
)
)
elif c.signal == "oversold":
scenarios.append(
StockScenario(
condition=StockCondition(
rsi_below=settings.RSI_OVERSOLD_THRESHOLD,
),
action=ScenarioAction.BUY,
confidence=80,
allocation_pct=10.0,
stop_loss_pct=-3.0,
take_profit_pct=5.0,
rationale=(
f"Rule-based BUY: oversold signal, "
f"RSI={c.rsi:.0f} (fallback planner)"
),
)
)
# Always add stop-loss guard
scenarios.append(
StockScenario(
condition=StockCondition(price_change_pct_below=-3.0),
action=ScenarioAction.SELL,
confidence=90,
stop_loss_pct=-3.0,
rationale="Rule-based stop-loss (fallback planner)",
)
)
stock_playbooks.append(
StockPlaybook(
stock_code=c.stock_code,
scenarios=scenarios,
)
)
logger.info(
"Smart fallback playbook for %s: %d stocks with rule-based BUY/SELL conditions",
market,
len(stock_playbooks),
)
return DayPlaybook(
date=today,
market=market,
market_outlook=MarketOutlook.NEUTRAL,
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

@@ -2,10 +2,6 @@
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.brain.gemini_client import GeminiClient
# ---------------------------------------------------------------------------
@@ -274,97 +270,3 @@ class TestBatchDecisionParsing:
assert decisions["AAPL"].action == "HOLD"
assert decisions["AAPL"].confidence == 0
# ---------------------------------------------------------------------------
# Prompt Override (used by pre_market_planner)
# ---------------------------------------------------------------------------
class TestPromptOverride:
"""decide() must use prompt_override when present in market_data."""
@pytest.mark.asyncio
async def test_prompt_override_is_sent_to_gemini(self, settings):
"""When prompt_override is in market_data, it should be used as the prompt."""
client = GeminiClient(settings)
custom_prompt = "You are a playbook generator. Return JSON with scenarios."
mock_response = MagicMock()
mock_response.text = '{"action": "HOLD", "confidence": 50, "rationale": "test"}'
with patch.object(
client._client.aio.models,
"generate_content",
new_callable=AsyncMock,
return_value=mock_response,
) as mock_generate:
market_data = {
"stock_code": "PLANNER",
"current_price": 0,
"prompt_override": custom_prompt,
}
await client.decide(market_data)
# Verify the custom prompt was sent, not a built prompt
mock_generate.assert_called_once()
actual_prompt = mock_generate.call_args[1].get(
"contents", mock_generate.call_args[0][1] if len(mock_generate.call_args[0]) > 1 else None
)
assert actual_prompt == custom_prompt
@pytest.mark.asyncio
async def test_prompt_override_skips_optimization(self, settings):
"""prompt_override should bypass prompt optimization."""
client = GeminiClient(settings)
client._enable_optimization = True
custom_prompt = "Custom playbook prompt"
mock_response = MagicMock()
mock_response.text = '{"action": "HOLD", "confidence": 50, "rationale": "ok"}'
with patch.object(
client._client.aio.models,
"generate_content",
new_callable=AsyncMock,
return_value=mock_response,
) as mock_generate:
market_data = {
"stock_code": "PLANNER",
"current_price": 0,
"prompt_override": custom_prompt,
}
await client.decide(market_data)
actual_prompt = mock_generate.call_args[1].get(
"contents", mock_generate.call_args[0][1] if len(mock_generate.call_args[0]) > 1 else None
)
assert actual_prompt == custom_prompt
@pytest.mark.asyncio
async def test_without_prompt_override_uses_build_prompt(self, settings):
"""Without prompt_override, decide() should use build_prompt as before."""
client = GeminiClient(settings)
mock_response = MagicMock()
mock_response.text = '{"action": "HOLD", "confidence": 50, "rationale": "ok"}'
with patch.object(
client._client.aio.models,
"generate_content",
new_callable=AsyncMock,
return_value=mock_response,
) as mock_generate:
market_data = {
"stock_code": "005930",
"current_price": 72000,
}
await client.decide(market_data)
actual_prompt = mock_generate.call_args[1].get(
"contents", mock_generate.call_args[0][1] if len(mock_generate.call_args[0]) > 1 else None
)
# Should contain stock code from build_prompt, not be a custom override
assert "005930" in actual_prompt

View File

@@ -3,7 +3,7 @@
from __future__ import annotations
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, patch
import pytest
@@ -90,12 +90,12 @@ class TestTokenManagement:
await broker.close()
@pytest.mark.asyncio
async def test_token_refresh_cooldown_waits_then_retries(self, settings):
"""Token refresh should wait out cooldown then retry (issue #54)."""
async def test_token_refresh_cooldown_prevents_rapid_retries(self, settings):
"""Token refresh should enforce cooldown after failure (issue #54)."""
broker = KISBroker(settings)
broker._refresh_cooldown = 0.1 # Short cooldown for testing
broker._refresh_cooldown = 2.0 # Short cooldown for testing
# All attempts fail with 403 (EGW00133)
# First refresh attempt fails with 403 (EGW00133)
mock_resp_403 = AsyncMock()
mock_resp_403.status = 403
mock_resp_403.text = AsyncMock(
@@ -109,8 +109,8 @@ class TestTokenManagement:
with pytest.raises(ConnectionError, match="Token refresh failed"):
await broker._ensure_token()
# Second attempt within cooldown should wait then retry (and still get 403)
with pytest.raises(ConnectionError, match="Token refresh failed"):
# Second attempt within cooldown should fail with cooldown error
with pytest.raises(ConnectionError, match="Token refresh on cooldown"):
await broker._ensure_token()
await broker.close()
@@ -296,280 +296,3 @@ class TestHashKey:
mock_acquire.assert_called_once()
await broker.close()
# ---------------------------------------------------------------------------
# fetch_market_rankings — TR_ID, path, params (issue #155)
# ---------------------------------------------------------------------------
def _make_ranking_mock(items: list[dict]) -> AsyncMock:
"""Build a mock HTTP response returning ranking items."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"output": items})
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
return mock_resp
class TestFetchMarketRankings:
"""Verify correct TR_ID, API path, and params per ranking_type (issue #155)."""
@pytest.fixture
def broker(self, settings) -> KISBroker:
b = KISBroker(settings)
b._access_token = "tok"
b._token_expires_at = float("inf")
b._rate_limiter.acquire = AsyncMock()
return b
@pytest.mark.asyncio
async def test_volume_uses_correct_tr_id_and_path(self, broker: KISBroker) -> None:
mock_resp = _make_ranking_mock([])
with patch("aiohttp.ClientSession.get", return_value=mock_resp) as mock_get:
await broker.fetch_market_rankings(ranking_type="volume")
call_kwargs = mock_get.call_args
url = call_kwargs[0][0] if call_kwargs[0] else call_kwargs[1].get("url", "")
headers = call_kwargs[1].get("headers", {})
params = call_kwargs[1].get("params", {})
assert "volume-rank" in url
assert headers.get("tr_id") == "FHPST01710000"
assert params.get("FID_COND_SCR_DIV_CODE") == "20171"
assert params.get("FID_TRGT_EXLS_CLS_CODE") == "0000000000"
@pytest.mark.asyncio
async def test_fluctuation_uses_correct_tr_id_and_path(self, broker: KISBroker) -> None:
mock_resp = _make_ranking_mock([])
with patch("aiohttp.ClientSession.get", return_value=mock_resp) as mock_get:
await broker.fetch_market_rankings(ranking_type="fluctuation")
call_kwargs = mock_get.call_args
url = call_kwargs[0][0] if call_kwargs[0] else call_kwargs[1].get("url", "")
headers = call_kwargs[1].get("headers", {})
params = call_kwargs[1].get("params", {})
assert "ranking/fluctuation" in url
assert headers.get("tr_id") == "FHPST01700000"
assert params.get("fid_cond_scr_div_code") == "20170"
@pytest.mark.asyncio
async def test_volume_returns_parsed_rows(self, broker: KISBroker) -> None:
items = [
{
"mksc_shrn_iscd": "005930",
"hts_kor_isnm": "삼성전자",
"stck_prpr": "75000",
"acml_vol": "10000000",
"prdy_ctrt": "2.5",
"vol_inrt": "150",
}
]
mock_resp = _make_ranking_mock(items)
with patch("aiohttp.ClientSession.get", return_value=mock_resp):
result = await broker.fetch_market_rankings(ranking_type="volume")
assert len(result) == 1
assert result[0]["stock_code"] == "005930"
assert result[0]["price"] == 75000.0
assert result[0]["change_rate"] == 2.5
# ---------------------------------------------------------------------------
# KRX tick unit / round-down helpers (issue #157)
# ---------------------------------------------------------------------------
from src.broker.kis_api import kr_tick_unit, kr_round_down # noqa: E402
class TestKrTickUnit:
"""kr_tick_unit and kr_round_down must implement KRX price tick rules."""
@pytest.mark.parametrize(
"price, expected_tick",
[
(1999, 1),
(2000, 5),
(4999, 5),
(5000, 10),
(19999, 10),
(20000, 50),
(49999, 50),
(50000, 100),
(199999, 100),
(200000, 500),
(499999, 500),
(500000, 1000),
(1000000, 1000),
],
)
def test_tick_unit_boundaries(self, price: int, expected_tick: int) -> None:
assert kr_tick_unit(price) == expected_tick
@pytest.mark.parametrize(
"price, expected_rounded",
[
(188150, 188100), # 100원 단위, 50원 잔여 → 내림
(188100, 188100), # 이미 정렬됨
(75050, 75000), # 100원 단위, 50원 잔여 → 내림
(49950, 49950), # 50원 단위 정렬됨
(49960, 49950), # 50원 단위, 10원 잔여 → 내림
(1999, 1999), # 1원 단위 → 그대로
(5003, 5000), # 10원 단위, 3원 잔여 → 내림
],
)
def test_round_down_to_tick(self, price: int, expected_rounded: int) -> None:
assert kr_round_down(price) == expected_rounded
# ---------------------------------------------------------------------------
# get_current_price (issue #157)
# ---------------------------------------------------------------------------
class TestGetCurrentPrice:
"""get_current_price must use inquire-price API and return (price, change, foreigner)."""
@pytest.fixture
def broker(self, settings) -> KISBroker:
b = KISBroker(settings)
b._access_token = "tok"
b._token_expires_at = float("inf")
b._rate_limiter.acquire = AsyncMock()
return b
@pytest.mark.asyncio
async def test_returns_correct_fields(self, broker: KISBroker) -> None:
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(
return_value={
"rt_cd": "0",
"output": {
"stck_prpr": "188600",
"prdy_ctrt": "3.97",
"frgn_ntby_qty": "12345",
},
}
)
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.get", return_value=mock_resp) as mock_get:
price, change_pct, foreigner = await broker.get_current_price("005930")
assert price == 188600.0
assert change_pct == 3.97
assert foreigner == 12345.0
call_kwargs = mock_get.call_args
url = call_kwargs[0][0] if call_kwargs[0] else call_kwargs[1].get("url", "")
headers = call_kwargs[1].get("headers", {})
assert "inquire-price" in url
assert headers.get("tr_id") == "FHKST01010100"
@pytest.mark.asyncio
async def test_http_error_raises_connection_error(self, broker: KISBroker) -> None:
mock_resp = AsyncMock()
mock_resp.status = 500
mock_resp.text = AsyncMock(return_value="Internal Server Error")
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
with patch("aiohttp.ClientSession.get", return_value=mock_resp):
with pytest.raises(ConnectionError, match="get_current_price failed"):
await broker.get_current_price("005930")
# ---------------------------------------------------------------------------
# send_order tick rounding and ORD_DVSN (issue #157)
# ---------------------------------------------------------------------------
class TestSendOrderTickRounding:
"""send_order must apply KRX tick rounding and correct ORD_DVSN codes."""
@pytest.fixture
def broker(self, settings) -> KISBroker:
b = KISBroker(settings)
b._access_token = "tok"
b._token_expires_at = float("inf")
b._rate_limiter.acquire = AsyncMock()
return b
@pytest.mark.asyncio
async def test_limit_order_rounds_down_to_tick(self, broker: KISBroker) -> None:
"""Price 188150 (not on 100-won tick) must be rounded to 188100."""
mock_hash = AsyncMock()
mock_hash.status = 200
mock_hash.json = AsyncMock(return_value={"HASH": "h"})
mock_hash.__aenter__ = AsyncMock(return_value=mock_hash)
mock_hash.__aexit__ = AsyncMock(return_value=False)
mock_order = AsyncMock()
mock_order.status = 200
mock_order.json = AsyncMock(return_value={"rt_cd": "0"})
mock_order.__aenter__ = AsyncMock(return_value=mock_order)
mock_order.__aexit__ = AsyncMock(return_value=False)
with patch(
"aiohttp.ClientSession.post", side_effect=[mock_hash, mock_order]
) as mock_post:
await broker.send_order("005930", "BUY", 1, price=188150)
order_call = mock_post.call_args_list[1]
body = order_call[1].get("json", {})
assert body["ORD_UNPR"] == "188100" # rounded down
assert body["ORD_DVSN"] == "00" # 지정가
@pytest.mark.asyncio
async def test_limit_order_ord_dvsn_is_00(self, broker: KISBroker) -> None:
"""send_order with price>0 must use ORD_DVSN='00' (지정가)."""
mock_hash = AsyncMock()
mock_hash.status = 200
mock_hash.json = AsyncMock(return_value={"HASH": "h"})
mock_hash.__aenter__ = AsyncMock(return_value=mock_hash)
mock_hash.__aexit__ = AsyncMock(return_value=False)
mock_order = AsyncMock()
mock_order.status = 200
mock_order.json = AsyncMock(return_value={"rt_cd": "0"})
mock_order.__aenter__ = AsyncMock(return_value=mock_order)
mock_order.__aexit__ = AsyncMock(return_value=False)
with patch(
"aiohttp.ClientSession.post", side_effect=[mock_hash, mock_order]
) as mock_post:
await broker.send_order("005930", "BUY", 1, price=50000)
order_call = mock_post.call_args_list[1]
body = order_call[1].get("json", {})
assert body["ORD_DVSN"] == "00"
@pytest.mark.asyncio
async def test_market_order_ord_dvsn_is_01(self, broker: KISBroker) -> None:
"""send_order with price=0 must use ORD_DVSN='01' (시장가)."""
mock_hash = AsyncMock()
mock_hash.status = 200
mock_hash.json = AsyncMock(return_value={"HASH": "h"})
mock_hash.__aenter__ = AsyncMock(return_value=mock_hash)
mock_hash.__aexit__ = AsyncMock(return_value=False)
mock_order = AsyncMock()
mock_order.status = 200
mock_order.json = AsyncMock(return_value={"rt_cd": "0"})
mock_order.__aenter__ = AsyncMock(return_value=mock_order)
mock_order.__aexit__ = AsyncMock(return_value=False)
with patch(
"aiohttp.ClientSession.post", side_effect=[mock_hash, mock_order]
) as mock_post:
await broker.send_order("005930", "SELL", 1, price=0)
order_call = mock_post.call_args_list[1]
body = order_call[1].get("json", {})
assert body["ORD_DVSN"] == "01"
assert body["ORD_UNPR"] == "0"

View File

@@ -1,25 +1,21 @@
"""Tests for dashboard endpoint handlers."""
"""Tests for FastAPI dashboard endpoints."""
from __future__ import annotations
import json
import sqlite3
from collections.abc import Callable
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
import pytest
from fastapi import HTTPException
from fastapi.responses import FileResponse
pytest.importorskip("fastapi")
from fastapi.testclient import TestClient
from src.dashboard.app import create_dashboard_app
from src.db import init_db
def _seed_db(conn: sqlite3.Connection) -> None:
today = datetime.now(UTC).date().isoformat()
conn.execute(
"""
INSERT INTO playbooks (
@@ -38,24 +34,6 @@ def _seed_db(conn: sqlite3.Connection) -> None:
1,
),
)
conn.execute(
"""
INSERT INTO playbooks (
date, market, status, playbook_json, generated_at,
token_count, scenario_count, match_count
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
today,
"US_NASDAQ",
"ready",
json.dumps({"market": "US_NASDAQ", "stock_playbooks": []}),
f"{today}T08:30:00+00:00",
100,
1,
0,
),
)
conn.execute(
"""
INSERT INTO contexts (layer, timeframe, key, value, created_at, updated_at)
@@ -93,7 +71,7 @@ def _seed_db(conn: sqlite3.Connection) -> None:
""",
(
"d-kr-1",
f"{today}T09:10:00+00:00",
"2026-02-14T09:10:00+00:00",
"005930",
"KR",
"KRX",
@@ -113,9 +91,9 @@ def _seed_db(conn: sqlite3.Connection) -> None:
""",
(
"d-us-1",
f"{today}T21:10:00+00:00",
"2026-02-14T21:10:00+00:00",
"AAPL",
"US_NASDAQ",
"US",
"NASDAQ",
"SELL",
80,
@@ -132,7 +110,7 @@ def _seed_db(conn: sqlite3.Connection) -> None:
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
f"{today}T09:11:00+00:00",
"2026-02-14T09:11:00+00:00",
"005930",
"BUY",
85,
@@ -154,7 +132,7 @@ def _seed_db(conn: sqlite3.Connection) -> None:
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
f"{today}T21:11:00+00:00",
"2026-02-14T21:11:00+00:00",
"AAPL",
"SELL",
80,
@@ -162,7 +140,7 @@ def _seed_db(conn: sqlite3.Connection) -> None:
1,
200,
-1.0,
"US_NASDAQ",
"US",
"NASDAQ",
None,
"d-us-1",
@@ -171,148 +149,122 @@ def _seed_db(conn: sqlite3.Connection) -> None:
conn.commit()
def _app(tmp_path: Path) -> Any:
def _client(tmp_path: Path) -> TestClient:
db_path = tmp_path / "dashboard_test.db"
conn = init_db(str(db_path))
_seed_db(conn)
conn.close()
return create_dashboard_app(str(db_path))
def _endpoint(app: Any, path: str) -> Callable[..., Any]:
for route in app.routes:
if getattr(route, "path", None) == path:
return route.endpoint
raise AssertionError(f"route not found: {path}")
app = create_dashboard_app(str(db_path))
return TestClient(app)
def test_index_serves_html(tmp_path: Path) -> None:
app = _app(tmp_path)
index = _endpoint(app, "/")
resp = index()
assert isinstance(resp, FileResponse)
assert "index.html" in str(resp.path)
client = _client(tmp_path)
resp = client.get("/")
assert resp.status_code == 200
assert "The Ouroboros Dashboard API" in resp.text
def test_status_endpoint(tmp_path: Path) -> None:
app = _app(tmp_path)
get_status = _endpoint(app, "/api/status")
body = get_status()
client = _client(tmp_path)
resp = client.get("/api/status")
assert resp.status_code == 200
body = resp.json()
assert "KR" in body["markets"]
assert "US_NASDAQ" in body["markets"]
assert "US" in body["markets"]
assert "totals" in body
def test_playbook_found(tmp_path: Path) -> None:
app = _app(tmp_path)
get_playbook = _endpoint(app, "/api/playbook/{date_str}")
body = get_playbook("2026-02-14", market="KR")
assert body["market"] == "KR"
client = _client(tmp_path)
resp = client.get("/api/playbook/2026-02-14?market=KR")
assert resp.status_code == 200
assert resp.json()["market"] == "KR"
def test_playbook_not_found(tmp_path: Path) -> None:
app = _app(tmp_path)
get_playbook = _endpoint(app, "/api/playbook/{date_str}")
with pytest.raises(HTTPException, match="playbook not found"):
get_playbook("2026-02-15", market="KR")
client = _client(tmp_path)
resp = client.get("/api/playbook/2026-02-15?market=KR")
assert resp.status_code == 404
def test_scorecard_found(tmp_path: Path) -> None:
app = _app(tmp_path)
get_scorecard = _endpoint(app, "/api/scorecard/{date_str}")
body = get_scorecard("2026-02-14", market="KR")
assert body["scorecard"]["total_pnl"] == 1.5
client = _client(tmp_path)
resp = client.get("/api/scorecard/2026-02-14?market=KR")
assert resp.status_code == 200
assert resp.json()["scorecard"]["total_pnl"] == 1.5
def test_scorecard_not_found(tmp_path: Path) -> None:
app = _app(tmp_path)
get_scorecard = _endpoint(app, "/api/scorecard/{date_str}")
with pytest.raises(HTTPException, match="scorecard not found"):
get_scorecard("2026-02-15", market="KR")
client = _client(tmp_path)
resp = client.get("/api/scorecard/2026-02-15?market=KR")
assert resp.status_code == 404
def test_performance_all(tmp_path: Path) -> None:
app = _app(tmp_path)
get_performance = _endpoint(app, "/api/performance")
body = get_performance(market="all")
client = _client(tmp_path)
resp = client.get("/api/performance?market=all")
assert resp.status_code == 200
body = resp.json()
assert body["market"] == "all"
assert body["combined"]["total_trades"] == 2
assert len(body["by_market"]) == 2
def test_performance_market_filter(tmp_path: Path) -> None:
app = _app(tmp_path)
get_performance = _endpoint(app, "/api/performance")
body = get_performance(market="KR")
client = _client(tmp_path)
resp = client.get("/api/performance?market=KR")
assert resp.status_code == 200
body = resp.json()
assert body["market"] == "KR"
assert body["metrics"]["total_trades"] == 1
def test_performance_empty_market(tmp_path: Path) -> None:
app = _app(tmp_path)
get_performance = _endpoint(app, "/api/performance")
body = get_performance(market="JP")
assert body["metrics"]["total_trades"] == 0
client = _client(tmp_path)
resp = client.get("/api/performance?market=JP")
assert resp.status_code == 200
assert resp.json()["metrics"]["total_trades"] == 0
def test_context_layer_all(tmp_path: Path) -> None:
app = _app(tmp_path)
get_context_layer = _endpoint(app, "/api/context/{layer}")
body = get_context_layer("L7_REALTIME", timeframe=None, limit=100)
client = _client(tmp_path)
resp = client.get("/api/context/L7_REALTIME")
assert resp.status_code == 200
body = resp.json()
assert body["layer"] == "L7_REALTIME"
assert body["count"] == 1
def test_context_layer_timeframe_filter(tmp_path: Path) -> None:
app = _app(tmp_path)
get_context_layer = _endpoint(app, "/api/context/{layer}")
body = get_context_layer("L6_DAILY", timeframe="2026-02-14", limit=100)
client = _client(tmp_path)
resp = client.get("/api/context/L6_DAILY?timeframe=2026-02-14")
assert resp.status_code == 200
body = resp.json()
assert body["count"] == 1
assert body["entries"][0]["key"] == "scorecard_KR"
def test_decisions_endpoint(tmp_path: Path) -> None:
app = _app(tmp_path)
get_decisions = _endpoint(app, "/api/decisions")
body = get_decisions(market="KR", limit=50)
client = _client(tmp_path)
resp = client.get("/api/decisions?market=KR")
assert resp.status_code == 200
body = resp.json()
assert body["count"] == 1
assert body["decisions"][0]["decision_id"] == "d-kr-1"
def test_scenarios_active_filters_non_matched(tmp_path: Path) -> None:
app = _app(tmp_path)
get_active_scenarios = _endpoint(app, "/api/scenarios/active")
body = get_active_scenarios(
market="KR",
date_str=datetime.now(UTC).date().isoformat(),
limit=50,
)
client = _client(tmp_path)
resp = client.get("/api/scenarios/active?market=KR&date_str=2026-02-14")
assert resp.status_code == 200
body = resp.json()
assert body["count"] == 1
assert body["matches"][0]["stock_code"] == "005930"
def test_scenarios_active_empty_when_no_matches(tmp_path: Path) -> None:
app = _app(tmp_path)
get_active_scenarios = _endpoint(app, "/api/scenarios/active")
body = get_active_scenarios(market="US", date_str="2026-02-14", limit=50)
assert body["count"] == 0
def test_pnl_history_all_markets(tmp_path: Path) -> None:
app = _app(tmp_path)
get_pnl_history = _endpoint(app, "/api/pnl/history")
body = get_pnl_history(days=30, market="all")
assert body["market"] == "all"
assert isinstance(body["labels"], list)
assert isinstance(body["pnl"], list)
assert len(body["labels"]) == len(body["pnl"])
def test_pnl_history_market_filter(tmp_path: Path) -> None:
app = _app(tmp_path)
get_pnl_history = _endpoint(app, "/api/pnl/history")
body = get_pnl_history(days=30, market="KR")
assert body["market"] == "KR"
# KR has 1 trade with pnl=2.0
assert len(body["labels"]) >= 1
assert body["pnl"][0] == 2.0
client = _client(tmp_path)
resp = client.get("/api/scenarios/active?market=US&date_str=2026-02-14")
assert resp.status_code == 200
assert resp.json()["count"] == 0

View File

@@ -1,60 +0,0 @@
"""Tests for database helper functions."""
from src.db import get_open_position, init_db, log_trade
def test_get_open_position_returns_latest_buy() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="005930",
action="BUY",
confidence=90,
rationale="entry",
quantity=2,
price=70000.0,
market="KR",
exchange_code="KRX",
decision_id="d-buy-1",
)
position = get_open_position(conn, "005930", "KR")
assert position is not None
assert position["decision_id"] == "d-buy-1"
assert position["price"] == 70000.0
assert position["quantity"] == 2
def test_get_open_position_returns_none_when_latest_is_sell() -> None:
conn = init_db(":memory:")
log_trade(
conn=conn,
stock_code="005930",
action="BUY",
confidence=90,
rationale="entry",
quantity=1,
price=70000.0,
market="KR",
exchange_code="KRX",
decision_id="d-buy-1",
)
log_trade(
conn=conn,
stock_code="005930",
action="SELL",
confidence=95,
rationale="exit",
quantity=1,
price=71000.0,
market="KR",
exchange_code="KRX",
decision_id="d-sell-1",
)
assert get_open_position(conn, "005930", "KR") is None
def test_get_open_position_returns_none_when_no_trades() -> None:
conn = init_db(":memory:")
assert get_open_position(conn, "AAPL", "US_NASDAQ") is None

View File

@@ -14,9 +14,6 @@ from src.evolution.scorecard import DailyScorecard
from src.logging.decision_logger import DecisionLogger
from src.main import (
_apply_dashboard_flag,
_determine_order_quantity,
_extract_held_codes_from_balance,
_extract_held_qty_from_balance,
_handle_market_close,
_run_context_scheduler,
_run_evolution_loop,
@@ -71,141 +68,6 @@ def _make_sell_match(stock_code: str = "005930") -> ScenarioMatch:
)
class TestExtractHeldQtyFromBalance:
"""Tests for _extract_held_qty_from_balance()."""
def _domestic_balance(self, stock_code: str, ord_psbl_qty: int) -> dict:
return {
"output1": [{"pdno": stock_code, "ord_psbl_qty": str(ord_psbl_qty)}],
"output2": [{"dnca_tot_amt": "1000000"}],
}
def test_domestic_returns_ord_psbl_qty(self) -> None:
balance = self._domestic_balance("005930", 7)
assert _extract_held_qty_from_balance(balance, "005930", is_domestic=True) == 7
def test_domestic_fallback_to_hldg_qty(self) -> None:
balance = {"output1": [{"pdno": "005930", "hldg_qty": "3"}]}
assert _extract_held_qty_from_balance(balance, "005930", is_domestic=True) == 3
def test_domestic_returns_zero_when_not_found(self) -> None:
balance = self._domestic_balance("005930", 5)
assert _extract_held_qty_from_balance(balance, "000660", is_domestic=True) == 0
def test_domestic_returns_zero_when_output1_empty(self) -> None:
balance = {"output1": [], "output2": [{}]}
assert _extract_held_qty_from_balance(balance, "005930", is_domestic=True) == 0
def test_overseas_returns_ovrs_cblc_qty(self) -> None:
balance = {"output1": [{"ovrs_pdno": "AAPL", "ovrs_cblc_qty": "10"}]}
assert _extract_held_qty_from_balance(balance, "AAPL", is_domestic=False) == 10
def test_overseas_fallback_to_hldg_qty(self) -> None:
balance = {"output1": [{"ovrs_pdno": "AAPL", "hldg_qty": "4"}]}
assert _extract_held_qty_from_balance(balance, "AAPL", is_domestic=False) == 4
def test_case_insensitive_match(self) -> None:
balance = {"output1": [{"pdno": "005930", "ord_psbl_qty": "2"}]}
assert _extract_held_qty_from_balance(balance, "005930", is_domestic=True) == 2
class TestExtractHeldCodesFromBalance:
"""Tests for _extract_held_codes_from_balance()."""
def test_returns_codes_with_positive_qty(self) -> None:
balance = {
"output1": [
{"pdno": "005930", "ord_psbl_qty": "5"},
{"pdno": "000660", "ord_psbl_qty": "3"},
]
}
result = _extract_held_codes_from_balance(balance, is_domestic=True)
assert set(result) == {"005930", "000660"}
def test_excludes_zero_qty_holdings(self) -> None:
balance = {
"output1": [
{"pdno": "005930", "ord_psbl_qty": "0"},
{"pdno": "000660", "ord_psbl_qty": "2"},
]
}
result = _extract_held_codes_from_balance(balance, is_domestic=True)
assert "005930" not in result
assert "000660" in result
def test_returns_empty_when_output1_missing(self) -> None:
balance: dict = {}
assert _extract_held_codes_from_balance(balance, is_domestic=True) == []
def test_overseas_uses_ovrs_pdno(self) -> None:
balance = {"output1": [{"ovrs_pdno": "AAPL", "ovrs_cblc_qty": "3"}]}
result = _extract_held_codes_from_balance(balance, is_domestic=False)
assert result == ["AAPL"]
class TestDetermineOrderQuantity:
"""Test _determine_order_quantity() — SELL uses broker_held_qty."""
def test_sell_returns_broker_held_qty(self) -> None:
result = _determine_order_quantity(
action="SELL",
current_price=105.0,
total_cash=50000.0,
candidate=None,
settings=None,
broker_held_qty=7,
)
assert result == 7
def test_sell_returns_zero_when_broker_qty_zero(self) -> None:
result = _determine_order_quantity(
action="SELL",
current_price=105.0,
total_cash=50000.0,
candidate=None,
settings=None,
broker_held_qty=0,
)
assert result == 0
def test_buy_without_position_sizing_returns_one(self) -> None:
result = _determine_order_quantity(
action="BUY",
current_price=50000.0,
total_cash=1000000.0,
candidate=None,
settings=None,
)
assert result == 1
def test_buy_with_zero_cash_returns_zero(self) -> None:
result = _determine_order_quantity(
action="BUY",
current_price=50000.0,
total_cash=0.0,
candidate=None,
settings=None,
)
assert result == 0
def test_buy_with_position_sizing_calculates_correctly(self) -> None:
settings = MagicMock(spec=Settings)
settings.POSITION_SIZING_ENABLED = True
settings.POSITION_VOLATILITY_TARGET_SCORE = 50.0
settings.POSITION_BASE_ALLOCATION_PCT = 10.0
settings.POSITION_MAX_ALLOCATION_PCT = 30.0
settings.POSITION_MIN_ALLOCATION_PCT = 1.0
# 1,000,000 * 10% = 100,000 budget // 50,000 price = 2 shares
result = _determine_order_quantity(
action="BUY",
current_price=50000.0,
total_cash=1000000.0,
candidate=None,
settings=settings,
)
assert result == 2
class TestSafeFloat:
"""Test safe_float() helper function."""
@@ -249,7 +111,14 @@ class TestTradingCycleTelegramIntegration:
def mock_broker(self) -> MagicMock:
"""Create mock broker."""
broker = MagicMock()
broker.get_current_price = AsyncMock(return_value=(50000.0, 1.23, 100.0))
broker.get_orderbook = AsyncMock(
return_value={
"output1": {
"stck_prpr": "50000",
"frgn_ntby_qty": "100",
}
}
)
broker.get_balance = AsyncMock(
return_value={
"output2": [
@@ -868,83 +737,6 @@ class TestOverseasBalanceParsing:
# Verify price API was called
mock_overseas_broker_with_empty_price.get_overseas_price.assert_called_once()
@pytest.fixture
def mock_overseas_broker_with_buy_scenario(self) -> MagicMock:
"""Create mock overseas broker that returns a valid price for BUY orders."""
broker = MagicMock()
broker.get_overseas_price = AsyncMock(
return_value={"output": {"last": "182.50"}}
)
broker.get_overseas_balance = AsyncMock(
return_value={
"output2": [
{
"frcr_evlu_tota": "100000.00",
"frcr_dncl_amt_2": "50000.00",
"frcr_buy_amt_smtl": "50000.00",
}
]
}
)
broker.send_overseas_order = AsyncMock(return_value={"msg1": "주문접수"})
return broker
@pytest.fixture
def mock_scenario_engine_buy(self) -> MagicMock:
"""Create mock scenario engine that returns BUY."""
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_buy_match("AAPL"))
return engine
@pytest.mark.asyncio
async def test_overseas_buy_order_uses_limit_price(
self,
mock_domestic_broker: MagicMock,
mock_overseas_broker_with_buy_scenario: MagicMock,
mock_scenario_engine_buy: MagicMock,
mock_playbook: DayPlaybook,
mock_risk: MagicMock,
mock_db: MagicMock,
mock_decision_logger: MagicMock,
mock_context_store: MagicMock,
mock_criticality_assessor: MagicMock,
mock_telegram: MagicMock,
mock_overseas_market: MagicMock,
) -> None:
"""Overseas BUY order must use current_price (limit), not 0 (market).
KIS VTS rejects market orders for overseas paper trading.
Regression test for issue #149.
"""
mock_telegram.notify_trade_execution = AsyncMock()
with patch("src.main.log_trade"):
await trading_cycle(
broker=mock_domestic_broker,
overseas_broker=mock_overseas_broker_with_buy_scenario,
scenario_engine=mock_scenario_engine_buy,
playbook=mock_playbook,
risk=mock_risk,
db_conn=mock_db,
decision_logger=mock_decision_logger,
context_store=mock_context_store,
criticality_assessor=mock_criticality_assessor,
telegram=mock_telegram,
market=mock_overseas_market,
stock_code="AAPL",
scan_candidates={},
)
# Verify limit order was sent with actual price + 0.5% premium (issue #151), not 0.0
mock_overseas_broker_with_buy_scenario.send_overseas_order.assert_called_once()
call_kwargs = mock_overseas_broker_with_buy_scenario.send_overseas_order.call_args
sent_price = call_kwargs[1].get("price") or call_kwargs[0][4]
expected_price = round(182.5 * 1.005, 4) # 0.5% premium for BUY limit orders
assert sent_price == expected_price, (
f"Expected limit price {expected_price} (182.5 * 1.005) but got {sent_price}. "
"KIS VTS only accepts limit orders; BUY uses 0.5% premium to improve fill rate."
)
class TestScenarioEngineIntegration:
"""Test scenario engine integration in trading_cycle."""
@@ -953,7 +745,11 @@ class TestScenarioEngineIntegration:
def mock_broker(self) -> MagicMock:
"""Create mock broker with standard domestic data."""
broker = MagicMock()
broker.get_current_price = AsyncMock(return_value=(50000.0, 2.50, 100.0))
broker.get_orderbook = AsyncMock(
return_value={
"output1": {"stck_prpr": "50000", "frgn_ntby_qty": "100"}
}
)
broker.get_balance = AsyncMock(
return_value={
"output2": [
@@ -1034,7 +830,6 @@ class TestScenarioEngineIntegration:
assert market_data["rsi"] == 25.0
assert market_data["volume_ratio"] == 3.5
assert market_data["current_price"] == 50000.0
assert market_data["price_change_pct"] == 2.5
# Portfolio data should include pnl
assert "portfolio_pnl_pct" in portfolio_data
@@ -1375,17 +1170,18 @@ async def test_sell_updates_original_buy_decision_outcome() -> None:
)
broker = MagicMock()
broker.get_current_price = AsyncMock(return_value=(120.0, 0.0, 0.0))
broker.get_orderbook = AsyncMock(
return_value={"output1": {"stck_prpr": "120", "frgn_ntby_qty": "0"}}
)
broker.get_balance = AsyncMock(
return_value={
"output1": [{"pdno": "005930", "ord_psbl_qty": "1"}],
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "10000",
"pchs_amt_smtl_amt": "90000",
}
],
]
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
@@ -1436,418 +1232,6 @@ async def test_sell_updates_original_buy_decision_outcome() -> None:
assert updated_buy.outcome_accuracy == 1
@pytest.mark.asyncio
async def test_hold_overridden_to_sell_when_stop_loss_triggered() -> None:
"""HOLD decision should be overridden to SELL when stop-loss threshold is breached."""
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=90,
rationale="entry",
context_snapshot={},
input_data={},
)
log_trade(
conn=db_conn,
stock_code="005930",
action="BUY",
confidence=90,
rationale="entry",
quantity=1,
price=100.0,
market="KR",
exchange_code="KRX",
decision_id=buy_decision_id,
)
broker = MagicMock()
broker.get_current_price = AsyncMock(return_value=(95.0, -5.0, 0.0))
broker.get_balance = AsyncMock(
return_value={
"output1": [{"pdno": "005930", "ord_psbl_qty": "1"}],
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "10000",
"pchs_amt_smtl_amt": "90000",
}
],
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
scenario = StockScenario(
condition=StockCondition(rsi_below=30),
action=ScenarioAction.BUY,
confidence=88,
stop_loss_pct=-2.0,
rationale="stop loss policy",
)
playbook = DayPlaybook(
date=date(2026, 2, 8),
market="KR",
stock_playbooks=[
{"stock_code": "005930", "stock_name": "Samsung", "scenarios": [scenario]}
],
)
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
await trading_cycle(
broker=broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=playbook,
risk=MagicMock(),
db_conn=db_conn,
decision_logger=decision_logger,
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=telegram,
market=market,
stock_code="005930",
scan_candidates={},
)
broker.send_order.assert_called_once()
assert broker.send_order.call_args.kwargs["order_type"] == "SELL"
@pytest.mark.asyncio
async def test_hold_overridden_to_sell_when_take_profit_triggered() -> None:
"""HOLD decision should be overridden to SELL when take-profit threshold is reached."""
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=90,
rationale="entry",
context_snapshot={},
input_data={},
)
log_trade(
conn=db_conn,
stock_code="005930",
action="BUY",
confidence=90,
rationale="entry",
quantity=1,
price=100.0,
market="KR",
exchange_code="KRX",
decision_id=buy_decision_id,
)
broker = MagicMock()
# Current price 106.0 → +6% gain, above take_profit_pct=3.0
broker.get_current_price = AsyncMock(return_value=(106.0, 6.0, 0.0))
broker.get_balance = AsyncMock(
return_value={
"output1": [{"pdno": "005930", "ord_psbl_qty": "1"}],
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "10000",
"pchs_amt_smtl_amt": "90000",
}
],
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
scenario = StockScenario(
condition=StockCondition(rsi_below=30),
action=ScenarioAction.BUY,
confidence=88,
stop_loss_pct=-2.0,
take_profit_pct=3.0,
rationale="take profit policy",
)
playbook = DayPlaybook(
date=date(2026, 2, 8),
market="KR",
stock_playbooks=[
{"stock_code": "005930", "stock_name": "Samsung", "scenarios": [scenario]}
],
)
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
await trading_cycle(
broker=broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=playbook,
risk=MagicMock(),
db_conn=db_conn,
decision_logger=decision_logger,
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=telegram,
market=market,
stock_code="005930",
scan_candidates={},
)
broker.send_order.assert_called_once()
assert broker.send_order.call_args.kwargs["order_type"] == "SELL"
@pytest.mark.asyncio
async def test_hold_not_overridden_when_between_stop_loss_and_take_profit() -> None:
"""HOLD should remain HOLD when P&L is within stop-loss and take-profit bounds."""
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=90,
rationale="entry",
context_snapshot={},
input_data={},
)
log_trade(
conn=db_conn,
stock_code="005930",
action="BUY",
confidence=90,
rationale="entry",
quantity=1,
price=100.0,
market="KR",
exchange_code="KRX",
decision_id=buy_decision_id,
)
broker = MagicMock()
# Current price 101.0 → +1% gain, within [-2%, +3%] range
broker.get_current_price = AsyncMock(return_value=(101.0, 1.0, 0.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"})
scenario = StockScenario(
condition=StockCondition(rsi_below=30),
action=ScenarioAction.BUY,
confidence=88,
stop_loss_pct=-2.0,
take_profit_pct=3.0,
rationale="within range policy",
)
playbook = DayPlaybook(
date=date(2026, 2, 8),
market="KR",
stock_playbooks=[
{"stock_code": "005930", "stock_name": "Samsung", "scenarios": [scenario]}
],
)
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
await trading_cycle(
broker=broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=playbook,
risk=MagicMock(),
db_conn=db_conn,
decision_logger=decision_logger,
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=telegram,
market=market,
stock_code="005930",
scan_candidates={},
)
broker.send_order.assert_not_called()
@pytest.mark.asyncio
async def test_sell_order_uses_broker_balance_qty_not_db() -> None:
"""SELL quantity must come from broker balance output1, not DB.
The DB records order quantity which may differ from actual fill quantity.
This test verifies that we use the broker-confirmed orderable quantity.
"""
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=90,
rationale="entry",
context_snapshot={},
input_data={},
)
# DB records 10 shares ordered — but only 5 actually filled (partial fill scenario)
log_trade(
conn=db_conn,
stock_code="005930",
action="BUY",
confidence=90,
rationale="entry",
quantity=10, # ordered quantity (may differ from fill)
price=100.0,
market="KR",
exchange_code="KRX",
decision_id=buy_decision_id,
)
broker = MagicMock()
# Stop-loss triggers (price dropped below -2%)
broker.get_current_price = AsyncMock(return_value=(95.0, -5.0, 0.0))
broker.get_balance = AsyncMock(
return_value={
# Broker confirms only 5 shares are actually orderable (partial fill)
"output1": [{"pdno": "005930", "ord_psbl_qty": "5"}],
"output2": [
{
"tot_evlu_amt": "100000",
"dnca_tot_amt": "10000",
"pchs_amt_smtl_amt": "90000",
}
],
}
)
broker.send_order = AsyncMock(return_value={"msg1": "OK"})
scenario = StockScenario(
condition=StockCondition(rsi_below=30),
action=ScenarioAction.BUY,
confidence=88,
stop_loss_pct=-2.0,
rationale="stop loss policy",
)
playbook = DayPlaybook(
date=date(2026, 2, 8),
market="KR",
stock_playbooks=[
{"stock_code": "005930", "stock_name": "Samsung", "scenarios": [scenario]}
],
)
engine = MagicMock(spec=ScenarioEngine)
engine.evaluate = MagicMock(return_value=_make_hold_match())
market = MagicMock()
market.name = "Korea"
market.code = "KR"
market.exchange_code = "KRX"
market.is_domestic = True
telegram = MagicMock()
telegram.notify_trade_execution = AsyncMock()
telegram.notify_fat_finger = AsyncMock()
telegram.notify_circuit_breaker = AsyncMock()
telegram.notify_scenario_matched = AsyncMock()
await trading_cycle(
broker=broker,
overseas_broker=MagicMock(),
scenario_engine=engine,
playbook=playbook,
risk=MagicMock(),
db_conn=db_conn,
decision_logger=decision_logger,
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=telegram,
market=market,
stock_code="005930",
scan_candidates={},
)
broker.send_order.assert_called_once()
call_kwargs = broker.send_order.call_args.kwargs
assert call_kwargs["order_type"] == "SELL"
# Must use broker-confirmed qty (5), NOT DB-recorded ordered qty (10)
assert call_kwargs["quantity"] == 5
@pytest.mark.asyncio
async def test_handle_market_close_runs_daily_review_flow() -> None:
"""Market close should aggregate, create scorecard, lessons, and notify."""
@@ -2043,7 +1427,7 @@ async def test_run_evolution_loop_notifies_when_pr_generated() -> None:
await _run_evolution_loop(
evolution_optimizer=optimizer,
telegram=telegram,
market_code="US_NASDAQ",
market_code="US",
market_date="2026-02-14",
)
@@ -2067,7 +1451,7 @@ async def test_run_evolution_loop_notification_error_is_ignored() -> None:
await _run_evolution_loop(
evolution_optimizer=optimizer,
telegram=telegram,
market_code="US_NYSE",
market_code="US",
market_date="2026-02-14",
)

View File

@@ -7,7 +7,6 @@ import pytest
from src.markets.schedule import (
MARKETS,
expand_market_codes,
get_next_market_open,
get_open_markets,
is_market_open,
@@ -200,28 +199,3 @@ class TestGetNextMarketOpen:
enabled_markets=["INVALID", "KR"], now=test_time
)
assert market.code == "KR"
class TestExpandMarketCodes:
"""Test shorthand market expansion."""
def test_expand_us_shorthand(self) -> None:
assert expand_market_codes(["US"]) == ["US_NASDAQ", "US_NYSE", "US_AMEX"]
def test_expand_cn_shorthand(self) -> None:
assert expand_market_codes(["CN"]) == ["CN_SHA", "CN_SZA"]
def test_expand_vn_shorthand(self) -> None:
assert expand_market_codes(["VN"]) == ["VN_HAN", "VN_HCM"]
def test_expand_mixed_codes(self) -> None:
assert expand_market_codes(["KR", "US", "JP"]) == [
"KR",
"US_NASDAQ",
"US_NYSE",
"US_AMEX",
"JP",
]
def test_expand_preserves_unknown_code(self) -> None:
assert expand_market_codes(["KR", "UNKNOWN"]) == ["KR", "UNKNOWN"]

View File

@@ -1,643 +0,0 @@
"""Tests for OverseasBroker — rankings, price, balance, order, and helpers."""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock
import aiohttp
import pytest
from src.broker.kis_api import KISBroker
from src.broker.overseas import OverseasBroker, _PRICE_EXCHANGE_MAP, _RANKING_EXCHANGE_MAP
from src.config import Settings
def _make_async_cm(mock_resp: AsyncMock) -> MagicMock:
"""Create an async context manager that returns mock_resp on __aenter__."""
cm = MagicMock()
cm.__aenter__ = AsyncMock(return_value=mock_resp)
cm.__aexit__ = AsyncMock(return_value=False)
return cm
@pytest.fixture
def mock_settings() -> Settings:
"""Provide mock settings with correct default TR_IDs/paths."""
return Settings(
KIS_APP_KEY="test_key",
KIS_APP_SECRET="test_secret",
KIS_ACCOUNT_NO="12345678-01",
GEMINI_API_KEY="test_gemini_key",
)
@pytest.fixture
def mock_broker(mock_settings: Settings) -> KISBroker:
"""Provide a mock KIS broker."""
broker = KISBroker(mock_settings)
broker.get_orderbook = AsyncMock() # type: ignore[method-assign]
return broker
@pytest.fixture
def overseas_broker(mock_broker: KISBroker) -> OverseasBroker:
"""Provide an OverseasBroker wrapping a mock KISBroker."""
return OverseasBroker(mock_broker)
def _setup_broker_mocks(overseas_broker: OverseasBroker, mock_session: MagicMock) -> None:
"""Wire up common broker mocks."""
overseas_broker._broker._rate_limiter.acquire = AsyncMock()
overseas_broker._broker._get_session = MagicMock(return_value=mock_session)
overseas_broker._broker._auth_headers = AsyncMock(return_value={})
class TestRankingExchangeMap:
"""Test exchange code mapping for ranking API."""
def test_nasd_maps_to_nas(self) -> None:
assert _RANKING_EXCHANGE_MAP["NASD"] == "NAS"
def test_nyse_maps_to_nys(self) -> None:
assert _RANKING_EXCHANGE_MAP["NYSE"] == "NYS"
def test_amex_maps_to_ams(self) -> None:
assert _RANKING_EXCHANGE_MAP["AMEX"] == "AMS"
def test_sehk_maps_to_hks(self) -> None:
assert _RANKING_EXCHANGE_MAP["SEHK"] == "HKS"
def test_unmapped_exchange_passes_through(self) -> None:
assert _RANKING_EXCHANGE_MAP.get("UNKNOWN", "UNKNOWN") == "UNKNOWN"
def test_tse_unchanged(self) -> None:
assert _RANKING_EXCHANGE_MAP["TSE"] == "TSE"
class TestConfigDefaults:
"""Test that config defaults match KIS official API specs."""
def test_fluct_tr_id(self, mock_settings: Settings) -> None:
assert mock_settings.OVERSEAS_RANKING_FLUCT_TR_ID == "HHDFS76290000"
def test_volume_tr_id(self, mock_settings: Settings) -> None:
assert mock_settings.OVERSEAS_RANKING_VOLUME_TR_ID == "HHDFS76270000"
def test_fluct_path(self, mock_settings: Settings) -> None:
assert mock_settings.OVERSEAS_RANKING_FLUCT_PATH == "/uapi/overseas-stock/v1/ranking/updown-rate"
def test_volume_path(self, mock_settings: Settings) -> None:
assert mock_settings.OVERSEAS_RANKING_VOLUME_PATH == "/uapi/overseas-stock/v1/ranking/volume-surge"
class TestFetchOverseasRankings:
"""Test fetch_overseas_rankings method."""
@pytest.mark.asyncio
async def test_fluctuation_uses_correct_params(
self, overseas_broker: OverseasBroker
) -> None:
"""Fluctuation ranking should use HHDFS76290000, updown-rate path, and correct params."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(
return_value={"output": [{"symb": "AAPL", "name": "Apple"}]}
)
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
overseas_broker._broker._auth_headers = AsyncMock(
return_value={"authorization": "Bearer test"}
)
result = await overseas_broker.fetch_overseas_rankings("NASD", "fluctuation")
assert len(result) == 1
assert result[0]["symb"] == "AAPL"
call_args = mock_session.get.call_args
url = call_args[0][0]
params = call_args[1]["params"]
assert "/uapi/overseas-stock/v1/ranking/updown-rate" in url
assert params["EXCD"] == "NAS"
assert params["NDAY"] == "0"
assert params["GUBN"] == "1"
assert params["VOL_RANG"] == "0"
overseas_broker._broker._auth_headers.assert_called_with("HHDFS76290000")
@pytest.mark.asyncio
async def test_volume_uses_correct_params(
self, overseas_broker: OverseasBroker
) -> None:
"""Volume ranking should use HHDFS76270000, volume-surge path, and correct params."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(
return_value={"output": [{"symb": "TSLA", "name": "Tesla"}]}
)
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
overseas_broker._broker._auth_headers = AsyncMock(
return_value={"authorization": "Bearer test"}
)
result = await overseas_broker.fetch_overseas_rankings("NYSE", "volume")
assert len(result) == 1
call_args = mock_session.get.call_args
url = call_args[0][0]
params = call_args[1]["params"]
assert "/uapi/overseas-stock/v1/ranking/volume-surge" in url
assert params["EXCD"] == "NYS"
assert params["MIXN"] == "0"
assert params["VOL_RANG"] == "0"
assert "NDAY" not in params
assert "GUBN" not in params
overseas_broker._broker._auth_headers.assert_called_with("HHDFS76270000")
@pytest.mark.asyncio
async def test_404_returns_empty_list(
self, overseas_broker: OverseasBroker
) -> None:
"""HTTP 404 should return empty list (fallback) instead of raising."""
mock_resp = AsyncMock()
mock_resp.status = 404
mock_resp.text = AsyncMock(return_value="Not Found")
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
result = await overseas_broker.fetch_overseas_rankings("AMEX", "fluctuation")
assert result == []
@pytest.mark.asyncio
async def test_non_404_error_raises(
self, overseas_broker: OverseasBroker
) -> None:
"""Non-404 HTTP errors should raise ConnectionError."""
mock_resp = AsyncMock()
mock_resp.status = 500
mock_resp.text = AsyncMock(return_value="Internal Server Error")
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
with pytest.raises(ConnectionError, match="500"):
await overseas_broker.fetch_overseas_rankings("NASD")
@pytest.mark.asyncio
async def test_empty_response_returns_empty(
self, overseas_broker: OverseasBroker
) -> None:
"""Empty output in response should return empty list."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"output": []})
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
result = await overseas_broker.fetch_overseas_rankings("NASD")
assert result == []
@pytest.mark.asyncio
async def test_ranking_disabled_returns_empty(
self, overseas_broker: OverseasBroker
) -> None:
"""When OVERSEAS_RANKING_ENABLED=False, should return empty immediately."""
overseas_broker._broker._settings.OVERSEAS_RANKING_ENABLED = False
result = await overseas_broker.fetch_overseas_rankings("NASD")
assert result == []
@pytest.mark.asyncio
async def test_limit_truncates_results(
self, overseas_broker: OverseasBroker
) -> None:
"""Results should be truncated to the specified limit."""
rows = [{"symb": f"SYM{i}"} for i in range(20)]
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"output": rows})
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
result = await overseas_broker.fetch_overseas_rankings("NASD", limit=5)
assert len(result) == 5
@pytest.mark.asyncio
async def test_network_error_raises(
self, overseas_broker: OverseasBroker
) -> None:
"""Network errors should raise ConnectionError."""
cm = MagicMock()
cm.__aenter__ = AsyncMock(side_effect=aiohttp.ClientError("timeout"))
cm.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=cm)
_setup_broker_mocks(overseas_broker, mock_session)
with pytest.raises(ConnectionError, match="Network error"):
await overseas_broker.fetch_overseas_rankings("NASD")
@pytest.mark.asyncio
async def test_exchange_code_mapping_applied(
self, overseas_broker: OverseasBroker
) -> None:
"""All major exchanges should use mapped codes in API params."""
for original, mapped in [("NASD", "NAS"), ("NYSE", "NYS"), ("AMEX", "AMS")]:
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"output": [{"symb": "X"}]})
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
await overseas_broker.fetch_overseas_rankings(original)
call_params = mock_session.get.call_args[1]["params"]
assert call_params["EXCD"] == mapped, f"{original} should map to {mapped}"
class TestGetOverseasPrice:
"""Test get_overseas_price method."""
@pytest.mark.asyncio
async def test_success(self, overseas_broker: OverseasBroker) -> None:
"""Successful price fetch returns JSON data."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"output": {"last": "150.00"}})
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
overseas_broker._broker._auth_headers = AsyncMock(return_value={"authorization": "Bearer t"})
result = await overseas_broker.get_overseas_price("NASD", "AAPL")
assert result["output"]["last"] == "150.00"
call_args = mock_session.get.call_args
params = call_args[1]["params"]
assert params["EXCD"] == "NAS" # NASD → NAS via _PRICE_EXCHANGE_MAP
assert params["SYMB"] == "AAPL"
@pytest.mark.asyncio
async def test_http_error_raises(self, overseas_broker: OverseasBroker) -> None:
"""Non-200 response should raise ConnectionError."""
mock_resp = AsyncMock()
mock_resp.status = 400
mock_resp.text = AsyncMock(return_value="Bad Request")
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
with pytest.raises(ConnectionError, match="get_overseas_price failed"):
await overseas_broker.get_overseas_price("NASD", "AAPL")
@pytest.mark.asyncio
async def test_network_error_raises(self, overseas_broker: OverseasBroker) -> None:
"""Network error should raise ConnectionError."""
cm = MagicMock()
cm.__aenter__ = AsyncMock(side_effect=aiohttp.ClientError("conn refused"))
cm.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=cm)
_setup_broker_mocks(overseas_broker, mock_session)
with pytest.raises(ConnectionError, match="Network error"):
await overseas_broker.get_overseas_price("NASD", "AAPL")
class TestGetOverseasBalance:
"""Test get_overseas_balance method."""
@pytest.mark.asyncio
async def test_success(self, overseas_broker: OverseasBroker) -> None:
"""Successful balance fetch returns JSON data."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"output1": [{"pdno": "AAPL"}]})
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
result = await overseas_broker.get_overseas_balance("NASD")
assert result["output1"][0]["pdno"] == "AAPL"
@pytest.mark.asyncio
async def test_http_error_raises(self, overseas_broker: OverseasBroker) -> None:
"""Non-200 should raise ConnectionError."""
mock_resp = AsyncMock()
mock_resp.status = 500
mock_resp.text = AsyncMock(return_value="Server Error")
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
with pytest.raises(ConnectionError, match="get_overseas_balance failed"):
await overseas_broker.get_overseas_balance("NASD")
@pytest.mark.asyncio
async def test_network_error_raises(self, overseas_broker: OverseasBroker) -> None:
"""Network error should raise ConnectionError."""
cm = MagicMock()
cm.__aenter__ = AsyncMock(side_effect=TimeoutError("timeout"))
cm.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=cm)
_setup_broker_mocks(overseas_broker, mock_session)
with pytest.raises(ConnectionError, match="Network error"):
await overseas_broker.get_overseas_balance("NYSE")
class TestSendOverseasOrder:
"""Test send_overseas_order method."""
@pytest.mark.asyncio
async def test_buy_market_order(self, overseas_broker: OverseasBroker) -> None:
"""Market buy order should use VTTT1002U and ORD_DVSN=01."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"rt_cd": "0"})
mock_session = MagicMock()
mock_session.post = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
overseas_broker._broker._get_hash_key = AsyncMock(return_value="hashval")
result = await overseas_broker.send_overseas_order("NASD", "AAPL", "BUY", 10)
assert result["rt_cd"] == "0"
# Verify BUY TR_ID
overseas_broker._broker._auth_headers.assert_called_with("VTTT1002U")
call_args = mock_session.post.call_args
body = call_args[1]["json"]
assert body["ORD_DVSN"] == "01" # market order
assert body["OVRS_ORD_UNPR"] == "0"
@pytest.mark.asyncio
async def test_sell_limit_order(self, overseas_broker: OverseasBroker) -> None:
"""Limit sell order should use VTTT1006U and ORD_DVSN=00."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"rt_cd": "0"})
mock_session = MagicMock()
mock_session.post = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
overseas_broker._broker._get_hash_key = AsyncMock(return_value="hashval")
result = await overseas_broker.send_overseas_order("NYSE", "MSFT", "SELL", 5, price=350.0)
assert result["rt_cd"] == "0"
overseas_broker._broker._auth_headers.assert_called_with("VTTT1006U")
call_args = mock_session.post.call_args
body = call_args[1]["json"]
assert body["ORD_DVSN"] == "00" # limit order
assert body["OVRS_ORD_UNPR"] == "350.0"
@pytest.mark.asyncio
async def test_order_http_error_raises(self, overseas_broker: OverseasBroker) -> None:
"""Non-200 should raise ConnectionError."""
mock_resp = AsyncMock()
mock_resp.status = 400
mock_resp.text = AsyncMock(return_value="Bad Request")
mock_session = MagicMock()
mock_session.post = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
overseas_broker._broker._get_hash_key = AsyncMock(return_value="hashval")
with pytest.raises(ConnectionError, match="send_overseas_order failed"):
await overseas_broker.send_overseas_order("NASD", "AAPL", "BUY", 1)
@pytest.mark.asyncio
async def test_order_network_error_raises(self, overseas_broker: OverseasBroker) -> None:
"""Network error should raise ConnectionError."""
cm = MagicMock()
cm.__aenter__ = AsyncMock(side_effect=aiohttp.ClientError("conn reset"))
cm.__aexit__ = AsyncMock(return_value=False)
mock_session = MagicMock()
mock_session.post = MagicMock(return_value=cm)
_setup_broker_mocks(overseas_broker, mock_session)
overseas_broker._broker._get_hash_key = AsyncMock(return_value="hashval")
with pytest.raises(ConnectionError, match="Network error"):
await overseas_broker.send_overseas_order("NASD", "TSLA", "SELL", 2)
class TestGetCurrencyCode:
"""Test _get_currency_code mapping."""
def test_us_exchanges(self, overseas_broker: OverseasBroker) -> None:
assert overseas_broker._get_currency_code("NASD") == "USD"
assert overseas_broker._get_currency_code("NYSE") == "USD"
assert overseas_broker._get_currency_code("AMEX") == "USD"
def test_japan(self, overseas_broker: OverseasBroker) -> None:
assert overseas_broker._get_currency_code("TSE") == "JPY"
def test_hong_kong(self, overseas_broker: OverseasBroker) -> None:
assert overseas_broker._get_currency_code("SEHK") == "HKD"
def test_china(self, overseas_broker: OverseasBroker) -> None:
assert overseas_broker._get_currency_code("SHAA") == "CNY"
assert overseas_broker._get_currency_code("SZAA") == "CNY"
def test_vietnam(self, overseas_broker: OverseasBroker) -> None:
assert overseas_broker._get_currency_code("HNX") == "VND"
assert overseas_broker._get_currency_code("HSX") == "VND"
def test_unknown_defaults_usd(self, overseas_broker: OverseasBroker) -> None:
assert overseas_broker._get_currency_code("UNKNOWN") == "USD"
class TestExtractRankingRows:
"""Test _extract_ranking_rows helper."""
def test_output_key(self, overseas_broker: OverseasBroker) -> None:
data = {"output": [{"a": 1}, {"b": 2}]}
assert overseas_broker._extract_ranking_rows(data) == [{"a": 1}, {"b": 2}]
def test_output1_key(self, overseas_broker: OverseasBroker) -> None:
data = {"output1": [{"c": 3}]}
assert overseas_broker._extract_ranking_rows(data) == [{"c": 3}]
def test_output2_key(self, overseas_broker: OverseasBroker) -> None:
data = {"output2": [{"d": 4}]}
assert overseas_broker._extract_ranking_rows(data) == [{"d": 4}]
def test_no_list_returns_empty(self, overseas_broker: OverseasBroker) -> None:
data = {"output": "not a list"}
assert overseas_broker._extract_ranking_rows(data) == []
def test_empty_data(self, overseas_broker: OverseasBroker) -> None:
assert overseas_broker._extract_ranking_rows({}) == []
def test_filters_non_dict_rows(self, overseas_broker: OverseasBroker) -> None:
data = {"output": [{"a": 1}, "invalid", {"b": 2}]}
assert overseas_broker._extract_ranking_rows(data) == [{"a": 1}, {"b": 2}]
class TestPriceExchangeMap:
"""Test _PRICE_EXCHANGE_MAP is applied in get_overseas_price (issue #151)."""
def test_price_map_equals_ranking_map(self) -> None:
assert _PRICE_EXCHANGE_MAP is _RANKING_EXCHANGE_MAP
@pytest.mark.parametrize("original,expected", [
("NASD", "NAS"),
("NYSE", "NYS"),
("AMEX", "AMS"),
])
def test_us_exchange_code_mapping(self, original: str, expected: str) -> None:
assert _PRICE_EXCHANGE_MAP[original] == expected
@pytest.mark.asyncio
async def test_get_overseas_price_sends_mapped_code(
self, overseas_broker: OverseasBroker
) -> None:
"""NASD → NAS must be sent to HHDFS00000300."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"output": {"last": "200.00"}})
mock_session = MagicMock()
mock_session.get = MagicMock(return_value=_make_async_cm(mock_resp))
_setup_broker_mocks(overseas_broker, mock_session)
await overseas_broker.get_overseas_price("NASD", "AAPL")
params = mock_session.get.call_args[1]["params"]
assert params["EXCD"] == "NAS"
class TestOrderRtCdCheck:
"""Test that send_overseas_order checks rt_cd and logs accordingly (issue #151)."""
@pytest.fixture
def overseas_broker(self, mock_settings: Settings) -> OverseasBroker:
broker = MagicMock(spec=KISBroker)
broker._settings = mock_settings
broker._account_no = "12345678"
broker._product_cd = "01"
broker._base_url = "https://openapivts.koreainvestment.com:9443"
broker._rate_limiter = AsyncMock()
broker._rate_limiter.acquire = AsyncMock()
broker._auth_headers = AsyncMock(return_value={"authorization": "Bearer t"})
broker._get_hash_key = AsyncMock(return_value="hashval")
return OverseasBroker(broker)
@pytest.mark.asyncio
async def test_success_rt_cd_returns_data(
self, overseas_broker: OverseasBroker
) -> None:
"""rt_cd='0' → order accepted, data returned."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(return_value={"rt_cd": "0", "msg1": "완료"})
mock_session = MagicMock()
mock_session.post = MagicMock(return_value=_make_async_cm(mock_resp))
overseas_broker._broker._get_session = MagicMock(return_value=mock_session)
result = await overseas_broker.send_overseas_order("NASD", "AAPL", "BUY", 10, price=150.0)
assert result["rt_cd"] == "0"
@pytest.mark.asyncio
async def test_error_rt_cd_returns_data_with_msg(
self, overseas_broker: OverseasBroker
) -> None:
"""rt_cd != '0' → order rejected, data still returned (caller checks rt_cd)."""
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.json = AsyncMock(
return_value={"rt_cd": "1", "msg1": "주문가능금액이 부족합니다."}
)
mock_session = MagicMock()
mock_session.post = MagicMock(return_value=_make_async_cm(mock_resp))
overseas_broker._broker._get_session = MagicMock(return_value=mock_session)
result = await overseas_broker.send_overseas_order("NASD", "AAPL", "BUY", 10, price=150.0)
assert result["rt_cd"] == "1"
assert "부족" in result["msg1"]
class TestPaperOverseasCash:
"""Test PAPER_OVERSEAS_CASH config setting (issue #151)."""
def test_default_value(self) -> None:
settings = Settings(
KIS_APP_KEY="k",
KIS_APP_SECRET="s",
KIS_ACCOUNT_NO="12345678-01",
GEMINI_API_KEY="g",
)
assert settings.PAPER_OVERSEAS_CASH == 50000.0
def test_env_override(self) -> None:
import os
os.environ["PAPER_OVERSEAS_CASH"] = "25000"
settings = Settings(
KIS_APP_KEY="k",
KIS_APP_SECRET="s",
KIS_ACCOUNT_NO="12345678-01",
GEMINI_API_KEY="g",
)
assert settings.PAPER_OVERSEAS_CASH == 25000.0
del os.environ["PAPER_OVERSEAS_CASH"]
def test_zero_disables_fallback(self) -> None:
import os
os.environ["PAPER_OVERSEAS_CASH"] = "0"
settings = Settings(
KIS_APP_KEY="k",
KIS_APP_SECRET="s",
KIS_ACCOUNT_NO="12345678-01",
GEMINI_API_KEY="g",
)
assert settings.PAPER_OVERSEAS_CASH == 0.0
del os.environ["PAPER_OVERSEAS_CASH"]

View File

@@ -164,23 +164,18 @@ class TestGeneratePlaybook:
assert pb.market_outlook == MarketOutlook.NEUTRAL
@pytest.mark.asyncio
async def test_gemini_failure_returns_smart_fallback(self) -> None:
async def test_gemini_failure_returns_defensive(self) -> None:
planner = _make_planner()
planner._gemini.decide = AsyncMock(side_effect=RuntimeError("API timeout"))
# oversold candidate (signal="oversold", rsi=28.5)
candidates = [_candidate()]
pb = await planner.generate_playbook("KR", candidates, today=date(2026, 2, 8))
assert pb.default_action == ScenarioAction.HOLD
# Smart fallback uses NEUTRAL outlook (not NEUTRAL_TO_BEARISH)
assert pb.market_outlook == MarketOutlook.NEUTRAL
assert pb.market_outlook == MarketOutlook.NEUTRAL_TO_BEARISH
assert pb.stock_count == 1
# Oversold candidate → first scenario is BUY, second is SELL stop-loss
scenarios = pb.stock_playbooks[0].scenarios
assert scenarios[0].action == ScenarioAction.BUY
assert scenarios[0].condition.rsi_below == 30
assert scenarios[1].action == ScenarioAction.SELL
# 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:
@@ -662,171 +657,3 @@ class TestDefensivePlaybook:
assert pb.stock_count == 0
assert pb.market == "US"
assert pb.market_outlook == MarketOutlook.NEUTRAL
# ---------------------------------------------------------------------------
# Smart fallback playbook
# ---------------------------------------------------------------------------
class TestSmartFallbackPlaybook:
"""Tests for _smart_fallback_playbook — rule-based BUY/SELL on Gemini failure."""
def _make_settings(self) -> Settings:
return Settings(
KIS_APP_KEY="test",
KIS_APP_SECRET="test",
KIS_ACCOUNT_NO="12345678-01",
GEMINI_API_KEY="test",
RSI_OVERSOLD_THRESHOLD=30,
VOL_MULTIPLIER=2.0,
)
def test_momentum_candidate_gets_buy_on_volume(self) -> None:
candidates = [
_candidate(code="CHOW", signal="momentum", volume_ratio=13.64, rsi=100.0)
]
settings = self._make_settings()
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "US_AMEX", candidates, settings
)
assert pb.stock_count == 1
sp = pb.stock_playbooks[0]
assert sp.stock_code == "CHOW"
# First scenario: BUY with volume_ratio_above
buy_sc = sp.scenarios[0]
assert buy_sc.action == ScenarioAction.BUY
assert buy_sc.condition.volume_ratio_above == 2.0
assert buy_sc.condition.rsi_below is None
assert buy_sc.confidence == 80
# Second scenario: stop-loss SELL
sell_sc = sp.scenarios[1]
assert sell_sc.action == ScenarioAction.SELL
assert sell_sc.condition.price_change_pct_below == -3.0
def test_oversold_candidate_gets_buy_on_rsi(self) -> None:
candidates = [
_candidate(code="005930", signal="oversold", rsi=22.0, volume_ratio=3.5)
]
settings = self._make_settings()
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "KR", candidates, settings
)
sp = pb.stock_playbooks[0]
buy_sc = sp.scenarios[0]
assert buy_sc.action == ScenarioAction.BUY
assert buy_sc.condition.rsi_below == 30
assert buy_sc.condition.volume_ratio_above is None
def test_all_candidates_have_stop_loss_sell(self) -> None:
candidates = [
_candidate(code="AAA", signal="momentum", volume_ratio=5.0),
_candidate(code="BBB", signal="oversold", rsi=25.0),
]
settings = self._make_settings()
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "US_NASDAQ", candidates, settings
)
assert pb.stock_count == 2
for sp in pb.stock_playbooks:
sell_scenarios = [s for s in sp.scenarios if s.action == ScenarioAction.SELL]
assert len(sell_scenarios) == 1
assert sell_scenarios[0].condition.price_change_pct_below == -3.0
assert sell_scenarios[0].condition.price_change_pct_below == -3.0
def test_market_outlook_is_neutral(self) -> None:
candidates = [_candidate(signal="momentum", volume_ratio=5.0)]
settings = self._make_settings()
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "US_AMEX", candidates, settings
)
assert pb.market_outlook == MarketOutlook.NEUTRAL
def test_default_action_is_hold(self) -> None:
candidates = [_candidate(signal="momentum", volume_ratio=5.0)]
settings = self._make_settings()
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "US_AMEX", candidates, settings
)
assert pb.default_action == ScenarioAction.HOLD
def test_has_global_reduce_all_rule(self) -> None:
candidates = [_candidate(signal="momentum", volume_ratio=5.0)]
settings = self._make_settings()
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "US_AMEX", candidates, settings
)
assert len(pb.global_rules) == 1
rule = pb.global_rules[0]
assert rule.action == ScenarioAction.REDUCE_ALL
assert "portfolio_pnl_pct" in rule.condition
def test_empty_candidates_returns_empty_playbook(self) -> None:
settings = self._make_settings()
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "US_AMEX", [], settings
)
assert pb.stock_count == 0
def test_vol_multiplier_applied_from_settings(self) -> None:
"""VOL_MULTIPLIER=3.0 should set volume_ratio_above=3.0 for momentum."""
candidates = [_candidate(signal="momentum", volume_ratio=5.0)]
settings = self._make_settings()
settings = settings.model_copy(update={"VOL_MULTIPLIER": 3.0})
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "US_AMEX", candidates, settings
)
buy_sc = pb.stock_playbooks[0].scenarios[0]
assert buy_sc.condition.volume_ratio_above == 3.0
def test_rsi_oversold_threshold_applied_from_settings(self) -> None:
"""RSI_OVERSOLD_THRESHOLD=25 should set rsi_below=25 for oversold."""
candidates = [_candidate(signal="oversold", rsi=22.0)]
settings = self._make_settings()
settings = settings.model_copy(update={"RSI_OVERSOLD_THRESHOLD": 25})
pb = PreMarketPlanner._smart_fallback_playbook(
date(2026, 2, 17), "KR", candidates, settings
)
buy_sc = pb.stock_playbooks[0].scenarios[0]
assert buy_sc.condition.rsi_below == 25
@pytest.mark.asyncio
async def test_generate_playbook_uses_smart_fallback_on_gemini_error(self) -> None:
"""generate_playbook() should use smart fallback (not defensive) on API failure."""
planner = _make_planner()
planner._gemini.decide = AsyncMock(side_effect=ConnectionError("429 quota exceeded"))
# momentum candidate
candidates = [
_candidate(code="CHOW", signal="momentum", volume_ratio=13.64, rsi=100.0)
]
pb = await planner.generate_playbook(
"US_AMEX", candidates, today=date(2026, 2, 18)
)
# Should NOT be all-SELL defensive; should have BUY for momentum
assert pb.stock_count == 1
buy_scenarios = [
s for s in pb.stock_playbooks[0].scenarios
if s.action == ScenarioAction.BUY
]
assert len(buy_scenarios) == 1
assert buy_scenarios[0].condition.volume_ratio_above == 2.0 # VOL_MULTIPLIER default

View File

@@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, MagicMock
from src.analysis.smart_scanner import ScanCandidate, SmartVolatilityScanner
from src.analysis.volatility import VolatilityAnalyzer
from src.broker.kis_api import KISBroker
from src.broker.overseas import OverseasBroker
from src.config import Settings
@@ -44,70 +43,61 @@ def scanner(mock_broker: MagicMock, mock_settings: Settings) -> SmartVolatilityS
analyzer = VolatilityAnalyzer()
return SmartVolatilityScanner(
broker=mock_broker,
overseas_broker=None,
volatility_analyzer=analyzer,
settings=mock_settings,
)
@pytest.fixture
def mock_overseas_broker() -> MagicMock:
"""Create mock overseas broker."""
broker = MagicMock(spec=OverseasBroker)
broker.get_overseas_price = AsyncMock()
broker.fetch_overseas_rankings = AsyncMock(return_value=[])
return broker
class TestSmartVolatilityScanner:
"""Test suite for SmartVolatilityScanner."""
@pytest.mark.asyncio
async def test_scan_domestic_prefers_volatility_with_liquidity_bonus(
async def test_scan_finds_oversold_candidates(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Domestic scan should score by volatility first and volume rank second."""
fluctuation_rows = [
"""Test that scanner identifies oversold stocks with high volume."""
# Mock rankings
mock_broker.fetch_market_rankings.return_value = [
{
"stock_code": "005930",
"name": "Samsung",
"price": 70000,
"volume": 5000000,
"change_rate": -5.0,
"change_rate": -3.5,
"volume_increase_rate": 250,
},
{
"stock_code": "035420",
"name": "NAVER",
"price": 250000,
"volume": 3000000,
"change_rate": 3.0,
"volume_increase_rate": 200,
},
]
volume_rows = [
{"stock_code": "035420", "name": "NAVER", "price": 250000, "volume": 3000000},
{"stock_code": "005930", "name": "Samsung", "price": 70000, "volume": 5000000},
]
mock_broker.fetch_market_rankings.side_effect = [fluctuation_rows, volume_rows]
mock_broker.get_daily_prices.return_value = [
{"open": 1, "high": 1, "low": 1, "close": 1, "volume": 1000000},
{"open": 1, "high": 1, "low": 1, "close": 1, "volume": 1000000},
]
# Mock daily prices - trending down (oversold)
prices = []
for i in range(20):
prices.append({
"date": f"2026020{i:02d}",
"open": 75000 - i * 200,
"high": 75500 - i * 200,
"low": 74500 - i * 200,
"close": 75000 - i * 250, # Steady decline
"volume": 2000000,
})
mock_broker.get_daily_prices.return_value = prices
candidates = await scanner.scan()
assert len(candidates) >= 1
# Samsung has higher absolute move, so it should lead despite lower volume rank bonus.
assert candidates[0].stock_code == "005930"
assert candidates[0].signal == "oversold"
# Should find at least one candidate (depending on exact RSI calculation)
mock_broker.fetch_market_rankings.assert_called_once()
mock_broker.get_daily_prices.assert_called_once_with("005930", days=20)
# If qualified, should have oversold signal
if candidates:
assert candidates[0].signal in ["oversold", "momentum"]
assert candidates[0].volume_ratio >= scanner.vol_multiplier
@pytest.mark.asyncio
async def test_scan_domestic_finds_momentum_candidate(
async def test_scan_finds_momentum_candidates(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Positive change should be represented as momentum signal."""
fluctuation_rows = [
"""Test that scanner identifies momentum stocks with high volume."""
mock_broker.fetch_market_rankings.return_value = [
{
"stock_code": "035420",
"name": "NAVER",
@@ -117,67 +107,124 @@ class TestSmartVolatilityScanner:
"volume_increase_rate": 300,
},
]
mock_broker.fetch_market_rankings.side_effect = [fluctuation_rows, fluctuation_rows]
mock_broker.get_daily_prices.return_value = [
{"open": 1, "high": 1, "low": 1, "close": 1, "volume": 1000000},
{"open": 1, "high": 1, "low": 1, "close": 1, "volume": 1000000},
]
# Mock daily prices - trending up (momentum)
prices = []
for i in range(20):
prices.append({
"date": f"2026020{i:02d}",
"open": 230000 + i * 500,
"high": 231000 + i * 500,
"low": 229000 + i * 500,
"close": 230500 + i * 500, # Steady rise
"volume": 1000000,
})
mock_broker.get_daily_prices.return_value = prices
candidates = await scanner.scan()
assert [c.stock_code for c in candidates] == ["035420"]
assert candidates[0].signal == "momentum"
mock_broker.fetch_market_rankings.assert_called_once()
@pytest.mark.asyncio
async def test_scan_domestic_filters_low_volatility(
async def test_scan_filters_low_volume(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Domestic scan should drop symbols below volatility threshold."""
fluctuation_rows = [
"""Test that stocks with low volume ratio are filtered out."""
mock_broker.fetch_market_rankings.return_value = [
{
"stock_code": "000660",
"name": "SK Hynix",
"price": 150000,
"volume": 500000,
"change_rate": 0.2,
"volume_increase_rate": 50,
"change_rate": -5.0,
"volume_increase_rate": 50, # Only 50% increase (< 200%)
},
]
mock_broker.fetch_market_rankings.side_effect = [fluctuation_rows, fluctuation_rows]
mock_broker.get_daily_prices.return_value = [
{"open": 1, "high": 150100, "low": 149900, "close": 150000, "volume": 1000000},
{"open": 1, "high": 150100, "low": 149900, "close": 150000, "volume": 1000000},
]
# Low volume
prices = []
for i in range(20):
prices.append({
"date": f"2026020{i:02d}",
"open": 150000 - i * 100,
"high": 151000 - i * 100,
"low": 149000 - i * 100,
"close": 150000 - i * 150, # Declining (would be oversold)
"volume": 1000000, # Current 500k < 2x prev day 1M
})
mock_broker.get_daily_prices.return_value = prices
candidates = await scanner.scan()
# Should be filtered out due to low volume ratio
assert len(candidates) == 0
@pytest.mark.asyncio
async def test_scan_filters_neutral_rsi(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Test that stocks with neutral RSI are filtered out."""
mock_broker.fetch_market_rankings.return_value = [
{
"stock_code": "051910",
"name": "LG Chem",
"price": 500000,
"volume": 3000000,
"change_rate": 0.5,
"volume_increase_rate": 300, # High volume
},
]
# Flat prices (neutral RSI ~50)
prices = []
for i in range(20):
prices.append({
"date": f"2026020{i:02d}",
"open": 500000 + (i % 2) * 100, # Small oscillation
"high": 500500,
"low": 499500,
"close": 500000 + (i % 2) * 50,
"volume": 1000000,
})
mock_broker.get_daily_prices.return_value = prices
candidates = await scanner.scan()
# Should be filtered out (RSI ~50, not < 30 or > 70)
assert len(candidates) == 0
@pytest.mark.asyncio
async def test_scan_uses_fallback_on_api_error(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Domestic scan should remain operational using fallback symbols."""
mock_broker.fetch_market_rankings.side_effect = [
ConnectionError("API unavailable"),
ConnectionError("API unavailable"),
]
mock_broker.get_daily_prices.return_value = [
{"open": 1, "high": 103, "low": 97, "close": 100, "volume": 1000000},
{"open": 1, "high": 103, "low": 97, "close": 100, "volume": 800000},
]
"""Test fallback to static list when ranking API fails."""
mock_broker.fetch_market_rankings.side_effect = ConnectionError("API unavailable")
# Fallback stocks should still be analyzed
prices = []
for i in range(20):
prices.append({
"date": f"2026020{i:02d}",
"open": 50000 - i * 50,
"high": 51000 - i * 50,
"low": 49000 - i * 50,
"close": 50000 - i * 75, # Declining
"volume": 1000000,
})
mock_broker.get_daily_prices.return_value = prices
candidates = await scanner.scan(fallback_stocks=["005930", "000660"])
# Should not crash
assert isinstance(candidates, list)
assert len(candidates) >= 1
@pytest.mark.asyncio
async def test_scan_returns_top_n_only(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Test that scan returns at most top_n candidates."""
fluctuation_rows = [
# Return many stocks
mock_broker.fetch_market_rankings.return_value = [
{
"stock_code": f"00{i}000",
"name": f"Stock{i}",
@@ -188,17 +235,62 @@ class TestSmartVolatilityScanner:
}
for i in range(1, 10)
]
mock_broker.fetch_market_rankings.side_effect = [fluctuation_rows, fluctuation_rows]
mock_broker.get_daily_prices.return_value = [
{"open": 1, "high": 105, "low": 95, "close": 100, "volume": 1000000},
{"open": 1, "high": 105, "low": 95, "close": 100, "volume": 900000},
]
# All oversold with high volume
def make_prices(code: str) -> list[dict]:
prices = []
for i in range(20):
prices.append({
"date": f"2026020{i:02d}",
"open": 10000 - i * 100,
"high": 10500 - i * 100,
"low": 9500 - i * 100,
"close": 10000 - i * 150,
"volume": 1000000,
})
return prices
mock_broker.get_daily_prices.side_effect = make_prices
candidates = await scanner.scan()
# Should respect top_n limit (3)
assert len(candidates) <= scanner.top_n
@pytest.mark.asyncio
async def test_scan_skips_insufficient_price_history(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Test that stocks with insufficient history are skipped."""
mock_broker.fetch_market_rankings.return_value = [
{
"stock_code": "005930",
"name": "Samsung",
"price": 70000,
"volume": 5000000,
"change_rate": -5.0,
"volume_increase_rate": 300,
},
]
# Only 5 days of data (need 15+ for RSI)
mock_broker.get_daily_prices.return_value = [
{
"date": f"2026020{i:02d}",
"open": 70000,
"high": 71000,
"low": 69000,
"close": 70000,
"volume": 2000000,
}
for i in range(5)
]
candidates = await scanner.scan()
# Should skip due to insufficient data
assert len(candidates) == 0
@pytest.mark.asyncio
async def test_get_stock_codes(
self, scanner: SmartVolatilityScanner
@@ -231,124 +323,6 @@ class TestSmartVolatilityScanner:
assert codes == ["005930", "035420"]
@pytest.mark.asyncio
async def test_scan_overseas_uses_dynamic_symbols(
self, mock_broker: MagicMock, mock_overseas_broker: MagicMock, mock_settings: Settings
) -> None:
"""Overseas scan should use provided dynamic universe symbols."""
analyzer = VolatilityAnalyzer()
scanner = SmartVolatilityScanner(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
volatility_analyzer=analyzer,
settings=mock_settings,
)
market = MagicMock()
market.name = "NASDAQ"
market.code = "US_NASDAQ"
market.exchange_code = "NASD"
market.is_domestic = False
mock_overseas_broker.get_overseas_price.side_effect = [
{"output": {"last": "210.5", "rate": "1.6", "tvol": "1500000"}},
{"output": {"last": "330.1", "rate": "0.2", "tvol": "900000"}},
]
candidates = await scanner.scan(
market=market,
fallback_stocks=["AAPL", "MSFT"],
)
assert [c.stock_code for c in candidates] == ["AAPL"]
assert candidates[0].signal == "momentum"
assert candidates[0].price == 210.5
@pytest.mark.asyncio
async def test_scan_overseas_uses_ranking_api_first(
self, mock_broker: MagicMock, mock_overseas_broker: MagicMock, mock_settings: Settings
) -> None:
"""Overseas scan should prioritize ranking API when available."""
analyzer = VolatilityAnalyzer()
scanner = SmartVolatilityScanner(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
volatility_analyzer=analyzer,
settings=mock_settings,
)
market = MagicMock()
market.name = "NASDAQ"
market.code = "US_NASDAQ"
market.exchange_code = "NASD"
market.is_domestic = False
mock_overseas_broker.fetch_overseas_rankings.return_value = [
{"symb": "NVDA", "last": "780.2", "rate": "2.4", "tvol": "1200000"},
{"symb": "MSFT", "last": "420.0", "rate": "0.3", "tvol": "900000"},
]
candidates = await scanner.scan(market=market, fallback_stocks=["AAPL", "TSLA"])
assert mock_overseas_broker.fetch_overseas_rankings.call_count >= 1
mock_overseas_broker.get_overseas_price.assert_not_called()
assert [c.stock_code for c in candidates] == ["NVDA"]
@pytest.mark.asyncio
async def test_scan_overseas_without_symbols_returns_empty(
self, mock_broker: MagicMock, mock_overseas_broker: MagicMock, mock_settings: Settings
) -> None:
"""Overseas scan should return empty list when no symbol universe exists."""
analyzer = VolatilityAnalyzer()
scanner = SmartVolatilityScanner(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
volatility_analyzer=analyzer,
settings=mock_settings,
)
market = MagicMock()
market.name = "NASDAQ"
market.code = "US_NASDAQ"
market.exchange_code = "NASD"
market.is_domestic = False
candidates = await scanner.scan(market=market, fallback_stocks=[])
assert candidates == []
@pytest.mark.asyncio
async def test_scan_overseas_picks_high_intraday_range_even_with_low_change(
self, mock_broker: MagicMock, mock_overseas_broker: MagicMock, mock_settings: Settings
) -> None:
"""Volatility selection should consider intraday range, not only change rate."""
analyzer = VolatilityAnalyzer()
scanner = SmartVolatilityScanner(
broker=mock_broker,
overseas_broker=mock_overseas_broker,
volatility_analyzer=analyzer,
settings=mock_settings,
)
market = MagicMock()
market.name = "NASDAQ"
market.code = "US_NASDAQ"
market.exchange_code = "NASD"
market.is_domestic = False
# change rate is tiny, but high-low range is large (15%).
mock_overseas_broker.fetch_overseas_rankings.return_value = [
{
"symb": "ABCD",
"last": "100",
"rate": "0.2",
"high": "110",
"low": "95",
"tvol": "800000",
}
]
candidates = await scanner.scan(market=market, fallback_stocks=[])
assert [c.stock_code for c in candidates] == ["ABCD"]
class TestRSICalculation:
"""Test RSI calculation in VolatilityAnalyzer."""

View File

@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, patch
import aiohttp
import pytest
from src.notifications.telegram_client import NotificationFilter, NotificationPriority, TelegramClient
from src.notifications.telegram_client import NotificationPriority, TelegramClient
class TestTelegramClientInit:
@@ -481,187 +481,3 @@ class TestClientCleanup:
# Should not raise exception
await client.close()
class TestNotificationFilter:
"""Test granular notification filter behavior."""
def test_default_filter_allows_all(self) -> None:
"""Default NotificationFilter has all flags enabled."""
f = NotificationFilter()
assert f.trades is True
assert f.market_open_close is True
assert f.fat_finger is True
assert f.system_events is True
assert f.playbook is True
assert f.scenario_match is True
assert f.errors is True
def test_client_uses_default_filter_when_none_given(self) -> None:
"""TelegramClient creates a default NotificationFilter when none provided."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
assert isinstance(client._filter, NotificationFilter)
assert client._filter.scenario_match is True
def test_client_stores_provided_filter(self) -> None:
"""TelegramClient stores a custom NotificationFilter."""
nf = NotificationFilter(scenario_match=False, trades=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
assert client._filter.scenario_match is False
assert client._filter.trades is False
assert client._filter.market_open_close is True # default still True
@pytest.mark.asyncio
async def test_scenario_match_filtered_does_not_send(self) -> None:
"""notify_scenario_matched skips send when scenario_match=False."""
nf = NotificationFilter(scenario_match=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
with patch("aiohttp.ClientSession.post") as mock_post:
await client.notify_scenario_matched(
stock_code="005930", action="BUY", condition_summary="rsi<30", confidence=85.0
)
mock_post.assert_not_called()
@pytest.mark.asyncio
async def test_trades_filtered_does_not_send(self) -> None:
"""notify_trade_execution skips send when trades=False."""
nf = NotificationFilter(trades=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
with patch("aiohttp.ClientSession.post") as mock_post:
await client.notify_trade_execution(
stock_code="005930", market="KR", action="BUY",
quantity=10, price=70000.0, confidence=85.0
)
mock_post.assert_not_called()
@pytest.mark.asyncio
async def test_market_open_close_filtered_does_not_send(self) -> None:
"""notify_market_open/close skip send when market_open_close=False."""
nf = NotificationFilter(market_open_close=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
with patch("aiohttp.ClientSession.post") as mock_post:
await client.notify_market_open("Korea")
await client.notify_market_close("Korea", pnl_pct=1.5)
mock_post.assert_not_called()
@pytest.mark.asyncio
async def test_circuit_breaker_always_sends_regardless_of_filter(self) -> None:
"""notify_circuit_breaker always sends (no filter flag)."""
nf = NotificationFilter(
trades=False, market_open_close=False, fat_finger=False,
system_events=False, playbook=False, scenario_match=False, errors=False,
)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
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_circuit_breaker(pnl_pct=-3.5, threshold=-3.0)
assert mock_post.call_count == 1
@pytest.mark.asyncio
async def test_errors_filtered_does_not_send(self) -> None:
"""notify_error skips send when errors=False."""
nf = NotificationFilter(errors=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
with patch("aiohttp.ClientSession.post") as mock_post:
await client.notify_error("TestError", "something went wrong", "KR")
mock_post.assert_not_called()
@pytest.mark.asyncio
async def test_playbook_filtered_does_not_send(self) -> None:
"""notify_playbook_generated/failed skip send when playbook=False."""
nf = NotificationFilter(playbook=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
with patch("aiohttp.ClientSession.post") as mock_post:
await client.notify_playbook_generated("KR", 3, 10, 1200)
await client.notify_playbook_failed("KR", "timeout")
mock_post.assert_not_called()
@pytest.mark.asyncio
async def test_system_events_filtered_does_not_send(self) -> None:
"""notify_system_start/shutdown skip send when system_events=False."""
nf = NotificationFilter(system_events=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
with patch("aiohttp.ClientSession.post") as mock_post:
await client.notify_system_start("paper", ["KR"])
await client.notify_system_shutdown("Normal shutdown")
mock_post.assert_not_called()
def test_set_flag_valid_key(self) -> None:
"""set_flag returns True and updates field for a known key."""
nf = NotificationFilter()
assert nf.set_flag("scenario", False) is True
assert nf.scenario_match is False
def test_set_flag_invalid_key(self) -> None:
"""set_flag returns False for an unknown key."""
nf = NotificationFilter()
assert nf.set_flag("unknown_key", False) is False
def test_as_dict_keys_match_KEYS(self) -> None:
"""as_dict() returns every key defined in KEYS."""
nf = NotificationFilter()
d = nf.as_dict()
assert set(d.keys()) == set(NotificationFilter.KEYS.keys())
def test_set_notification_valid_key(self) -> None:
"""TelegramClient.set_notification toggles filter at runtime."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
assert client._filter.scenario_match is True
assert client.set_notification("scenario", False) is True
assert client._filter.scenario_match is False
def test_set_notification_all_off(self) -> None:
"""set_notification('all', False) disables every filter flag."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
assert client.set_notification("all", False) is True
for v in client.filter_status().values():
assert v is False
def test_set_notification_all_on(self) -> None:
"""set_notification('all', True) enables every filter flag."""
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True,
notification_filter=NotificationFilter(
trades=False, market_open_close=False, scenario_match=False,
fat_finger=False, system_events=False, playbook=False, errors=False,
),
)
assert client.set_notification("all", True) is True
for v in client.filter_status().values():
assert v is True
def test_set_notification_unknown_key(self) -> None:
"""set_notification returns False for an unknown key."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
assert client.set_notification("unknown", False) is False
def test_filter_status_reflects_current_state(self) -> None:
"""filter_status() matches the current NotificationFilter state."""
nf = NotificationFilter(trades=False, scenario_match=False)
client = TelegramClient(
bot_token="123:abc", chat_id="456", enabled=True, notification_filter=nf
)
status = client.filter_status()
assert status["trades"] is False
assert status["scenario"] is False
assert status["market"] is True

View File

@@ -682,10 +682,6 @@ class TestBasicCommands:
"/help - Show available commands\n"
"/status - Trading status (mode, markets, P&L)\n"
"/positions - Current holdings\n"
"/report - Daily summary report\n"
"/scenarios - Today's playbook scenarios\n"
"/review - Recent scorecards\n"
"/dashboard - Dashboard URL/status\n"
"/stop - Pause trading\n"
"/resume - Resume trading"
)
@@ -711,106 +707,10 @@ class TestBasicCommands:
assert "/help" in payload["text"]
assert "/status" in payload["text"]
assert "/positions" in payload["text"]
assert "/report" in payload["text"]
assert "/scenarios" in payload["text"]
assert "/review" in payload["text"]
assert "/dashboard" in payload["text"]
assert "/stop" in payload["text"]
assert "/resume" in payload["text"]
class TestExtendedCommands:
"""Test additional bot commands."""
@pytest.mark.asyncio
async def test_report_command(self) -> None:
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
async def mock_report() -> None:
await client.send_message("<b>📈 Daily Report</b>\n\nTrades: 1")
handler.register_command("report", mock_report)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await handler._handle_update(
{"update_id": 1, "message": {"chat": {"id": 456}, "text": "/report"}}
)
payload = mock_post.call_args.kwargs["json"]
assert "Daily Report" in payload["text"]
@pytest.mark.asyncio
async def test_scenarios_command(self) -> None:
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
async def mock_scenarios() -> None:
await client.send_message("<b>🧠 Today's Scenarios</b>\n\n- AAPL: BUY (85)")
handler.register_command("scenarios", mock_scenarios)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await handler._handle_update(
{"update_id": 1, "message": {"chat": {"id": 456}, "text": "/scenarios"}}
)
payload = mock_post.call_args.kwargs["json"]
assert "Today's Scenarios" in payload["text"]
@pytest.mark.asyncio
async def test_review_command(self) -> None:
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
async def mock_review() -> None:
await client.send_message("<b>📝 Recent Reviews</b>\n\n- 2026-02-14 KR")
handler.register_command("review", mock_review)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await handler._handle_update(
{"update_id": 1, "message": {"chat": {"id": 456}, "text": "/review"}}
)
payload = mock_post.call_args.kwargs["json"]
assert "Recent Reviews" in payload["text"]
@pytest.mark.asyncio
async def test_dashboard_command(self) -> None:
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
mock_resp = AsyncMock()
mock_resp.status = 200
mock_resp.__aenter__ = AsyncMock(return_value=mock_resp)
mock_resp.__aexit__ = AsyncMock(return_value=False)
async def mock_dashboard() -> None:
await client.send_message("<b>🖥️ Dashboard</b>\n\nURL: http://127.0.0.1:8080")
handler.register_command("dashboard", mock_dashboard)
with patch("aiohttp.ClientSession.post", return_value=mock_resp) as mock_post:
await handler._handle_update(
{"update_id": 1, "message": {"chat": {"id": 456}, "text": "/dashboard"}}
)
payload = mock_post.call_args.kwargs["json"]
assert "Dashboard" in payload["text"]
class TestGetUpdates:
"""Test getUpdates API interaction."""
@@ -875,91 +775,3 @@ class TestGetUpdates:
updates = await handler._get_updates()
assert updates == []
class TestCommandWithArgs:
"""Test register_command_with_args and argument dispatch."""
def test_register_command_with_args_stored(self) -> None:
"""register_command_with_args stores handler in _commands_with_args."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
async def my_handler(args: list[str]) -> None:
pass
handler.register_command_with_args("notify", my_handler)
assert "notify" in handler._commands_with_args
assert handler._commands_with_args["notify"] is my_handler
@pytest.mark.asyncio
async def test_args_handler_receives_arguments(self) -> None:
"""Args handler is called with the trailing tokens."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
received: list[list[str]] = []
async def capture(args: list[str]) -> None:
received.append(args)
handler.register_command_with_args("notify", capture)
update = {
"message": {
"chat": {"id": "456"},
"text": "/notify scenario off",
}
}
await handler._handle_update(update)
assert received == [["scenario", "off"]]
@pytest.mark.asyncio
async def test_args_handler_takes_priority_over_no_args_handler(self) -> None:
"""When both handlers exist for same command, args handler wins."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
no_args_called = []
args_called = []
async def no_args_handler() -> None:
no_args_called.append(True)
async def args_handler(args: list[str]) -> None:
args_called.append(args)
handler.register_command("notify", no_args_handler)
handler.register_command_with_args("notify", args_handler)
update = {
"message": {
"chat": {"id": "456"},
"text": "/notify all off",
}
}
await handler._handle_update(update)
assert args_called == [["all", "off"]]
assert no_args_called == []
@pytest.mark.asyncio
async def test_args_handler_with_no_trailing_args(self) -> None:
"""/notify with no args still dispatches to args handler with empty list."""
client = TelegramClient(bot_token="123:abc", chat_id="456", enabled=True)
handler = TelegramCommandHandler(client)
received: list[list[str]] = []
async def capture(args: list[str]) -> None:
received.append(args)
handler.register_command_with_args("notify", capture)
update = {
"message": {
"chat": {"id": "456"},
"text": "/notify",
}
}
await handler._handle_update(update)
assert received == [[]]