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
153
154
155
156
157
158
159
160
161
162
163
164
|
from collections import deque
from decimal import Decimal
import pandas as pd
from shared.models import Candle, OrderSide, Signal
from strategies.base import BaseStrategy
class EmaCrossoverStrategy(BaseStrategy):
name: str = "ema_crossover"
def __init__(self) -> None:
super().__init__()
self._closes: deque[float] = deque(maxlen=500)
self._short_period: int = 9
self._long_period: int = 21
self._quantity: Decimal = Decimal("0.01")
self._prev_short_above: bool | None = None
self._pending_signal: str | None = None # "BUY" or "SELL" if waiting for pullback
self._pullback_enabled: bool = True
self._pullback_tolerance: float = 0.002 # 0.2% tolerance around short EMA
@property
def warmup_period(self) -> int:
return self._long_period
def configure(self, params: dict) -> None:
self._short_period = int(params.get("short_period", 9))
self._long_period = int(params.get("long_period", 21))
self._quantity = Decimal(str(params.get("quantity", "0.01")))
self._pullback_enabled = bool(params.get("pullback_enabled", True))
self._pullback_tolerance = float(params.get("pullback_tolerance", 0.002))
if self._short_period >= self._long_period:
raise ValueError(
f"EMA short_period must be < long_period, "
f"got short={self._short_period}, long={self._long_period}"
)
if self._short_period < 2:
raise ValueError(f"EMA short_period must be >= 2, got {self._short_period}")
if self._long_period < 2:
raise ValueError(f"EMA long_period must be >= 2, got {self._long_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:
super().reset()
self._closes.clear()
self._prev_short_above = None
self._pending_signal = None
def _ema_conviction(self, short_ema: float, long_ema: float, price: float) -> float:
"""Map EMA gap to conviction (0.1-1.0). Larger gap = stronger crossover."""
if price == 0:
return 0.5
gap_pct = abs(short_ema - long_ema) / price
# Scale: 0% gap -> 0.1, 1%+ gap -> ~1.0
return min(1.0, max(0.1, gap_pct * 100))
def on_candle(self, candle: Candle) -> Signal | None:
self._update_filter_data(candle)
self._closes.append(float(candle.close))
if len(self._closes) < self._long_period:
return None
series = pd.Series(list(self._closes))
short_ema = series.ewm(span=self._short_period, adjust=False).mean().iloc[-1]
long_ema = series.ewm(span=self._long_period, adjust=False).mean().iloc[-1]
close = float(candle.close)
short_above = short_ema > long_ema
signal = None
if self._prev_short_above is not None:
prev = self._prev_short_above
conviction = self._ema_conviction(short_ema, long_ema, close)
# Golden Cross detected
if not prev and short_above:
if self._pullback_enabled:
self._pending_signal = "BUY"
# Don't signal yet — wait for pullback
else:
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.BUY,
price=candle.close,
quantity=self._quantity,
conviction=conviction,
reason=f"Golden Cross: short EMA ({short_ema:.2f}) crossed above long EMA ({long_ema:.2f})",
)
# Death Cross detected
elif prev and not short_above:
if self._pullback_enabled:
self._pending_signal = "SELL"
else:
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.SELL,
price=candle.close,
quantity=self._quantity,
conviction=conviction,
reason=f"Death Cross: short EMA ({short_ema:.2f}) crossed below long EMA ({long_ema:.2f})",
)
self._prev_short_above = short_above
if signal is not None:
return self._apply_filters(signal)
# Check for pullback entry
if self._pending_signal == "BUY":
distance = abs(close - short_ema) / short_ema if short_ema > 0 else 999
if distance <= self._pullback_tolerance:
self._pending_signal = None
conv = min(0.5 + (1.0 - distance / self._pullback_tolerance) * 0.5, 1.0)
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.BUY,
price=candle.close,
quantity=self._quantity,
conviction=conv,
reason=f"EMA Golden Cross pullback entry (distance={distance:.4f})",
)
return self._apply_filters(signal)
# Cancel if crossover reverses
if not short_above:
self._pending_signal = None
if self._pending_signal == "SELL":
distance = abs(close - short_ema) / short_ema if short_ema > 0 else 999
if distance <= self._pullback_tolerance:
self._pending_signal = None
conv = min(0.5 + (1.0 - distance / self._pullback_tolerance) * 0.5, 1.0)
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.SELL,
price=candle.close,
quantity=self._quantity,
conviction=conv,
reason=f"EMA Death Cross pullback entry (distance={distance:.4f})",
)
return self._apply_filters(signal)
# Cancel if crossover reverses
if short_above:
self._pending_signal = None
return None
|
