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

fix(management): 宽表生命周期hist改时点语义(≤as_of累积)——历史日dormant不再被未来消息虚增;补正向同源/percentile None判别+防御性flush

Browse Source 
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-09-09 14:44:35 +08:00
parent 5bfcebf20c
commit ad3beb7625
2 changed files with 80 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yunke-server/app/storage/employee_metrics.py
+17 -9
View File
@@ -15,6 +15,7 @@ import logging
import math
from bisect import bisect_right
from collections.abc import Iterable
from itertools import accumulate
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession
@@ -170,8 +171,10 @@ async def _lifecycle_daily(
) -> dict[int, dict[str, dict[str, int]]]:
"""逐日 {day: {user: {state: count}}}——customer 当前归属 + 互动日 sweep。
数据源:msg_daily_stats friend 行全历史(互动日/条数)。16k 客户 × 窗口
天数 bisect,夜间全量约秒级;4h tick 只算当日+3 天。
数据源:msg_daily_stats friend 行全历史(互动日/条数)。hist 为 ≤as_of
的历史消息数(时点语义:与 last 同一 bisect 位取按日累积数组 cum[idx],
as_of 之后的消息不虚增历史日的沉睡门槛)。16k 客户 × 窗口天数
bisect+classify,夜间全量实测见批次A核实文档;4h tick 只算当日+3 天。
"""
rows = (
await session.execute(
@@ -184,12 +187,13 @@ async def _lifecycle_daily(
)
)
).all()
days_map: dict[str, set[int]] = {}
cnt: dict[str, int] = {}
# peer → {day: 消息条数};days_list 与累积数组 cum 对齐:
# cum[i] = 该客户 ≤days_list[i] 的历史消息数(hist 时点语义)
day_msgs: dict[str, dict[int, int]] = {}
for peer, day, snd, rcv in rows:
days_map.setdefault(peer, set()).add(day)
cnt[peer] = cnt.get(peer, 0) + int(snd or 0) + int(rcv or 0)
interaction = {p: sorted(ds) for p, ds in days_map.items()}
dm = day_msgs.setdefault(peer, {})
dm[day] = dm.get(day, 0) + int(snd or 0) + int(rcv or 0)
interaction = {p: sorted(dm) for p, dm in day_msgs.items()}
custs = (
await session.execute(
@@ -212,14 +216,15 @@ async def _lifecycle_daily(
if uid is None:
continue
first = days_list[0]
hist = cnt.get(peer, 0)
cum = list(accumulate(day_msgs[peer][d] for d in days_list))
for d in days:
idx = bisect_right(days_list, d) - 1
if idx < 0:
continue
last = days_list[idx]
prev = days_list[idx - 1] if idx > 0 else None
state = classify(first, last, prev, d, hist)
hist_asof = cum[idx]
state = classify(first, last, prev, d, hist_asof)
if state:
out[d].setdefault(uid, {}).setdefault(state, 0)
out[d][uid][state] += 1
@@ -345,4 +350,7 @@ async def recompute_metric_days(
)
)
written += 1
# 防御性 flush:防 autoflush=False 类配置下逐日 add 的 ORM 对象
# 攒到事务尾(夜间全量约 52k 行;与 Task3 周期性 flush 同教训)
await session.flush()
return written
yunke-server/tests/test_employee_metrics.py
+63
View File
@@ -123,6 +123,7 @@ async def test_metric_daily_friend_only_caliber(make_session):
assert row.msg_sent == 3 and row.msg_recv == 2 # room 的 10 不进
assert row.active_rooms == 1
assert row.cust_threads == 1 and row.threads_answered == 0 # 未回复串
assert row.first_reply_p50_ms is None and row.first_reply_p90_ms is None # 空样本→None(≠0)
async def test_metric_daily_ghost_filtered(make_session):
@@ -173,6 +174,35 @@ async def test_metric_daily_same_source_as_fact(make_session):
sa.select(EmployeeMetricDaily))).scalars().all()) == 2
async def test_metric_daily_same_source_as_fact_answered(make_session):
"""同源判别正向路径(v2 §1.1 #5):被回串——threads_answered ≡ fact 行数,
单样本 p50=p90=fr_ms(锁分位非 None 映射)。"""
from app.storage.first_reply_facts import recompute_pairs
s = make_session
await _seed_org(s)
# 客户 10:00 提问,员工 10:25 回复 → fr=25min,串末条归属 09-01
fr_ms = 25 * 60 * 1000
s.add(Message(msg_id="an0", channel="wechat", owner_account_id="wx1",
talker_id="pa", timestamp=_ts(2026, 9, 1, 10, 0),
is_send=False, type="text", text="q"))
s.add(Message(msg_id="an1", channel="wechat", owner_account_id="wx1",
talker_id="pa", timestamp=_ts(2026, 9, 1, 10, 25),
is_send=True, type="text", text="a"))
await _seed_msg_rows(s, day=20260901, peer="pa", send=1, recv=1)
await s.commit()
await recompute_pairs(s, [("wechat", "wx1", "pa")])
await s.commit()
await recompute_metric_days(s, [20260901])
await s.commit()
n_facts = (await s.execute(
sa.select(sa.func.count()).select_from(FirstReplyFact))).scalar_one()
row = (await s.execute(sa.select(EmployeeMetricDaily))).scalar_one()
assert row.cust_threads == 1
assert row.threads_answered == 1 == n_facts
assert row.fr_sum_ms == fr_ms
assert row.first_reply_p50_ms == fr_ms == row.first_reply_p90_ms
async def test_metric_daily_lifecycle_counts(make_session):
"""生命周期列:cooling/dormant/reactivated 判定进宽表。"""
s = make_session
@@ -204,3 +234,36 @@ async def test_metric_daily_lifecycle_counts(make_session):
await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()}
assert got[20261010].dormant_customer_count == 1 # 39 天静默+历史 8 条
assert got[20261011].reactivated_customer_count == 1 # 当日回流且 gap>30
async def test_metric_daily_lifecycle_hist_point_in_time(make_session):
"""hist 时点语义判别:回填历史日 M,沉睡门槛只计 ≤M 的消息。
客户 pd:X=09-01 共 2 条 + Y=10-02 共 3 条(全历史 5 条):
- M=10-01(X+30,<Y):gap≥30 达沉睡判定,但 hist_asof=2 < 5 → None
(全历史口径 hist=5 会把 M 误判 dormant——M 之后的消息虚增);
- D=11-01(Y+30):hist_asof=5 ≥ 5 → dormant。"""
s = make_session
await _seed_org(s)
s.add(Customer(friend_id="pd", channel="wechat", owner_account_id="wx1",
delete_flag=0))
for i, (day, h, is_send) in enumerate([
(20260901, 10, False), (20260901, 11, True), # X 2 条
(20261002, 10, False), (20261002, 11, True), (20261002, 12, False), # Y 3 条
]):
s.add(Message(msg_id=f"pd{i}", channel="wechat", owner_account_id="wx1",
talker_id="pd",
timestamp=_ts(day // 10000, day // 100 % 100, day % 100, h, 0),
is_send=is_send, type="text", text="x"))
await _seed_msg_rows(s, day=20260901, peer="pd", send=1, recv=1)
await _seed_msg_rows(s, day=20261002, peer="pd", send=1, recv=2)
# M 是无消息日:state=None 时纯生命周期源不出行,补一条 room 行撑出
# 宽表行,让 dormant==0 有断言落点
await _seed_msg_rows(s, day=20261001, kind="room", peer="r1", send=1, recv=0)
await s.commit()
await recompute_metric_days(s, [20261001, 20261101])
await s.commit()
got = {r.day: r for r in (
await s.execute(sa.select(EmployeeMetricDaily))).scalars().all()}
assert got[20261001].dormant_customer_count == 0 # hist_asof=2<5 → None
assert got[20261101].dormant_customer_count == 1 # hist_asof=5 ≥ 5