93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
|
|
|
|
import pandas as pd
|
|
|
|
from .test_helpers import check_and_notify
|
|
|
|
|
|
def test_compute_macd_hist_monkeypatch(monkeypatch):
|
|
# Arrange: monkeypatch pandas_ta.macd to return a DataFrame with MACDh column
|
|
dummy_macd = pd.DataFrame({"MACDh_12_26_9": [None, 0.5, 1.2, 2.3]})
|
|
|
|
def fake_macd(series, fast, slow, signal):
|
|
return dummy_macd
|
|
|
|
# 올바른 모듈 경로로 monkey patch
|
|
monkeypatch.setattr("src.indicators.ta.macd", fake_macd)
|
|
|
|
close = pd.Series([1, 2, 3, 4])
|
|
|
|
# Act: import directly from indicators
|
|
from src.indicators import compute_macd_hist
|
|
|
|
hist = compute_macd_hist(close)
|
|
|
|
# Assert
|
|
assert isinstance(hist, pd.Series)
|
|
assert list(hist.dropna()) == [0.5, 1.2, 2.3]
|
|
|
|
|
|
def test_check_and_notify_positive_sends(monkeypatch):
|
|
# Prepare a fake OHLCV DataFrame with required OHLCV columns
|
|
idx = pd.date_range(end=pd.Timestamp.now(), periods=250, freq="h")
|
|
df = pd.DataFrame(
|
|
{
|
|
"open": list(range(100, 350)),
|
|
"high": list(range(105, 355)),
|
|
"low": list(range(95, 345)),
|
|
"close": list(range(100, 350)),
|
|
"volume": [1000] * 250,
|
|
},
|
|
index=idx,
|
|
)
|
|
|
|
# Monkeypatch at the point of use: src.signals imports from indicators
|
|
from src import signals
|
|
|
|
# Patch fetch_ohlcv to return complete OHLCV data
|
|
monkeypatch.setattr(signals, "fetch_ohlcv", lambda symbol, timeframe, limit=200, log_buffer=None: df)
|
|
|
|
# Fake pandas_ta.macd to return MACD crossover (signal cross)
|
|
def fake_macd(close, fast=12, slow=26, signal=9):
|
|
macd_df = pd.DataFrame(index=close.index)
|
|
# Create crossover: prev < signal, curr > signal
|
|
macd_values = [-0.5] * (len(close) - 1) + [1.5] # Last value crosses above
|
|
signal_values = [0.5] * len(close) # Constant signal line
|
|
macd_df["MACD_12_26_9"] = pd.Series(macd_values, index=close.index)
|
|
macd_df["MACDs_12_26_9"] = pd.Series(signal_values, index=close.index)
|
|
macd_df["MACDh_12_26_9"] = pd.Series(
|
|
[v - s for v, s in zip(macd_values, signal_values, strict=True)], index=close.index
|
|
)
|
|
return macd_df
|
|
|
|
monkeypatch.setattr(signals.ta, "macd", fake_macd)
|
|
|
|
# Fake pandas_ta.adx to return valid ADX data
|
|
def fake_adx(high, low, close, length=14):
|
|
adx_df = pd.DataFrame(index=close.index)
|
|
adx_df[f"ADX_{length}"] = pd.Series([30.0] * len(close), index=close.index)
|
|
return adx_df
|
|
|
|
monkeypatch.setattr(signals.ta, "adx", fake_adx)
|
|
|
|
# Capture calls to safe_send_telegram
|
|
called = {"count": 0}
|
|
|
|
def fake_safe_send(token, chat_id, text, **kwargs):
|
|
called["count"] += 1
|
|
return True
|
|
|
|
# Monkeypatch test_helpers module
|
|
from . import test_helpers
|
|
|
|
monkeypatch.setattr(test_helpers, "safe_send_telegram", fake_safe_send)
|
|
|
|
# Act: call check_and_notify (not dry-run)
|
|
check_and_notify("upbit", "KRW-BTC", "1h", "token", "chat", limit=10, dry_run=False)
|
|
|
|
# Assert: safe_send_telegram was called
|
|
assert called["count"] == 1
|