131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
# agent-office/tests/test_lotto_signals.py
|
||
import pytest
|
||
|
||
from app.curator import signals
|
||
|
||
|
||
def test_sim_consensus_top10_geomean():
|
||
"""top-10 consensus 평균이 기하평균 기반인지."""
|
||
best_picks = [
|
||
{"scores": [10, 10, 10, 10, 10]}, # high & uniform
|
||
{"scores": [9, 9, 9, 9, 9]},
|
||
{"scores": [8, 8, 8, 8, 8]},
|
||
{"scores": [7, 7, 7, 7, 7]},
|
||
{"scores": [6, 6, 6, 6, 6]},
|
||
{"scores": [5, 5, 5, 5, 5]},
|
||
{"scores": [4, 4, 4, 4, 4]},
|
||
{"scores": [3, 3, 3, 3, 3]},
|
||
{"scores": [2, 2, 2, 2, 2]},
|
||
{"scores": [1, 1, 1, 1, 1]}, # top 10
|
||
{"scores": [0, 0, 0, 0, 0]}, # bottom 10
|
||
] * 1 + [{"scores": [0, 0, 0, 0, 0]}] * 10
|
||
result = signals.sim_consensus_score(best_picks)
|
||
assert 0.0 <= result <= 1.0
|
||
assert result > 0.4
|
||
|
||
|
||
def test_sim_consensus_geomean_penalizes_imbalance():
|
||
"""5종 중 한 종만 폭주하는 outlier 후보는 균형 후보보다 작아야 한다."""
|
||
balanced = [{"scores": [5, 5, 5, 5, 5]}] * 20
|
||
imbalanced = [{"scores": [25, 0, 0, 0, 0]}] * 20
|
||
s_balanced = signals.sim_consensus_score(balanced)
|
||
s_imbalanced = signals.sim_consensus_score(imbalanced)
|
||
assert s_imbalanced < s_balanced
|
||
|
||
|
||
def test_strategy_drift_score():
|
||
"""drift = 전략별 가중치 변화 절댓값 합."""
|
||
w_prev = {"gap_focus": 0.30, "hot_focus": 0.25, "pair_bias": 0.45}
|
||
w_curr = {"gap_focus": 0.40, "hot_focus": 0.20, "pair_bias": 0.40}
|
||
result = signals.strategy_drift_score(w_prev, w_curr)
|
||
assert abs(result - 0.20) < 1e-9
|
||
|
||
|
||
def test_strategy_drift_new_strategy_appears():
|
||
"""이전에 없던 전략이 등장하면 그 가중치 전체가 drift에 가산."""
|
||
w_prev = {"gap_focus": 0.5, "hot_focus": 0.5}
|
||
w_curr = {"gap_focus": 0.4, "hot_focus": 0.4, "newbie": 0.2}
|
||
result = signals.strategy_drift_score(w_prev, w_curr)
|
||
assert abs(result - 0.4) < 1e-9
|
||
|
||
|
||
def test_confidence_score_passthrough():
|
||
"""confidence는 큐레이션 결과의 값 그대로 (0~1 clamp 확인)."""
|
||
assert signals.confidence_score({"confidence": 0.85}) == 0.85
|
||
assert signals.confidence_score({"confidence": 1.2}) == 1.0
|
||
assert signals.confidence_score({"confidence": -0.1}) == 0.0
|
||
assert signals.confidence_score({}) is None
|
||
|
||
|
||
def test_adaptive_baseline_cold_start():
|
||
"""window 크기 < 4 → warmup, z=None."""
|
||
bl = signals.AdaptiveBaseline(window=[1.0, 1.1, 0.9], window_max=8)
|
||
z, fire = bl.evaluate(value=1.5, z_normal=1.5, z_urgent=2.5)
|
||
assert fire == "warmup"
|
||
assert z is None
|
||
|
||
|
||
def test_adaptive_baseline_preparing():
|
||
"""window 4~7 → 보수적 임계치 z=2.0."""
|
||
bl = signals.AdaptiveBaseline(window=[1.0, 1.0, 1.0, 1.0], window_max=8)
|
||
z, fire = bl.evaluate(value=3.0, z_normal=1.5, z_urgent=2.5)
|
||
assert fire in ("normal", "urgent")
|
||
|
||
|
||
def test_adaptive_baseline_normal_window_full():
|
||
"""window 8 풀, value가 평균보다 1.5σ 이상이면 normal."""
|
||
bl = signals.AdaptiveBaseline(
|
||
window=[1.0, 1.1, 0.9, 1.0, 1.0, 1.1, 0.9, 1.0],
|
||
window_max=8,
|
||
)
|
||
z, fire = bl.evaluate(value=1.12, z_normal=1.5, z_urgent=2.5)
|
||
assert fire == "normal"
|
||
assert z is not None and z >= 1.5
|
||
|
||
|
||
def test_adaptive_baseline_urgent():
|
||
"""z >= 2.5 → urgent."""
|
||
bl = signals.AdaptiveBaseline(
|
||
window=[1.0, 1.1, 0.9, 1.0, 1.0, 1.1, 0.9, 1.0],
|
||
window_max=8,
|
||
)
|
||
z, fire = bl.evaluate(value=2.0, z_normal=1.5, z_urgent=2.5)
|
||
assert fire == "urgent"
|
||
|
||
|
||
def test_adaptive_baseline_push_updates_window():
|
||
"""push 시 FIFO 동작."""
|
||
bl = signals.AdaptiveBaseline(window=[1, 2, 3, 4, 5, 6, 7, 8], window_max=8)
|
||
bl.push(9.0)
|
||
assert bl.window == [2, 3, 4, 5, 6, 7, 8, 9.0]
|
||
|
||
|
||
def test_decide_fire_level_two_normals_escalate():
|
||
sigs = [
|
||
{"metric": "sim", "z": 1.6, "fire": "normal"},
|
||
{"metric": "drift", "z": 1.7, "fire": "normal"},
|
||
{"metric": "conf", "z": 0.5, "fire": "noop"},
|
||
]
|
||
assert signals.decide_overall_fire(sigs) == "urgent"
|
||
|
||
|
||
def test_decide_fire_level_single_normal():
|
||
sigs = [
|
||
{"metric": "sim", "z": 1.6, "fire": "normal"},
|
||
{"metric": "drift", "z": 0.3, "fire": "noop"},
|
||
]
|
||
assert signals.decide_overall_fire(sigs) == "normal"
|
||
|
||
|
||
def test_decide_fire_level_single_urgent():
|
||
sigs = [
|
||
{"metric": "sim", "z": 3.0, "fire": "urgent"},
|
||
{"metric": "drift", "z": 0.2, "fire": "noop"},
|
||
]
|
||
assert signals.decide_overall_fire(sigs) == "urgent"
|
||
|
||
|
||
def test_decide_fire_level_all_noop():
|
||
sigs = [{"metric": "sim", "z": 0.5, "fire": "noop"}]
|
||
assert signals.decide_overall_fire(sigs) == "noop"
|