gihyeon/cointrader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add signal score detail to bot logs for HOLD reason debugging

Browse Source 
get_signal() now returns (signal, detail) tuple with long/short scores,
ADX value, volume surge status, and HOLD reason for easier diagnosis.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
21in7
2026-03-07 02:20:44 +09:00
parent c577019793
commit 0a8748913e
5 changed files with 46 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
scripts/train_model.py
View File

@@ -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

10
src/bot.py
View File

@@ -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()

34
src/indicators.py
View File

@@ -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,

8
tests/test_bot.py
View File

@@ -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:

11
tests/test_indicators.py
View File

@@ -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