Compare commits

...

11 Commits

Author SHA1 Message Date
agentson
e7de9aa894 Add scanner diagnostics for zero-candidate trade stalls 2026-02-18 00:11:53 +09:00
agentson
aeed881d85 fix: wait on token refresh cooldown instead of failing fast
Some checks failed
CI / test (pull_request) Has been cancelled
2026-02-18 00:03:42 +09:00
agentson
d0bbdb5dc1 fix: harden overseas ranking fallback and scanner visibility 2026-02-17 23:39:20 +09:00
agentson
22ffdafacc chore: add overnight helper scripts
Some checks failed
CI / test (pull_request) Has been cancelled
- add morning report launcher\n- add overnight stop script\n- add watchdog health monitor script\n\nRefs #137
2026-02-17 23:24:15 +09:00
agentson
c49765e951 fix: make overnight runner use venv python and tmux-safe window target
Some checks failed
CI / test (pull_request) Has been cancelled
- prefer .venv/bin/python when APP_CMD is unset\n- pass DASHBOARD_PORT into launch command (default 8080)\n- target tmux window by name instead of fixed index\n\nRefs #137
2026-02-17 23:21:04 +09:00
64000b9967 Merge pull request 'feat: unify domestic scanner and sizing; update docs' (#136) from feat/overseas-ranking-current-state into main
Some checks failed
CI / test (push) Has been cancelled
Reviewed-on: #136
2026-02-17 06:35:43 +09:00
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
15 changed files with 1329 additions and 277 deletions

View File

@@ -69,6 +69,10 @@ High-frequency trading with individual stock analysis:
- `get_next_market_open()` finds next market to open and when - `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) - 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/`) ### 2. Analysis (`src/analysis/`)
**VolatilityAnalyzer** (`volatility.py`) — Technical indicator calculations **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 **SmartVolatilityScanner** (`smart_scanner.py`) — Python-first filtering pipeline
- **Step 1**: Fetch volume rankings from KIS API (top 30 stocks) - **Domestic (KR)**:
- **Step 2**: Calculate RSI and volume ratio for each stock - **Step 1**: Fetch domestic fluctuation ranking as primary universe
- **Step 3**: Apply filters: - **Step 2**: Fetch domestic volume ranking for liquidity bonus
- Volume ratio >= `VOL_MULTIPLIER` (default 2.0x previous day) - **Step 3**: Compute volatility-first score (max of daily change% and intraday range%)
- RSI < `RSI_OVERSOLD_THRESHOLD` (30) OR RSI > `RSI_MOMENTUM_THRESHOLD` (70) - **Step 4**: Apply liquidity bonus and return top N candidates
- **Step 4**: Score candidates by RSI extremity (60%) + volume surge (40%) - **Overseas (US/JP/HK/CN/VN)**:
- **Step 5**: Return top N candidates (default 3) for AI analysis - **Step 1**: Fetch overseas ranking universe (fluctuation rank + volume rank bonus)
- **Fallback**: Uses static watchlist if ranking API unavailable - **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 - **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/`) ### 3. Brain (`src/brain/`)
**GeminiClient** (`gemini_client.py`) — AI decision engine powered by Google Gemini **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) │ │ Smart Scanner (Python-first)
│ - Fetch volume rankings (KIS) │ - Domestic: fluctuation rank
- Get 20d price history per stock + volume rank bonus
- Calculate RSI(14) + vol ratio + volatility-first scoring
│ - Filter: vol>2x AND RSI extreme │ - Overseas: ranking universe
│ + volatility-first scoring │
│ - Fallback: dynamic universe │
│ - Return top 3 qualified stocks │ │ - Return top 3 qualified stocks │
└──────────────────┬───────────────┘ └──────────────────┬───────────────┘
@@ -568,6 +583,25 @@ S3_REGION=...
NEWS_API_KEY=... NEWS_API_KEY=...
NEWS_API_PROVIDER=... NEWS_API_PROVIDER=...
MARKET_DATA_API_KEY=... 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`. Tests use in-memory SQLite (`DB_PATH=":memory:"`) and dummy credentials via `tests/conftest.py`.

