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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
"""Tests for the Bollinger Bands strategy."""
from datetime import datetime, timezone
from decimal import Decimal
from shared.models import Candle, OrderSide
from strategies.bollinger_strategy import BollingerStrategy
def make_candle(close: float) -> Candle:
return Candle(
symbol="AAPL",
timeframe="1m",
open_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
open=Decimal(str(close)),
high=Decimal(str(close)),
low=Decimal(str(close)),
close=Decimal(str(close)),
volume=Decimal("1.0"),
)
def _make_strategy() -> BollingerStrategy:
s = BollingerStrategy()
s.configure({"period": 5, "num_std": 1.0, "min_bandwidth": 0.0, "squeeze_threshold": 0.0})
return s
def test_bollinger_warmup_period():
strategy = BollingerStrategy()
strategy.configure({"period": 20})
assert strategy.warmup_period == 20
strategy.configure({"period": 5})
assert strategy.warmup_period == 5
def test_bollinger_no_signal_insufficient_data():
strategy = _make_strategy()
# Feed fewer candles than the period
for price in [100.0, 101.0, 102.0, 101.0]:
result = strategy.on_candle(make_candle(price))
assert result is None
def test_bollinger_buy_on_lower_band_recovery():
strategy = _make_strategy()
# Feed stable prices to build up the window
for _ in range(5):
strategy.on_candle(make_candle(100.0))
# Drop well below the lower band
signal = strategy.on_candle(make_candle(50.0))
# No buy yet -- still below lower band
assert signal is None
# Recover back inside the bands
signal = strategy.on_candle(make_candle(100.0))
assert signal is not None
assert signal.side == OrderSide.BUY
assert "lower" in signal.reason.lower() or "bollinger" in signal.reason.lower()
def test_bollinger_sell_on_upper_band_recovery():
strategy = _make_strategy()
# Feed stable prices to build up the window
for _ in range(5):
strategy.on_candle(make_candle(100.0))
# Spike well above the upper band
signal = strategy.on_candle(make_candle(150.0))
# No sell yet -- still above upper band
assert signal is None
# Recover back inside the bands
signal = strategy.on_candle(make_candle(100.0))
assert signal is not None
assert signal.side == OrderSide.SELL
assert "upper" in signal.reason.lower() or "bollinger" in signal.reason.lower()
def test_bollinger_reset_clears_state():
strategy = _make_strategy()
# Build some state
for _ in range(5):
strategy.on_candle(make_candle(100.0))
strategy.on_candle(make_candle(50.0)) # penetrate lower band
strategy.reset()
# After reset, insufficient data again
result = strategy.on_candle(make_candle(100.0))
assert result is None
# Internal state should be cleared
assert len(strategy._closes) == 1
assert strategy._was_below_lower is False
assert strategy._was_above_upper is False
assert strategy._in_squeeze is False
assert strategy._squeeze_bars == 0
def test_bollinger_squeeze_detection():
"""Tight bandwidth → no signal during squeeze."""
# Use a strategy with a high squeeze threshold so constant prices trigger squeeze
s = BollingerStrategy()
s.configure(
{
"period": 5,
"num_std": 2.0,
"min_bandwidth": 0.0,
"squeeze_threshold": 0.5, # Very high threshold to ensure squeeze triggers
}
)
# Feed identical prices → bandwidth = 0 (below any threshold)
for _ in range(6):
result = s.on_candle(make_candle(100.0))
# With identical prices, std=0, bandwidth=0 < 0.5 → squeeze, no signal
assert s._in_squeeze is True
assert result is None
def test_bollinger_squeeze_breakout_buy():
"""Squeeze ends with price above SMA → BUY signal."""
s = BollingerStrategy()
s.configure(
{
"period": 5,
"num_std": 1.0,
"min_bandwidth": 0.0,
"squeeze_threshold": 0.01,
}
)
# Feed identical prices to create a squeeze (bandwidth = 0)
for _ in range(6):
s.on_candle(make_candle(100.0))
assert s._in_squeeze is True
# Now feed a price that creates enough spread to exit squeeze AND is above SMA
signal = s.on_candle(make_candle(120.0))
assert signal is not None
assert signal.side == OrderSide.BUY
assert "squeeze breakout UP" in signal.reason
def test_bollinger_pct_b_conviction():
"""Signals near band extremes have higher conviction via %B."""
s = BollingerStrategy()
s.configure(
{
"period": 5,
"num_std": 1.0,
"min_bandwidth": 0.0,
"squeeze_threshold": 0.0, # Disable squeeze for this test
}
)
# Build up with stable prices
for _ in range(5):
s.on_candle(make_candle(100.0))
# Drop below lower band
s.on_candle(make_candle(50.0))
# Recover just at the lower band edge — %B close to 0 → high conviction
signal = s.on_candle(make_candle(100.0))
assert signal is not None
assert signal.side == OrderSide.BUY
# conviction = max(1.0 - pct_b, 0.3), with pct_b near lower → conviction should be >= 0.3
assert signal.conviction >= 0.3
|
