"""Tests for Combined strategy.""" import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from decimal import Decimal from datetime import datetime, timezone import pytest from shared.models import Candle, Signal, OrderSide from strategies.combined_strategy import CombinedStrategy from strategies.base import BaseStrategy class AlwaysBuyStrategy(BaseStrategy): name = "always_buy" @property def warmup_period(self) -> int: return 0 def configure(self, params: dict) -> None: pass def on_candle(self, candle: Candle) -> Signal | None: return Signal( strategy=self.name, symbol=candle.symbol, side=OrderSide.BUY, price=candle.close, quantity=Decimal("0.01"), reason="always buy", ) class AlwaysSellStrategy(BaseStrategy): name = "always_sell" @property def warmup_period(self) -> int: return 0 def configure(self, params: dict) -> None: pass def on_candle(self, candle: Candle) -> Signal | None: return Signal( strategy=self.name, symbol=candle.symbol, side=OrderSide.SELL, price=candle.close, quantity=Decimal("0.01"), reason="always sell", ) class NeutralStrategy(BaseStrategy): name = "neutral" @property def warmup_period(self) -> int: return 0 def configure(self, params: dict) -> None: pass def on_candle(self, candle: Candle) -> Signal | None: return None def _candle(price=100.0): return Candle( symbol="BTCUSDT", timeframe="1m", open_time=datetime(2025, 1, 1, tzinfo=timezone.utc), open=Decimal(str(price)), high=Decimal(str(price+10)), low=Decimal(str(price-10)), close=Decimal(str(price)), volume=Decimal("10"), ) def test_combined_no_strategies(): c = CombinedStrategy() c.configure({"threshold": 0.5}) assert c.on_candle(_candle()) is None def test_combined_unanimous_buy(): c = CombinedStrategy() c.configure({"threshold": 0.5}) c.add_strategy(AlwaysBuyStrategy(), weight=1.0) c.add_strategy(AlwaysBuyStrategy(), weight=1.0) sig = c.on_candle(_candle()) assert sig is not None assert sig.side == OrderSide.BUY def test_combined_unanimous_sell(): c = CombinedStrategy() c.configure({"threshold": 0.5}) c.add_strategy(AlwaysSellStrategy(), weight=1.0) c.add_strategy(AlwaysSellStrategy(), weight=1.0) sig = c.on_candle(_candle()) assert sig is not None assert sig.side == OrderSide.SELL def test_combined_conflicting_signals_cancel(): c = CombinedStrategy() c.configure({"threshold": 0.5}) c.add_strategy(AlwaysBuyStrategy(), weight=1.0) c.add_strategy(AlwaysSellStrategy(), weight=1.0) sig = c.on_candle(_candle()) assert sig is None # Score = 0, below threshold def test_combined_weighted_buy(): c = CombinedStrategy() c.configure({"threshold": 0.3}) c.add_strategy(AlwaysBuyStrategy(), weight=3.0) c.add_strategy(AlwaysSellStrategy(), weight=1.0) sig = c.on_candle(_candle()) assert sig is not None assert sig.side == OrderSide.BUY # Score = (3-1)/4 = 0.5 >= 0.3 def test_combined_neutral_doesnt_affect_score(): c = CombinedStrategy() c.configure({"threshold": 0.5}) c.add_strategy(AlwaysBuyStrategy(), weight=1.0) c.add_strategy(NeutralStrategy(), weight=1.0) sig = c.on_candle(_candle()) assert sig is not None assert sig.side == OrderSide.BUY # Score = 1/2 = 0.5 >= 0.5 def test_combined_warmup_is_max(): c = CombinedStrategy() c.configure({}) s1 = AlwaysBuyStrategy() s2 = NeutralStrategy() c.add_strategy(s1) c.add_strategy(s2) assert c.warmup_period == 0 def test_combined_reset_resets_all(): c = CombinedStrategy() c.configure({}) c.add_strategy(AlwaysBuyStrategy()) c.on_candle(_candle()) c.reset() # Should not crash def test_combined_invalid_threshold(): c = CombinedStrategy() with pytest.raises(ValueError): c.configure({"threshold": -1}) def test_combined_invalid_weight(): c = CombinedStrategy() c.configure({}) with pytest.raises(ValueError): c.add_strategy(AlwaysBuyStrategy(), weight=-1.0)