68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
"""RSI (Relative Strength Index) strategy."""
|
|
|
|
import pandas as pd
|
|
from decimal import Decimal
|
|
from typing import Optional, Dict, Any
|
|
from src.strategies.base import BaseStrategy, StrategySignal, SignalType
|
|
from src.data.indicators import get_indicators
|
|
|
|
class RSIStrategy(BaseStrategy):
|
|
"""RSI-based trading strategy."""
|
|
|
|
def __init__(self, name: str, parameters: Optional[Dict[str, Any]] = None, timeframes: Optional[list] = None):
|
|
"""Initialize RSI strategy.
|
|
|
|
Parameters:
|
|
rsi_period: RSI period (default 14)
|
|
oversold: Oversold threshold (default 30)
|
|
overbought: Overbought threshold (default 70)
|
|
"""
|
|
super().__init__(name, parameters, timeframes)
|
|
self.rsi_period = self.parameters.get('rsi_period', 14)
|
|
self.oversold = self.parameters.get('oversold', 30)
|
|
self.overbought = self.parameters.get('overbought', 70)
|
|
self.indicators = get_indicators()
|
|
self._price_history = []
|
|
|
|
async def on_tick(self, symbol: str, price: Decimal, timeframe: str, data: Dict[str, Any]) -> Optional[StrategySignal]:
|
|
"""Generate signal based on RSI."""
|
|
|
|
# Add price to history
|
|
self._price_history.append(float(price))
|
|
if len(self._price_history) < self.rsi_period + 1:
|
|
return None
|
|
|
|
# Calculate RSI
|
|
prices = pd.Series(self._price_history[-self.rsi_period-1:])
|
|
rsi = self.indicators.rsi(prices, self.rsi_period)
|
|
|
|
if len(rsi) == 0:
|
|
return None
|
|
|
|
current_rsi = rsi.iloc[-1]
|
|
|
|
# Generate signals
|
|
if current_rsi < self.oversold:
|
|
return StrategySignal(
|
|
signal_type=SignalType.BUY,
|
|
symbol=symbol,
|
|
strength=1.0 - (current_rsi / self.oversold),
|
|
price=price,
|
|
metadata={'rsi': float(current_rsi)}
|
|
)
|
|
elif current_rsi > self.overbought:
|
|
return StrategySignal(
|
|
signal_type=SignalType.SELL,
|
|
symbol=symbol,
|
|
strength=(current_rsi - self.overbought) / (100 - self.overbought),
|
|
price=price,
|
|
metadata={'rsi': float(current_rsi)}
|
|
)
|
|
|
|
return None
|
|
|
|
def on_signal(self, signal: StrategySignal) -> Optional[StrategySignal]:
|
|
"""Process signal."""
|
|
return signal if self.should_execute(signal) else None
|
|
|