View File

@@ -0,0 +1,29 @@
# Issue: Realtime 모드에서 거래가 지속적으로 0건
## Summary
`realtime` 실행 중 주문 단계까지 진입하지 못하고, 스캐너 단계에서 후보가 0건으로 반복 종료된다.
## Observed
- 로그에서 반복적으로 `Smart Scanner: No candidates ... — no trades` 출력
- 해외 시장에서 `Overseas ranking endpoint unavailable (404)` 다수 발생
- fallback 심볼 스캔도 `0 candidates`로 종료
- `data/trade_logs.db` 기준 최근 구간에 `BUY/SELL` 없음
## Impact
- 매매 전략 품질과 무관하게 주문 경로가 실행되지 않아 실질 거래 불가
- 장애 원인을 로그만으로 즉시 분해하기 어려움
## Root-Cause Hypothesis
- 스캐너 필터(가격/변동성) 단계에서 대부분 탈락
- 해외 랭킹 API 불가 시 입력 유니버스가 빈 상태가 되어 후보 생성 실패
- 기존 로그는 최종 결과(0 candidates)만 보여 원인별 분해가 어려움
## Acceptance Criteria
- 스캔 1회마다 탈락 사유가 구조화되어 로그에 남아야 함
- 국내/해외(랭킹/폴백) 경로 모두 동일한 진단 지표를 제공해야 함
- 운영자가 로그만 보고 `왜 0 candidates인지`를 즉시 판단 가능해야 함
## Scope
- 이번 이슈는 **진단 가능성 개선(Observability)** 에 한정
- 후보 생성 전략 변경(기본 유니버스 강제 추가 등)은 별도 이슈로 분리

View File

@@ -0,0 +1,32 @@
# PR: Smart Scanner 진단 로그 추가 (0 candidates 원인 분해)
## Linked Issue
- `docs/issues/ISSUE-2026-02-17-no-trades-zero-candidates.md`
## What Changed
- `src/analysis/smart_scanner.py`에 스캔 진단 카운터 추가
- 국내 스캔 진단 로그 추가
- 해외 랭킹 스캔 진단 로그 추가
- 해외 fallback 심볼 스캔 진단 로그 추가
## Diagnostics Keys
- `total_rows`
- `missing_code`
- `invalid_price`
- `low_volatility`
- `connection_error` (해당 경로에서만)
- `unexpected_error` (해당 경로에서만)
- `qualified`
## Expected Log Examples
- `Domestic scan diagnostics: {...}`
- `Overseas ranking scan diagnostics for US_NASDAQ: {...}`
- `Overseas fallback scan diagnostics for US_NYSE: {...}`
## Out of Scope
- 해외 랭킹 404 시 기본 심볼 유니버스 강제 주입
- 국내 경로 fallback 정책 변경
## Validation
- `.venv/bin/python -m py_compile src/analysis/smart_scanner.py`

View File

@@ -111,3 +111,57 @@
- 이전 시도(2개 커밋)는 기존 내용을 과도하게 삭제하여 폐기, main 기준으로 재작업 - 이전 시도(2개 커밋)는 기존 내용을 과도하게 삭제하여 폐기, main 기준으로 재작업
**이슈/PR:** #131, PR #134 **이슈/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_*` 설정 추가
**효과:**
- 국내/해외 스캐너 기준이 변동성 중심으로 일관화
- 고변동 구간에서 자동 익스포저 축소, 저변동 구간에서 과소진입 완화

54
scripts/morning_report.sh Executable file
View File

@@ -0,0 +1,54 @@
#!/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

87
scripts/run_overnight.sh Executable file
View File

@@ -0,0 +1,87 @@
#!/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

