feat: add 6 new strategies (value, MACD, Bollinger, VWAP, liquidity, MA stack)

shared/strategies/vwap.py

@@ -0,0 +1,91 @@
+"""VWAP crossover strategy — trade on price crossing the Volume Weighted Average Price."""
+
+from datetime import datetime, timezone
+
+from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal
+from shared.strategies.base import BaseStrategy
+
+
+class VWAPStrategy(BaseStrategy):
+    """Generate signals when price crosses above or below VWAP.
+
+    Tracks previous price and VWAP per ticker. The first call for any
+    ticker stores state and returns None.
+
+    **Buy signal** (LONG):
+        Price crosses from below VWAP to above VWAP.
+
+    **Sell signal** (SHORT):
+        Price crosses from above VWAP to below VWAP.
+
+    Signal strength = ``distance_pct * 20 * vol_ratio``, clamped to [0, 1],
+    where ``distance_pct = abs(price - vwap) / vwap`` and ``vol_ratio``
+    is a simple volume multiplier (1.0 by default).
+    """
+
+    name: str = "vwap"
+
+    def __init__(self) -> None:
+        self._prev_price: dict[str, float] = {}
+        self._prev_vwap: dict[str, float] = {}
+
+    async def evaluate(
+        self,
+        ticker: str,
+        market: MarketSnapshot,
+        sentiment: SentimentContext | None = None,
+    ) -> TradeSignal | None:
+        if market.vwap is None:
+            return None
+
+        price = market.current_price
+        vwap = market.vwap
+
+        # First call for this ticker — store state only.
+        if ticker not in self._prev_price:
+            self._prev_price[ticker] = price
+            self._prev_vwap[ticker] = vwap
+            return None
+
+        prev_price = self._prev_price[ticker]
+        prev_vwap = self._prev_vwap[ticker]
+
+        # Update stored state.
+        self._prev_price[ticker] = price
+        self._prev_vwap[ticker] = vwap
+
+        # Detect crossover.
+        prev_above = prev_price > prev_vwap
+        curr_above = price > vwap
+
+        if prev_above == curr_above:
+            # No crossover.
+            return None
+
+        if curr_above:
+            direction = SignalDirection.LONG
+        else:
+            direction = SignalDirection.SHORT
+
+        # Compute strength.
+        distance_pct = abs(price - vwap) / vwap if vwap != 0 else 0.0
+
+        # Volume ratio: use bars average volume if available.
+        vol_ratio = 1.0
+        if market.bars:
+            volumes = [b.get("volume", 0) for b in market.bars if "volume" in b]
+            if volumes:
+                avg_vol = sum(volumes) / len(volumes)
+                if avg_vol > 0:
+                    vol_ratio = market.volume / avg_vol
+
+        raw_strength = distance_pct * 20.0 * vol_ratio
+        strength = max(0.0, min(1.0, raw_strength))
+
+        return TradeSignal(
+            ticker=ticker,
+            direction=direction,
+            strength=strength,
+            strategy_sources=[self.name],
+            timestamp=datetime.now(tz=timezone.utc),
+        )