Compare commits

...

6 Commits

Author SHA1 Message Date
agentson
733e6b36e9 feat: unify domestic scanner and sizing; update docs
Some checks failed
CI / test (pull_request) Has been cancelled
2026-02-17 06:29:36 +09:00
agentson
0659cc0aca docs: reflect overseas ranking integration and volatility-first selection 2026-02-17 06:29:16 +09:00
agentson
748b9b848e feat: prioritize overseas volatility scoring over raw rankings 2026-02-17 06:25:45 +09:00
agentson
6a1ad230ee feat: add overseas ranking integration with dynamic fallback 2026-02-17 06:25:45 +09:00
90bbc78867 Merge pull request 'docs: sync V2 status and process docs (#131)' (#134) from feature/issue-131-docs-v2-status-sync into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #134
Reviewed-by: jihoson <kiparang7th@gmail.com>
2026-02-16 21:50:49 +09:00
agentson
1ef5dcb2b3 docs: README.md v2 현행화 (#131)
Some checks failed
CI / test (pull_request) Has been cancelled
- 아키텍처 다이어그램에 v2 컴포넌트 (Strategy, Context, Evolution) 추가
- 핵심 모듈 테이블: 6개 → 14개 모듈 반영
- 테스트: 35개/3파일 → 551개/25파일
- 지원 시장 10개 거래소 테이블 추가
- 텔레그램 양방향 명령어 9종 레퍼런스
- 프로젝트 구조 트리 전면 갱신
- 문서 링크 섹션 추가

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-16 21:48:49 +09:00
9 changed files with 1029 additions and 312 deletions

160
README.md
View File

@@ -10,28 +10,41 @@ KIS(한국투자증권) API로 매매하고, Google Gemini로 판단하며, 자
│ (매매 실행) │ │ (거래 루프) │ │ (의사결정) │
└─────────────┘ └──────┬──────┘ └─────────────┘
┌────────────┐
│Risk Manager
│ (안전장치) │
└──────┬──────┘
────────────
│ Evolution
│ (전략 진화) │
└─────────────┘
┌────────────┼────────────┐
│ │
┌──────┴──────┐ ┌──┴───┐ ┌──────┴──────┐
│Risk Manager │ │ DB │ │ Telegram │
(안전장치) │ │ │ │ (알림+명령)
└──────┬──────┘ └──────┘ └─────────────
┌────────┼────────┐
┌────┴────┐┌──┴──┐┌────┴─────┐
│Strategy ││Ctx ││Evolution │
│(플레이북)││(메모리)││ (진화) │
└─────────┘└─────┘└──────────┘
```
**v2 핵심**: "Plan Once, Execute Locally" — 장 시작 전 AI가 시나리오 플레이북을 1회 생성하고, 거래 시간에는 로컬 시나리오 매칭만 수행하여 API 비용과 지연 시간을 대폭 절감.
## 핵심 모듈
| 모듈 | 파일 | 설명 |
| 모듈 | 위치 | 설명 |
|------|------|------|
| 설정 | `src/config.py` | Pydantic 기반 환경변수 로딩 및 타입 검증 |
| 브로커 | `src/broker/kis_api.py` | KIS API 비동기 래퍼 (토큰 갱신, 레이트 리미터, 해시키) |
| 두뇌 | `src/brain/gemini_client.py` | Gemini 프롬프트 구성 JSON 응답 파싱 |
| 방패 | `src/core/risk_manager.py` | 서킷 브레이커 + 팻 핑거 체크 |
| 알림 | `src/notifications/telegram_client.py` | 텔레그램 실시간 거래 알림 (선택사항) |
| 진화 | `src/evolution/optimizer.py` | 실패 패턴 분석 → 새 전략 생성 → 테스트 → PR |
| DB | `src/db.py` | SQLite 거래 로그 기록 |
| 설정 | `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개 테이블) |
## 안전장치
@@ -42,6 +55,7 @@ KIS(한국투자증권) API로 매매하고, Google Gemini로 판단하며, 자
| 신뢰도 임계값 | Gemini 신뢰도 80 미만이면 강제 HOLD |
| 레이트 리미터 | Leaky Bucket 알고리즘으로 API 호출 제한 |
| 토큰 자동 갱신 | 만료 1분 전 자동으로 Access Token 재발급 |
| 손절 모니터링 | 플레이북 시나리오 기반 실시간 포지션 보호 |
## 빠른 시작
@@ -67,7 +81,11 @@ pytest -v --cov=src --cov-report=term-missing
### 4. 실행 (모의투자)
```bash
# 기본 실행
python -m src.main --mode=paper
# 대시보드 활성화
python -m src.main --mode=paper --dashboard
```
### 5. Docker 실행
@@ -76,7 +94,20 @@ python -m src.main --mode=paper
docker compose up -d ouroboros
```
## 텔레그램 알림 (선택사항)
## 지원 시장
| 국가 | 거래소 | 코드 |
|------|--------|------|
| 🇰🇷 한국 | KRX | KR |
| 🇺🇸 미국 | NASDAQ, NYSE, AMEX | US_NASDAQ, US_NYSE, US_AMEX |
| 🇯🇵 일본 | TSE | JP |
| 🇭🇰 홍콩 | SEHK | HK |
| 🇨🇳 중국 | 상하이, 선전 | CN_SHA, CN_SZA |
| 🇻🇳 베트남 | 하노이, 호치민 | VN_HNX, VN_HSX |
`ENABLED_MARKETS` 환경변수로 활성 시장 선택 (기본: `KR,US`).
## 텔레그램 (선택사항)
거래 실행, 서킷 브레이커 발동, 시스템 상태 등을 텔레그램으로 실시간 알림 받을 수 있습니다.
@@ -102,25 +133,51 @@ docker compose up -d ouroboros
- 장 시작/종료 알림
- 📝 시스템 시작/종료 상태
**안전장치**: 알림 실패해도 거래는 계속 진행됩니다. 텔레그램 API 오류나 설정 누락이 있어도 거래 시스템은 정상 작동합니다.
### 양방향 명령어
`TELEGRAM_COMMANDS_ENABLED=true` (기본값) 설정 시 9개 대화형 명령어 지원:
| 명령어 | 설명 |
|--------|------|
| `/help` | 사용 가능한 명령어 목록 |
| `/status` | 거래 상태 (모드, 시장, P&L) |
| `/positions` | 계좌 요약 (잔고, 현금, P&L) |
| `/report` | 일일 요약 (거래 수, P&L, 승률) |
| `/scenarios` | 오늘의 플레이북 시나리오 |
| `/review` | 최근 스코어카드 (L6_DAILY) |
| `/dashboard` | 대시보드 URL 표시 |
| `/stop` | 거래 일시 정지 |
| `/resume` | 거래 재개 |
**안전장치**: 알림 실패해도 거래는 계속 진행됩니다.
## 테스트
35개 테스트가 TDD 방식으로 구현 전에 먼저 작성되었습니다.
551개 테스트가 25개 파일에 걸쳐 구현되어 있습니다. 최소 커버리지 80%.
```
tests/test_risk.py — 서킷 브레이커, 팻 핑거, 통합 검증 (11개)
tests/test_broker.py — 토큰 관리, 타임아웃, HTTP 에러, 해시키 (6개)
tests/test_brain.py JSON 파싱, 신뢰도 임계값, 비정상 응답 처리 (15개)
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개 파일
```
**상세**: [docs/testing.md](docs/testing.md)
## 기술 스택
- **언어**: Python 3.11+ (asyncio 기반)
- **브로커**: KIS Open API (REST)
- **브로커**: KIS Open API (REST, 국내+해외)
- **AI**: Google Gemini Pro
- **DB**: SQLite
- **검증**: pytest + coverage
- **DB**: SQLite (5개 테이블: trades, contexts, decision_logs, playbooks, context_metadata)
- **대시보드**: FastAPI + uvicorn
- **검증**: pytest + coverage (551 tests)
- **CI/CD**: GitHub Actions
- **배포**: Docker + Docker Compose
@@ -128,27 +185,50 @@ tests/test_brain.py — JSON 파싱, 신뢰도 임계값, 비정상 응답 처
```
The-Ouroboros/
├── .github/workflows/ci.yml # CI 파이프라인
├── docs/
│ ├── agents.md # AI 에이전트 페르소나 정의
── skills.md # 사용 가능한 도구 목록
│ ├── 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/
│ ├── config.py # Pydantic 설정
│ ├── logging_config.py # JSON 구조화 로깅
│ ├── db.py # SQLite 거래 기록
│ ├── main.py # 비동기 거래 루프
│ ├── broker/kis_api.py # KIS API 클라이언트
│ ├── brain/gemini_client.py # Gemini 의사결정 엔진
│ ├── core/risk_manager.py # 리스크 관리
│ ├── notifications/telegram_client.py # 텔레그램 알림
│ ├── evolution/optimizer.py # 전략 진화 엔진
── strategies/base.py # 전략 베이스 클래스
├── tests/ # TDD 테스트 스위트
│ ├── 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 # 의존성 및 도구 설정
```
## 문서
- **[아키텍처](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) 파일을 참조하세요.

View File

@@ -69,6 +69,10 @@ High-frequency trading with individual stock analysis:
- `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)
**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
### 2. Analysis (`src/analysis/`)
**VolatilityAnalyzer** (`volatility.py`) — Technical indicator calculations
@@ -82,16 +86,25 @@ High-frequency trading with individual stock analysis:
**SmartVolatilityScanner** (`smart_scanner.py`) — Python-first filtering pipeline
- **Step 1**: Fetch volume rankings from KIS API (top 30 stocks)
- **Step 2**: Calculate RSI and volume ratio for each stock
- **Step 3**: Apply filters:
- Volume ratio >= `VOL_MULTIPLIER` (default 2.0x previous day)
- RSI < `RSI_OVERSOLD_THRESHOLD` (30) OR RSI > `RSI_MOMENTUM_THRESHOLD` (70)
- **Step 4**: Score candidates by RSI extremity (60%) + volume surge (40%)
- **Step 5**: Return top N candidates (default 3) for AI analysis
- **Fallback**: Uses static watchlist if ranking API unavailable
- **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
### 3. Brain (`src/brain/`)
**GeminiClient** (`gemini_client.py`) — AI decision engine powered by Google Gemini
@@ -363,11 +376,13 @@ High-frequency trading with individual stock analysis:
┌──────────────────────────────────┐
│ Smart Scanner (Python-first) │
│ - Fetch volume rankings (KIS)
- Get 20d price history per stock
- Calculate RSI(14) + vol ratio
│ - Filter: vol>2x AND RSI extreme
│ 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 │
└──────────────────┬───────────────┘
@@ -568,6 +583,25 @@ S3_REGION=...
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`.

View File

@@ -111,3 +111,57 @@
- 이전 시도(2개 커밋)는 기존 내용을 과도하게 삭제하여 폐기, main 기준으로 재작업
**이슈/PR:** #131, PR #134
### 해외 스캐너 개선: 랭킹 연동 + 변동성 우선 선별
**배경:**
- `run_overnight` 실운영에서 미국장 동안 거래가 0건 지속
- 원인: 해외 시장에서도 국내 랭킹/일봉 API 경로를 사용하던 구조적 불일치
**요구사항:**
1. 해외 시장도 랭킹 API 기반 유니버스 탐색 지원
2. 단순 상승률/거래대금 상위가 아니라, **변동성이 큰 종목**을 우선 선별
3. 고정 티커 fallback 금지
**구현 결과:**
- `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_*` 설정 추가
**효과:**
- 국내/해외 스캐너 기준이 변동성 중심으로 일관화
- 고변동 구간에서 자동 익스포저 축소, 저변동 구간에서 과소진입 완화

View File

@@ -1,8 +1,4 @@
"""Smart Volatility Scanner with RSI and volume filters.
Fetches market rankings from KIS API and applies technical filters
to identify high-probability trading candidates.
"""
"""Smart Volatility Scanner with volatility-first market ranking logic."""
from __future__ import annotations
@@ -12,7 +8,9 @@ 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__)
@@ -32,19 +30,19 @@ class ScanCandidate:
class SmartVolatilityScanner:
"""Scans market rankings and applies RSI/volume filters.
"""Scans market rankings and applies volatility-first filters.
Flow:
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
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
"""
def __init__(
self,
broker: KISBroker,
overseas_broker: OverseasBroker | None,
volatility_analyzer: VolatilityAnalyzer,
settings: Settings,
) -> None:
@@ -56,6 +54,7 @@ class SmartVolatilityScanner:
settings: Application settings
"""
self.broker = broker
self.overseas_broker = overseas_broker
self.analyzer = volatility_analyzer
self.settings = settings
@@ -67,107 +66,129 @@ 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
"""
# Step 1: Fetch rankings
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.
try:
rankings = await self.broker.fetch_market_rankings(
ranking_type="volume",
limit=30, # Fetch more than needed for filtering
fluct_rows = await self.broker.fetch_market_rankings(
ranking_type="fluctuation",
limit=50,
)
logger.info("Fetched %d stocks from volume rankings", len(rankings))
except ConnectionError as exc:
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 []
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(
ranking_type="volume",
limit=50,
)
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)
# Step 2: Analyze each stock
candidates: list[ScanCandidate] = []
for stock in rankings:
stock_code = stock["stock_code"]
for stock in fluct_rows:
stock_code = _extract_stock_code(stock)
if not stock_code:
continue
try:
# Fetch daily prices for RSI calculation
daily_prices = await self.broker.get_daily_prices(stock_code, days=20)
price = _extract_last_price(stock)
change_rate = _extract_change_rate_pct(stock)
volume = _extract_volume(stock)
if len(daily_prices) < 15: # Need at least 14+1 for RSI
logger.debug("Insufficient price history for %s", stock_code)
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:
continue
# Calculate RSI
close_prices = [p["close"] for p in daily_prices]
rsi = self.analyzer.calculate_rsi(close_prices, period=14)
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 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,
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,
)
)
except ConnectionError as exc:
logger.warning("Failed to analyze %s: %s", stock_code, exc)
@@ -176,10 +197,161 @@ class SmartVolatilityScanner:
logger.error("Unexpected error analyzing %s: %s", stock_code, exc)
continue
# Sort by score and return top N
logger.info("Domestic ranking scan found %d candidates", len(candidates))
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 []
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)
return candidates
def get_stock_codes(self, candidates: list[ScanCandidate]) -> list[str]:
"""Extract stock codes from candidates for watchlist update.
@@ -190,3 +362,78 @@ 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

@@ -64,6 +64,65 @@ 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 amount).
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()
if ranking_type == "volume":
tr_id = self._broker._settings.OVERSEAS_RANKING_VOLUME_TR_ID
path = self._broker._settings.OVERSEAS_RANKING_VOLUME_PATH
else:
tr_id = self._broker._settings.OVERSEAS_RANKING_FLUCT_TR_ID
path = self._broker._settings.OVERSEAS_RANKING_FLUCT_PATH
headers = await self._broker._auth_headers(tr_id)
url = f"{self._broker._base_url}{path}"
# Try common param variants used by KIS overseas quotation APIs.
param_variants = [
{"AUTH": "", "EXCD": exchange_code, "NREC": str(max(limit, 30))},
{"AUTH": "", "OVRS_EXCG_CD": exchange_code, "NREC": str(max(limit, 30))},
{"AUTH": "", "EXCD": exchange_code},
{"AUTH": "", "OVRS_EXCG_CD": exchange_code},
]
last_error: str | None = None
for params in param_variants:
try:
async with session.get(url, headers=headers, params=params) as resp:
text = await resp.text()
if resp.status != 200:
last_error = f"HTTP {resp.status}: {text}"
continue
data = await resp.json()
rows = self._extract_ranking_rows(data)
if rows:
return rows[:limit]
# keep trying another param variant if response has no usable rows
last_error = f"empty output (keys={list(data.keys())})"
except (TimeoutError, aiohttp.ClientError) as exc:
last_error = str(exc)
continue
raise ConnectionError(
f"fetch_overseas_rankings failed for {exchange_code}/{ranking_type}: {last_error}"
)
async def get_overseas_balance(self, exchange_code: str) -> dict[str, Any]:
"""
Fetch overseas account balance.
@@ -198,3 +257,11 @@ 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,6 +38,11 @@ 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"
@@ -83,6 +88,18 @@ class Settings(BaseSettings):
TELEGRAM_COMMANDS_ENABLED: bool = True
TELEGRAM_POLLING_INTERVAL: float = 1.0 # seconds
# 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 = "HHDFS76200100"
OVERSEAS_RANKING_VOLUME_TR_ID: str = "HHDFS76200200"
OVERSEAS_RANKING_FLUCT_PATH: str = (
"/uapi/overseas-price/v1/quotations/inquire-updown-rank"
)
OVERSEAS_RANKING_VOLUME_PATH: str = (
"/uapi/overseas-price/v1/quotations/inquire-volume-rank"
)
# Dashboard (optional)
DASHBOARD_ENABLED: bool = False
DASHBOARD_HOST: str = "127.0.0.1"

View File

@@ -235,3 +235,21 @@ def get_open_position(
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]]

View File

@@ -29,7 +29,13 @@ from src.context.store import ContextStore
from src.core.criticality import CriticalityAssessor
from src.core.priority_queue import PriorityTaskQueue
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager
from src.db import get_latest_buy_trade, get_open_position, init_db, log_trade
from src.db import (
get_latest_buy_trade,
get_open_position,
get_recent_symbols,
init_db,
log_trade,
)
from src.evolution.daily_review import DailyReviewer
from src.evolution.optimizer import EvolutionOptimizer
from src.logging.decision_logger import DecisionLogger
@@ -81,6 +87,102 @@ DAILY_TRADE_SESSIONS = 4 # Number of trading sessions per day
TRADE_SESSION_INTERVAL_HOURS = 6 # Hours between sessions
def _extract_symbol_from_holding(item: dict[str, Any]) -> str:
"""Extract symbol from overseas holding payload variants."""
for key in (
"ovrs_pdno",
"pdno",
"ovrs_item_name",
"prdt_name",
"symb",
"symbol",
"stock_code",
):
value = item.get(key)
if isinstance(value, str):
symbol = value.strip().upper()
if symbol and symbol.replace(".", "").replace("-", "").isalnum():
return symbol
return ""
def _determine_order_quantity(
*,
action: str,
current_price: float,
total_cash: float,
candidate: ScanCandidate | None,
settings: Settings | None,
) -> int:
"""Determine order quantity using volatility-aware position sizing."""
if action != "BUY":
return 1
if current_price <= 0 or total_cash <= 0:
return 0
if settings is None or not settings.POSITION_SIZING_ENABLED:
return 1
target_score = max(1.0, settings.POSITION_VOLATILITY_TARGET_SCORE)
observed_score = candidate.score if candidate else target_score
observed_score = max(1.0, min(100.0, observed_score))
# Higher observed volatility score => smaller allocation.
scaled_pct = settings.POSITION_BASE_ALLOCATION_PCT * (target_score / observed_score)
allocation_pct = min(
settings.POSITION_MAX_ALLOCATION_PCT,
max(settings.POSITION_MIN_ALLOCATION_PCT, scaled_pct),
)
budget = total_cash * (allocation_pct / 100.0)
quantity = int(budget // current_price)
if quantity <= 0:
return 0
return quantity
async def build_overseas_symbol_universe(
db_conn: Any,
overseas_broker: OverseasBroker,
market: MarketInfo,
active_stocks: dict[str, list[str]],
) -> list[str]:
"""Build dynamic overseas symbol universe from runtime, DB, and holdings."""
symbols: list[str] = []
# 1) Keep current active stocks first to avoid sudden churn between cycles.
symbols.extend(active_stocks.get(market.code, []))
# 2) Add recent symbols from own trading history (no fixed list).
symbols.extend(get_recent_symbols(db_conn, market.code, limit=30))
# 3) Add current overseas holdings from broker balance if available.
try:
balance_data = await overseas_broker.get_overseas_balance(market.exchange_code)
output1 = balance_data.get("output1", [])
if isinstance(output1, dict):
output1 = [output1]
if isinstance(output1, list):
for row in output1:
if not isinstance(row, dict):
continue
symbol = _extract_symbol_from_holding(row)
if symbol:
symbols.append(symbol)
except Exception as exc:
logger.warning("Failed to build overseas holdings universe for %s: %s", market.code, exc)
seen: set[str] = set()
ordered_unique: list[str] = []
for symbol in symbols:
normalized = symbol.strip().upper()
if not normalized or normalized in seen:
continue
seen.add(normalized)
ordered_unique.append(normalized)
return ordered_unique
async def trading_cycle(
broker: KISBroker,
overseas_broker: OverseasBroker,
@@ -95,6 +197,7 @@ async def trading_cycle(
market: MarketInfo,
stock_code: str,
scan_candidates: dict[str, dict[str, ScanCandidate]],
settings: Settings | None = None,
) -> None:
"""Execute one trading cycle for a single stock."""
cycle_start_time = asyncio.get_event_loop().time()
@@ -332,8 +435,23 @@ async def trading_cycle(
trade_price = current_price
trade_pnl = 0.0
if decision.action in ("BUY", "SELL"):
# Determine order size (simplified: 1 lot)
quantity = 1
quantity = _determine_order_quantity(
action=decision.action,
current_price=current_price,
total_cash=total_cash,
candidate=candidate,
settings=settings,
)
if quantity <= 0:
logger.info(
"Skip %s %s (%s): no affordable quantity (cash=%.2f, price=%.2f)",
decision.action,
stock_code,
market.name,
total_cash,
current_price,
)
return
order_amount = current_price * quantity
# 4. Risk check BEFORE order
@@ -482,8 +600,28 @@ async def run_daily_session(
# Dynamic stock discovery via scanner (no static watchlists)
candidates_list: list[ScanCandidate] = []
fallback_stocks: list[str] | None = None
if not market.is_domestic:
fallback_stocks = await build_overseas_symbol_universe(
db_conn=db_conn,
overseas_broker=overseas_broker,
market=market,
active_stocks={},
)
if not fallback_stocks:
logger.warning(
"No dynamic overseas symbol universe for %s; scanner cannot run",
market.code,
)
try:
candidates_list = await smart_scanner.scan() if smart_scanner else []
candidates_list = (
await smart_scanner.scan(
market=market,
fallback_stocks=fallback_stocks,
)
if smart_scanner
else []
)
except Exception as exc:
logger.error("Smart Scanner failed for %s: %s", market.name, exc)
@@ -679,7 +817,23 @@ async def run_daily_session(
trade_price = stock_data["current_price"]
trade_pnl = 0.0
if decision.action in ("BUY", "SELL"):
quantity = 1
quantity = _determine_order_quantity(
action=decision.action,
current_price=stock_data["current_price"],
total_cash=total_cash,
candidate=candidate_map.get(stock_code),
settings=settings,
)
if quantity <= 0:
logger.info(
"Skip %s %s (%s): no affordable quantity (cash=%.2f, price=%.2f)",
decision.action,
stock_code,
market.name,
total_cash,
stock_data["current_price"],
)
continue
order_amount = stock_data["current_price"] * quantity
# Risk check
@@ -1263,6 +1417,7 @@ async def run(settings: Settings) -> None:
# Initialize smart scanner (Python-first, AI-last pipeline)
smart_scanner = SmartVolatilityScanner(
broker=broker,
overseas_broker=overseas_broker,
volatility_analyzer=volatility_analyzer,
settings=settings,
)
@@ -1442,7 +1597,25 @@ async def run(settings: Settings) -> None:
try:
logger.info("Smart Scanner: Scanning %s market", market.name)
candidates = await smart_scanner.scan()
fallback_stocks: list[str] | None = None
if not market.is_domestic:
fallback_stocks = await build_overseas_symbol_universe(
db_conn=db_conn,
overseas_broker=overseas_broker,
market=market,
active_stocks=active_stocks,
)
if not fallback_stocks:
logger.warning(
"No dynamic overseas symbol universe for %s;"
" scanner cannot run",
market.code,
)
candidates = await smart_scanner.scan(
market=market,
fallback_stocks=fallback_stocks,
)
if candidates:
# Use scanner results directly as trading candidates
@@ -1566,6 +1739,7 @@ async def run(settings: Settings) -> None:
market,
stock_code,
scan_candidates,
settings,
)
break # Success — exit retry loop
except CircuitBreakerTripped as exc:

View File

@@ -8,6 +8,7 @@ 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
@@ -43,61 +44,70 @@ 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_finds_oversold_candidates(
async def test_scan_domestic_prefers_volatility_with_liquidity_bonus(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Test that scanner identifies oversold stocks with high volume."""
# Mock rankings
mock_broker.fetch_market_rankings.return_value = [
"""Domestic scan should score by volatility first and volume rank second."""
fluctuation_rows = [
{
"stock_code": "005930",
"name": "Samsung",
"price": 70000,
"volume": 5000000,
"change_rate": -3.5,
"change_rate": -5.0,
"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()
# 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
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"
@pytest.mark.asyncio
async def test_scan_finds_momentum_candidates(
async def test_scan_domestic_finds_momentum_candidate(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Test that scanner identifies momentum stocks with high volume."""
mock_broker.fetch_market_rankings.return_value = [
"""Positive change should be represented as momentum signal."""
fluctuation_rows = [
{
"stock_code": "035420",
"name": "NAVER",
@@ -107,124 +117,67 @@ class TestSmartVolatilityScanner:
"volume_increase_rate": 300,
},
]
# 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
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},
]
candidates = await scanner.scan()
mock_broker.fetch_market_rankings.assert_called_once()
assert [c.stock_code for c in candidates] == ["035420"]
assert candidates[0].signal == "momentum"
@pytest.mark.asyncio
async def test_scan_filters_low_volume(
async def test_scan_domestic_filters_low_volatility(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None:
"""Test that stocks with low volume ratio are filtered out."""
mock_broker.fetch_market_rankings.return_value = [
"""Domestic scan should drop symbols below volatility threshold."""
fluctuation_rows = [
{
"stock_code": "000660",
"name": "SK Hynix",
"price": 150000,
"volume": 500000,
"change_rate": -5.0,
"volume_increase_rate": 50, # Only 50% increase (< 200%)
"change_rate": 0.2,
"volume_increase_rate": 50,
},
]
# 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
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},
]
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:
"""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
"""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},
]
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."""
# Return many stocks
mock_broker.fetch_market_rankings.return_value = [
fluctuation_rows = [
{
"stock_code": f"00{i}000",
"name": f"Stock{i}",
@@ -235,62 +188,17 @@ class TestSmartVolatilityScanner:
}
for i in range(1, 10)
]
# 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
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},
]
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
@@ -323,6 +231,124 @@ 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."""