33 lines
1.4 KiB
Python
33 lines
1.4 KiB
Python
 | from datetime import datetime, timedelta, timezone
| ||
from utils.indicators import atr
| |||
| |||
class Filters:
| |||
def __init__(self, cfg: dict):
| |||
self.cfg = cfg
| |||
self.news = [(datetime.fromisoformat(d), int(b), int(a)) for d, b, a in cfg.get("news_blackouts", [])]
| |||
| |||
def spread_ok(self, symbol, info, tick) -> bool:
| |||
lim = self.cfg["max_spread_points"]
| |||
return (tick.ask - tick.bid) / info.point <= lim.get(symbol, lim["default"])
| |||
| |||
def session_ok(self) -> bool:
| |||
h = datetime.now(timezone.utc).hour
| |||
return any(a <= h < b for a, b in self.cfg["sessions_utc"])
| |||
| |||
def news_ok(self) -> bool:
| |||
now = datetime.now(timezone.utc).replace(tzinfo=None)
| |||
return not any(t - timedelta(minutes=b) <= now <= t + timedelta(minutes=a) for t, b, a in self.news)
| |||
| |||
def regime_ok(self, df) -> bool:
| |||
r = self.cfg.get("regime", {})
| |||
if not r.get("enabled"):
| |||
return True
| |||
a = atr(df, 14)
| |||
return a.iloc[-1] / a.iloc[-100:].mean() >= r["min_atr_ratio"]
| |||
| |||
def all_ok(self, symbol, info, tick, df):
| |||
if not self.session_ok(): return False, "hors session"
| |||
if not self.news_ok(): return False, "fenêtre news"
| |||
if not self.spread_ok(symbol, info, tick): return False, "spread élevé"
| |||
if not self.regime_ok(df): return False, "volatilité trop faible"
| |||
return True, ""
|