"""Tests for the MACD strategy.""" from datetime import datetime, timezone from decimal import Decimal from shared.models import Candle, OrderSide from strategies.macd_strategy import MacdStrategy def _candle(price: float) -> Candle: return Candle( symbol="BTC/USDT", timeframe="1m", open_time=datetime(2024, 1, 1, tzinfo=timezone.utc), open=Decimal(str(price)), high=Decimal(str(price)), low=Decimal(str(price)), close=Decimal(str(price)), volume=Decimal("1.0"), ) def _make_strategy(**kwargs) -> MacdStrategy: params = {"fast_period": 3, "slow_period": 6, "signal_period": 3, "quantity": "0.01"} params.update(kwargs) s = MacdStrategy() s.configure(params) return s def test_macd_warmup_period(): s = _make_strategy() assert s.warmup_period == 6 + 3 # slow_period + signal_period def test_macd_no_signal_insufficient_data(): s = _make_strategy() # Feed fewer candles than warmup_period for price in [100.0, 101.0, 102.0]: result = s.on_candle(_candle(price)) assert result is None def test_macd_buy_signal_on_bullish_crossover(): s = _make_strategy() # Declining prices drive MACD histogram negative, then rising prices cross positive prices = [100, 99, 98, 97, 96, 95, 94, 93, 92, 91, 90, 89, 88] prices += [89, 91, 94, 98, 103, 109, 116, 124, 133, 143] signal = None for p in prices: result = s.on_candle(_candle(float(p))) if result is not None: signal = result assert signal is not None, "Expected a BUY signal from bullish crossover" assert signal.side == OrderSide.BUY def test_macd_sell_signal_on_bearish_crossover(): s = _make_strategy() # Rising prices drive MACD histogram positive, then declining prices cross negative prices = [100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112] prices += [111, 109, 106, 102, 97, 91, 84, 76, 67, 57] signal = None for p in prices: result = s.on_candle(_candle(float(p))) if result is not None: signal = result assert signal is not None, "Expected a SELL signal from bearish crossover" assert signal.side == OrderSide.SELL def test_macd_reset_clears_state(): s = _make_strategy() for p in [100, 101, 102, 103, 104, 105, 106, 107, 108]: s.on_candle(_candle(float(p))) assert len(s._closes) > 0 s.reset() assert len(s._closes) == 0 assert s._prev_histogram is None assert s._prev_macd is None assert s._prev_signal is None def test_macd_signal_line_crossover(): """Test that MACD signal-line crossover generates signals.""" s = _make_strategy() # Declining then rising prices should produce a signal-line bullish crossover prices = [100, 99, 98, 97, 96, 95, 94, 93, 92, 91, 90, 89, 88] prices += [89, 91, 94, 98, 103, 109, 116, 124, 133, 143] signals = [] for p in prices: result = s.on_candle(_candle(float(p))) if result is not None: signals.append(result) buy_signals = [sig for sig in signals if sig.side == OrderSide.BUY] assert len(buy_signals) > 0, "Expected at least one BUY signal" # Check that at least one is a signal-line crossover or histogram crossover all_reasons = [sig.reason for sig in buy_signals] assert any("crossover" in r for r in all_reasons), ( f"Expected crossover signal, got: {all_reasons}" ) def test_macd_conviction_varies_with_distance(): """Test that conviction varies based on MACD distance from zero line.""" s1 = _make_strategy() s2 = _make_strategy() # Small price movements -> MACD near zero -> lower conviction small_prices = [100, 99.5, 99, 98.5, 98, 97.5, 97, 96.5, 96, 95.5, 95, 94.5, 94] small_prices += [94.5, 95, 95.5, 96, 96.5, 97, 97.5, 98, 98.5, 99] small_signals = [] for p in small_prices: result = s1.on_candle(_candle(float(p))) if result is not None: small_signals.append(result) # Large price movements -> MACD far from zero -> higher conviction large_prices = [100, 95, 90, 85, 80, 75, 70, 65, 60, 55, 50, 45, 40] large_prices += [45, 55, 70, 90, 115, 145, 180, 220, 265, 315] large_signals = [] for p in large_prices: result = s2.on_candle(_candle(float(p))) if result is not None: large_signals.append(result) # Both should produce signals assert len(small_signals) > 0, "Expected signals from small movements" assert len(large_signals) > 0, "Expected signals from large movements" # The large-movement signals should generally have higher conviction # (or at least different conviction, since distance from zero affects it) small_conv = small_signals[-1].conviction large_conv = large_signals[-1].conviction # Large movements should produce conviction >= small movements assert large_conv >= small_conv, ( f"Expected large movement conviction ({large_conv}) >= small ({small_conv})" )