refactor(phase2): Add Scanner module and strategy README

2026-03-06 10:55:19 +09:00
parent 8920699eee
commit a322745cf3
2 changed files with 407 additions and 0 deletions
--- a/backend/app/strategy/README.md
+++ 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  
+**작성자**: 서영우
--- a/backend/app/strategy/scanner.py
+++ 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