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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
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
self._prev_macd: float | None = None
self._prev_signal: 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}")
self._init_filters(
require_trend=True,
adx_threshold=float(params.get("adx_threshold", 25.0)),
min_volume_ratio=float(params.get("min_volume_ratio", 0.5)),
atr_stop_multiplier=float(params.get("atr_stop_multiplier", 2.0)),
atr_tp_multiplier=float(params.get("atr_tp_multiplier", 3.0)),
)
def reset(self) -> None:
self._closes.clear()
self._prev_histogram = None
self._prev_macd = None
self._prev_signal = None
def _macd_conviction(self, histogram_value: float, price: float) -> float:
"""Map histogram magnitude to conviction (0.1-1.0).
Normalize by price to make it scale-independent.
"""
if price == 0:
return 0.5
normalized = abs(histogram_value) / price * 1000 # scale to reasonable range
return min(1.0, max(0.1, normalized))
def on_candle(self, candle: Candle) -> Signal | None:
self._update_filter_data(candle)
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])
macd_val = float(macd_line.iloc[-1])
signal_val = float(signal_line.iloc[-1])
result_signal = None
# Signal-line crossover detection (MACD crosses signal line directly)
if self._prev_macd is not None and self._prev_signal is not None:
# Bullish: MACD crosses above signal
if self._prev_macd <= self._prev_signal and macd_val > signal_val:
distance_from_zero = abs(macd_val) / float(candle.close) * 1000
conv = min(max(distance_from_zero, 0.3), 1.0)
result_signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.BUY,
price=candle.close,
quantity=self._quantity,
conviction=conv,
reason="MACD signal-line bullish crossover",
)
# Bearish: MACD crosses below signal
elif self._prev_macd >= self._prev_signal and macd_val < signal_val:
distance_from_zero = abs(macd_val) / float(candle.close) * 1000
conv = min(max(distance_from_zero, 0.3), 1.0)
result_signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.SELL,
price=candle.close,
quantity=self._quantity,
conviction=conv,
reason="MACD signal-line bearish crossover",
)
# Histogram crossover detection (existing logic, as secondary signal)
if result_signal is None and self._prev_histogram is not None:
conviction = self._macd_conviction(current_histogram, float(candle.close))
# Bullish crossover: histogram crosses from negative to positive
if self._prev_histogram <= 0 and current_histogram > 0:
result_signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.BUY,
price=candle.close,
quantity=self._quantity,
conviction=conviction,
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:
result_signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.SELL,
price=candle.close,
quantity=self._quantity,
conviction=conviction,
reason=f"MACD bearish crossover: histogram {self._prev_histogram:.6f} -> {current_histogram:.6f}",
)
self._prev_histogram = current_histogram
self._prev_macd = macd_val
self._prev_signal = signal_val
if result_signal is not None:
return self._apply_filters(result_signal)
return None
|
