diff --git a/TODOS.md b/TODOS.md index ad6ce7e..120a8af 100644 --- a/TODOS.md +++ b/TODOS.md @@ -23,6 +23,18 @@ Standalone helper `atm dump --duration 4h --dir samples/` that just captures fra Expand README with concrete Windows Task Scheduler XML export + import steps, example triggers for 16:30→18:30 and 21:00→23:00, and DST-change checklist reminder. +## P2-alert-mute-flags + +Per-kind mute toggles for notifications in case arm/prime turn out too noisy in live sessions: + +- `cfg.notify.arm: bool = true` +- `cfg.notify.prime: bool = true` +- `cfg.notify.late_start: bool = true` + +Default all true. Gate each `notifier.send()` in `_handle_tick()` on the flag. Start after 3+ live sessions confirm the signal/noise ratio. + +Blocked on: Faza 1 baseline evidence. + ## P3-faza2-exec Auto-execution on TradeLocker. Blocked on TOS audit (see `docs/phase2-prop-firm-audit.md`). Not started until GO decision + 20+ Faza 1 sessions. diff --git a/src/atm/main.py b/src/atm/main.py index 10f52d2..b542b4d 100644 --- a/src/atm/main.py +++ b/src/atm/main.py @@ -5,14 +5,25 @@ import argparse import os import sys import time +from datetime import datetime from pathlib import Path -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Protocol, cast from atm.config import Config # stdlib-only (tomllib); safe at module level +from atm.notifier import Alert +from atm.state_machine import State, StateMachine, Transition if TYPE_CHECKING: from atm.state_machine import DotColor + +class _NotifierLike(Protocol): + def send(self, alert: Alert) -> None: ... + + +class _AuditLike(Protocol): + def log(self, event: dict) -> None: ... + # Module-level reference — set lazily by _cmd_dryrun; tests may monkeypatch it. dryrun = None @@ -324,14 +335,106 @@ def _cmd_report(args) -> None: # Live loop # --------------------------------------------------------------------------- +def _handle_tick( + fsm: StateMachine, + color: str, + now: float, + notifier: _NotifierLike, + audit: _AuditLike, + first_accepted: bool, +) -> 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 + late-start short-circuit (FSM untouched; caller should skip FIRE handling). + + Pure in the sense that all state lives in the arguments — safe to unit-test + with a FakeNotifier and FakeAudit. + """ + # Late start: the very first accepted color is already at FIRE phase. + # User came online after the trade signal fired — warn and skip FSM feed. + if first_accepted and color in ("light_green", "light_red") and fsm.state == State.IDLE: + direction = "BUY" if color == "light_green" else "SELL" + audit.log({ + "ts": now, + "event": "late_start", + "observed_color": color, + }) + notifier.send(Alert( + kind="late_start", + title=f"ATM started late — {direction} already fired", + body=f"Observed {color} at startup. Check chart manually.", + direction=direction, + )) + return None + + # Catchup synth-arm: first accepted color is already at PRIME phase. + # Drive FSM through a synthetic arm so the real PRIME transition fires a + # normal prime alert below. Audit entry is tagged catchup:true. + catchup = False + if first_accepted and color in ("dark_green", "dark_red") and fsm.state == State.IDLE: + assert fsm.state == State.IDLE, "synth-arm invariant: FSM must be IDLE" + arm_color = "turquoise" if color == "dark_green" else "yellow" + direction = "BUY" if color == "dark_green" else "SELL" + tr_synth = fsm.feed(cast("DotColor", arm_color), now) + audit.log({ + "ts": now, + "event": "tick", + "color": arm_color, + "state": tr_synth.next.value, + "reason": tr_synth.reason, + "catchup": True, + "synthesized_from": color, + }) + notifier.send(Alert( + kind="arm", + title=f"{direction} armed ({arm_color}) — catchup", + body=f"catchup — session already armed at startup " + f"@ {datetime.fromtimestamp(now).isoformat(timespec='seconds')}", + direction=direction, + )) + catchup = True + + # Normal FSM feed + tr = fsm.feed(cast("DotColor", color), now) + tick_event: dict = { + "ts": now, + "event": "tick", + "color": color, + "state": tr.next.value, + "reason": tr.reason, + } + if catchup: + tick_event["catchup"] = True + audit.log(tick_event) + + # ARM: turquoise (BUY) / yellow (SELL) — only on fresh IDLE→ARMED + if tr.reason == "arm": + direction = "BUY" if tr.next == State.ARMED_BUY else "SELL" + notifier.send(Alert( + kind="arm", + title=f"{direction} armed ({color})", + body=f"@ {datetime.fromtimestamp(now).isoformat(timespec='seconds')}", + direction=direction, + )) + # PRIME: dark_green (BUY) / dark_red (SELL) — only on ARMED→PRIMED + elif tr.reason == "prime": + direction = "BUY" if tr.next == State.PRIMED_BUY else "SELL" + suffix = " — catchup" if catchup else "" + notifier.send(Alert( + kind="prime", + title=f"{direction} primed ({color}){suffix}", + body=f"@ {datetime.fromtimestamp(now).isoformat(timespec='seconds')}", + direction=direction, + )) + return tr + + def run_live(cfg, duration_s=None, capture_stub: bool = False) -> None: """Main live monitoring loop. Imports are lazy to keep --help fast.""" try: from atm.detector import Detector - from atm.state_machine import StateMachine from atm.canary import Canary from atm.levels import LevelsExtractor - from atm.notifier import Alert from atm.notifier.fanout import FanoutNotifier from atm.notifier.discord import DiscordNotifier from atm.notifier.telegram import TelegramNotifier @@ -351,7 +454,6 @@ def run_live(cfg, duration_s=None, capture_stub: bool = False) -> None: notifier = FanoutNotifier(backends, Path(cfg.dead_letter_path)) # Sanity check: capture one frame, confirm canary matches calibration. - from datetime import datetime first_frame = capture() if first_frame is None: print("WARN: first capture returned None — window/region missing", flush=True) @@ -378,6 +480,7 @@ def run_live(cfg, duration_s=None, capture_stub: bool = False) -> None: heartbeat_due = start + cfg.heartbeat_min * 60 levels_extractor = None last_saved_color: str | None = None + first_accepted = True samples_dir = Path("samples") samples_dir.mkdir(exist_ok=True) fires_dir = Path("logs") / "fires" @@ -401,11 +504,13 @@ def run_live(cfg, duration_s=None, capture_stub: bool = False) -> None: # detection res = detector.step(now) if res.accepted and res.color: - tr = fsm.feed(cast("DotColor", res.color), now) - audit.log({ - "ts": now, "event": "tick", - "color": res.color, "state": tr.next.value, "reason": tr.reason, - }) + is_first = first_accepted + first_accepted = False + tr = _handle_tick(fsm, res.color, now, notifier, audit, is_first) + if tr is None: + # late_start short-circuit: FSM untouched, skip FIRE + corpus save + time.sleep(cfg.loop_interval_s) + continue # corpus: save full frame on each new distinct color for later labeling if res.color != last_saved_color: ts_str = datetime.fromtimestamp(now).strftime("%Y%m%d_%H%M%S") diff --git a/src/atm/notifier/__init__.py b/src/atm/notifier/__init__.py index d556859..c6afeb5 100644 --- a/src/atm/notifier/__init__.py +++ b/src/atm/notifier/__init__.py @@ -5,7 +5,7 @@ from typing import Protocol @dataclass class Alert: - kind: str # "trigger" | "heartbeat" | "levels" | "warn" + kind: str # "trigger" | "heartbeat" | "levels" | "warn" | "arm" | "prime" | "late_start" title: str body: str image_path: Path | None = None # annotated screenshot diff --git a/tests/test_handle_tick.py b/tests/test_handle_tick.py new file mode 100644 index 0000000..e7cc1ff --- /dev/null +++ b/tests/test_handle_tick.py @@ -0,0 +1,220 @@ +"""Unit tests for atm.main._handle_tick. + +Covers the six cases from the arm+prime notification plan: + 1. fresh BUY arm (turquoise) + 2. fresh SELL arm (yellow) + 3. BUY prime (turquoise → dark_green) + 4. silent refresh (turquoise → turquoise) + 5. catchup synth-arm from dark_green + prime alert, audit tagged catchup:true + 6. late-start guard (light_green from IDLE with first_accepted=True) +""" +from __future__ import annotations + +from atm.main import _handle_tick +from atm.notifier import Alert +from atm.state_machine import State, StateMachine + + +class FakeNotifier: + """Captures alerts instead of fanning out. send() is synchronous.""" + + def __init__(self) -> None: + self.alerts: list[Alert] = [] + + def send(self, alert: Alert) -> None: + self.alerts.append(alert) + + +class FakeAudit: + """Captures audit events in-memory.""" + + def __init__(self) -> None: + self.events: list[dict] = [] + + def log(self, event: dict) -> None: + self.events.append(event) + + +# --------------------------------------------------------------------------- +# 1. fresh BUY arm +# --------------------------------------------------------------------------- + +def test_handle_tick_arm_buy(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False) + + assert tr is not None + assert tr.next == State.ARMED_BUY + assert tr.reason == "arm" + assert len(notif.alerts) == 1 + assert notif.alerts[0].kind == "arm" + assert notif.alerts[0].direction == "BUY" + assert "turquoise" in notif.alerts[0].title + + +# --------------------------------------------------------------------------- +# 2. fresh SELL arm +# --------------------------------------------------------------------------- + +def test_handle_tick_arm_sell(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "yellow", 1.0, notif, audit, first_accepted=False) + + assert tr is not None + assert tr.next == State.ARMED_SELL + assert len(notif.alerts) == 1 + assert notif.alerts[0].kind == "arm" + assert notif.alerts[0].direction == "SELL" + assert "yellow" in notif.alerts[0].title + + +# --------------------------------------------------------------------------- +# 3. BUY prime — turquoise then dark_green +# --------------------------------------------------------------------------- + +def test_handle_tick_prime_buy(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + _handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False) + tr2 = _handle_tick(fsm, "dark_green", 2.0, notif, audit, first_accepted=False) + + assert tr2 is not None + assert tr2.next == State.PRIMED_BUY + assert len(notif.alerts) == 2 + assert notif.alerts[0].kind == "arm" + assert notif.alerts[1].kind == "prime" + assert notif.alerts[1].direction == "BUY" + # Non-catchup prime alert must not mention catchup + assert "catchup" not in notif.alerts[1].title.lower() + + +# --------------------------------------------------------------------------- +# 4. refresh is silent — second turquoise in ARMED_BUY emits no alert +# --------------------------------------------------------------------------- + +def test_handle_tick_refresh_no_alert(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + _handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=False) + tr2 = _handle_tick(fsm, "turquoise", 2.0, notif, audit, first_accepted=False) + + assert tr2 is not None + assert tr2.reason == "refresh" + assert len(notif.alerts) == 1 # only the first (real) arm + + +# --------------------------------------------------------------------------- +# 5. catchup synth-arm from dark_green with first_accepted=True +# --------------------------------------------------------------------------- + +def test_handle_tick_catchup_dark_green(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "dark_green", 1.0, notif, audit, first_accepted=True) + + # Synth arm + real prime feed → FSM lands in PRIMED_BUY + assert tr is not None + assert tr.next == State.PRIMED_BUY + assert tr.reason == "prime" + + # Two alerts: synth-arm (catchup-tagged) + prime + assert len(notif.alerts) == 2 + assert notif.alerts[0].kind == "arm" + assert notif.alerts[0].direction == "BUY" + assert "catchup" in notif.alerts[0].title.lower() or "catchup" in notif.alerts[0].body.lower() + assert notif.alerts[1].kind == "prime" + assert notif.alerts[1].direction == "BUY" + assert "catchup" in notif.alerts[1].title.lower() or "catchup" in notif.alerts[1].body.lower() + + # Audit: both tick entries tagged catchup:true + catchup_events = [e for e in audit.events if e.get("catchup")] + assert len(catchup_events) == 2 + synth = next(e for e in catchup_events if e.get("synthesized_from") == "dark_green") + assert synth["color"] == "turquoise" + assert synth["reason"] == "arm" + + +def test_handle_tick_catchup_dark_red(): + """Mirror of dark_green — SELL side.""" + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "dark_red", 1.0, notif, audit, first_accepted=True) + + assert tr is not None + assert tr.next == State.PRIMED_SELL + assert len(notif.alerts) == 2 + assert notif.alerts[0].direction == "SELL" + assert notif.alerts[1].direction == "SELL" + + +# --------------------------------------------------------------------------- +# 6. late-start: light_green from IDLE with first_accepted=True +# --------------------------------------------------------------------------- + +def test_handle_tick_late_start(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "light_green", 1.0, notif, audit, first_accepted=True) + + # FSM untouched + assert tr is None + assert fsm.state == State.IDLE + + # Exactly one late_start alert + assert len(notif.alerts) == 1 + assert notif.alerts[0].kind == "late_start" + assert notif.alerts[0].direction == "BUY" + + # Audit has a late_start event + assert any(e.get("event") == "late_start" for e in audit.events) + + +def test_handle_tick_late_start_sell(): + """Mirror of light_green — light_red triggers SELL late_start.""" + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "light_red", 1.0, notif, audit, first_accepted=True) + + assert tr is None + assert fsm.state == State.IDLE + assert len(notif.alerts) == 1 + assert notif.alerts[0].kind == "late_start" + assert notif.alerts[0].direction == "SELL" + + +# --------------------------------------------------------------------------- +# Extra guard: first_accepted=True with turquoise is a normal arm (decision 3A) +# --------------------------------------------------------------------------- + +def test_handle_tick_first_turquoise_no_catchup_label(): + fsm = StateMachine(lockout_s=60) + notif = FakeNotifier() + audit = FakeAudit() + + tr = _handle_tick(fsm, "turquoise", 1.0, notif, audit, first_accepted=True) + + assert tr is not None + assert tr.next == State.ARMED_BUY + assert len(notif.alerts) == 1 + assert notif.alerts[0].kind == "arm" + # Decision 3A: cannot distinguish fresh arm from catchup-at-arm-phase + assert "catchup" not in notif.alerts[0].title.lower() + assert "catchup" not in notif.alerts[0].body.lower()