diff --git a/scripts/train_model.py b/scripts/train_model.py index aabbfa4..9b084b0 100644 --- a/scripts/train_model.py +++ b/scripts/train_model.py @@ -73,7 +73,7 @@ def _process_index(args: tuple) -> dict | None: if df_ind.iloc[-1].isna().any(): return None - signal = ind.get_signal(df_ind) + signal, _ = ind.get_signal(df_ind) if signal == "HOLD": return None diff --git a/src/bot.py b/src/bot.py index 100d84d..6ff7863 100644 --- a/src/bot.py +++ b/src/bot.py @@ -139,7 +139,7 @@ class TradingBot: ind = Indicators(df) df_with_indicators = ind.calculate_all() - raw_signal = ind.get_signal( + raw_signal, signal_detail = ind.get_signal( df_with_indicators, signal_threshold=self.config.signal_threshold, adx_threshold=self.config.adx_threshold, @@ -147,7 +147,13 @@ class TradingBot: ) current_price = df_with_indicators["close"].iloc[-1] - logger.info(f"[{self.symbol}] 신호: {raw_signal} | 현재가: {current_price:.4f} USDT") + adx_str = f"ADX={signal_detail['adx']:.1f}" if signal_detail['adx'] is not None else "ADX=N/A" + vol_str = "Vol급증" if signal_detail['vol_surge'] else "Vol정상" + score_str = f"L={signal_detail['long']} S={signal_detail['short']}" + if raw_signal == "HOLD" and signal_detail['hold_reason']: + logger.info(f"[{self.symbol}] 신호: HOLD | {score_str} | {adx_str} | {vol_str} | 사유: {signal_detail['hold_reason']} | 현재가: {current_price:.4f}") + else: + logger.info(f"[{self.symbol}] 신호: {raw_signal} | {score_str} | {adx_str} | {vol_str} | 현재가: {current_price:.4f}") position = await self.exchange.get_position() diff --git a/src/indicators.py b/src/indicators.py index b67cddb..f55363b 100644 --- a/src/indicators.py +++ b/src/indicators.py @@ -58,23 +58,29 @@ class Indicators: signal_threshold: int = 3, adx_threshold: float = 25, volume_multiplier: float = 2.5, - ) -> str: + ) -> tuple[str, dict]: """ 복합 지표 기반 매매 신호 생성. signal_threshold: 최소 가중치 합계 (기본 3) adx_threshold: ADX 최소값 필터 (0=비활성화, 25=ADX<25이면 HOLD) volume_multiplier: 거래량 급증 배수 (기본 1.5) + + Returns: + (signal, detail) — signal은 "LONG"/"SHORT"/"HOLD", + detail은 {"long": int, "short": int, "vol_surge": bool, "adx": float|None, "hold_reason": str} """ last = df.iloc[-1] prev = df.iloc[-2] # ADX 필터 adx = last.get("adx", None) - if adx is not None and not pd.isna(adx): - logger.debug(f"ADX: {adx:.1f}") - if adx_threshold > 0 and adx < adx_threshold: - return "HOLD" + adx_val = adx if adx is not None and not pd.isna(adx) else None + if adx_val is not None: + logger.debug(f"ADX: {adx_val:.1f}") + if adx_threshold > 0 and adx_val < adx_threshold: + detail = {"long": 0, "short": 0, "vol_surge": False, "adx": adx_val, "hold_reason": f"ADX({adx_val:.1f}) < {adx_threshold}"} + return "HOLD", detail long_signals = 0 short_signals = 0 @@ -112,11 +118,23 @@ class Indicators: # 6. 거래량 확인 (신호 강화) vol_surge = last["volume"] > last["vol_ma20"] * volume_multiplier + detail = {"long": long_signals, "short": short_signals, "vol_surge": vol_surge, "adx": adx_val, "hold_reason": ""} + if long_signals >= signal_threshold and (vol_surge or long_signals >= signal_threshold + 1): - return "LONG" + return "LONG", detail elif short_signals >= signal_threshold and (vol_surge or short_signals >= signal_threshold + 1): - return "SHORT" - return "HOLD" + return "SHORT", detail + + # HOLD 사유 구성 + best_side = "LONG" if long_signals >= short_signals else "SHORT" + best_score = max(long_signals, short_signals) + reasons = [] + if best_score < signal_threshold: + reasons.append(f"{best_side} 점수({best_score}) < 임계값({signal_threshold})") + elif not vol_surge and best_score < signal_threshold + 1: + reasons.append(f"거래량 미급증 & {best_side} 점수({best_score}) < {signal_threshold + 1}") + detail["hold_reason"] = ", ".join(reasons) if reasons else "점수 부족" + return "HOLD", detail def get_atr_stop( self, df: pd.DataFrame, side: str, entry_price: float, diff --git a/tests/test_bot.py b/tests/test_bot.py index 07de871..0d4e8a3 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -92,7 +92,7 @@ async def test_bot_processes_signal(config, sample_df): with patch("src.bot.Indicators") as MockInd: mock_ind = MagicMock() mock_ind.calculate_all.return_value = sample_df - mock_ind.get_signal.return_value = "LONG" + mock_ind.get_signal.return_value = ("LONG", {"long": 3, "short": 0, "vol_surge": True, "adx": 30.0, "hold_reason": ""}) mock_ind.get_atr_stop.return_value = (0.48, 0.56) MockInd.return_value = mock_ind await bot.process_candle(sample_df) @@ -178,7 +178,7 @@ async def test_process_candle_calls_close_and_reenter_on_reverse_signal(config, with patch("src.bot.Indicators") as MockInd: mock_ind = MagicMock() mock_ind.calculate_all.return_value = sample_df - mock_ind.get_signal.return_value = "SHORT" # 현재 LONG 포지션에 반대 시그널 + mock_ind.get_signal.return_value = ("SHORT", {"long": 0, "short": 3, "vol_surge": True, "adx": 30.0, "hold_reason": ""}) # 현재 LONG 포지션에 반대 시그널 MockInd.return_value = mock_ind await bot.process_candle(sample_df) @@ -207,7 +207,7 @@ async def test_process_candle_passes_raw_signal_to_close_and_reenter_even_if_ml_ with patch("src.bot.Indicators") as MockInd: mock_ind = MagicMock() mock_ind.calculate_all.return_value = sample_df - mock_ind.get_signal.return_value = "SHORT" + mock_ind.get_signal.return_value = ("SHORT", {"long": 0, "short": 3, "vol_surge": True, "adx": 30.0, "hold_reason": ""}) MockInd.return_value = mock_ind await bot.process_candle(sample_df) @@ -243,7 +243,7 @@ async def test_process_candle_fetches_oi_and_funding(config, sample_df): with patch("src.bot.Indicators") as mock_ind_cls: mock_ind = MagicMock() mock_ind.calculate_all.return_value = sample_df - mock_ind.get_signal.return_value = "LONG" + mock_ind.get_signal.return_value = ("LONG", {"long": 3, "short": 0, "vol_surge": True, "adx": 30.0, "hold_reason": ""}) mock_ind_cls.return_value = mock_ind with patch("src.bot.build_features_aligned") as mock_build: diff --git a/tests/test_indicators.py b/tests/test_indicators.py index 43acc63..af0ca2f 100644 --- a/tests/test_indicators.py +++ b/tests/test_indicators.py @@ -66,10 +66,11 @@ def test_adx_filter_blocks_low_adx(sample_df): df.loc[df.index[-1], "volume"] = df.loc[df.index[-1], "vol_ma20"] * 3 df["adx"] = 15.0 # 기본 adx_threshold=25이므로 ADX=15은 HOLD - signal = ind.get_signal(df) + signal, detail = ind.get_signal(df) assert signal == "HOLD" + assert "ADX" in detail["hold_reason"] # adx_threshold=0이면 ADX 필터 비활성화 → LONG - signal = ind.get_signal(df, adx_threshold=0) + signal, detail = ind.get_signal(df, adx_threshold=0) assert signal == "LONG" @@ -78,13 +79,15 @@ def test_adx_nan_falls_through(sample_df): ind = Indicators(sample_df) df = ind.calculate_all() df["adx"] = float("nan") - signal = ind.get_signal(df) + signal, detail = ind.get_signal(df) # NaN이면 차단하지 않고 기존 로직 실행 → LONG/SHORT/HOLD 중 하나 assert signal in ("LONG", "SHORT", "HOLD") + assert detail["adx"] is None def test_signal_returns_direction(sample_df): ind = Indicators(sample_df) df = ind.calculate_all() - signal = ind.get_signal(df) + signal, detail = ind.get_signal(df) assert signal in ("LONG", "SHORT", "HOLD") + assert "long" in detail and "short" in detail