ForexBot2/violent_watch.py

"""
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 _log_watch(report: Any, watch: "ViolentWatch") -> None:
    """Persist the assessment onto the scan's signals.db row (fail-safe).

    The row is inserted inside analyze_pair — BEFORE a surface attaches
    the watch — so these columns arrive by post-hoc update, exactly like
    the panel verdict. Level and kinds are written even when the display
    line is off: evaluating the watch needs the assessment, not the text.
    """
    try:
        from signal_log import log_watch_update
        log_watch_update(report, watch)
    except Exception:
        logger.debug("violent watch: logging failed", exc_info=True)


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()
        _log_watch(report, watch)
        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"