Hermes CI Fix
6b6043536a
升级1: 宏观敏感性弹性历史校准 - 内置宏观历史数据(oil/usd/cpi 2026-01~07月度) - 变化率弹性: 同period匹配KPI历史vs因素历史算弹性 - 合理性校验: |弹性|超出[0.01,0.5]视为噪声回退规则(诚实标注) 升级2: 预测偏差告警闭环 - 新表 kpi_forecast_log(预测历史)+模型KpiForecastLog - 预测时落库(同KPI同预测期覆盖) - alert_rules 支持 rule_type=forecast_deviation(threshold_pct) - POST /alert-rules/run-forecast-deviation: 预测vs实际偏差>阈值生成预警(去重, 超2倍阈值红色) - 端到端验证: 模拟实际500vs预测399.55→偏差20.1%>5%→红色预警生成 回归: pytest 40 passed(predict+alerts)
697 lines
28 KiB
Python
697 lines
28 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.deps import get_entity_id
|
||
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)
|
||
entity_id = Column(Integer, default=1, comment="企业ID (P2多租户隔离 2026-08-23)")
|
||
kpi_id = Column(Integer, nullable=False, comment="关联KPI ID")
|
||
rule_type = Column(String(30), nullable=False, comment="static/dynamic/trend_up/trend_down")
|
||
trigger_on = Column(String(20), default="actual", comment="actual/forecast/both — 实际值/预测值/两者触发")
|
||
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),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""列出所有预警规则（账套隔离: 按token企业, 2026-08-23 P2）"""
|
||
query = db.query(AlertRule).filter(AlertRule.entity_id == entity_id)
|
||
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")
|
||
trigger_on = data.get("trigger_on", "actual")
|
||
|
||
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", "forecast_deviation"): # 升级2b: 预测偏差
|
||
raise HTTPException(400, f"不支持的规则类型: {rule_type}")
|
||
|
||
rule = AlertRule(
|
||
kpi_id=kpi_id,
|
||
rule_type=rule_type,
|
||
trigger_on=trigger_on,
|
||
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", "trigger_on", "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
|
||
import json; params = json.loads(rule.params) if isinstance(rule.params, str) else (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
|
||
|
||
|
||
# ============================================================
|
||
# 预测值检查 + 情景建议
|
||
# ============================================================
|
||
|
||
def _check_forecast_alerts(db: Session) -> int:
|
||
"""检查未来7天预测值是否超限 — 针对trigger_on='forecast'和'both'的规则"""
|
||
from app.utils.cash_forecast_engine import forecast_cash_flow, generate_scenario_suggestion
|
||
from app.models import CashForecast
|
||
|
||
rules = db.query(AlertRule).filter(
|
||
AlertRule.enabled == 1,
|
||
AlertRule.trigger_on.in_(["forecast", "both"]),
|
||
).all()
|
||
|
||
if not rules:
|
||
return 0
|
||
|
||
alerts_generated = 0
|
||
rule_kpi_cache = {}
|
||
|
||
for rule in rules:
|
||
try:
|
||
kpi = rule_kpi_cache.get(rule.kpi_id)
|
||
if kpi is None:
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == rule.kpi_id).first()
|
||
if kpi:
|
||
rule_kpi_cache[rule.kpi_id] = kpi
|
||
if not kpi:
|
||
continue
|
||
|
||
entity_id = kpi.entity_id or 1
|
||
# 获取最新的预测
|
||
latest_forecasts = db.query(CashForecast).filter(
|
||
CashForecast.entity_id == entity_id,
|
||
).order_by(CashForecast.forecast_date.asc()).limit(7).all()
|
||
|
||
if not latest_forecasts:
|
||
# 没有已有预测，执行一次实时预测
|
||
from app.utils.cash_forecast_engine import save_forecast_to_db
|
||
result = forecast_cash_flow(entity_id, db)
|
||
try:
|
||
save_forecast_to_db(entity_id, result, db)
|
||
except:
|
||
pass
|
||
latest_forecasts = db.query(CashForecast).filter(
|
||
CashForecast.entity_id == entity_id,
|
||
).order_by(CashForecast.forecast_date.asc()).limit(7).all()
|
||
|
||
if not latest_forecasts:
|
||
continue
|
||
|
||
# 检查预测值是否超限
|
||
import json; params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {})
|
||
params["kpi"] = kpi
|
||
for forecast in latest_forecasts:
|
||
value = forecast.predicted_cash
|
||
if value is None:
|
||
continue
|
||
alert_level, alert_message = _check_static(value, params, kpi)
|
||
if alert_level and alert_level != "green":
|
||
# 生成情景建议
|
||
sug_type = "cash_critical" if alert_level == "red" else "cash_low"
|
||
sug = generate_scenario_suggestion(
|
||
sug_type, kpi.kpi_name,
|
||
{"expected_receivables": 20, "forecast_date": forecast.forecast_date.isoformat()}
|
||
)
|
||
suggestion_text = f"{sug['title']}：{sug['description']}\\n建议行动：{'；'.join(sug['actions'])}"
|
||
|
||
existing = db.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == rule.kpi_id,
|
||
KPIAlert.alert_type == "forecast",
|
||
KPIAlert.alert_level == alert_level,
|
||
KPIAlert.alert_message == alert_message,
|
||
KPIAlert.status == "pending",
|
||
).first()
|
||
if not existing:
|
||
alert = KPIAlert(
|
||
kpi_id=rule.kpi_id,
|
||
alert_level=alert_level,
|
||
alert_message=alert_message,
|
||
alert_type="forecast",
|
||
suggestion=suggestion_text,
|
||
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 alerts_generated
|
||
|
||
|
||
@router.post("/check-forecast")
|
||
def run_forecast_alert_check(db: Session = Depends(get_db)):
|
||
"""执行预测值预警检查 — 检查未来7天预测值是否超限"""
|
||
generated = _check_forecast_alerts(db)
|
||
return {"message": f"预测值预警检查完成: 生成{generated}条", "generated": generated}
|
||
|
||
|
||
@router.post("/generate-suggestions")
|
||
def generate_alert_suggestions(db: Session = Depends(get_db)):
|
||
"""为所有未处理的预警生成情景建议"""
|
||
from app.utils.cash_forecast_engine import generate_scenario_suggestion
|
||
|
||
pending = db.query(KPIAlert).filter(
|
||
KPIAlert.status == "pending",
|
||
KPIAlert.suggestion.is_(None),
|
||
).all()
|
||
|
||
updated = 0
|
||
for alert in pending:
|
||
try:
|
||
sug_type = "cash_critical" if alert.alert_level == "red" else "cash_low"
|
||
if alert.alert_type == "forecast":
|
||
sug_type = "cash_critical" if alert.alert_level == "red" else "cash_low"
|
||
else:
|
||
sug_type = "cash_low"
|
||
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == alert.kpi_id).first()
|
||
kpi_name = kpi.kpi_name if kpi else "未知KPI"
|
||
|
||
sug = generate_scenario_suggestion(sug_type, kpi_name, {
|
||
"alert_level": alert.alert_level,
|
||
"alert_message": alert.alert_message,
|
||
})
|
||
alert.suggestion = f"{sug['title']}：{sug['description']}\\n建议行动：{'；'.join(sug['actions'])}"
|
||
updated += 1
|
||
except Exception as e:
|
||
logger.error(f"生成建议失败: alert_id={alert.id}, error={e}")
|
||
|
||
db.commit()
|
||
return {"message": f"已为{updated}条预警生成情景建议", "updated": updated}
|
||
|
||
|
||
@router.post("/run-forecast-deviation")
|
||
def run_forecast_deviation_check(
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""预测偏差检查（升级2b, 2026-08-25）— alert_rules type=forecast_deviation
|
||
对每条偏差规则: 取最新预测log(kpi_forecast_log) vs 该期实际值(kpi_values)，
|
||
偏差 > threshold_pct → 生成/更新 pending 预警（去重）"""
|
||
from app.models import KpiForecastLog
|
||
rules = db.query(AlertRule).filter(
|
||
AlertRule.entity_id == entity_id,
|
||
AlertRule.rule_type == "forecast_deviation",
|
||
AlertRule.enabled == 1,
|
||
).all()
|
||
if not rules:
|
||
return {"message": "无预测偏差规则，可先创建 rule_type=forecast_deviation 规则", "generated": 0}
|
||
|
||
generated = 0
|
||
for rule in rules:
|
||
try:
|
||
params = rule.params or {}
|
||
threshold = float(params.get("threshold_pct", 15))
|
||
# 最新预测
|
||
log = db.query(KpiForecastLog).filter(
|
||
KpiForecastLog.entity_id == entity_id,
|
||
KpiForecastLog.kpi_id == rule.kpi_id,
|
||
).order_by(KpiForecastLog.created_at.desc()).first()
|
||
if not log or log.forecast_value is None:
|
||
continue
|
||
# 该预测期的实际值（同period匹配；兼容 2026-H1 等半年度）
|
||
actual = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == rule.kpi_id,
|
||
KPIValue.period == log.period,
|
||
).order_by(KPIValue.id.desc()).first()
|
||
if not actual or not actual.actual_value:
|
||
continue
|
||
base = abs(actual.actual_value)
|
||
if base < 1e-9:
|
||
continue
|
||
deviation_pct = abs(log.forecast_value - actual.actual_value) / base * 100
|
||
if deviation_pct <= threshold:
|
||
continue
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == rule.kpi_id).first()
|
||
kpi_label = f"{kpi.kpi_name}({kpi.kpi_code})" if kpi else f"KPI#{rule.kpi_id}"
|
||
alert_level = "red" if deviation_pct > threshold * 2 else "yellow"
|
||
alert_message = (
|
||
f"预测偏差 {deviation_pct:.1f}% > 阈值{threshold}%："
|
||
f"{kpi_label} 预测{log.period}={log.forecast_value}，实际={actual.actual_value}"
|
||
)
|
||
# 去重: 同KPI+period 已有 pending 偏差预警
|
||
existing = db.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == rule.kpi_id,
|
||
KPIAlert.alert_message.like(f"%预测偏差%{log.period}%"),
|
||
KPIAlert.status == "pending",
|
||
).first()
|
||
if existing:
|
||
existing.alert_message = alert_message
|
||
existing.alert_level = alert_level
|
||
else:
|
||
db.add(KPIAlert(
|
||
kpi_id=rule.kpi_id,
|
||
kpi_value_id=actual.id,
|
||
alert_level=alert_level,
|
||
alert_message=alert_message,
|
||
alert_type="forecast",
|
||
status="pending",
|
||
))
|
||
generated += 1
|
||
except Exception as e:
|
||
logger.error(f"预测偏差检查失败 rule_id={rule.id}: {e}")
|
||
continue
|
||
db.commit()
|
||
return {"message": f"预测偏差检查完成: {generated}条", "generated": generated}
|