From 7a3dde2146d2caf0349502748b504eb95c858c7c Mon Sep 17 00:00:00 2001 From: 21in7 Date: Sun, 1 Mar 2026 12:50:51 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EB=B3=B5=ED=95=A9=20=EA=B8=B0=EC=88=A0?= =?UTF-8?q?=20=EC=A7=80=ED=91=9C=20=EB=AA=A8=EB=93=88=20=EA=B5=AC=ED=98=84?= =?UTF-8?q?=20(RSI/MACD/BB/EMA/ATR/StochRSI)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Made-with: Cursor --- src/indicators.py | 115 +++++++++++++++++++++++++++++++++++++++ tests/test_indicators.py | 52 ++++++++++++++++++ 2 files changed, 167 insertions(+) create mode 100644 src/indicators.py create mode 100644 tests/test_indicators.py diff --git a/src/indicators.py b/src/indicators.py new file mode 100644 index 0000000..882d87e --- /dev/null +++ b/src/indicators.py @@ -0,0 +1,115 @@ +import pandas as pd +import pandas_ta as ta +from loguru import logger + + +class Indicators: + """ + 복합 기술 지표 계산 및 매매 신호 생성. + 공격적 전략: 여러 지표가 동시에 같은 방향을 가리킬 때 진입. + """ + + def __init__(self, df: pd.DataFrame): + self.df = df.copy() + + def calculate_all(self) -> pd.DataFrame: + df = self.df + + # RSI (14) + df["rsi"] = ta.rsi(df["close"], length=14) + + # MACD (12, 26, 9) + macd = ta.macd(df["close"], fast=12, slow=26, signal=9) + df["macd"] = macd["MACD_12_26_9"] + df["macd_signal"] = macd["MACDs_12_26_9"] + df["macd_hist"] = macd["MACDh_12_26_9"] + + # 볼린저 밴드 (20, 2) + bb = ta.bbands(df["close"], length=20, std=2) + df["bb_upper"] = bb["BBU_20_2.0_2.0"] + df["bb_mid"] = bb["BBM_20_2.0_2.0"] + df["bb_lower"] = bb["BBL_20_2.0_2.0"] + + # EMA (9, 21, 50) + df["ema9"] = ta.ema(df["close"], length=9) + df["ema21"] = ta.ema(df["close"], length=21) + df["ema50"] = ta.ema(df["close"], length=50) + + # ATR (14) - 변동성 기반 손절 계산용 + df["atr"] = ta.atr(df["high"], df["low"], df["close"], length=14) + + # Stochastic RSI + stoch = ta.stochrsi(df["close"], length=14) + df["stoch_k"] = stoch["STOCHRSIk_14_14_3_3"] + df["stoch_d"] = stoch["STOCHRSId_14_14_3_3"] + + # 거래량 이동평균 + df["vol_ma20"] = ta.sma(df["volume"], length=20) + + return df + + def get_signal(self, df: pd.DataFrame) -> str: + """ + 복합 지표 기반 매매 신호 생성. + 공격적 전략: 3개 이상 지표 일치 시 진입. + """ + last = df.iloc[-1] + prev = df.iloc[-2] + + long_signals = 0 + short_signals = 0 + + # 1. RSI + if last["rsi"] < 35: + long_signals += 1 + elif last["rsi"] > 65: + short_signals += 1 + + # 2. MACD 크로스 + if prev["macd"] < prev["macd_signal"] and last["macd"] > last["macd_signal"]: + long_signals += 2 # 크로스는 강한 신호 + elif prev["macd"] > prev["macd_signal"] and last["macd"] < last["macd_signal"]: + short_signals += 2 + + # 3. 볼린저 밴드 돌파 + if last["close"] < last["bb_lower"]: + long_signals += 1 + elif last["close"] > last["bb_upper"]: + short_signals += 1 + + # 4. EMA 정배열/역배열 + if last["ema9"] > last["ema21"] > last["ema50"]: + long_signals += 1 + elif last["ema9"] < last["ema21"] < last["ema50"]: + short_signals += 1 + + # 5. Stochastic RSI 과매도/과매수 + if last["stoch_k"] < 20 and last["stoch_k"] > last["stoch_d"]: + long_signals += 1 + elif last["stoch_k"] > 80 and last["stoch_k"] < last["stoch_d"]: + short_signals += 1 + + # 6. 거래량 확인 (신호 강화) + vol_surge = last["volume"] > last["vol_ma20"] * 1.5 + + threshold = 3 + if long_signals >= threshold and (vol_surge or long_signals >= 4): + return "LONG" + elif short_signals >= threshold and (vol_surge or short_signals >= 4): + return "SHORT" + return "HOLD" + + def get_atr_stop( + self, df: pd.DataFrame, side: str, entry_price: float + ) -> tuple[float, float]: + """ATR 기반 손절/익절 가격 반환 (stop_loss, take_profit)""" + atr = df["atr"].iloc[-1] + multiplier_sl = 1.5 + multiplier_tp = 3.0 + if side == "LONG": + stop_loss = entry_price - atr * multiplier_sl + take_profit = entry_price + atr * multiplier_tp + else: + stop_loss = entry_price + atr * multiplier_sl + take_profit = entry_price - atr * multiplier_tp + return stop_loss, take_profit diff --git a/tests/test_indicators.py b/tests/test_indicators.py new file mode 100644 index 0000000..2a10be9 --- /dev/null +++ b/tests/test_indicators.py @@ -0,0 +1,52 @@ +import pandas as pd +import numpy as np +import pytest +from src.indicators import Indicators + + +@pytest.fixture +def sample_df(): + """100개 캔들 샘플 데이터""" + np.random.seed(42) + n = 100 + close = np.cumsum(np.random.randn(n) * 0.01) + 0.5 + df = pd.DataFrame({ + "open": close * (1 + np.random.randn(n) * 0.001), + "high": close * (1 + np.abs(np.random.randn(n)) * 0.005), + "low": close * (1 - np.abs(np.random.randn(n)) * 0.005), + "close": close, + "volume": np.random.randint(100000, 1000000, n).astype(float), + }) + return df + + +def test_rsi_range(sample_df): + ind = Indicators(sample_df) + df = ind.calculate_all() + assert "rsi" in df.columns + valid = df["rsi"].dropna() + assert (valid >= 0).all() and (valid <= 100).all() + + +def test_macd_columns(sample_df): + ind = Indicators(sample_df) + df = ind.calculate_all() + assert "macd" in df.columns + assert "macd_signal" in df.columns + assert "macd_hist" in df.columns + + +def test_bollinger_bands(sample_df): + ind = Indicators(sample_df) + df = ind.calculate_all() + assert "bb_upper" in df.columns + assert "bb_lower" in df.columns + valid = df.dropna() + assert (valid["bb_upper"] >= valid["bb_lower"]).all() + + +def test_signal_returns_direction(sample_df): + ind = Indicators(sample_df) + df = ind.calculate_all() + signal = ind.get_signal(df) + assert signal in ("LONG", "SHORT", "HOLD")