76
scripts/stop_overnight.sh Executable file
View File

@@ -0,0 +1,76 @@
#!/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"

42
scripts/watchdog.sh Executable file
View File

@@ -0,0 +1,42 @@
#!/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,8 +1,4 @@
"""Smart Volatility Scanner with RSI and volume filters. """Smart Volatility Scanner with volatility-first market ranking logic."""
Fetches market rankings from KIS API and applies technical filters
to identify high-probability trading candidates.
"""
from __future__ import annotations from __future__ import annotations
@@ -12,7 +8,9 @@ from typing import Any
from src.analysis.volatility import VolatilityAnalyzer from src.analysis.volatility import VolatilityAnalyzer
from src.broker.kis_api import KISBroker from src.broker.kis_api import KISBroker
from src.broker.overseas import OverseasBroker
from src.config import Settings from src.config import Settings
from src.markets.schedule import MarketInfo
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -32,19 +30,19 @@ class ScanCandidate:
class SmartVolatilityScanner: class SmartVolatilityScanner:
"""Scans market rankings and applies RSI/volume filters. """Scans market rankings and applies volatility-first filters.
Flow: Flow:
1. Fetch volume rankings from KIS API 1. Fetch fluctuation rankings as primary universe
2. For each ranked stock, fetch daily prices 2. Fetch volume rankings for liquidity bonus
3. Calculate RSI and volume ratio 3. Score by volatility first, liquidity second
4. Apply filters: volume > VOL_MULTIPLIER AND (RSI < 30 OR RSI > 70) 4. Return top N qualified candidates
5. Return top N qualified candidates
""" """
def __init__( def __init__(
self, self,
broker: KISBroker, broker: KISBroker,
overseas_broker: OverseasBroker | None,
volatility_analyzer: VolatilityAnalyzer, volatility_analyzer: VolatilityAnalyzer,
settings: Settings, settings: Settings,
) -> None: ) -> None:
@@ -56,6 +54,7 @@ class SmartVolatilityScanner:
settings: Application settings settings: Application settings
""" """
self.broker = broker self.broker = broker
self.overseas_broker = overseas_broker
self.analyzer = volatility_analyzer self.analyzer = volatility_analyzer
self.settings = settings self.settings = settings
@@ -67,119 +66,366 @@ class SmartVolatilityScanner:
async def scan( async def scan(
self, self,
market: MarketInfo | None = None,
fallback_stocks: list[str] | None = None, fallback_stocks: list[str] | None = None,
) -> list[ScanCandidate]: ) -> list[ScanCandidate]:
"""Execute smart scan and return qualified candidates. """Execute smart scan and return qualified candidates.
Args: Args:
market: Target market info (domestic vs overseas behavior)
fallback_stocks: Stock codes to use if ranking API fails fallback_stocks: Stock codes to use if ranking API fails
Returns: Returns:
List of ScanCandidate, sorted by score, up to top_n items 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: try:
rankings = await self.broker.fetch_market_rankings( fluct_rows = await self.broker.fetch_market_rankings(
ranking_type="volume", ranking_type="fluctuation",
limit=30, # Fetch more than needed for filtering limit=50,
) )
logger.info("Fetched %d stocks from volume rankings", len(rankings))
except ConnectionError as exc: except ConnectionError as exc:
logger.warning("Ranking API failed, using fallback: %s", exc) logger.warning("Domestic fluctuation ranking failed: %s", exc)
if fallback_stocks: fluct_rows = []
# Create minimal ranking data for fallback
rankings = [ # 2) Liquidity bonus from volume ranking.
{ try:
"stock_code": code, volume_rows = await self.broker.fetch_market_rankings(
"name": code, ranking_type="volume",
"price": 0, limit=50,
"volume": 0, )
"change_rate": 0, except ConnectionError as exc:
"volume_increase_rate": 0, logger.warning("Domestic volume ranking failed: %s", exc)
} volume_rows = []
for code in fallback_stocks
] if not fluct_rows and fallback_stocks:
else: logger.info(
return [] "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 []
diagnostics: dict[str, int | float] = {
"total_rows": len(fluct_rows),
"missing_code": 0,
"invalid_price": 0,
"low_volatility": 0,
"connection_error": 0,
"unexpected_error": 0,
"qualified": 0,
}
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] = [] candidates: list[ScanCandidate] = []
for stock in fluct_rows:
for stock in rankings: stock_code = _extract_stock_code(stock)
stock_code = stock["stock_code"]
if not stock_code: if not stock_code:
diagnostics["missing_code"] += 1
continue continue
try: try:
# Fetch daily prices for RSI calculation price = _extract_last_price(stock)
daily_prices = await self.broker.get_daily_prices(stock_code, days=20) change_rate = _extract_change_rate_pct(stock)
volume = _extract_volume(stock)
if len(daily_prices) < 15: # Need at least 14+1 for RSI intraday_range_pct = 0.0
logger.debug("Insufficient price history for %s", stock_code) 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:
diagnostics["invalid_price"] += 1
continue
if volatility_pct < 0.8:
diagnostics["low_volatility"] += 1
continue continue
# Calculate RSI volatility_score = min(volatility_pct / 10.0, 1.0) * 85.0
close_prices = [p["close"] for p in daily_prices] liquidity_score = volume_rank_bonus.get(stock_code, 0.0)
rsi = self.analyzer.calculate_rsi(close_prices, period=14) 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) candidates.append(
if len(daily_prices) >= 2: ScanCandidate(
prev_day_volume = daily_prices[-2]["volume"] stock_code=stock_code,
current_volume = stock.get("volume", 0) or daily_prices[-1]["volume"] name=stock.get("name", stock_code),
volume_ratio = ( price=price,
current_volume / prev_day_volume if prev_day_volume > 0 else 1.0 volume=volume,
) volume_ratio=max(1.0, volume_ratio, volatility_pct / 2.0),
else: rsi=implied_rsi,
volume_ratio = stock.get("volume_increase_rate", 0) / 100 + 1 # Fallback signal=signal,
score=score,
# 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,
) )
)
diagnostics["qualified"] += 1
except ConnectionError as exc: except ConnectionError as exc:
diagnostics["connection_error"] += 1
logger.warning("Failed to analyze %s: %s", stock_code, exc) logger.warning("Failed to analyze %s: %s", stock_code, exc)
continue continue
except Exception as exc: except Exception as exc:
diagnostics["unexpected_error"] += 1
logger.error("Unexpected error analyzing %s: %s", stock_code, exc) logger.error("Unexpected error analyzing %s: %s", stock_code, exc)
continue continue
# Sort by score and return top N logger.info(
"Domestic scan diagnostics: %s (volatility_threshold=0.8, top_n=%d)",
diagnostics,
self.top_n,
)
logger.info("Domestic ranking scan found %d candidates", len(candidates))
candidates.sort(key=lambda c: c.score, reverse=True) candidates.sort(key=lambda c: c.score, reverse=True)
return candidates[: self.top_n] 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 []
diagnostics: dict[str, int | float] = {
"total_rows": len(fluct_rows),
"missing_code": 0,
"invalid_price": 0,
"low_volatility": 0,
"qualified": 0,
}
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:
diagnostics["missing_code"] += 1
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:
diagnostics["invalid_price"] += 1
continue
if volatility_pct < 0.8:
diagnostics["low_volatility"] += 1
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,
)
)
diagnostics["qualified"] += 1
logger.info(
"Overseas ranking scan diagnostics for %s: %s (volatility_threshold=0.8, top_n=%d)",
market.code,
diagnostics,
self.top_n,
)
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,
)
diagnostics: dict[str, int | float] = {
"total_rows": len(symbols),
"invalid_price": 0,
"low_volatility": 0,
"connection_error": 0,
"unexpected_error": 0,
"qualified": 0,
}
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:
diagnostics["invalid_price"] += 1
continue
if volatility_pct < 0.8:
diagnostics["low_volatility"] += 1
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,
)
)
diagnostics["qualified"] += 1
except ConnectionError as exc:
diagnostics["connection_error"] += 1
logger.warning("Failed to analyze overseas %s: %s", stock_code, exc)
except Exception as exc:
diagnostics["unexpected_error"] += 1
logger.error("Unexpected error analyzing overseas %s: %s", stock_code, exc)
logger.info(
"Overseas fallback scan diagnostics for %s: %s (volatility_threshold=0.8, top_n=%d)",
market.code,
diagnostics,
self.top_n,
)
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]: def get_stock_codes(self, candidates: list[ScanCandidate]) -> list[str]:
"""Extract stock codes from candidates for watchlist update. """Extract stock codes from candidates for watchlist update.
@@ -190,3 +436,78 @@ class SmartVolatilityScanner:
List of stock codes List of stock codes
""" """
return [c.stock_code for c in candidates] 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

