139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
"""
|
||
Trend-Filtered RSI Pullback Strategy
|
||
=====================================
|
||
|
||
Core idea
|
||
---------
|
||
Use the H1 EMA(50)/EMA(200) crossover to establish the macro trend, then
|
||
enter M15 trades only in the trend direction when RSI pulls back to an
|
||
extreme and *exits* that extreme. This turns RSI from a counter-trend tool
|
||
into a high-probability trend-entry timer.
|
||
|
||
Entry logic
|
||
-----------
|
||
LONG : H1 EMA(50) > EMA(200) [bullish bias] AND M15 RSI crosses back
|
||
above `os_level` from below (RSI was oversold, now recovering).
|
||
SHORT : H1 EMA(50) < EMA(200) [bearish bias] AND M15 RSI crosses back
|
||
below `ob_level` from above (RSI was overbought, now rolling over).
|
||
|
||
Exits
|
||
-----
|
||
TP : entry ± tp_atr × ATR(14)
|
||
SL : entry ∓ sl_atr × ATR(14)
|
||
|
||
Session filter
|
||
--------------
|
||
Only trades during [session_start, session_end) UTC to avoid thin
|
||
Asian-overnight and weekend conditions.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional
|
||
|
||
import pandas as pd
|
||
|
||
from engine import indicators as ind
|
||
|
||
|
||
class RSITrendStrategy:
|
||
def __init__(
|
||
self,
|
||
rsi_period: int = 14,
|
||
atr_period: int = 14,
|
||
os_level: float = 30.0, # oversold threshold
|
||
ob_level: float = 70.0, # overbought threshold
|
||
sl_atr: float = 1.5,
|
||
tp_atr: float = 3.0,
|
||
trend_fast: int = 50, # H1 EMA periods
|
||
trend_slow: int = 200,
|
||
session_start: int = 7, # UTC hour (inclusive)
|
||
session_end: int = 20, # UTC hour (exclusive)
|
||
):
|
||
self.rsi_period = rsi_period
|
||
self.atr_period = atr_period
|
||
self.os_level = os_level
|
||
self.ob_level = ob_level
|
||
self.sl_atr = sl_atr
|
||
self.tp_atr = tp_atr
|
||
self.trend_fast = trend_fast
|
||
self.trend_slow = trend_slow
|
||
self.session_start = session_start
|
||
self.session_end = session_end
|
||
|
||
# ── Strategy protocol ─────────────────────────────────────────────────────
|
||
|
||
def prepare(self, df: pd.DataFrame) -> pd.DataFrame:
|
||
# M15 indicators
|
||
df["rsi"] = ind.rsi(df["close"], self.rsi_period)
|
||
df["atr"] = ind.atr(df, self.atr_period)
|
||
|
||
# Build H1 trend from M15 data (resample, compute EMAs, forward-fill)
|
||
h1 = df["close"].resample("1h", label="left", closed="left").last().dropna()
|
||
h1_ema_fast = ind.ema(h1, self.trend_fast)
|
||
h1_ema_slow = ind.ema(h1, self.trend_slow)
|
||
|
||
df["h1_ema_fast"] = h1_ema_fast.reindex(df.index, method="ffill")
|
||
df["h1_ema_slow"] = h1_ema_slow.reindex(df.index, method="ffill")
|
||
|
||
# EMA spread: positive = bullish, negative = bearish
|
||
df["trend"] = df["h1_ema_fast"] - df["h1_ema_slow"]
|
||
|
||
return df
|
||
|
||
def signal(
|
||
self,
|
||
df: pd.DataFrame,
|
||
i: int,
|
||
) -> Optional[tuple[str, float, float]]:
|
||
if i < 1:
|
||
return None
|
||
|
||
row = df.iloc[i]
|
||
prev = df.iloc[i - 1]
|
||
|
||
# guard: need valid indicators
|
||
if pd.isna(row["rsi"]) or pd.isna(row["atr"]) or pd.isna(row["trend"]):
|
||
return None
|
||
if pd.isna(prev["rsi"]):
|
||
return None
|
||
|
||
# session filter
|
||
hour = row.name.hour
|
||
if not (self.session_start <= hour < self.session_end):
|
||
return None
|
||
|
||
rsi_now = row["rsi"]
|
||
rsi_prev = prev["rsi"]
|
||
trend = row["trend"]
|
||
atr = row["atr"]
|
||
|
||
# entry is on next bar's open — approximate SL/TP from current close
|
||
ref = row["close"]
|
||
|
||
# LONG: bullish trend AND RSI exits oversold
|
||
if trend > 0 and rsi_prev < self.os_level and rsi_now >= self.os_level:
|
||
sl = ref - self.sl_atr * atr
|
||
tp = ref + self.tp_atr * atr
|
||
return "long", sl, tp
|
||
|
||
# SHORT: bearish trend AND RSI exits overbought
|
||
if trend < 0 and rsi_prev > self.ob_level and rsi_now <= self.ob_level:
|
||
sl = ref + self.sl_atr * atr
|
||
tp = ref - self.tp_atr * atr
|
||
return "short", sl, tp
|
||
|
||
return None
|
||
|
||
def describe(self, params: Optional[dict] = None) -> str:
|
||
p = params or {}
|
||
return (
|
||
f"RSITrend("
|
||
f"os={p.get('os_level', self.os_level)}, "
|
||
f"ob={p.get('ob_level', self.ob_level)}, "
|
||
f"sl={p.get('sl_atr', self.sl_atr)}, "
|
||
f"tp={p.get('tp_atr', self.tp_atr)}, "
|
||
f"fast={p.get('trend_fast', self.trend_fast)}, "
|
||
f"slow={p.get('trend_slow', self.trend_slow)})"
|
||
)
|