feat(alerts): fire_on_phase_skip backstop + public FSM lockout API

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 <noreply@anthropic.com>
This commit is contained in:
2026-04-18 11:55:39 +03:00
parent 9cf49caf8a
commit 8b53b8d3c9
5 changed files with 151 additions and 1 deletions

View File

@@ -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