91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""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),
|
|
)
|