From 8b53b8d3c98708cb64860a746593aa5e043ef425 Mon Sep 17 00:00:00 2001 From: Marius Mutu Date: Sat, 18 Apr 2026 11:55:39 +0300 Subject: [PATCH] feat(alerts): fire_on_phase_skip backstop + public FSM lockout API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the FSM observes ARMED → light_{green,red} directly (the dark prime was missed), the color classifier likely mis-labeled the dark phase as gray/background. Missing a fire is worse than a noisy alert, so the new [options.alerts] fire_on_phase_skip flag (default True) emits a phase_skip_fire alert on that transition with the standard 240s lockout to dedupe detector bounces. Adds public StateMachine.is_locked / record_fire so the handler does not reach into private attrs. _handle_tick now accepts an optional cfg to read the flag; falls back to True if cfg is absent (tests). Config gains AlertBehaviorCfg (new alerts field), parsed from [options.alerts]. Example TOML updated with an explanatory comment. Tests: test_phase_skip_fire_when_flag_on, test_phase_skip_no_fire_when_flag_off, test_phase_skip_lockout_suppresses_spam, test_state_machine_is_locked_and_record_fire_public_api. Co-Authored-By: Claude Sonnet 4.6 --- configs/example.toml | 7 ++++ src/atm/config.py | 19 +++++++++ src/atm/main.py | 28 +++++++++++++- src/atm/state_machine.py | 17 ++++++++ tests/test_handle_tick.py | 81 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 151 insertions(+), 1 deletion(-) diff --git a/configs/example.toml b/configs/example.toml index cdcb3c9..e738105 100644 --- a/configs/example.toml +++ b/configs/example.toml @@ -81,6 +81,13 @@ low_conf_run = 3 phaseb_timeout_s = 600 dead_letter_path = "logs/dead_letter.jsonl" +# Alert-behavior toggles (not screenshot-attachment; see attach_screenshots below). +# fire_on_phase_skip: emit a backstop "PHASE SKIP" alert when the FSM observes +# ARMED → light_green/light_red directly (skipping the dark prime). Default on +# because missing a fire is worse than a false-positive phase-skip alert. +[options.alerts] +fire_on_phase_skip = true + # Per-kind screenshot-attach toggles. All default to true on upgrade. # Accepts either a bare bool (legacy: attach_screenshots = true) or this table. [options.attach_screenshots] diff --git a/src/atm/config.py b/src/atm/config.py index 35aafa2..6d11768 100644 --- a/src/atm/config.py +++ b/src/atm/config.py @@ -97,6 +97,18 @@ class AlertsCfg: trigger: bool = True +@dataclass(frozen=True) +class AlertBehaviorCfg: + """Alert behavior knobs (not screenshot toggles). + + `fire_on_phase_skip`: backstop alert when FSM observes ARMED→light_{green,red} + directly (skipping the dark prime phase — often means dark color was + mis-classified as gray). Default True: missing a fire is worse than a noisy + phase-skip alert. Disable via `[options.alerts] fire_on_phase_skip = false`. + """ + fire_on_phase_skip: bool = True + + @dataclass(frozen=True) class Config: window_title: str @@ -117,6 +129,7 @@ class Config: phaseb_timeout_s: int = 600 dead_letter_path: str = "logs/dead_letter.jsonl" attach_screenshots: AlertsCfg = field(default_factory=AlertsCfg) + alerts: AlertBehaviorCfg = field(default_factory=AlertBehaviorCfg) config_version: str = "unknown" def __post_init__(self) -> None: @@ -184,6 +197,11 @@ class Config: ) else: attach = AlertsCfg() + + alerts_dict = opts.get("alerts", {}) or {} + alert_behavior = AlertBehaviorCfg( + fire_on_phase_skip=bool(alerts_dict.get("fire_on_phase_skip", True)), + ) return cls( window_title=data["window_title"], dot_roi=roi, @@ -203,5 +221,6 @@ class Config: phaseb_timeout_s=int(opts.get("phaseb_timeout_s", 600)), dead_letter_path=opts.get("dead_letter_path", "logs/dead_letter.jsonl"), attach_screenshots=attach, + alerts=alert_behavior, config_version=version, ) diff --git a/src/atm/main.py b/src/atm/main.py index 0a6a686..dc1416d 100644 --- a/src/atm/main.py +++ b/src/atm/main.py @@ -415,6 +415,7 @@ def _handle_tick( audit: _AuditLike, first_accepted: bool, snapshot: Snapshot | None = None, + cfg: Any = None, ) -> Transition | None: """Feed FSM for a single accepted color and dispatch arm/prime/late_start alerts. Returns the final Transition, or None when the color triggered a @@ -518,6 +519,31 @@ def _handle_tick( image_path=snap(prime_kind, prime_label), direction=direction, )) + # PHASE_SKIP fire backstop: ARMED→light_{green,red} directly (dark was missed). + # Emits a fire-equivalent alert when cfg.alerts.fire_on_phase_skip (default True). + # Uses public FSM lockout API (is_locked/record_fire) to reuse the standard + # 240s dedupe window so bouncing detectors do not spam the user. + elif tr.reason == "phase_skip" and color in ("light_green", "light_red"): + flag_on = True + if cfg is not None: + alerts_cfg = getattr(cfg, "alerts", None) + if alerts_cfg is not None: + flag_on = bool(getattr(alerts_cfg, "fire_on_phase_skip", True)) + if flag_on: + direction = "BUY" if color == "light_green" else "SELL" + if not fsm.is_locked(direction, now): + fsm.record_fire(direction, now) + dark_name = "dark_green" if direction == "BUY" else "dark_red" + notifier.send(Alert( + kind="phase_skip_fire", + title=f"PHASE SKIP {direction} — {dark_name} nu a fost detectat", + body=( + "Verifică chart-ul manual. Posibil necalibrare culoare " + f"(observat {color} direct după armare)." + ), + image_path=snap("phase_skip", f"phase_skip_{direction.lower()}"), + direction=direction, + )) return tr @@ -618,7 +644,7 @@ def _sync_detection_tick( canary_ok=True, ) - tr = _handle_tick(fsm, res.color, now, notifier, audit, is_first, snapshot=_snapshot) + tr = _handle_tick(fsm, res.color, now, notifier, audit, is_first, snapshot=_snapshot, cfg=cfg) if tr is None: return _TickSyncResult(frame=frame, res=res, first_consumed=is_first, late_start=True) diff --git a/src/atm/state_machine.py b/src/atm/state_machine.py index f4e92d2..88cc2f8 100644 --- a/src/atm/state_machine.py +++ b/src/atm/state_machine.py @@ -232,3 +232,20 @@ class StateMachine: if last is None: return False return (ts - last) < self._lockout_s + + # ------------------------------------------------------------------ + # Public lockout API — used by fire_on_phase_skip handler outside the + # FSM. Mirrors _is_locked / _last_fire without leaking private attrs. + # ------------------------------------------------------------------ + + def is_locked(self, direction: str, ts: float) -> bool: + """True if a FIRE in `direction` at ts would be within the lockout window.""" + return self._is_locked(direction, ts) + + def record_fire(self, direction: str, ts: float) -> None: + """Mark a FIRE for `direction` at ts, starting the lockout timer. + + Used by backstop handlers (e.g. fire_on_phase_skip) that emit a + fire-equivalent alert without going through the natural FSM path. + """ + self._last_fire[direction] = ts diff --git a/tests/test_handle_tick.py b/tests/test_handle_tick.py index 958ce70..943a5c9 100644 --- a/tests/test_handle_tick.py +++ b/tests/test_handle_tick.py @@ -10,6 +10,8 @@ Covers the six cases from the arm+prime notification plan: """ from __future__ import annotations +from types import SimpleNamespace + from atm.main import _handle_tick from atm.notifier import Alert from atm.state_machine import State, StateMachine @@ -486,3 +488,82 @@ def test_save_annotated_frame_succeeds(tmp_path, monkeypatch): assert "BUY" in result.name assert len(written) == 1 assert not any(e.get("event") == "snapshot_fail" for e in audit.events) + + +# --------------------------------------------------------------------------- +# Commit 3: fire_on_phase_skip backstop +# --------------------------------------------------------------------------- + +def _cfg_with_flag(enabled: bool): + return SimpleNamespace(alerts=SimpleNamespace(fire_on_phase_skip=enabled)) + + +def test_phase_skip_fire_when_flag_on(): + """ARMED_SELL → light_red directly with flag=True → phase_skip_fire alert.""" + fsm = StateMachine(lockout_s=240) + notif = FakeNotifier() + audit = FakeAudit() + + # Arm SELL (yellow from IDLE) + _handle_tick(fsm, "yellow", 1.0, notif, audit, first_accepted=False, + cfg=_cfg_with_flag(True)) + assert fsm.state == State.ARMED_SELL + notif.alerts.clear() + + # ARMED_SELL → light_red (skips dark_red) → phase_skip_fire + tr = _handle_tick(fsm, "light_red", 2.0, notif, audit, first_accepted=False, + cfg=_cfg_with_flag(True)) + assert tr is not None and tr.reason == "phase_skip" + + ps_alerts = [a for a in notif.alerts if a.kind == "phase_skip_fire"] + assert len(ps_alerts) == 1 + assert ps_alerts[0].direction == "SELL" + assert "SELL" in ps_alerts[0].title + + +def test_phase_skip_no_fire_when_flag_off(): + """Same scenario, flag=False → no phase_skip_fire emitted.""" + fsm = StateMachine(lockout_s=240) + notif = FakeNotifier() + audit = FakeAudit() + + _handle_tick(fsm, "yellow", 1.0, notif, audit, first_accepted=False, + cfg=_cfg_with_flag(False)) + notif.alerts.clear() + + _handle_tick(fsm, "light_red", 2.0, notif, audit, first_accepted=False, + cfg=_cfg_with_flag(False)) + + ps_alerts = [a for a in notif.alerts if a.kind == "phase_skip_fire"] + assert ps_alerts == [] + + +def test_phase_skip_lockout_suppresses_spam(): + """Two phase_skip events within lockout_s → only the first emits an alert.""" + fsm = StateMachine(lockout_s=240) + notif = FakeNotifier() + audit = FakeAudit() + cfg = _cfg_with_flag(True) + + # First cycle + _handle_tick(fsm, "yellow", 1.0, notif, audit, first_accepted=False, cfg=cfg) + _handle_tick(fsm, "light_red", 2.0, notif, audit, first_accepted=False, cfg=cfg) + # Second arm + phase_skip well within 240s + _handle_tick(fsm, "yellow", 60.0, notif, audit, first_accepted=False, cfg=cfg) + _handle_tick(fsm, "light_red", 61.0, notif, audit, first_accepted=False, cfg=cfg) + + ps_alerts = [a for a in notif.alerts if a.kind == "phase_skip_fire"] + assert len(ps_alerts) == 1, ( + f"expected 1 phase_skip_fire (lockout), got {len(ps_alerts)}" + ) + + +def test_state_machine_is_locked_and_record_fire_public_api(): + """Public lockout helpers mirror the private _is_locked / _last_fire behavior.""" + fsm = StateMachine(lockout_s=100) + assert fsm.is_locked("BUY", 0.0) is False + + fsm.record_fire("BUY", 10.0) + assert fsm.is_locked("BUY", 50.0) is True # within 100s + assert fsm.is_locked("BUY", 150.0) is False # past lockout + assert fsm.is_locked("SELL", 50.0) is False # other direction unaffected