1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
"""Tests for the MACD strategy."""
from datetime import datetime, timezone
from decimal import Decimal
from shared.models import Candle, OrderSide
from strategies.macd_strategy import MacdStrategy
def _candle(price: float) -> Candle:
return Candle(
symbol="BTC/USDT",
timeframe="1m",
open_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
open=Decimal(str(price)),
high=Decimal(str(price)),
low=Decimal(str(price)),
close=Decimal(str(price)),
volume=Decimal("1.0"),
)
def _make_strategy(**kwargs) -> MacdStrategy:
params = {"fast_period": 3, "slow_period": 6, "signal_period": 3, "quantity": "0.01"}
params.update(kwargs)
s = MacdStrategy()
s.configure(params)
return s
def test_macd_warmup_period():
s = _make_strategy()
assert s.warmup_period == 6 + 3 # slow_period + signal_period
def test_macd_no_signal_insufficient_data():
s = _make_strategy()
# Feed fewer candles than warmup_period
for price in [100.0, 101.0, 102.0]:
result = s.on_candle(_candle(price))
assert result is None
def test_macd_buy_signal_on_bullish_crossover():
s = _make_strategy()
# Declining prices drive MACD histogram negative, then rising prices cross positive
prices = [100, 99, 98, 97, 96, 95, 94, 93, 92, 91, 90, 89, 88]
prices += [89, 91, 94, 98, 103, 109, 116, 124, 133, 143]
signal = None
for p in prices:
result = s.on_candle(_candle(float(p)))
if result is not None:
signal = result
assert signal is not None, "Expected a BUY signal from bullish crossover"
assert signal.side == OrderSide.BUY
def test_macd_sell_signal_on_bearish_crossover():
s = _make_strategy()
# Rising prices drive MACD histogram positive, then declining prices cross negative
prices = [100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112]
prices += [111, 109, 106, 102, 97, 91, 84, 76, 67, 57]
signal = None
for p in prices:
result = s.on_candle(_candle(float(p)))
if result is not None:
signal = result
assert signal is not None, "Expected a SELL signal from bearish crossover"
assert signal.side == OrderSide.SELL
def test_macd_reset_clears_state():
s = _make_strategy()
for p in [100, 101, 102, 103, 104, 105, 106, 107, 108]:
s.on_candle(_candle(float(p)))
assert len(s._closes) > 0
s.reset()
assert len(s._closes) == 0
assert s._prev_histogram is None
assert s._prev_macd is None
assert s._prev_signal is None
def test_macd_signal_line_crossover():
"""Test that MACD signal-line crossover generates signals."""
s = _make_strategy()
# Declining then rising prices should produce a signal-line bullish crossover
prices = [100, 99, 98, 97, 96, 95, 94, 93, 92, 91, 90, 89, 88]
prices += [89, 91, 94, 98, 103, 109, 116, 124, 133, 143]
signals = []
for p in prices:
result = s.on_candle(_candle(float(p)))
if result is not None:
signals.append(result)
buy_signals = [sig for sig in signals if sig.side == OrderSide.BUY]
assert len(buy_signals) > 0, "Expected at least one BUY signal"
# Check that at least one is a signal-line crossover or histogram crossover
all_reasons = [sig.reason for sig in buy_signals]
assert any("crossover" in r for r in all_reasons), (
f"Expected crossover signal, got: {all_reasons}"
)
def test_macd_conviction_varies_with_distance():
"""Test that conviction varies based on MACD distance from zero line."""
s1 = _make_strategy()
s2 = _make_strategy()
# Small price movements -> MACD near zero -> lower conviction
small_prices = [100, 99.5, 99, 98.5, 98, 97.5, 97, 96.5, 96, 95.5, 95, 94.5, 94]
small_prices += [94.5, 95, 95.5, 96, 96.5, 97, 97.5, 98, 98.5, 99]
small_signals = []
for p in small_prices:
result = s1.on_candle(_candle(float(p)))
if result is not None:
small_signals.append(result)
# Large price movements -> MACD far from zero -> higher conviction
large_prices = [100, 95, 90, 85, 80, 75, 70, 65, 60, 55, 50, 45, 40]
large_prices += [45, 55, 70, 90, 115, 145, 180, 220, 265, 315]
large_signals = []
for p in large_prices:
result = s2.on_candle(_candle(float(p)))
if result is not None:
large_signals.append(result)
# Both should produce signals
assert len(small_signals) > 0, "Expected signals from small movements"
assert len(large_signals) > 0, "Expected signals from large movements"
# The large-movement signals should generally have higher conviction
# (or at least different conviction, since distance from zero affects it)
small_conv = small_signals[-1].conviction
large_conv = large_signals[-1].conviction
# Large movements should produce conviction >= small movements
assert large_conv >= small_conv, (
f"Expected large movement conviction ({large_conv}) >= small ({small_conv})"
)
|
