feat(analytics): 埋点/上报健康度自报计数(selfstat)+ admin 健康度聚合端点 (#127)

背景 / 目标
客户端埋点从「采集」到「上报落库」全链路存在丢失(采集端丢帧、网络丢包、上报失败),现有 analytics_event(行为事件)无法度量这条链路的健康度。本 MR 引入一条与行为埋点完全独立的自报计数(selfstat)链路:客户端周期上报「自 epoch 起算的累计计数」,服务端只 append 存原始快照,admin 查询时在 Python 侧差分聚合出两段成功率(埋点成功率 / 上报成功率),支持总览 / 按天趋势 / 按维度下钻。

改动概览
① App 侧 — 上报接入(不强制登录)

新增 POST /api/v1/analytics/selfstat:一份快照 = 设备头 + N 条 event 累计计数,单事务落库,返回 snapshot_id。
稳态兜底:落库异常不裸奔 500,logger.exception 后回 503(计数链路要稳,不阻塞端上主流程)。
events 允许空列表(某次只报设备级计数也合法),上限 200 条/次;与行为埋点 min_length=1 有意不同。
② Admin 侧 — 健康度聚合(只读,需 admin 鉴权)

新增 GET /admin/api/analytics-health/{overview,trend,breakdown}。
差分聚合:原始累计快照按 (device_id, epoch_id, event) 分区、created_at 升序做相邻差分;分区首行增量=累计值,负值(epoch 重置 / 乱序)夹 0;每条增量按其快照 created_at 归入北京天桶。
基线行:取 date_from 左侧每分区最后一条快照,保证区间第一条增量正确(参与差分后丢弃)。
两段率(分母为 0 → null):
埋点成功率 track_success_rate = (attempted − drop_capture) / attempted
上报成功率 report_success_rate = delivered / (delivered + drop_undelivered)
breakdown 维度限 event | app_ver | oem(路由正则校验);结果按上报成功率升序(最差在前,None 垫底)。
③ 数据模型 / 基建

新表 analytics_selfstat(快照头 + 设备维度 app_ver/oem/os + 设备级诊断量 batches_attempted/ok/fail、retries、queue_depth、端 sent_at + 服务端权威 created_at)与 analytics_selfstat_event(四类累计计数,FK → 快照头)。
Alembic 迁移 11c44afbea58(down_revision admin_user_plain_password):建两表 + 索引。
模型登记进 app/models/__init__.py(供 Alembic 发现);路由挂进 app/admin/main.py。
关键设计取舍
只存原始累计、查询时 Python 差分:admin 低频、量级小,跨 PG/SQLite 无方言坑(与 cps.py/coupon_data.py 同款约定)。
选「最新一条」用 max(id) 而非 max(created_at):id 严格单调,规避 SQLite 秒级时间戳撞车的歧义。
tz 口径统一:SQLite 返回 naive UTC、PG 返回 aware UTC,过滤前统一转 naive UTC 再比较。
基线子查询无下界:扫 date_from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。
测试(+14,全绿)
tests/test_analytics_selfstat.py(3):正常落库 / 空 events / 落库异常回 503。
tests/test_analytics_health.py(11):差分逻辑(首行=累计值、相邻差、epoch 重置换分区、乱序夹 0、同时间戳按 id 定序)、两段率公式、零分母→None、北京天边界、端点(overview 鉴权 / overview / breakdown)。
迁移 / 部署注意
部署需执行 alembic upgrade head(建两张新表)。纯新增表,无回填、无破坏性改动,向后兼容。
客户端需按约定 payload(snake_case、累计语义)对接 /api/v1/analytics/selfstat;admin 前端(独立仓 shaguabijia-admin-web)消费三个 analytics-health 端点,不在本 MR 范围。

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #127

app/admin/main.py

@@ -26,6 +26,7 @@ from app.admin.routers.cps import router as cps_router
 from app.admin.routers.dashboard import router as dashboard_router
 from app.admin.routers.device_liveness import router as device_liveness_router
 from app.admin.routers.ops_stat_config import router as ops_stat_config_router
+from app.admin.routers.analytics_health import router as analytics_health_router
 from app.admin.routers.event_logs import router as event_logs_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.feedback_qr import router as feedback_qr_router
@@ -97,6 +98,7 @@ admin_app.include_router(withdraw_router)
 admin_app.include_router(price_report_router)
 admin_app.include_router(feedback_router)
 admin_app.include_router(event_logs_router)
+admin_app.include_router(analytics_health_router)
 admin_app.include_router(feedback_qr_router)
 admin_app.include_router(admins_router)
 admin_app.include_router(roles_router)

app/admin/repositories/analytics_health.py

@@ -0,0 +1,146 @@
+"""埋点健康度聚合(埋点成功率 / 上报成功率)。
+
+只存原始累计快照,查询时在 Python 侧差分聚合(admin 低频、量级小,跨 PG/SQLite 无方言坑;
+与 cps.py / coupon_data.py 同款约定)。差分按 (device_id, epoch_id, event) 分区、created_at
+升序,相邻做差、负值夹 0;每增量按其快照 created_at 归入北京天桶。
+"""
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import UTC, datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core import rewards
+from app.models.analytics_selfstat import AnalyticsSelfStat as H
+from app.models.analytics_selfstat import AnalyticsSelfStatEvent as E
+
+_COUNTS = ("attempted", "drop_capture", "delivered", "drop_undelivered")
+
+
+def diff_snapshots(rows: list[dict]) -> list[dict]:
+    """累计快照行 → 每快照增量行(纯逻辑)。
+
+    rows 每行含 device_id/epoch_id/event/created_at/app_ver/oem/os + 四个累计计数。
+    返回每行含 dims + created_at + 四个增量 d_*(分区首行增量=累计值;负值夹 0)。
+    """
+    parts: dict[tuple, list[dict]] = defaultdict(list)
+    for r in rows:
+        parts[(r["device_id"], r["epoch_id"], r["event"])].append(r)
+
+    out: list[dict] = []
+    for group in parts.values():
+        group.sort(key=lambda r: (r["created_at"], r.get("id", 0)))
+        prev = {k: 0 for k in _COUNTS}
+        for r in group:
+            deltas = {f"d_{k}": max(0, int(r[k]) - prev[k]) for k in _COUNTS}
+            out.append({
+                "device_id": r["device_id"], "epoch_id": r["epoch_id"], "event": r["event"],
+                "created_at": r["created_at"], "app_ver": r["app_ver"],
+                "oem": r["oem"], "os": r["os"], **deltas,
+            })
+            prev = {k: int(r[k]) for k in _COUNTS}
+    return out
+
+
+def _cn_day(dt: datetime) -> str:
+    """created_at(UTC 口径)→ 北京日期字符串 YYYY-MM-DD。naive 当 UTC,tz-aware 直接换算。"""
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    return dt.astimezone(rewards.CN_TZ).date().isoformat()
+
+
+def _rates(sums: dict) -> dict:
+    """由四个增量和派生两段率(分母 0 → None)。"""
+    persisted_denom = sums["attempted"]
+    report_denom = sums["delivered"] + sums["drop_undelivered"]
+    return {
+        **sums,
+        "track_success_rate": (
+            (sums["attempted"] - sums["drop_capture"]) / persisted_denom
+            if persisted_denom else None
+        ),
+        "report_success_rate": (
+            sums["delivered"] / report_denom if report_denom else None
+        ),
+    }
+
+
+def _sum_deltas(deltas: list[dict]) -> dict:
+    return {k: sum(d[f"d_{k}"] for d in deltas) for k in _COUNTS}
+
+
+def _fetch_rows(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
+    """取 [from, to) 区间行 + 每分区在 from 左侧的最后一条基线行(供第一条区间增量做差)。"""
+    cols = (
+        H.id, H.device_id, H.epoch_id, E.event, H.created_at,
+        H.app_ver, H.oem, H.os,
+        E.attempted, E.drop_capture, E.delivered, E.drop_undelivered,
+    )
+    in_range = db.execute(
+        select(*cols).join(E, E.snapshot_id == H.id)
+        .where(H.created_at >= date_from, H.created_at < date_to)
+    ).mappings().all()
+
+    # 注:基线子查询无下界扫 from 左侧全量(spec §7 已接受的取舍;量级变大再上物化 rollup)。
+    # 用 max(id) 而非 max(created_at) 选"最新一条":id 严格单调,避免 SQLite 秒级时间戳撞车时选歧义。
+    sub = (
+        select(H.device_id, H.epoch_id, E.event, func.max(H.id).label("max_id"))
+        .join(E, E.snapshot_id == H.id)
+        .where(H.created_at < date_from)
+        .group_by(H.device_id, H.epoch_id, E.event)
+        .subquery()
+    )
+    baseline = db.execute(
+        select(*cols).join(E, E.snapshot_id == H.id).join(
+            sub, sub.c.max_id == H.id
+        )
+    ).mappings().all()
+
+    return [dict(r) for r in list(baseline) + list(in_range)]
+
+
+def _in_range_deltas(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
+    """差分后只保留 created_at ∈ [from, to) 的增量(基线行被差分用后丢弃)。
+
+    Python 侧过滤需对齐 tz 口径:SQLite 返回 naive UTC,PG 返回 aware UTC。
+    统一转成 naive UTC 再比较,兼容两种后端。
+    """
+    def _to_naive_utc(dt: datetime) -> datetime:
+        if dt.tzinfo is not None:
+            return dt.astimezone(UTC).replace(tzinfo=None)
+        return dt
+
+    from_naive = _to_naive_utc(date_from)
+    to_naive = _to_naive_utc(date_to)
+    deltas = diff_snapshots(_fetch_rows(db, date_from, date_to))
+    return [d for d in deltas if from_naive <= _to_naive_utc(d["created_at"]) < to_naive]
+
+
+def overview(db: Session, date_from: datetime, date_to: datetime) -> dict:
+    deltas = _in_range_deltas(db, date_from, date_to)
+    return _rates(_sum_deltas(deltas))
+
+
+def trend(db: Session, date_from: datetime, date_to: datetime) -> list[dict]:
+    deltas = _in_range_deltas(db, date_from, date_to)
+    by_day: dict[str, list[dict]] = defaultdict(list)
+    for d in deltas:
+        by_day[_cn_day(d["created_at"])].append(d)
+    return [
+        {"day": day, **_rates(_sum_deltas(items))}
+        for day, items in sorted(by_day.items())
+    ]
+
+
+def breakdown(db: Session, date_from: datetime, date_to: datetime, dim: str) -> list[dict]:
+    if dim not in ("event", "app_ver", "oem"):
+        raise ValueError(f"invalid dim: {dim!r}")
+    deltas = _in_range_deltas(db, date_from, date_to)
+    by_key: dict[str, list[dict]] = defaultdict(list)
+    for d in deltas:
+        by_key[d.get(dim) or "(unknown)"].append(d)
+    rows = [{"key": key, **_rates(_sum_deltas(items))} for key, items in by_key.items()]
+    rows.sort(key=lambda r: (r["report_success_rate"] is None, r["report_success_rate"] or 0.0))
+    return rows

app/admin/routers/analytics_health.py

@@ -0,0 +1,49 @@
+"""admin 埋点健康度:埋点成功率 / 上报成功率 总览 + 趋势 + 下钻(只读)。"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query
+
+from app.admin.deps import AdminDb, get_current_admin
+from app.admin.repositories import analytics_health as repo
+from app.admin.schemas.analytics_health import (
+    HealthBreakdownRow,
+    HealthMetrics,
+    HealthTrendPoint,
+)
+
+router = APIRouter(
+    prefix="/admin/api/analytics-health",
+    tags=["admin-analytics-health"],
+    dependencies=[Depends(get_current_admin)],
+)
+
+
+@router.get("/overview", response_model=HealthMetrics, summary="两段成功率总览")
+def overview(
+    db: AdminDb,
+    date_from: Annotated[datetime, Query()],
+    date_to: Annotated[datetime, Query()],
+) -> HealthMetrics:
+    return HealthMetrics(**repo.overview(db, date_from, date_to))
+
+
+@router.get("/trend", response_model=list[HealthTrendPoint], summary="按北京天趋势")
+def trend(
+    db: AdminDb,
+    date_from: Annotated[datetime, Query()],
+    date_to: Annotated[datetime, Query()],
+) -> list[HealthTrendPoint]:
+    return [HealthTrendPoint(**p) for p in repo.trend(db, date_from, date_to)]
+
+
+@router.get("/breakdown", response_model=list[HealthBreakdownRow], summary="按维度下钻")
+def breakdown(
+    db: AdminDb,
+    date_from: Annotated[datetime, Query()],
+    date_to: Annotated[datetime, Query()],
+    dim: Annotated[str, Query(pattern="^(event|app_ver|oem)$")] = "event",
+) -> list[HealthBreakdownRow]:
+    return [HealthBreakdownRow(**r) for r in repo.breakdown(db, date_from, date_to, dim)]

app/admin/schemas/analytics_health.py

@@ -0,0 +1,21 @@
+"""埋点健康度 admin 响应 schema。"""
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class HealthMetrics(BaseModel):
+    attempted: int
+    drop_capture: int
+    delivered: int
+    drop_undelivered: int
+    track_success_rate: float | None
+    report_success_rate: float | None
+
+
+class HealthTrendPoint(HealthMetrics):
+    day: str
+
+
+class HealthBreakdownRow(HealthMetrics):
+    key: str

app/api/v1/analytics.py

@@ -6,13 +6,18 @@ POST /api/v1/analytics/events — 批量接收新手引导(及后续)埋点,appe
 """
 from __future__ import annotations
 
-from fastapi import APIRouter, Request
+import logging
+
+from fastapi import APIRouter, HTTPException, Request
 
 from app.api.deps import DbSession
 from app.repositories import analytics as analytics_repo
+from app.repositories import analytics_selfstat as selfstat_repo
 from app.schemas.analytics import AnalyticsBatchIn, AnalyticsIngestOut
+from app.schemas.analytics_selfstat import SelfStatBatchIn, SelfStatIngestOut
 
 router = APIRouter(prefix="/api/v1/analytics", tags=["analytics"])
+logger = logging.getLogger("shagua.analytics")
 
 
 def _client_ip(request: Request) -> str:
@@ -29,3 +34,13 @@ def ingest_events(
 ) -> AnalyticsIngestOut:
     n = analytics_repo.record_batch(db, batch, client_ip=_client_ip(request))
     return AnalyticsIngestOut(received=n)
+
+
+@router.post("/selfstat", response_model=SelfStatIngestOut, summary="上报自报计数快照")
+def ingest_selfstat(batch: SelfStatBatchIn, db: DbSession) -> SelfStatIngestOut:
+    try:
+        snap_id = selfstat_repo.record_selfstat(db, batch)
+    except Exception:  # noqa: BLE001 — 计数链路要稳,落库失败不裸奔 500,记日志回明确错误
+        logger.exception("selfstat ingest failed device=%s epoch=%s", batch.device_id, batch.epoch_id)
+        raise HTTPException(status_code=503, detail="selfstat ingest failed") from None
+    return SelfStatIngestOut(snapshot_id=snap_id)

app/models/__init__.py

@@ -7,6 +7,10 @@ from app.models.ad_watch_log import AdWatchLog  # noqa: F401
 from app.models.admin import AdminAuditLog, AdminUser  # noqa: F401
 from app.models.admin_role import AdminRole  # noqa: F401
 from app.models.analytics_event import AnalyticsEvent  # noqa: F401
+from app.models.analytics_selfstat import (  # noqa: F401
+    AnalyticsSelfStat,
+    AnalyticsSelfStatEvent,
+)
 from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.cps_activity import CpsActivity  # noqa: F401

app/models/analytics_selfstat.py

@@ -0,0 +1,59 @@
+"""埋点/上报成功率自报计数快照表(append-only)。
+
+客户端周期上报「自 epoch 起算的累计计数」;服务端只存原始快照,查询时在 Python 侧差分聚合
+(见 app/admin/repositories/analytics_health.py)。与既有 analytics_event 表完全独立。
+
+- analytics_selfstat        :一快照一行(快照头 + 设备维度 + 设备级诊断量)
+- analytics_selfstat_event  :一 event 一行(四类累计计数),外键指向快照头
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class AnalyticsSelfStat(Base):
+    __tablename__ = "analytics_selfstat"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    device_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    epoch_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    # 设备维度(每设备固定,下钻用)
+    app_ver: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    oem: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    os: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # 设备级诊断量(累计;queue_depth 是瞬时 gauge)
+    batches_attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    batches_ok: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    batches_fail: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    retries: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    queue_depth: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    sent_at: Mapped[int | None] = mapped_column(BigInteger, nullable=True)  # 端上报时刻 epoch ms
+    # 服务端接收时间(权威,用于时间分桶与分区排序)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"<AnalyticsSelfStat id={self.id} device={self.device_id} epoch={self.epoch_id}>"
+
+
+class AnalyticsSelfStatEvent(Base):
+    __tablename__ = "analytics_selfstat_event"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    snapshot_id: Mapped[int] = mapped_column(
+        ForeignKey("analytics_selfstat.id"), index=True, nullable=False
+    )
+    event: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    attempted: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    drop_capture: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    delivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+    drop_undelivered: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"<AnalyticsSelfStatEvent snap={self.snapshot_id} event={self.event}>"

app/repositories/analytics_selfstat.py

@@ -0,0 +1,38 @@
+"""自报计数快照落库。一次事务:插 1 条快照头 + N 条 event 行,返回快照 id。"""
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from app.models.analytics_selfstat import AnalyticsSelfStat, AnalyticsSelfStatEvent
+from app.schemas.analytics_selfstat import SelfStatBatchIn
+
+
+def record_selfstat(db: Session, batch: SelfStatBatchIn) -> int:
+    snap = AnalyticsSelfStat(
+        device_id=batch.device_id,
+        epoch_id=batch.epoch_id,
+        app_ver=batch.app_ver,
+        oem=batch.oem,
+        os=batch.os,
+        batches_attempted=batch.batches_attempted,
+        batches_ok=batch.batches_ok,
+        batches_fail=batch.batches_fail,
+        retries=batch.retries,
+        queue_depth=batch.queue_depth,
+        sent_at=batch.sent_at,
+    )
+    db.add(snap)
+    db.flush()  # 拿到 snap.id
+    db.add_all([
+        AnalyticsSelfStatEvent(
+            snapshot_id=snap.id,
+            event=e.event,
+            attempted=e.attempted,
+            drop_capture=e.drop_capture,
+            delivered=e.delivered,
+            drop_undelivered=e.drop_undelivered,
+        )
+        for e in batch.events
+    ])
+    db.commit()
+    return snap.id

app/schemas/analytics_selfstat.py

@@ -0,0 +1,34 @@
+"""自报计数上报 schema。字段名对齐客户端 payload(snake_case),累计值语义。"""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class SelfStatEventIn(BaseModel):
+    event: str = Field(max_length=64)
+    attempted: int = 0
+    drop_capture: int = 0
+    delivered: int = 0
+    drop_undelivered: int = 0
+
+
+class SelfStatBatchIn(BaseModel):
+    device_id: str = Field(max_length=64)
+    epoch_id: str = Field(max_length=64)
+    sent_at: int | None = None
+    app_ver: str | None = Field(default=None, max_length=32)
+    oem: str | None = Field(default=None, max_length=32)
+    os: str | None = Field(default=None, max_length=32)
+    batches_attempted: int = 0
+    batches_ok: int = 0
+    batches_fail: int = 0
+    retries: int = 0
+    queue_depth: int = 0
+    # 允许空列表:某次快照只上报设备级计数(batches_*/retries/queue_depth)、无 event 细分时也合法
+    # (与 AnalyticsBatchIn 的 min_length=1 有意不同——那是行为事件、必须至少一条)。
+    events: list[SelfStatEventIn] = Field(default_factory=list, max_length=200)
+
+
+class SelfStatIngestOut(BaseModel):
+    ok: bool = True
+    snapshot_id: int