@@ -104,12 +104,14 @@ class KISBroker:
time_since_last_attempt = now - self._last_refresh_attempt time_since_last_attempt = now - self._last_refresh_attempt
if time_since_last_attempt < self._refresh_cooldown: if time_since_last_attempt < self._refresh_cooldown:
remaining = self._refresh_cooldown - time_since_last_attempt remaining = self._refresh_cooldown - time_since_last_attempt
error_msg = ( # Do not fail fast here. If token is unavailable, upstream calls
f"Token refresh on cooldown. " # will all fail for up to a minute and scanning returns no trades.
f"Retry in {remaining:.1f}s (KIS allows 1/minute)" logger.warning(
"Token refresh on cooldown. Waiting %.1fs before retry (KIS allows 1/minute)",
remaining,
) )
logger.warning(error_msg) await asyncio.sleep(remaining)
raise ConnectionError(error_msg) now = asyncio.get_event_loop().time()
logger.info("Refreshing KIS access token") logger.info("Refreshing KIS access token")
self._last_refresh_attempt = now self._last_refresh_attempt = now

View File

@@ -64,6 +64,84 @@ class OverseasBroker:
f"Network error fetching overseas price: {exc}" f"Network error fetching overseas price: {exc}"
) from 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":
configured_tr_id = self._broker._settings.OVERSEAS_RANKING_VOLUME_TR_ID
configured_path = self._broker._settings.OVERSEAS_RANKING_VOLUME_PATH
default_tr_id = "HHDFS76200200"
default_path = "/uapi/overseas-price/v1/quotations/inquire-volume-rank"
else:
configured_tr_id = self._broker._settings.OVERSEAS_RANKING_FLUCT_TR_ID
configured_path = self._broker._settings.OVERSEAS_RANKING_FLUCT_PATH
default_tr_id = "HHDFS76200100"
default_path = "/uapi/overseas-price/v1/quotations/inquire-updown-rank"
endpoint_specs: list[tuple[str, str]] = [(configured_tr_id, configured_path)]
if (configured_tr_id, configured_path) != (default_tr_id, default_path):
endpoint_specs.append((default_tr_id, default_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
saw_http_404 = False
for tr_id, path in endpoint_specs:
headers = await self._broker._auth_headers(tr_id)
url = f"{self._broker._base_url}{path}"
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}"
if resp.status == 404:
saw_http_404 = True
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
if saw_http_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 for {exchange_code}/{ranking_type}: {last_error}"
)
async def get_overseas_balance(self, exchange_code: str) -> dict[str, Any]: async def get_overseas_balance(self, exchange_code: str) -> dict[str, Any]:
""" """
Fetch overseas account balance. Fetch overseas account balance.
@@ -198,3 +276,11 @@ class OverseasBroker:
"HSX": "VND", "HSX": "VND",
} }
return currency_map.get(exchange_code, "USD") 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) RSI_MOMENTUM_THRESHOLD: int = Field(default=70, ge=50, le=100)
VOL_MULTIPLIER: float = Field(default=2.0, gt=1.0, le=10.0) VOL_MULTIPLIER: float = Field(default=2.0, gt=1.0, le=10.0)
SCANNER_TOP_N: int = Field(default=3, ge=1, le=10) 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 # Database
DB_PATH: str = "data/trade_logs.db" DB_PATH: str = "data/trade_logs.db"
@@ -83,6 +88,18 @@ class Settings(BaseSettings):
TELEGRAM_COMMANDS_ENABLED: bool = True TELEGRAM_COMMANDS_ENABLED: bool = True
TELEGRAM_POLLING_INTERVAL: float = 1.0 # seconds 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 (optional)
DASHBOARD_ENABLED: bool = False DASHBOARD_ENABLED: bool = False
DASHBOARD_HOST: str = "127.0.0.1" DASHBOARD_HOST: str = "127.0.0.1"

