92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
"""
|
|
Live signal generation — applies the RSI Trend strategy to a freshly fetched
|
|
candle window and returns a trade signal (or None).
|
|
|
|
The logic mirrors backtest.strategies.rsi_trend exactly so live and backtest
|
|
behaviour stay in sync.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import pandas as pd
|
|
|
|
from . import config
|
|
from engine import indicators as ind
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── signal result ─────────────────────────────────────────────────────────────
|
|
|
|
class Signal:
|
|
__slots__ = ("symbol", "direction", "sl", "tp", "atr", "rsi")
|
|
|
|
def __init__(self, symbol: str, direction: str, sl: float, tp: float,
|
|
atr: float, rsi: float):
|
|
self.symbol = symbol
|
|
self.direction = direction # 'long' | 'short'
|
|
self.sl = sl
|
|
self.tp = tp
|
|
self.atr = atr
|
|
self.rsi = rsi
|
|
|
|
def __repr__(self) -> str:
|
|
return (f"Signal({self.symbol} {self.direction.upper()} "
|
|
f"sl={self.sl:.5f} tp={self.tp:.5f} rsi={self.rsi:.1f})")
|
|
|
|
|
|
# ── signal check ─────────────────────────────────────────────────────────────
|
|
|
|
def check(df: pd.DataFrame, symbol: str) -> Optional[Signal]:
|
|
"""
|
|
Expects a DataFrame with columns [open, high, low, close] indexed by time,
|
|
with at least (LOOKBACK) rows. Returns a Signal or None.
|
|
|
|
We check bar [-2] (the last completed bar) relative to bar [-3] (prev).
|
|
Bar [-1] is the currently forming bar — we never use it for signals.
|
|
"""
|
|
if len(df) < config.LOOKBACK:
|
|
logger.debug("%s: insufficient bars (%d < %d)", symbol, len(df), config.LOOKBACK)
|
|
return None
|
|
|
|
# M15 indicators on full window
|
|
close = df["close"]
|
|
rsi_s = ind.rsi(close, config.RSI_PERIOD)
|
|
atr_s = ind.atr(df, config.ATR_PERIOD)
|
|
|
|
# H1 trend from resampled M15
|
|
h1 = close.resample("1h", label="left", closed="left").last().dropna()
|
|
ema_f = ind.ema(h1, config.TREND_FAST)
|
|
ema_s = ind.ema(h1, config.TREND_SLOW)
|
|
trend = (ema_f - ema_s).reindex(df.index, method="ffill")
|
|
|
|
# Use second-to-last bar as signal bar (last fully closed bar)
|
|
bar = df.iloc[-2]
|
|
prev = df.iloc[-3]
|
|
rsi_now = rsi_s.iloc[-2]
|
|
rsi_prev = rsi_s.iloc[-3]
|
|
atr_val = atr_s.iloc[-2]
|
|
trend_val = trend.iloc[-2]
|
|
|
|
if any(pd.isna(v) for v in [rsi_now, rsi_prev, atr_val, trend_val]):
|
|
return None
|
|
|
|
# session filter on signal bar
|
|
hour = bar.name.hour
|
|
if not (config.SESSION_START <= hour < config.SESSION_END):
|
|
return None
|
|
|
|
ref = bar["close"]
|
|
|
|
if trend_val > 0 and rsi_prev < config.OS_LEVEL and rsi_now >= config.OS_LEVEL:
|
|
sl = ref - config.SL_ATR * atr_val
|
|
tp = ref + config.TP_ATR * atr_val
|
|
return Signal(symbol, "long", sl, tp, atr_val, rsi_now)
|
|
|
|
if trend_val < 0 and rsi_prev > config.OB_LEVEL and rsi_now <= config.OB_LEVEL:
|
|
sl = ref + config.SL_ATR * atr_val
|
|
tp = ref - config.TP_ATR * atr_val
|
|
return Signal(symbol, "short", sl, tp, atr_val, rsi_now)
|
|
|
|
return None
|