175 lines
5.8 KiB
Python
175 lines
5.8 KiB
Python
"""
|
||
코스닥 전종목 스캐너 (CLAUDE.md 섹션 8.1)
|
||
|
||
매일 장 마감 후 코스닥 전종목(~1,600종목)을 스캔하여
|
||
RSI + 볼린저 밴드 + 거래량 복합 점수로 상위 20종목을 선별한다.
|
||
"""
|
||
import logging
|
||
from dataclasses import dataclass
|
||
from typing import List, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 가격 필터 (CLAUDE.md 섹션 6.4)
|
||
MIN_PRICE_KRW = 5_000
|
||
|
||
# 거래대금 상위 20% 임계값 (하위 80% 제외)
|
||
LIQUIDITY_PERCENTILE = 0.80
|
||
|
||
# 스캔 결과 상위 N개
|
||
TOP_N = 20
|
||
|
||
|
||
@dataclass
|
||
class ScanResult:
|
||
ticker: str
|
||
score: float
|
||
rsi14: float
|
||
pct_vs_bb_lower: float # (close - bb_lower) / bb_lower * 100
|
||
volume_ratio: float # current_volume / volume_ma20
|
||
close: float
|
||
reason: str = ""
|
||
|
||
|
||
@dataclass
|
||
class TickerData:
|
||
"""스캐너에 필요한 종목 데이터 구조체"""
|
||
ticker: str
|
||
close: float
|
||
closes: List[float] # 최근 60일 이상 종가
|
||
volume: int
|
||
volumes: List[int] # 최근 20일 거래량
|
||
turnover: float # 당일 거래대금 (원)
|
||
turnovers: List[float] # 최근 20일 거래대금
|
||
|
||
|
||
class KosdaqScanner:
|
||
"""코스닥 전종목 스캐너
|
||
|
||
매일 장 마감 후 scan_all()을 호출하면 조건을 충족하는 상위 20종목을
|
||
점수 내림차순으로 반환한다.
|
||
|
||
스코어링 공식 (CLAUDE.md 섹션 8.1):
|
||
score = rsi_score × 0.5 + bb_score × 0.3 + volume_score × 0.2
|
||
|
||
- rsi_score = max(0, (25 - rsi14) / 25) # RSI 낮을수록 ↑
|
||
- bb_score = max(0, (0 - pct_vs_lower) / 3) # BB 하단 근접 ↑
|
||
- volume_score = min(volume / volume_ma20, 2.0) # 거래량 급증 ↑
|
||
"""
|
||
|
||
def __init__(self, min_price: int = MIN_PRICE_KRW, top_n: int = TOP_N):
|
||
self.min_price = min_price
|
||
self.top_n = top_n
|
||
|
||
# ------------------------------------------------------------------
|
||
# Public API
|
||
# ------------------------------------------------------------------
|
||
|
||
def scan_all(self, tickers_data: List[TickerData]) -> List[ScanResult]:
|
||
"""코스닥 전종목 스캔 → 상위 N개 ScanResult 반환
|
||
|
||
Args:
|
||
tickers_data: 외부(KIS API 등)에서 수집한 종목 데이터 리스트
|
||
|
||
Returns:
|
||
점수 내림차순 정렬된 상위 TOP_N개 ScanResult
|
||
"""
|
||
liquidity_threshold = self._calc_liquidity_threshold(tickers_data)
|
||
|
||
candidates: List[ScanResult] = []
|
||
for td in tickers_data:
|
||
if not self._price_filter(td.close):
|
||
continue
|
||
if not self._liquidity_filter(td.turnover, liquidity_threshold):
|
||
continue
|
||
result = self._score(td)
|
||
if result is not None:
|
||
candidates.append(result)
|
||
|
||
candidates.sort(key=lambda x: x.score, reverse=True)
|
||
return candidates[: self.top_n]
|
||
|
||
# ------------------------------------------------------------------
|
||
# Filters
|
||
# ------------------------------------------------------------------
|
||
|
||
def _price_filter(self, close: float) -> bool:
|
||
"""CLAUDE.md 섹션 6.4: 국내 5,000원 이하 매수 금지"""
|
||
return close >= self.min_price
|
||
|
||
def _liquidity_filter(self, turnover: float, threshold: float) -> bool:
|
||
"""거래대금 상위 20% 이내 필터 (하위 80% 제외)"""
|
||
return turnover >= threshold
|
||
|
||
# ------------------------------------------------------------------
|
||
# Scoring
|
||
# ------------------------------------------------------------------
|
||
|
||
def _score(self, td: TickerData) -> Optional[ScanResult]:
|
||
"""RSI + 볼린저 밴드 + 거래량 복합 점수 계산
|
||
|
||
Returns:
|
||
ScanResult if 최소 조건 충족, else None
|
||
"""
|
||
from .mean_reversion_strategy import calculate_rsi, calculate_bollinger
|
||
|
||
closes = td.closes
|
||
if len(closes) < 25:
|
||
return None
|
||
|
||
rsi14 = calculate_rsi(closes, 14)
|
||
bb = calculate_bollinger(closes, 20)
|
||
if rsi14 is None or bb is None:
|
||
return None
|
||
|
||
_, _, bb_lower = bb
|
||
if bb_lower == 0:
|
||
return None
|
||
|
||
volume_ma20 = sum(td.volumes[-20:]) / 20 if len(td.volumes) >= 20 else None
|
||
if volume_ma20 is None or volume_ma20 == 0:
|
||
return None
|
||
|
||
pct_vs_lower = (td.close - bb_lower) / bb_lower * 100
|
||
volume_ratio = td.volume / volume_ma20
|
||
|
||
rsi_score = max(0.0, (25.0 - rsi14) / 25.0)
|
||
bb_score = max(0.0, (0.0 - pct_vs_lower) / 3.0)
|
||
volume_score = min(volume_ratio, 2.0)
|
||
|
||
total_score = rsi_score * 0.5 + bb_score * 0.3 + volume_score * 0.2
|
||
|
||
if total_score <= 0:
|
||
return None
|
||
|
||
reason_parts = []
|
||
if rsi14 <= 25:
|
||
reason_parts.append(f"RSI({rsi14:.1f})≤25")
|
||
if pct_vs_lower <= 0:
|
||
reason_parts.append(f"BB하단({pct_vs_lower:.1f}%)")
|
||
if volume_ratio >= 1.5:
|
||
reason_parts.append(f"거래량({volume_ratio:.1f}x)")
|
||
|
||
return ScanResult(
|
||
ticker=td.ticker,
|
||
score=round(total_score, 4),
|
||
rsi14=round(rsi14, 2),
|
||
pct_vs_bb_lower=round(pct_vs_lower, 2),
|
||
volume_ratio=round(volume_ratio, 2),
|
||
close=td.close,
|
||
reason=" + ".join(reason_parts) if reason_parts else "복합점수",
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Helpers
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def _calc_liquidity_threshold(tickers_data: List[TickerData]) -> float:
|
||
"""거래대금 상위 20% 임계값 계산 (80번째 백분위수)"""
|
||
turnovers = sorted([td.turnover for td in tickers_data if td.turnover > 0])
|
||
if not turnovers:
|
||
return 0.0
|
||
idx = int(len(turnovers) * LIQUIDITY_PERCENTILE)
|
||
return turnovers[min(idx, len(turnovers) - 1)]
|