shuim/auto-stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve: win-rate optimization for short_term_strategy and kosdaq_scanner

Browse Source
This commit is contained in:
YoungWooSeo
2026-02-22 17:39:36 +09:00
parent 9f63e4a24e
commit 0fe933c480
2 changed files with 187 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
backend/app/services/kosdaq_scanner.py
View File

@@ -1,11 +1,16 @@
"""
코스닥 전종목 스캐너 (CLAUDE.md 섹션 8.1)
코스닥 전종목 스캐너 (CLAUDE.md 섹션 8.1 — 승률 최적화)
매일 장 마감 후 코스닥 전종목(~1,600종목)을 스캔하여
RSI + 볼린저 밴드 + 거래량 복합 점수로 상위 20종목을 선별한다.
[승률 개선 항목]
1. RSI 방향 필터: RSI가 바닥에서 상승 전환 중인 종목만 선별
2. 최소 점수 임계값 0.15 적용 (저품질 신호 제외)
3. 승률 필드(rsi_turning) ScanResult에 노출
"""
import logging
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import List, Optional
logger = logging.getLogger(__name__)
@@ -19,6 +24,12 @@ LIQUIDITY_PERCENTILE = 0.80
# 스캔 결과 상위 N개
TOP_N = 20
# 승률 개선: 최소 점수 임계값 (저품질 신호 필터)
MIN_SCORE = 0.15
# RSI 방향 확인용 과거 봉 수
RSI_DIRECTION_LOOKBACK = 3
@dataclass
class ScanResult:
@@ -28,6 +39,7 @@ class ScanResult:
pct_vs_bb_lower: float # (close - bb_lower) / bb_lower * 100
volume_ratio: float # current_volume / volume_ma20
close: float
rsi_turning: bool = False # RSI 반등 신호 여부 (승률 지표)
reason: str = ""
@@ -44,17 +56,18 @@ class TickerData:
class KosdaqScanner:
"""코스닥 전종목 스캐너
"""코스닥 전종목 스캐너 (승률 최적화)
매일 장 마감 후 scan_all()을 호출하면 조건을 충족하는 상위 20종목을
점수 내림차순으로 반환한다.
스코어링 공식 (CLAUDE.md 섹션 8.1):
스코어링 공식:
score = rsi_score × 0.5 + bb_score × 0.3 + volume_score × 0.2
→ score > 0.15 이상인 종목만 선별 (MIN_SCORE 필터)
- 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) # 거래량 급증 ↑
승률 햵심 필터 (rsi_turning=True):
RSI가 바닥에서 반등 중인 종목은 ScanResult.rsi_turning=True로 표시되며,
포트폴리오 매니저 쪽에서 제일 먼저 진입 처리하는 것을 권장한다.
"""
def __init__(self, min_price: int = MIN_PRICE_KRW, top_n: int = TOP_N):
@@ -68,11 +81,8 @@ class KosdaqScanner:
def scan_all(self, tickers_data: List[TickerData]) -> List[ScanResult]:
"""코스닥 전종목 스캔 → 상위 N개 ScanResult 반환
Args:
tickers_data: 외부(KIS API 등)에서 수집한 종목 데이터 리스트
Returns:
점수 내림차순 정렬된 상위 TOP_N개 ScanResult
rsi_turning=True 종목은 리스트 앞쪽으로 정렬하여
포트폴리오 매니저가 우선 진입할 수 있도록 한다.
"""
liquidity_threshold = self._calc_liquidity_threshold(tickers_data)
@@ -86,7 +96,8 @@ class KosdaqScanner:
if result is not None:
candidates.append(result)
candidates.sort(key=lambda x: x.score, reverse=True)
# rsi_turning=True 종목 우선, 동점이면 점수 내림차순
candidates.sort(key=lambda x: (not x.rsi_turning, -x.score))
return candidates[: self.top_n]
# ------------------------------------------------------------------
@@ -98,7 +109,7 @@ class KosdaqScanner:
return close >= self.min_price
def _liquidity_filter(self, turnover: float, threshold: float) -> bool:
"""거래대금 상위 20% 이내 필터 (하위 80% 제외)"""
"""거래대금 상위 20% 이내 필터"""
return turnover >= threshold
# ------------------------------------------------------------------
@@ -108,17 +119,19 @@ class KosdaqScanner:
def _score(self, td: TickerData) -> Optional[ScanResult]:
"""RSI + 볼린저 밴드 + 거래량 복합 점수 계산
Returns:
ScanResult if 최소 조건 충족, else None
승률 개선:
- MIN_SCORE(0.15) 이하 종목 제외
- RSI 반등 여부(rsi_turning) 노출
"""
from .mean_reversion_strategy import calculate_rsi, calculate_bollinger
closes = td.closes
if len(closes) < 25:
min_len = 25 + RSI_DIRECTION_LOOKBACK
if len(closes) < min_len:
return None
rsi14 = calculate_rsi(closes, 14)
bb = calculate_bollinger(closes, 20)
bb = calculate_bollinger(closes, 20)
if rsi14 is None or bb is None:
return None
@@ -133,15 +146,19 @@ class KosdaqScanner:
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)
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
total_score = rsi_score * 0.5 + bb_score * 0.3 + volume_score * 0.2
if total_score <= 0:
# 승률 필터: 최소 점수 미달 제외
if total_score <= MIN_SCORE:
return None
# RSI 반등 방향 확인 (승률 지표)
rsi_prev = calculate_rsi(closes[:-RSI_DIRECTION_LOOKBACK], 14)
rsi_turning = rsi_prev is not None and rsi14 > rsi_prev
reason_parts = []
if rsi14 <= 25:
reason_parts.append(f"RSI({rsi14:.1f})≤25")
@@ -149,6 +166,8 @@ class KosdaqScanner:
reason_parts.append(f"BB하단({pct_vs_lower:.1f}%)")
if volume_ratio >= 1.5:
reason_parts.append(f"거래량({volume_ratio:.1f}x)")
if rsi_turning:
reason_parts.append("RSI반등↑")
return ScanResult(
ticker=td.ticker,
@@ -157,6 +176,7 @@ class KosdaqScanner:
pct_vs_bb_lower=round(pct_vs_lower, 2),
volume_ratio=round(volume_ratio, 2),
close=td.close,
rsi_turning=rsi_turning,
reason=" + ".join(reason_parts) if reason_parts else "복합점수",
)
@@ -166,7 +186,7 @@ class KosdaqScanner:
@staticmethod
def _calc_liquidity_threshold(tickers_data: List[TickerData]) -> float:
"""거래대금 상위 20% 임계값 계산 (80번째 백분위수)"""
"""거래대금 상위 20% 임계값 (80번째 백분위수)"""
turnovers = sorted([td.turnover for td in tickers_data if td.turnover > 0])
if not turnovers:
return 0.0

199
backend/app/services/short_term_strategy.py
View File

@@ -1,13 +1,23 @@
"""
15분봉 단기 전략 (CLAUDE.md 섹션 8.2)
15분봉 단기 전략 (CLAUDE.md 섹션 8.2 — 승률 최적화)
전체 자본의 최대 25% 이내에서 운용.
일봉 RSI ≤ 40 관심 종목에 한해 15분봉 신호를 감지한다.
[승률 개선 항목]
1. 가격 하한 필터 → ATR 비율 필터 (ATR/close ≥ 0.8%)
— 거래비용을 커버할 수 있는 변동성 종목만 선별
2. 거래량 배수 1.3 → 1.5 (허위 신호 감소)
3. RSI 방향 확인 — RSI가 바닥에서 반등 중인 경우만 진입
4. 반전 캔들 확인 — 양봉 or 긴 하단 꼬리 (optional)
5. 손절 조건 추가 — entry - 1.0×ATR (CRITICAL: 기존 코드에 누락되어 있던 것)
6. 장 초반 30분 / 마감 15분 진입 금지 (노이즈 구간)
⚠️ 슬리피지 주의: 수수료 0.3% + 슬리피지 0.2% 포함 백테스트 필수
"""
import logging
from dataclasses import dataclass
from dataclasses import dataclass, field
from datetime import time as dtime
from enum import Enum
from typing import List, Optional
@@ -15,14 +25,22 @@ from .mean_reversion_strategy import calculate_rsi, calculate_bollinger
logger = logging.getLogger(__name__)
# 자본 비율 상한 (CLAUDE.md 섹션 8.2 / 8.3)
# 자본 비율 상한
MAX_CAPITAL_RATIO = 0.25
# 비용 가정 — 백테스트 시 왕복으로 적용
COMMISSION_RATE = 0.003 # 수수료 0.3%
SLIPPAGE_RATE = 0.002 # 슬리피지 0.2%
COMMISSION_RATE = 0.003
SLIPPAGE_RATE = 0.002
TOTAL_COST_RATE = COMMISSION_RATE + SLIPPAGE_RATE
# 승률 개선 파라미터
MIN_ATR_RATIO = 0.008 # ATR/close ≥ 0.8% : 거래비용 커버 보장
RSI_DIRECTION_LOOKBACK = 3 # RSI 반등 확인용 과거 봉 수
# 장 운영 시간 필터
_SKIP_OPEN_UNTIL = dtime(9, 30) # 장 초반 30분 제외 (개장 직후 변동성 과다)
_SKIP_CLOSE_FROM = dtime(15, 15) # 마감 15분 전 제외 (종가 청산 압력)
class TimeFrame(str, Enum):
M15 = "15m"
@@ -36,6 +54,7 @@ class EntrySignal:
reason: str
rsi: float
close: float
atr_ratio: float = 0.0 # ATR/close — 수익 가능성 지표
@dataclass
@@ -46,29 +65,32 @@ class ExitSignal:
class ShortTermStrategy:
"""15분봉 단기 평균회귀 전략
"""
15분봉 단기 평균회귀 전략 (승률 우선 최적화)
진입 조건 (모두 충족 필요):
1. 일봉 RSI(14) ≤ 40 — 관심 종목 사전 필터
2. 15분봉 RSI(14) ≤ 30
3. 15분봉 종가 ≤ BB 하단 × 1.01
4. 15분봉 종가 > MA5 — Fake-out 방지
5. 15분봉 거래량 > 거래량 MA20 × 1.3
1. 일봉 RSI(14) ≤ 40 — 관심 종목 사전 필터
2. 15분봉 RSI(14) ≤ 30 — 과매도
3. 15분봉 RSI 반등 중 — RSI_now > RSI_3봉전 (바닥 확인)
4. 15분봉 종가 ≤ BB 하단 × 1.01
5. 15분봉 종가 > MA5 — Fake-out 방지
6. 15분봉 거래량 > MA20 × 1.5 — 1.3에서 강화
7. ATR/close ≥ 0.8% — 거래비용 커버 가능한 변동성
8. 반전 캔들 (optional) — 양봉 or 핵머 캔들
9. 시간 필터 (optional) — 09:30~15:15만 허용
청산 조건 (둘 중 하나 충족):
- 15분봉 RSI(14) ≥ 55
- 수익률 ≥ 1.5 × ATR_15m_pct
자본 제한:
- 이 전략 배분 총액이 전체 자본 × 25% 초과 시 신규 진입 차단
- MDD -10% 돌파 시 즉시 중단 (호출 측에서 판단)
청산 조건 (우선순위 순):
1. 손절: close ≤ entry - 1.0×ATR [승률 핵심 — 기존 코드에 누락]
2. RSI(14) ≥ 55 회복
3. 수익률 ≥ 1.5 × ATR_15m_pct
"""
TIMEFRAME = TimeFrame.M15
DAILY_RSI_THRESHOLD = 40.0 # 일봉 사전 필터
M15_RSI_ENTRY = 30.0 # 15분봉 진입 RSI
M15_RSI_EXIT = 55.0 # 15분봉 청산 RSI
VOLUME_MULTIPLIER = 1.3 # 거래량 배수
DAILY_RSI_THRESHOLD = 40.0
M15_RSI_ENTRY = 30.0
M15_RSI_EXIT = 55.0
VOLUME_MULTIPLIER = 1.5 # 1.3 → 1.5 강화
MIN_ATR_RATIO = MIN_ATR_RATIO
def __init__(self, max_capital_ratio: float = MAX_CAPITAL_RATIO):
self.max_capital_ratio = max_capital_ratio
@@ -84,34 +106,55 @@ class ShortTermStrategy:
m15_closes: List[float],
m15_volumes: List[int],
current_volume: int,
atr_15m: Optional[float] = None,
m15_opens: Optional[List[float]] = None,
m15_lows: Optional[List[float]] = None,
trade_time: Optional[dtime] = None,
) -> Optional[EntrySignal]:
"""15분봉 진입 신호 감지
"""
15분봉 진입 신호 감지
Args:
ticker: 종목코드
daily_rsi: 일봉 RSI(14)
m15_closes: 15분봉 종가 리스트 (최소 25개)
m15_closes: 15분봉 종가 리스트 (최소 28개)
m15_volumes: 15분봉 거래량 리스트 (최소 20개)
current_volume: 현재 봉 거래량
atr_15m: 15분봉 ATR — None이면 ATR 비율 필터 스킵
m15_opens: 15분봉 시가 — None이면 반전 캔들 스킵
m15_lows: 15분봉 저가 — None이면 반전 캔들 스킵
trade_time: 현재 시각 — None이면 시간 필터 스킵
Returns:
EntrySignal if 모든 조건 충족, else None
EntrySignal if 조건 충족, else None
"""
# 조건 1: 일봉 사전 필터
if daily_rsi > self.DAILY_RSI_THRESHOLD:
return None
if len(m15_closes) < 25 or len(m15_volumes) < 20:
# 조건 9: 시간 필터 (제공된 경우)
if trade_time is not None:
if not (_SKIP_OPEN_UNTIL <= trade_time < _SKIP_CLOSE_FROM):
return None
min_len = 25 + RSI_DIRECTION_LOOKBACK
if len(m15_closes) < min_len or len(m15_volumes) < 20:
return None
close = m15_closes[-1]
# 조건 2: 15분봉 RSI
# 조건 2: 15분봉 RSI 과매도
rsi_15m = calculate_rsi(m15_closes, 14)
if rsi_15m is None or rsi_15m > self.M15_RSI_ENTRY:
return None
# 조건 3: BB 하단 근처
# 조건 3: RSI 반등 방향 확인 — 바닥에서 회복 중인지 확인 (승률 햵심)
rsi_prev = calculate_rsi(m15_closes[:-RSI_DIRECTION_LOOKBACK], 14)
if rsi_prev is not None and rsi_15m <= rsi_prev:
# RSI가 여전히 하락 중 → 변곡점 전이므로 진입 금지
return None
# 조건 4: BB 하단 근처
bb = calculate_bollinger(m15_closes, 20)
if bb is None:
return None
@@ -119,25 +162,40 @@ class ShortTermStrategy:
if close > bb_lower * 1.01:
return None
# 조건 4: MA5 상향 (Fake-out 방지)
# 조건 5: MA5 상향 (Fake-out 방지)
ma5 = sum(m15_closes[-5:]) / 5
if close <= ma5:
return None
# 조건 5: 거래량 급증
# 조건 6: 거래량 급증 (1.5배 강화)
vol_ma20 = sum(m15_volumes[-20:]) / 20
if vol_ma20 == 0 or current_volume <= vol_ma20 * self.VOLUME_MULTIPLIER:
return None
# 조건 7: ATR 비율 필터 — 5,000원 필터 대체
atr_ratio = 0.0
if atr_15m is not None:
atr_ratio = atr_15m / close if close > 0 else 0.0
if atr_ratio < self.MIN_ATR_RATIO:
# 변동성 부족 → 1.5×ATR 익절 목표가 왕복 비용에 도달안함
return None
# 조건 8: 반전 캔들 (데이터 제공 시)
if m15_opens is not None and m15_lows is not None:
if not _is_reversal_candle(m15_opens, m15_lows, m15_closes):
return None
return EntrySignal(
ticker=ticker,
timeframe=self.TIMEFRAME,
reason=(
f"15분봉 RSI({rsi_15m:.1f})≤30 + BB하단 + MA5돌파 "
f"+ 거래량({current_volume / vol_ma20:.1f}x)"
f"RSI({rsi_15m:.1f})≤30↑ + BB하단 + MA5돌파"
f" + 거래량({current_volume / vol_ma20:.1f}x)"
+ (f" + ATR비율({atr_ratio * 100:.1f}%)" if atr_15m else "")
),
rsi=round(rsi_15m, 2),
close=close,
atr_ratio=round(atr_ratio, 4),
)
def exit_signal(
@@ -147,16 +205,13 @@ class ShortTermStrategy:
entry_price: float,
atr_15m: Optional[float] = None,
) -> Optional[ExitSignal]:
"""15분봉 청산 신호
"""
15분봉 청산 신호
Args:
ticker: 종목코드
m15_closes: 15분봉 종가 리스트
entry_price: 진입 가격
atr_15m: 15분봉 ATR (None 이면 ATR 조건 스킵)
Returns:
ExitSignal if 청산 조건 충족, else None
청산 우선순위:
1. 손절 : close ≤ max(entry - 1.0×ATR, entry × 0.98) [핵심 추가]
2. RSI ≥ 55 회복
3. 수익률 ≥ 1.5 × ATR_15m_pct
"""
if len(m15_closes) < 15:
return None
@@ -164,7 +219,24 @@ class ShortTermStrategy:
close = m15_closes[-1]
rsi_15m = calculate_rsi(m15_closes, 14)
# 청산 조건 1: RSI ≥ 55 회복
# 청산 우선순위 1: 손절 [기존 코드에서 누락된 핸심 로직]
if entry_price > 0:
if atr_15m:
atr_stop = entry_price - 1.0 * atr_15m
hard_stop = entry_price * 0.98 # -2% 하드 스탑
stop_price = max(atr_stop, hard_stop) # 더 타이트한 쪽
else:
stop_price = entry_price * 0.98 # ATR 없으면 -2% 고정
if close <= stop_price:
loss_pct = (close - entry_price) / entry_price * 100
return ExitSignal(
action="SELL_ALL",
reason=f"손절 close({close:.0f}) ≤ stop({stop_price:.0f}) | {loss_pct:.1f}%",
sell_pct=1.0,
)
# 청산 우선순위 2: RSI ≥ 55 회복
if rsi_15m is not None and rsi_15m >= self.M15_RSI_EXIT:
return ExitSignal(
action="SELL_ALL",
@@ -172,7 +244,7 @@ class ShortTermStrategy:
sell_pct=1.0,
)
# 청산 조건 2: 수익률 ≥ 1.5 × ATR_15m_pct
# 청산 우선순위 3: 수익률 ≥ 1.5 × ATR_15m_pct
if atr_15m and entry_price > 0:
profit_pct = (close - entry_price) / entry_price
atr_pct = atr_15m / entry_price
@@ -180,8 +252,8 @@ class ShortTermStrategy:
return ExitSignal(
action="SELL_ALL",
reason=(
f"익절 수익률({profit_pct * 100:.1f}%) "
f"≥ 1.5×ATR({atr_pct * 100:.1f}%)"
f"익절 수익률({profit_pct * 100:.1f}%)"
f" ≥ 1.5×ATR({atr_pct * 100:.1f}%)"
),
sell_pct=1.0,
)
@@ -191,18 +263,33 @@ class ShortTermStrategy:
def is_capital_available(
self, total_assets: float, current_short_term_allocation: float
) -> bool:
"""단기 전략 자본 한도 초과 여부 확인
CLAUDE.md 섹션 8.3: 단기 전략에 최대 25% 배분.
MDD -10% 돌파 시에는 호출 측에서 별도로 차단해야 한다.
"""
limit = total_assets * self.max_capital_ratio
return current_short_term_allocation < limit
"""CLAUDE.md 8.3: 단기 전략 자본 한도 충족 여부"""
return current_short_term_allocation < total_assets * self.max_capital_ratio
@staticmethod
def net_profit_pct(gross_profit_pct: float) -> float:
"""수수료 + 슬리피지 차감 후 순수익률 (왕복 기준)
백테스트 결과 검증 시 이 함수로 실질 수익률을 계산하라.
"""
"""수수료 + 슬리피지 차감 후 순수익률 (왕복 기준)"""
return gross_profit_pct - TOTAL_COST_RATE * 2
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _is_reversal_candle(
opens: List[float],
lows: List[float],
closes: List[float],
) -> bool:
"""반전 캔들 확인 — 양봉 또는 핸머(카드) 여부
조건 (둘 중 하나):
- 양봉: close > open
- 해머: 하단 꼬리 > 몸통의 2배 (핀바 / 행장형)
"""
if not opens or not lows or not closes:
return False
o, l, c = opens[-1], lows[-1], closes[-1]
body = abs(c - o)
lower_shadow = min(o, c) - l
return (c > o) or (lower_shadow > max(body * 2.0, 1e-6))