浅念
228 lines
8.7 KiB
Python
228 lines
8.7 KiB
Python
"""管理域评分引擎(批次B)——试算版(v2 §0 决议:全默认+试算标记)。
|
||
|
||
三类标准化(v1 §10.2):
|
||
- target : floor..target 线性 0..100(≤floor 得 0,≥target 得 100)
|
||
- reverse : ≤target_ms 100 分,≥floor_ms 0 分,中间线性降
|
||
- band : 区间内 100,超上限 80(不再加),低于下限 None(负载不足→不评分)
|
||
样本保护(v1 §10.3):指标样本量=串类 cust_threads/band 类 active_friends;
|
||
<soft_threads(10) insufficient 不评分;soft(10)~min_threads(30) 低置信照评。
|
||
维度分=指标均分(等权,None 剔除按剩余归一);总分=可用维度加权重归一
|
||
(满分 Σweight×100/Σweight,一期可用 70 → 100/70 放大回百分制)——
|
||
action/risk pending_data 不参与,不默认 0/100。
|
||
|
||
band 基数=周期 distinct active_friends(basis=period_distinct,B2 决议:
|
||
比单日期末值更抗噪声且与 B1 口径一致)。
|
||
|
||
team 中位数键=派生指标名(spec 的 metric 字段,无则 code 本身):
|
||
fr_p90_ms/active_friends/call_count 可直接用 team_medians 输出,
|
||
reply_*_rate/active_share 需调用方对派生值取中位后传入。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
|
||
from app.management.employee_metrics import PeriodMetrics
|
||
from app.management.metric_definitions import METRIC_DEFINITIONS
|
||
|
||
_GRADE_ORDER = ("A", "B", "C", "D") # 低于 D 线为 E
|
||
|
||
|
||
def normalize(spec: dict, value: float | None) -> float | None:
|
||
"""三类标准化(v1 §10.2):value None→None;band 低于下限→None。
|
||
|
||
- target : (v-floor)/(target-floor) 线性 0..100,两端截断
|
||
- reverse : (floor_ms-v)/(floor_ms-target_ms) 线性降 100..0,两端截断
|
||
- band : [low,high] 内 100,>high 80(不再加),<low None(不评分)
|
||
"""
|
||
if value is None:
|
||
return None
|
||
mtype = spec["type"]
|
||
if mtype == "target":
|
||
floor, target = spec["floor"], spec["target"]
|
||
if value <= floor:
|
||
return 0.0
|
||
if value >= target:
|
||
return 100.0
|
||
return (value - floor) / (target - floor) * 100.0
|
||
if mtype == "reverse":
|
||
target_ms, floor_ms = spec["target_ms"], spec["floor_ms"]
|
||
if value <= target_ms:
|
||
return 100.0
|
||
if value >= floor_ms:
|
||
return 0.0
|
||
return (floor_ms - value) / (floor_ms - target_ms) * 100.0
|
||
if mtype == "band":
|
||
low, high = spec["low"], spec["high"]
|
||
if value < low:
|
||
return None
|
||
if value > high:
|
||
return 80.0
|
||
return 100.0
|
||
raise ValueError(f"未知指标类型 {mtype!r}(须 target/reverse/band)")
|
||
|
||
|
||
def _derive_metrics(pm: PeriodMetrics) -> dict[str, float | None]:
|
||
"""从 PeriodMetrics 派生可评分指标值(键=指标名,None=无值≠0)。"""
|
||
out: dict[str, float | None] = {
|
||
"fr_p90_ms": pm.fr_p90_ms,
|
||
"active_friends": pm.active_friends, # band 用,周期 distinct
|
||
"call_count": pm.call_count, # band 用,period_total
|
||
}
|
||
if pm.cust_threads:
|
||
out["reply_2h_rate"] = pm.reply_2h_count / pm.cust_threads
|
||
out["reply_24h_rate"] = pm.reply_24h_count / pm.cust_threads
|
||
else:
|
||
# 无串样本→无率(严格区分 0%),分子分母同源
|
||
out["reply_2h_rate"] = None
|
||
out["reply_24h_rate"] = None
|
||
denom = pm.end_active_friends + pm.cooling_customer + pm.dormant_customer
|
||
out["active_share"] = (
|
||
pm.end_active_friends / denom if denom > 0 else None
|
||
)
|
||
return out
|
||
|
||
|
||
def _fmt_ms(ms: float) -> str:
|
||
minutes = ms / 60_000
|
||
if minutes >= 60 and minutes % 60 == 0:
|
||
return f"{minutes // 60:g}h"
|
||
return f"{minutes:g}min"
|
||
|
||
|
||
def _formula_note(spec: dict) -> str:
|
||
"""指标计分公式的中文说明(B4 透出,阈值全部来自 config)。"""
|
||
mtype = spec["type"]
|
||
if mtype == "target":
|
||
return (
|
||
f"floor {spec['floor']}..target {spec['target']} 线性 0..100 分"
|
||
"(≤floor 0,≥target 100)"
|
||
)
|
||
if mtype == "reverse":
|
||
return (
|
||
f"≤{_fmt_ms(spec['target_ms'])} 100 分,"
|
||
f"≥{_fmt_ms(spec['floor_ms'])} 0 分,中间线性降"
|
||
)
|
||
return (
|
||
f"{spec['low']}..{spec['high']} 区间内 100 分,超上限 80 分,"
|
||
f"低于 {spec['low']} 不评分(负载不足);basis={spec.get('basis', '-')}"
|
||
)
|
||
|
||
|
||
def score_employee(
|
||
config: dict,
|
||
current: PeriodMetrics,
|
||
baseline: PeriodMetrics | None = None,
|
||
team: dict[str, float] | None = None,
|
||
prev: PeriodMetrics | None = None,
|
||
) -> dict:
|
||
"""单员工试算评分(纯函数,B4 端点直接透出)。
|
||
|
||
prev/baseline/team 为四对比位槽位:prev/baseline 按同口径派生后对比,
|
||
team 为指标级中位数(键=派生指标名,可 None)。返回结构见模块 docstring
|
||
——metrics_detail 键=配置里的 metric_code(band 型如 active_friends_band)。
|
||
"""
|
||
cur_vals = _derive_metrics(current)
|
||
prev_vals = _derive_metrics(prev) if prev is not None else None
|
||
base_vals = _derive_metrics(baseline) if baseline is not None else None
|
||
guards = config.get("sample_guards", {})
|
||
soft = guards.get("soft_threads", 10)
|
||
minimum = guards.get("min_threads", 30)
|
||
|
||
dimensions: dict[str, dict] = {}
|
||
metrics_detail: dict[str, dict] = {}
|
||
|
||
for dim_key, dim in config.get("dimensions", {}).items():
|
||
entry: dict = {
|
||
"name": dim.get("name", dim_key),
|
||
"weight": dim.get("weight", 0),
|
||
"score": None,
|
||
"status": "pending",
|
||
}
|
||
if dim.get("status") == "pending_data" or not dim.get("metrics"):
|
||
# 一期 action/risk:数据未接入——不参与归一,不带 metrics
|
||
dimensions[dim_key] = entry
|
||
continue
|
||
|
||
dim_scores: list[float] = []
|
||
any_insufficient = False # 有值但样本不足→维度 insufficient
|
||
dim_metrics: dict[str, dict] = {}
|
||
for code, spec in dim["metrics"].items():
|
||
src = spec.get("metric", code)
|
||
value = cur_vals.get(src)
|
||
# 样本量:串类=cust_threads,band 类=active_friends
|
||
sample_size = (
|
||
current.active_friends
|
||
if spec.get("type") == "band" else current.cust_threads
|
||
)
|
||
if sample_size >= minimum:
|
||
sample_status = "ok"
|
||
elif sample_size >= soft:
|
||
sample_status = "low_confidence"
|
||
else:
|
||
sample_status = "insufficient"
|
||
score: float | None = None
|
||
if value is not None:
|
||
if sample_status == "insufficient":
|
||
any_insufficient = True # 有可评值但不评分
|
||
else:
|
||
score = normalize(spec, value)
|
||
if score is not None:
|
||
dim_scores.append(score)
|
||
dim_metrics[code] = {
|
||
"value": value,
|
||
"score": score,
|
||
"sample_size": sample_size,
|
||
"sample_status": sample_status,
|
||
"formula_note": _formula_note(spec),
|
||
}
|
||
metrics_detail[code] = {
|
||
"value": value,
|
||
"prev": prev_vals.get(src) if prev_vals else None,
|
||
"baseline": base_vals.get(src) if base_vals else None,
|
||
"median": team.get(src) if team else None,
|
||
"sample_size": sample_size,
|
||
"sample_status": sample_status,
|
||
"score": score,
|
||
"unit": METRIC_DEFINITIONS.get(src, {}).get("unit", ""),
|
||
"name": METRIC_DEFINITIONS.get(src, {}).get("name", code),
|
||
}
|
||
|
||
if dim_scores:
|
||
entry["score"] = sum(dim_scores) / len(dim_scores)
|
||
entry["status"] = "scored"
|
||
elif any_insufficient:
|
||
entry["status"] = "insufficient"
|
||
else:
|
||
entry["status"] = "pending" # 全部无值(期末无存量/无串样本等)
|
||
entry["metrics"] = dim_metrics
|
||
dimensions[dim_key] = entry
|
||
|
||
# 可用维度加权重归一:Σ(w×s)/Σw(满分=Σw×100/Σw=100),四舍五入取整
|
||
available = [
|
||
(d["weight"], d["score"]) for d in dimensions.values()
|
||
if d["score"] is not None
|
||
]
|
||
total: int | None = None
|
||
if available:
|
||
wsum = sum(w for w, _ in available)
|
||
total = int(math.floor(
|
||
sum(w * s for w, s in available) / wsum + 0.5
|
||
))
|
||
grade: str | None = None
|
||
if total is not None:
|
||
grade = "E"
|
||
bounds = config.get("grade_bounds", {})
|
||
for g in _GRADE_ORDER: # A..D 阈值递减,首中即停
|
||
if total >= bounds[g]:
|
||
grade = g
|
||
break
|
||
|
||
return {
|
||
"total_score": total,
|
||
"grade": grade,
|
||
"score_status": "scored" if total is not None else "insufficient_sample",
|
||
"trial": bool(config.get("trial")),
|
||
"dimensions": dimensions,
|
||
"metrics_detail": metrics_detail,
|
||
}
|