Add OperatingHoursCfg (enabled/timezone/weekdays/start_hhmm/stop_hhmm) so
the run loop can align with NYSE session hours instead of the user's
local wall clock (fixes DST drift between NY and Europe/Bucharest).
- Config parses [options.operating_hours] and resolves ZoneInfo at load,
fail-fast on invalid tz or weekday names. The tz is cached on
_tz_cache so the detection loop pays zero per-tick cost.
- LifecycleState tracks user_paused + last_window_state across ticks.
- Module-scope _should_skip(now, state, cfg, canary) returns skip reason
or None. Weekday check uses datetime.weekday() + a fixed MON..SUN list
(locale-free; strftime('%a') is localized).
- _maybe_log_transition emits market_open / market_closed once per edge.
R2: when last_window_state is None (startup), just seed — do not send
a spurious market_open alert when run_live_async launches in-window.
- _run_tick consults the lifecycle guard before scheduling the heavy
detection thread, so drain + transition logging still happen when the
tick is skipped.
- CLI flags --tz / --weekdays / --oh-start / --oh-stop override TOML.
(Kept distinct from the existing --start-at/--stop-at sleep-until-time
semantics to avoid breaking current deployments — deviation noted.)
- configs/example.toml documents the new [options.operating_hours] table.
Tests: parametrized window matrix (tests #8), transition logging (#9),
notification side-effect (#10), R2 #20 startup suppression, R2 #22
locale-independent weekday, plus guards for user_paused / canary
precedence and config-parse error paths.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
156 lines
5.7 KiB
Python
156 lines
5.7 KiB
Python
"""Tests for atm.config — focused on attach_screenshots parsing (legacy bool vs new dict)."""
|
|
from __future__ import annotations
|
|
|
|
from atm.config import AlertsCfg, Config
|
|
|
|
|
|
_BASE = {
|
|
"window_title": "X",
|
|
"dot_roi": {"x": 0, "y": 0, "w": 10, "h": 10},
|
|
"chart_roi": {"x": 0, "y": 0, "w": 100, "h": 100},
|
|
"colors": {
|
|
"turquoise": {"rgb": [64, 224, 208], "tolerance": 30.0},
|
|
"yellow": {"rgb": [255, 215, 0], "tolerance": 30.0},
|
|
"dark_green": {"rgb": [0, 100, 0], "tolerance": 30.0},
|
|
"dark_red": {"rgb": [139, 0, 0], "tolerance": 30.0},
|
|
"light_green": {"rgb": [0, 230, 118], "tolerance": 30.0},
|
|
"light_red": {"rgb": [255, 82, 82], "tolerance": 30.0},
|
|
"gray": {"rgb": [128, 128, 128], "tolerance": 25.0},
|
|
},
|
|
"y_axis": {"p1_y": 100, "p1_price": 485.0, "p2_y": 200, "p2_price": 484.0},
|
|
"canary": {
|
|
"roi": {"x": 0, "y": 0, "w": 10, "h": 10},
|
|
"baseline_phash": "0" * 16,
|
|
"drift_threshold": 8,
|
|
},
|
|
"discord": {"webhook_url": "https://example.com/hook"},
|
|
"telegram": {"bot_token": "tok", "chat_id": "123"},
|
|
}
|
|
|
|
|
|
def _with_opts(opts: dict) -> dict:
|
|
d = {k: v for k, v in _BASE.items()}
|
|
d["options"] = opts
|
|
return d
|
|
|
|
|
|
def test_attach_screenshots_default_all_true() -> None:
|
|
"""Missing attach_screenshots → all fields True."""
|
|
cfg = Config._from_dict(_with_opts({}))
|
|
assert cfg.attach_screenshots == AlertsCfg(
|
|
late_start=True, catchup=True, arm=True, prime=True, trigger=True,
|
|
)
|
|
|
|
|
|
def test_attach_screenshots_legacy_bool_true() -> None:
|
|
"""Legacy: attach_screenshots = true → all fields True."""
|
|
cfg = Config._from_dict(_with_opts({"attach_screenshots": True}))
|
|
assert cfg.attach_screenshots.arm is True
|
|
assert cfg.attach_screenshots.catchup is True
|
|
assert cfg.attach_screenshots.trigger is True
|
|
|
|
|
|
def test_attach_screenshots_legacy_bool_false() -> None:
|
|
"""Legacy: attach_screenshots = false → all fields False."""
|
|
cfg = Config._from_dict(_with_opts({"attach_screenshots": False}))
|
|
assert cfg.attach_screenshots.arm is False
|
|
assert cfg.attach_screenshots.catchup is False
|
|
assert cfg.attach_screenshots.trigger is False
|
|
assert cfg.attach_screenshots.late_start is False
|
|
|
|
|
|
def test_attach_screenshots_partial_dict() -> None:
|
|
"""Dict form with only some keys → specified False, others default True."""
|
|
cfg = Config._from_dict(_with_opts({
|
|
"attach_screenshots": {"arm": False, "prime": False},
|
|
}))
|
|
assert cfg.attach_screenshots.arm is False
|
|
assert cfg.attach_screenshots.prime is False
|
|
# Unspecified → dataclass default True
|
|
assert cfg.attach_screenshots.trigger is True
|
|
assert cfg.attach_screenshots.catchup is True
|
|
assert cfg.attach_screenshots.late_start is True
|
|
|
|
|
|
def test_attach_screenshots_full_dict() -> None:
|
|
"""Dict form with all keys specified."""
|
|
cfg = Config._from_dict(_with_opts({
|
|
"attach_screenshots": {
|
|
"late_start": False,
|
|
"catchup": True,
|
|
"arm": False,
|
|
"prime": True,
|
|
"trigger": True,
|
|
},
|
|
}))
|
|
assert cfg.attach_screenshots.late_start is False
|
|
assert cfg.attach_screenshots.catchup is True
|
|
assert cfg.attach_screenshots.arm is False
|
|
assert cfg.attach_screenshots.prime is True
|
|
assert cfg.attach_screenshots.trigger is True
|
|
|
|
|
|
def test_attach_screenshots_unknown_keys_ignored() -> None:
|
|
"""Unknown keys are silently dropped (dataclass won't accept them)."""
|
|
cfg = Config._from_dict(_with_opts({
|
|
"attach_screenshots": {"arm": False, "nonexistent_knob": True},
|
|
}))
|
|
assert cfg.attach_screenshots.arm is False
|
|
# Should not raise even with unknown key
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commit 3: AlertBehaviorCfg (fire_on_phase_skip)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_alerts_default_fire_on_phase_skip_true() -> None:
|
|
cfg = Config._from_dict(_with_opts({}))
|
|
assert cfg.alerts.fire_on_phase_skip is True
|
|
|
|
|
|
def test_alerts_fire_on_phase_skip_can_be_disabled() -> None:
|
|
cfg = Config._from_dict(_with_opts({"alerts": {"fire_on_phase_skip": False}}))
|
|
assert cfg.alerts.fire_on_phase_skip is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commit 4: OperatingHoursCfg parsing + tz cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_operating_hours_default_disabled() -> None:
|
|
cfg = Config._from_dict(_with_opts({}))
|
|
assert cfg.operating_hours.enabled is False
|
|
assert cfg.operating_hours.timezone == "America/New_York"
|
|
assert cfg.operating_hours._tz_cache is None
|
|
|
|
|
|
def test_operating_hours_enabled_caches_tz() -> None:
|
|
cfg = Config._from_dict(_with_opts({
|
|
"operating_hours": {
|
|
"enabled": True,
|
|
"timezone": "America/New_York",
|
|
"weekdays": ["MON", "TUE", "WED", "THU", "FRI"],
|
|
"start_hhmm": "09:30",
|
|
"stop_hhmm": "16:00",
|
|
}
|
|
}))
|
|
assert cfg.operating_hours.enabled is True
|
|
assert cfg.operating_hours._tz_cache is not None
|
|
assert str(cfg.operating_hours._tz_cache) == "America/New_York"
|
|
|
|
|
|
def test_operating_hours_invalid_tz_raises_valueerror() -> None:
|
|
import pytest
|
|
with pytest.raises(ValueError, match="operating_hours.timezone"):
|
|
Config._from_dict(_with_opts({
|
|
"operating_hours": {"enabled": True, "timezone": "Not/A_Zone"},
|
|
}))
|
|
|
|
|
|
def test_operating_hours_invalid_weekday_raises_valueerror() -> None:
|
|
import pytest
|
|
with pytest.raises(ValueError, match="weekdays"):
|
|
Config._from_dict(_with_opts({
|
|
"operating_hours": {"enabled": True, "weekdays": ["XYZ"]},
|
|
}))
|