cma-management/backend/app/api/alert_rules.py

""""
预警规则智能化 — 任务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
from app.utils.oplog import log_op
# 裁定 R1（2026-09-19 财务口径）：demo 账套不参与预警生成/遍历（单一事实源）
# 裁定 R5（2026-09-19 财务口径）：期间非规整月 YYYY-MM → 该 kpi_alerts 行不入库（生成侧守卫）
from app.utils.alert_entity_scope import guard_alert_period, is_excluded_entity
# 「实测区 / 预估区」分区口径（财务裁定 2026-10-02）：**预估行禁作预警依据**——
# 本文件的取数读路径一律加 actual_only_clause()，动态阈值白名单去掉 estimated。
from app.utils.kpi_value_scope import actual_only_clause
# G1（2026-10-17）：资金缺口阈值的单一事实源（默认 20/10 万 + 参数校验）
from app.utils.cash_thresholds import (
    CASH_GAP_RULE_TYPE,
    validate_params as validate_cash_gap_params,
)

logger = logging.getLogger("alert_rules")

# 支持的规则类型（P2 2026-09-10: 新增 customer_concentration 客户集中度）
VALID_RULE_TYPES = (
    "static",           # 静态阈值
    "dynamic",          # 动态阈值
    "trend_up",         # 环比上升
    "trend_down",       # 环比下降
    "forecast_deviation",   # 预测偏差（升级2b 2026-08-25）
    "customer_concentration",  # 客户集中度（P2 2026-09-10）
    "cash_gap_threshold",   # 资金缺口阈值配置（G1 2026-10-17，配置行，不进预警生成链）
)

# ============================================================
# 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)
    entity_id = Column(Integer, default=1, comment="企业ID (多租户隔离 2026-08-31 安全修复)")
    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), entity_id: int = Depends(get_entity_id)):
    """获取单个KPI的所有预警规则（账套隔离: 按token企业）"""
    rules = db.query(AlertRule).filter(
        AlertRule.kpi_id == kpi_id, AlertRule.entity_id == entity_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")), entity_id: int = Depends(get_entity_id)):
    """创建预警规则（账套隔离: 写入token企业, 2026-08-31 安全修复）"""
    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 kpi.entity_id != entity_id:
        raise HTTPException(404, "KPI不存在")
    if rule_type not in VALID_RULE_TYPES:  # 升级2b: 预测偏差 / P2-20260910: 客户集中度
        raise HTTPException(400, f"不支持的规则类型: {rule_type}")
    # G1（2026-10-17）：资金缺口阈值为引擎级参数，入库前校验数值（非法配置一律拒收，
    # 防「配置写坏 → 预警静默回落默认值」的假生效）。
    if rule_type == CASH_GAP_RULE_TYPE:
        err = validate_cash_gap_params(data.get("params"))
        if err:
            raise HTTPException(400, err)

    rule = AlertRule(
        entity_id=entity_id,
        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)
    
    # 日志
    log_op(db, "create_alert_rule", target_type="alert_rule", target_id=rule.id,
           detail=f"KPI={kpi.kpi_code}({kpi.kpi_name}) type={rule_type}",
           entity_id=entity_id)
    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), entity_id: int = Depends(get_entity_id)):
    """更新预警规则（账套隔离: 禁止跨企业修改）"""
    rule = db.query(AlertRule).filter(AlertRule.id == rule_id, AlertRule.entity_id == entity_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])
    # G1（2026-10-17）：改完仍是资金阈值配置行 → 参数必须合法（拒收非法值）
    if rule.rule_type == CASH_GAP_RULE_TYPE:
        err = validate_cash_gap_params(rule.params)
        if err:
            db.rollback()
            raise HTTPException(400, err)
    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), entity_id: int = Depends(get_entity_id)):
    """删除预警规则（账套隔离: 禁止跨企业删除）"""
    rule = db.query(AlertRule).filter(AlertRule.id == rule_id, AlertRule.entity_id == entity_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), entity_id: int = Depends(get_entity_id)):
    """批量创建预警规则（账套隔离: 全部写入token企业）
    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")
        if rule_type not in VALID_RULE_TYPES:
            continue
        if rule_type == CASH_GAP_RULE_TYPE and validate_cash_gap_params(rule_data.get("params")):
            continue
        kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
        if not kpi or kpi.entity_id != entity_id:
            continue
        # 检查是否已存在相同类型的规则（同企业内）
        existing = db.query(AlertRule).filter(
            AlertRule.kpi_id == kpi_id,
            AlertRule.rule_type == rule_type,
            AlertRule.entity_id == entity_id,
        ).first()
        if existing:
            continue
        rule = AlertRule(
            entity_id=entity_id,
            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), entity_id: int = Depends(get_entity_id)):
    """为当前企业尚未配置预警规则的KPI生成默认规则（账套隔离 2026-08-31）"""
    # 找到当前企业所有active KPI
    all_kpis = db.query(KPIDefinition).filter(
        KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).all()
    
    created = 0
    for kpi in all_kpis:
        # 检查是否已有任何规则（同企业内）
        existing = db.query(AlertRule).filter(
            AlertRule.kpi_id == kpi.id, AlertRule.entity_id == entity_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(
                entity_id=entity_id,
                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(
            entity_id=entity_id,
            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), entity_id: int = Depends(get_entity_id)):
    """执行当前企业所有KPI的预警检查 — 生成新的预警记录（账套隔离 2026-08-31）

    R1（裁定 2026-09-19）：demo 演示账套不参与预警生成 → 直接返回 skipped。
    """
    if is_excluded_entity(db, entity_id):
        logger.info("预警检查跳过（demo 账套 R1）: entity_id=%s", entity_id)
        return {"message": f"账套 {entity_id} 为 demo 演示账套，按 R1 不参与预警生成", "generated": 0,
                "skipped": True, "reason": "R1_demo_entity_excluded"}
    rules = db.query(AlertRule).filter(
        AlertRule.enabled == 1, AlertRule.entity_id == entity_id).all()
    kpi_cache = {}
    value_cache = {}
    
    alerts_generated = 0
    # R5（裁定 2026-09-19）：期间非规整月（含伪期间 2026-H1/H2）→ 该行不入库，跳过并记数
    skipped_non_month_period = 0
    skipped_non_month_period_kpis = []
    
    for rule in rules:
        try:
            # G1（2026-10-17）：cash_gap_threshold 是资金引擎的参数配置行，不是 KPI 预警规则，
            # 不进 KPIValue 比价链（否则未来若绑定到有值的 KPI 会误生成预警）。
            if rule.rule_type == CASH_GAP_RULE_TYPE:
                continue
            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 or kpi.entity_id != entity_id:
                continue

            import json; params = json.loads(rule.params) if isinstance(rule.params, str) else (rule.params or {})

            # 客户集中度规则（P2 2026-09-10）：不依赖同期KPIValue，直接看客户收入占比
            if rule.rule_type == "customer_concentration":
                alert_level, alert_message = _check_customer_concentration(rule.kpi_id, params, db, entity_id)
                if alert_level and alert_level != "green":
                    existing_alert = db.query(KPIAlert).filter(
                        KPIAlert.entity_id == entity_id,
                        KPIAlert.kpi_id == rule.kpi_id,
                        KPIAlert.alert_type == "concentration",
                        KPIAlert.status == "pending",
                    ).first()
                    if existing_alert:
                        existing_alert.alert_level = alert_level
                        existing_alert.alert_message = alert_message
                    else:
                        db.add(KPIAlert(
                            entity_id=entity_id,
                            kpi_id=rule.kpi_id,
                            kpi_value_id=None,
                            alert_level=alert_level,
                            alert_message=alert_message,
                            alert_type="concentration",
                            status="pending",
                        ))
                        alerts_generated += 1
                continue

            # 获取最新值（仅实测区：预估行禁作预警依据，财务裁定 2026-10-02）
            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,
                    actual_only_clause(),
                ).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
            
            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":
                # R5 生成侧守卫（裁定 2026-09-19）：期间非规整月 YYYY-MM → 该 kpi_alerts 行不入库。
                # 注：`latest_value` 按 period 字符串倒序取，`2026-H1` 会排在 `2026-09` 之前 →
                # 伪期间确实可能被取为「最新值」，此处是伪期间入库唯一的拦截点。
                if not guard_alert_period(period, source="alert_rules.run_all_alert_checks",
                                          detail="kpi=%s rule_id=%s" % (kpi.kpi_code, rule.id)):
                    skipped_non_month_period += 1
                    skipped_non_month_period_kpis.append("%s[%s]" % (kpi.kpi_code, period))
                    continue
                # 检查是否已有相同预警
                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(
                        entity_id=entity_id,
                        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()
    if skipped_non_month_period:
        logger.info("R5 期间守卫：本次跳过 %s 条不可归期预警（%s）",
                    skipped_non_month_period, skipped_non_month_period_kpis)
    return {"message": f"预警检查完成: 生成{alerts_generated}条", "generated": alerts_generated,
            "skipped_non_month_period": skipped_non_month_period,
            "skipped_non_month_period_kpis": skipped_non_month_period_kpis}


@router.get("/dynamic-thresholds")
def get_dynamic_thresholds(kpi_id: Optional[int] = None, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
    """获取动态阈值缓存（账套隔离 2026-08-31）"""
    query = db.query(DynamicThresholdCache).filter(DynamicThresholdCache.entity_id == entity_id)
    if kpi_id:
        query = query.filter(DynamicThresholdCache.kpi_id == kpi_id)
    rows = query.order_by(DynamicThresholdCache.id.desc()).limit(50).all()
    # 列名取自表元数据，值取自 ORM 行（原实现把 Column 对象当行取值，AttributeError: 'Comparator' object has no attribute 'id'）
    columns = [col.name for col in DynamicThresholdCache.__table__.columns]
    return {"data": [{name: getattr(row, name) for name in columns} for row in rows]}


@router.post("/calculate-dynamic")
def calculate_dynamic_thresholds(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
    """计算当前企业所有KPI的动态阈值（账套隔离 2026-08-31）"""
    kpis = db.query(KPIDefinition).filter(
        KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).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),
            # 白名单去掉 estimated（财务裁定 2026-10-02）：预估行不参与动态阈值计算
            KPIValue.data_status.in_(["verified"]),
            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.entity_id == entity_id,
            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(
                entity_id=entity_id,
                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):
        # red 阈值字面即命中条件（如 "<600" = 低于600触发红灯；">25" = 高于25触发红灯）
        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),
        actual_only_clause(),
    ).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:])
            # invert=True 时取反：命中 = 值低于阈值（低于下限触发红灯等场景）
            return value < limit if invert else value >= limit
        elif threshold_str.startswith("<="):
            limit = float(threshold_str[2:])
            return value > limit if invert else value <= limit
        elif threshold_str.startswith(">"):
            limit = float(threshold_str[1:])
            return value <= limit if invert else value > limit
        elif threshold_str.startswith("<"):
            limit = float(threshold_str[1:])
            return value >= limit if invert else value < limit
        else:
            return False
    except (ValueError, TypeError):
        return False


