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, ""
|