1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|
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:
super().__init__()
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")))
if self._fast_period >= self._slow_period:
raise ValueError(
f"MACD fast_period must be < slow_period, "
f"got fast={self._fast_period}, slow={self._slow_period}"
)
if self._fast_period < 2:
raise ValueError(f"MACD fast_period must be >= 2, got {self._fast_period}")
if self._slow_period < 2:
raise ValueError(f"MACD slow_period must be >= 2, got {self._slow_period}")
if self._signal_period < 2:
raise ValueError(f"MACD signal_period must be >= 2, got {self._signal_period}")
if self._quantity <= 0:
raise ValueError(f"Quantity must be positive, got {self._quantity}")
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
|
