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
178
179
180
|
"""Tests for the Volume Profile strategy."""
from datetime import datetime, timezone
from decimal import Decimal
from shared.models import Candle, OrderSide
from strategies.volume_profile_strategy import VolumeProfileStrategy
def make_candle(close: float, volume: float = 1.0) -> 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(str(volume)),
)
def test_volume_profile_warmup_period():
strategy = VolumeProfileStrategy()
strategy.configure({"lookback_period": 10, "num_bins": 5})
assert strategy.warmup_period == 10
def test_volume_profile_no_signal_insufficient_data():
strategy = VolumeProfileStrategy()
strategy.configure({"lookback_period": 10, "num_bins": 5})
# Feed fewer candles than lookback_period
for _ in range(5):
result = strategy.on_candle(make_candle(100.0, 10.0))
assert result is None
def test_volume_profile_buy_at_value_area_low():
"""Concentrate volume around 95-105, price drops to 88, bounces back to 99."""
strategy = VolumeProfileStrategy()
strategy.configure(
{
"lookback_period": 10,
"num_bins": 5,
"value_area_pct": 0.7,
"quantity": "0.01",
}
)
# Build profile: 10 candles with volume concentrated around 95-105
profile_data = [
(95, 50),
(97, 50),
(99, 100),
(100, 100),
(101, 100),
(103, 50),
(105, 50),
(100, 100),
(99, 100),
(101, 50),
]
for price, vol in profile_data:
strategy.on_candle(make_candle(price, vol))
# Price drops below value area low
strategy.on_candle(make_candle(88.0, 1.0))
# Price bounces back into value area (between va_low and poc)
signal = strategy.on_candle(make_candle(99.0, 1.0))
assert signal is not None
assert signal.side == OrderSide.BUY
def test_volume_profile_sell_at_value_area_high():
"""Concentrate volume around 95-105, price rises to 112, pulls back to 101."""
strategy = VolumeProfileStrategy()
strategy.configure(
{
"lookback_period": 10,
"num_bins": 5,
"value_area_pct": 0.7,
"quantity": "0.01",
}
)
# Build profile: 10 candles with volume concentrated around 95-105
profile_data = [
(95, 50),
(97, 50),
(99, 100),
(100, 100),
(101, 100),
(103, 50),
(105, 50),
(100, 100),
(99, 100),
(101, 50),
]
for price, vol in profile_data:
strategy.on_candle(make_candle(price, vol))
# Price rises above value area high
strategy.on_candle(make_candle(112.0, 1.0))
# Price pulls back into value area (between poc and va_high)
signal = strategy.on_candle(make_candle(101.0, 1.0))
assert signal is not None
assert signal.side == OrderSide.SELL
def test_volume_profile_reset_clears_state():
strategy = VolumeProfileStrategy()
strategy.configure({"lookback_period": 10, "num_bins": 5})
# Feed enough candles to establish profile
for _ in range(10):
strategy.on_candle(make_candle(100.0, 10.0))
strategy.reset()
# After reset, should not have enough data
result = strategy.on_candle(make_candle(100.0, 10.0))
assert result is None
def test_volume_profile_hvn_detection():
"""Feed clustered volume at specific price levels to produce HVN nodes."""
strategy = VolumeProfileStrategy()
strategy.configure({"lookback_period": 20, "num_bins": 10, "value_area_pct": 0.7})
# Create a profile with very high volume at price ~100 and low volume elsewhere
# Prices range from 90 to 110, heavy volume concentrated at 100
candles_data = []
# Low volume at extremes
for p in [90, 91, 92, 109, 110]:
candles_data.append((p, 1.0))
# Very high volume around 100
for _ in range(15):
candles_data.append((100, 100.0))
for price, vol in candles_data:
strategy.on_candle(make_candle(price, vol))
# Access the internal method to verify HVN detection
result = strategy._compute_value_area()
assert result is not None
poc, va_low, va_high, hvn_levels, lvn_levels = result
# The bin containing price ~100 should have very high volume -> HVN
assert len(hvn_levels) > 0
# At least one HVN should be near 100
assert any(abs(h - 100) < 5 for h in hvn_levels)
def test_volume_profile_reset_thorough():
"""Verify all state is cleared on reset."""
strategy = VolumeProfileStrategy()
strategy.configure({"lookback_period": 10, "num_bins": 5})
# Build up state
for _ in range(10):
strategy.on_candle(make_candle(100.0, 10.0))
# Set below/above VA flags
strategy.on_candle(make_candle(50.0, 1.0)) # below VA
strategy.on_candle(make_candle(200.0, 1.0)) # above VA
strategy.reset()
# Verify all state cleared
assert len(strategy._candles) == 0
assert strategy._was_below_va is False
assert strategy._was_above_va is False
# Should not produce signal since no data
result = strategy.on_candle(make_candle(100.0, 10.0))
assert result is None
|
