auto-stock/backend/app/strategy/README.md

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() 메서드 구현:

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. 전략 등록 및 실행

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 데이터 구조

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. 보조지표 사용

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. 종목 스캔

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

변환 패턴:

# 기존
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 완료 체크리스트

• strategy/base_strategy.py 생성
• strategy/signal.py 생성
• strategy/strategy_manager.py 생성
• strategy/indicators.py 생성
• strategy/scanner.py 생성
• strategy/strategies/day_trading.py 예시 생성
• README 문서 작성

다음 단계 (Phase 3)

• WebSocket 모듈 분리 (realtime/)
• 실시간 시세 수신 구조 개선
• 장 스케줄 관리

---

작성일: 2026-03-06
작성자: 서영우