480 lines
18 KiB
Python
480 lines
18 KiB
Python
""""
|
||
预警规则智能化 — 任务6
|
||
后端组件: alert_rules 模型 + API + 预警引擎
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import text, Column, Integer, String, Text, Float, DateTime, JSON, Boolean, func
|
||
from typing import Optional, List
|
||
from datetime import datetime, timedelta
|
||
import json
|
||
import logging
|
||
|
||
from app.database import get_db, Base
|
||
from app.auth_middleware import require_auth, require_role
|
||
from app.models import KPIDefinition, KPIValue, KPIAlert, OperationLog
|
||
|
||
logger = logging.getLogger("alert_rules")
|
||
|
||
# ============================================================
|
||
# AlertRule 模型
|
||
# ============================================================
|
||
class AlertRule(Base):
|
||
"""预警规则配置"""
|
||
__tablename__ = "alert_rules"
|
||
id = Column(Integer, primary_key=True, index=True)
|
||
kpi_id = Column(Integer, nullable=False, comment="关联KPI ID")
|
||
rule_type = Column(String(30), nullable=False, comment="static/dynamic/trend_up/trend_down")
|
||
enabled = Column(Integer, default=1, comment="1启用 0禁用")
|
||
params = Column(JSON, nullable=True, comment="规则参数")
|
||
# static: {"green": ">=90", "yellow": ">=80", "red": "<80"}
|
||
# dynamic: {"sensitivity": 1.0} — 阈值 = mean ± sensitivity * stddev, period_months=3
|
||
# trend_up: {"threshold_pct": 10} — 环比上升超过 threshold_pct% 触发
|
||
# trend_down: {"threshold_pct": 10} — 环比下降超过 threshold_pct% 触发
|
||
created_at = Column(DateTime, server_default=func.now())
|
||
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
|
||
|
||
class DynamicThresholdCache(Base):
|
||
"""动态阈值缓存 — 存储近3个月历史统计"""
|
||
__tablename__ = "dynamic_threshold_cache"
|
||
id = Column(Integer, primary_key=True, index=True)
|
||
kpi_id = Column(Integer, nullable=False, comment="关联KPI ID")
|
||
period = Column(String(20), nullable=False, comment="计算期间 2026-07")
|
||
mean_value = Column(Float, nullable=True, comment="近3月均值")
|
||
stddev_value = Column(Float, nullable=True, comment="近3月标准差")
|
||
dynamic_green = Column(String(100), nullable=True, comment="动态绿灯阈值")
|
||
dynamic_yellow = Column(String(100), nullable=True, comment="动态黄灯阈值")
|
||
dynamic_red = Column(String(100), nullable=True, comment="动态红灯阈值")
|
||
calculated_at = Column(DateTime, server_default=func.now())
|
||
|
||
|
||
router = APIRouter(prefix="/api/cma/alert-rules", tags=["预警规则"],
|
||
dependencies=[Depends(require_role("ceo", "finance", "it"))],
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# API Endpoints
|
||
# ============================================================
|
||
|
||
@router.get("")
|
||
def list_alert_rules(
|
||
kpi_id: Optional[int] = None,
|
||
rule_type: Optional[str] = None,
|
||
enabled: Optional[int] = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""列出所有预警规则"""
|
||
query = db.query(AlertRule)
|
||
if kpi_id:
|
||
query = query.filter(AlertRule.kpi_id == kpi_id)
|
||
if rule_type:
|
||
query = query.filter(AlertRule.rule_type == rule_type)
|
||
if enabled is not None:
|
||
query = query.filter(AlertRule.enabled == enabled)
|
||
|
||
rules = query.order_by(AlertRule.id).all()
|
||
result = []
|
||
for r in rules:
|
||
d = {c.name: getattr(r, c.name) for c in AlertRule.__table__.columns}
|
||
# 关联KPI信息
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == r.kpi_id).first()
|
||
if kpi:
|
||
d["kpi_code"] = kpi.kpi_code
|
||
d["kpi_name"] = kpi.kpi_name
|
||
result.append(d)
|
||
return {"data": result, "total": len(result)}
|
||
|
||
|
||
@router.get("/kpi/{kpi_id}")
|
||
def get_kpi_rules(kpi_id: int, db: Session = Depends(get_db)):
|
||
"""获取单个KPI的所有预警规则"""
|
||
rules = db.query(AlertRule).filter(AlertRule.kpi_id == kpi_id).order_by(AlertRule.id).all()
|
||
return {"data": [{c.name: getattr(r, c.name) for c in AlertRule.__table__.columns} for r in rules]}
|
||
|
||
|
||
@router.post("")
|
||
def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(require_role("ceo", "finance", "it"))):
|
||
"""创建预警规则"""
|
||
kpi_id = data.get("kpi_id")
|
||
rule_type = data.get("rule_type", "static")
|
||
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
if not kpi:
|
||
raise HTTPException(404, "KPI不存在")
|
||
if rule_type not in ("static", "dynamic", "trend_up", "trend_down"):
|
||
raise HTTPException(400, f"不支持的规则类型: {rule_type}")
|
||
|
||
rule = AlertRule(
|
||
kpi_id=kpi_id,
|
||
rule_type=rule_type,
|
||
enabled=data.get("enabled", 1),
|
||
params=data.get("params"),
|
||
)
|
||
db.add(rule)
|
||
db.commit()
|
||
db.refresh(rule)
|
||
|
||
# 日志
|
||
db.add(OperationLog(
|
||
action="create_alert_rule", target_type="alert_rule",
|
||
detail=f"KPI={kpi.kpi_code}({kpi.kpi_name}) type={rule_type}",
|
||
))
|
||
db.commit()
|
||
|
||
return {"data": {c.name: getattr(rule, c.name) for c in AlertRule.__table__.columns}}
|
||
|
||
|
||
@router.put("/{rule_id}")
|
||
def update_alert_rule(rule_id: int, data: dict, db: Session = Depends(get_db)):
|
||
"""更新预警规则"""
|
||
rule = db.query(AlertRule).filter(AlertRule.id == rule_id).first()
|
||
if not rule:
|
||
raise HTTPException(404, "预警规则不存在")
|
||
|
||
for field in ("rule_type", "enabled", "params"):
|
||
if field in data:
|
||
setattr(rule, field, data[field])
|
||
db.commit()
|
||
db.refresh(rule)
|
||
return {"data": {c.name: getattr(rule, c.name) for c in AlertRule.__table__.columns}}
|
||
|
||
|
||
@router.delete("/{rule_id}")
|
||
def delete_alert_rule(rule_id: int, db: Session = Depends(get_db)):
|
||
"""删除预警规则"""
|
||
rule = db.query(AlertRule).filter(AlertRule.id == rule_id).first()
|
||
if rule:
|
||
db.delete(rule)
|
||
db.commit()
|
||
return {"message": "已删除"}
|
||
|
||
|
||
@router.post("/batch")
|
||
def batch_create_rules(data: dict, db: Session = Depends(get_db)):
|
||
"""批量创建预警规则
|
||
data.rules: [{"kpi_id": id, "rule_type": "static", "params": {...}}, ...]
|
||
"""
|
||
rules_data = data.get("rules", [])
|
||
created = 0
|
||
for rule_data in rules_data:
|
||
kpi_id = rule_data.get("kpi_id")
|
||
rule_type = rule_data.get("rule_type", "static")
|
||
# 检查是否已存在相同类型的规则
|
||
existing = db.query(AlertRule).filter(
|
||
AlertRule.kpi_id == kpi_id,
|
||
AlertRule.rule_type == rule_type,
|
||
).first()
|
||
if existing:
|
||
continue
|
||
rule = AlertRule(
|
||
kpi_id=kpi_id,
|
||
rule_type=rule_type,
|
||
enabled=rule_data.get("enabled", 1),
|
||
params=rule_data.get("params"),
|
||
)
|
||
db.add(rule)
|
||
created += 1
|
||
db.commit()
|
||
return {"message": f"批量创建完成: 新增{created}条", "created": created}
|
||
|
||
|
||
@router.post("/generate-defaults")
|
||
def generate_default_rules(db: Session = Depends(get_db)):
|
||
"""为所有尚未配置预警规则的KPI生成默认规则"""
|
||
# 找到所有active KPI
|
||
all_kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
||
|
||
created = 0
|
||
for kpi in all_kpis:
|
||
# 检查是否已有任何规则
|
||
existing = db.query(AlertRule).filter(AlertRule.kpi_id == kpi.id).first()
|
||
if existing:
|
||
continue
|
||
|
||
kpi_id = kpi.id
|
||
|
||
# 1. 静态阈值规则（基于kpi_definitions的阈值）
|
||
if kpi.threshold_green or kpi.threshold_yellow or kpi.threshold_red:
|
||
rule = AlertRule(
|
||
kpi_id=kpi_id,
|
||
rule_type="static",
|
||
enabled=1,
|
||
params={
|
||
"green": kpi.threshold_green,
|
||
"yellow": kpi.threshold_yellow,
|
||
"red": kpi.threshold_red,
|
||
}
|
||
)
|
||
db.add(rule)
|
||
created += 1
|
||
|
||
# 2. 动态趋势规则（所有KPI默认加 trend_down）
|
||
rule2 = AlertRule(
|
||
kpi_id=kpi_id,
|
||
rule_type="trend_down",
|
||
enabled=1,
|
||
params={"threshold_pct": 10},
|
||
)
|
||
db.add(rule2)
|
||
created += 1
|
||
|
||
db.commit()
|
||
return {"message": f"默认规则生成完成: 共{created}条", "created": created}
|
||
|
||
|
||
@router.post("/check-all")
|
||
def run_all_alert_checks(db: Session = Depends(get_db)):
|
||
"""执行所有KPI的预警检查 — 生成新的预警记录"""
|
||
rules = db.query(AlertRule).filter(AlertRule.enabled == 1).all()
|
||
kpi_cache = {}
|
||
value_cache = {}
|
||
|
||
alerts_generated = 0
|
||
|
||
for rule in rules:
|
||
try:
|
||
kpi = kpi_cache.get(rule.kpi_id)
|
||
if kpi is None:
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == rule.kpi_id).first()
|
||
if kpi:
|
||
kpi_cache[rule.kpi_id] = kpi
|
||
|
||
if not kpi:
|
||
continue
|
||
|
||
# 获取最新值
|
||
latest_value = value_cache.get(rule.kpi_id)
|
||
if latest_value is None:
|
||
latest_value = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == rule.kpi_id,
|
||
).order_by(KPIValue.period.desc()).first()
|
||
if latest_value:
|
||
value_cache[rule.kpi_id] = latest_value
|
||
|
||
if not latest_value or latest_value.actual_value is None:
|
||
continue
|
||
|
||
value = latest_value.actual_value
|
||
period = latest_value.period
|
||
params = rule.params or {}
|
||
|
||
alert_level = None
|
||
alert_message = None
|
||
|
||
if rule.rule_type == "static":
|
||
alert_level, alert_message = _check_static(value, params, kpi)
|
||
elif rule.rule_type == "dynamic":
|
||
alert_level, alert_message = _check_dynamic(kpi.id, value, params, db)
|
||
elif rule.rule_type == "trend_up":
|
||
alert_level, alert_message = _check_trend(kpi.id, value, "up", params, db)
|
||
elif rule.rule_type == "trend_down":
|
||
alert_level, alert_message = _check_trend(kpi.id, value, "down", params, db)
|
||
|
||
if alert_level and alert_level != "green":
|
||
# 检查是否已有相同预警
|
||
existing_alert = db.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == rule.kpi_id,
|
||
KPIAlert.kpi_value_id == latest_value.id,
|
||
KPIAlert.alert_level == alert_level,
|
||
KPIAlert.alert_message == alert_message,
|
||
KPIAlert.status == "pending",
|
||
).first()
|
||
if not existing_alert:
|
||
alert = KPIAlert(
|
||
kpi_id=rule.kpi_id,
|
||
kpi_value_id=latest_value.id,
|
||
alert_level=alert_level,
|
||
alert_message=alert_message,
|
||
status="pending",
|
||
)
|
||
db.add(alert)
|
||
alerts_generated += 1
|
||
|
||
except Exception as e:
|
||
logger.error(f"预警检查失败: rule_id={rule.id}, error={e}")
|
||
continue
|
||
|
||
db.commit()
|
||
return {"message": f"预警检查完成: 生成{alerts_generated}条", "generated": alerts_generated}
|
||
|
||
|
||
@router.get("/dynamic-thresholds")
|
||
def get_dynamic_thresholds(kpi_id: Optional[int] = None, db: Session = Depends(get_db)):
|
||
"""获取动态阈值缓存"""
|
||
query = db.query(DynamicThresholdCache)
|
||
if kpi_id:
|
||
query = query.filter(DynamicThresholdCache.kpi_id == kpi_id)
|
||
cache = query.order_by(DynamicThresholdCache.id.desc()).limit(50).all()
|
||
return {"data": [{c.name: getattr(c, c.name) for c in DynamicThresholdCache.__table__.columns} for c in cache]}
|
||
|
||
|
||
@router.post("/calculate-dynamic")
|
||
def calculate_dynamic_thresholds(db: Session = Depends(get_db)):
|
||
"""计算所有KPI的动态阈值（基于近3个月历史均值±标准差）"""
|
||
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
|
||
current_period = datetime.now().strftime("%Y-%m")
|
||
|
||
computed = 0
|
||
for kpi in kpis:
|
||
# 取近3个月的历史值（不含当月）
|
||
from sqlalchemy import text as sa_text, func as sa_func
|
||
values = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi.id,
|
||
KPIValue.actual_value.isnot(None),
|
||
KPIValue.data_status.in_(["verified", "estimated"]),
|
||
KPIValue.period < current_period,
|
||
).order_by(KPIValue.period.desc()).limit(3).all()
|
||
|
||
if len(values) < 2:
|
||
continue
|
||
|
||
vals = [v.actual_value for v in values if v.actual_value is not None]
|
||
if len(vals) < 2:
|
||
continue
|
||
|
||
mean_val = sum(vals) / len(vals)
|
||
if len(vals) > 1:
|
||
variance = sum((v - mean_val) ** 2 for v in vals) / len(vals)
|
||
stddev = variance ** 0.5
|
||
else:
|
||
stddev = mean_val * 0.1 # 仅1个值时的合理估算
|
||
|
||
# 生成动态阈值（±1标准差）
|
||
dynamic_green = f">={mean_val + stddev:.2f}"
|
||
dynamic_yellow = f">={mean_val:.2f}"
|
||
dynamic_red = f"<{mean_val:.2f}"
|
||
|
||
# 检查是否已有缓存
|
||
existing = db.query(DynamicThresholdCache).filter(
|
||
DynamicThresholdCache.kpi_id == kpi.id,
|
||
DynamicThresholdCache.period == current_period,
|
||
).first()
|
||
|
||
if existing:
|
||
existing.mean_value = mean_val
|
||
existing.stddev_value = stddev
|
||
existing.dynamic_green = dynamic_green
|
||
existing.dynamic_yellow = dynamic_yellow
|
||
existing.dynamic_red = dynamic_red
|
||
else:
|
||
cache = DynamicThresholdCache(
|
||
kpi_id=kpi.id,
|
||
period=current_period,
|
||
mean_value=mean_val,
|
||
stddev_value=stddev,
|
||
dynamic_green=dynamic_green,
|
||
dynamic_yellow=dynamic_yellow,
|
||
dynamic_red=dynamic_red,
|
||
)
|
||
db.add(cache)
|
||
computed += 1
|
||
|
||
db.commit()
|
||
return {"message": f"动态阈值计算完成: {computed}个KPI", "computed": computed}
|
||
|
||
|
||
# ============================================================
|
||
# 检查引擎
|
||
# ============================================================
|
||
|
||
def _check_static(value: float, params: dict, kpi) -> tuple:
|
||
"""静态阈值检查"""
|
||
green = params.get("green")
|
||
yellow = params.get("yellow")
|
||
red = params.get("red")
|
||
|
||
# 从KPI定义获取阈值
|
||
if not green and not yellow and not red:
|
||
green = kpi.threshold_green
|
||
yellow = kpi.threshold_yellow
|
||
red = kpi.threshold_red
|
||
|
||
if _eval_threshold(value, green):
|
||
return ("green", f"[静态] {kpi.kpi_name}={value}, 绿灯{green}")
|
||
elif _eval_threshold(value, yellow):
|
||
return ("yellow", f"[静态] {kpi.kpi_name}={value}, 黄灯{yellow}")
|
||
elif red and _eval_threshold(value, red, invert=True):
|
||
return ("red", f"[静态] {kpi.kpi_name}={value}, 红灯{red}")
|
||
|
||
return (None, None)
|
||
|
||
|
||
def _check_dynamic(kpi_id: int, value: float, params: dict, db: Session) -> tuple:
|
||
"""动态阈值检查 — 基于历史均值±标准差"""
|
||
current_period = datetime.now().strftime("%Y-%m")
|
||
cache = db.query(DynamicThresholdCache).filter(
|
||
DynamicThresholdCache.kpi_id == kpi_id,
|
||
DynamicThresholdCache.period == current_period,
|
||
).first()
|
||
|
||
if not cache:
|
||
return (None, None)
|
||
|
||
sensitivity = params.get("sensitivity", 1.0)
|
||
mean_val = cache.mean_value or 0
|
||
stddev_val = (cache.stddev_value or 0) * sensitivity
|
||
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
kpi_name = kpi.kpi_name if kpi else f"KPI#{kpi_id}"
|
||
|
||
if value >= mean_val + stddev_val:
|
||
return ("green", f"[动态] {kpi_name}={value}, 均值={mean_val:.1f}, 标准差={stddev_val:.1f}")
|
||
elif value >= mean_val:
|
||
return ("yellow", f"[动态] {kpi_name}={value}, 均值={mean_val:.1f}, 标准差={stddev_val:.1f}")
|
||
else:
|
||
return ("red", f"[动态] {kpi_name}={value}, 低于均值={mean_val:.1f}, 标准差={stddev_val:.1f}")
|
||
|
||
|
||
def _check_trend(kpi_id: int, value: float, direction: str, params: dict, db: Session) -> tuple:
|
||
"""趋势检查 — 环比变化"""
|
||
threshold_pct = params.get("threshold_pct", 10)
|
||
|
||
# 获取上月值
|
||
prev_value = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi_id,
|
||
KPIValue.actual_value.isnot(None),
|
||
).order_by(KPIValue.period.desc()).offset(1).limit(1).first()
|
||
|
||
if not prev_value or not prev_value.actual_value or prev_value.actual_value == 0:
|
||
return (None, None)
|
||
|
||
change_pct = round((value - prev_value.actual_value) / abs(prev_value.actual_value) * 100, 2)
|
||
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
kpi_name = kpi.kpi_name if kpi else f"KPI#{kpi_id}"
|
||
|
||
if direction == "up" and change_pct > threshold_pct:
|
||
level = "yellow" if change_pct < threshold_pct * 2 else "red"
|
||
return (level, f"[趋势↑] {kpi_name}环比上升{change_pct}%(阈值>{threshold_pct}%), 当前={value}, 上月={prev_value.actual_value}")
|
||
elif direction == "down" and change_pct < -threshold_pct:
|
||
level = "yellow" if abs(change_pct) < threshold_pct * 2 else "red"
|
||
return (level, f"[趋势↓] {kpi_name}环比下降{abs(change_pct)}%(阈值>{threshold_pct}%), 当前={value}, 上月={prev_value.actual_value}")
|
||
|
||
return (None, None)
|
||
|
||
|
||
def _eval_threshold(value: float, threshold_str: str, invert: bool = False) -> bool:
|
||
"""评估阈值: '>=90', '<80', '>5', '<=2' 等"""
|
||
if not threshold_str:
|
||
return False
|
||
threshold_str = str(threshold_str).strip()
|
||
|
||
try:
|
||
if threshold_str.startswith(">="):
|
||
limit = float(threshold_str[2:])
|
||
return value >= limit if not invert else value >= limit
|
||
elif threshold_str.startswith("<="):
|
||
limit = float(threshold_str[2:])
|
||
return value <= limit if not invert else value <= limit
|
||
elif threshold_str.startswith(">"):
|
||
limit = float(threshold_str[1:])
|
||
return value > limit if not invert else value > limit
|
||
elif threshold_str.startswith("<"):
|
||
limit = float(threshold_str[1:])
|
||
return value < limit if not invert else value < limit
|
||
else:
|
||
return False
|
||
except (ValueError, TypeError):
|
||
return False
|