import pandas as pd
import numpy as np
from .indicators import calculate_rsi, calculate_ema, calculate_atr
from .risk_manager import get_position_size, calculate_trailing_stop
class RSIMomentumStrategy:
"""
RSI + Moving Average Momentum strategy for Indian Equities.
- Entry Long: RSI < oversold threshold and price > 50-period EMA
- Exit Long: RSI > overbought threshold or trailing stop breach
- Dynamic volatility sizing via ATR risk model
"""
def __init__(self,
rsi_period: int = 14,
ma_period: int = 50,
rsi_oversold: float = 30.0,
rsi_overbought: float = 70.0,
risk_per_trade: float = 0.02,
max_slippage_bps: float = 4.5):
self.rsi_period = rsi_period
self.ma_period = ma_period
self.rsi_oversold = rsi_oversold
self.rsi_overbought = rsi_overbought
self.risk_per_trade = risk_per_trade
self.max_slippage_bps = max_slippage_bps
self.current_position = 0.0
self.entry_price = 0.0
def generate_signals(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Generates vectorized discrete signals for offline backtesting.
"""
df = df.copy()
df['rsi'] = calculate_rsi(df['close'], self.rsi_period)
df['ema'] = calculate_ema(df['close'], self.ma_period)
df['atr'] = calculate_atr(df, period=14)
df['signal'] = 0
long_condition = (df['rsi'] < self.rsi_oversold) & (df['close'] > df['ema'])
short_condition = (df['rsi'] > self.rsi_overbought) | (df['close'] < df['ema'])
df.loc[long_condition, 'signal'] = 1
df.loc[short_condition, 'signal'] = -1
return df
def on_candle_close(self, candle: dict, account_equity: float) -> dict:
"""
Event-driven execution handler called on every bar close.
"""
price = candle['close']
rsi = candle['rsi']
ema = candle['ema']
atr = candle.get('atr', 2.5)
# Evaluate Long Entry
if self.current_position == 0:
if rsi < self.rsi_oversold and price > ema:
size = get_position_size(account_equity, price, atr, self.risk_per_trade)
self.current_position = size
self.entry_price = price
return {
"action": "BUY",
"quantity": size,
"limit_price": price,
"stop_loss": price - (2.0 * atr),
"target": price + (4.0 * atr)
}
# Evaluate Long Exit
elif self.current_position > 0:
trailing_stop = calculate_trailing_stop(self.entry_price, price, atr)
if rsi > self.rsi_overbought or price < trailing_stop:
qty = self.current_position
self.current_position = 0
return {
"action": "SELL",
"quantity": qty,
"limit_price": price,
"reason": "RSI_EXIT" if rsi > self.rsi_overbought else "TRAILING_STOP"
}
return {"action": "HOLD"}
