diff options
Diffstat (limited to 'services')
3 files changed, 159 insertions, 0 deletions
diff --git a/services/strategy-engine/strategies/config/macd_strategy.yaml b/services/strategy-engine/strategies/config/macd_strategy.yaml new file mode 100644 index 0000000..9d55c39 --- /dev/null +++ b/services/strategy-engine/strategies/config/macd_strategy.yaml @@ -0,0 +1,4 @@ +fast_period: 12 +slow_period: 26 +signal_period: 9 +quantity: "0.01" diff --git a/services/strategy-engine/strategies/macd_strategy.py b/services/strategy-engine/strategies/macd_strategy.py new file mode 100644 index 0000000..049574e --- /dev/null +++ b/services/strategy-engine/strategies/macd_strategy.py @@ -0,0 +1,75 @@ +from collections import deque +from decimal import Decimal + +import pandas as pd + +from shared.models import Candle, Signal, OrderSide +from strategies.base import BaseStrategy + + +class MacdStrategy(BaseStrategy): + name: str = "macd" + + def __init__(self) -> None: + self._fast_period: int = 12 + self._slow_period: int = 26 + self._signal_period: int = 9 + self._quantity: Decimal = Decimal("0.01") + self._closes: deque[float] = deque(maxlen=500) + self._prev_histogram: float | None = None + + @property + def warmup_period(self) -> int: + return self._slow_period + self._signal_period + + def configure(self, params: dict) -> None: + self._fast_period = int(params.get("fast_period", 12)) + self._slow_period = int(params.get("slow_period", 26)) + self._signal_period = int(params.get("signal_period", 9)) + self._quantity = Decimal(str(params.get("quantity", "0.01"))) + + def reset(self) -> None: + self._closes.clear() + self._prev_histogram = None + + def on_candle(self, candle: Candle) -> Signal | None: + self._closes.append(float(candle.close)) + + if len(self._closes) < self.warmup_period: + return None + + series = pd.Series(list(self._closes)) + + fast_ema = series.ewm(span=self._fast_period, adjust=False).mean() + slow_ema = series.ewm(span=self._slow_period, adjust=False).mean() + macd_line = fast_ema - slow_ema + signal_line = macd_line.ewm(span=self._signal_period, adjust=False).mean() + histogram = macd_line - signal_line + + current_histogram = float(histogram.iloc[-1]) + signal = None + + if self._prev_histogram is not None: + # Bullish crossover: histogram crosses from negative to positive + if self._prev_histogram <= 0 and current_histogram > 0: + signal = Signal( + strategy=self.name, + symbol=candle.symbol, + side=OrderSide.BUY, + price=candle.close, + quantity=self._quantity, + reason=f"MACD bullish crossover: histogram {self._prev_histogram:.6f} -> {current_histogram:.6f}", + ) + # Bearish crossover: histogram crosses from positive to negative + elif self._prev_histogram >= 0 and current_histogram < 0: + signal = Signal( + strategy=self.name, + symbol=candle.symbol, + side=OrderSide.SELL, + price=candle.close, + quantity=self._quantity, + reason=f"MACD bearish crossover: histogram {self._prev_histogram:.6f} -> {current_histogram:.6f}", + ) + + self._prev_histogram = current_histogram + return signal diff --git a/services/strategy-engine/tests/test_macd_strategy.py b/services/strategy-engine/tests/test_macd_strategy.py new file mode 100644 index 0000000..e1ae2a3 --- /dev/null +++ b/services/strategy-engine/tests/test_macd_strategy.py @@ -0,0 +1,80 @@ +"""Tests for the MACD strategy.""" +from datetime import datetime, timezone +from decimal import Decimal + +import pytest + +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 |