diff --git a/shared/strategies/__init__.py b/shared/strategies/__init__.py index 0567b2e..96532d2 100644 --- a/shared/strategies/__init__.py +++ b/shared/strategies/__init__.py @@ -10,16 +10,40 @@ MeanReversionStrategy RSI-based mean reversion strategy. NewsDrivenStrategy News sentiment driven strategy. +ValueStrategy + Fundamental valuation strategy. +MACDCrossoverStrategy + MACD / signal line crossover strategy. +BollingerBreakoutStrategy + Bollinger Band breakout strategy. +VWAPStrategy + VWAP crossover strategy. +LiquidityStrategy + Volume anomaly and divergence strategy. +MAStackStrategy + Moving average stack alignment strategy. """ from shared.strategies.base import BaseStrategy +from shared.strategies.bollinger_breakout import BollingerBreakoutStrategy +from shared.strategies.liquidity import LiquidityStrategy +from shared.strategies.macd_crossover import MACDCrossoverStrategy +from shared.strategies.ma_stack import MAStackStrategy from shared.strategies.mean_reversion import MeanReversionStrategy from shared.strategies.momentum import MomentumStrategy from shared.strategies.news_driven import NewsDrivenStrategy +from shared.strategies.value import ValueStrategy +from shared.strategies.vwap import VWAPStrategy __all__ = [ "BaseStrategy", + "BollingerBreakoutStrategy", + "LiquidityStrategy", + "MACDCrossoverStrategy", + "MAStackStrategy", "MeanReversionStrategy", "MomentumStrategy", "NewsDrivenStrategy", + "ValueStrategy", + "VWAPStrategy", ] diff --git a/shared/strategies/bollinger_breakout.py b/shared/strategies/bollinger_breakout.py new file mode 100644 index 0000000..440fd33 --- /dev/null +++ b/shared/strategies/bollinger_breakout.py @@ -0,0 +1,84 @@ +"""Bollinger Band breakout strategy — trade on price breaching Bollinger Bands.""" + +from datetime import datetime, timezone + +from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal +from shared.strategies.base import BaseStrategy + + +class BollingerBreakoutStrategy(BaseStrategy): + """Generate signals when price breaks through Bollinger Bands. + + **Buy signal** (LONG): + - Price > upper band AND volume > 1.5x average volume (momentum + breakout), or + - Price < lower band (mean reversion bounce). + + **Sell signal**: None — this strategy only generates LONG signals. + + Signal strength = distance from the relevant band / band_width, + clamped to [0, 1]. + """ + + name: str = "bollinger_breakout" + + async def evaluate( + self, + ticker: str, + market: MarketSnapshot, + sentiment: SentimentContext | None = None, + ) -> TradeSignal | None: + if ( + market.bollinger_upper is None + or market.bollinger_mid is None + or market.bollinger_lower is None + ): + return None + + price = market.current_price + upper = market.bollinger_upper + lower = market.bollinger_lower + band_width = upper - lower + + if band_width <= 0: + return None + + avg_volume = self._avg_volume(market.bars) + + direction: SignalDirection | None = None + distance: float = 0.0 + + if price > upper and market.volume > 1.5 * avg_volume: + # Momentum breakout above upper band on high volume. + direction = SignalDirection.LONG + distance = price - upper + elif price < lower: + # Mean reversion bounce off lower band. + direction = SignalDirection.LONG + distance = lower - price + else: + return None + + raw_strength = distance / band_width + 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), + ) + + @staticmethod + def _avg_volume(bars: list[dict]) -> float: + """Compute average volume from historical bars. + + Returns a fallback of 0.0 if no bars are available so the volume + comparison always works (volume > 1.5 * 0 is always true when + volume > 0). + """ + if not bars: + return 0.0 + volumes = [b.get("volume", 0) for b in bars if "volume" in b] + return sum(volumes) / len(volumes) if volumes else 0.0 diff --git a/shared/strategies/liquidity.py b/shared/strategies/liquidity.py new file mode 100644 index 0000000..5b2d983 --- /dev/null +++ b/shared/strategies/liquidity.py @@ -0,0 +1,77 @@ +"""Liquidity strategy — trade on volume anomalies and volume-price divergence.""" + +from datetime import datetime, timezone + +from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal +from shared.strategies.base import BaseStrategy + + +class LiquidityStrategy(BaseStrategy): + """Generate signals based on relative volume and volume-price relationships. + + Requires at least 5 bars of historical data to compute average volume. + + **No signal** if relative_volume < 1.0 (thin liquidity). + + **Buy signal** (LONG): + relative_volume >= 2.0 AND price is rising (close > open). + + **Sell signal** (SHORT): + - relative_volume >= 2.0 AND price is falling (close < open), or + - Price is rising on declining volume (relative_volume < 0.7) — + bearish divergence. + + Signal strength = ``relative_volume / 4.0``, clamped to [0, 1]. + """ + + name: str = "liquidity" + + async def evaluate( + self, + ticker: str, + market: MarketSnapshot, + sentiment: SentimentContext | None = None, + ) -> TradeSignal | None: + if len(market.bars) < 5: + return None + + # Compute average volume from bars. + volumes = [b.get("volume", 0) for b in market.bars if "volume" in b] + if not volumes: + return None + avg_volume = sum(volumes) / len(volumes) + if avg_volume <= 0: + return None + + relative_volume = market.volume / avg_volume + + price_rising = market.close > market.open + + direction: SignalDirection | None = None + + # Bearish divergence: price rising on declining volume. + if price_rising and relative_volume < 0.7: + direction = SignalDirection.SHORT + elif relative_volume < 1.0: + # Thin liquidity — no signal. + return None + elif relative_volume >= 2.0: + if price_rising: + direction = SignalDirection.LONG + elif market.close < market.open: + direction = SignalDirection.SHORT + else: + return None + else: + return None + + raw_strength = relative_volume / 4.0 + 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), + ) diff --git a/shared/strategies/ma_stack.py b/shared/strategies/ma_stack.py new file mode 100644 index 0000000..71ea632 --- /dev/null +++ b/shared/strategies/ma_stack.py @@ -0,0 +1,114 @@ +"""Moving Average Stack strategy — trade on alignment of multiple moving averages.""" + +from datetime import datetime, timezone + +from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal +from shared.strategies.base import BaseStrategy + + +class MAStackStrategy(BaseStrategy): + """Generate signals based on moving average alignment (stacking). + + Checks whether the price and four moving averages (EMA-9, EMA-21, + SMA-50, SMA-200) are aligned in bullish or bearish order. Also + detects golden/death cross (SMA-50 vs SMA-200). + + **Buy signal** (LONG): + Bull alignment score >= 3 (at least 3 of 4 ordering conditions met). + + **Sell signal** (SHORT): + Bear alignment score >= 3. + + Signal strength = ``score / 4.0 + cross_bonus``, clamped to [0, 1]. + Cross bonus = 0.15 for golden cross, -0.15 for death cross (applied + only when direction agrees). + """ + + name: str = "ma_stack" + + def __init__(self) -> None: + self._prev_sma_50: dict[str, float] = {} + self._prev_sma_200: dict[str, float] = {} + + async def evaluate( + self, + ticker: str, + market: MarketSnapshot, + sentiment: SentimentContext | None = None, + ) -> TradeSignal | None: + if ( + market.ema_9 is None + or market.ema_21 is None + or market.sma_50 is None + or market.sma_200 is None + ): + return None + + price = market.current_price + ema_9 = market.ema_9 + ema_21 = market.ema_21 + sma_50 = market.sma_50 + sma_200 = market.sma_200 + + # Count bull alignment: price > ema_9 > ema_21 > sma_50 > sma_200 + bull_score = 0 + if price > ema_9: + bull_score += 1 + if ema_9 > ema_21: + bull_score += 1 + if ema_21 > sma_50: + bull_score += 1 + if sma_50 > sma_200: + bull_score += 1 + + # Count bear alignment: price < ema_9 < ema_21 < sma_50 < sma_200 + bear_score = 0 + if price < ema_9: + bear_score += 1 + if ema_9 < ema_21: + bear_score += 1 + if ema_21 < sma_50: + bear_score += 1 + if sma_50 < sma_200: + bear_score += 1 + + # Detect golden/death cross (SMA-50 vs SMA-200). + cross_bonus = 0.0 + if ticker in self._prev_sma_50: + prev_50 = self._prev_sma_50[ticker] + prev_200 = self._prev_sma_200[ticker] + # Golden cross: SMA-50 crosses above SMA-200. + if prev_50 <= prev_200 and sma_50 > sma_200: + cross_bonus = 0.15 + # Death cross: SMA-50 crosses below SMA-200. + elif prev_50 >= prev_200 and sma_50 < sma_200: + cross_bonus = -0.15 + + # Update stored state. + self._prev_sma_50[ticker] = sma_50 + self._prev_sma_200[ticker] = sma_200 + + # Determine direction. + if bull_score >= 3: + direction = SignalDirection.LONG + score = bull_score + # Only apply positive cross bonus for LONG. + bonus = max(0.0, cross_bonus) + elif bear_score >= 3: + direction = SignalDirection.SHORT + score = bear_score + # Only apply negative cross bonus (as positive value) for SHORT. + bonus = abs(min(0.0, cross_bonus)) + else: + return None + + raw_strength = score / 4.0 + bonus + 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), + ) diff --git a/shared/strategies/macd_crossover.py b/shared/strategies/macd_crossover.py new file mode 100644 index 0000000..db38f92 --- /dev/null +++ b/shared/strategies/macd_crossover.py @@ -0,0 +1,82 @@ +"""MACD crossover strategy — trade on MACD/signal line crossovers.""" + +from datetime import datetime, timezone + +from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal +from shared.strategies.base import BaseStrategy + + +class MACDCrossoverStrategy(BaseStrategy): + """Detect MACD / signal line crossovers for entry signals. + + Tracks previous MACD and signal values per ticker. On the first call + for any ticker the strategy stores state and returns None. + + **Buy signal** (LONG): + Bullish crossover — previous ``macd - signal <= 0`` and current + ``macd - signal > 0``. + + **Sell signal** (SHORT): + Bearish crossover — previous ``macd - signal >= 0`` and current + ``macd - signal < 0``. + + Signal strength = ``abs(histogram) / atr`` (or ``abs(histogram) / 2.0`` + when ATR is unavailable), clamped to [0, 1]. + """ + + name: str = "macd_crossover" + + def __init__(self) -> None: + self._prev_macd: dict[str, float] = {} + self._prev_signal: dict[str, float] = {} + + async def evaluate( + self, + ticker: str, + market: MarketSnapshot, + sentiment: SentimentContext | None = None, + ) -> TradeSignal | None: + if market.macd is None or market.macd_signal is None: + return None + + macd = market.macd + signal = market.macd_signal + + # First call for this ticker — store state only. + if ticker not in self._prev_macd: + self._prev_macd[ticker] = macd + self._prev_signal[ticker] = signal + return None + + prev_diff = self._prev_macd[ticker] - self._prev_signal[ticker] + curr_diff = macd - signal + + # Update stored state. + self._prev_macd[ticker] = macd + self._prev_signal[ticker] = signal + + direction: SignalDirection | None = None + + if prev_diff <= 0 and curr_diff > 0: + direction = SignalDirection.LONG + elif prev_diff >= 0 and curr_diff < 0: + direction = SignalDirection.SHORT + else: + return None + + # Compute strength. + histogram = market.macd_histogram if market.macd_histogram is not None else curr_diff + if market.atr is not None and market.atr > 0: + raw_strength = abs(histogram) / market.atr + else: + raw_strength = abs(histogram) / 2.0 + + 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), + ) diff --git a/shared/strategies/value.py b/shared/strategies/value.py new file mode 100644 index 0000000..0109f0b --- /dev/null +++ b/shared/strategies/value.py @@ -0,0 +1,98 @@ +"""Value strategy — trade on fundamental valuation metrics.""" + +from datetime import datetime, timezone + +from shared.schemas.trading import MarketSnapshot, SentimentContext, SignalDirection, TradeSignal +from shared.strategies.base import BaseStrategy + + +class ValueStrategy(BaseStrategy): + """Generate signals from fundamental financial data. + + Computes a composite score from PEG ratio, P/E ratio, EPS, revenue + growth, profit margin, and debt-to-equity ratio. + + **Buy signal** (LONG): + Composite score > 0.3 (undervalued). + + **Sell signal** (SHORT): + Composite score < -0.3 (overvalued). + + Signal strength = ``abs(score) / 2.0``, clamped to [0, 1]. + """ + + name: str = "value" + + async def evaluate( + self, + ticker: str, + market: MarketSnapshot, + sentiment: SentimentContext | None = None, + ) -> TradeSignal | None: + if market.fundamentals is None: + return None + + f = market.fundamentals + + if f.peg_ratio is None or f.pe_ratio is None: + return None + + score = 0.0 + + # PEG ratio scoring + if f.peg_ratio < 1.0: + score += 0.3 + elif f.peg_ratio > 3.0: + score -= 0.3 + + # P/E ratio scoring + if f.pe_ratio < 15: + score += 0.3 + elif f.pe_ratio > 40: + score -= 0.3 + + # EPS scoring + if f.eps_ttm is not None: + if f.eps_ttm > 0: + score += 0.2 + elif f.eps_ttm < 0: + score -= 0.3 + + # Revenue growth scoring + if f.revenue_growth_yoy is not None: + if f.revenue_growth_yoy > 0.1: + score += 0.2 + elif f.revenue_growth_yoy < -0.1: + score -= 0.2 + + # Profit margin scoring + if f.profit_margin is not None: + if f.profit_margin > 0.15: + score += 0.1 + elif f.profit_margin < 0: + score -= 0.2 + + # Debt-to-equity scoring + if f.debt_to_equity is not None: + if f.debt_to_equity > 3.0: + score -= 0.2 + elif f.debt_to_equity < 0.5: + score += 0.1 + + # Determine direction + if score > 0.3: + direction = SignalDirection.LONG + elif score < -0.3: + direction = SignalDirection.SHORT + else: + return None + + strength = max(0.0, min(1.0, abs(score) / 2.0)) + + return TradeSignal( + ticker=ticker, + direction=direction, + strength=strength, + strategy_sources=[self.name], + timestamp=datetime.now(tz=timezone.utc), + ) diff --git a/shared/strategies/vwap.py b/shared/strategies/vwap.py new file mode 100644 index 0000000..df4ec1e --- /dev/null +++ b/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), + ) diff --git a/tests/test_new_strategies.py b/tests/test_new_strategies.py new file mode 100644 index 0000000..40e82d4 --- /dev/null +++ b/tests/test_new_strategies.py @@ -0,0 +1,796 @@ +"""Comprehensive tests for the 6 new trading strategy implementations.""" + +from datetime import datetime, timezone + +import pytest + +from shared.schemas.trading import ( + FundamentalsSnapshot, + MarketSnapshot, + SignalDirection, +) +from shared.strategies import ( + BaseStrategy, + BollingerBreakoutStrategy, + LiquidityStrategy, + MACDCrossoverStrategy, + MAStackStrategy, + ValueStrategy, + VWAPStrategy, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _market( + ticker: str = "AAPL", + price: float = 150.0, + volume: float = 1_000_000, + **kwargs, +) -> MarketSnapshot: + """Build a MarketSnapshot with sensible defaults and optional overrides.""" + defaults = dict( + ticker=ticker, + current_price=price, + open=price - 1, + high=price + 2, + low=price - 2, + close=price, + volume=volume, + ) + defaults.update(kwargs) + return MarketSnapshot(**defaults) + + +def _fundamentals( + ticker: str = "AAPL", + **kwargs, +) -> FundamentalsSnapshot: + """Build a FundamentalsSnapshot with sensible defaults.""" + defaults = dict( + ticker=ticker, + eps_ttm=5.0, + pe_ratio=12.0, + peg_ratio=0.8, + revenue_growth_yoy=0.15, + profit_margin=0.20, + debt_to_equity=0.4, + market_cap=2_000_000_000_000, + fetched_at=datetime.now(timezone.utc), + ) + defaults.update(kwargs) + return FundamentalsSnapshot(**defaults) + + +def _bars(count: int = 10, base_volume: float = 1_000_000) -> list[dict]: + """Generate a list of bar dicts with consistent volume.""" + return [ + { + "timestamp": datetime.now(timezone.utc).isoformat(), + "open": 149.0 + i, + "high": 151.0 + i, + "low": 148.0 + i, + "close": 150.0 + i, + "volume": base_volume, + } + for i in range(count) + ] + + +# =================================================================== +# ValueStrategy +# =================================================================== + + +class TestValueStrategy: + """Tests for :class:`ValueStrategy`.""" + + @pytest.fixture() + def strategy(self) -> ValueStrategy: + return ValueStrategy() + + @pytest.mark.asyncio + async def test_value_long_signal(self, strategy: ValueStrategy) -> None: + """LONG when fundamentals are strongly positive (undervalued).""" + f = _fundamentals( + peg_ratio=0.5, + pe_ratio=10.0, + eps_ttm=5.0, + revenue_growth_yoy=0.2, + profit_margin=0.25, + debt_to_equity=0.3, + ) + market = _market(fundamentals=f) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert signal.ticker == "AAPL" + assert 0 < signal.strength <= 1.0 + assert strategy.name in signal.strategy_sources + + @pytest.mark.asyncio + async def test_value_short_signal(self, strategy: ValueStrategy) -> None: + """SHORT when fundamentals are strongly negative (overvalued).""" + f = _fundamentals( + peg_ratio=4.0, + pe_ratio=50.0, + eps_ttm=-2.0, + revenue_growth_yoy=-0.2, + profit_margin=-0.1, + debt_to_equity=4.0, + ) + market = _market(fundamentals=f) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.SHORT + assert 0 < signal.strength <= 1.0 + + @pytest.mark.asyncio + async def test_value_no_fundamentals(self, strategy: ValueStrategy) -> None: + """Return None when fundamentals are missing.""" + market = _market() + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_value_missing_peg_ratio(self, strategy: ValueStrategy) -> None: + """Return None when peg_ratio is None.""" + f = _fundamentals(peg_ratio=None) + market = _market(fundamentals=f) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_value_missing_pe_ratio(self, strategy: ValueStrategy) -> None: + """Return None when pe_ratio is None.""" + f = _fundamentals(pe_ratio=None) + market = _market(fundamentals=f) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_value_neutral_score(self, strategy: ValueStrategy) -> None: + """No signal when fundamentals are mediocre (score near 0).""" + f = _fundamentals( + peg_ratio=2.0, # neutral (between 1 and 3) + pe_ratio=25.0, # neutral (between 15 and 40) + eps_ttm=None, + revenue_growth_yoy=None, + profit_margin=None, + debt_to_equity=None, + ) + market = _market(fundamentals=f) + signal = await strategy.evaluate("AAPL", market) + assert signal is None + + @pytest.mark.asyncio + async def test_value_strength_clamped(self, strategy: ValueStrategy) -> None: + """Strength must be within [0, 1].""" + # Maximum positive score scenario. + f = _fundamentals( + peg_ratio=0.3, + pe_ratio=8.0, + eps_ttm=10.0, + revenue_growth_yoy=0.5, + profit_margin=0.4, + debt_to_equity=0.2, + ) + market = _market(fundamentals=f) + signal = await strategy.evaluate("AAPL", market) + assert signal is not None + assert 0.0 <= signal.strength <= 1.0 + + +# =================================================================== +# MACDCrossoverStrategy +# =================================================================== + + +class TestMACDCrossoverStrategy: + """Tests for :class:`MACDCrossoverStrategy`.""" + + @pytest.fixture() + def strategy(self) -> MACDCrossoverStrategy: + return MACDCrossoverStrategy() + + @pytest.mark.asyncio + async def test_macd_bullish_crossover(self, strategy: MACDCrossoverStrategy) -> None: + """LONG on bullish crossover (MACD crosses above signal).""" + # First call: MACD below signal. + market1 = _market(macd=-1.0, macd_signal=0.5, macd_histogram=-1.5, atr=2.0) + result1 = await strategy.evaluate("AAPL", market1) + assert result1 is None # First call stores state. + + # Second call: MACD above signal (crossover). + market2 = _market(macd=1.0, macd_signal=0.5, macd_histogram=0.5, atr=2.0) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert 0 < signal.strength <= 1.0 + assert strategy.name in signal.strategy_sources + + @pytest.mark.asyncio + async def test_macd_bearish_crossover(self, strategy: MACDCrossoverStrategy) -> None: + """SHORT on bearish crossover (MACD crosses below signal).""" + # First call: MACD above signal. + market1 = _market(macd=1.0, macd_signal=0.5, macd_histogram=0.5, atr=2.0) + await strategy.evaluate("AAPL", market1) + + # Second call: MACD below signal (crossover). + market2 = _market(macd=-1.0, macd_signal=0.5, macd_histogram=-1.5, atr=2.0) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.direction == SignalDirection.SHORT + assert 0 < signal.strength <= 1.0 + + @pytest.mark.asyncio + async def test_macd_no_crossover(self, strategy: MACDCrossoverStrategy) -> None: + """No signal when MACD stays on the same side of signal.""" + # Both calls have MACD above signal. + market1 = _market(macd=2.0, macd_signal=0.5, macd_histogram=1.5) + await strategy.evaluate("AAPL", market1) + + market2 = _market(macd=3.0, macd_signal=0.5, macd_histogram=2.5) + signal = await strategy.evaluate("AAPL", market2) + assert signal is None + + @pytest.mark.asyncio + async def test_macd_first_call_returns_none(self, strategy: MACDCrossoverStrategy) -> None: + """First call for a ticker always returns None (storing state).""" + market = _market(macd=1.0, macd_signal=0.5, macd_histogram=0.5) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_macd_missing_data(self, strategy: MACDCrossoverStrategy) -> None: + """Return None when MACD or signal is missing.""" + market_no_macd = _market(macd=None, macd_signal=0.5) + assert await strategy.evaluate("AAPL", market_no_macd) is None + + market_no_signal = _market(macd=1.0, macd_signal=None) + assert await strategy.evaluate("AAPL", market_no_signal) is None + + @pytest.mark.asyncio + async def test_macd_strength_with_atr(self, strategy: MACDCrossoverStrategy) -> None: + """Strength = abs(histogram) / atr when atr is available.""" + market1 = _market(macd=-1.0, macd_signal=0.5, macd_histogram=-1.5, atr=2.0) + await strategy.evaluate("AAPL", market1) + + market2 = _market(macd=1.0, macd_signal=0.5, macd_histogram=0.5, atr=2.0) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.strength == pytest.approx(0.5 / 2.0, abs=1e-9) + + @pytest.mark.asyncio + async def test_macd_strength_without_atr(self, strategy: MACDCrossoverStrategy) -> None: + """Strength = abs(histogram) / 2.0 when atr is None.""" + market1 = _market(macd=-1.0, macd_signal=0.5, macd_histogram=-1.5, atr=None) + await strategy.evaluate("AAPL", market1) + + market2 = _market(macd=1.0, macd_signal=0.5, macd_histogram=0.8, atr=None) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.strength == pytest.approx(0.8 / 2.0, abs=1e-9) + + @pytest.mark.asyncio + async def test_macd_strength_clamped(self, strategy: MACDCrossoverStrategy) -> None: + """Strength should be clamped to [0, 1].""" + market1 = _market(macd=-1.0, macd_signal=0.5, macd_histogram=-1.5, atr=0.1) + await strategy.evaluate("AAPL", market1) + + # Large histogram with small ATR -> raw strength > 1. + market2 = _market(macd=5.0, macd_signal=0.5, macd_histogram=4.5, atr=0.1) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.strength == 1.0 + + +# =================================================================== +# BollingerBreakoutStrategy +# =================================================================== + + +class TestBollingerBreakoutStrategy: + """Tests for :class:`BollingerBreakoutStrategy`.""" + + @pytest.fixture() + def strategy(self) -> BollingerBreakoutStrategy: + return BollingerBreakoutStrategy() + + @pytest.mark.asyncio + async def test_bollinger_upper_breakout(self, strategy: BollingerBreakoutStrategy) -> None: + """LONG when price > upper band on high volume.""" + bars = _bars(count=10, base_volume=500_000) + market = _market( + price=160.0, + volume=1_000_000, # > 1.5 * 500_000 + bollinger_upper=155.0, + bollinger_mid=150.0, + bollinger_lower=145.0, + bars=bars, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert 0 < signal.strength <= 1.0 + assert strategy.name in signal.strategy_sources + + @pytest.mark.asyncio + async def test_bollinger_lower_bounce(self, strategy: BollingerBreakoutStrategy) -> None: + """LONG when price < lower band (mean reversion bounce).""" + market = _market( + price=140.0, + bollinger_upper=155.0, + bollinger_mid=150.0, + bollinger_lower=145.0, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert 0 < signal.strength <= 1.0 + + @pytest.mark.asyncio + async def test_bollinger_no_signal_between_bands(self, strategy: BollingerBreakoutStrategy) -> None: + """No signal when price is between the bands.""" + market = _market( + price=150.0, + bollinger_upper=155.0, + bollinger_mid=150.0, + bollinger_lower=145.0, + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is None + + @pytest.mark.asyncio + async def test_bollinger_upper_low_volume(self, strategy: BollingerBreakoutStrategy) -> None: + """No signal when price > upper but volume is too low.""" + bars = _bars(count=10, base_volume=1_000_000) + market = _market( + price=160.0, + volume=1_000_000, # Not > 1.5 * 1_000_000 + bollinger_upper=155.0, + bollinger_mid=150.0, + bollinger_lower=145.0, + bars=bars, + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is None + + @pytest.mark.asyncio + async def test_bollinger_missing_bands(self, strategy: BollingerBreakoutStrategy) -> None: + """Return None when any Bollinger band is missing.""" + market = _market(bollinger_upper=None, bollinger_mid=150.0, bollinger_lower=145.0) + assert await strategy.evaluate("AAPL", market) is None + + market = _market(bollinger_upper=155.0, bollinger_mid=None, bollinger_lower=145.0) + assert await strategy.evaluate("AAPL", market) is None + + market = _market(bollinger_upper=155.0, bollinger_mid=150.0, bollinger_lower=None) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_bollinger_strength_proportional(self, strategy: BollingerBreakoutStrategy) -> None: + """Strength is proportional to distance from band / band_width.""" + # Price well below lower band. + market = _market( + price=140.0, + bollinger_upper=155.0, + bollinger_mid=150.0, + bollinger_lower=145.0, + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is not None + expected = (145.0 - 140.0) / (155.0 - 145.0) # 5 / 10 = 0.5 + assert signal.strength == pytest.approx(expected, abs=1e-9) + + @pytest.mark.asyncio + async def test_bollinger_strength_clamped(self, strategy: BollingerBreakoutStrategy) -> None: + """Strength must not exceed 1.0.""" + # Price far below lower band. + market = _market( + price=120.0, + bollinger_upper=155.0, + bollinger_mid=150.0, + bollinger_lower=145.0, + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is not None + assert signal.strength == 1.0 + + +# =================================================================== +# VWAPStrategy +# =================================================================== + + +class TestVWAPStrategy: + """Tests for :class:`VWAPStrategy`.""" + + @pytest.fixture() + def strategy(self) -> VWAPStrategy: + return VWAPStrategy() + + @pytest.mark.asyncio + async def test_vwap_long_crossover(self, strategy: VWAPStrategy) -> None: + """LONG when price crosses from below VWAP to above VWAP.""" + # First call: price below VWAP. + market1 = _market(price=148.0, vwap=150.0) + result1 = await strategy.evaluate("AAPL", market1) + assert result1 is None + + # Second call: price above VWAP. + market2 = _market(price=152.0, vwap=150.0) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert 0 < signal.strength <= 1.0 + assert strategy.name in signal.strategy_sources + + @pytest.mark.asyncio + async def test_vwap_short_crossover(self, strategy: VWAPStrategy) -> None: + """SHORT when price crosses from above VWAP to below VWAP.""" + # First call: price above VWAP. + market1 = _market(price=152.0, vwap=150.0) + await strategy.evaluate("AAPL", market1) + + # Second call: price below VWAP. + market2 = _market(price=148.0, vwap=150.0) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.direction == SignalDirection.SHORT + assert 0 < signal.strength <= 1.0 + + @pytest.mark.asyncio + async def test_vwap_no_crossover(self, strategy: VWAPStrategy) -> None: + """No signal when price stays on the same side of VWAP.""" + # Both calls: price above VWAP. + market1 = _market(price=152.0, vwap=150.0) + await strategy.evaluate("AAPL", market1) + + market2 = _market(price=155.0, vwap=150.0) + signal = await strategy.evaluate("AAPL", market2) + assert signal is None + + @pytest.mark.asyncio + async def test_vwap_first_call_returns_none(self, strategy: VWAPStrategy) -> None: + """First call for a ticker always returns None.""" + market = _market(vwap=150.0) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_vwap_missing_vwap(self, strategy: VWAPStrategy) -> None: + """Return None when VWAP is missing.""" + market = _market(vwap=None) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_vwap_strength_bounds(self, strategy: VWAPStrategy) -> None: + """Strength should be within [0, 1].""" + market1 = _market(price=100.0, vwap=150.0) + await strategy.evaluate("AAPL", market1) + + market2 = _market(price=200.0, vwap=150.0) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert 0.0 <= signal.strength <= 1.0 + + +# =================================================================== +# LiquidityStrategy +# =================================================================== + + +class TestLiquidityStrategy: + """Tests for :class:`LiquidityStrategy`.""" + + @pytest.fixture() + def strategy(self) -> LiquidityStrategy: + return LiquidityStrategy() + + @pytest.mark.asyncio + async def test_liquidity_long_high_volume_rising(self, strategy: LiquidityStrategy) -> None: + """LONG when relative_volume >= 2.0 and price is rising.""" + bars = _bars(count=10, base_volume=500_000) + market = _market( + price=152.0, + volume=1_200_000, # relative_volume = 1_200_000 / 500_000 = 2.4 + bars=bars, + **{"open": 148.0, "close": 152.0}, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert 0 < signal.strength <= 1.0 + assert strategy.name in signal.strategy_sources + + @pytest.mark.asyncio + async def test_liquidity_short_high_volume_falling(self, strategy: LiquidityStrategy) -> None: + """SHORT when relative_volume >= 2.0 and price is falling.""" + bars = _bars(count=10, base_volume=500_000) + market = _market( + price=148.0, + volume=1_200_000, + bars=bars, + **{"open": 152.0, "close": 148.0}, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.SHORT + assert 0 < signal.strength <= 1.0 + + @pytest.mark.asyncio + async def test_liquidity_short_divergence(self, strategy: LiquidityStrategy) -> None: + """SHORT on divergence: price rising on declining volume.""" + bars = _bars(count=10, base_volume=1_000_000) + market = _market( + price=152.0, + volume=600_000, # relative_volume = 0.6 < 0.7 + bars=bars, + **{"open": 148.0, "close": 152.0}, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.SHORT + + @pytest.mark.asyncio + async def test_liquidity_no_signal_thin(self, strategy: LiquidityStrategy) -> None: + """No signal when relative_volume < 1.0 (thin, non-divergent).""" + bars = _bars(count=10, base_volume=1_000_000) + market = _market( + price=148.0, + volume=800_000, # relative_volume = 0.8; not rising so no divergence + bars=bars, + **{"open": 152.0, "close": 148.0}, + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is None + + @pytest.mark.asyncio + async def test_liquidity_insufficient_bars(self, strategy: LiquidityStrategy) -> None: + """Return None when fewer than 5 bars are available.""" + bars = _bars(count=3) + market = _market(volume=2_000_000, bars=bars) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_liquidity_no_bars(self, strategy: LiquidityStrategy) -> None: + """Return None when bars are empty.""" + market = _market(volume=2_000_000, bars=[]) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_liquidity_strength_bounds(self, strategy: LiquidityStrategy) -> None: + """Strength should be within [0, 1].""" + bars = _bars(count=10, base_volume=100_000) + market = _market( + price=155.0, + volume=1_000_000, # relative_volume = 10.0 + bars=bars, + **{"open": 148.0, "close": 155.0}, + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is not None + assert 0.0 <= signal.strength <= 1.0 + + +# =================================================================== +# MAStackStrategy +# =================================================================== + + +class TestMAStackStrategy: + """Tests for :class:`MAStackStrategy`.""" + + @pytest.fixture() + def strategy(self) -> MAStackStrategy: + return MAStackStrategy() + + @pytest.mark.asyncio + async def test_ma_stack_bullish(self, strategy: MAStackStrategy) -> None: + """LONG when MAs are in full bullish alignment.""" + market = _market( + price=200.0, + ema_9=195.0, + ema_21=190.0, + sma_50=180.0, + sma_200=170.0, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert 0 < signal.strength <= 1.0 + assert strategy.name in signal.strategy_sources + + @pytest.mark.asyncio + async def test_ma_stack_bearish(self, strategy: MAStackStrategy) -> None: + """SHORT when MAs are in full bearish alignment.""" + market = _market( + price=140.0, + ema_9=145.0, + ema_21=150.0, + sma_50=160.0, + sma_200=170.0, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.direction == SignalDirection.SHORT + assert 0 < signal.strength <= 1.0 + + @pytest.mark.asyncio + async def test_ma_stack_neutral(self, strategy: MAStackStrategy) -> None: + """No signal when MAs are not sufficiently aligned.""" + # Mixed alignment: only 2 bull conditions met. + market = _market( + price=155.0, + ema_9=153.0, + ema_21=157.0, # ema_9 < ema_21, breaks bull + sma_50=150.0, + sma_200=160.0, # sma_50 < sma_200, breaks bull + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is None + + @pytest.mark.asyncio + async def test_ma_stack_missing_ma(self, strategy: MAStackStrategy) -> None: + """Return None when any required MA is missing.""" + base = dict(ema_9=195.0, ema_21=190.0, sma_50=180.0, sma_200=170.0) + + for key in ("ema_9", "ema_21", "sma_50", "sma_200"): + overrides = {**base, key: None} + market = _market(**overrides) + assert await strategy.evaluate("AAPL", market) is None + + @pytest.mark.asyncio + async def test_ma_stack_golden_cross_bonus(self, strategy: MAStackStrategy) -> None: + """Golden cross adds bonus to bullish signal strength.""" + # First call: SMA-50 below SMA-200. + market1 = _market( + price=200.0, + ema_9=195.0, + ema_21=190.0, + sma_50=169.0, # below sma_200 + sma_200=170.0, + ) + await strategy.evaluate("AAPL", market1) + + # Second call: SMA-50 above SMA-200 (golden cross). + market2 = _market( + price=200.0, + ema_9=195.0, + ema_21=190.0, + sma_50=171.0, # above sma_200 + sma_200=170.0, + ) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.direction == SignalDirection.LONG + # With golden cross, strength = 4/4 + 0.15 = 1.15, clamped to 1.0 + assert signal.strength == 1.0 + + @pytest.mark.asyncio + async def test_ma_stack_death_cross_bonus(self, strategy: MAStackStrategy) -> None: + """Death cross adds bonus to bearish signal strength.""" + # First call: SMA-50 above SMA-200. + market1 = _market( + price=140.0, + ema_9=145.0, + ema_21=150.0, + sma_50=171.0, # above sma_200 + sma_200=170.0, + ) + await strategy.evaluate("AAPL", market1) + + # Second call: SMA-50 below SMA-200 (death cross). + market2 = _market( + price=140.0, + ema_9=145.0, + ema_21=150.0, + sma_50=169.0, # below sma_200 + sma_200=170.0, + ) + signal = await strategy.evaluate("AAPL", market2) + + assert signal is not None + assert signal.direction == SignalDirection.SHORT + # With death cross, strength = 4/4 + 0.15 = 1.15, clamped to 1.0 + assert signal.strength == 1.0 + + @pytest.mark.asyncio + async def test_ma_stack_strength_without_cross(self, strategy: MAStackStrategy) -> None: + """Strength = score/4 without cross bonus.""" + # Full bull alignment, no cross (first call). + market = _market( + price=200.0, + ema_9=195.0, + ema_21=190.0, + sma_50=180.0, + sma_200=170.0, + ) + signal = await strategy.evaluate("AAPL", market) + + assert signal is not None + assert signal.strength == pytest.approx(4.0 / 4.0, abs=1e-9) + + @pytest.mark.asyncio + async def test_ma_stack_partial_bull(self, strategy: MAStackStrategy) -> None: + """LONG with 3 of 4 bull conditions met, strength = 3/4.""" + # price > ema_9 > ema_21, ema_21 < sma_50 (breaks one), sma_50 > sma_200 + market = _market( + price=200.0, + ema_9=195.0, + ema_21=185.0, + sma_50=190.0, # ema_21 < sma_50, so that condition fails + sma_200=170.0, + ) + signal = await strategy.evaluate("AAPL", market) + assert signal is not None + assert signal.direction == SignalDirection.LONG + assert signal.strength == pytest.approx(3.0 / 4.0, abs=1e-9) + + +# =================================================================== +# Cross-strategy tests (new strategies) +# =================================================================== + + +class TestNewStrategyCrossChecks: + """Tests that apply across all new strategy implementations.""" + + ALL_NEW = ( + ValueStrategy, + MACDCrossoverStrategy, + BollingerBreakoutStrategy, + VWAPStrategy, + LiquidityStrategy, + MAStackStrategy, + ) + + def test_all_new_strategies_are_base_strategy_subclass(self) -> None: + """All new concrete strategies must inherit from BaseStrategy.""" + for cls in self.ALL_NEW: + assert issubclass(cls, BaseStrategy), f"{cls.__name__} is not a BaseStrategy subclass" + + def test_new_strategy_names_unique(self) -> None: + """Every new strategy must have a distinct name.""" + strategies = [cls() for cls in self.ALL_NEW] + names = [s.name for s in strategies] + assert len(names) == len(set(names)), f"Duplicate strategy names detected: {names}" + + def test_new_strategy_names_non_empty(self) -> None: + """Every new strategy name must be a non-empty string.""" + for cls in self.ALL_NEW: + instance = cls() + assert isinstance(instance.name, str) + assert len(instance.name) > 0 + + def test_no_name_clash_with_existing(self) -> None: + """New strategy names must not clash with existing strategies.""" + from shared.strategies import MeanReversionStrategy, MomentumStrategy, NewsDrivenStrategy + + existing_names = { + MomentumStrategy().name, + MeanReversionStrategy().name, + NewsDrivenStrategy().name, + } + for cls in self.ALL_NEW: + instance = cls() + assert instance.name not in existing_names, ( + f"{cls.__name__}.name '{instance.name}' clashes with an existing strategy" + )