355 lines
11 KiB
Python
355 lines
11 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
|
||
MIN_EFFICIENCY_DURATION = 600
|
||
DEFAULT_MATURITY_DAYS = 2
|
||
DEFAULT_SHRINK_MINUTES = 120.0
|
||
_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 content_hours(duration_seconds: object) -> float | None:
|
||
"""把视频总时长（秒）换算为内容小时；缺失或非正数返回 ``None``。
|
||
|
||
多 P 视频的 ``duration`` 已是各 P 之和，因此这里不需要再乘 P 数。
|
||
"""
|
||
if duration_seconds is None or isinstance(duration_seconds, bool):
|
||
return None
|
||
try:
|
||
seconds = float(duration_seconds)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if seconds <= 0:
|
||
return None
|
||
return seconds / 3600.0
|
||
|
||
|
||
def views_per_hour(view: object, duration_seconds: object) -> float | None:
|
||
"""播放/内容小时；时长缺失或非正数时返回 ``None``，不做除零。"""
|
||
hours = content_hours(duration_seconds)
|
||
if hours is None:
|
||
return None
|
||
return _metric_value({"view": view}, "view") / hours
|
||
|
||
|
||
def interactions_per_hour(
|
||
metrics: Mapping[str, object] | None,
|
||
duration_seconds: object,
|
||
*,
|
||
interaction_metrics: Sequence[str] = ("like", "coin", "favorite"),
|
||
) -> float | None:
|
||
"""互动/内容小时；统计缺失或时长无效时返回 ``None``。"""
|
||
hours = content_hours(duration_seconds)
|
||
if hours is None or metrics is None:
|
||
return None
|
||
interactions = sum(_metric_value(metrics, metric) for metric in interaction_metrics)
|
||
return interactions / hours
|
||
|
||
|
||
def age_days(pubdate: object, reference: int | float) -> int | None:
|
||
"""发布到参考时点（通常为最新数据日）的整天数；无效发布时间返回 ``None``。"""
|
||
if pubdate is None or isinstance(pubdate, bool):
|
||
return None
|
||
try:
|
||
published = int(pubdate)
|
||
now = int(reference)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if published <= 0:
|
||
return None
|
||
return max(0, (now - published) // 86400)
|
||
|
||
|
||
def is_mature(
|
||
pubdate: object,
|
||
reference: int | float,
|
||
maturity_days: int = DEFAULT_MATURITY_DAYS,
|
||
) -> bool:
|
||
"""是否已达到成熟窗口；发布时间缺失视为未成熟。"""
|
||
days = age_days(pubdate, reference)
|
||
return days is not None and days >= max(0, int(maturity_days))
|
||
|
||
|
||
def median_of(values: Iterable[object]) -> float | None:
|
||
"""忽略无效值的中位数；没有有效值返回 ``None``。"""
|
||
numbers: list[float] = []
|
||
for value in values:
|
||
if value is None or isinstance(value, bool):
|
||
continue
|
||
try:
|
||
numbers.append(float(value))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if not numbers:
|
||
return None
|
||
return float(_stat_median(numbers))
|
||
|
||
|
||
def effective_shrink_minutes(
|
||
per_up_minutes: Iterable[object],
|
||
fallback: float = DEFAULT_SHRINK_MINUTES,
|
||
) -> float:
|
||
"""自适应收缩强度：各 UP 总内容分钟的中位数，缺失时回退固定值。"""
|
||
positives: list[float] = []
|
||
for minutes in per_up_minutes:
|
||
if minutes is None or isinstance(minutes, bool):
|
||
continue
|
||
try:
|
||
value = float(minutes)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if value > 0:
|
||
positives.append(value)
|
||
value = median_of(positives)
|
||
if value is None or value <= 0:
|
||
return float(fallback)
|
||
return value
|
||
|
||
|
||
def eb_rate(
|
||
total: object,
|
||
exposure: object,
|
||
prior_rate: float,
|
||
prior_strength: float,
|
||
) -> float | None:
|
||
"""Gamma–Poisson 经验贝叶斯收缩后的速率。
|
||
|
||
``exposure`` 与 ``prior_strength`` 必须使用同一单位（本项目中为分钟），
|
||
返回单位与 ``prior_rate`` 一致（每分钟）。
|
||
"""
|
||
if prior_strength < 0:
|
||
raise ValueError("prior_strength 不能为负")
|
||
try:
|
||
total_value = max(0.0, float(total))
|
||
exposure_value = max(0.0, float(exposure))
|
||
prior = max(0.0, float(prior_rate))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if prior_strength <= 0:
|
||
return None if exposure_value <= 0 else total_value / exposure_value
|
||
return (total_value + prior_strength * prior) / (exposure_value + prior_strength)
|
||
|
||
|
||
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"}
|