feat: heartbeat suprimate afara orelor, format compact, status simplificat
- _in_trading_window(): helper nou — suprima heartbeat in afara ferestrei de tranzactionare - _heartbeat_loop: format compact romanian (STATE | semnale: N | Xh), fara statistici backend - ATM pornit: "canary:" -> "senzor: ok/deviat (dist/thresh)" - ATM oprit: simplificat la "durata: Xh | semnale: N" - /status: 2-3 linii compacte, etichete pauza in romana, fara "Canary" - _maybe_log_transition: parametru status_body optional; titluri "Piata deschisa/inchisa" - _brief_status(): helper nou pentru mesajele de tranzitie piata - 10 teste noi (trading_window, status compact, fereastra); 194 passed Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -666,7 +666,7 @@ def test_market_transition_sends_notification():
|
||||
_main._maybe_log_transition(None, lifecycle, mid, _A(), _N())
|
||||
assert len(alerts) == 1
|
||||
assert alerts[0].kind == "status"
|
||||
assert "market" in alerts[0].title.lower() or "piața" in alerts[0].body.lower()
|
||||
assert "piața" in alerts[0].title.lower() or "monitorizare" in alerts[0].body.lower()
|
||||
|
||||
|
||||
def test_startup_in_window_suppresses_market_open():
|
||||
@@ -920,7 +920,7 @@ async def test_status_command_reports_pause_reason():
|
||||
status = [a for a in ctx.notifier.alerts if a.kind == "status"]
|
||||
assert status
|
||||
body = status[0].body
|
||||
assert "user_paused" in body or "pauzat:user_paused" in body
|
||||
assert "pauză manuală" in body or "pauza" in body.lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -979,3 +979,123 @@ async def test_lifecycle_with_drift_then_resume_then_fire(monkeypatch, tmp_path)
|
||||
# FSM reached fire via normal path
|
||||
assert tr is not None and tr.trigger == "SELL"
|
||||
assert fsm.state == State.IDLE
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _in_trading_window tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
import datetime as _dt
|
||||
|
||||
|
||||
def test_in_trading_window_disabled():
|
||||
"""Returns True when operating_hours disabled."""
|
||||
import atm.main as _main
|
||||
cfg = types.SimpleNamespace(operating_hours=types.SimpleNamespace(enabled=False))
|
||||
assert _main._in_trading_window(0.0, cfg) is True
|
||||
|
||||
|
||||
def test_in_trading_window_no_oh():
|
||||
"""Returns True when cfg has no operating_hours attr."""
|
||||
import atm.main as _main
|
||||
cfg = types.SimpleNamespace()
|
||||
assert _main._in_trading_window(0.0, cfg) is True
|
||||
|
||||
|
||||
def test_in_trading_window_in_window():
|
||||
"""Returns True during configured hours (Monday 12:00 NY)."""
|
||||
import atm.main as _main
|
||||
cfg = _oh_cfg()
|
||||
tz = cfg.operating_hours._tz_cache
|
||||
ts = _dt.datetime(2026, 4, 20, 12, 0, tzinfo=tz).timestamp()
|
||||
assert _main._in_trading_window(ts, cfg) is True
|
||||
|
||||
|
||||
def test_in_trading_window_out_of_hours():
|
||||
"""Returns False before market open (Monday 08:00 NY)."""
|
||||
import atm.main as _main
|
||||
cfg = _oh_cfg()
|
||||
tz = cfg.operating_hours._tz_cache
|
||||
ts = _dt.datetime(2026, 4, 20, 8, 0, tzinfo=tz).timestamp()
|
||||
assert _main._in_trading_window(ts, cfg) is False
|
||||
|
||||
|
||||
def test_in_trading_window_weekend():
|
||||
"""Returns False on weekend (Sunday 12:00 NY)."""
|
||||
import atm.main as _main
|
||||
cfg = _oh_cfg()
|
||||
tz = cfg.operating_hours._tz_cache
|
||||
ts = _dt.datetime(2026, 4, 19, 12, 0, tzinfo=tz).timestamp()
|
||||
assert _main._in_trading_window(ts, cfg) is False
|
||||
|
||||
|
||||
def test_heartbeat_suppressed_outside_hours(monkeypatch):
|
||||
"""_in_trading_window=False prevents heartbeat from sending."""
|
||||
import atm.main as _main
|
||||
monkeypatch.setattr(_main, "_in_trading_window", lambda ts, cfg: False)
|
||||
sent = []
|
||||
# Simulate the check: outside window → no send
|
||||
if not _main._in_trading_window(0.0, None):
|
||||
pass # heartbeat_due reset, continue — no send
|
||||
assert sent == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_status_compact_active():
|
||||
"""/status produces compact 2-line format; 'Canary' absent."""
|
||||
import atm.main as _main
|
||||
from atm.commands import Command
|
||||
|
||||
ctx = _dispatch_ctx()
|
||||
ctx.detector.rolling = []
|
||||
ctx.fsm.state = types.SimpleNamespace(value="ARMED_BUY")
|
||||
ctx.state.fire_count = 3
|
||||
|
||||
await _main._dispatch_command(ctx, Command(action="status"))
|
||||
|
||||
status = [a for a in ctx.notifier.alerts if a.kind == "status"]
|
||||
assert status
|
||||
body = status[0].body
|
||||
lines = body.splitlines()
|
||||
assert len(lines) == 2 # no fereastră line (oh.enabled=False)
|
||||
assert "ARMED_BUY" in lines[0]
|
||||
assert "semnale: 3" in lines[0]
|
||||
assert "Canary" not in body
|
||||
assert "canary" not in body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_status_compact_paused_manual():
|
||||
"""/status shows 'pauză manuală' on line 1 when user_paused."""
|
||||
import atm.main as _main
|
||||
from atm.commands import Command
|
||||
|
||||
ctx = _dispatch_ctx()
|
||||
ctx.lifecycle.user_paused = True
|
||||
ctx.detector.rolling = []
|
||||
ctx.fsm.state = types.SimpleNamespace(value="IDLE")
|
||||
|
||||
await _main._dispatch_command(ctx, Command(action="status"))
|
||||
|
||||
status = [a for a in ctx.notifier.alerts if a.kind == "status"]
|
||||
body = status[0].body
|
||||
assert body.startswith("pauză manuală")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_status_window_line_when_oh_enabled():
|
||||
"""/status adds fereastră line when operating_hours enabled."""
|
||||
import atm.main as _main
|
||||
from atm.commands import Command
|
||||
|
||||
cfg = _oh_cfg()
|
||||
lifecycle = _main.LifecycleState(last_window_state="open")
|
||||
ctx = _dispatch_ctx(lifecycle=lifecycle, cfg=cfg)
|
||||
ctx.detector.rolling = []
|
||||
ctx.fsm.state = types.SimpleNamespace(value="IDLE")
|
||||
|
||||
await _main._dispatch_command(ctx, Command(action="status"))
|
||||
|
||||
status = [a for a in ctx.notifier.alerts if a.kind == "status"]
|
||||
body = status[0].body
|
||||
assert "fereastră: deschisă" in body
|
||||
|
||||
Reference in New Issue
Block a user