feat(cli): atm validate-calibration — offline color classification gate

Adds `atm validate-calibration LABEL_FILE` subcommand that runs the Detector
on a set of labeled PNG frames and reports per-sample PASS/FAIL with top-3
candidate colors and RGB-distance suggestions for failures. Exits 0 on 100%
PASS, 1 on any FAIL, 2 on missing/malformed label file.

- New module src/atm/validate.py with ValidationReport + SampleRecord
  dataclasses; reuses Detector.step(frame), does not reimplement color
  classification.
- main.py: new `validate-calibration` subparser and _cmd_validate_calibration
  handler wired into the dispatch map.
- samples/calibration_labels.json seeded with 3 entries from the 2026-04-17
  incident, plus a README describing the schema.
- tests/test_validate.py covers the 3 planned cases: PASS, FAIL w/ top-3
  + suggestion, missing file (graceful error, no traceback).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.gitignore

@@ -46,18 +46,14 @@ ENV/
 # ATM runtime artefacts
 logs/*.jsonl
 logs/dead_letter.jsonl
-logs/detections/
-logs/fires
-logs/pause.flag
 samples/*.png
 samples/*.jpg
 samples/labels.json
 trades.jsonl
 
-# configs: keep template only; ignore generated calibration and runtime state
+# configs: keep template + current marker, not generated calibration
 configs/*.toml
 !configs/example.toml
-configs/current.txt
 
 # Claude scheduler state
 .claude/

CLAUDE.md

@@ -1,35 +0,0 @@
-# ATM — Automated Trading Monitor
-
-Personal Faza-1 tool for the M2D strategy. Python 3.11+.
-
-## Quick Reference
-
-```bash
-pip install -e ".[windows]"    # Windows: live capture
-pip install -e .               # Linux/macOS: dev/dryrun only
-atm calibrate                  # Tk wizard
-atm debug --delay 5            # one-shot capture + detect
-atm run --start-at 16:30 --stop-at 23:00  # live session
-atm dryrun samples             # corpus gate
-pytest                         # run tests
-```
-
-## Skill routing
-
-When the user's request matches an available skill, ALWAYS invoke it using the Skill
-tool as your FIRST action. Do NOT answer directly, do NOT use other tools first.
-The skill has specialized workflows that produce better results than ad-hoc answers.
-
-Key routing rules:
-- Product ideas, "is this worth building", brainstorming → invoke office-hours
-- Bugs, errors, "why is this broken", 500 errors → invoke investigate
-- Ship, deploy, push, create PR → invoke ship
-- QA, test the site, find bugs → invoke qa
-- Code review, check my diff → invoke review
-- Update docs after shipping → invoke document-release
-- Weekly retro → invoke retro
-- Design system, brand → invoke design-consultation
-- Visual audit, design polish → invoke design-review
-- Architecture review → invoke plan-eng-review
-- Save progress, checkpoint, resume → invoke checkpoint
-- Code quality, health check → invoke health

TODOS.md

@@ -49,17 +49,9 @@ Read-only web view of today's audit JSONL + recent triggers. Useful for review a
 
 ---
 
-## P2-yaxis-recalib-detect — Y-axis recalibration detection
-
-Price overlay (from Telegram commands feature) uses `y_axis` linear interpolation to show current price on screenshots. When the user rescales the chart y-axis (common after overnight price gaps), the calibration becomes stale and prices shown are incorrect. Canary check detects layout drift but NOT y-axis range changes.
-
-- Possible approaches: OCR on y-axis labels (fragile), track price range consistency across sessions, or simple "calibration age" warning after N hours.
-- Start after price overlay is live and the false-price frequency is known.
-- Depends on: Telegram commands + price overlay feature being shipped.
-
 ## Quality debt
 
-- [x] **Integration test for run_live loop**: lifecycle async test added in `tests/test_main.py` (IDLE→ARMED→PRIMED auto-poll→FIRE auto-stop).
+- [ ] **Integration test for run_live loop**: currently mocked at module level. Add a short-duration in-memory loop test that threads real detector/state_machine/audit together (no network).
 - [ ] **Coverage report**: run `pytest --cov=atm --cov-report=term-missing`, aim for ≥ 85% per module.
 - [ ] **Typing strictness**: run `pyright src/` with strict mode, fix reported issues.
 - [ ] **Perf baseline**: profile one detection cycle on a representative frame; ensure < 100ms so 5s loop has ample headroom.

pyproject.toml

@@ -13,7 +13,6 @@ dependencies = [
     "pillow>=10.0",
     "requests>=2.31",
     "rich>=13.0",
-    "httpx>=0.27",
 ]
 
 [project.optional-dependencies]
@@ -25,7 +24,6 @@ windows = [
 dev = [
     "pytest>=8.0",
     "pytest-cov>=5.0",
-    "pytest-asyncio>=0.23",
 ]
 
 [project.scripts]

samples/calibration_labels.README.md

@@ -0,0 +1,33 @@
+# calibration_labels.json — schema
+
+Used by `atm validate-calibration` to check that the current color calibration
+classifies known-good screenshots correctly before a live session.
+
+## Schema
+
+A JSON array of entries. Each entry:
+
+| Field      | Type    | Required | Description                                                    |
+|------------|---------|----------|----------------------------------------------------------------|
+| `path`     | string  | yes      | Path to a PNG frame (relative to CWD or absolute).             |
+| `expected` | string  | yes      | Expected color name: one of `turquoise`, `yellow`, `dark_green`, `dark_red`, `light_green`, `light_red`, `gray`. |
+| `note`     | string  | no       | Freeform annotation; shown in SUGGESTIONS output.              |
+
+## Usage
+
+```bash
+atm validate-calibration samples/calibration_labels.json
+```
+
+Exit codes:
+- `0` — every sample PASS
+- `1` — one or more FAIL
+- `2` — label file missing or malformed JSON
+
+## Adding new samples
+
+1. Find a screenshot in `logs/fires/` whose dot color you can verify by eye.
+2. Append an entry with `path`, `expected`, and an optional `note`.
+3. Re-run validation. If it FAILs, the SUGGESTIONS section will tell you the
+   RGB distance between the observed pixel and the expected color's center —
+   use that as input for `atm calibrate`.

samples/calibration_labels.json

@@ -0,0 +1,17 @@
+[
+  {
+    "path": "logs/fires/20260417_201500_arm_sell.png",
+    "expected": "yellow",
+    "note": "first arm of SELL cycle 2026-04-17"
+  },
+  {
+    "path": "logs/fires/20260417_205302_ss.png",
+    "expected": "dark_red",
+    "note": "user confirmed via screenshot (missed live alert)"
+  },
+  {
+    "path": "logs/fires/20260417_210441_ss.png",
+    "expected": "light_red",
+    "note": "fire phase (missed live alert)"
+  }
+]

src/atm/audit.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import json
-import threading
 from datetime import datetime, date
 from pathlib import Path
 from typing import Callable, IO
@@ -17,25 +16,21 @@ class AuditLog:
         self._clock: Callable[[], datetime] = clock or datetime.now
         self._current_date: date | None = None
         self._fh: IO[str] | None = None
-        self._lock = threading.Lock()
 
     def log(self, event: dict) -> None:
         now = self._clock()
         today = now.date()
+
+        if today != self._current_date:
+            self._open(today)
+
         if "ts" not in event:
             event = {**event, "ts": now.isoformat()}
-        with self._lock:
+
-            if today != self._current_date:
+        assert self._fh is not None
-                self._open(today)
+        self._fh.write(json.dumps(event, separators=(",", ":")) + "\n")
-            assert self._fh is not None
-            self._fh.write(json.dumps(event, separators=(",", ":")) + "\n")
 
     def close(self) -> None:
-        with self._lock:
-            self._close_locked()
-
-    def _close_locked(self) -> None:
-        """Close file handle; must be called while holding self._lock."""
         if self._fh is not None:
             try:
                 self._fh.close()
@@ -52,7 +47,7 @@ class AuditLog:
         return self._base_dir / f"{self._current_date}.jsonl"
 
     def _open(self, today: date) -> None:
-        self._close_locked()  # already holding self._lock
+        self.close()
         self._base_dir.mkdir(parents=True, exist_ok=True)
         path = self._base_dir / f"{today}.jsonl"
         self._fh = open(path, "a", buffering=1, encoding="utf-8")

src/atm/commands.py

@@ -1,163 +0,0 @@
-"""Telegram command poller + Command dataclass.
-
-Uses httpx (async) for long-polling getUpdates. The sync TelegramNotifier
-continues to use requests — this module is the only httpx consumer.
-"""
-from __future__ import annotations
-
-import asyncio
-import logging
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, Literal
-
-import httpx
-
-if TYPE_CHECKING:
-    from .config import TelegramCfg
-
-logger = logging.getLogger(__name__)
-
-CommandAction = Literal["set_interval", "stop", "status", "ss"]
-
-_BASE = "https://api.telegram.org/bot{token}/{method}"
-
-
-@dataclass
-class Command:
-    action: CommandAction
-    value: int | None = None  # seconds; only for set_interval
-
-
-class TelegramPoller:
-    """Long-poll Telegram getUpdates, emit Commands into asyncio.Queue.
-
-    Security: rejects messages from chat_ids not in cfg.allowed_chat_ids.
-    Degrades (stops polling) after 3 consecutive 401 responses and warns
-    via Discord (caller responsibility — poller only logs + sets degraded flag).
-    """
-
-    def __init__(
-        self,
-        cfg: TelegramCfg,
-        cmd_queue: asyncio.Queue[Command],
-        audit,  # _AuditLike
-    ) -> None:
-        self._cfg = cfg
-        self._cmd_queue = cmd_queue
-        self._audit = audit
-        self._offset = 0
-        self._consecutive_401 = 0
-        self._degraded = False
-        # fallback: if allowed_chat_ids is empty, accept only the primary chat
-        self._allowed = set(cfg.allowed_chat_ids) or {cfg.chat_id}
-
-    @property
-    def degraded(self) -> bool:
-        return self._degraded
-
-    async def run(self) -> None:
-        async with httpx.AsyncClient() as client:
-            await self._drain(client)
-            while True:
-                if self._degraded:
-                    await asyncio.sleep(5)
-                    continue
-                try:
-                    await self._poll_once(client)
-                except asyncio.CancelledError:
-                    raise
-                except (httpx.HTTPError, httpx.TimeoutException) as exc:
-                    self._audit.log({"event": "poller_error", "error": str(exc)})
-                    await asyncio.sleep(5)
-                except Exception as exc:  # json, unexpected
-                    self._audit.log({"event": "poller_error", "error": str(exc)})
-                    await asyncio.sleep(5)
-
-    async def _drain(self, client: httpx.AsyncClient) -> None:
-        """Discard all pending updates at startup so stale commands don't replay."""
-        try:
-            resp = await client.get(
-                _BASE.format(token=self._cfg.bot_token, method="getUpdates"),
-                params={"timeout": 0, "offset": self._offset},
-                timeout=10,
-            )
-            body = resp.json()
-            if body.get("ok") and body.get("result"):
-                self._offset = body["result"][-1]["update_id"] + 1
-        except Exception as exc:
-            logger.warning("TelegramPoller startup drain failed: %s", exc)
-
-    async def _poll_once(self, client: httpx.AsyncClient) -> None:
-        resp = await client.get(
-            _BASE.format(token=self._cfg.bot_token, method="getUpdates"),
-            params={"timeout": self._cfg.poll_timeout_s, "offset": self._offset},
-            timeout=self._cfg.poll_timeout_s + 5,
-        )
-
-        if resp.status_code == 401:
-            self._consecutive_401 += 1
-            if self._consecutive_401 >= 3:
-                self._degraded = True
-                self._audit.log({"event": "poller_degraded", "reason": "3_consecutive_401"})
-            return
-        self._consecutive_401 = 0
-
-        body = resp.json()
-        if not body.get("ok"):
-            return
-
-        for update in body.get("result", []):
-            self._offset = update["update_id"] + 1
-            await self._process_update(update)
-
-    async def _process_update(self, update: dict) -> None:
-        if "callback_query" in update:
-            # Inline button pressed — may be expired; reply with fallback
-            cbq = update["callback_query"]
-            chat_id = str(cbq.get("from", {}).get("id", ""))
-            if chat_id not in self._allowed:
-                logger.info("Rejected callback_query from chat_id=%s", chat_id)
-                return
-            # Caller handles answerCallbackQuery; just note in audit
-            self._audit.log({"event": "command_received", "action": "callback_query", "chat_id": chat_id})
-            return
-
-        msg = update.get("message") or update.get("edited_message")
-        if not msg:
-            return
-
-        chat_id = str(msg.get("chat", {}).get("id", ""))
-        if chat_id not in self._allowed:
-            logger.info("Rejected message from chat_id=%s", chat_id)
-            return
-
-        text = (msg.get("text") or "").strip().lower()
-        cmd = self._parse_command(text)
-        if cmd is None:
-            return
-
-        self._audit.log({
-            "event": "command_received",
-            "action": cmd.action,
-            "value": cmd.value,
-            "chat_id": chat_id,
-        })
-        await self._cmd_queue.put(cmd)
-
-    def _parse_command(self, text: str) -> Command | None:
-        t = text.lstrip("/").strip()
-        if not t:
-            return None
-        if t == "stop":
-            return Command(action="stop")
-        if t == "status":
-            return Command(action="status")
-        if t in ("ss", "screenshot"):
-            return Command(action="ss")
-        # "3" → set_interval 3 minutes → 180s; "interval 3" also accepted
-        parts = t.split()
-        if len(parts) == 1 and parts[0].isdigit():
-            return Command(action="set_interval", value=int(parts[0]) * 60)
-        if len(parts) == 2 and parts[0] in ("interval", "set_interval") and parts[1].isdigit():
-            return Command(action="set_interval", value=int(parts[1]) * 60)
-        return None

src/atm/config.py

@@ -78,9 +78,6 @@ class DiscordCfg:
 class TelegramCfg:
     bot_token: str
     chat_id: str
-    allowed_chat_ids: tuple[str, ...] = ()
-    poll_timeout_s: int = 30
-    auto_poll_interval_s: int = 180
 
     def __post_init__(self) -> None:
         if not self.bot_token or not self.chat_id:
@@ -159,14 +156,9 @@ class Config:
             drift_threshold=int(data["canary"].get("drift_threshold", 8)),
         )
         discord = DiscordCfg(webhook_url=data["discord"]["webhook_url"])
-        tg = data["telegram"]
-        _allowed = [str(c) for c in tg.get("allowed_chat_ids", [])] or [str(tg["chat_id"])]
         telegram = TelegramCfg(
-            bot_token=tg["bot_token"],
+            bot_token=data["telegram"]["bot_token"],
-            chat_id=str(tg["chat_id"]),
+            chat_id=str(data["telegram"]["chat_id"]),
-            allowed_chat_ids=tuple(_allowed),
-            poll_timeout_s=int(tg.get("poll_timeout_s", 30)),
-            auto_poll_interval_s=int(tg.get("auto_poll_interval_s", 180)),
         )
         opts = data.get("options", {})
         region = None

src/atm/detector.py

@@ -28,7 +28,6 @@ class DetectionResult:
     match: ColorMatch | None  # None if no dot
     accepted: bool  # post-debounce; True only when match repeats debounce_depth times
     color: str | None  # accepted color name (UNKNOWN excluded)
-    dot_pos_abs: tuple[int, int] | None = None  # absolute (x, y) in frame; set when dot_found
 
 
 class Detector:
@@ -61,14 +60,8 @@ class Detector:
         self._debounce: deque[str | None] = deque(maxlen=cfg.debounce_depth)
         self._rolling: deque[DetectionResult] = deque(maxlen=20)
 
-    def step(self, ts: float, frame=None) -> DetectionResult:
+    def step(self, ts: float) -> DetectionResult:
-        """Run one detection tick.
+        frame = self._capture()
-
-        frame: pre-captured BGR ndarray (from asyncio.to_thread capture). When
-        None (default), calls self._capture() — preserving the sync-loop behaviour.
-        """
-        if frame is None:
-            frame = self._capture()
 
         if frame is None:
             self._debounce.append(None)
@@ -124,7 +117,6 @@ class Detector:
             match=match,
             accepted=accepted,
             color=color,
-            dot_pos_abs=(self._cfg.dot_roi.x + x, self._cfg.dot_roi.y + y),
         )
         self._rolling.append(r)
         return r

src/atm/main.py

@@ -2,15 +2,12 @@
 from __future__ import annotations
 
 import argparse
-import asyncio
-import contextlib
 import os
 import sys
 import time
-from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable, Protocol, cast
+from typing import TYPE_CHECKING, Callable, Protocol, cast
 
 from atm.config import Config  # stdlib-only (tomllib); safe at module level
 from atm.notifier import Alert
@@ -118,6 +115,16 @@ def main(argv=None) -> None:
         metavar="PATH", help="Journal JSONL file (default: trades.jsonl)",
     )
 
+    # validate-calibration
+    p_valid = sub.add_parser(
+        "validate-calibration",
+        help="Offline: run Detector on labeled frames and report PASS/FAIL",
+    )
+    p_valid.add_argument(
+        "label_file", type=Path, metavar="LABEL_FILE",
+        help="JSON array with [{path, expected, note?}, ...] entries",
+    )
+
     args = parser.parse_args(argv)
 
     _dispatch = {
@@ -128,6 +135,7 @@ def main(argv=None) -> None:
         "debug": _cmd_debug,
         "journal": _cmd_journal,
         "report": _cmd_report,
+        "validate-calibration": _cmd_validate_calibration,
     }
     _dispatch[args.command](args)
 
@@ -340,6 +348,37 @@ def _cmd_report(args) -> None:
     )
 
 
+def _cmd_validate_calibration(args) -> None:
+    """Run offline calibration validation; exit 0 on 100% PASS, 1 otherwise."""
+    try:
+        from atm.validate import validate_calibration, ValidationError
+    except ImportError as exc:
+        sys.exit(f"validate module not available: {exc}")
+
+    label_file = Path(args.label_file)
+    try:
+        cfg = Config.load_current(Path("configs"))
+    except FileNotFoundError as exc:
+        sys.exit(f"config not found: {exc}")
+
+    try:
+        config_name = ""
+        cur_ptr = Path("configs") / "current.txt"
+        if cur_ptr.exists():
+            config_name = cur_ptr.read_text(encoding="utf-8").strip()
+    except Exception:
+        config_name = ""
+
+    try:
+        report = validate_calibration(label_file, cfg, config_name=config_name)
+    except ValidationError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        sys.exit(2)
+
+    print(report.render())
+    sys.exit(0 if report.all_pass else 1)
+
+
 # ---------------------------------------------------------------------------
 # Live loop
 # ---------------------------------------------------------------------------
@@ -351,8 +390,6 @@ def _save_annotated_frame(
     label: str,
     now: float,
     audit: _AuditLike | None = None,
-    dot_pos_abs: "tuple[int, int] | None" = None,
-    canary_ok: bool = True,
 ) -> "Path | None":
     """Save BGR frame with cyan dot_roi rect to ``logs/fires/{ts}_{label}.png``.
 
@@ -360,10 +397,6 @@ def _save_annotated_frame(
     audit (when provided) so disk-full / permission issues don't become silent
     regressions. Never raises — snapshot is a best-effort enhancement, the
     text alert must still go out.
-
-    dot_pos_abs + canary_ok: when both are set the price overlay is drawn
-    (y-axis linear interpolation via cfg.y_axis). Skipped when canary drifted
-    since calibration may be stale.
     """
     try:
         import cv2  # type: ignore[import-untyped]
@@ -380,22 +413,6 @@ def _save_annotated_frame(
         annotated = frame.copy()
         x, y, w, h = cfg.dot_roi.x, cfg.dot_roi.y, cfg.dot_roi.w, cfg.dot_roi.h
         cv2.rectangle(annotated, (x, y), (x + w, y + h), (0, 255, 255), 2)
-        if dot_pos_abs is not None and canary_ok and hasattr(cfg, "y_axis"):
-            try:
-                _, dot_y = dot_pos_abs
-                ya = cfg.y_axis
-                slope = (ya.p2_price - ya.p1_price) / (ya.p2_y - ya.p1_y)
-                price = ya.p1_price + (dot_y - ya.p1_y) * slope
-                w_frame = annotated.shape[1]
-                text = f"${price:.2f}"
-                font = cv2.FONT_HERSHEY_SIMPLEX
-                scale, thickness = 1.2, 3
-                (tw, th), _ = cv2.getTextSize(text, font, scale, thickness)
-                tx, ty = w_frame - tw - 10, th + 10
-                cv2.rectangle(annotated, (tx - 4, 4), (tx + tw + 4, ty + 4), (0, 0, 0), -1)
-                cv2.putText(annotated, text, (tx, ty), font, scale, (255, 255, 255), thickness, cv2.LINE_AA)
-            except Exception:
-                pass  # price overlay is best-effort; never break the screenshot
         cv2.imwrite(str(fpath), annotated)
         return fpath
     except Exception as exc:
@@ -521,113 +538,7 @@ def _handle_tick(
     return tr
 
 
-@dataclass
-class _TickSyncResult:
-    frame: Any = None
-    res: Any = None       # DetectionResult | None
-    tr: Any = None        # Transition | None
-    first_consumed: bool = False
-    late_start: bool = False
-    new_color: str | None = None  # corpus sample color when changed
-
-
-def _sync_detection_tick(
-    capture: Callable,
-    canary: Any,
-    cfg: Any,
-    detector: Any,
-    fsm: Any,
-    notifier: _NotifierLike,
-    audit: _AuditLike,
-    detection_log: _AuditLike,
-    fires_dir: Path,
-    first_accepted: bool,
-    last_saved_color: "str | None",
-    now: float,
-    samples_dir: Path,
-) -> _TickSyncResult:
-    """One full detection tick (blocking I/O). Runs in asyncio.to_thread."""
-    frame = capture()
-    if frame is None:
-        audit.log({"ts": now, "event": "window_lost"})
-        return _TickSyncResult()
-
-    cr = canary.check(frame)
-    if canary.is_paused:
-        audit.log({"ts": now, "event": "paused", "drift": cr.distance})
-        return _TickSyncResult(frame=frame)
-
-    res = detector.step(now, frame)
-    detection_log.log({
-        "ts": now, "event": "frame",
-        "window_found": res.window_found,
-        "dot_found": res.dot_found,
-        "rgb": list(res.rgb) if res.rgb is not None else None,
-        "match_name": res.match.name if res.match is not None else None,
-        "distance": round(res.match.distance, 2) if res.match is not None else None,
-        "confidence": round(res.match.confidence, 3) if res.match is not None else None,
-        "accepted": res.accepted,
-        "color": res.color,
-    })
-
-    if not (res.accepted and res.color):
-        return _TickSyncResult(frame=frame, res=res)
-
-    is_first = first_accepted
-
-    def _snapshot(kind: str, label: str) -> "Path | None":
-        if not getattr(cfg.attach_screenshots, kind, True):
-            return None
-        return _save_annotated_frame(
-            frame, cfg, fires_dir, label, now, audit=audit,
-            dot_pos_abs=getattr(res, "dot_pos_abs", None),
-            canary_ok=True,
-        )
-
-    tr = _handle_tick(fsm, res.color, now, notifier, audit, is_first, snapshot=_snapshot)
-
-    if tr is None:
-        return _TickSyncResult(frame=frame, res=res, first_consumed=is_first, late_start=True)
-
-    new_color: str | None = None
-    if res.color != last_saved_color:
-        ts_str = datetime.fromtimestamp(now).strftime("%Y%m%d_%H%M%S")
-        sample_path = samples_dir / f"{ts_str}_{res.color}.png"
-        try:
-            import cv2  # type: ignore[import-untyped]
-            cv2.imwrite(str(sample_path), frame)
-        except Exception as exc:
-            audit.log({"ts": now, "event": "sample_save_failed", "error": str(exc)})
-        new_color = res.color
-
-    if tr.trigger and not tr.locked:
-        fire_path: "Path | None" = None
-        if cfg.attach_screenshots.trigger:
-            fire_path = _save_annotated_frame(
-                frame, cfg, fires_dir, tr.trigger, now, audit=audit,
-                dot_pos_abs=getattr(res, "dot_pos_abs", None),
-                canary_ok=True,
-            )
-        notifier.send(Alert(
-            kind="trigger",
-            title=f"Semnal {tr.trigger}",
-            body=f"@ {datetime.fromtimestamp(now).isoformat(timespec='seconds')}",
-            image_path=fire_path,
-            direction=tr.trigger,
-        ))
-
-    return _TickSyncResult(
-        frame=frame, res=res, tr=tr,
-        first_consumed=is_first, new_color=new_color,
-    )
-
-
 def run_live(cfg, duration_s=None, capture_stub: bool = False) -> None:
-    """Sync entry point — delegates to asyncio event loop."""
-    asyncio.run(run_live_async(cfg, duration_s=duration_s, capture_stub=capture_stub))
-
-
-async def run_live_async(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
@@ -637,8 +548,6 @@ async def run_live_async(cfg, duration_s=None, capture_stub: bool = False) -> No
         from atm.notifier.discord import DiscordNotifier
         from atm.notifier.telegram import TelegramNotifier
         from atm.audit import AuditLog
-        from atm.commands import TelegramPoller, Command
-        from atm.scheduler import ScreenshotScheduler
     except ImportError as exc:
         sys.exit(f"run-loop dependencies not available: {exc}")
 
@@ -654,6 +563,7 @@ async def run_live_async(cfg, duration_s=None, capture_stub: bool = False) -> No
     ]
 
     def _on_drop(backend_name: str, dropped: Alert) -> None:
+        """Audit la depășire coadă — face eșecul silențios vizibil."""
         audit.log({
             "ts": time.time(),
             "event": "queue_overflow_drop",
@@ -664,7 +574,7 @@ async def run_live_async(cfg, duration_s=None, capture_stub: bool = False) -> No
 
     notifier = FanoutNotifier(backends, Path(cfg.dead_letter_path), on_drop=_on_drop)
 
-    # Initial frame + canary check
+    # Verificare inițială: captură un frame, confirmă că canary se potrivește cu calibrarea.
     first_frame = capture()
     if first_frame is None:
         print("WARN: first capture returned None — window/region missing", flush=True)
@@ -674,9 +584,9 @@ async def run_live_async(cfg, duration_s=None, capture_stub: bool = False) -> No
         canary_status = f"drift={first_check.distance}/{cfg.canary.drift_threshold}"
         if first_check.drifted:
             print(f"WARN: canary drift at startup ({canary_status}). Wrong window in front?", flush=True)
-            canary.resume()
+            canary.resume()  # clear the auto-pause so user can Ctrl+C and fix
 
-    dur_note = " dur=∞" if duration_s is None else f" dur={duration_s/3600:.2f}h"
+    dur_note = f" dur=∞" if duration_s is None else f" dur={duration_s/3600:.2f}h"
     notifier.send(Alert(
         kind="heartbeat",
         title="ATM pornit",
@@ -688,148 +598,91 @@ async def run_live_async(cfg, duration_s=None, capture_stub: bool = False) -> No
     audit.log({"event": "started", "config": cfg.config_version, "canary": canary_status})
 
     start = time.monotonic()
-    heartbeat_due = time.monotonic() + cfg.heartbeat_min * 60
+    heartbeat_due = time.time() + 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"
     fires_dir.mkdir(parents=True, exist_ok=True)
+    import cv2  # type: ignore[import-untyped]
+
     try:
-        import cv2  # noqa: F401  fail fast if cv2 is missing  # type: ignore[import-untyped]
+        while duration_s is None or (time.monotonic() - start) < duration_s:
-    except ImportError:
-        pass
-
-    cmd_queue: asyncio.Queue[Command] = asyncio.Queue()
-    first_accepted = True
-    last_saved_color: str | None = None
-    levels_extractor = None
-    fire_count = 0
-
-    def _bound_save(frame: Any, label: str, now: float) -> "Path | None":
-        return _save_annotated_frame(frame, cfg, fires_dir, label, now, audit=audit)
-
-    scheduler = ScreenshotScheduler(
-        capture=capture,
-        save_fn=_bound_save,
-        notifier=notifier,
-        audit=audit,
-    )
-    poller = TelegramPoller(cfg.telegram, cmd_queue, audit)
-
-    # ------------------------------------------------------------------
-    # Nested async coroutines — capture nonlocal state from run_live_async
-    # ------------------------------------------------------------------
-
-    async def _heartbeat_loop() -> None:
-        nonlocal heartbeat_due
-        while True:
-            await asyncio.sleep(60)
-            if time.monotonic() > heartbeat_due:
-                try:
-                    stats = notifier.stats()
-                    audit.log({"ts": time.time(), "event": "notifier_stats", "stats": stats})
-                    body_lines = ["confidence ok"]
-                    for name, s in stats.items():
-                        body_lines.append(
-                            f"{name}: sent={s['sent']} failed={s['failed']} "
-                            f"dropped={s['dropped']} retries={s['retries']}"
-                        )
-                    notifier.send(Alert(kind="heartbeat", title="activ", body="\n".join(body_lines)))
-                except Exception:
-                    notifier.send(Alert(kind="heartbeat", title="activ", body="încredere ok"))
-                heartbeat_due = time.monotonic() + cfg.heartbeat_min * 60
-
-    async def _dispatch_command(cmd: Command) -> None:
-        nonlocal fire_count
-        if cmd.action == "set_interval":
-            secs = cmd.value or cfg.telegram.auto_poll_interval_s
-            scheduler.start(secs)
-            audit.log({"ts": time.time(), "event": "scheduler_started", "reason": "set_interval", "interval_s": secs})
-            notifier.send(Alert(kind="status", title=f"Polling activ — interval {secs // 60} min", body=""))
-        elif cmd.action == "stop":
-            if scheduler.is_running:
-                scheduler.stop()
-                audit.log({"ts": time.time(), "event": "scheduler_stopped", "reason": "command_stop"})
-                notifier.send(Alert(kind="status", title="Polling oprit", body=""))
-            else:
-                notifier.send(Alert(kind="status", title="Polling nu este activ", body=""))
-        elif cmd.action == "status":
-            uptime_s = time.monotonic() - start
-            last_roll = detector.rolling[-1] if detector.rolling else None
-            last_conf = f"{last_roll.match.confidence:.2f}" if last_roll and last_roll.match else "—"
-            last_color = (
-                (last_roll.color or last_roll.match.name) if last_roll and last_roll.match else "—"
-            ) if last_roll else "—"
-            sched_info = (
-                f"activ @{scheduler.interval_s // 60}min" if scheduler.interval_s else "activ"
-            ) if scheduler.is_running else "oprit"
-            canary_info = "drift (pauze)" if canary.is_paused else "ok"
-            body = (
-                f"Stare: {fsm.state.value}\n"
-                f"Ultima detecție: {last_color} (conf {last_conf})\n"
-                f"Uptime: {uptime_s / 3600:.1f}h | Semnale: {fire_count}\n"
-                f"Poller: {sched_info} | Canary: {canary_info}"
-            )
-            notifier.send(Alert(kind="status", title="ATM Status", body=body))
-        elif cmd.action == "ss":
-            now_ss = time.time()
-            frame_ss = await asyncio.to_thread(capture)
-            if frame_ss is None:
-                notifier.send(Alert(
-                    kind="warn",
-                    title="Captură eșuată — verificați fereastra TradeStation",
-                    body="",
-                ))
-                return
-            path_ss = await asyncio.to_thread(
-                _save_annotated_frame, frame_ss, cfg, fires_dir, "ss", now_ss, audit,
-            )
-            audit.log({"ts": now_ss, "event": "screenshot_sent", "path": str(path_ss) if path_ss else None})
-            notifier.send(Alert(kind="screenshot", title="Screenshot manual", body="", image_path=path_ss))
-
-    async def _detection_loop() -> None:
-        nonlocal first_accepted, last_saved_color, levels_extractor, fire_count
-
-        while True:
-            if duration_s is not None and (time.monotonic() - start) >= duration_s:
-                break
-
             now = time.time()
-
+            frame = capture()
-            result: _TickSyncResult = await asyncio.to_thread(
+            if frame is None:
-                _sync_detection_tick,
+                audit.log({"ts": now, "event": "window_lost"})
-                capture, canary, cfg, detector, fsm, notifier, audit, detection_log,
+                time.sleep(cfg.loop_interval_s)
-                fires_dir, first_accepted, last_saved_color, now, samples_dir,
-            )
-
-            if result.first_consumed:
-                first_accepted = False
-            if result.new_color is not None:
-                last_saved_color = result.new_color
-
-            tr = result.tr
-            res = result.res
-
-            if result.late_start or res is None:
-                await asyncio.sleep(cfg.loop_interval_s)
                 continue
+            # canary check
+            cr = canary.check(frame)
+            if canary.is_paused:
+                audit.log({"ts": now, "event": "paused", "drift": cr.distance})
+                time.sleep(cfg.loop_interval_s)
+                continue
+            # detection
+            res = detector.step(now)
+            detection_log.log({
+                "ts": now,
+                "event": "frame",
+                "window_found": res.window_found,
+                "dot_found": res.dot_found,
+                "rgb": list(res.rgb) if res.rgb is not None else None,
+                "match_name": res.match.name if res.match is not None else None,
+                "distance": round(res.match.distance, 2) if res.match is not None else None,
+                "confidence": round(res.match.confidence, 3) if res.match is not None else None,
+                "accepted": res.accepted,
+                "color": res.color,
+            })
+            if res.accepted and res.color:
+                is_first = first_accepted
+                first_accepted = False
 
-            if tr is not None and res.accepted and res.color:
+                # Per-iteration closure — binds current frame/now, gates on config.
-                if tr.reason == "prime" and not scheduler.is_running:
+                def _snapshot(kind: str, label: str) -> "Path | None":
-                    scheduler.start(cfg.telegram.auto_poll_interval_s)
+                    if not getattr(cfg.attach_screenshots, kind, True):
-                    audit.log({"ts": time.time(), "event": "scheduler_started", "reason": "primed"})
+                        return None
-                elif tr.reason in ("fire", "cooled", "phase_skip", "opposite_rearm") and scheduler.is_running:
+                    return _save_annotated_frame(
-                    scheduler.stop()
+                        frame, cfg, fires_dir, label, now, audit=audit,
-                    audit.log({"ts": time.time(), "event": "scheduler_stopped", "reason": tr.reason})
+                    )
 
-            if tr is not None and tr.trigger and not tr.locked:
+                tr = _handle_tick(
-                fire_count += 1
+                    fsm, res.color, now, notifier, audit, is_first,
-                if scheduler.is_running:
+                    snapshot=_snapshot,
-                    scheduler.stop()
+                )
-                    audit.log({"ts": time.time(), "event": "scheduler_stopped", "reason": "fire"})
+                if tr is None:
-                levels_extractor = LevelsExtractor(cfg, tr.trigger, now)
+                    # pornire târzie: FSM neatins, sari peste FIRE + salvare corpus
-
+                    time.sleep(cfg.loop_interval_s)
-            if levels_extractor is not None and result.frame is not None:
+                    continue
-                lr = levels_extractor.step(result.frame, now)
+                # corpus: salvează frame complet la fiecare culoare nouă distinctă, pt etichetare ulterioară
+                if res.color != last_saved_color:
+                    ts_str = datetime.fromtimestamp(now).strftime("%Y%m%d_%H%M%S")
+                    sample_path = samples_dir / f"{ts_str}_{res.color}.png"
+                    try:
+                        cv2.imwrite(str(sample_path), frame)
+                    except Exception as exc:
+                        audit.log({"ts": now, "event": "sample_save_failed", "error": str(exc)})
+                    last_saved_color = res.color
+                # FIRE: adnotează frame-ul + salvează, atașează la alertă
+                if tr.trigger and not tr.locked:
+                    fire_path: "Path | None" = None
+                    if cfg.attach_screenshots.trigger:
+                        fire_path = _save_annotated_frame(
+                            frame, cfg, fires_dir, tr.trigger, now, audit=audit,
+                        )
+                    notifier.send(Alert(
+                        kind="trigger",
+                        title=f"Semnal {tr.trigger}",
+                        body=f"@ {datetime.fromtimestamp(now).isoformat(timespec='seconds')}",
+                        image_path=fire_path,
+                        direction=tr.trigger,
+                    ))
+                    levels_extractor = LevelsExtractor(cfg, tr.trigger, now)
+            # phase-B levels
+            if levels_extractor is not None:
+                lr = levels_extractor.step(frame, now)
                 if lr.status in ("complete", "timeout"):
                     if lr.status == "complete" and lr.levels:
                         notifier.send(Alert(
@@ -842,45 +695,24 @@ async def run_live_async(cfg, duration_s=None, capture_stub: bool = False) -> No
                             ),
                         ))
                     levels_extractor = None
-
+            # heartbeat — include statistici per-backend ca eșecurile silențioase
-            while True:
+            # să apară la fiecare 30 min fără să aștepte oprirea.
+            if time.time() > heartbeat_due:
                 try:
-                    cmd = cmd_queue.get_nowait()
+                    stats = notifier.stats()
-                except asyncio.QueueEmpty:
+                    audit.log({"ts": time.time(), "event": "notifier_stats", "stats": stats})
-                    break
+                    body_lines = ["confidence ok"]
-                try:
+                    for name, s in stats.items():
-                    await _dispatch_command(cmd)
+                        body_lines.append(
-                except Exception as _cmd_exc:
+                            f"{name}: sent={s['sent']} failed={s['failed']} "
-                    _msg = f"/{cmd.action}: {_cmd_exc}"
+                            f"dropped={s['dropped']} retries={s['retries']}"
-                    audit.log({"ts": time.time(), "event": "command_error", "action": cmd.action, "error": str(_cmd_exc)})
+                        )
-                    print(f"ERR command_dispatch {_msg}", flush=True)
+                    notifier.send(Alert(kind="heartbeat", title="activ", body="\n".join(body_lines)))
-                    notifier.send(Alert(kind="warn", title=f"Eroare comandă /{cmd.action}", body=str(_cmd_exc)))
+                except Exception:
-
+                    notifier.send(Alert(kind="heartbeat", title="activ", body="încredere ok"))
-            await asyncio.sleep(cfg.loop_interval_s)
+                heartbeat_due = time.time() + cfg.heartbeat_min * 60
-
+            time.sleep(cfg.loop_interval_s)
-    # Launch background tasks
-    t_scheduler = asyncio.create_task(scheduler.run(), name="scheduler")
-    t_poller = asyncio.create_task(poller.run(), name="poller")
-    t_heartbeat = asyncio.create_task(_heartbeat_loop(), name="heartbeat")
-
-    try:
-        await _detection_loop()
     finally:
-        # 7-step graceful shutdown
-        # 1. cancel scheduler
-        t_scheduler.cancel()
-        with contextlib.suppress(asyncio.CancelledError, Exception):
-            await t_scheduler
-        # 2. cancel poller
-        t_poller.cancel()
-        with contextlib.suppress(asyncio.CancelledError, Exception):
-            await t_poller
-        # 3. cancel heartbeat
-        t_heartbeat.cancel()
-        with contextlib.suppress(asyncio.CancelledError, Exception):
-            await t_heartbeat
-        # 4. drain detection — complete (we awaited _detection_loop directly)
-        # 5. send shutdown alert
         try:
             stats = notifier.stats()
             lines = [f"după {time.monotonic() - start:.0f}s"]
@@ -889,12 +721,13 @@ async def run_live_async(cfg, duration_s=None, capture_stub: bool = False) -> No
                     f"{name}: sent={s['sent']} failed={s['failed']} "
                     f"dropped={s['dropped']} retries={s['retries']}"
                 )
-            notifier.send(Alert(kind="heartbeat", title="ATM oprit", body="\n".join(lines)))
+            notifier.send(Alert(
+                kind="heartbeat", title="ATM oprit",
+                body="\n".join(lines),
+            ))
         except Exception:
             pass
-        # 6. notifier.stop() — flush + join FanoutNotifier threads
         notifier.stop()
-        # 7. audit.close()
         audit.close()
         detection_log.close()
 

src/atm/notifier/__init__.py

@@ -5,13 +5,11 @@ from typing import Protocol
 
 @dataclass
 class Alert:
-    # flat union: "trigger"|"heartbeat"|"levels"|"warn"|"arm"|"prime"|"late_start"|"screenshot"|"status"
+    kind: str          # "trigger" | "heartbeat" | "levels" | "warn" | "arm" | "prime" | "late_start"
-    kind: str
     title: str
     body: str
     image_path: Path | None = None  # annotated screenshot
     direction: str | None = None    # "BUY"/"SELL" when kind=trigger
-    silent: bool = False            # disable_notification for Telegram; ignored by Discord
 
 
 class Notifier(Protocol):

src/atm/notifier/telegram.py

@@ -33,7 +33,6 @@ class TelegramNotifier:
                         "chat_id": self._chat_id,
                         "caption": text,
                         "parse_mode": "HTML",
-                        "disable_notification": str(alert.silent).lower(),
                     },
                     files={"photo": fh},
                     timeout=10,
@@ -45,7 +44,6 @@ class TelegramNotifier:
                     "chat_id": self._chat_id,
                     "text": text,
                     "parse_mode": "HTML",
-                    "disable_notification": alert.silent,
                 },
                 timeout=10,
             )

src/atm/scheduler.py

@@ -1,118 +0,0 @@
-"""ScreenshotScheduler — periodic capture + annotate + send.
-
-Runs as an asyncio task. capture() and cv2 work execute in asyncio.to_thread
-to avoid blocking the event loop. Decision 13: scheduler calls capture()
-directly, NOT via Detector.
-"""
-from __future__ import annotations
-
-import asyncio
-import logging
-import time
-from pathlib import Path
-from typing import Callable
-
-from .notifier import Alert
-
-logger = logging.getLogger(__name__)
-
-
-class ScreenshotScheduler:
-    """Periodic screenshot sender.
-
-    Constructor params are explicit (decision 11 outside-voice finding).
-    """
-
-    def __init__(
-        self,
-        capture: Callable,               # () -> ndarray | None
-        save_fn: Callable,               # (frame, label, now) -> Path | None
-        notifier,                        # _NotifierLike
-        audit,                           # _AuditLike
-        interval_s: int | None = None,
-    ) -> None:
-        self._capture = capture
-        self._save_fn = save_fn
-        self._notifier = notifier
-        self._audit = audit
-        self._interval_s = interval_s
-        self._is_running = False
-        self._next_due: float | None = None  # monotonic
-
-    # ------------------------------------------------------------------
-    # Public state
-    # ------------------------------------------------------------------
-
-    @property
-    def is_running(self) -> bool:
-        return self._is_running
-
-    @property
-    def interval_s(self) -> int | None:
-        return self._interval_s
-
-    @property
-    def next_due(self) -> float | None:
-        return self._next_due
-
-    # ------------------------------------------------------------------
-    # Control (called from async event loop)
-    # ------------------------------------------------------------------
-
-    def start(self, interval_s: int) -> None:
-        self._interval_s = interval_s
-        self._is_running = True
-        self._next_due = time.monotonic() + interval_s
-
-    def stop(self) -> None:
-        self._is_running = False
-        self._next_due = None
-
-    # ------------------------------------------------------------------
-    # Task body
-    # ------------------------------------------------------------------
-
-    async def run(self) -> None:
-        """Runs until cancelled."""
-        while True:
-            await asyncio.sleep(1)
-            if not self._is_running or self._next_due is None:
-                continue
-            if time.monotonic() >= self._next_due:
-                await self._take_screenshot()
-                if self._is_running and self._interval_s is not None:
-                    self._next_due = time.monotonic() + self._interval_s
-
-    async def _take_screenshot(self) -> None:
-        now = time.time()
-        try:
-            frame = await asyncio.to_thread(self._capture)
-        except Exception as exc:
-            logger.warning("ScreenshotScheduler capture failed: %s", exc)
-            self._audit.log({"ts": now, "event": "screenshot_sent", "status": "capture_failed", "error": str(exc)})
-            self._notifier.send(Alert(
-                kind="warn",
-                title="Captură eșuată — verificați fereastra TradeStation",
-                body="",
-                silent=True,
-            ))
-            return
-
-        if frame is None:
-            self._notifier.send(Alert(
-                kind="warn",
-                title="Captură eșuată — verificați fereastra TradeStation",
-                body="",
-                silent=True,
-            ))
-            return
-
-        path = await asyncio.to_thread(self._save_fn, frame, "poll", now)
-        self._audit.log({"ts": now, "event": "screenshot_sent", "path": str(path) if path else None})
-        self._notifier.send(Alert(
-            kind="screenshot",
-            title="Screenshot periodic",
-            body="",
-            image_path=path,
-            silent=True,
-        ))

src/atm/validate.py

@@ -0,0 +1,229 @@
+"""Offline calibration validation: run Detector on labeled frames, report PASS/FAIL.
+
+Used by the `atm validate-calibration` subcommand. Reports per-sample detection
+results against expected labels, and for failures, computes RGB distance to
+each color threshold and emits tuning suggestions.
+
+Reuses `Detector.step(frame)` - does NOT reimplement color classification.
+"""
+from __future__ import annotations
+
+import json
+import math
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from .config import Config
+
+
+@dataclass
+class SampleRecord:
+    path: str
+    expected: str
+    detected: str | None
+    confidence: float
+    rgb: tuple[int, int, int] | None
+    top3: list[tuple[str, float]]  # [(name, score), ...] ranked by RGB distance
+    passed: bool
+    note: str = ""
+    error: str | None = None  # non-None if frame load failed / schema bad
+
+
+@dataclass
+class ValidationReport:
+    records: list[SampleRecord] = field(default_factory=list)
+    config_name: str = ""
+
+    @property
+    def total(self) -> int:
+        return len(self.records)
+
+    @property
+    def passed(self) -> int:
+        return sum(1 for r in self.records if r.passed)
+
+    @property
+    def failed(self) -> int:
+        return self.total - self.passed
+
+    @property
+    def all_pass(self) -> bool:
+        return self.total > 0 and self.failed == 0
+
+    def render(self) -> str:
+        lines: list[str] = []
+        hdr = f"Testing {self.total} frames"
+        if self.config_name:
+            hdr += f" against config {self.config_name}"
+        hdr += "..."
+        lines.append(hdr)
+        lines.append("")
+
+        for r in self.records:
+            name = Path(r.path).name or r.path
+            if r.error:
+                lines.append(f"  [FAIL] {name}")
+                lines.append(f"         error: {r.error}")
+                continue
+            tag = "PASS" if r.passed else "FAIL"
+            rgb_str = f"RGB {r.rgb}" if r.rgb is not None else "RGB n/a"
+            detected = r.detected if r.detected is not None else "none"
+            lines.append(f"  [{tag}] {name}")
+            lines.append(
+                f"         expected={r.expected}  detected={detected}  "
+                f"(conf {r.confidence:.2f}, {rgb_str})"
+            )
+            if not r.passed and r.top3:
+                top3_str = " ".join(f"{n}({c:.2f})" for n, c in r.top3)
+                lines.append(f"         Top 3 candidates: {top3_str}")
+
+        lines.append("")
+        pct = (self.passed / self.total * 100.0) if self.total else 0.0
+        lines.append(f"SUMMARY: {self.passed}/{self.total} PASS ({pct:.0f}%)")
+
+        fails = [r for r in self.records if not r.passed]
+        if fails:
+            lines.append("FAILED:")
+            for r in fails:
+                name = Path(r.path).name or r.path
+                if r.error:
+                    lines.append(f"  - {name}: {r.error}")
+                    continue
+                detected = r.detected if r.detected is not None else "none"
+                lines.append(
+                    f"  - {name}: expected {r.expected}, got {detected}"
+                )
+
+            sug_lines = [
+                r._suggestion  # type: ignore[attr-defined]
+                for r in fails
+                if getattr(r, "_suggestion", "")
+            ]
+            if sug_lines:
+                lines.append("")
+                lines.append("SUGGESTIONS:")
+                for s in sug_lines:
+                    lines.append(f"  - {s}")
+
+        return "\n".join(lines)
+
+    def __str__(self) -> str:
+        return self.render()
+
+
+class ValidationError(Exception):
+    """Raised for missing label files or invalid schema."""
+
+
+def _rgb_distance(a: tuple[int, int, int], b: tuple[int, int, int]) -> float:
+    return math.sqrt(sum((a[i] - b[i]) ** 2 for i in range(3)))
+
+
+def _load_labels(label_file: Path) -> list[dict[str, Any]]:
+    if not label_file.exists():
+        raise ValidationError(f"label file not found: {label_file}")
+    try:
+        data = json.loads(label_file.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise ValidationError(f"invalid JSON in {label_file}: {exc}") from exc
+    if not isinstance(data, list):
+        raise ValidationError(
+            f"label file must be a JSON array; got {type(data).__name__}"
+        )
+    return data
+
+
+def validate_calibration(
+    label_file: Path,
+    cfg: Config,
+    config_name: str = "",
+) -> ValidationReport:
+    """Run Detector on each labeled frame; return a ValidationReport.
+
+    Reuses `Detector.step(frame)`. Loads frames via cv2.imread.
+    Raises ValidationError if the label file is missing or malformed.
+    """
+    import cv2  # local import keeps module import cheap
+    from .detector import Detector
+
+    entries = _load_labels(label_file)
+    report = ValidationReport(config_name=config_name)
+
+    palette = {
+        name: spec.rgb
+        for name, spec in cfg.colors.items()
+        if name != "background"
+    }
+
+    detector = Detector(cfg=cfg, capture=lambda: None)
+
+    for entry in entries:
+        path = str(entry.get("path", ""))
+        expected = str(entry.get("expected", ""))
+        note = str(entry.get("note", ""))
+
+        if not path or not expected:
+            rec = SampleRecord(
+                path=path, expected=expected, detected=None, confidence=0.0,
+                rgb=None, top3=[], passed=False, note=note,
+                error="missing 'path' or 'expected' field",
+            )
+            rec._suggestion = ""  # type: ignore[attr-defined]
+            report.records.append(rec)
+            continue
+
+        frame = cv2.imread(path)
+        if frame is None:
+            rec = SampleRecord(
+                path=path, expected=expected, detected=None, confidence=0.0,
+                rgb=None, top3=[], passed=False, note=note,
+                error=f"cv2.imread failed for {path}",
+            )
+            rec._suggestion = ""  # type: ignore[attr-defined]
+            report.records.append(rec)
+            continue
+
+        result = detector.step(ts=0.0, frame=frame)
+
+        match = result.match
+        if match is None:
+            detected: str | None = None
+            confidence = 0.0
+        else:
+            detected = match.name if match.name != "UNKNOWN" else None
+            confidence = match.confidence
+
+        rgb = result.rgb
+
+        # Top 3 candidates: rank palette entries by RGB distance to observed.
+        top3: list[tuple[str, float]] = []
+        if rgb is not None:
+            scored: list[tuple[str, float]] = []
+            for name, ref in palette.items():
+                scored.append((name, _rgb_distance(rgb, ref)))
+            scored.sort(key=lambda t: t[1])
+            top3 = [(n, 1.0 / (1.0 + d / 20.0)) for n, d in scored[:3]]
+
+        passed = detected == expected
+
+        rec = SampleRecord(
+            path=path, expected=expected, detected=detected,
+            confidence=confidence, rgb=rgb, top3=top3, passed=passed, note=note,
+        )
+
+        if not passed and rgb is not None and expected in palette:
+            ref = palette[expected]
+            tol = cfg.colors[expected].tolerance
+            dist = _rgb_distance(rgb, ref)
+            rec._suggestion = (  # type: ignore[attr-defined]
+                f"{expected} praguri curente: RGB{ref} +/- {tol:.0f}. "
+                f"Pixelul observat {rgb} e la distanta {dist:.1f} "
+                f"-> recalibreaza cu acest sample."
+            )
+        else:
+            rec._suggestion = ""  # type: ignore[attr-defined]
+
+        report.records.append(rec)
+
+    return report

tests/test_main.py

@@ -1,7 +1,6 @@
 """Tests for atm.main unified CLI."""
 from __future__ import annotations
 
-import asyncio
 import os
 import subprocess
 import sys
@@ -187,7 +186,7 @@ def test_run_live_catchup_sell_from_gray_then_dark_red(monkeypatch, tmp_path):
         ]
         def __init__(self, *a, **kw):
             self._i = 0
-        def step(self, ts, frame=None):
+        def step(self, ts):
             if self._i >= len(self._script):
                 raise _StopLoop
             color, accepted = self._script[self._i]
@@ -229,17 +228,6 @@ def test_run_live_catchup_sell_from_gray_then_dark_red(monkeypatch, tmp_path):
         def step(self, *a, **kw):
             return types.SimpleNamespace(status="pending", levels=None)
 
-    class _StubPoller:
-        def __init__(self, *a, **kw): pass
-        async def run(self): await asyncio.sleep(9999)
-
-    class _StubScheduler:
-        def __init__(self, *a, **kw):
-            self.is_running = False
-        def start(self, interval_s): self.is_running = True
-        def stop(self): self.is_running = False
-        async def run(self): await asyncio.sleep(9999)
-
     monkeypatch.setattr("atm.detector.Detector", ScriptedDetector)
     monkeypatch.setattr("atm.canary.Canary", FakeCanary)
     monkeypatch.setattr("atm.notifier.fanout.FanoutNotifier", FakeFanout)
@@ -249,8 +237,6 @@ def test_run_live_catchup_sell_from_gray_then_dark_red(monkeypatch, tmp_path):
     monkeypatch.setattr("atm.levels.LevelsExtractor", _Stub)
     monkeypatch.setattr("atm.main._build_capture", fake_build_capture)
     monkeypatch.setattr("atm.main.time.sleep", lambda s: None)
-    monkeypatch.setattr("atm.commands.TelegramPoller", _StubPoller)
-    monkeypatch.setattr("atm.scheduler.ScreenshotScheduler", _StubScheduler)
 
     with pytest.raises(_StopLoop):
         _main.run_live(cfg, duration_s=None)
@@ -269,135 +255,3 @@ def test_run_live_catchup_sell_from_gray_then_dark_red(monkeypatch, tmp_path):
 
     assert len(trigger) == 1
     assert trigger[0].direction == "SELL"
-
-
-# ---------------------------------------------------------------------------
-# MUST-HAVE: async lifecycle integration test
-# IDLE → ARMED → PRIMED (auto-poll scheduler starts) → FIRE (scheduler stops)
-# Tests: scheduler starts on prime, stops on fire, fire alert sent.
-# ---------------------------------------------------------------------------
-
-@pytest.mark.asyncio
-async def test_lifecycle_idle_armed_primed_autopoll_fire_stop(monkeypatch, tmp_path):
-    import numpy as np
-    import atm.main as _main
-    from atm.detector import DetectionResult
-
-    captured_alerts: list = []
-    scheduler_events: list[str] = []
-
-    class FakeFanout:
-        def __init__(self, *a, **kw): pass
-        def send(self, alert): captured_alerts.append(alert)
-        def stop(self): pass
-        def stats(self): return {}
-
-    class FakeCanaryResult:
-        distance = 0
-        drifted = False
-        paused = False
-
-    class FakeCanary:
-        def __init__(self, *a, **kw): self.is_paused = False
-        def check(self, frame): return FakeCanaryResult()
-        def resume(self): pass
-
-    # Scheduler tracks start/stop calls
-    class FakeScheduler:
-        def __init__(self, *a, **kw):
-            self.is_running = False
-            self.interval_s = None
-        def start(self, interval_s):
-            self.is_running = True
-            self.interval_s = interval_s
-            scheduler_events.append(f"start:{interval_s}")
-        def stop(self):
-            self.is_running = False
-            scheduler_events.append("stop")
-        async def run(self):
-            await asyncio.sleep(9999)
-
-    class FakePoller:
-        def __init__(self, *a, **kw): pass
-        async def run(self): await asyncio.sleep(9999)
-
-    class _StopLoop(Exception): pass
-
-    class ScriptedDetector:
-        # turquoise→ARM, dark_green→PRIME, light_green→FIRE
-        _script = [
-            ("turquoise", True),
-            ("dark_green", True),
-            ("light_green", True),
-        ]
-        def __init__(self, *a, **kw): self._i = 0
-        def step(self, ts, frame=None):
-            if self._i >= len(self._script):
-                raise _StopLoop
-            color, accepted = self._script[self._i]
-            self._i += 1
-            return DetectionResult(ts=ts, window_found=True, dot_found=True,
-                                   rgb=(1, 1, 1), match=None, accepted=accepted, color=color)
-        @property
-        def rolling(self): return []
-
-    def fake_build_capture(cfg, capture_stub=False):
-        return lambda: np.zeros((50, 50, 3), dtype=np.uint8)
-
-    cfg = MagicMock()
-    cfg.lockout_s = 60
-    cfg.heartbeat_min = 999
-    cfg.loop_interval_s = 0
-    cfg.config_version = "test"
-    cfg.dead_letter_path = str(tmp_path / "dl.jsonl")
-    cfg.canary.drift_threshold = 10
-    cfg.dot_roi.x = 0; cfg.dot_roi.y = 0; cfg.dot_roi.w = 10; cfg.dot_roi.h = 10
-    cfg.chart_window_region = None
-    cfg.telegram.auto_poll_interval_s = 180
-    cfg.telegram.bot_token = "tok"
-    cfg.telegram.chat_id = "123"
-    cfg.telegram.allowed_chat_ids = ("123",)
-
-    fake_sched = FakeScheduler()
-
-    monkeypatch.chdir(tmp_path)
-
-    class _Stub:
-        def __init__(self, *a, **kw): pass
-        def log(self, *a, **kw): pass
-        def close(self, *a, **kw): pass
-        def step(self, *a, **kw): return types.SimpleNamespace(status="pending", levels=None)
-
-    monkeypatch.setattr("atm.detector.Detector", ScriptedDetector)
-    monkeypatch.setattr("atm.canary.Canary", FakeCanary)
-    monkeypatch.setattr("atm.notifier.fanout.FanoutNotifier", FakeFanout)
-    monkeypatch.setattr("atm.notifier.discord.DiscordNotifier", _Stub)
-    monkeypatch.setattr("atm.notifier.telegram.TelegramNotifier", _Stub)
-    monkeypatch.setattr("atm.audit.AuditLog", _Stub)
-    monkeypatch.setattr("atm.levels.LevelsExtractor", _Stub)
-    monkeypatch.setattr("atm.main._build_capture", fake_build_capture)
-    monkeypatch.setattr("atm.commands.TelegramPoller", FakePoller)
-    monkeypatch.setattr("atm.scheduler.ScreenshotScheduler", lambda *a, **kw: fake_sched)
-
-    with pytest.raises(_StopLoop):
-        await _main.run_live_async(cfg, duration_s=None)
-
-    arm_alerts = [a for a in captured_alerts if a.kind == "arm"]
-    prime_alerts = [a for a in captured_alerts if a.kind == "prime"]
-    trigger_alerts = [a for a in captured_alerts if a.kind == "trigger"]
-
-    assert len(arm_alerts) == 1, f"expected 1 arm, got {[a.title for a in captured_alerts]}"
-    assert arm_alerts[0].direction == "BUY"
-
-    assert len(prime_alerts) == 1
-    assert prime_alerts[0].direction == "BUY"
-
-    assert len(trigger_alerts) == 1
-    assert trigger_alerts[0].direction == "BUY"
-
-    # Scheduler must have started (on PRIME) and stopped (on FIRE)
-    assert "start:180" in scheduler_events, f"scheduler not started: {scheduler_events}"
-    assert "stop" in scheduler_events, f"scheduler not stopped: {scheduler_events}"
-    start_idx = scheduler_events.index("start:180")
-    stop_idx = scheduler_events.index("stop")
-    assert start_idx < stop_idx, "scheduler started after it stopped"

tests/test_validate.py

@@ -0,0 +1,214 @@
+"""Tests for atm.validate — offline calibration validation.
+
+Covers the 3 tests from plan section D':
+  17. test_validate_calibration_pass
+  18. test_validate_calibration_fail_reports_top_candidates
+  19. test_validate_calibration_file_not_found
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from atm.config import (
+    CanaryRegion,
+    ColorSpec,
+    Config,
+    DiscordCfg,
+    ROI,
+    TelegramCfg,
+    YAxisCalib,
+)
+from atm.detector import DetectionResult
+from atm.vision import ColorMatch
+
+
+def _make_config() -> Config:
+    """Minimal Config with a palette large enough to support top-3 candidates."""
+    colors = {
+        "turquoise":   ColorSpec(rgb=(0, 200, 200),    tolerance=30),
+        "yellow":      ColorSpec(rgb=(255, 255, 0),    tolerance=30),
+        "dark_green":  ColorSpec(rgb=(0, 100, 0),      tolerance=30),
+        "dark_red":    ColorSpec(rgb=(165, 42, 42),    tolerance=30),
+        "light_green": ColorSpec(rgb=(144, 238, 144),  tolerance=30),
+        "light_red":   ColorSpec(rgb=(255, 182, 193),  tolerance=30),
+        "gray":        ColorSpec(rgb=(128, 128, 128),  tolerance=30),
+        "background":  ColorSpec(rgb=(18, 18, 18),     tolerance=15),
+    }
+    return Config(
+        window_title="test",
+        dot_roi=ROI(x=0, y=0, w=100, h=100),
+        chart_roi=ROI(x=0, y=0, w=100, h=100),
+        colors=colors,
+        y_axis=YAxisCalib(p1_y=0, p1_price=100.0, p2_y=100, p2_price=0.0),
+        canary=CanaryRegion(
+            roi=ROI(x=0, y=0, w=10, h=10),
+            baseline_phash="0" * 64,
+        ),
+        discord=DiscordCfg(webhook_url="http://localhost/fake"),
+        telegram=TelegramCfg(bot_token="fake_token", chat_id="123"),
+        debounce_depth=1,
+    )
+
+
+def _write_labels(tmp_path: Path, entries: list[dict]) -> Path:
+    f = tmp_path / "labels.json"
+    f.write_text(json.dumps(entries), encoding="utf-8")
+    return f
+
+
+def _write_blank_png(tmp_path: Path, name: str) -> Path:
+    """Write a trivially-valid 10x10 BGR image so cv2.imread returns non-None."""
+    import cv2
+    p = tmp_path / name
+    arr = np.zeros((10, 10, 3), dtype=np.uint8)
+    cv2.imwrite(str(p), arr)
+    return p
+
+
+# ---------------------------------------------------------------------------
+# Test 17: PASS path — mocked Detector.step returns expected color
+# ---------------------------------------------------------------------------
+
+def test_validate_calibration_pass(monkeypatch, tmp_path):
+    from atm import validate as validate_mod
+
+    img_path = _write_blank_png(tmp_path, "yellow_sample.png")
+    labels = _write_labels(
+        tmp_path,
+        [{"path": str(img_path), "expected": "yellow", "note": "test"}],
+    )
+
+    def fake_step(self, ts, frame=None):
+        return DetectionResult(
+            ts=ts,
+            window_found=True,
+            dot_found=True,
+            rgb=(250, 250, 5),
+            match=ColorMatch(name="yellow", distance=6.0, confidence=0.94),
+            accepted=True,
+            color="yellow",
+        )
+
+    monkeypatch.setattr("atm.detector.Detector.step", fake_step)
+
+    report = validate_mod.validate_calibration(labels, _make_config())
+
+    assert report.total == 1
+    assert report.passed == 1
+    assert report.failed == 0
+    assert report.all_pass is True
+    rec = report.records[0]
+    assert rec.passed is True
+    assert rec.detected == "yellow"
+    assert rec.expected == "yellow"
+    assert "[PASS]" in report.render()
+
+    # CLI wiring: exit 0
+    import atm.main as _main
+
+    class _Args:
+        label_file = labels
+
+    monkeypatch.setattr("atm.config.Config.load_current", classmethod(lambda cls, d: _make_config()))
+    with pytest.raises(SystemExit) as exc_info:
+        _main._cmd_validate_calibration(_Args())
+    assert exc_info.value.code == 0
+
+
+# ---------------------------------------------------------------------------
+# Test 18: FAIL path — Detector returns wrong color; report lists top 3
+#                      candidates and a SUGGESTIONS line with RGB distance.
+# ---------------------------------------------------------------------------
+
+def test_validate_calibration_fail_reports_top_candidates(monkeypatch, tmp_path):
+    from atm import validate as validate_mod
+
+    img_path = _write_blank_png(tmp_path, "dark_red_sample.png")
+    labels = _write_labels(
+        tmp_path,
+        [{"path": str(img_path), "expected": "dark_red", "note": "missed dark_red"}],
+    )
+
+    # Observed RGB closer to gray than dark_red (like the real 2026-04-17 miss).
+    def fake_step(self, ts, frame=None):
+        return DetectionResult(
+            ts=ts,
+            window_found=True,
+            dot_found=True,
+            rgb=(135, 62, 67),
+            match=ColorMatch(name="gray", distance=45.0, confidence=0.12),
+            accepted=True,
+            color="gray",
+        )
+
+    monkeypatch.setattr("atm.detector.Detector.step", fake_step)
+
+    report = validate_mod.validate_calibration(labels, _make_config())
+
+    assert report.total == 1
+    assert report.failed == 1
+    assert report.all_pass is False
+
+    rec = report.records[0]
+    assert rec.passed is False
+    assert rec.detected == "gray"
+    assert rec.expected == "dark_red"
+    # Top 3 candidates populated (name, score) sorted by RGB distance.
+    assert len(rec.top3) == 3
+    names = [n for n, _ in rec.top3]
+    # dark_red should appear in top candidates since observed RGB(135,62,67)
+    # is reasonably close to dark_red(165,42,42).
+    assert "dark_red" in names
+
+    rendered = report.render()
+    assert "[FAIL]" in rendered
+    assert "Top 3 candidates:" in rendered
+    assert "SUGGESTIONS:" in rendered
+    # The suggestion must mention the expected color's RGB and the measured distance.
+    assert "dark_red" in rendered
+    assert "(165, 42, 42)" in rendered
+
+    # CLI wiring: exit 1
+    import atm.main as _main
+
+    class _Args:
+        label_file = labels
+
+    monkeypatch.setattr("atm.config.Config.load_current", classmethod(lambda cls, d: _make_config()))
+    with pytest.raises(SystemExit) as exc_info:
+        _main._cmd_validate_calibration(_Args())
+    assert exc_info.value.code == 1
+
+
+# ---------------------------------------------------------------------------
+# Test 19: missing label file — clean error, non-zero exit, no stack trace
+# ---------------------------------------------------------------------------
+
+def test_validate_calibration_file_not_found(monkeypatch, tmp_path, capsys):
+    from atm import validate as validate_mod
+
+    missing = tmp_path / "nope.json"
+
+    # Library-level: raises ValidationError (not bare FileNotFoundError).
+    with pytest.raises(validate_mod.ValidationError) as exc_info:
+        validate_mod.validate_calibration(missing, _make_config())
+    assert "not found" in str(exc_info.value).lower()
+
+    # CLI-level: graceful sys.exit with non-zero code, message on stderr.
+    import atm.main as _main
+
+    class _Args:
+        label_file = missing
+
+    monkeypatch.setattr("atm.config.Config.load_current", classmethod(lambda cls, d: _make_config()))
+    with pytest.raises(SystemExit) as exc_info:
+        _main._cmd_validate_calibration(_Args())
+    assert exc_info.value.code != 0
+    err = capsys.readouterr().err
+    assert "not found" in err.lower()
+    # Ensure no python traceback leaked through.
+    assert "Traceback" not in err