af1e82e9a5
- index.html manhuang.js 版本对齐内容 hash，避免面板吃到旧 JS - derive_shelf_events(limit=0) 返回空列表；key 查找兼容非 str - --record-outcome 同时 --fail/--dispute 只计一次 fail - build-tab-caches 价差与 shelf_events 共用一次差分
177 lines
6.3 KiB
Python
177 lines
6.3 KiB
Python
"""货架事件差分 + 商家信任：驱动 joy.dashboard.manhuang_shelf 真函数。"""
|
||
from __future__ import annotations
|
||
|
||
from joy.dashboard.manhuang_shelf import (
|
||
annotate_shelf_event_noise,
|
||
derive_shelf_events,
|
||
enrich_shops_trust,
|
||
history_best_from_skus,
|
||
normalize_availability,
|
||
price_moves_from_events,
|
||
score_shop_trust,
|
||
)
|
||
|
||
|
||
def test_normalize_availability_price_and_unknown():
|
||
assert normalize_availability({"p": 3.5}) == "in_stock"
|
||
assert normalize_availability({"price": 1.0, "av": "in_stock"}) == "in_stock"
|
||
assert normalize_availability({"p": None}) == "unknown"
|
||
assert normalize_availability({"p": None, "probe_ok": False}) == "unknown"
|
||
assert normalize_availability({"av": "sold_out"}) == "sold_out"
|
||
assert normalize_availability({"availability": "delist"}) == "delist"
|
||
# 无价不得默认售罄
|
||
assert normalize_availability({}) == "unknown"
|
||
|
||
|
||
def test_derive_price_and_stock_events_and_unknown_not_sold_out():
|
||
prev = {
|
||
"openai": {"p": 10.0, "ch": "FunPay", "av": "in_stock"},
|
||
"claude": {"p": 8.0, "ch": "GGSel", "av": "in_stock"},
|
||
"cursor": {"p": 20.0, "ch": "Plati", "av": "in_stock"},
|
||
"sms": {"p": 0.05, "ch": "5sim", "av": "in_stock"},
|
||
}
|
||
cur = {
|
||
"openai": {"p": 7.5, "ch": "FunPay", "av": "in_stock"}, # price_down
|
||
"claude": {"p": 9.0, "ch": "GGSel", "av": "in_stock"}, # price_up
|
||
"cursor": {"p": None, "ch": "Plati", "av": "sold_out"}, # sold_out
|
||
"sms": {"p": None, "ch": "5sim", "av": "unknown", "probe_ok": False}, # fail
|
||
"giftcard_us": {"p": 1.8, "ch": "GGSel", "av": "in_stock"}, # listed
|
||
}
|
||
events = derive_shelf_events(prev, cur, ts="2026-07-13T00:00:00Z")
|
||
types_by_sku: dict[str, set[str]] = {}
|
||
for e in events:
|
||
types_by_sku.setdefault(e["sku_id"], set()).add(e["type"])
|
||
|
||
assert "price_down" in types_by_sku["openai"]
|
||
assert "price_up" in types_by_sku["claude"]
|
||
assert "sold_out" in types_by_sku["cursor"]
|
||
# 失败探针：不得 sold_out
|
||
assert "sold_out" not in types_by_sku.get("sms", set())
|
||
assert "delist" not in types_by_sku.get("sms", set())
|
||
assert "listed" in types_by_sku.get("giftcard_us", set()) or "restock" in types_by_sku.get(
|
||
"giftcard_us", set()
|
||
)
|
||
|
||
# 价差子集
|
||
moves = price_moves_from_events(events)
|
||
ids = {m["id"] for m in moves}
|
||
assert "openai" in ids
|
||
assert "claude" in ids
|
||
openai_m = next(m for m in moves if m["id"] == "openai")
|
||
assert openai_m["prev"] == 10.0
|
||
assert openai_m["cur"] == 7.5
|
||
assert openai_m["delta"] == -2.5
|
||
|
||
|
||
def test_failed_probe_alone_never_sold_out():
|
||
prev = {"x": {"p": 5.0, "ch": "A", "av": "in_stock"}}
|
||
cur = {"x": {"p": None, "ch": "A", "probe_ok": False}} # no av sold_out
|
||
events = derive_shelf_events(prev, cur)
|
||
assert not any(e["type"] in ("sold_out", "delist") for e in events)
|
||
|
||
|
||
def test_restock_from_sold_out():
|
||
prev = {"y": {"p": None, "ch": "B", "av": "sold_out"}}
|
||
cur = {"y": {"p": 4.0, "ch": "B", "av": "in_stock"}}
|
||
events = derive_shelf_events(prev, cur)
|
||
assert any(e["type"] == "restock" and e["sku_id"] == "y" for e in events)
|
||
|
||
|
||
def test_channel_switch_event():
|
||
prev = {"z": {"p": 3.0, "ch": "FunPay", "av": "in_stock"}}
|
||
cur = {"z": {"p": 3.0, "ch": "GGSel", "av": "in_stock"}}
|
||
events = derive_shelf_events(prev, cur)
|
||
assert any(e["type"] == "channel_switch" for e in events)
|
||
|
||
|
||
def test_score_shop_trust_ours_and_dispute():
|
||
good = score_shop_trust({
|
||
"risk": "mid",
|
||
"trust": {
|
||
"platform": {"rating": 4.5, "sales_n": 200},
|
||
"ours": {"orders_ok": 5, "orders_fail": 0, "quality": 0.9},
|
||
},
|
||
})
|
||
assert good["score"] >= 70
|
||
assert good["band"] in ("high", "mid")
|
||
assert "ours.orders" in good["inputs_used"]
|
||
|
||
bad = score_shop_trust({
|
||
"risk": "high",
|
||
"trust": {"ours": {"orders_ok": 0, "orders_fail": 2, "dispute": True}},
|
||
})
|
||
assert bad["score"] == 0
|
||
assert bad["band"] == "block"
|
||
|
||
unproven = score_shop_trust({"risk": "low", "trust": {}})
|
||
assert unproven["band"] == "unproven"
|
||
assert 0 <= unproven["score"] <= 100
|
||
|
||
|
||
def test_enrich_shops_trust_and_history_best():
|
||
shops = enrich_shops_trust([
|
||
{
|
||
"id": "shop-a",
|
||
"risk": "mid",
|
||
"trust": {"ours": {"orders_ok": 3, "orders_fail": 0}},
|
||
}
|
||
])
|
||
assert shops[0]["trust"]["score"] == shops[0]["trust_score"]
|
||
assert shops[0]["trust_band"]
|
||
|
||
best = history_best_from_skus([
|
||
{
|
||
"id": "openai",
|
||
"tier": "p0",
|
||
"best": {"price": 8.0, "channel": "FunPay"},
|
||
"sources": [{"ok": True, "price": 8.0}],
|
||
},
|
||
{
|
||
"id": "empty",
|
||
"best": None,
|
||
"sources": [{"ok": False, "error": "timeout"}],
|
||
},
|
||
])
|
||
assert best["openai"]["av"] == "in_stock"
|
||
assert best["openai"]["p"] == 8.0
|
||
assert best["empty"]["av"] == "unknown"
|
||
|
||
|
||
def test_limit_zero_returns_empty():
|
||
prev = {"a": {"p": 1.0, "ch": "X", "av": "in_stock"}}
|
||
cur = {"a": {"p": 2.0, "ch": "X", "av": "in_stock"}}
|
||
assert derive_shelf_events(prev, cur, limit=0) == []
|
||
assert len(derive_shelf_events(prev, cur, limit=1)) == 1
|
||
|
||
|
||
def test_giftcard_face_noise_flag():
|
||
prev = {"giftcard_us": {"p": 100.0, "ch": "GGSel", "av": "in_stock"}}
|
||
cur = {"giftcard_us": {"p": 1.81, "ch": "GGSel", "av": "in_stock"}}
|
||
events = derive_shelf_events(prev, cur)
|
||
downs = [e for e in events if e.get("type") == "price_down"]
|
||
assert downs and downs[0].get("noise") is True
|
||
assert downs[0].get("noise_reason") in ("face_vs_listing", "extreme_ratio")
|
||
moves = price_moves_from_events(events)
|
||
assert moves and moves[0].get("noise") is True
|
||
clean = annotate_shelf_event_noise({
|
||
"type": "price_down", "sku_id": "openai", "prev": 10.0, "cur": 9.5, "delta": -0.5,
|
||
})
|
||
assert not clean.get("noise")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 无 pytest 时：python3 tests/test_manhuang_shelf.py
|
||
import traceback
|
||
|
||
_failed = 0
|
||
for _name, _fn in sorted(globals().items()):
|
||
if not _name.startswith("test_") or not callable(_fn):
|
||
continue
|
||
try:
|
||
_fn()
|
||
print(f"PASS {_name}")
|
||
except Exception:
|
||
_failed += 1
|
||
print(f"FAIL {_name}")
|
||
traceback.print_exc()
|
||
raise SystemExit(1 if _failed else 0)
|
