xinyouhe/yunke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(analytics): 环比同期对齐口径+趋势窗校验+异常新员工明细

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-04 17:34:14 +08:00
parent 02e100849b
commit ad153aa2f7
3 changed files with 130 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/specs/2026-09-04-yunke-analytics-module-design.md
+1 -1
View File
@@ -195,7 +195,7 @@ stat_daily_customer/group 物化表（msg_daily_stats 按 peer_kind 分组即等
| 待交接 | 归属员工 is_on_job=0（依赖 R1） |
| 该回未回 | 1:1 对话尾部消息 is_send=false 且属有效类型，且 max(最后消息, 最后通话) 距今 > N 小时（默认 2h），剔除 delete_flag/is_block |
| 等待时长 | now − max(最后消息 ts, 最后通话 ts) |
| 环比 | 仅周/月卡片（本周 vs 上周、本月 vs 上月）；**周起始=周一**（中国业务惯例，Python 侧计算） |
| 环比 | 仅周/月卡片；**同期对齐**：周=本周各日 vs 上周各对应日（-7天），月=本月各日 vs 上月同日（短月钳月末）；周起始=周一（中国业务惯例，Python 侧计算） |
| 日汇总 day | Asia/Shanghai 自然日 yyyymmdd，纯 Python 计算 |
| 异常警示 | 昨日消息总量 vs 前 7 日均值偏离 >50%，明细到员工级 |
yunke-server/app/analytics/metrics.py
+42 -15
View File
@@ -5,6 +5,7 @@
"""
from __future__ import annotations
import calendar
import datetime as dt
from collections.abc import Iterable
from typing import Any
@@ -53,13 +54,26 @@ def scope_day_keys(scope: str, today: int) -> list[int]:
def _prev_scope_days(scope: str, days: list[int]) -> list[int]:
"""紧邻的前一个等长窗口(环比;今日 scope 不调用)。"""
n = len(days)
last = _day_date(days[0]) - dt.timedelta(days=1)
return [
_date_key(last - dt.timedelta(days=i))
for i in range(n - 1, -1, -1)
]
"""同期对齐环比窗(与 days 等长逐日对齐;今日 scope 不调用)。
week=各日-7天(本周一..今 vs 上周一..上周同日);month=上月同日
(短月钳到上月末,如 3/31→2/28)。同期对齐而非滚动紧邻窗——后者
周会跨周末、月取上月尾巴,管理者会误读成「上周/上月」(spec §8)。
"""
if scope == "week":
return [_date_key(_day_date(d) - dt.timedelta(days=7)) for d in days]
if scope == "month":
out = []
for d in days:
cur = _day_date(d)
y, m = cur.year, cur.month - 1
if m == 0:
y, m = y - 1, 12
out.append(
_date_key(dt.date(y, m, min(cur.day, calendar.monthrange(y, m)[1])))
)
return out
raise ValueError(f"bad scope: {scope!r}")
async def _day_rows(
@@ -80,12 +94,15 @@ async def _day_rows(
).all()
async def _call_count(session: AsyncSession, channel_scope_days: list[int]) -> int:
"""phone_call 落窗计数(3k 行小表;start_call_time ms 与 day 键边界换算)。"""
if not channel_scope_days:
async def _call_count(session: AsyncSession, days: list[int]) -> int:
"""phone_call 落窗计数(3k 行小表;start_call_time ms 与 day 键边界换算)。
通话不按 messaging channel 过滤(spec §8:含全部 call-type)。
"""
if not days:
return 0
lo, _hi = day_bounds_ms(channel_scope_days[0])
hi_day_start, _ = day_bounds_ms(channel_scope_days[-1])
lo, _hi = day_bounds_ms(days[0])
hi_day_start, _ = day_bounds_ms(days[-1])
hi = hi_day_start + 86_400_000 # 含末日全天
val = (
await session.execute(
@@ -207,6 +224,14 @@ async def anomaly(session: AsyncSession, channel: str) -> dict | None:
"yesterday": y,
"prev7_mean": round(pmean, 1),
})
elif pmean == 0 and y > 0:
# 新员工:前 7 日无量、昨日有量——昨日新增的大异常源,不能从明细消失
staff_out.append({
"user_id": uid,
"yesterday": y,
"prev7_mean": 0.0,
"new": True,
})
if totals_p <= 0:
return None
delta = (totals_y - totals_p) / totals_p
@@ -215,7 +240,7 @@ async def anomaly(session: AsyncSession, channel: str) -> dict | None:
"total_delta_pct": round(delta, 3),
"yesterday_total": totals_y,
"prev7_mean": round(totals_p, 1),
"staff": sorted(staff_out, key=lambda s: s["yesterday"]),
"staff": sorted(staff_out, key=lambda s: -s["yesterday"]),
}
return None
@@ -223,7 +248,9 @@ async def anomaly(session: AsyncSession, channel: str) -> dict | None:
async def overview(
session: AsyncSession, *, channel: str, scope: str, trend_days: int = 30
) -> dict[str, Any]:
"""A1:卡片 + 环比(仅周/月) + 近 N 天趋势 + 异常警示。"""
"""A1:卡片 + 环比(仅周/月,同期对齐) + 近 N 天趋势 + 异常警示。"""
if trend_days < 1 or trend_days > 90:
raise ValueError("trend_days must be 1..90")
today = today_key()
days = scope_day_keys(scope, today)
rows = await _day_rows(session, channel, days)
@@ -238,7 +265,7 @@ async def overview(
"first_contact": await _first_contact(session, channel, days),
}
# 环比仅周/月(今日 vs 昨日全天会误导,spec §8)
# 环比仅周/月,同期对齐(今日 vs 昨日全天会误导,spec §8)
if scope in ("week", "month"):
prev_days = _prev_scope_days(scope, days)
prev_rows = await _day_rows(session, channel, prev_days)
yunke-server/tests/test_analytics_overview.py
+87 -1
View File
@@ -1,6 +1,8 @@
"""A1 总览指标:卡片/首沟通口径/环比(仅周月)/异常警示。"""
"""A1 总览指标:卡片/首沟通口径/环比同期对齐(仅周月)/异常警示。"""
from __future__ import annotations
import pytest
from app.analytics import metrics
from app.db.models import (
MsgDailyStats,
@@ -118,3 +120,87 @@ async def test_anomaly_fallback_multi_account(make_client, monkeypatch):
assert staff["u1"]["prev7_mean"] == 100.0
await make_client(seed)
def test_prev_scope_days_aligned():
"""环比窗=同期对齐(非滚动紧邻):周各日-7天;月=上月同日短月钳月末。"""
week = metrics.scope_day_keys("week", 20260902) # 周三:本周一..三
assert week == [20260831, 20260901, 20260902]
# 上周周一..上周三(各日-7)——不是跨周末的滚动紧邻窗
assert metrics._prev_scope_days("week", week) == [20260824, 20260825, 20260826]
month = metrics.scope_day_keys("month", 20260902)
assert month == [20260901, 20260902]
assert metrics._prev_scope_days("month", month) == [20260801, 20260802]
# 月末短月钳制:2026 非闰年 3/31→2/28;闰年 2024 3/31→2/29;跨年 1/1→上年 12/1
assert metrics._prev_scope_days("month", [20260331]) == [20260228]
assert metrics._prev_scope_days("month", [20240331]) == [20240229]
assert metrics._prev_scope_days("month", [20260101]) == [20251201]
async def test_overview_trend_days_validation(make_client, monkeypatch):
"""trend_days 越界(1..90 外)直接 ValueError,不静默出奇怪窗口。"""
monkeypatch.setattr(metrics, "today_key", lambda: _TODAY)
async def seed(session):
with pytest.raises(ValueError, match="trend_days"):
await metrics.overview(session, channel="wechat", scope="today",
trend_days=0)
with pytest.raises(ValueError, match="trend_days"):
await metrics.overview(session, channel="wechat", scope="today",
trend_days=91)
await make_client(seed)
async def test_week_delta_none_when_prev_week_empty(make_client, monkeypatch):
"""上周边无数据(prev_total=0)→ delta=None,不除零不误报。"""
monkeypatch.setattr(metrics, "today_key", lambda: _TODAY)
async def seed(session):
session.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
peer_kind="friend", peer_id="t1", day=20260904,
send_cnt=5, recv_cnt=0, last_ts=1788500000000,
last_is_send=True))
await session.commit()
out = await metrics.overview(session, channel="wechat", scope="week")
assert out["cards"]["prev_msg_total"] == 0
assert out["cards"]["msg_total_delta"] is None
await make_client(seed)
async def test_anomaly_none_when_no_prev_data(make_client, monkeypatch):
"""全员无前 7 日数据(totals_p<=0)→ 异常警示 None(无基数可算)。"""
monkeypatch.setattr(metrics, "today_key", lambda: _TODAY)
async def seed(session):
session.add(StatDailyStaff(day=20260903, channel="wechat", user_id="u1",
msg_sent=100, msg_recv=0,
active_friends=1, active_rooms=0))
await session.commit()
assert await metrics.anomaly(session, channel="wechat") is None
await make_client(seed)
async def test_anomaly_new_staff_detail_and_sort(make_client, monkeypatch):
"""新员工(前7日无量、昨日>0)进明细带 new 标记;明细按昨日量降序。"""
monkeypatch.setattr(metrics, "today_key", lambda: _TODAY)
async def seed(session):
for day, sent in [(20260827, 100), (20260828, 100), (20260829, 100),
(20260830, 100), (20260831, 100), (20260901, 100),
(20260902, 100), (20260903, 250)]:
session.add(StatDailyStaff(day=day, channel="wechat", user_id="u1",
msg_sent=sent, msg_recv=0,
active_friends=1, active_rooms=0))
# u2 昨日新出现(前 7 日无量)——修复前从明细里消失
session.add(StatDailyStaff(day=20260903, channel="wechat", user_id="u2",
msg_sent=80, msg_recv=0,
active_friends=1, active_rooms=0))
await session.commit()
a = await metrics.anomaly(session, channel="wechat")
assert a is not None
assert [s["user_id"] for s in a["staff"]] == ["u1", "u2"] # 降序:250 > 80
staff = {s["user_id"]: s for s in a["staff"]}
assert staff["u2"]["new"] is True
assert staff["u2"]["yesterday"] == 80
assert "new" not in staff["u1"] # 老员工不带标记
await make_client(seed)