# ============================================================
# 预测值检查 + 情景建议
# ============================================================

def _check_forecast_alerts(db: Session, entity_id: int = 1) -> int:
    """检查未来7天预测值是否超限 — 针对trigger_on='forecast'和'both'的规则（账套隔离 2026-08-31）"""
    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.entity_id == entity_id,
        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 or kpi.entity_id != entity_id:
                continue

            # 获取最新的预测（按规则所属企业）
            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(
                            entity_id=entity_id,
                            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), entity_id: int = Depends(get_entity_id)):
    """执行预测值预警检查 — 检查未来7天预测值是否超限（账套隔离 2026-08-31）

    R1（裁定 2026-09-19）：demo 演示账套不参与预警生成。
    """
    if is_excluded_entity(db, entity_id):
        logger.info("预测值预警检查跳过（demo 账套 R1）: entity_id=%s", entity_id)
        return {"message": f"账套 {entity_id} 为 demo 演示账套，按 R1 不参与预警生成",
                "generated": 0, "skipped": True, "reason": "R1_demo_entity_excluded"}
    generated = _check_forecast_alerts(db, entity_id=entity_id)
    return {"message": f"预测值预警检查完成: 生成{generated}条", "generated": generated}


@router.post("/generate-suggestions")
def generate_alert_suggestions(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
    """为当前企业所有未处理的预警生成情景建议（账套隔离 2026-08-31）"""
    from app.utils.cash_forecast_engine import generate_scenario_suggestion

    pending = db.query(KPIAlert).filter(
        KPIAlert.entity_id == entity_id,
        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 预警（去重）

    R1（裁定 2026-09-19）：demo 演示账套不参与预警生成。
    """
    if is_excluded_entity(db, entity_id):
        logger.info("预测偏差检查跳过（demo 账套 R1）: entity_id=%s", entity_id)
        return {"message": f"账套 {entity_id} 为 demo 演示账套，按 R1 不参与预警生成",
                "generated": 0, "skipped": True, "reason": "R1_demo_entity_excluded"}
    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 等半年度）
            # 仅实测区：预估行不得与预测值自比（否则偏差恒 0，预警失效）
            actual = db.query(KPIValue).filter(
                KPIValue.kpi_id == rule.kpi_id,
                KPIValue.period == log.period,
                actual_only_clause(),
            ).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(
                    entity_id=entity_id,
                    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}


# ============================================================
# 客户集中度预警（rule_type=customer_concentration, P2 2026-09-10）
# ============================================================

def _customer_revenue_shares(db: Session, entity_id: int, lookback_months: int = 12, top_n: int = 1):
    """聚合客户（往来单位）收入占比 — 数据源 zxt_masterbill（智享通**销售类**单据, unit_id=客户）

    返回 (shares, total, period_range)：
      shares = [(unit_id, amount, pct), ...] 按金额倒序取前 top_n 名
      total  = 窗口内客户收入合计
      period_range = (start_date, end_date) ISO 字符串
    无可用数据时返回 ([], 0.0, None)。

    **口径单一事实源**：SQL 在 `app/utils/cpa_caliber.py`（CPA 派生器共用同一实现，
    防「同一客户收入两份结果」）。用独立连接执行，避免表不存在/方言差异污染调用方 session 事务。

    2026-10-04 P0 口径纠错（任务 finance-20261004T0905-cpa-revenue-caliber-p0）：
      ① 增列**单据类型白名单** `bill_type IN (1,4,14)`（销售单/退货/换货）——非销售单据
         （进货/验收/其它入库/预付款/收款/费用）的 `unit_id` 是**供应商**，混入会把
         「客户集中度」度量成「往来单位集中度」（旧口径 Top1/Top2 恰是最大供应商）；
      ② 取舍口径统一为 `(sum_money + sum_tax_money) > 0`（原 `money_only` 与之在本库
         当前数据上**逐行同集、差集 0 行**，故行为**零变化**；原「17.2% 低估」表述
         经财务复验证否，见 `cpa_caliber` 模块 docstring）。
      预警阈值 50% 在新口径下仍不触发（Top1 22.52% < 50%）⇒ 阈值与分级不漂移。
    """
    from datetime import date, timedelta
    from app.database import get_engine
    from app.utils import cpa_caliber

    end = date.today()
    start = end - timedelta(days=max(1, int(lookback_months)) * 30)
    shares, total, rng = cpa_caliber.fetch_revenue_rows(
        get_engine(), entity_id, start, end, top_n)
    if not shares:
        return ([], 0.0, None)
    return (shares, total, rng)


def _check_customer_concentration(kpi_id: int, params: dict, db: Session, entity_id: int) -> tuple:
    """客户集中度检查 — 单一（或前N）客户收入占比超阈值触发预警

    params（全部可配）:
      threshold_pct   触发阈值，默认 50（黄灯）
      red_ratio       红灯倍数，默认 1.5（占比 ≥ 阈值×1.5 → 红灯）
      top_n           前N客户合计占比，默认 1（单一客户）
      lookback_months 回看月数，默认 12
      source          auto(默认) / zxt_masterbill / kpi
                      auto = 优先客户收入数据，无数据回退到 KPI 口径
    """
    threshold = float(params.get("threshold_pct", 50))
    red_ratio = float(params.get("red_ratio", 1.5))
    top_n = int(params.get("top_n", 1))
    lookback = int(params.get("lookback_months", 12))
    source = str(params.get("source") or "auto").lower()

    kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
    kpi_label = f"{kpi.kpi_name}({kpi.kpi_code})" if kpi else f"KPI#{kpi_id}"

    share = None
    detail = ""

    if source in ("auto", "zxt", "zxt_masterbill"):
        shares, total, rng = _customer_revenue_shares(db, entity_id, lookback, top_n)
        if total > 0 and shares:
            share = round(sum(s[2] for s in shares), 2)
            head = f"前{top_n}客户" if top_n > 1 else "单一客户"
            items = "、".join([f"客户#{s[0]} {s[1]:.0f}元/{s[2]}%" for s in shares])
            detail = f"{head}收入占比={share}%（合计{total:.0f}元，{rng[0]}~{rng[1]}：{items}）"

    if share is None and source in ("auto", "kpi"):
        # 仅实测区：预估行禁作预警依据（财务裁定 2026-10-02）
        latest = db.query(KPIValue).filter(
            KPIValue.kpi_id == kpi_id,
            KPIValue.actual_value.isnot(None),
            actual_only_clause(),
        ).order_by(KPIValue.period.desc(), KPIValue.id.desc()).first()
        if latest:
            share = round(float(latest.actual_value), 2)
            detail = f"客户集中度KPI值={share}%（期间{latest.period}）"

    if share is None:
        return (None, None)
    if share <= threshold:
        return (None, None)

    level = "red" if share >= threshold * red_ratio else "yellow"
    return (level, f"[客户集中度] {kpi_label} {detail}，阈值{threshold}%")


@router.post("/run-customer-concentration")
def run_customer_concentration_check(
    db: Session = Depends(get_db),
    entity_id: int = Depends(get_entity_id),
):
    """客户集中度检查（P2 2026-09-10）— 遍历本企业 customer_concentration 规则，
    计算客户收入占比，超阈值生成/更新 pending 预警（去重）。

    R1（裁定 2026-09-19）：demo 演示账套不参与预警生成。
    """
    if is_excluded_entity(db, entity_id):
        logger.info("客户集中度检查跳过（demo 账套 R1）: entity_id=%s", entity_id)
        return {"message": f"账套 {entity_id} 为 demo 演示账套，按 R1 不参与预警生成",
                "generated": 0, "skipped": True, "reason": "R1_demo_entity_excluded"}
    rules = db.query(AlertRule).filter(
        AlertRule.entity_id == entity_id,
        AlertRule.rule_type == "customer_concentration",
        AlertRule.enabled == 1,
    ).all()
    if not rules:
        return {"message": "无客户集中度规则，可先创建 rule_type=customer_concentration 规则", "generated": 0}

    generated = 0
    for rule in rules:
        try:
            params = rule.params
            if isinstance(params, str):
                params = json.loads(params)
            params = params or {}
            alert_level, alert_message = _check_customer_concentration(rule.kpi_id, params, db, entity_id)
            if not alert_level or alert_level == "green":
                continue
            existing = db.query(KPIAlert).filter(
                KPIAlert.entity_id == entity_id,
                KPIAlert.kpi_id == rule.kpi_id,
                KPIAlert.alert_type == "concentration",
                KPIAlert.status == "pending",
            ).first()
            if existing:
                existing.alert_level = alert_level
                existing.alert_message = alert_message
            else:
                db.add(KPIAlert(
                    entity_id=entity_id,
                    kpi_id=rule.kpi_id,
                    kpi_value_id=None,
                    alert_level=alert_level,
                    alert_message=alert_message,
                    alert_type="concentration",
                    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}