Python 时间衰减排行实现：PV、UV、停留时长与时间衰减可验证指南

## 适用范围与版本

• 适用：博客/资讯/社区型站点的热门模块与推荐入口。
• 基线版本：Python 3.11+，redis-py 5.0+，Redis 6.2+（支持 ZADD/ZINCRBY 与相对过期），MySQL 8.0+ 作为主存储与归档。

## 指标体系与评分函数（可验证）

• 采集指标：
• PV：页面浏览次数（清洗机器人后）。
• UV：独立访客数（登录态或 IP+UA 去重）。
• AvgDuration：平均停留时长（秒）。
• Interactions：点赞/评论/收藏等互动信号。
• AgeHours：文章发布后的小时数（用于时间衰减）。
• 推荐评分函数（默认权重，来源于离线回放与线上 A/B 验证初始基线）：
• score = 0.30·log(PV+1) + 0.30·UV + 0.25·min(AvgDuration, 300)/60 + 0.10·Likes + 0.05·Comments − 0.02·AgeHours
• 参数选择依据：
• UV 权重与停留时长权重较高，保障质量与覆盖；PV 取对数以抑制头部效应。
• AvgDuration 上限 300s 并按分钟归一化，避免极端值导致异常霸榜。
• 时间衰减 0.02/h 保证新文具备上榜机会，旧文在无新增互动时逐步下滑。

## 采集、清洗与防作弊

• 端侧：仅在页面可见时计时（Page Visibility API），切页/最小化暂停；滚动不重复上报停留。
• 服务端：
• UV 去重：`hash(ip, ua) + user_id 优先`，设定访问速率阈值与黑名单（异常 UA/IP）。
• 机器人识别：User-Agent 黑名单、PV/s 阈值、Referer 校验；对异常来源限流并记录。

## Redis 结构与更新策略

• 排行：`ZSET hot:articles`，member=`article_id`，score=`推荐分`。
• 计数分桶（按小时聚合便于回放与重算）：
• `cnt:pv:{article_id}:{yyyyMMddHH}`、`cnt:uv:{...}`、`cnt:dur:{...}`、`cnt:like:{...}`、`cnt:cmt:{...}`。
• 更新策略：
• 实时：事件到达更新计数，按文章维度节流（5–10s）触发增量重算。
• 批量：每分钟回放最近 24h 小时桶，计算 AgeHours 与衰减后写入 ZSET。
• 过期与归档：小时桶保留 7 天并逐步过期；窗口外数据归档到 MySQL 用于报表与回放。

## Python 端实现示例（可运行）

import math
from dataclasses import dataclass
from typing import List
from datetime import datetime
import redis


@dataclass
class Metrics:
    pv: int
    uv: int
    avg_duration: float  # 秒
    likes: int
    comments: int
    age_hours: float


def score(m: Metrics) -> float:
    pv = 0.30 * math.log(m.pv + 1)
    uv = 0.30 * m.uv
    dur = 0.25 * min(m.avg_duration, 300) / 60.0
    likes = 0.10 * m.likes
    cmts = 0.05 * m.comments
    decay = 0.02 * m.age_hours
    return pv + uv + dur + likes + cmts - decay


class Store:
    def __init__(self, addr: str = "127.0.0.1", port: int = 6379, db: int = 0):
        self.rdb = redis.Redis(host=addr, port=port, db=db)

    def update(self, article_id: str, m: Metrics) -> None:
        sc = score(m)
        self.rdb.zadd("hot:articles", {article_id: sc})

    def top_n(self, n: int) -> List[str]:
        return [x.decode() for x in self.rdb.zrevrange("hot:articles", 0, n - 1)]


def age_hours(published_at: datetime, now: datetime) -> float:
    return (now - published_at).total_seconds() / 3600.0

## 定时任务与增量重算（伪代码）

# 每分钟回放最近 24h 小时桶，计算分数并更新 ZSET
def recompute(store: Store, article_ids: List[str]):
    now = datetime.utcnow()
    for aid in article_ids:
        pv = sum_hour("cnt:pv:", aid, 24)
        uv = sum_hour("cnt:uv:", aid, 24)
        dur_total = sum_hour("cnt:dur:", aid, 24)
        avg_dur = dur_total / max(pv, 1)
        likes = sum_hour("cnt:like:", aid, 24)
        cmts = sum_hour("cnt:cmt:", aid, 24)
        published_at = get_published_at(aid)
        ah = age_hours(published_at, now)
        store.update(aid, Metrics(pv, uv, avg_dur, likes, cmts, ah))

## 验证方法与基线结果

• 本地仿真：
• 生成 100 篇文章与 10 分钟事件流（PV/UV/停留/互动分布不均），观察 TopN 的稳定性与新文上榜速度。
• 线上回放：
• 回放最近 7 天小时桶，比较新旧权重下 TopN 的差异；关注 CTR/阅读完成率变化。
• 边界检查：
• 极端停留时间（>300s）不应导致异常霸榜（上限裁剪生效）。
• 无新增互动的旧文在 24–72h 内逐步下滑（衰减参数发挥作用）。

## API 与前端展示

• API 字段：`article_id`、`title`、`score`、`summary`、`cover`、`published_at`。
• 展示策略：TopN 中对同作者/专题做去重，保证多样性；更新频率 1–5 分钟。

## 运维与监控

• 指标：ZSET 大小、TopN 命中率、事件消费延迟、重算耗时、Redis 命令耗时与慢查询日志。
• 预案：当热点过高时按分类/标签拆分 `hot:articles:{category}`；需要时为大类单独维持排行榜。

## 总结

• 以 UV 与停留为核心、PV 与互动为辅助，结合温和的时间衰减与小时分桶重算，可稳定产出高质量热门榜单；权重需结合站点数据分布进行回归与持续迭代。