[6] 전략 끼워넣기 – BaseStrategy 가상함수와 샘플 전략 구현

# Python — 추상 클래스
class BaseStrategy(ABC):
    @abstractmethod
    def generate_signal(self, stock_code, data, timeframe='daily'):
        ...  # 순수 가상함수 — 반드시 구현해야 함

    def on_init(self, broker, data_provider, executor):
        ...  # 기본 구현 있음 — 필요하면 오버라이드

    def on_market_open(self):
        ...  # 기본 구현 있음

    def on_order_filled(self, order):
        ...  # 기본 구현 있음

    def on_market_close(self):
        ...  # 기본 구현 있음

from strategies.base import BaseStrategy, Signal, SignalType

class SampleStrategy(BaseStrategy):
    name = "SampleStrategy"
    version = "1.0.0"

def on_init(self, broker, data_provider, executor) -> bool:
    # 프레임워크가 넘겨주는 세 가지를 저장
    self._broker = broker
    self._data_provider = data_provider
    self._executor = executor

    # config.yaml에서 파라미터 로드
    params = self.config.get("parameters", {})
    self._ma_short = params.get("ma_short_period", 5)
    self._ma_long = params.get("ma_long_period", 20)
    self._rsi_period = params.get("rsi_period", 14)
    self._rsi_oversold = params.get("rsi_oversold", 30)
    self._rsi_overbought = params.get("rsi_overbought", 70)
    self._volume_multiplier = params.get("volume_multiplier", 1.5)
    self._min_buy_signals = params.get("min_buy_signals", 2)

    risk = self.config.get("risk_management", {})
    self._stop_loss_pct = risk.get("stop_loss_pct", 0.05)
    self._take_profit_pct = risk.get("take_profit_pct", 0.10)
    self._max_daily_trades = risk.get("max_daily_trades", 5)

    # 상태 변수 초기화
    self.positions = {}       # 보유 종목 추적
    self.daily_trades = 0     # 오늘 거래 횟수

    self._is_initialized = True
    return True

def generate_signal(self, stock_code, data, timeframe='daily'):
    # 데이터 부족하면 패스
    if data is None or len(data) < self._ma_long + 2:
        return None

    # 하루 거래 한도 초과
    if self.daily_trades >= self._max_daily_trades:
        return None

    close = data["close"]
    volume = data["volume"]

    # 지표 계산
    sma_short = close.rolling(self._ma_short).mean()
    sma_long = close.rolling(self._ma_long).mean()
    rsi = self._calculate_rsi(close, self._rsi_period)
    avg_volume = volume.rolling(self._ma_long).mean()

    # ── 보유 중이면 매도 조건 확인 ──
    if stock_code in self.positions:
        reasons = []
        current_price = float(close.iloc[-1])
        entry_price = self.positions[stock_code]["entry_price"]
        pnl_pct = (current_price - entry_price) / entry_price

        if sma_short.iloc[-1] < sma_long.iloc[-1] and sma_short.iloc[-2] >= sma_long.iloc[-2]:
            reasons.append("데드크로스")
        if rsi.iloc[-1] > self._rsi_overbought:
            reasons.append("RSI 과매수")
        if pnl_pct >= self._take_profit_pct:
            reasons.append(f"익절 도달 ({pnl_pct:+.1%})")
        if pnl_pct <= -self._stop_loss_pct:
            reasons.append(f"손절 도달 ({pnl_pct:+.1%})")

        if reasons:
            return Signal(
                signal_type=SignalType.SELL,
                stock_code=stock_code,
                confidence=min(90.0, 50.0 + len(reasons) * 20),
                reasons=reasons,
            )
        return None

    # ── 미보유면 매수 조건 확인 ──
    reasons = []
    if sma_short.iloc[-1] > sma_long.iloc[-1] and sma_short.iloc[-2] <= sma_long.iloc[-2]:
        reasons.append(f"{self._ma_short}일선이 {self._ma_long}일선 골든크로스")
    if rsi.iloc[-2] < self._rsi_oversold and rsi.iloc[-1] >= self._rsi_oversold:
        reasons.append("RSI 과매도 탈출")
    if volume.iloc[-1] > avg_volume.iloc[-1] * self._volume_multiplier:
        reasons.append("거래량 급증")

    if len(reasons) >= self._min_buy_signals:
        current_price = float(close.iloc[-1])
        return Signal(
            signal_type=SignalType.BUY,
            stock_code=stock_code,
            confidence=min(95.0, 50.0 + len(reasons) * 15),
            target_price=current_price * (1 + self._take_profit_pct),
            stop_loss=current_price * (1 - self._stop_loss_pct),
            reasons=reasons,
        )

    return None

def on_order_filled(self, order) -> None:
    self.daily_trades += 1

    if order.is_buy:
        self.positions[order.stock_code] = {
            "quantity": order.quantity,
            "entry_price": order.price,
            "entry_time": order.filled_at,
        }
    elif order.is_sell:
        if order.stock_code in self.positions:
            entry = self.positions[order.stock_code]["entry_price"]
            pnl_pct = (order.price - entry) / entry * 100
            self.logger.info(f"{order.stock_code} 매도 완료: {pnl_pct:+.1f}%")
            del self.positions[order.stock_code]

def on_market_open(self) -> None:
    self.daily_trades = 0
    self.daily_profit = 0.0
    self.logger.info("장 시작 — 일일 카운터 초기화")

def on_market_close(self) -> None:
    self.logger.info(
        f"장 마감 — 거래 {self.daily_trades}건, "
        f"잔여 포지션 {len(self.positions)}개"
    )

# main.py — 이 코드는 수정할 필요 없음
strategies = StrategyLoader.discover_strategies()
# → {'sample': Path('strategies/sample')} 자동 발견

strategy = StrategyLoader.load_strategy('sample')
# → config.yaml 로드 → SampleStrategy 클래스 찾기 → 인스턴스 생성

strategy.on_init(broker, data_provider, executor)
# → 프레임워크가 모듈 주입

# 매매 루프 시작
# → generate_signal()이 3초마다 호출됨

# strategies/my_momentum/strategy.py
class MyMomentumStrategy(BaseStrategy):
    name = "MyMomentum"
    version = "1.0.0"

    def generate_signal(self, stock_code, data, timeframe='daily'):
        # 5일 연속 양봉이면 매수
        close = data["close"]
        consecutive_up = all(
            close.iloc[-i] > close.iloc[-i-1] for i in range(1, 6)
        )
        if consecutive_up:
            return Signal(
                signal_type=SignalType.BUY,
                stock_code=stock_code,
                confidence=80.0,
                reasons=["5일 연속 상승"],
            )
        return None