796 lines
29 KiB
Python
796 lines
29 KiB
Python
"""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"
|
|
)
|