""" ForexBot2 — violent-move watch (DISPLAY-ONLY heads-up) ======================================================= Flags the conditions under which a "violent bar" (range >= 3x the prior ATR14) is unusually likely, so a human reading an alert knows the next hour may be wild. It is a WARNING LINE and a standalone heads-up message — nothing here ever changes a signal, grade, gate, trade setup or backtest result. Hard independence rule (mirrors market_outlook.py / agent_panel.py): this module is NEVER imported by `signal_aggregator`, `engine`, or anything under `backtest/`. Surfaces (telegram_bot, main, the dashboard publisher) attach the result POST-HOC via `attach_watch`, exactly the way panels are attached. Evidence — 3-year study, backtest/research/sizing_campaign_2026-09.md (addenda): * A violent bar is a bar whose range >= 3x the prior ATR14. * Hot clock slots run about 3-4x the normal rate: bars closing 12:00-14:59 UTC on weekdays, the 01:xx UTC hour on any weekday, and the Sunday reopen (22:xx-23:xx UTC). * The 30 minutes after a high-impact US release run about 10x normal. Share of releases followed by a violent bar within the hour: NFP 97%, CPI 91%, FOMC 88%, retail sales 64%, PPI 54%, PCE 50%, jobless claims 44%. * Indicator readings taken BEFORE the bar each lift the odds 1.3-3x: price outside the Bollinger bands, RSI < 30 or >= 70, close outside the Keltner channel, a rising ADX (slope > 3), volatility regime "extreme", market regime "transitioning", and MACD histogram negative and accelerating down (the engine emits STRONG_SELL for exactly that case). * Only the three PRICE-EXTREME readings (Bollinger, RSI, Keltner) earn a level. A bar-by-bar sweep of the 3-year window showed the first draft — which let any two readings raise the level — flagged 27% of all bars "hot" at merely ~2x odds (~13 heads-ups/day). The shipped rule (calendar, or a clock slot plus one price extreme) flags 8.6% of bars: next-bar odds 3-5x normal, ~20% chance within the hour, about 4 hot hours a day. The remaining readings (ADX slope, volatility regime, regime transition, MACD acceleration) are CONTEXT: they are printed once a level exists, and never create one. * Direction: bearish readings preceded a DOWN burst 70-78% of the time on BOTH M5 and M15. Bullish readings preceded an UP burst only ~60% of the time and only on M5 — so an "up" call is made on M5 alone. * Open trades are NOT to be touched. Every protective action tested (early close, tightened stop, partial exit) lost money; the stop already IS the protection. The heads-up says so explicitly. """ import logging from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional, Tuple from config import SignalStrength, Timeframe logger = logging.getLogger(__name__) # Calendar window either side of a high-impact US release (minutes). RELEASE_WINDOW_MINUTES = 30 # Hot clock slots (UTC), from the 3-year bar study. US_DATA_HOURS = (12, 13, 14) ASIAN_OPEN_HOUR = 1 SUNDAY_REOPEN_HOURS = (22, 23) # ADX slope above this counts as "trend strength rising". ADX_SLOPE_HOT = 3.0 _LEVEL_NONE = "none" _LEVEL_WATCH = "watch" _LEVEL_HOT = "hot" @dataclass class ViolentWatch: """One scan's violent-move assessment. Advisory/display-only.""" level: str = _LEVEL_NONE # "none" | "watch" | "hot" triggers: List[str] = field(default_factory=list) # plain language kinds: List[str] = field(default_factory=list) # calendar/clock/extreme readings: int = 0 # PRICE-EXTREME readings (0..3) likely_direction: str = "" # "down" | "up" | "" def line(self) -> str: """One-line alert text; "" when nothing fired.""" if self.level == _LEVEL_NONE or not self.triggers: return "" if self.level == _LEVEL_HOT: head = "🌪 Violent-move watch (HOT): " odds = " — odds of a violent bar about 3–5× normal" else: head = "🌪 Violent-move watch (watch): " odds = " — about 2× normal" tail = "" if self.likely_direction == "down": tail = " — if one comes it has usually gone DOWN" elif self.likely_direction == "up": tail = " — if one comes it has usually gone UP" return head + " · ".join(self.triggers) + odds + tail # ── Calendar ───────────────────────────────────────────────────────── def _is_high_impact_usd(event: Any) -> bool: """USD + high impact, tolerating either the `impact` string or the `is_high_impact` property (EconomicEvent exposes both).""" try: if str(getattr(event, "currency", "")).upper() != "USD": return False impact = getattr(event, "impact", None) if isinstance(impact, str): return impact.strip().lower() == "high" return bool(getattr(event, "is_high_impact", False)) except Exception: return False def _calendar_trigger(events: Optional[List[Any]], now: datetime) -> Optional[str]: """Soonest high-impact USD release inside +/-30 min of `now`.""" if not events: return None window = timedelta(minutes=RELEASE_WINDOW_MINUTES) best: Optional[Tuple[float, str]] = None for ev in events or []: if not _is_high_impact_usd(ev): continue when = getattr(ev, "time", None) if not isinstance(when, datetime) or when.tzinfo is None: continue try: delta = when - now except Exception: continue name = str(getattr(ev, "event", "") or "US release").strip() if timedelta(0) < delta <= window: minutes = max(1, int(round(delta.total_seconds() / 60))) text = f"US high-impact release in {minutes} min ({name})" rank = abs(delta.total_seconds()) elif -window <= delta < timedelta(0): minutes = max(1, int(round(-delta.total_seconds() / 60))) text = f"US release just out ({name}, {minutes} min ago)" rank = abs(delta.total_seconds()) else: continue if best is None or rank < best[0]: best = (rank, text) return None if best is None else best[1] # ── Clock ──────────────────────────────────────────────────────────── def _clock_trigger(now: datetime) -> Optional[str]: """Hot clock slot for `now` (UTC), or None.""" try: moment = now.astimezone(timezone.utc) if now.tzinfo else now except Exception: return None weekday, hour = moment.weekday(), moment.hour # Mon=0 .. Sun=6 if weekday <= 4 and hour in US_DATA_HOURS: return "US data hours (12:00–15:00 UTC)" if weekday <= 4 and hour == ASIAN_OPEN_HOUR: return "Asian open hour (01:00 UTC)" if weekday == 6 and hour in SUNDAY_REOPEN_HOURS: return "Sunday reopen" return None # ── Indicator readings ─────────────────────────────────────────────── def _indicator_value(analysis: Any, name: str) -> Optional[float]: try: ind = (analysis.indicators or {}).get(name) if ind is None: return None value = float(ind.value) except Exception: return None if value != value: # NaN return None return value def _indicator_signal(analysis: Any, name: str) -> Optional[SignalStrength]: try: ind = (analysis.indicators or {}).get(name) return None if ind is None else ind.signal except Exception: return None def _keltner_side(analysis: Any) -> Optional[str]: """"above" / "below" / None — is the close outside the channel? The engine's keltner indicator carries `value = close - kc_mid` and stashes kc_upper/kc_lower/kc_mid in `key_levels` (engine.py _compute_keltner). Comparing the value against the band half-widths is therefore exact AND survives the squeeze branches, which overwrite the indicator's signal/description but still publish the same levels. The description text is only a fallback for analyses that were built without key_levels. """ value = _indicator_value(analysis, "keltner") levels = getattr(analysis, "key_levels", None) or {} try: upper = levels.get("kc_upper") lower = levels.get("kc_lower") mid = levels.get("kc_mid") if value is not None and None not in (upper, lower, mid): if value > float(upper) - float(mid): return "above" if value < float(lower) - float(mid): return "below" return None except Exception: pass try: desc = str((analysis.indicators or {})["keltner"].description).lower() except Exception: return None if "above keltner upper" in desc: return "above" if "below keltner lower" in desc: return "below" return None def _collect_readings(analysis: Any) -> Tuple[List[str], List[str], int, int]: """(price_extreme_phrases, context_phrases, bearish, bullish). Only the PRICE-EXTREME group earns a level (see the module docstring); the context group is printed alongside once a level exists but can never create one. Both groups feed direction. """ extreme: List[str] = [] context: List[str] = [] bear = bull = 0 if analysis is None: return extreme, context, bear, bull # ── Price extremes (level-bearing) ── bb = _indicator_value(analysis, "bollinger") if bb is not None: if bb < 0: extreme.append("price below the Bollinger bands") bear += 1 elif bb > 1: extreme.append("price above the Bollinger bands") bull += 1 rsi = _indicator_value(analysis, "rsi") if rsi is not None: if rsi < 30: extreme.append(f"RSI {rsi:.0f}, oversold extreme") bear += 1 elif rsi >= 70: extreme.append(f"RSI {rsi:.0f}, overbought extreme") bull += 1 kc_side = _keltner_side(analysis) if kc_side is not None: extreme.append("price outside the Keltner channel") if kc_side == "above": bull += 1 else: bear += 1 # ── Context (never raises the level) ── slope = getattr(analysis, "_adx_slope", None) try: if slope is not None and float(slope) > ADX_SLOPE_HOT: context.append("trend strength rising") except (TypeError, ValueError): pass if str(getattr(analysis, "volatility_regime", "")) == "extreme": context.append("volatility extreme") regime = str(getattr(analysis, "market_regime", "")) if regime == "transitioning": context.append("regime transitioning") macd = _indicator_signal(analysis, "macd") if macd == SignalStrength.STRONG_SELL: context.append("downside momentum accelerating") bear += 1 elif macd == SignalStrength.STRONG_BUY: context.append("upside momentum accelerating") bull += 1 # Regime contributes direction only — it has no phrase of its own # beyond "regime transitioning" above. if regime == "trending_down": bear += 1 elif regime == "trending_up": bull += 1 return extreme, context, bear, bull # ── Assessment ─────────────────────────────────────────────────────── def assess(report: Any, now: datetime, events: Optional[List[Any]] = None, primary_tf: Optional[Timeframe] = None) -> ViolentWatch: """Score violent-move conditions. Every input may be missing or malformed — the answer then degrades to "no trigger". Never raises.""" triggers: List[str] = [] kinds: List[str] = [] calendar = _calendar_trigger(events, now) if calendar: triggers.append(calendar) kinds.append("calendar") clock = _clock_trigger(now) if clock: triggers.append(clock) kinds.append("clock") analysis = getattr(report, "primary_analysis", None) extreme_phrases, context_phrases, bear, bull = _collect_readings(analysis) triggers.extend(extreme_phrases) readings = len(extreme_phrases) if readings >= 1: kinds.append("extreme") # Level rule (bar-by-bar validated, see module docstring): only # price extremes and the clock/calendar create a level. if calendar or (clock and readings >= 1): level = _LEVEL_HOT elif clock or readings >= 2: level = _LEVEL_WATCH else: level = _LEVEL_NONE # Context readings are colour, printed only once a level exists. if level != _LEVEL_NONE: triggers.extend(context_phrases) direction = "" if bear >= 2 and bear > bull: direction = "down" elif bull >= 2 and bull > bear: # Bullish readings only earned a call on M5 in the study. tf_value = getattr(primary_tf, "value", primary_tf) if tf_value == Timeframe.M5.value: direction = "up" return ViolentWatch(level=level, triggers=triggers, kinds=kinds, readings=readings, likely_direction=direction) def attach_watch(report: Any, primary_tf: Optional[Timeframe], settings: Dict[str, Any], now: Optional[datetime] = None) -> ViolentWatch: """Stamp `report.watch_line` with the display line (advisory only). Contract: NEVER raises. A broken watch must not cost an alert send or a dashboard publish — a failure logs and returns level "none". """ try: settings = settings or {} if not settings.get("enabled", False): return ViolentWatch() moment = now or datetime.now(timezone.utc) events: Optional[List[Any]] = None try: # Lazy import keeps this module import-light and free of any # signal-path dependency. The calendar rides NewsEngine's # class-level cache, so a warm cache costs no network I/O. from config import PlatformConfig from news_engine import NewsEngine events = NewsEngine(PlatformConfig().news).get_week_events(["USD"]) except Exception as exc: logger.debug("violent watch: calendar unavailable (%s)", exc) watch = assess(report, moment, events, primary_tf) if settings.get("alert_line", True): report.watch_line = watch.line() return watch except Exception: logger.exception("violent watch: assessment failed") return ViolentWatch() def should_send_headsup(state: Dict[str, Any], now: datetime, kinds: List[str], max_per_day: int) -> Tuple[bool, str]: """Rate-limit the standalone heads-up message. `state` is a mutable dict owned by the caller: {"day": date, "count": int, "sent": {(hour_iso, kinds): True}} One message per UTC hour per trigger combination, capped per UTC day. """ day = now.date() if state.get("day") != day: state["day"] = day state["count"] = 0 state["sent"] = {} sent = state.setdefault("sent", {}) if int(state.get("count", 0)) >= int(max_per_day): return False, "daily cap" hour_bucket = now.replace(minute=0, second=0, microsecond=0).isoformat() key = (hour_bucket, tuple(sorted(kinds))) if key in sent: return False, "already sent this hour for these triggers" sent[key] = True state["count"] = int(state.get("count", 0)) + 1 return True, "ok"