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
|
from collections import deque
from decimal import Decimal
import pandas as pd
from shared.models import Candle, Signal, OrderSide
from strategies.base import BaseStrategy
def _compute_rsi(series: pd.Series, period: int) -> float | None:
"""Compute RSI using Wilder's smoothing (EMA-based)."""
if len(series) < period + 1:
return None
delta = series.diff()
gain = delta.clip(lower=0)
loss = -delta.clip(upper=0)
avg_gain = gain.ewm(com=period - 1, min_periods=period).mean()
avg_loss = loss.ewm(com=period - 1, min_periods=period).mean()
rs = avg_gain / avg_loss.replace(0, float("nan"))
rsi = 100 - (100 / (1 + rs))
value = rsi.iloc[-1]
if pd.isna(value):
return None
return float(value)
class RsiStrategy(BaseStrategy):
name: str = "rsi"
def __init__(self) -> None:
super().__init__()
self._closes: deque[float] = deque(maxlen=200)
self._period: int = 14
self._oversold: float = 30.0
self._overbought: float = 70.0
self._quantity: Decimal = Decimal("0.01")
@property
def warmup_period(self) -> int:
return self._period + 1
def configure(self, params: dict) -> None:
self._period = int(params.get("period", 14))
self._oversold = float(params.get("oversold", 30))
self._overbought = float(params.get("overbought", 70))
self._quantity = Decimal(str(params.get("quantity", "0.01")))
if self._period < 2:
raise ValueError(f"RSI period must be >= 2, got {self._period}")
if not (0 < self._oversold < self._overbought < 100):
raise ValueError(
f"RSI thresholds must be 0 < oversold < overbought < 100, "
f"got oversold={self._oversold}, overbought={self._overbought}"
)
if self._quantity <= 0:
raise ValueError(f"Quantity must be positive, got {self._quantity}")
self._init_filters(
require_trend=False,
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()
def _rsi_conviction(self, rsi_value: float) -> float:
"""Map RSI value to conviction strength (0.0-1.0).
For BUY (oversold): lower RSI = higher conviction.
For SELL (overbought): higher RSI = higher conviction.
Linear scale from the threshold to the extreme (0 or 100).
"""
if rsi_value < self._oversold:
# RSI 0 -> 1.0, RSI at oversold threshold -> 0.0
return min(1.0, max(0.1, (self._oversold - rsi_value) / self._oversold))
elif rsi_value > self._overbought:
# RSI 100 -> 1.0, RSI at overbought threshold -> 0.0
return min(1.0, max(0.1, (rsi_value - self._overbought) / (100.0 - self._overbought)))
return 0.0
def on_candle(self, candle: Candle) -> Signal | None:
self._update_filter_data(candle)
self._closes.append(float(candle.close))
if len(self._closes) < self._period + 1:
return None
series = pd.Series(list(self._closes))
rsi_value = _compute_rsi(series, self._period)
if rsi_value is None:
return None
if rsi_value < self._oversold:
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.BUY,
price=candle.close,
quantity=self._quantity,
conviction=self._rsi_conviction(rsi_value),
reason=f"RSI {rsi_value:.2f} below oversold threshold {self._oversold}",
)
return self._apply_filters(signal)
elif rsi_value > self._overbought:
signal = Signal(
strategy=self.name,
symbol=candle.symbol,
side=OrderSide.SELL,
price=candle.close,
quantity=self._quantity,
conviction=self._rsi_conviction(rsi_value),
reason=f"RSI {rsi_value:.2f} above overbought threshold {self._overbought}",
)
return self._apply_filters(signal)
return None
|