View File

@@ -235,3 +235,21 @@ def get_open_position(
if not row or row[0] != "BUY": if not row or row[0] != "BUY":
return None return None
return {"decision_id": row[1], "price": row[2], "quantity": row[3]} 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.criticality import CriticalityAssessor
from src.core.priority_queue import PriorityTaskQueue from src.core.priority_queue import PriorityTaskQueue
from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager from src.core.risk_manager import CircuitBreakerTripped, FatFingerRejected, RiskManager
from src.db import 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.daily_review import DailyReviewer
from src.evolution.optimizer import EvolutionOptimizer from src.evolution.optimizer import EvolutionOptimizer
from src.logging.decision_logger import DecisionLogger 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 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( async def trading_cycle(
broker: KISBroker, broker: KISBroker,
overseas_broker: OverseasBroker, overseas_broker: OverseasBroker,
@@ -95,6 +197,7 @@ async def trading_cycle(
market: MarketInfo, market: MarketInfo,
stock_code: str, stock_code: str,
scan_candidates: dict[str, dict[str, ScanCandidate]], scan_candidates: dict[str, dict[str, ScanCandidate]],
settings: Settings | None = None,
) -> None: ) -> None:
"""Execute one trading cycle for a single stock.""" """Execute one trading cycle for a single stock."""
cycle_start_time = asyncio.get_event_loop().time() cycle_start_time = asyncio.get_event_loop().time()
@@ -332,8 +435,23 @@ async def trading_cycle(
trade_price = current_price trade_price = current_price
trade_pnl = 0.0 trade_pnl = 0.0
if decision.action in ("BUY", "SELL"): if decision.action in ("BUY", "SELL"):
# Determine order size (simplified: 1 lot) quantity = _determine_order_quantity(
quantity = 1 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 order_amount = current_price * quantity
# 4. Risk check BEFORE order # 4. Risk check BEFORE order
@@ -482,8 +600,28 @@ async def run_daily_session(
# Dynamic stock discovery via scanner (no static watchlists) # Dynamic stock discovery via scanner (no static watchlists)
candidates_list: list[ScanCandidate] = [] 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: 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: except Exception as exc:
logger.error("Smart Scanner failed for %s: %s", market.name, exc) logger.error("Smart Scanner failed for %s: %s", market.name, exc)
@@ -679,7 +817,23 @@ async def run_daily_session(
trade_price = stock_data["current_price"] trade_price = stock_data["current_price"]
trade_pnl = 0.0 trade_pnl = 0.0
if decision.action in ("BUY", "SELL"): 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 order_amount = stock_data["current_price"] * quantity
# Risk check # Risk check
@@ -1263,6 +1417,7 @@ async def run(settings: Settings) -> None:
# Initialize smart scanner (Python-first, AI-last pipeline) # Initialize smart scanner (Python-first, AI-last pipeline)
smart_scanner = SmartVolatilityScanner( smart_scanner = SmartVolatilityScanner(
broker=broker, broker=broker,
overseas_broker=overseas_broker,
volatility_analyzer=volatility_analyzer, volatility_analyzer=volatility_analyzer,
settings=settings, settings=settings,
) )
@@ -1442,7 +1597,25 @@ async def run(settings: Settings) -> None:
try: try:
logger.info("Smart Scanner: Scanning %s market", market.name) 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: if candidates:
# Use scanner results directly as trading candidates # Use scanner results directly as trading candidates
@@ -1566,6 +1739,7 @@ async def run(settings: Settings) -> None:
market, market,
stock_code, stock_code,
scan_candidates, scan_candidates,
settings,
) )
break # Success — exit retry loop break # Success — exit retry loop
except CircuitBreakerTripped as exc: 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.smart_scanner import ScanCandidate, SmartVolatilityScanner
from src.analysis.volatility import VolatilityAnalyzer from src.analysis.volatility import VolatilityAnalyzer
from src.broker.kis_api import KISBroker from src.broker.kis_api import KISBroker
from src.broker.overseas import OverseasBroker
from src.config import Settings from src.config import Settings
@@ -43,61 +44,70 @@ def scanner(mock_broker: MagicMock, mock_settings: Settings) -> SmartVolatilityS
analyzer = VolatilityAnalyzer() analyzer = VolatilityAnalyzer()
return SmartVolatilityScanner( return SmartVolatilityScanner(
broker=mock_broker, broker=mock_broker,
overseas_broker=None,
volatility_analyzer=analyzer, volatility_analyzer=analyzer,
settings=mock_settings, 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: class TestSmartVolatilityScanner:
"""Test suite for SmartVolatilityScanner.""" """Test suite for SmartVolatilityScanner."""
@pytest.mark.asyncio @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 self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None: ) -> None:
"""Test that scanner identifies oversold stocks with high volume.""" """Domestic scan should score by volatility first and volume rank second."""
# Mock rankings fluctuation_rows = [
mock_broker.fetch_market_rankings.return_value = [
{ {
"stock_code": "005930", "stock_code": "005930",
"name": "Samsung", "name": "Samsung",
"price": 70000, "price": 70000,
"volume": 5000000, "volume": 5000000,
"change_rate": -3.5, "change_rate": -5.0,
"volume_increase_rate": 250, "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() candidates = await scanner.scan()
# Should find at least one candidate (depending on exact RSI calculation) assert len(candidates) >= 1
mock_broker.fetch_market_rankings.assert_called_once() # Samsung has higher absolute move, so it should lead despite lower volume rank bonus.
mock_broker.get_daily_prices.assert_called_once_with("005930", days=20) assert candidates[0].stock_code == "005930"
assert candidates[0].signal == "oversold"
# 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 @pytest.mark.asyncio
async def test_scan_finds_momentum_candidates( async def test_scan_domestic_finds_momentum_candidate(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None: ) -> None:
"""Test that scanner identifies momentum stocks with high volume.""" """Positive change should be represented as momentum signal."""
mock_broker.fetch_market_rankings.return_value = [ fluctuation_rows = [
{ {
"stock_code": "035420", "stock_code": "035420",
"name": "NAVER", "name": "NAVER",
@@ -107,124 +117,67 @@ class TestSmartVolatilityScanner:
"volume_increase_rate": 300, "volume_increase_rate": 300,
}, },
] ]
mock_broker.fetch_market_rankings.side_effect = [fluctuation_rows, fluctuation_rows]
# Mock daily prices - trending up (momentum) mock_broker.get_daily_prices.return_value = [
prices = [] {"open": 1, "high": 1, "low": 1, "close": 1, "volume": 1000000},
for i in range(20): {"open": 1, "high": 1, "low": 1, "close": 1, "volume": 1000000},
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() 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 @pytest.mark.asyncio
async def test_scan_filters_low_volume( async def test_scan_domestic_filters_low_volatility(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None: ) -> None:
"""Test that stocks with low volume ratio are filtered out.""" """Domestic scan should drop symbols below volatility threshold."""
mock_broker.fetch_market_rankings.return_value = [ fluctuation_rows = [
{ {
"stock_code": "000660", "stock_code": "000660",
"name": "SK Hynix", "name": "SK Hynix",
"price": 150000, "price": 150000,
"volume": 500000, "volume": 500000,
"change_rate": -5.0, "change_rate": 0.2,
"volume_increase_rate": 50, # Only 50% increase (< 200%) "volume_increase_rate": 50,
}, },
] ]
mock_broker.fetch_market_rankings.side_effect = [fluctuation_rows, fluctuation_rows]
# Low volume mock_broker.get_daily_prices.return_value = [
prices = [] {"open": 1, "high": 150100, "low": 149900, "close": 150000, "volume": 1000000},
for i in range(20): {"open": 1, "high": 150100, "low": 149900, "close": 150000, "volume": 1000000},
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() 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 assert len(candidates) == 0
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_scan_uses_fallback_on_api_error( async def test_scan_uses_fallback_on_api_error(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None: ) -> None:
"""Test fallback to static list when ranking API fails.""" """Domestic scan should remain operational using fallback symbols."""
mock_broker.fetch_market_rankings.side_effect = ConnectionError("API unavailable") mock_broker.fetch_market_rankings.side_effect = [
ConnectionError("API unavailable"),
# Fallback stocks should still be analyzed ConnectionError("API unavailable"),
prices = [] ]
for i in range(20): mock_broker.get_daily_prices.return_value = [
prices.append({ {"open": 1, "high": 103, "low": 97, "close": 100, "volume": 1000000},
"date": f"2026020{i:02d}", {"open": 1, "high": 103, "low": 97, "close": 100, "volume": 800000},
"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"]) candidates = await scanner.scan(fallback_stocks=["005930", "000660"])
# Should not crash
assert isinstance(candidates, list) assert isinstance(candidates, list)
assert len(candidates) >= 1
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_scan_returns_top_n_only( async def test_scan_returns_top_n_only(
self, scanner: SmartVolatilityScanner, mock_broker: MagicMock self, scanner: SmartVolatilityScanner, mock_broker: MagicMock
) -> None: ) -> None:
"""Test that scan returns at most top_n candidates.""" """Test that scan returns at most top_n candidates."""
# Return many stocks fluctuation_rows = [
mock_broker.fetch_market_rankings.return_value = [
{ {
"stock_code": f"00{i}000", "stock_code": f"00{i}000",
"name": f"Stock{i}", "name": f"Stock{i}",
@@ -235,62 +188,17 @@ class TestSmartVolatilityScanner:
} }
for i in range(1, 10) for i in range(1, 10)
] ]
mock_broker.fetch_market_rankings.side_effect = [fluctuation_rows, fluctuation_rows]
# All oversold with high volume mock_broker.get_daily_prices.return_value = [
def make_prices(code: str) -> list[dict]: {"open": 1, "high": 105, "low": 95, "close": 100, "volume": 1000000},
prices = [] {"open": 1, "high": 105, "low": 95, "close": 100, "volume": 900000},
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() candidates = await scanner.scan()
# Should respect top_n limit (3) # Should respect top_n limit (3)
assert len(candidates) <= scanner.top_n 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 @pytest.mark.asyncio
async def test_get_stock_codes( async def test_get_stock_codes(
self, scanner: SmartVolatilityScanner self, scanner: SmartVolatilityScanner
@@ -323,6 +231,124 @@ class TestSmartVolatilityScanner:
assert codes == ["005930", "035420"] 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: class TestRSICalculation:
"""Test RSI calculation in VolatilityAnalyzer.""" """Test RSI calculation in VolatilityAnalyzer."""