164 lines
6.9 KiB
Python
164 lines
6.9 KiB
Python
"""阈值智能推荐 API
|
||
|
||
2026-09-11 (project-20260911-threshold-fix) 修复：
|
||
推荐值必须与后端解析器（app.utils.threshold_parser）语法一致 —— 原实现给黄灯/适中型
|
||
生成区间/文字标签（`52.33~107.67`、`偏离55%`），用户点“自动建议”填入后保存即静默失效。
|
||
现在统一生成可解析表达式：标量比较（支持负数）/ 闭区间 `A~B` / 多条件 `C1|C2`。
|
||
"""
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy.orm import Session
|
||
from app.database import get_db
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.models import KPIDefinition, KPIValue, OperationLog
|
||
from app.utils.threshold_parser import THRESHOLD_HELP, is_valid_threshold
|
||
from typing import Any, Optional
|
||
import json
|
||
|
||
router = APIRouter(prefix="/api/cma/thresholds", tags=["阈值分析"],
|
||
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
|
||
)
|
||
|
||
KPI_TYPES = {
|
||
"higher_better": ["SALES_TOTAL", "CUSTOMER_COUNT", "SALES_PROFIT_RATE",
|
||
"CUSTOMER_SATISFACTION", "ORDER_DELIVERY_RATE", "TRAINING_COMPLETION",
|
||
"RECEIVABLE_TURNOVER", "TURNOVER_RATE"],
|
||
"lower_better": ["COST_CONTROL_RATE"],
|
||
"middle_best": ["TOP5_CUSTOMER_RATIO"],
|
||
}
|
||
|
||
|
||
def _r(v) -> float:
|
||
"""保留2位小数（避免浮点噪声写进表达式）"""
|
||
return round(float(v), 2)
|
||
|
||
|
||
def _fmt(v) -> str:
|
||
"""数值 → 表达式文本。禁用科学计数法：`:g` 在 |v|>=1e6 时会输出 1.6e+06，
|
||
解析器不认（会导致推荐值再次静默失效）。"""
|
||
s = f"{float(v):.2f}".rstrip("0").rstrip(".")
|
||
return s if s not in ("", "-") else "0"
|
||
|
||
|
||
def build_suggestion(kpi_type: str, target: float, tolerance: Optional[float] = None) -> dict:
|
||
"""按类型生成可解析的红黄绿阈值建议（仅生成表达式，不含 I/O）。
|
||
|
||
口径（label = 该等级的触发条件，判定优先级 red→green→yellow）：
|
||
higher_better：绿 `>=0.8T`、黄 `>=0.5T`（更宽的兜底带）、红 `<0.5T`
|
||
lower_better ：绿 `<=1.2T`、黄 `<=2.0T`、红 `>2.0T`
|
||
middle_best ：绿 `T±tol` 闭区间、黄 `T±2tol` 外环带（green ⊂ yellow）、红 `<T-2tol|>T+2tol`
|
||
"""
|
||
if kpi_type == "higher_better":
|
||
green_min = _r(target * 0.8)
|
||
yellow_min = _r(target * 0.5)
|
||
red_max = yellow_min
|
||
return {
|
||
"type": "higher_better",
|
||
"description": "越高越好型",
|
||
"green": {"min": green_min, "max": None, "label": f">={_fmt(green_min)}"},
|
||
"yellow": {"min": yellow_min, "max": green_min, "label": f">={_fmt(yellow_min)}"},
|
||
"red": {"min": None, "max": red_max, "label": f"<{_fmt(red_max)}"},
|
||
}
|
||
if kpi_type == "lower_better":
|
||
green_max = _r(target * 1.2)
|
||
yellow_max = _r(target * 2.0)
|
||
red_min = yellow_max
|
||
return {
|
||
"type": "lower_better",
|
||
"description": "越低越好型",
|
||
"green": {"min": None, "max": green_max, "label": f"<={_fmt(green_max)}"},
|
||
"yellow": {"min": green_max, "max": yellow_max, "label": f"<={_fmt(yellow_max)}"},
|
||
"red": {"min": red_min, "max": None, "label": f">{_fmt(red_min)}"},
|
||
}
|
||
# middle_best（适中最好型）：绿=内区间 T±tol，黄=外环带 T±2tol（green ⊂ yellow），红=环带之外
|
||
tol = float(tolerance) if tolerance else abs(float(target)) * 0.2
|
||
if tol <= 0:
|
||
tol = 1.0
|
||
a, b = _r(target - tol), _r(target + tol)
|
||
c, d = _r(target - 2 * tol), _r(target + 2 * tol)
|
||
return {
|
||
"type": "middle_best",
|
||
"description": "适中最好型",
|
||
"green": {"min": a, "max": b, "label": f"{_fmt(a)}~{_fmt(b)}"},
|
||
"yellow": {"min": c, "max": d, "label": f"{_fmt(c)}~{_fmt(d)}"},
|
||
"red": {"min": None, "max": None, "label": f"<{_fmt(c)}|>{_fmt(d)}"},
|
||
}
|
||
|
||
|
||
def suggest_by_type(kpi):
|
||
"""无历史数据时按类型推荐（同样输出可解析表达式，含 yellow/red 标签）"""
|
||
target = kpi.target_value or 100
|
||
kpi_type = _kpi_type(kpi)
|
||
suggestion = build_suggestion(kpi_type, target)
|
||
return {"kpi_id": kpi.id, "kpi_name": kpi.kpi_name, "message": "无历史数据",
|
||
"format_hint": THRESHOLD_HELP, "suggestion": suggestion}
|
||
|
||
|
||
def _kpi_type(kpi) -> str:
|
||
"""KPI → 推荐类型（**保持历史默认语义**）。
|
||
|
||
原实现为 if higher_better / elif lower_better / **else → middle_best**，
|
||
即「不在任何清单里的编码」默认走适中型（如 C_REBATE_RATE 渠补率、
|
||
BH_SALES_10 销售-任富海，其历史生成的 '52.33~107.67' / '偏离…%' 即出自该分支）。
|
||
此处默认值必须同为 middle_best，否则默认分支的推荐口径会被改变。
|
||
"""
|
||
code = getattr(kpi, "kpi_code", None)
|
||
if code in KPI_TYPES["higher_better"]:
|
||
return "higher_better"
|
||
if code in KPI_TYPES["lower_better"]:
|
||
return "lower_better"
|
||
return "middle_best"
|
||
|
||
|
||
def _sanity_check(suggestion: dict) -> dict:
|
||
"""自检：推荐值必须过解析器（防再次生成不可解析格式），并挂 format 提示给前端。"""
|
||
bad = []
|
||
for level in ("green", "yellow", "red"):
|
||
label = (suggestion.get(level) or {}).get("label")
|
||
if label and not is_valid_threshold(label):
|
||
bad.append({"level": level, "label": label})
|
||
suggestion["parseable"] = not bad
|
||
if bad:
|
||
suggestion["unparseable"] = bad
|
||
return suggestion
|
||
|
||
|
||
@router.get("/suggest/{kpi_id}")
|
||
def suggest_threshold(kpi_id: int, db: Session = Depends(get_db)):
|
||
"""根据历史数据自动推荐阈值（输出格式保证被 threshold_parser 解析生效）"""
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
if not kpi:
|
||
return {"error": "KPI不存在"}
|
||
|
||
values = db.query(KPIValue).filter(KPIValue.kpi_id == kpi_id).order_by(KPIValue.period.asc()).all()
|
||
|
||
if not values:
|
||
# 无历史数据，按行业标准推荐
|
||
return suggest_by_type(kpi)
|
||
|
||
nums = [v.actual_value for v in values if v.actual_value is not None]
|
||
|
||
if len(nums) < 2:
|
||
return suggest_by_type(kpi)
|
||
|
||
# 注: SQLAlchemy 模型属性未做 Mapped 注解, Pyright 会推断为 Column;
|
||
# 这里统一标注 Any 抑制静态噪音(运行期均为 float)。
|
||
avg: Any = sum(nums) / len(nums)
|
||
# 计算标准差
|
||
variance: Any = sum((x - avg) ** 2 for x in nums) / len(nums)
|
||
std: Any = variance ** 0.5
|
||
|
||
target: Any = kpi.target_value or avg
|
||
kpi_type = _kpi_type(kpi)
|
||
|
||
if kpi_type == "middle_best":
|
||
tolerance: Any = max(std * 1.5, target * 0.2)
|
||
suggestion = build_suggestion(kpi_type, target, tolerance)
|
||
else:
|
||
suggestion = build_suggestion(kpi_type, target)
|
||
suggestion = _sanity_check(suggestion)
|
||
suggestion["current_avg"] = round(avg, 2)
|
||
suggestion["target"] = target
|
||
|
||
return {"kpi_id": kpi_id, "kpi_name": kpi.kpi_name,
|
||
"format_hint": THRESHOLD_HELP, "suggestion": suggestion}
|