"""Tests for the EMA Crossover strategy.""" from datetime import datetime, timezone from decimal import Decimal from shared.models import Candle, OrderSide from strategies.ema_crossover_strategy import EmaCrossoverStrategy def make_candle(close: float) -> Candle: return Candle( symbol="BTC/USDT", timeframe="1m", open_time=datetime(2024, 1, 1, tzinfo=timezone.utc), open=Decimal(str(close)), high=Decimal(str(close)), low=Decimal(str(close)), close=Decimal(str(close)), volume=Decimal("1.0"), ) def _make_strategy(short: int = 3, long: int = 6, pullback_enabled: bool = False) -> EmaCrossoverStrategy: s = EmaCrossoverStrategy() s.configure({"short_period": short, "long_period": long, "quantity": "0.01", "pullback_enabled": pullback_enabled}) return s def test_ema_warmup_period(): strategy = _make_strategy(short=3, long=6) assert strategy.warmup_period == 6 def test_ema_no_signal_insufficient_data(): strategy = _make_strategy(short=3, long=6) # Feed fewer candles than warmup_period for price in [100, 101, 102, 103, 104]: result = strategy.on_candle(make_candle(price)) assert result is None def test_ema_buy_signal_golden_cross(): strategy = _make_strategy(short=3, long=6) # Declining prices so short EMA stays below long EMA declining = [100, 98, 96, 94, 92, 90, 88, 86, 84, 82] for price in declining: strategy.on_candle(make_candle(price)) # Sharp rise to force short EMA above long EMA (golden cross) rising = [120, 140, 160] signal = None for price in rising: result = strategy.on_candle(make_candle(price)) if result is not None: signal = result assert signal is not None assert signal.side == OrderSide.BUY assert "Golden Cross" in signal.reason def test_ema_sell_signal_death_cross(): strategy = _make_strategy(short=3, long=6) # Rising prices so short EMA stays above long EMA rising = [80, 82, 84, 86, 88, 90, 92, 94, 96, 98] for price in rising: strategy.on_candle(make_candle(price)) # Sharp decline to force short EMA below long EMA (death cross) declining = [60, 40, 20] signal = None for price in declining: result = strategy.on_candle(make_candle(price)) if result is not None: signal = result assert signal is not None assert signal.side == OrderSide.SELL assert "Death Cross" in signal.reason def test_ema_reset_clears_state(): strategy = _make_strategy(short=3, long=6) # Feed enough data to produce a signal for price in [100, 98, 96, 94, 92, 90, 88, 86, 84, 82]: strategy.on_candle(make_candle(price)) strategy.reset() # After reset, insufficient data again — no signal result = strategy.on_candle(make_candle(100)) assert result is None # Internal state should be cleared assert len(strategy._closes) == 1 assert strategy._prev_short_above is None assert strategy._pending_signal is None def test_ema_pullback_entry(): """Crossover detected, then pullback to short EMA triggers signal.""" strategy = EmaCrossoverStrategy() strategy.configure({ "short_period": 3, "long_period": 6, "quantity": "0.01", "pullback_enabled": True, "pullback_tolerance": 0.05, # 5% tolerance for test simplicity }) # Declining prices so short EMA stays below long EMA declining = [100, 98, 96, 94, 92, 90, 88, 86, 84, 82] for price in declining: strategy.on_candle(make_candle(price)) # Sharp rise to force golden cross — with pullback enabled, no signal yet rising = [120, 140, 160] for price in rising: strategy.on_candle(make_candle(price)) # With pullback enabled, crossover should NOT produce immediate signal # but _pending_signal should be set assert strategy._pending_signal == "BUY" # Now feed a candle whose close is near the short EMA (pullback) # The short EMA will be tracking recent prices; feed a price that pulls back # toward it. We use a moderate price to get close to short EMA. import pandas as pd series = pd.Series(list(strategy._closes)) short_ema_val = series.ewm(span=3, adjust=False).mean().iloc[-1] # Feed a candle at approximately the short EMA value result = strategy.on_candle(make_candle(short_ema_val)) assert result is not None assert result.side == OrderSide.BUY assert "pullback" in result.reason def test_ema_pullback_cancelled_on_reversal(): """Crossover detected, then reversal cancels the pending signal.""" strategy = EmaCrossoverStrategy() strategy.configure({ "short_period": 3, "long_period": 6, "quantity": "0.01", "pullback_enabled": True, "pullback_tolerance": 0.001, # Very tight tolerance — won't trigger easily }) # Declining prices declining = [100, 98, 96, 94, 92, 90, 88, 86, 84, 82] for price in declining: strategy.on_candle(make_candle(price)) # Sharp rise to force golden cross for price in [120, 140, 160]: strategy.on_candle(make_candle(price)) assert strategy._pending_signal == "BUY" # Now sharp decline to reverse the crossover (death cross) for price in [60, 40, 20]: strategy.on_candle(make_candle(price)) # The BUY pending signal should be cancelled because short EMA fell below long EMA. # A new death cross may set _pending_signal to "SELL", but the original "BUY" is gone. assert strategy._pending_signal != "BUY" def test_ema_immediate_mode(): """With pullback_enabled=False, original immediate entry works.""" strategy = EmaCrossoverStrategy() strategy.configure({ "short_period": 3, "long_period": 6, "quantity": "0.01", "pullback_enabled": False, }) # Declining prices so short EMA stays below long EMA declining = [100, 98, 96, 94, 92, 90, 88, 86, 84, 82] for price in declining: strategy.on_candle(make_candle(price)) # Sharp rise to force golden cross — immediate mode should fire signal rising = [120, 140, 160] signal = None for price in rising: result = strategy.on_candle(make_candle(price)) if result is not None: signal = result assert signal is not None assert signal.side == OrderSide.BUY assert "Golden Cross" in signal.reason # No pending signal should be set assert strategy._pending_signal is None