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
|
"""Tests for the VWAP strategy."""
from datetime import datetime, timezone
from decimal import Decimal
import pytest
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,
) -> Candle:
if high is None:
high = close
if low is None:
low = close
return Candle(
symbol="BTC/USDT",
timeframe="1m",
open_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
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
|
