73 lines
2.8 KiB
Python
73 lines
2.8 KiB
Python
# ============================================================
|
||
# 偏好时间衰减加权黄金测试（plan.md §8.3 / 矩阵 66 行剩余项）
|
||
# 覆盖：新样本主导偏好（衰减）、久远样本权重趋近 0、
|
||
# 舍入后当天信号近似整数、既有计数语义兼容。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
from pathlib import Path
|
||
|
||
from server.knowledge.preferences import PreferenceStore
|
||
|
||
|
||
def _store(tmp_path: Path) -> PreferenceStore:
|
||
return PreferenceStore(str(tmp_path / "pref.json"))
|
||
|
||
|
||
def _sample(strategy: str, *, days_ago: float, project_id: str | None = "p1") -> dict:
|
||
at = (datetime.datetime.now() - datetime.timedelta(days=days_ago)).strftime("%Y-%m-%d %H:%M")
|
||
return {"strategy": strategy, "source": "schedule.run", "actor": "u1",
|
||
"projectId": project_id, "at": at}
|
||
|
||
|
||
def test_recent_signal_dominates_stale(tmp_path: Path):
|
||
"""时间衰减：30 天前的旧偏好不应压过今天的信号（偏好漂移）。"""
|
||
ps = _store(tmp_path)
|
||
ps.samples = [
|
||
_sample("DELIVERY_FIRST", days_ago=30),
|
||
_sample("CAPACITY_BALANCE", days_ago=0),
|
||
]
|
||
scores = ps._decayed_scores("p1")
|
||
assert scores["CAPACITY_BALANCE"] > scores["DELIVERY_FIRST"]
|
||
assert scores["DELIVERY_FIRST"] < 0.5 # 30 天（>1 半衰期）显著衰减
|
||
strategy, _ = ps.preferred_strategy(project_id="p1")
|
||
assert strategy == "CAPACITY_BALANCE"
|
||
|
||
|
||
def test_same_source_two_recent_samples_half_life(tmp_path: Path):
|
||
"""半衰期：14 天前样本权重约为今天的 0.5。"""
|
||
ps = _store(tmp_path)
|
||
ps.samples = [
|
||
_sample("A", days_ago=14),
|
||
_sample("B", days_ago=0),
|
||
]
|
||
scores = ps._decayed_scores("p1")
|
||
ratio = scores["A"] / scores["B"]
|
||
assert 0.45 <= ratio <= 0.55, f"半衰期比值应≈0.5，got {ratio}"
|
||
|
||
|
||
def test_today_scores_round_to_ints(tmp_path: Path):
|
||
"""当天样本（同源同策略×2）评分≈2（保留既有计数语义兼容）。"""
|
||
ps = _store(tmp_path)
|
||
ps.samples = [
|
||
_sample("DELIVERY_FIRST", days_ago=0),
|
||
_sample("DELIVERY_FIRST", days_ago=0),
|
||
]
|
||
strategy, scores = ps.preferred_strategy(project_id="p1")
|
||
assert strategy == "DELIVERY_FIRST"
|
||
assert abs(scores["DELIVERY_FIRST"] - 2.0) < 0.01
|
||
|
||
|
||
def test_decay_preserves_apply_weight(tmp_path: Path):
|
||
"""来源权重叠加衰减：采用(2×decay) > 试排(1×decay) 同时间。"""
|
||
ps = _store(tmp_path)
|
||
ps.samples = [
|
||
_sample("CAPACITY_BALANCE", days_ago=0),
|
||
_sample("DELIVERY_FIRST", days_ago=0),
|
||
]
|
||
# 同时间同来源权重应相等；改其中一个为采用验证权重
|
||
ps.samples[0]["source"] = "scenario.apply"
|
||
scores = ps._decayed_scores("p1")
|
||
assert abs(scores["CAPACITY_BALANCE"] - 2 * scores["DELIVERY_FIRST"]) < 0.05
|