feat(analytics): admin health aggregation (python-side diff)

app/admin/repositories/analytics_health.py

@@ -0,0 +1,138 @@
+"""埋点健康度聚合(埋点成功率 / 上报成功率)。
+
+只存原始累计快照,查询时在 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 and_, 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"])
+        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.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()
+
+    sub = (
+        select(H.device_id, H.epoch_id, E.event, func.max(H.created_at).label("mx"))
+        .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,
+            and_(
+                sub.c.device_id == H.device_id,
+                sub.c.epoch_id == H.epoch_id,
+                sub.c.event == E.event,
+                sub.c.mx == H.created_at,
+            ),
+        )
+    ).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) 的增量(基线行被差分用后丢弃)。"""
+    deltas = diff_snapshots(_fetch_rows(db, date_from, date_to))
+    return [d for d in deltas if date_from <= d["created_at"] < date_to]
+
+
+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]:
+    assert dim in ("event", "app_ver", "oem")
+    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

tests/test_analytics_health.py

@@ -0,0 +1,54 @@
+"""埋点健康度聚合:纯差分函数 + admin 端点。"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from app.admin.repositories.analytics_health import diff_snapshots
+
+
+def _row(device, epoch, event, ts, **cum) -> dict:
+    base = {"attempted": 0, "drop_capture": 0, "delivered": 0, "drop_undelivered": 0}
+    base.update(cum)
+    return {
+        "device_id": device, "epoch_id": epoch, "event": event,
+        "created_at": datetime(2026, 7, 1, ts, 0, tzinfo=timezone.utc),
+        "app_ver": "0.2.12(62)", "oem": "ColorOS", "os": "Android 14",
+        **base,
+    }
+
+
+def test_diff_first_row_is_full_cumulative() -> None:
+    rows = [_row("d", "e", "video_play", 1, attempted=100, delivered=90)]
+    out = diff_snapshots(rows)
+    assert len(out) == 1
+    assert out[0]["d_attempted"] == 100
+    assert out[0]["d_delivered"] == 90
+
+
+def test_diff_consecutive_delta() -> None:
+    rows = [
+        _row("d", "e", "video_play", 1, attempted=100, delivered=90),
+        _row("d", "e", "video_play", 2, attempted=150, delivered=140),
+    ]
+    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
+    assert out[1]["d_attempted"] == 50
+    assert out[1]["d_delivered"] == 50
+
+
+def test_diff_epoch_reset_new_partition() -> None:
+    rows = [
+        _row("d", "e1", "video_play", 1, attempted=100),
+        _row("d", "e2", "video_play", 2, attempted=5),
+    ]
+    out = {(r["epoch_id"]): r["d_attempted"] for r in diff_snapshots(rows)}
+    assert out["e1"] == 100
+    assert out["e2"] == 5
+
+
+def test_diff_clamps_negative_on_reorder() -> None:
+    rows = [
+        _row("d", "e", "video_play", 1, attempted=100),
+        _row("d", "e", "video_play", 2, attempted=80),
+    ]
+    out = sorted(diff_snapshots(rows), key=lambda r: r["created_at"])
+    assert out[1]["d_attempted"] == 0