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
|
"""Tests for the VWAP strategy."""
from datetime import datetime, timezone
from decimal import Decimal
from shared.models import Candle, OrderSide
from strategies.vwap_strategy import VwapStrategy
def make_candle(
close: float,
high: float | None = None,
low: float | None = None,
volume: float = 1.0,
open_time: datetime | None = None,
) -> Candle:
if high is None:
high = close
if low is None:
low = close
if open_time is None:
open_time = datetime(2024, 1, 1, tzinfo=timezone.utc)
return Candle(
symbol="AAPL",
timeframe="1m",
open_time=open_time,
open=Decimal(str(close)),
high=Decimal(str(high)),
low=Decimal(str(low)),
close=Decimal(str(close)),
volume=Decimal(str(volume)),
)
def _configured_strategy() -> VwapStrategy:
strategy = VwapStrategy()
strategy.configure({"deviation_threshold": 0.01, "quantity": "0.01"})
return strategy
def test_vwap_warmup_period():
strategy = VwapStrategy()
assert strategy.warmup_period == 30
def test_vwap_no_signal_insufficient_data():
strategy = _configured_strategy()
# Feed fewer candles than warmup_period
for _ in range(29):
signal = strategy.on_candle(make_candle(100.0))
assert signal is None
def test_vwap_buy_signal_below_vwap_recovery():
strategy = _configured_strategy()
# Build VWAP around 100 with 30 candles (satisfy warmup)
for _ in range(30):
strategy.on_candle(make_candle(100.0, high=101.0, low=99.0))
# Drop price well below VWAP to trigger _was_below_vwap
for _ in range(3):
strategy.on_candle(make_candle(95.0, high=96.0, low=94.0))
# Recover back to VWAP (close ~100, deviation within threshold)
signal = strategy.on_candle(make_candle(100.0, high=101.0, low=99.0))
assert signal is not None
assert signal.side == OrderSide.BUY
assert "VWAP" in signal.reason
def test_vwap_sell_signal_above_vwap_recovery():
strategy = _configured_strategy()
# Build VWAP around 100 with 30 candles (satisfy warmup)
for _ in range(30):
strategy.on_candle(make_candle(100.0, high=101.0, low=99.0))
# Rise price well above VWAP to trigger _was_above_vwap
for _ in range(3):
strategy.on_candle(make_candle(105.0, high=106.0, low=104.0))
# Recover back to VWAP (close ~100, deviation within threshold)
signal = strategy.on_candle(make_candle(100.0, high=101.0, low=99.0))
assert signal is not None
assert signal.side == OrderSide.SELL
assert "VWAP" in signal.reason
def test_vwap_reset_clears_state():
strategy = _configured_strategy()
# Build some state
for _ in range(35):
strategy.on_candle(make_candle(100.0))
strategy.reset()
assert strategy._cumulative_tp_vol == 0.0
assert strategy._cumulative_vol == 0.0
assert strategy._candle_count == 0
assert strategy._was_below_vwap is False
assert strategy._was_above_vwap is False
assert strategy._current_date is None
assert len(strategy._tp_values) == 0
assert len(strategy._vwap_values) == 0
def test_vwap_daily_reset():
"""Candles from two different dates cause VWAP to reset."""
strategy = _configured_strategy()
day1 = datetime(2024, 1, 1, tzinfo=timezone.utc)
day2 = datetime(2024, 1, 2, tzinfo=timezone.utc)
# Feed 35 candles on day 1 to build VWAP state
for i in range(35):
strategy.on_candle(make_candle(100.0, high=101.0, low=99.0, open_time=day1))
# Verify state is built up
assert strategy._candle_count == 35
assert strategy._cumulative_vol > 0
assert strategy._current_date == "2024-01-01"
# Feed first candle of day 2 — should reset
strategy.on_candle(make_candle(100.0, high=101.0, low=99.0, open_time=day2))
# After reset, candle_count should be 1 (the new candle)
assert strategy._candle_count == 1
assert strategy._current_date == "2024-01-02"
def test_vwap_reset_clears_date():
"""Verify reset() clears _current_date and deviation band state."""
strategy = _configured_strategy()
for _ in range(35):
strategy.on_candle(make_candle(100.0))
assert strategy._current_date is not None
strategy.reset()
assert strategy._current_date is None
assert len(strategy._tp_values) == 0
assert len(strategy._vwap_values) == 0
|
