feat: KPI多粒度目标智能派生 — 基准值+按类型派生+手动覆盖

- 后端: target_calc_type字段(accumulate累计/ratio比率) + infer_calc_type名称/单位推断 - 派生规则: 累计型月×3=季×12=年(季×4=年); 比率型季/年沿用基准不可乘 - 虚拟派生不落库: kpi_to_dict返回derived_targets+derived_flags(自动标记) - DB: 302个KPI回填类型(226累计/76比率) - 前端KPIList: 指标类型选择 + 季/年自动派生预览(↳自动=N) - 前端KPIDetail: 元数据卡片自动标记 + 编辑表单指标类型 - 回归: pytest 451 passed
2026-08-21 10:59:43 +08:00
parent d5edcefb48
commit b06821a426
4 changed files with 214 additions and 5 deletions
@@ -488,6 +488,103 @@ def _validate_kpi_data(data: dict, db: Session, current_kpi_id: Optional[int] =
    return kpi_issues_message(issues)


+# ════════════════════════════════════════════════════════════
+# KPI多粒度目标：指标类型推断 + 周期目标派生（docs/kpi-design-rule.md 落地）
+# 规则：累计型 月×3=季、月×12=年(季×4=年)；比率型 季/年沿用基准(可手调)
+# 派生为"虚拟展示值"：DB只存用户手填真值，API返回时补派生值+derived标记
+# ════════════════════════════════════════════════════════════
+RATIO_NAME_HINTS = ['率', '比', '满意度', '周转', '时长', '周期', '天数', '指数', 'NPS', 'LTV', 'CAC',
+                    '份额', '集中度', '响应', '完成', '达成', '人均', '单价', '净推荐', '覆盖', '保留',
+                    '复购', '转介绍', '投诉', '合规', '认证', '掌握', '胜任', '认知', '采纳', '引用',
+                    '复用', '一致性', '准确', '间隙', '时效', '及时']
+ACCUM_NAME_HINTS = ['营收', '收入', '利润', '净利', '销售', '客户数', '新客', '新增', '产量', '销量',
+                    '金额', '现金流', '回款', '毛利额', '产值', '储备', '数量', '篇数', '报告产出',
+                    '提案', '发现数', '知识沉淀', '招待费']
+RATIO_UNIT_HINTS = ['%', '倍', '天', '分', '小时', '分钟']
+ACCUM_UNIT_HINTS = ['万元', '元', '个', '件', '人', '篇', '份', '万']
+
+
+def infer_calc_type(kpi_code: str = "", kpi_name: str = "", unit: str = "") -> str:
+    """推断指标类型: accumulate累计(可乘) / ratio比率(不可乘)。名称关键词优先于单位"""
+    n = (kpi_name or "") + " " + (kpi_code or "")
+    u = unit or ""
+    if any(k in n for k in RATIO_NAME_HINTS):
+        return "ratio"
+    if any(k in n for k in ACCUM_NAME_HINTS):
+        return "accumulate"
+    if u in RATIO_UNIT_HINTS or u.startswith("小时"):
+        return "ratio"
+    if u in ACCUM_UNIT_HINTS:
+        return "accumulate"
+    return "ratio"  # 兜底比率（率值不能乘，更安全）
+
+
+def derive_cycle_targets(kpi) -> dict:
+    """按指标类型派生月/季/年目标（虚拟值，不落库）。
+    返回: {"derived": {monthly/quarterly/yearly: 显示值}, "flags": {monthly/quarterly/yearly: 是否派生}}
+    """
+    calc_type = (getattr(kpi, "target_calc_type", None) or infer_calc_type(
+        kpi.kpi_code or "", kpi.kpi_name or "", kpi.unit or "")).lower()
+    m = kpi.target_monthly
+    q = kpi.target_quarterly
+    y = kpi.target_yearly
+    freq = (kpi.frequency or "monthly").lower()
+
+    # 基准值（考核周期优先，回退 target_value）
+    base = None
+    if freq == "yearly":
+        base = y
+    elif freq in ("quarterly", "half_year"):
+        base = q
+    elif freq in ("monthly", "weekly"):
+        base = m
+    if base is None:
+        base = kpi.target_value
+    # 无基准值则不派生
+    if base is None:
+        return {"derived": {"monthly": m, "quarterly": q, "yearly": y},
+                "flags": {"monthly": False, "quarterly": False, "yearly": False}}
+
+    dm, dq, dy = m, q, y
+    fm, fq, fy = False, False, False
+    if calc_type == "accumulate":
+        # 锚点月值：手填月目标优先；月基准且手填月空时用 target_value 回退
+        anchor_m = dm
+        if anchor_m is None and base is not None and freq in ("monthly", "weekly"):
+            anchor_m = base
+        if anchor_m is not None:
+            if dm is None:
+                dm = anchor_m  # target_value 回退显示为月基准
+            if dq is None:
+                dq, fq = anchor_m * 3, True
+            if dy is None:
+                dy, fy = anchor_m * 12, True
+        elif dq is not None:
+            # 季基准（累计型）：年=季×4；月不反推（避免小数噪声）
+            if dy is None:
+                dy, fy = dq * 4, True
+    else:  # ratio：季/年沿用基准，不乘
+        if dq is None:
+            dq, fq = base, True
+        if dy is None:
+            dy, fy = base, True
+    return {"derived": {"monthly": dm, "quarterly": dq, "yearly": dy},
+            "flags": {"monthly": fm, "quarterly": fq, "yearly": fy}}
+
+
+def apply_calc_type_inference(data: dict, infer_missing: bool = True) -> dict:
+    """create/update 前：未显式传 target_calc_type 时按名称/单位推断。
+    infer_missing=False（update场景）：仅当用户显式传了空值时推断，未传则保留DB原值"""
+    if "target_calc_type" in data:
+        if not data.get("target_calc_type"):
+            data["target_calc_type"] = infer_calc_type(
+                data.get("kpi_code", ""), data.get("kpi_name", ""), data.get("unit", ""))
+    elif infer_missing:
+        data["target_calc_type"] = infer_calc_type(
+            data.get("kpi_code", ""), data.get("kpi_name", ""), data.get("unit", ""))
+    return data
+
+
@router.post("")
 def create_kpi(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
    # 检查编码唯一性
@@ -498,6 +595,7 @@ def create_kpi(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
    errs = _validate_kpi_data(data, db=db, is_update=False)
    if errs:
        raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs})
+    data = apply_calc_type_inference(data)
    kpi = KPIDefinition(**data)
    db.add(kpi)
    db.commit()
@@ -515,6 +613,7 @@ def update_kpi(kpi_id: int, data: dict, db: Session = Depends(get_db), user=WRIT
    errs = _validate_kpi_data(data, db=db, current_kpi_id=kpi_id, is_update=True)
    if errs:
        raise HTTPException(422, detail={"message": "数据校验不通过", "errors": errs})
+    data = apply_calc_type_inference(data, infer_missing=False)
    for k, v in data.items():
        if hasattr(kpi, k) and v is not None:
            setattr(kpi, k, v)
@@ -543,6 +642,14 @@ def restore_kpi(kpi_id: int, db: Session = Depends(get_db), user=WRITE_ROLES):

 def kpi_to_dict(k):
    d = {c.name: getattr(k, c.name) for c in k.__table__.columns}
+    # 多粒度目标派生：月/季/年显示值 + derived标记（虚拟，不落库）
+    try:
+        der = derive_cycle_targets(k)
+        d["derived_targets"] = der["derived"]
+        d["derived_flags"] = der["flags"]
+    except Exception:
+        d["derived_targets"] = {"monthly": k.target_monthly, "quarterly": k.target_quarterly, "yearly": k.target_yearly}
+        d["derived_flags"] = {"monthly": False, "quarterly": False, "yearly": False}
    # 附加战略地图信息
    if k.map_id:
        from app.database import get_session_local