diff options
Diffstat (limited to 'services/strategy-engine/strategies/vwap_strategy.py')
| -rw-r--r-- | services/strategy-engine/strategies/vwap_strategy.py | 49 |
1 files changed, 47 insertions, 2 deletions
diff --git a/services/strategy-engine/strategies/vwap_strategy.py b/services/strategy-engine/strategies/vwap_strategy.py index c525ff3..0348752 100644 --- a/services/strategy-engine/strategies/vwap_strategy.py +++ b/services/strategy-engine/strategies/vwap_strategy.py @@ -1,3 +1,4 @@ +from collections import deque from decimal import Decimal from shared.models import Candle, Signal, OrderSide @@ -16,6 +17,9 @@ class VwapStrategy(BaseStrategy): self._candle_count: int = 0 self._was_below_vwap: bool = False self._was_above_vwap: bool = False + self._current_date: str | None = None # Track date for daily reset + self._tp_values: deque[float] = deque(maxlen=500) # For std calculation + self._vwap_values: deque[float] = deque(maxlen=500) @property def warmup_period(self) -> int: @@ -41,11 +45,15 @@ class VwapStrategy(BaseStrategy): ) def reset(self) -> None: + super().reset() 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 + self._current_date = None + self._tp_values.clear() + self._vwap_values.clear() def _vwap_conviction(self, deviation: float) -> float: """Map VWAP deviation magnitude to conviction (0.1-1.0). @@ -58,6 +66,20 @@ class VwapStrategy(BaseStrategy): def on_candle(self, candle: Candle) -> Signal | None: self._update_filter_data(candle) + + # Daily reset + candle_date = candle.open_time.strftime("%Y-%m-%d") + if self._current_date is not None and candle_date != self._current_date: + # New day — reset VWAP + 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 + self._tp_values.clear() + self._vwap_values.clear() + self._current_date = candle_date + high = float(candle.high) low = float(candle.low) close = float(candle.close) @@ -77,6 +99,19 @@ class VwapStrategy(BaseStrategy): vwap = self._cumulative_tp_vol / self._cumulative_vol if vwap == 0.0: return None + + # Track values for deviation band calculation + self._tp_values.append(typical_price) + self._vwap_values.append(vwap) + + # Standard deviation of (TP - VWAP) for bands + std_dev = 0.0 + if len(self._tp_values) >= 2: + diffs = [tp - v for tp, v in zip(self._tp_values, self._vwap_values)] + mean_diff = sum(diffs) / len(diffs) + variance = sum((d - mean_diff) ** 2 for d in diffs) / len(diffs) + std_dev = variance ** 0.5 + deviation = (close - vwap) / vwap if deviation < -self._deviation_threshold: @@ -84,10 +119,20 @@ class VwapStrategy(BaseStrategy): if deviation > self._deviation_threshold: self._was_above_vwap = True + # Determine conviction based on deviation bands + def _band_conviction(price: float) -> float: + if std_dev > 0 and len(self._tp_values) >= 2: + dist_from_vwap = abs(price - vwap) + if dist_from_vwap >= 2 * std_dev: + return 0.9 + elif dist_from_vwap >= std_dev: + return 0.6 + return 0.5 + # 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) + conviction = _band_conviction(close) signal = Signal( strategy=self.name, symbol=candle.symbol, @@ -102,7 +147,7 @@ class VwapStrategy(BaseStrategy): # 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) + conviction = _band_conviction(close) signal = Signal( strategy=self.name, symbol=candle.symbol, |