stockautomtion/app/engine/strategies/technical.py

from __future__ import annotations

import ta
import pandas as pd

from app.engine.strategies.base import BaseStrategy, Signal


class TechnicalStrategy(BaseStrategy):
    """기술적 분석 기반 전략 (MACD, RSI, 볼린저밴드)"""

    def evaluate(self, current_price: dict, price_history: list[dict], holding_qty: int) -> Signal:
        df = self._to_dataframe(price_history)
        if len(df) < 30:
            return Signal(action="hold", stock_code=self.stock_code)

        price = current_price.get("current_price", 0)
        indicators = self.params.get("indicators", ["rsi"])
        qty = self.params.get("qty", 1)

        buy_signals = []
        sell_signals = []

        if "rsi" in indicators:
            rsi_signal = self._evaluate_rsi(df, price, holding_qty, qty)
            if rsi_signal.action == "buy":
                buy_signals.append(rsi_signal)
            elif rsi_signal.action == "sell":
                sell_signals.append(rsi_signal)

        if "macd" in indicators:
            macd_signal = self._evaluate_macd(df, price, holding_qty, qty)
            if macd_signal.action == "buy":
                buy_signals.append(macd_signal)
            elif macd_signal.action == "sell":
                sell_signals.append(macd_signal)

        if "bollinger" in indicators:
            bb_signal = self._evaluate_bollinger(df, price, holding_qty, qty)
            if bb_signal.action == "buy":
                buy_signals.append(bb_signal)
            elif bb_signal.action == "sell":
                sell_signals.append(bb_signal)

        if buy_signals:
            best = max(buy_signals, key=lambda s: s.confidence)
            return best
        if sell_signals:
            best = max(sell_signals, key=lambda s: s.confidence)
            return best

        return Signal(action="hold", stock_code=self.stock_code)

    def _evaluate_rsi(
        self, df: pd.DataFrame, price: int, holding_qty: int, qty: int
    ) -> Signal:
        period = self.params.get("rsi_period", 14)
        oversold = self.params.get("rsi_oversold", 30)
        overbought = self.params.get("rsi_overbought", 70)

        rsi = ta.momentum.RSIIndicator(df["close"], window=period).rsi()
        current_rsi = rsi.iloc[-1] if not rsi.empty else 50

        if current_rsi <= oversold and holding_qty == 0:
            return Signal(
                action="buy",
                stock_code=self.stock_code,
                qty=qty,
                price=price,
                reason=f"RSI 과매도: {current_rsi:.1f} <= {oversold}",
                confidence=(oversold - current_rsi) / oversold if oversold > 0 else 0,
            )

        if current_rsi >= overbought and holding_qty > 0:
            return Signal(
                action="sell",
                stock_code=self.stock_code,
                qty=min(qty, holding_qty),
                price=price,
                reason=f"RSI 과매수: {current_rsi:.1f} >= {overbought}",
                confidence=(current_rsi - overbought) / (100 - overbought) if overbought < 100 else 0,
            )

        return Signal(action="hold", stock_code=self.stock_code)

    def _evaluate_macd(
        self, df: pd.DataFrame, price: int, holding_qty: int, qty: int
    ) -> Signal:
        fast = self.params.get("macd_fast", 12)
        slow = self.params.get("macd_slow", 26)
        signal_period = self.params.get("macd_signal", 9)

        macd_ind = ta.trend.MACD(df["close"], window_fast=fast, window_slow=slow, window_sign=signal_period)
        macd_line = macd_ind.macd()
        signal_line = macd_ind.macd_signal()

        if len(macd_line) < 2 or len(signal_line) < 2:
            return Signal(action="hold", stock_code=self.stock_code)

        prev_macd = macd_line.iloc[-2]
        prev_signal = signal_line.iloc[-2]
        curr_macd = macd_line.iloc[-1]
        curr_signal = signal_line.iloc[-1]

        if prev_macd <= prev_signal and curr_macd > curr_signal and holding_qty == 0:
            return Signal(
                action="buy",
                stock_code=self.stock_code,
                qty=qty,
                price=price,
                reason=f"MACD 골든크로스: MACD({curr_macd:.2f}) > Signal({curr_signal:.2f})",
                confidence=0.8,
            )

        if prev_macd >= prev_signal and curr_macd < curr_signal and holding_qty > 0:
            return Signal(
                action="sell",
                stock_code=self.stock_code,
                qty=min(qty, holding_qty),
                price=price,
                reason=f"MACD 데드크로스: MACD({curr_macd:.2f}) < Signal({curr_signal:.2f})",
                confidence=0.8,
            )

        return Signal(action="hold", stock_code=self.stock_code)

    def _evaluate_bollinger(
        self, df: pd.DataFrame, price: int, holding_qty: int, qty: int
    ) -> Signal:
        period = self.params.get("bb_period", 20)
        std_dev = self.params.get("bb_std", 2.0)

        bb = ta.volatility.BollingerBands(df["close"], window=period, window_dev=std_dev)
        upper = bb.bollinger_hband().iloc[-1]
        lower = bb.bollinger_lband().iloc[-1]
        mid = bb.bollinger_mavg().iloc[-1]

        if price <= lower and holding_qty == 0:
            return Signal(
                action="buy",
                stock_code=self.stock_code,
                qty=qty,
                price=price,
                reason=f"볼린저밴드 하단 돌파: 가격({price}) <= 하단({lower:.0f})",
                confidence=0.7,
            )

        if price >= upper and holding_qty > 0:
            return Signal(
                action="sell",
                stock_code=self.stock_code,
                qty=min(qty, holding_qty),
                price=price,
                reason=f"볼린저밴드 상단 돌파: 가격({price}) >= 상단({upper:.0f})",
                confidence=0.7,
            )

        return Signal(action="hold", stock_code=self.stock_code)