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
|
"""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="BTC/USDT",
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
|
