Files
cma-management/backend/app/api/thresholds.py
T
Hermes CI Fix 3dddd36866 init: 管理会计OS初始代码
包含前后端完整代码:
- 前端:Vue3+Vite+ElementPlus
- 后端:FastAPI+SQLAlchemy
- 模块:驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动
- 当前版本:v1.0.0
2026-05-28 17:32:22 +08:00

98 lines
4.4 KiB
Python

"""阈值智能推荐 API"""
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
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"],
}
@router.get("/suggest/{kpi_id}")
def suggest_threshold(kpi_id: int, db: Session = Depends(get_db)):
"""根据历史数据自动推荐阈值"""
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)
avg = sum(nums) / len(nums)
# 计算标准差
variance = sum((x - avg) ** 2 for x in nums) / len(nums)
std = variance ** 0.5
target = kpi.target_value or avg
# 根据KPI类型生成推荐区间
if kpi.kpi_code in KPI_TYPES["higher_better"]:
green_min = round(target * 0.8, 2)
yellow_min = round(target * 0.5, 2)
red_max = round(target * 0.5, 2)
suggestion = {
"type": "higher_better",
"description": "越高越好型",
"green": {"min": green_min, "max": None, "label": f">={green_min}"},
"yellow": {"min": yellow_min, "max": green_min, "label": f"{yellow_min}~{green_min}"},
"red": {"min": None, "max": red_max, "label": f"<{red_max}"},
"current_avg": round(avg, 2),
"target": target,
}
elif kpi.kpi_code in KPI_TYPES["lower_better"]:
green_max = round(target * 1.2, 2)
yellow_max = round(target * 2.0, 2)
suggestion = {
"type": "lower_better",
"description": "越低越好型",
"green": {"min": None, "max": green_max, "label": f"<={green_max}"},
"yellow": {"min": green_max, "max": yellow_max, "label": f"{green_max}~{yellow_max}"},
"red": {"min": yellow_max, "max": None, "label": f">{yellow_max}"},
"current_avg": round(avg, 2),
"target": target,
}
else:
tolerance = max(std * 1.5, target * 0.2)
suggestion = {
"type": "middle_best",
"description": "适中最好型",
"green": {"min": round(target - tolerance, 2), "max": round(target + tolerance, 2), "label": f"{round(target-tolerance,2)}~{round(target+tolerance,2)}"},
"yellow": {"min": round(target - tolerance*2, 2), "max": round(target + tolerance*2, 2), "label": f"偏离{(tolerance*2):.0f}%"},
"red": {"min": None, "max": round(target - tolerance*2, 2), "label": f"偏离>{tolerance*2:.0f}%"},
"current_avg": round(avg, 2),
"target": target,
}
return {"kpi_id": kpi_id, "kpi_name": kpi.kpi_name, "suggestion": suggestion}
def suggest_by_type(kpi):
"""无历史数据时按类型推荐"""
target = kpi.target_value or 100
if kpi.kpi_code in KPI_TYPES["higher_better"]:
return {"kpi_id": kpi.id, "kpi_name": kpi.kpi_name, "message": "无历史数据",
"suggestion": {"type": "higher_better", "green": {"min": round(target*0.8,2)}, "yellow": {"min": round(target*0.5,2)}, "red": {"max": round(target*0.5,2)}}}
elif kpi.kpi_code in KPI_TYPES["lower_better"]:
return {"kpi_id": kpi.id, "kpi_name": kpi.kpi_name, "message": "无历史数据",
"suggestion": {"type": "lower_better", "green": {"max": round(target*1.2,2)}, "yellow": {"max": round(target*2,2)}, "red": {"min": round(target*2,2)}}}
else:
return {"kpi_id": kpi.id, "kpi_name": kpi.kpi_name, "message": "无历史数据",
"suggestion": {"type": "middle_best", "green": {"min": round(target*0.8,2), "max": round(target*1.2,2)}}}