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
|
from decimal import Decimal
from shared.models import Candle, Signal, OrderSide
from strategies.base import BaseStrategy
class VwapStrategy(BaseStrategy):
name: str = "vwap"
def __init__(self) -> None:
super().__init__()
self._deviation_threshold: float = 0.002
self._quantity: Decimal = Decimal("0.01")
self._cumulative_tp_vol: float = 0.0
self._cumulative_vol: float = 0.0
self._candle_count: int = 0
self._was_below_vwap: bool = False
self._was_above_vwap: bool = False
@property
def warmup_period(self) -> int:
return 30
def configure(self, params: dict) -> None:
self._deviation_threshold = float(params.get("deviation_threshold", 0.002))
self._quantity = Decimal(str(params.get("quantity", "0.01")))
if self._deviation_threshold <= 0:
raise ValueError(
f"VWAP deviation_threshold must be > 0, got {self._deviation_threshold}"
)
if self._quantity <= 0:
raise ValueError(f"Quantity must be positive, got {self._quantity}")
self._init_filters(
require_trend=False,
adx_threshold=float(params.get("adx_threshold", 25.0)),
min_volume_ratio=float(params.get("min_volume_ratio", 0.5)),
atr_stop_multiplier=float(params.get("atr_stop_multiplier", 2.0)),
atr_tp_multiplier=float(params.get("atr_tp_multiplier", 3.0)),
)
def reset(self) -> None:
self._cumulative_tp_vol = 0.0
self._cumulative_vol = 0.0
self._candle_count = 0
self._was_below_vwap = False
self._was_above_vwap = False
def _vwap_conviction(self, deviation: float) -> float:
"""Map VWAP deviation magnitude to conviction (0.1-1.0).
Further from VWAP = stronger mean reversion signal.
"""
magnitude = abs(deviation)
# Scale: at threshold -> 0.3, at 5x threshold -> ~1.0
return min(1.0, max(0.1, magnitude / self._deviation_threshold * 0.3))
def on_candle(self, candle: Candle) -> Signal | None:
self._update_filter_data(candle)
high = float(candle.high)
low = float(candle.low)
close = float(candle.close)
volume = float(candle.volume)
typical_price = (high + low + close) / 3.0
self._cumulative_tp_vol += typical_price * volume
self._cumulative_vol += volume
self._candle_count += 1
if self._candle_count < self.warmup_period:
return None
if self._cumulative_vol == 0.0:
return None
vwap = self._cumulative_tp_vol / self._cumulative_vol
if vwap == 0.0:
return None
deviation = (close - vwap) / vwap
if deviation < -self._deviation_threshold:
self._was_below_vwap = True
if deviation > self._deviation_threshold:
self._was_above_vwap = True
# Mean reversion from below: was below VWAP, now back near it
if self._was_below_vwap and abs(deviation) <= self._deviation_threshold:
self._was_below_vwap = False
conviction = self._vwap_conviction(deviation)
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.BUY,
price=candle.close,
quantity=self._quantity,
conviction=conviction,
reason=f"VWAP mean reversion BUY: deviation {deviation:.4f} within threshold {self._deviation_threshold}",
)
return self._apply_filters(signal)
# Mean reversion from above: was above VWAP, now back near it
if self._was_above_vwap and abs(deviation) <= self._deviation_threshold:
self._was_above_vwap = False
conviction = self._vwap_conviction(deviation)
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.SELL,
price=candle.close,
quantity=self._quantity,
conviction=conviction,
reason=f"VWAP mean reversion SELL: deviation {deviation:.4f} within threshold {self._deviation_threshold}",
)
return self._apply_filters(signal)
return None
|
