From a322745cf304a44c3354d8bb6f427e1d44224374 Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 10:55:19 +0900 Subject: [PATCH] refactor(phase2): Add Scanner module and strategy README --- backend/app/strategy/README.md | 288 ++++++++++++++++++++++++++++++++ backend/app/strategy/scanner.py | 119 +++++++++++++ 2 files changed, 407 insertions(+) create mode 100644 backend/app/strategy/README.md create mode 100644 backend/app/strategy/scanner.py diff --git a/backend/app/strategy/README.md b/backend/app/strategy/README.md new file mode 100644 index 0000000..30cffb6 --- /dev/null +++ b/backend/app/strategy/README.md @@ -0,0 +1,288 @@ +# Strategy Module + +매매 전략을 관리하고 실행하는 핵심 모듈 + +## 모듈 구조 + +``` +strategy/ +├── __init__.py # 모듈 내보내기 +├── base_strategy.py # BaseStrategy 추상 클래스 +├── signal.py # Signal 데이터 클래스 +├── strategy_manager.py # 전략 관리자 (등록/실행) +├── indicators.py # 보조지표 (RSI, BB, MACD 등) +├── scanner.py # 종목 스캔 +├── strategies/ # 구체적 전략 구현 +│ ├── __init__.py +│ └── day_trading.py # 데이트레이딩 전략 +└── README.md +``` + +## 사용 방법 + +### 1. 전략 구현 + +`BaseStrategy`를 상속받아 `analyze()` 메서드 구현: + +```python +from app.strategy import BaseStrategy, Signal, SignalType +from datetime import datetime +import pytz + +class MyStrategy(BaseStrategy): + def __init__(self): + super().__init__(name="MyStrategy", config={"param1": 10}) + + async def analyze(self, stock_code, market_data, portfolio_data=None): + current_price = market_data.get("current_price") + + # 매수 조건 체크 + if self._should_buy(market_data): + return Signal( + stock_code=stock_code, + signal_type=SignalType.BUY, + strategy_name=self.name, + timestamp=datetime.now(pytz.timezone("Asia/Seoul")), + price=current_price, + confidence=0.8, + reason="매수 조건 충족" + ) + + return None + + def _should_buy(self, market_data): + # 실제 로직 구현 + return True +``` + +### 2. 전략 등록 및 실행 + +```python +from app.strategy import strategy_manager +from app.strategy.strategies import DayTradingStrategy + +# 전략 인스턴스 생성 +day_trading = DayTradingStrategy( + name="DayTrading_ORB", + config={ + "max_positions": 3, + "stop_loss_pct": 0.025 + } +) + +# 전략 등록 (priority 낮을수록 먼저 실행) +strategy_manager.register_strategy(day_trading, priority=10) + +# 단일 전략 실행 +signal = await strategy_manager.run_strategy( + "DayTrading_ORB", + stock_code="005930", + market_data={ + "current_price": 75000, + "minute_chart": [...], + } +) + +if signal: + print(f"신호 발생: {signal}") + +# 모든 전략 동시 실행 +signals = await strategy_manager.run_all_strategies( + stock_code="005930", + market_data={...}, + parallel=True # 병렬 실행 +) + +# 신호 충돌 해결 +final_signal = strategy_manager.resolve_conflicts( + signals, + method="highest_confidence" # or "priority", "majority_vote" +) +``` + +### 3. Signal 데이터 구조 + +```python +from app.strategy import Signal, SignalType +from datetime import datetime +import pytz + +signal = Signal( + stock_code="005930", + signal_type=SignalType.BUY, + strategy_name="MyStrategy", + timestamp=datetime.now(pytz.timezone("Asia/Seoul")), + + # 주문 정보 + quantity=10, + price=75000, # None = 시장가 + confidence=0.85, + + # 손익 관리 + stop_loss_price=73000, + take_profit_price=78000, + trailing_stop_pct=0.02, # 2% 트레일링 + + # 메타데이터 + reason="RSI 과매도 반등", + metadata={"rsi": 25, "bb_lower": 74000} +) + +# JSON 변환 +data = signal.to_dict() +``` + +### 4. 보조지표 사용 + +```python +from app.strategy.indicators import rsi, bollinger_bands, sma, macd, atr + +# 종가 리스트 +closes = [75000, 75500, 76000, 74500, 75000, ...] + +# RSI +rsi_value = rsi(closes, period=14) +if rsi_value and rsi_value < 30: + print("과매도 구간") + +# 볼린저 밴드 +bb = bollinger_bands(closes, period=20, std_mult=2.0) +if bb: + upper, middle, lower = bb + if closes[-1] < lower: + print("하단 돌파") + +# SMA +sma_20 = sma(closes, 20) +sma_60 = sma(closes, 60) +if sma_20 and sma_60 and sma_20 > sma_60: + print("골든 크로스 발생") + +# ATR +highs = [76000, 76500, ...] +lows = [74500, 75000, ...] +atr_value = atr(highs, lows, closes, period=14) +``` + +### 5. 종목 스캔 + +```python +from app.strategy.scanner import Scanner, filter_price_range, filter_volume_min + +# 스캀너 생성 +scanner = Scanner() + +# 필터 추가 +scanner.add_filter(filter_price_range(10000, 100000)) # 1만~10만원 +scanner.add_filter(filter_volume_min(100000)) # 최소 거래량 10만주 + +# 커스텀 필터 +def custom_filter(data): + # 등락률 5% 이상 + return data.get("change_rate", 0) >= 5.0 + +scanner.add_filter(custom_filter) + +# 스캔 실행 +kospi_codes = ["005930", "000660", "035420", ...] # KOSPI 종목 리스트 +matched = await scanner.scan(kospi_codes, max_concurrent=10) + +for stock in matched: + print(f"{stock['stock_name']} ({stock['stock_code']}): " + f"{stock['current_price']}원, +{stock['change_rate']}%") +``` + +## 주요 기능 + +### BaseStrategy +- ✅ 전략 추상화 (템플릿 메서드 패턴) +- ✅ `analyze()` 구현 강제 +- ✅ `on_signal_executed()` 콜백 +- ✅ `on_error()` 에러 핸들링 +- ✅ 설정 관리 (config) +- ✅ 활성화/비활성화 + +### StrategyManager +- ✅ 전략 등록/삭제 +- ✅ 우선순위 관리 +- ✅ 단일/복수 전략 실행 +- ✅ 병렬 실행 지원 +- ✅ 신호 충돌 해결 + - `highest_confidence`: 가장 높은 신뢰도 + - `priority`: 전략 우선순위 + - `majority_vote`: 다수결 +- ✅ 신호 이력 관리 +- ✅ 통계 정보 + +### Signal +- ✅ 표준화된 신호 데이터 구조 +- ✅ 6가지 신호 타입 (BUY/SELL/HOLD/CLOSE/STOP_LOSS/TAKE_PROFIT) +- ✅ 손익 관리 필드 +- ✅ 신뢰도 평가 +- ✅ JSON 직렬화 + +### Indicators +- ✅ SMA (Simple Moving Average) +- ✅ EMA (Exponential Moving Average) +- ✅ RSI (Relative Strength Index) +- ✅ Bollinger Bands +- ✅ MACD +- ✅ ATR (Average True Range) +- ✅ VWAP (Volume Weighted Average Price) + +### Scanner +- ✅ 조건 기반 종목 필터링 +- ✅ 동시 다중 종목 스캔 +- ✅ 사전 정의 필터 + - 가격 범위 + - 거래량 최소값 + - 등락률 범위 + - 시가총액 최소값 +- ✅ 커스텀 필터 추가 가능 + +## 기존 코드 마이그레이션 + +기존 `services/` 폴더의 전략 파일들을 새 구조로 변환: + +| 기존 파일 | 새 위치 | +|----------|----------| +| `day_trading_strategy.py` | `strategy/strategies/day_trading.py` | +| `swing_trading_strategy.py` | `strategy/strategies/swing_trading.py` | +| `momentum_strategy.py` | `strategy/strategies/momentum.py` | + +변환 패턴: +```python +# 기존 +def check_signal(stock_code, market_data): + if condition: + return DayTradeSignal(...) + return None + +# 새 구조 +class DayTradingStrategy(BaseStrategy): + async def analyze(self, stock_code, market_data, portfolio_data=None): + if condition: + return Signal(...) + return None +``` + +## Phase 2 완료 체크리스트 + +- [x] `strategy/base_strategy.py` 생성 +- [x] `strategy/signal.py` 생성 +- [x] `strategy/strategy_manager.py` 생성 +- [x] `strategy/indicators.py` 생성 +- [x] `strategy/scanner.py` 생성 +- [x] `strategy/strategies/day_trading.py` 예시 생성 +- [x] README 문서 작성 + +## 다음 단계 (Phase 3) + +- [ ] WebSocket 모듈 분리 (`realtime/`) +- [ ] 실시간 시세 수신 구조 개선 +- [ ] 장 스케줄 관리 + +--- + +**작성일**: 2026-03-06 +**작성자**: 서영우 diff --git a/backend/app/strategy/scanner.py b/backend/app/strategy/scanner.py new file mode 100644 index 0000000..25e1d7f --- /dev/null +++ b/backend/app/strategy/scanner.py @@ -0,0 +1,119 @@ +""" +Stock Scanner + +시장에서 특정 조건에 맞는 종목을 스캔하는 모듈 +""" +from typing import List, Dict, Any, Optional, Callable +import logging +import asyncio + +from app.trading import market_data_provider + +logger = logging.getLogger(__name__) + + +class Scanner: + """ + 종목 스캔 엔진 + + 특정 조건을 만족하는 종목을 찾는다. + """ + + def __init__(self): + self.filters: List[Callable] = [] + + def add_filter(self, filter_func: Callable[[Dict[str, Any]], bool]): + """ + 필터 추가 + + Args: + filter_func: market_data를 받아 True/False 반환하는 함수 + """ + self.filters.append(filter_func) + + async def scan( + self, + stock_codes: List[str], + max_concurrent: int = 5 + ) -> List[Dict[str, Any]]: + """ + 종목 목록을 스캔하여 조건에 맞는 종목 반환 + + Args: + stock_codes: 스캔할 종목코드 목록 + max_concurrent: 동시 처리 수 + + Returns: + 조건에 맞는 종목 데이터 목록 + """ + semaphore = asyncio.Semaphore(max_concurrent) + + async def scan_one(stock_code: str) -> Optional[Dict[str, Any]]: + async with semaphore: + try: + # 시세 데이터 수집 + price_data = await market_data_provider.get_current_price(stock_code) + + if not price_data or price_data.get("rt_cd") != "0": + return None + + output = price_data.get("output", {}) + market_data = { + "stock_code": stock_code, + "stock_name": output.get("hts_kor_isnm", ""), + "current_price": float(output.get("stck_prpr", 0)), + "change_rate": float(output.get("prdy_ctrt", 0)), + "volume": int(output.get("acml_vol", 0)), + "market_cap": int(output.get("hts_avls", 0)) * 100000000, # 억 → 원 + } + + # 필터 적용 + for filter_func in self.filters: + if not filter_func(market_data): + return None + + return market_data + + except Exception as e: + logger.warning(f"⚠️ 스캔 실패 {stock_code}: {e}") + return None + + tasks = [scan_one(code) for code in stock_codes] + results = await asyncio.gather(*tasks) + + matched = [r for r in results if r is not None] + logger.info(f"🔍 스캔 완료: {len(matched)}/{len(stock_codes)} 종목 매칭") + + return matched + + +# 사전 정의 필터 + +def filter_price_range(min_price: float, max_price: float) -> Callable: + """가격 범위 필터""" + def _filter(data: Dict[str, Any]) -> bool: + price = data.get("current_price", 0) + return min_price <= price <= max_price + return _filter + + +def filter_volume_min(min_volume: int) -> Callable: + """최소 거래량 필터""" + def _filter(data: Dict[str, Any]) -> bool: + return data.get("volume", 0) >= min_volume + return _filter + + +def filter_change_rate(min_rate: float, max_rate: float) -> Callable: + """등락률 범위 필터""" + def _filter(data: Dict[str, Any]) -> bool: + rate = data.get("change_rate", 0) + return min_rate <= rate <= max_rate + return _filter + + +def filter_market_cap_min(min_cap: int) -> Callable: + """최소 시가총액 필터 (원)""" + def _filter(data: Dict[str, Any]) -> bool: + return data.get("market_cap", 0) >= min_cap + return _filter