feat(analytics): add heat, growth and interaction metrics

reaction_board/analytics.py

@@ -0,0 +1,229 @@
+# -*- coding: utf-8 -*-
+"""看板指标计算。
+
+这里的函数不访问数据库、网络或时钟，便于离线单元测试。SQL 查询与数据整形
+放在 ``reaction_board.web.queries``，前端只负责展示。
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+from statistics import median as _stat_median
+from statistics import pstdev
+from typing import Iterable, Mapping, Sequence
+
+METRICS: tuple[str, ...] = (
+    "view",
+    "like",
+    "coin",
+    "favorite",
+    "share",
+    "reply",
+    "danmaku",
+)
+
+DEFAULT_HEAT_WEIGHTS: dict[str, float] = {
+    "view": 0.30,
+    "like": 0.20,
+    "coin": 0.20,
+    "favorite": 0.15,
+    "share": 0.10,
+    "reply": 0.03,
+    "danmaku": 0.02,
+}
+
+MIN_VIEW_FOR_INTERACTION = 100
+MIN_VIEW_FOR_GROWTH_RATE = 200
+_IQR_TO_SIGMA = 1.349
+
+
+def percentile(values: Sequence[float], p: float) -> float:
+    """线性插值分位数；``p`` 取 0~1。空序列返回 0。"""
+    if not values:
+        return 0.0
+    ordered = sorted(float(value) for value in values)
+    if len(ordered) == 1:
+        return ordered[0]
+    p = min(max(float(p), 0.0), 1.0)
+    position = (len(ordered) - 1) * p
+    lower = math.floor(position)
+    upper = math.ceil(position)
+    if lower == upper:
+        return ordered[lower]
+    weight = position - lower
+    return ordered[lower] * (1 - weight) + ordered[upper] * weight
+
+
+def robust_center_scale(values: Sequence[float]) -> tuple[float, float]:
+    """中位数与稳健标准差（IQR / 1.349）。
+
+    当 IQR 为 0（指标全相同或样本过少）时退化为总体标准差；仍为 0 时返回
+    ``(center, 0)``，调用方应把该维度的 z 值视为 0。
+    """
+    if not values:
+        return 0.0, 0.0
+    numbers = [float(value) for value in values]
+    center = float(_stat_median(numbers))
+    scale = (percentile(numbers, 0.75) - percentile(numbers, 0.25)) / _IQR_TO_SIGMA
+    if scale <= 1e-12:
+        scale = float(pstdev(numbers)) if len(numbers) > 1 else 0.0
+    if scale <= 1e-12:
+        scale = 0.0
+    return center, scale
+
+
+def _metric_value(row: Mapping[str, object], metric: str) -> float:
+    value = row.get(metric)
+    if value is None or isinstance(value, bool):
+        return 0.0
+    try:
+        return max(0.0, float(value))
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def normalized_weights(weights: Mapping[str, float] | None = None) -> dict[str, float]:
+    source = weights or DEFAULT_HEAT_WEIGHTS
+    result: dict[str, float] = {}
+    for metric in METRICS:
+        value = source.get(metric, DEFAULT_HEAT_WEIGHTS[metric])
+        try:
+            result[metric] = max(0.0, float(value))
+        except (TypeError, ValueError):
+            result[metric] = 0.0
+    total = sum(result.values())
+    if total <= 0:
+        return {metric: 0.0 for metric in METRICS}
+    return {metric: value / total for metric, value in result.items()}
+
+
+def heat_scores(
+    rows: Iterable[Mapping[str, object]],
+    weights: Mapping[str, float] | None = None,
+) -> dict[int, float]:
+    """对每行最新指标计算综合热度。
+
+    每个指标先 ``log1p``，再按中位数/IQR 做稳健标准化，最后加权求和。结果只
+    用于同一批数据内的排序，不跨日期比较。
+    """
+    items = list(rows)
+    if not items:
+        return {}
+    normalized = normalized_weights(weights)
+    log_values: dict[str, list[float]] = {}
+    centers: dict[str, float] = {}
+    scales: dict[str, float] = {}
+    for metric in METRICS:
+        values = [math.log1p(_metric_value(row, metric)) for row in items]
+        log_values[metric] = values
+        centers[metric], scales[metric] = robust_center_scale(values)
+
+    scores: dict[int, float] = {}
+    for index, row in enumerate(items):
+        if "aid" not in row:
+            raise ValueError("heat_scores 需要每行包含 aid")
+        score = 0.0
+        for metric in METRICS:
+            weight = normalized[metric]
+            if weight == 0:
+                continue
+            scale = scales[metric]
+            if scale <= 0:
+                z = 0.0
+            else:
+                z = (log_values[metric][index] - centers[metric]) / scale
+            score += weight * z
+        scores[int(row["aid"])] = score
+    return scores
+
+
+def score_percentiles(scores: Mapping[int, float]) -> dict[int, float]:
+    """把分数映射为 0~100 的百分位，最高者 100，最低者 0。"""
+    if not scores:
+        return {}
+    ordered = sorted(scores.values())
+    total = len(ordered)
+    if total == 1:
+        return {aid: 100.0 for aid in scores}
+    result: dict[int, float] = {}
+    for aid, score in scores.items():
+        below = sum(1 for value in ordered if value < score)
+        result[aid] = round(below / (total - 1) * 100, 2)
+    return result
+
+
+def interaction_rate(
+    metrics: Mapping[str, object],
+    *,
+    min_view: int = MIN_VIEW_FOR_INTERACTION,
+) -> float | None:
+    """互动率 = (点赞 + 投币 + 收藏) / 播放；播放不足门槛返回 None。"""
+    view = _metric_value(metrics, "view")
+    if view < min_view or view <= 0:
+        return None
+    interactions = sum(_metric_value(metrics, metric) for metric in ("like", "coin", "favorite"))
+    return interactions / view
+
+
+@dataclass(frozen=True)
+class Growth:
+    available: bool
+    delta: int | None = None
+    per_day: float | None = None
+    rate: float | None = None
+    gap_days: int | None = None
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "available": self.available,
+            "delta": self.delta,
+            "per_day": self.per_day,
+            "rate": self.rate,
+            "gap_days": self.gap_days,
+        }
+
+
+def compute_growth(
+    current: int | float | None,
+    previous: int | float | None,
+    gap_days: int | float | None,
+) -> Growth:
+    """计算累计差值、日均差值和相对增速。
+
+    缺失基线或间隔不足 1 天时返回不可用，不把 0 当作真实增长。
+    """
+    if current is None or previous is None or gap_days is None:
+        return Growth(available=False)
+    gap = int(gap_days)
+    if gap <= 0:
+        return Growth(available=False)
+    current_value = max(0, int(current))
+    previous_value = max(0, int(previous))
+    delta = current_value - previous_value
+    return Growth(
+        available=True,
+        delta=delta,
+        per_day=delta / gap,
+        rate=delta / max(previous_value, 1),
+        gap_days=gap,
+    )
+
+
+def normalize_cover(url: object) -> str | None:
+    """把 B 站封面统一为 https；空值返回 None。"""
+    if not isinstance(url, str):
+        return None
+    value = url.strip()
+    if not value:
+        return None
+    if value.startswith("//"):
+        return "https:" + value
+    if value.startswith("http://"):
+        return "https://" + value[len("http://"):]
+    return value
+
+
+def is_loopback(host: str) -> bool:
+    value = (host or "").strip().lower()
+    return value in {"127.0.0.1", "localhost", "::1"}

tests/test_analytics.py

@@ -0,0 +1,173 @@
+# -*- coding: utf-8 -*-
+"""看板指标纯函数测试（不联网）: uv run python tests/test_analytics.py"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.stdout.reconfigure(encoding="utf-8")
+
+from reaction_board.analytics import (  # noqa: E402
+    DEFAULT_HEAT_WEIGHTS,
+    METRICS,
+    compute_growth,
+    heat_scores,
+    interaction_rate,
+    is_loopback,
+    normalize_cover,
+    normalized_weights,
+    percentile,
+    robust_center_scale,
+    score_percentiles,
+)
+
+
+def row(aid, **metrics):
+    base = {metric: 0 for metric in METRICS}
+    base.update(metrics)
+    base["aid"] = aid
+    return base
+
+
+def test_percentile_and_robust_scale():
+    assert percentile([], 0.5) == 0.0
+    assert percentile([7], 0.25) == 7.0
+    assert percentile([1, 2, 3, 4], 0.5) == 2.5
+    assert percentile([1, 2, 3, 4], 0) == 1.0
+    assert percentile([1, 2, 3, 4], 1) == 4.0
+
+    center, scale = robust_center_scale([1, 2, 3, 4, 5])
+    assert center == 3.0
+    assert scale > 0
+
+    # IQR 为 0 时退化为总体标准差
+    center, scale = robust_center_scale([10, 10, 10, 20])
+    assert center == 10.0
+    assert scale > 0
+
+    # 全相同则 scale=0，调用方把 z 视为 0
+    assert robust_center_scale([5, 5, 5]) == (5.0, 0.0)
+    assert robust_center_scale([]) == (0.0, 0.0)
+
+
+def test_normalized_weights():
+    weights = normalized_weights()
+    assert abs(sum(weights.values()) - 1.0) < 1e-12
+    assert abs(weights["view"] - DEFAULT_HEAT_WEIGHTS["view"]) < 1e-12
+
+    doubled = normalized_weights({metric: value * 2 for metric, value in DEFAULT_HEAT_WEIGHTS.items()})
+    assert doubled == weights
+
+    zero = normalized_weights({metric: 0 for metric in METRICS})
+    assert all(value == 0 for value in zero.values())
+
+
+def test_heat_scores_order_and_empty():
+    rows = [
+        row(1, view=100, like=10, coin=5, favorite=4, share=1, reply=1, danmaku=2),
+        row(2, view=1000, like=100, coin=50, favorite=40, share=10, reply=10, danmaku=20),
+        row(3, view=10_000, like=1000, coin=500, favorite=400, share=100, reply=100, danmaku=200),
+    ]
+    scores = heat_scores(rows)
+    assert set(scores) == {1, 2, 3}
+    assert scores[1] < scores[2] < scores[3]
+
+    assert heat_scores([]) == {}
+
+    # 所有指标完全相同：所有热度为 0，而不是除零
+    same = heat_scores([row(1, view=5), row(2, view=5)])
+    assert same == {1: 0.0, 2: 0.0}
+
+
+def test_heat_scores_weights_and_missing_values():
+    rows = [
+        row(1, view=100),
+        row(2, view=200),
+    ]
+    view_only = heat_scores(rows, {"view": 1.0})
+    assert view_only[2] > view_only[1]
+    assert all(isinstance(value, float) for value in view_only.values())
+
+    # 缺失/非法指标按 0 处理，不抛异常
+    rows = [
+        {"aid": 1, "view": None, "like": "bad"},
+        {"aid": 2, "view": 10},
+    ]
+    scores = heat_scores(rows)
+    assert scores[2] > scores[1]
+
+
+def test_score_percentiles():
+    assert score_percentiles({}) == {}
+    assert score_percentiles({5: 1.0}) == {5: 100.0}
+    result = score_percentiles({1: 0.0, 2: 1.0, 3: 2.0})
+    assert result[1] == 0.0
+    assert result[2] == 50.0
+    assert result[3] == 100.0
+
+
+def test_interaction_rate_thresholds():
+    assert interaction_rate({"view": 99, "like": 1, "coin": 1, "favorite": 1}) is None
+    assert interaction_rate({"view": 100, "like": 1, "coin": 1, "favorite": 2}) == 0.04
+    assert interaction_rate({"view": 1000, "like": 10}) == 0.01
+    assert interaction_rate({"view": 0, "like": 10}) is None
+    assert interaction_rate({"view": None}) is None
+    assert interaction_rate({"view": 100, "like": None, "coin": "x"}) == 0.0
+
+
+def test_compute_growth():
+    unavailable = compute_growth(100, None, 7)
+    assert unavailable.available is False
+    assert unavailable.delta is None
+
+    no_gap = compute_growth(100, 90, 0)
+    assert no_gap.available is False
+
+    growth = compute_growth(1000, 800, 7)
+    assert growth.available is True
+    assert growth.delta == 200
+    assert growth.per_day == 200 / 7
+    assert abs(growth.rate - 0.25) < 1e-12
+    assert growth.gap_days == 7
+
+    # 基线为 0 时用 1 做分母，避免除零
+    from_zero = compute_growth(50, 0, 3)
+    assert from_zero.delta == 50
+    assert from_zero.rate == 50.0
+
+    # 累计指标倒退（接口修正）时保留负值，由调用方决定是否展示
+    negative = compute_growth(90, 100, 2)
+    assert negative.delta == -10
+    assert negative.per_day == -5.0
+
+
+def test_normalize_cover_and_hosts():
+    assert normalize_cover(None) is None
+    assert normalize_cover("") is None
+    assert normalize_cover("  ") is None
+    assert normalize_cover(123) is None
+    assert normalize_cover("//i0.hdslb.com/a.jpg") == "https://i0.hdslb.com/a.jpg"
+    assert normalize_cover("http://i0.hdslb.com/a.jpg") == "https://i0.hdslb.com/a.jpg"
+    assert normalize_cover("https://i0.hdslb.com/a.jpg") == "https://i0.hdslb.com/a.jpg"
+
+    assert is_loopback("127.0.0.1")
+    assert is_loopback("localhost")
+    assert is_loopback("::1")
+    assert not is_loopback("0.0.0.0")
+    assert not is_loopback("192.168.1.2")
+
+
+def main_test():
+    test_percentile_and_robust_scale()
+    test_normalized_weights()
+    test_heat_scores_order_and_empty()
+    test_heat_scores_weights_and_missing_values()
+    test_score_percentiles()
+    test_interaction_rate_thresholds()
+    test_compute_growth()
+    test_normalize_cover_and_hosts()
+    print("OK: analytics 热度/增长/互动率/封面测试全部通过")
+
+
+if __name__ == "__main__":
+    main_test()