230 lines
6.8 KiB
Python
230 lines
6.8 KiB
Python
# -*- 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"}
|