gahusb
94a034ef38
신호 expires_at 계산용 TTL (default 300s). 환경별로 조정 가능. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
67 lines
2.1 KiB
Python
67 lines
2.1 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
|
|
|
|
|
|
# ----- SIGNAL_TTL_SECONDS env -----
|
|
|
|
def test_signal_ttl_seconds_default(monkeypatch):
|
|
monkeypatch.delenv("SIGNAL_TTL_SECONDS", raising=False)
|
|
from ai_trade.config import Settings
|
|
s = Settings()
|
|
assert s.signal_ttl_seconds == 300
|
|
|
|
|
|
def test_signal_ttl_seconds_env_override(monkeypatch):
|
|
monkeypatch.setenv("SIGNAL_TTL_SECONDS", "60")
|
|
from ai_trade.config import Settings
|
|
s = Settings()
|
|
assert s.signal_ttl_seconds == 60
|