gahusb
2a11d05f4a
코드 리뷰 F5 — Phase 5 consumer(agent-office /signal) prereq: PollState.signal_cycle_id (process auto-increment) + get_active_signals(now) + purge_expired_signals(now) helper. expires_at 없는 legacy signal은 expired 취급. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
"""F5 — state.signals lifecycle (expires_at + cycle_id)."""
|
|
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from ai_trade.state import PollState
|
|
|
|
KST = ZoneInfo("Asia/Seoul")
|
|
|
|
|
|
def test_initial_signal_cycle_id_is_zero():
|
|
state = PollState()
|
|
assert state.signal_cycle_id == 0
|
|
|
|
|
|
def test_get_active_signals_excludes_expired():
|
|
state = PollState()
|
|
now = datetime(2026, 5, 25, 10, 0, tzinfo=KST)
|
|
future = (now + timedelta(seconds=300)).isoformat()
|
|
past = (now - timedelta(seconds=60)).isoformat()
|
|
state.signals = {
|
|
"A": {"ticker": "A", "expires_at": future, "cycle_id": 1, "action": "buy"},
|
|
"B": {"ticker": "B", "expires_at": past, "cycle_id": 1, "action": "buy"},
|
|
}
|
|
active = state.get_active_signals(now)
|
|
tickers = [s["ticker"] for s in active]
|
|
assert "A" in tickers
|
|
assert "B" not in tickers
|
|
|
|
|
|
def test_get_active_signals_treats_missing_expires_as_expired():
|
|
"""expires_at 없는 legacy 신호는 expired로 간주."""
|
|
state = PollState()
|
|
now = datetime(2026, 5, 25, 10, 0, tzinfo=KST)
|
|
state.signals = {"C": {"ticker": "C", "action": "buy"}}
|
|
assert state.get_active_signals(now) == []
|
|
|
|
|
|
def test_purge_expired_signals_removes_expired():
|
|
state = PollState()
|
|
now = datetime(2026, 5, 25, 10, 0, tzinfo=KST)
|
|
future = (now + timedelta(seconds=300)).isoformat()
|
|
past = (now - timedelta(seconds=60)).isoformat()
|
|
state.signals = {
|
|
"A": {"ticker": "A", "expires_at": future, "cycle_id": 1},
|
|
"B": {"ticker": "B", "expires_at": past, "cycle_id": 1},
|
|
}
|
|
removed = state.purge_expired_signals(now)
|
|
assert "A" in state.signals
|
|
assert "B" not in state.signals
|
|
assert removed == 1
|