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
|
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 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
@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")))
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:
self._closes.clear()
self._prev_short_above = 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]
short_above = short_ema > long_ema
signal = None
if self._prev_short_above is not None:
conviction = self._ema_conviction(short_ema, long_ema, float(candle.close))
if not self._prev_short_above and short_above:
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})",
)
elif self._prev_short_above and not short_above:
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)
return None
|
