first commit

app/engine/strategies/technical.py

@@ -0,0 +1,158 @@
+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)