"""管理域确定性异常规则(批次B Task B3,v1 §5.4 异常卡片)。 4 条规则,输入=当前+个人基线 PeriodMetrics,输出异常卡片 dict 列表。 全部确定性——不使用 LLM 推断,reason/title/suggested_action 均为固定 中文模板填数,reason 只描述数据事实(百分比/样本数/未回串数)。 规则表(触发比较均为严格 > / <): - reply_p90_worsened : cur.fr_p90_ms > base×1.5 且 cur.cust_threads≥ soft_threads;>×2.5 升 critical - reply_rate_dropped : derive(reply_2h_rate) cur < base 同式×0.7 且 cur.cust_threads≥soft_threads(率由 scoring. derive_metrics 派生,与评分同一口径) - activity_drop : cur 日均 (msg_sent+msg_recv)/n_days < base 日均 (…/28)×0.5 且 base.days_with_data≥min_days (基线太稀疏不可比;基线窗口恒 28 天=baseline_window 语义,当前窗口天数由 n_days 传入) - workload_overload : cur.active_friends(周期 distinct) > workload 维 active_friends_band.high(与评分 band 同源阈值) 无基线不判:baseline=None 直接返回 [](新员工无个人对比位即不出卡, 含与基线无关的 overload——新入职负载爬坡期误报多,不判);某侧值 None/不可算(无串样本的率、无 fact 的分位、0ms 退化基线)→ 对应规则 不触发。 纯函数(无 session/IO):B4 dashboard/employees 对全员逐个调用。 阈值不硬编码:soft_threads/min_days 取 config.sample_guards,过载上限取 config.dimensions.workload.metrics.active_friends_band.high;乘数 (1.5/2.5/0.7/0.5)是规则语义本身,定义为模块常量而非可调参数。 """ from __future__ import annotations from app.management.employee_metrics import PeriodMetrics from app.management.scoring import derive_metrics # 规则乘数(语义常量,非可调参数——调整即换规则版本) P90_WARN_MULTIPLIER = 1.5 P90_CRITICAL_MULTIPLIER = 2.5 RATE_DROP_MULTIPLIER = 0.7 ACTIVITY_DROP_MULTIPLIER = 0.5 BASELINE_WINDOW_DAYS = 28 # 个人基线窗口恒 28 天(employee_metrics.baseline_window) # ---- 固定文案模板(.format 填数,reason 只含数据事实) ---- TITLE_P90 = "回复 P90 较个人基线上升 {pct}%" TITLE_RATE = "2h 回复率较个人基线下降 {pct}%" TITLE_ACTIVITY = "日均消息量较个人基线下降 {pct}%" TITLE_OVERLOAD = "活跃客户 {current} 人超过负载上限 {limit} 人" REASON_P90 = ( "近 {n_days} 天首响 P90 为 {cur}(基线前 28 天为 {base}),上升 {pct}%;" "期内 {threads} 串客户消息中 {unreplied} 串未回" ) REASON_RATE = ( "近 {n_days} 天 2h 回复率为 {cur:.0%}({cur_cnt}/{cur_thr} 串)," "基线(前 28 天)为 {base:.0%}({base_cnt}/{base_thr} 串);" "期内 {unreplied} 串未回" ) REASON_ACTIVITY = ( "近 {n_days} 天日均消息 {cur:.1f} 条/天,基线(前 28 天)日均 {base:.1f} 条/天," "下降 {pct}%;当前活跃客户 {friends} 人" ) REASON_OVERLOAD = ( "近 {n_days} 天活跃客户 {current} 人(周期 distinct 口径)," "超过配置上限 {limit} 人" ) ACTION_P90 = "查看超时未回客户明细,与员工确认近期负载与接待安排" ACTION_RATE = "查看未回串明细,排查回复流程与时段覆盖问题" ACTION_ACTIVITY = "与员工确认在岗状态与客户维护情况,并核对账号数据接入是否正常" ACTION_OVERLOAD = "评估客户分配与负载均衡,考虑分流部分客户或安排协助" def _fmt_ms(ms: float) -> str: # 与 scoring._fmt_ms 同式(模块私有不跨文件引用) minutes = ms / 60_000 if minutes >= 60 and minutes % 60 == 0: return f"{minutes // 60:g}h" return f"{minutes:g}min" def _pct_up(cur: float, base: float) -> int: """较基线上升百分比(四舍五入取整;触发后调用,base>0 已保证)。""" return int(round((cur - base) / base * 100)) def _pct_down(cur: float, base: float) -> int: """较基线下降百分比(触发后调用,base>0 已由触发条件保证)。""" return int(round((base - cur) / base * 100)) def detect_anomalies( user_id: str, current: PeriodMetrics, baseline: PeriodMetrics | None, *, n_days: int, config: dict, ) -> list[dict]: """单员工确定性异常检测(纯函数,卡片结构见 v1 §5.4/模块 docstring)。 输出按规则表序:p90 恶化 → 回复率骤降 → 活动骤降 → 过载 (排序/聚合由 B4 调用方按 severity/affected_count 处理)。 """ if baseline is None: return [] # 无基线不判(新员工) guards = config.get("sample_guards", {}) soft = guards.get("soft_threads", 10) min_days = guards.get("min_days", 5) band_high = ( config.get("dimensions", {}) .get("workload", {}) .get("metrics", {}) .get("active_friends_band", {}) .get("high") ) cards: list[dict] = [] unreplied = current.cust_threads - current.threads_answered threads_ok = current.cust_threads >= soft # ---- 1. reply_p90_worsened:P90 较个人基线恶化 ×1.5/×2.5 ---- if ( threads_ok and current.fr_p90_ms is not None and baseline.fr_p90_ms is not None and baseline.fr_p90_ms > 0 # 0ms 基线为退化数据,不可比 ): base_p90 = baseline.fr_p90_ms if current.fr_p90_ms > base_p90 * P90_CRITICAL_MULTIPLIER: severity = "critical" elif current.fr_p90_ms > base_p90 * P90_WARN_MULTIPLIER: severity = "warning" else: severity = None if severity is not None: pct = _pct_up(current.fr_p90_ms, base_p90) cards.append({ "rule_code": "reply_p90_worsened", "severity": severity, "employee_user_id": user_id, "title": TITLE_P90.format(pct=pct), "current_value": current.fr_p90_ms, "baseline_value": base_p90, "unit": "ms", "sample_size": current.cust_threads, "affected_count": unreplied, "reason": REASON_P90.format( n_days=n_days, cur=_fmt_ms(current.fr_p90_ms), base=_fmt_ms(base_p90), pct=pct, threads=current.cust_threads, unreplied=unreplied, ), "drilldown": {"type": "metric", "query": { "metric_code": "first_reply_p90", "user_id": user_id}}, "suggested_action": ACTION_P90, }) # ---- 2. reply_rate_dropped:2h 回复率较基线降 30%+ ---- if threads_ok: cur_rate = derive_metrics(current)["reply_2h_rate"] base_rate = derive_metrics(baseline)["reply_2h_rate"] if ( cur_rate is not None and base_rate is not None and cur_rate < base_rate * RATE_DROP_MULTIPLIER ): cards.append({ "rule_code": "reply_rate_dropped", "severity": "warning", "employee_user_id": user_id, "title": TITLE_RATE.format( pct=_pct_down(cur_rate, base_rate)), "current_value": cur_rate, "baseline_value": base_rate, "unit": "0-1", "sample_size": current.cust_threads, "affected_count": unreplied, "reason": REASON_RATE.format( n_days=n_days, cur=cur_rate, cur_cnt=current.reply_2h_count, cur_thr=current.cust_threads, base=base_rate, base_cnt=baseline.reply_2h_count, base_thr=baseline.cust_threads, unreplied=unreplied, ), "drilldown": {"type": "metric", "query": { "metric_code": "unreplied", "user_id": user_id}}, "suggested_action": ACTION_RATE, }) # ---- 3. activity_drop:日均消息量较基线腰斩 ---- if n_days > 0 and baseline.days_with_data >= min_days: cur_daily = (current.msg_sent + current.msg_recv) / n_days base_daily = ( baseline.msg_sent + baseline.msg_recv ) / BASELINE_WINDOW_DAYS if cur_daily < base_daily * ACTIVITY_DROP_MULTIPLIER: cards.append({ "rule_code": "activity_drop", "severity": "warning", "employee_user_id": user_id, "title": TITLE_ACTIVITY.format( pct=_pct_down(cur_daily, base_daily)), "current_value": cur_daily, "baseline_value": base_daily, "unit": "条", "sample_size": current.active_friends, "affected_count": current.active_friends, "reason": REASON_ACTIVITY.format( n_days=n_days, cur=cur_daily, base=base_daily, pct=_pct_down(cur_daily, base_daily), friends=current.active_friends, ), "drilldown": {"type": "employee", "query": {"user_id": user_id}}, "suggested_action": ACTION_ACTIVITY, }) # ---- 4. workload_overload:活跃客户超 band 上限 ---- if band_high is not None and current.active_friends > band_high: cards.append({ "rule_code": "workload_overload", "severity": "warning", "employee_user_id": user_id, "title": TITLE_OVERLOAD.format( current=current.active_friends, limit=band_high), "current_value": current.active_friends, "baseline_value": None, # 与个人基线无关,阈值在 band 配置 "unit": "人", "sample_size": current.active_friends, "affected_count": current.active_friends, "reason": REASON_OVERLOAD.format( n_days=n_days, current=current.active_friends, limit=band_high, ), "drilldown": {"type": "employee", "query": {"user_id": user_id}}, "suggested_action": ACTION_OVERLOAD, }) return cards