feat(management): 期聚合查询——fact精确分位/期末存量跨渠道合并/distinct活跃/前窗基线窗口

Co-Authored-By: Claude Code <noreply@anthropic.com>

yunke-server/app/management/employee_metrics.py

@@ -0,0 +1,349 @@
+"""管理域期聚合查询(批次B)——评分/异常/employees API 的单一数据源。
+
+窗口:days=[start_day..end_day] 闭区间 yyyymmdd;上期=等长紧邻前窗;
+个人基线=紧邻当前窗之前的 28 天。所有查询走 (day, ...) 索引,不扫消息表
+(除 distinct 活跃客户走 msg_daily_stats friend 行)。
+
+口径要点:
+- 分位/回复计数:first_reply_fact 窗内行拉回 Python 按员工合并排序取
+  最近邻秩 ceil(q·n)(整除算术 (n+1)//2 / (9n+9)//10——SQLite 无原生
+  ceil,MySQL `/` 返回浮点,Python 侧取秩零方言风险)。合并按员工而非
+  owner:一员工多账号(wx1+wx2→u1,channel=None 时人人如此)的样本必须
+  跨账号归并,逐 owner 取分位再映射会互相覆盖。fact 全库全期 ≤6.8万行,
+  Python 侧聚合可行。
+- 期末存量(new/cooling/dormant):状态快照取「每 (user, channel) 最新
+  ≤end 行」——channel=None 时两渠道各取各的最新行再相加(直接取全渠道
+  最大 day 的那一行会丢另一渠道的存量)。
+- active_rooms_avg = SUM(active_rooms)/全窗天数(含无数据日摊薄),
+  非有数据天数——「日均负载」语义。
+- days_with_data = COUNT(DISTINCT day):单渠道时与 COUNT(*) 等价
+  (PK 含 day+channel,每日至多一行),channel=None 双渠道同日也只计 1 天。
+- 通话四列恒全渠道(employee_call_daily 无渠道维),channel 参数不过滤。
+"""
+from __future__ import annotations
+
+import datetime as dt
+import statistics
+from collections import namedtuple
+from collections.abc import Iterable
+from zoneinfo import ZoneInfo
+
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import (
+    EmployeeCallDaily,
+    EmployeeMetricDaily,
+    FirstReplyFact,
+    MsgDailyStats,
+    Staff,
+)
+from app.storage.staff_summary import account_map
+
+_SH = ZoneInfo("Asia/Shanghai")
+METRIC_VERSION = "v1"
+
+PeriodMetrics = namedtuple("PeriodMetrics", [
+    "user_id", "days_with_data", "msg_sent", "msg_recv", "cust_threads",
+    "threads_answered", "fr_sum_ms", "fr_p50_ms", "fr_p90_ms",
+    "reply_2h_count", "reply_24h_count",
+    "active_friends",          # 期内 distinct(直查,非日值求和)
+    "active_rooms_avg",        # 日均(全窗天数摊薄)
+    "new_customer", "cooling_customer", "dormant_customer",  # 期末单日存量
+    "reactivated_customer",    # 期内求和
+    "call_count", "call_seconds", "answered_call", "call_followup",
+])
+
+
+def _to_date(k: int) -> dt.date:
+    return dt.date(k // 10000, k // 100 % 100, k % 100)
+
+
+def _to_key(d: dt.date) -> int:
+    return d.year * 10000 + d.month * 100 + d.day
+
+
+def day_range(start: int, end: int) -> list[int]:
+    """闭区间日键列表。"""
+    out, cur = [], _to_date(start)
+    while cur <= _to_date(end):
+        out.append(_to_key(cur))
+        cur += dt.timedelta(days=1)
+    return out
+
+
+def prev_window(start: int, end: int) -> tuple[int, int]:
+    """等长紧邻前窗。"""
+    n = len(day_range(start, end))
+    pe = _to_date(start) - dt.timedelta(days=1)
+    ps = pe - dt.timedelta(days=n - 1)
+    return _to_key(ps), _to_key(pe)
+
+
+def baseline_window(start: int) -> tuple[int, int]:
+    """个人基线:紧邻当前窗之前的 28 天。"""
+    be = _to_date(start) - dt.timedelta(days=1)
+    bs = be - dt.timedelta(days=27)
+    return _to_key(bs), _to_key(be)
+
+
+def _percentile(vals: list[int], q_num: int, q_den: int) -> int | None:
+    """最近邻秩分位 ceil(q·n) 的样本值,q=q_num/q_den(1/2、9/10)。
+
+    整数算术 rank=(q_num*n + q_den - 1)//q_den 对正整数 n 恒落在 [1,n]
+    (与 math.ceil(q·n) 等价,与批次A _percentile 同口径);空列表→None
+    (与 0 严格区分)。
+    """
+    if not vals:
+        return None
+    s = sorted(vals)
+    n = len(s)
+    rank = max(1, min(n, (q_num * n + q_den - 1) // q_den))
+    return s[rank - 1]
+
+
+# 累加器初值:全员(staff)都出结果行,无数据员工为零值而非缺行——
+# employees API 全员列表语义。
+_ZERO: dict[str, object] = {
+    "days_with_data": 0, "msg_sent": 0, "msg_recv": 0, "cust_threads": 0,
+    "threads_answered": 0, "fr_sum_ms": 0,
+    "fr_p50_ms": None, "fr_p90_ms": None,
+    "reply_2h_count": 0, "reply_24h_count": 0,
+    "active_friends": 0, "active_rooms_sum": 0,
+    "new_customer": 0, "cooling_customer": 0, "dormant_customer": 0,
+    "reactivated_customer": 0,
+    "call_count": 0, "call_seconds": 0, "answered_call": 0, "call_followup": 0,
+}
+
+
+async def period_metrics(
+    session: AsyncSession,
+    start_day: int,
+    end_day: int,
+    channel: str | None,
+    user_ids: Iterable[str] | None = None,
+) -> dict[str, PeriodMetrics]:
+    """窗口 [start_day, end_day] × channel(None=全渠道)的员工期指标。
+
+    user_ids 传入时与 staff 表求交(幽灵过滤);返回含全部目标员工的
+    PeriodMetrics(无数据员工为零值行)。channel=None 时两渠道
+    account_map 合并(wx_id 与 WA uid 命名空间不撞,无冲突)。
+    """
+    staff_ids = {
+        u for u in (await session.execute(sa.select(Staff.user_id))).scalars() if u
+    }
+    target = staff_ids if user_ids is None else staff_ids & set(user_ids)
+    if not target:
+        return {}
+
+    if channel is None:
+        amap: dict[str, str] = {}
+        for ch in ("wechat", "whatsapp"):
+            amap.update(await account_map(session, ch))
+    else:
+        amap = await account_map(session, channel)
+
+    n_days = len(day_range(start_day, end_day))
+    acc: dict[str, dict] = {uid: dict(_ZERO) for uid in target}
+
+    # ---- 宽表 SUM:消息计数/串类/日均负载/复活 ----
+    wide_conds = [
+        EmployeeMetricDaily.day.between(start_day, end_day),
+        EmployeeMetricDaily.metric_version == METRIC_VERSION,
+        EmployeeMetricDaily.user_id.in_(target),
+    ]
+    if channel is not None:
+        wide_conds.append(EmployeeMetricDaily.channel == channel)
+    wide = (await session.execute(
+        sa.select(
+            EmployeeMetricDaily.user_id.label("uid"),
+            sa.func.count(sa.distinct(EmployeeMetricDaily.day)).label("nd"),
+            sa.func.sum(EmployeeMetricDaily.msg_sent).label("sent"),
+            sa.func.sum(EmployeeMetricDaily.msg_recv).label("recv"),
+            sa.func.sum(EmployeeMetricDaily.cust_threads).label("thr"),
+            sa.func.sum(EmployeeMetricDaily.threads_answered).label("ans"),
+            sa.func.sum(EmployeeMetricDaily.fr_sum_ms).label("frs"),
+            sa.func.sum(EmployeeMetricDaily.active_rooms).label("rooms"),
+            sa.func.sum(
+                EmployeeMetricDaily.reactivated_customer_count
+            ).label("react"),
+        ).where(*wide_conds).group_by(EmployeeMetricDaily.user_id)
+    )).all()
+    for uid, nd, sent, recv, thr, ans, frs, rooms, react in wide:
+        a = acc[uid]
+        a["days_with_data"] = int(nd or 0)
+        a["msg_sent"] = int(sent or 0)
+        a["msg_recv"] = int(recv or 0)
+        a["cust_threads"] = int(thr or 0)
+        a["threads_answered"] = int(ans or 0)
+        a["fr_sum_ms"] = int(frs or 0)
+        a["active_rooms_sum"] = int(rooms or 0)
+        a["reactivated_customer"] = int(react or 0)
+
+    # ---- 期末存量:每 (user, channel) 最新 ≤end 行,跨渠道相加 ----
+    stock_conds = [
+        EmployeeMetricDaily.day <= end_day,
+        EmployeeMetricDaily.metric_version == METRIC_VERSION,
+        EmployeeMetricDaily.user_id.in_(target),
+    ]
+    if channel is not None:
+        stock_conds.append(EmployeeMetricDaily.channel == channel)
+    latest = (await session.execute(
+        sa.select(
+            EmployeeMetricDaily.user_id.label("uid"),
+            EmployeeMetricDaily.channel.label("ch"),
+            sa.func.max(EmployeeMetricDaily.day).label("d"),
+        ).where(*stock_conds).group_by(
+            EmployeeMetricDaily.user_id, EmployeeMetricDaily.channel
+        )
+    )).all()
+    if latest:
+        # (user, channel, day) 含 metric_version 唯一(主键)——OR 点查
+        # 精确取回各渠道最新行,再按 user 跨渠道 SUM
+        stocks = (await session.execute(
+            sa.select(
+                EmployeeMetricDaily.user_id.label("uid"),
+                sa.func.sum(EmployeeMetricDaily.new_customer_count).label("nw"),
+                sa.func.sum(
+                    EmployeeMetricDaily.cooling_customer_count
+                ).label("cool"),
+                sa.func.sum(
+                    EmployeeMetricDaily.dormant_customer_count
+                ).label("dorm"),
+            ).where(sa.or_(*(
+                sa.and_(
+                    EmployeeMetricDaily.user_id == uid,
+                    EmployeeMetricDaily.channel == ch,
+                    EmployeeMetricDaily.day == d,
+                    EmployeeMetricDaily.metric_version == METRIC_VERSION,
+                )
+                for uid, ch, d in latest
+            ))).group_by(EmployeeMetricDaily.user_id)
+        )).all()
+        for uid, nw, cool, dorm in stocks:
+            a = acc[uid]
+            a["new_customer"] = int(nw or 0)
+            a["cooling_customer"] = int(cool or 0)
+            a["dormant_customer"] = int(dorm or 0)
+
+    # ---- fact 分位 + 2h/24h 回复计数(同一过滤) ----
+    fr_conds = [FirstReplyFact.day.between(start_day, end_day)]
+    if channel is not None:
+        fr_conds.append(FirstReplyFact.channel == channel)
+    fr_rows = (await session.execute(
+        sa.select(
+            FirstReplyFact.owner_account_id.label("owner"),
+            FirstReplyFact.fr_ms.label("fr"),
+        ).where(*fr_conds)
+    )).all()
+    fr_by_user: dict[str, list[int]] = {}
+    for owner, fr in fr_rows:
+        uid = amap.get(owner or "")
+        if uid is not None and uid in target:
+            fr_by_user.setdefault(uid, []).append(int(fr))
+    for uid, frs in fr_by_user.items():
+        acc[uid]["fr_p50_ms"] = _percentile(frs, 1, 2)
+        acc[uid]["fr_p90_ms"] = _percentile(frs, 9, 10)
+
+    reply = (await session.execute(
+        sa.select(
+            FirstReplyFact.owner_account_id.label("owner"),
+            sa.func.sum(sa.case(
+                (FirstReplyFact.fr_ms <= 7_200_000, 1), else_=0
+            )).label("c2h"),
+            sa.func.sum(sa.case(
+                (FirstReplyFact.fr_ms <= 86_400_000, 1), else_=0
+            )).label("c24h"),
+        ).where(*fr_conds).group_by(FirstReplyFact.owner_account_id)
+    )).all()
+    for owner, c2h, c24h in reply:
+        uid = amap.get(owner or "")
+        if uid is None or uid not in target:
+            continue
+        acc[uid]["reply_2h_count"] += int(c2h or 0)
+        acc[uid]["reply_24h_count"] += int(c24h or 0)
+
+    # ---- distinct 活跃客户:窗内 (owner, peer) 去重(friend 行) ----
+    af_conds = [
+        MsgDailyStats.day.between(start_day, end_day),
+        MsgDailyStats.peer_kind == "friend",
+    ]
+    if channel is not None:
+        af_conds.append(MsgDailyStats.channel == channel)
+    pairs = (await session.execute(
+        sa.select(MsgDailyStats.owner_account_id, MsgDailyStats.peer_id)
+        .distinct().where(*af_conds)
+    )).all()
+    active: dict[str, set[tuple[str, str]]] = {}
+    for owner, peer in pairs:
+        uid = amap.get(owner or "")
+        if uid is None or uid not in target:
+            continue
+        active.setdefault(uid, set()).add((owner or "", peer or ""))
+    for uid, pset in active.items():
+        acc[uid]["active_friends"] = len(pset)
+
+    # ---- 通话:恒全渠道(表无渠道维,channel 参数不过滤) ----
+    calls = (await session.execute(
+        sa.select(
+            EmployeeCallDaily.user_id.label("uid"),
+            sa.func.sum(EmployeeCallDaily.call_count).label("cc"),
+            sa.func.sum(EmployeeCallDaily.call_seconds).label("cs"),
+            sa.func.sum(EmployeeCallDaily.answered_call_count).label("ca"),
+            sa.func.sum(EmployeeCallDaily.call_followup_count).label("cf"),
+        ).where(
+            EmployeeCallDaily.day.between(start_day, end_day),
+            EmployeeCallDaily.user_id.in_(target),
+        ).group_by(EmployeeCallDaily.user_id)
+    )).all()
+    for uid, cc, cs, ca, cf in calls:
+        a = acc[uid]
+        a["call_count"] = int(cc or 0)
+        a["call_seconds"] = int(cs or 0)
+        a["answered_call"] = int(ca or 0)
+        a["call_followup"] = int(cf or 0)
+
+    return {
+        uid: PeriodMetrics(
+            user_id=uid,
+            days_with_data=a["days_with_data"],
+            msg_sent=a["msg_sent"],
+            msg_recv=a["msg_recv"],
+            cust_threads=a["cust_threads"],
+            threads_answered=a["threads_answered"],
+            fr_sum_ms=a["fr_sum_ms"],
+            fr_p50_ms=a["fr_p50_ms"],
+            fr_p90_ms=a["fr_p90_ms"],
+            reply_2h_count=a["reply_2h_count"],
+            reply_24h_count=a["reply_24h_count"],
+            active_friends=a["active_friends"],
+            active_rooms_avg=(a["active_rooms_sum"] / n_days) if n_days else 0.0,
+            new_customer=a["new_customer"],
+            cooling_customer=a["cooling_customer"],
+            dormant_customer=a["dormant_customer"],
+            reactivated_customer=a["reactivated_customer"],
+            call_count=a["call_count"],
+            call_seconds=a["call_seconds"],
+            answered_call=a["answered_call"],
+            call_followup=a["call_followup"],
+        )
+        for uid, a in acc.items()
+    }
+
+
+def team_medians(metrics: dict[str, PeriodMetrics]) -> dict[str, float | None]:
+    """每数值字段的全员中位数(纯函数,评分归一用)。
+
+    None(无样本的分位等)剔除后取中位——statistics.median:奇数取中位
+    样本、偶数取中间两数均值;全员 None → None(调用方按无基线处理)。
+    """
+    out: dict[str, float | None] = {}
+    for field in PeriodMetrics._fields:
+        if field == "user_id":
+            continue
+        vals = [
+            v for v in (getattr(m, field) for m in metrics.values())
+            if v is not None
+        ]
+        out[field] = statistics.median(vals) if vals else None
+    return out

yunke-server/tests/test_management_period_metrics.py

@@ -0,0 +1,276 @@
+"""管理域期聚合查询判别测试(批次B Task B1)。
+
+判据(v2 §3 / B1 规格):
+- 分位=fact 表最近邻秩 ceil(q·n) 精确值(整除算术,无 ceil 方言坑),
+  多账号/跨渠道样本按员工合并;
+- 宽表 SUM 按 (day∈窗, metric_version='v1', channel) 过滤;
+  active_rooms_avg 摊到全窗天数(含无数据日);
+- 期末生命周期存量=每 (user, channel) 最新 ≤end 行,跨渠道相加;
+- distinct 活跃客户=窗内 (owner, peer) 去重(peer_kind='friend');
+- 窗口工具:闭区间 day_range/等长前窗/28 天个人基线窗;
+- 幽灵员工(不在 staff)过滤;user_ids 交集;通话恒全渠道。
+"""
+import datetime as dt
+from zoneinfo import ZoneInfo
+
+from app.db.models import (
+    EmployeeCallDaily,
+    EmployeeMetricDaily,
+    FirstReplyFact,
+    MsgDailyStats,
+    Staff,
+    WechatAccount,
+    WhatsappAccount,
+)
+from app.management.employee_metrics import (
+    PeriodMetrics,
+    baseline_window,
+    day_range,
+    period_metrics,
+    prev_window,
+    team_medians,
+)
+
+_SH = ZoneInfo("Asia/Shanghai")
+
+
+def _ts(*args) -> int:
+    return int(dt.datetime(*args, tzinfo=_SH).timestamp() * 1000)
+
+
+async def _seed_org(s, users=("u1", "u2")):
+    for u in users:
+        s.add(Staff(user_id=u, name=u, is_on_job=True))
+    s.add(WechatAccount(wx_id="wx1", user_id="u1"))
+    s.add(WechatAccount(wx_id="wx2", user_id="u2"))
+    await s.commit()
+
+
+def _mday(day, *, channel="wechat", user_id="u1", metric_version="v1", **kw):
+    cols = {
+        "msg_sent": 0, "msg_recv": 0, "active_friends": 0, "active_rooms": 0,
+        "cust_threads": 0, "threads_answered": 0, "fr_sum_ms": 0,
+        "new_customer_count": 0, "cooling_customer_count": 0,
+        "dormant_customer_count": 0, "reactivated_customer_count": 0,
+        "data_complete": True,
+    }
+    cols.update(kw)
+    return EmployeeMetricDaily(
+        day=day, channel=channel, user_id=user_id,
+        metric_version=metric_version, **cols
+    )
+
+
+def _fact(fr_ms, *, channel="wechat", owner="wx1", peer="p1", day=20260901, i=0):
+    end = _ts(2026, 9, 1, 10, 0) + i * 60_000
+    return FirstReplyFact(
+        channel=channel, owner_account_id=owner, peer_id=peer,
+        thread_end_ts=end, reply_ts=end + fr_ms,
+        reply_msg_id=f"{owner}-{peer}-m{i}", day=day, fr_ms=fr_ms,
+    )
+
+
+async def test_percentile_exact(make_session):
+    """5 样本 p50=180s/p90=300s(ceil(2.5)=3、ceil(4.5)=5,1-based);
+    2 样本 p50=10s/p90=20s;窗外行不参与。"""
+    s = make_session
+    await _seed_org(s)
+    for i, sec in enumerate([60, 120, 180, 240, 300]):
+        s.add(_fact(sec * 1000, i=i))
+    for i, sec in enumerate([10, 20]):
+        s.add(_fact(sec * 1000, owner="wx2", peer="q1", i=i))
+    # 窗外(0906>end)长尾:混入则 p90=999s
+    s.add(_fact(999_000, i=9, day=20260906))
+    await s.commit()
+    m = await period_metrics(s, 20260901, 20260905, "wechat")
+    assert m["u1"].fr_p50_ms == 180_000
+    assert m["u1"].fr_p90_ms == 300_000
+    assert m["u2"].fr_p50_ms == 10_000
+    assert m["u2"].fr_p90_ms == 20_000
+
+
+async def test_period_sums_vs_daily_rows(make_session):
+    """宽表两天行 SUM 正确;days_with_data=2;active_rooms_avg 摊到全窗
+    5 天(含无数据日);channel/metric_version/窗外行都不计。"""
+    s = make_session
+    await _seed_org(s)
+    s.add(_mday(20260901, msg_sent=3, msg_recv=5, cust_threads=2,
+                threads_answered=1, fr_sum_ms=7_000, active_rooms=4,
+                reactivated_customer_count=1))
+    s.add(_mday(20260903, msg_sent=4, msg_recv=6, cust_threads=3,
+                threads_answered=2, fr_sum_ms=11_000, active_rooms=6,
+                reactivated_customer_count=2))
+    # 反例三连:错渠道 / 错版本 / 窗外
+    s.add(_mday(20260901, channel="whatsapp", msg_sent=100))
+    s.add(_mday(20260901, metric_version="v2", msg_sent=200))
+    s.add(_mday(20260906, msg_sent=400))
+    await s.commit()
+    m = await period_metrics(s, 20260901, 20260905, "wechat")
+    u1 = m["u1"]
+    assert u1.msg_sent == 7
+    assert u1.msg_recv == 11
+    assert u1.cust_threads == 5
+    assert u1.threads_answered == 3
+    assert u1.fr_sum_ms == 18_000
+    assert u1.days_with_data == 2
+    assert u1.active_rooms_avg == 2.0  # (4+6)/5 全窗天数,非 2 有数据日
+    assert u1.reactivated_customer == 3
+
+
+async def test_end_of_period_stock(make_session):
+    """期末存量=每渠道最新 ≤end 行:窗 [0901,0903] 取 0901 行、
+    [0901,0905] 取 0905 行;channel=None 跨渠道相加(4+2=6)。"""
+    s = make_session
+    await _seed_org(s)
+    s.add(_mday(20260901, new_customer_count=1, cooling_customer_count=2,
+                dormant_customer_count=3))
+    s.add(_mday(20260905, dormant_customer_count=4))
+    s.add(_mday(20260905, channel="whatsapp", dormant_customer_count=2))
+    await s.commit()
+
+    m1 = await period_metrics(s, 20260901, 20260903, "wechat")
+    assert m1["u1"].new_customer == 1
+    assert m1["u1"].cooling_customer == 2
+    assert m1["u1"].dormant_customer == 3
+
+    m2 = await period_metrics(s, 20260901, 20260905, "wechat")
+    assert m2["u1"].dormant_customer == 4  # wechat 渠道内最新行
+    assert m2["u1"].new_customer == 0
+
+    m3 = await period_metrics(s, 20260901, 20260905, None)
+    assert m3["u1"].dormant_customer == 6  # wechat 4 + whatsapp 2
+
+
+async def test_distinct_active_cross_day(make_session):
+    """同 peer 两天两行只计 1;room 行与幽灵 owner 不计。"""
+    s = make_session
+    await _seed_org(s)
+    for day, peer in [(20260901, "p1"), (20260902, "p1"), (20260902, "p2")]:
+        s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
+                            peer_kind="friend", peer_id=peer, day=day))
+    # room 行不计入活跃客户
+    s.add(MsgDailyStats(channel="wechat", owner_account_id="wx1",
+                        peer_kind="room", peer_id="r1", day=20260902))
+    # 幽灵 owner(account_map 无映射)不计
+    s.add(MsgDailyStats(channel="wechat", owner_account_id="wx9",
+                        peer_kind="friend", peer_id="p9", day=20260902))
+    await s.commit()
+    m = await period_metrics(s, 20260901, 20260902, "wechat")
+    assert m["u1"].active_friends == 2
+
+
+def test_window_helpers():
+    # day_range 闭区间 + 月末边界
+    assert day_range(20260901, 20260903) == [20260901, 20260902, 20260903]
+    assert day_range(20260831, 20260901) == [20260831, 20260901]
+    # prev_window 等长紧邻前窗(7 天窗 → 7 天窗;单日 → 单日)
+    assert prev_window(20260908, 20260914) == (20260901, 20260907)
+    assert prev_window(20260901, 20260901) == (20260831, 20260831)
+    # baseline_window:紧邻窗前 28 天([0915] → [0818, 0914])
+    assert baseline_window(20260915) == (20260818, 20260914)
+
+
+async def test_ghost_and_channel_none(make_session):
+    """幽灵(不在 staff)宽表行/fact 行不计;channel=None 合并两渠道
+    amap——u1(wx1+wa1)的宽表 SUM 与 fact 分位都跨渠道归并;
+    无数据员工(u2)也以零值行出现。"""
+    s = make_session
+    await _seed_org(s)
+    s.add(WhatsappAccount(uid="wa1", user_id="u1"))
+    # 账号存在但映射到非 staff 用户 → 幽灵
+    s.add(WechatAccount(wx_id="wxg", user_id="ghost"))
+    await s.commit()
+
+    s.add(_mday(20260901, msg_sent=3))
+    s.add(_mday(20260901, channel="whatsapp", msg_sent=4))
+    s.add(_mday(20260901, user_id="ghost", msg_sent=50))
+    s.add(_fact(100_000, owner="wx1", i=0))
+    s.add(_fact(300_000, channel="whatsapp", owner="wa1", peer="w1", i=1))
+    s.add(_fact(50_000, owner="wxg", i=2))
+    await s.commit()
+
+    m = await period_metrics(s, 20260901, 20260905, None)
+    assert set(m) == {"u1", "u2"}  # ghost 不在;零数据 u2 在
+    u1 = m["u1"]
+    assert u1.msg_sent == 7  # 两渠道行都归 u1
+    assert u1.fr_p50_ms == 100_000  # [100s,300s] 合并后 ceil(0.5*2)=1
+    assert u1.fr_p90_ms == 300_000  # ceil(0.9*2)=2
+    assert u1.reply_2h_count == 2
+    assert u1.reply_24h_count == 2
+    assert m["u2"].msg_sent == 0
+    assert m["u2"].fr_p50_ms is None
+
+
+async def test_reply_counts(make_session):
+    """fr=90min/3h/25h → 2h_count=1、24h_count=2(2h 是 24h 的子集)。"""
+    s = make_session
+    await _seed_org(s)
+    s.add(_fact(5_400_000, i=0))    # 90min ≤2h
+    s.add(_fact(10_800_000, i=1))   # 3h ∈(2h,24h]
+    s.add(_fact(90_000_000, i=2))   # 25h >24h
+    await s.commit()
+    m = await period_metrics(s, 20260901, 20260905, "wechat")
+    assert m["u1"].reply_2h_count == 1
+    assert m["u1"].reply_24h_count == 2
+    assert m["u2"].reply_2h_count == 0
+
+
+async def test_user_ids_filter_and_calls(make_session):
+    """user_ids 与 staff 求交(未知 id 剔除/空集→{});通话四列按窗 SUM
+    且不受 channel 参数影响(无渠道维,恒全渠道)。"""
+    s = make_session
+    await _seed_org(s)
+    s.add(_mday(20260901, user_id="u1", msg_sent=1))
+    s.add(_mday(20260901, user_id="u2", msg_sent=2))
+    s.add(EmployeeCallDaily(day=20260901, user_id="u1", call_count=3,
+                            call_seconds=600, answered_call_count=2,
+                            call_followup_count=1))
+    s.add(EmployeeCallDaily(day=20260902, user_id="u1", call_count=1,
+                            call_seconds=60, answered_call_count=1,
+                            call_followup_count=0))
+    s.add(EmployeeCallDaily(day=20260901, user_id="u2", call_count=9))
+    await s.commit()
+
+    m = await period_metrics(s, 20260901, 20260905, "wechat", user_ids=["u1", "nobody"])
+    assert set(m) == {"u1"}
+    assert m["u1"].msg_sent == 1
+    u1 = m["u1"]
+    assert u1.call_count == 4
+    assert u1.call_seconds == 660
+    assert u1.answered_call == 3
+    assert u1.call_followup == 1
+
+    empty = await period_metrics(s, 20260901, 20260905, "wechat", user_ids=[])
+    assert empty == {}
+
+    # channel="whatsapp" 时通话仍计入(恒全渠道)
+    mwa = await period_metrics(s, 20260901, 20260905, "whatsapp")
+    assert mwa["u1"].call_count == 4
+
+
+def test_team_medians():
+    """奇数样本取中位、偶数取均值;None 剔除;全 None → None;user_id 不参与。"""
+
+    def _pm(**kw):
+        base = dict(zip(PeriodMetrics._fields, [None] * len(PeriodMetrics._fields)))
+        base.update(user_id="x", days_with_data=0, msg_sent=0, msg_recv=0,
+                    cust_threads=0, threads_answered=0, fr_sum_ms=0,
+                    reply_2h_count=0, reply_24h_count=0, active_friends=0,
+                    active_rooms_avg=0.0, new_customer=0, cooling_customer=0,
+                    dormant_customer=0, reactivated_customer=0,
+                    call_count=0, call_seconds=0, answered_call=0,
+                    call_followup=0)
+        base.update(kw)
+        return PeriodMetrics(**base)
+
+    metrics = {
+        "u1": _pm(user_id="u1", msg_sent=10, fr_p50_ms=None),
+        "u2": _pm(user_id="u2", msg_sent=20, fr_p50_ms=100),
+        "u3": _pm(user_id="u3", msg_sent=30, fr_p50_ms=200),
+    }
+    med = team_medians(metrics)
+    assert med["msg_sent"] == 20
+    assert med["fr_p50_ms"] == 150.0  # None 剔除后 [100,200] 均值
+    assert "user_id" not in med
+    assert team_medians({})["msg_sent"] is None
+    assert team_medians({"u1": _pm()})["fr_p50_ms"] is None