From 3beec14d96ed534a01e19a1a7fd669a9ca374c97 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Sat, 19 Sep 2026 10:05:11 +0800 Subject: [PATCH] fix(alerts): enforce caliber rulings R1-R4 for alert generation R1: exclude status=demo entities from alert traversal (alert_generator / deviation_engine / kpi_bep_deriver / alert_rules / cash engine) R2: only generate alerts within the last 6 calendar months (incl. current) R3: only regular month periods (YYYY-MM) participate R4: decouple generation from push - empty notification_channels never blocks Single source of truth: app/utils/alert_entity_scope.py --- backend/app/api/alert_rules.py | 38 +- backend/app/utils/alert_entity_scope.py | 96 +++++ backend/app/utils/cash_forecast_engine.py | 8 + backend/app/utils/deviation_engine.py | 23 +- backend/scripts/alert_generator.py | 126 +++++- backend/scripts/kpi_bep_deriver.py | 8 +- .../tests/test_alert_caliber_r1r4_20260919.py | 363 ++++++++++++++++++ 7 files changed, 641 insertions(+), 21 deletions(-) create mode 100644 backend/app/utils/alert_entity_scope.py create mode 100644 backend/tests/test_alert_caliber_r1r4_20260919.py diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index 4b9350e6..1efb4cfa 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -14,6 +14,8 @@ 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 +# 裁定 R1（2026-09-19 财务口径）：demo 账套不参与预警生成/遍历（单一事实源） +from app.utils.alert_entity_scope import is_excluded_entity logger = logging.getLogger("alert_rules") @@ -257,7 +259,14 @@ def generate_default_rules(db: Session = Depends(get_db), entity_id: int = Depen @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）""" + """执行当前企业所有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 = {} @@ -642,7 +651,14 @@ def _check_forecast_alerts(db: Session, entity_id: int = 1) -> int: @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）""" + """执行预测值预警检查 — 检查未来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} @@ -690,7 +706,14 @@ def run_forecast_deviation_check( ): """预测偏差检查（升级2b, 2026-08-25）— alert_rules type=forecast_deviation 对每条偏差规则: 取最新预测log(kpi_forecast_log) vs 该期实际值(kpi_values)， - 偏差 > threshold_pct → 生成/更新 pending 预警（去重）""" + 偏差 > 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, @@ -861,7 +884,14 @@ def run_customer_concentration_check( entity_id: int = Depends(get_entity_id), ): """客户集中度检查（P2 2026-09-10）— 遍历本企业 customer_concentration 规则， - 计算客户收入占比，超阈值生成/更新 pending 预警（去重）。""" + 计算客户收入占比，超阈值生成/更新 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", diff --git a/backend/app/utils/alert_entity_scope.py b/backend/app/utils/alert_entity_scope.py new file mode 100644 index 00000000..6c92b0cd --- /dev/null +++ b/backend/app/utils/alert_entity_scope.py @@ -0,0 +1,96 @@ +# -*- coding: utf-8 -*- +"""预警「参与范围」口径 —— 单一事实源（财务裁定 R1/R2/R3，2026-09-19） + +依据：`wecom-finance/reports/CMA_预警口径裁定_20260919.md`（任总 2026-09-18「按建议执行」） + +- **R1 账套范围**：`entities.status = 'demo'` 的账套（演示账套）**不参与预警生成/遍历** + —— demo 仅用于演示展示，不产生业务预警。 +- **R2 期间窗口**：预警遍历期间 = **当月往前 6 个自然月（含当月）**；更早期间**不再新生成**。 + 与裁定 2b（各 KPI 最近可用期间）叠加：`max(可用期间下限, 当月-6月)`。 +- **R3 规整月期间**：仅 `YYYY-MM` 参与；`H1/H2/FY`、纯年份等伪期间**一律不参与**。 + +约定：所有预警生成入口（alert_generator / deviation_engine / kpi_bep_deriver / +alert_rules 生成端点 / cash_forecast_engine）**只调本模块的入口函数**， +禁止在调用点各写一份 `status == 'demo'` / 期间正则 / 窗口算术（避免多处漂移）。 +""" +import logging +import re +from datetime import datetime +from typing import Optional + +from app.models import Entity, KPIDefinition + +logger = logging.getLogger("cma.alert_scope") + +#: R1：不参与预警的账套状态 +ALERT_EXCLUDED_ENTITY_STATUSES = ("demo",) + +#: R2：期间窗口长度（含当月，6 = 当月 + 往前 5 个月） +ALERT_WINDOW_MONTHS = 6 + +#: R3：规整月期间（严格校验 01–12，`2026-13` 之类非法月同样不参与） +MONTH_PERIOD_RE = r"^[0-9]{4}-(?:0[1-9]|1[0-2])$" +_MONTH_PERIOD_RE = re.compile(MONTH_PERIOD_RE) + + +def excluded_entity_ids(db_session) -> set: + """R1：返回不参与预警遍历的账套 id 集合（当前 status='demo' → entity 3/4/5/6）。 + + fail-open：读不到 entities 表（测试库/权限异常）时返回空集并记 WARNING—— + 口径缺失不应把预警主流程整条打断（生成链路另有 fail-safe）。 + """ + try: + rows = db_session.query(Entity.id).filter( + Entity.status.in_(ALERT_EXCLUDED_ENTITY_STATUSES) + ).all() + return {int(r[0]) for r in rows} + except Exception as e: # pragma: no cover - 防御分支 + logger.warning("读取 entities.status=%s 失败，本次按「不排除任何账套」处理: %s", + list(ALERT_EXCLUDED_ENTITY_STATUSES), e) + return set() + + +def is_excluded_entity(db_session, entity_id: Optional[int]) -> bool: + """R1：单个账套是否被排除（None → 按默认账套 1 处理，与 KPIAlert.entity_id 缺省一致）""" + return int(entity_id or 1) in excluded_entity_ids(db_session) + + +def kpi_excluded(db_session, kpi: KPIDefinition, excluded: Optional[set] = None) -> bool: + """R1：KPI 归属账套是否被排除（可传入预取的 excluded 集合，避免循环内重复查询）""" + eids = excluded if excluded is not None else excluded_entity_ids(db_session) + return int(getattr(kpi, "entity_id", None) or 1) in eids + + +def window_lower_period(now: Optional[datetime] = None) -> str: + """R2：期间窗口下限 = 当月往前 6 个自然月（含当月）→ 当月-5 月。 + + 2026-09 → `2026-04`（窗口 = 2026-04 ~ 2026-09）。 + """ + now = now or datetime.now() + months = now.month - (ALERT_WINDOW_MONTHS - 1) + year = now.year + while months <= 0: + months += 12 + year -= 1 + return "%04d-%02d" % (year, months) + + +def is_month_period(period) -> bool: + """R3：是否规整月期间（YYYY-MM）""" + return bool(_MONTH_PERIOD_RE.fullmatch(str(period or ""))) + + +def period_in_window(period, now: Optional[datetime] = None) -> bool: + """R3 + R2：期间是否可参与预警生成（规整月 + 落在窗口内）""" + if not is_month_period(period): + return False + return str(period) >= window_lower_period(now) + + +def scope_summary(db_session, now: Optional[datetime] = None) -> dict: + """口径证据（写日志/报告用）：被排除账套 + 窗口下限""" + return { + "excluded_entities": sorted(excluded_entity_ids(db_session)), + "window_lower_period": window_lower_period(now), + "window_months": ALERT_WINDOW_MONTHS, + } diff --git a/backend/app/utils/cash_forecast_engine.py b/backend/app/utils/cash_forecast_engine.py index 1d6c01a5..4d39cb1c 100644 --- a/backend/app/utils/cash_forecast_engine.py +++ b/backend/app/utils/cash_forecast_engine.py @@ -806,9 +806,17 @@ def check_cash_alerts(db: Session, entity_id: int = 1) -> dict: （此前博海走无排序兜底 → 命中 uk_entity_kpi 索引必中 BH_ADM_EXPENSE(437 管理费用)） F3 去重命中时刷新文本/等级/逾期天数/归属（此前直接 continue → 金额与天数永久冻结在首版） F4 余额快照批次行按开关处理（默认 skip，不再逐户生成伪"到期未收款"） + + R1（裁定 2026-09-19）：demo 演示账套不产生业务预警 → 直接返回 skipped。 """ import json as _json from app.models import KPIAlert, CashPlan, KPIDefinition + from app.utils.alert_entity_scope import is_excluded_entity + + if is_excluded_entity(db, entity_id): + logger.info("资金预警检查跳过（demo 账套 R1）: entity_id=%s", entity_id) + return {"entity_id": entity_id, "skipped": True, "reason": "R1_demo_entity_excluded", + "new_alerts": [], "refreshed": []} result = forecast_cash_flow_with_plans(entity_id, db, days=30) new_alerts = [] diff --git a/backend/app/utils/deviation_engine.py b/backend/app/utils/deviation_engine.py index 0d91d8dc..e1f2617d 100644 --- a/backend/app/utils/deviation_engine.py +++ b/backend/app/utils/deviation_engine.py @@ -14,6 +14,10 @@ from typing import Optional from app.database import get_session_local from app.models import KPIDefinition, KPIValue, KPIAlert, BudgetPlan, SystemConfig +# 裁定 R1/R2/R3（2026-09-19 财务口径）：demo 账套排除 / 期间窗口 / 规整月期间 +from app.utils.alert_entity_scope import ( + excluded_entity_ids, kpi_excluded, period_in_window, window_lower_period, +) logger = logging.getLogger("cma.deviation") @@ -311,16 +315,31 @@ def build_deviation_alert(db, kpi, period: str, entity_id: int = 1, min_rate: fl def run_deviation_check(db_session, period: str = None) -> int: - """运行差异预警检查，返回新增预警数（统一走 build_deviation_alert，P2-⑤）""" + """运行差异预警检查，返回新增预警数（统一走 build_deviation_alert，P2-⑤） + + 2026-09-19 财务口径裁定 R1/R2/R3（`alert_entity_scope`）： + · R1 —— `status='demo'` 的账套不参与遍历（demo 仅演示，不产生业务预警）； + · R2/R3 —— 期间须为规整月 `YYYY-MM` 且落在窗口（当月往前 6 个自然月，含当月）内。 + """ if period is None: period = datetime.now().strftime("%Y-%m") + if not period_in_window(period): + logger.warning(" 差异预警跳过（R2/R3）: 期间 %s 非规整月或早于窗口下限 %s", + period, window_lower_period()) + return 0 + excluded = excluded_entity_ids(db_session) kpis = db_session.query(KPIDefinition).filter( KPIDefinition.status == "active" ).all() new_count = 0 + skipped_demo = 0 for kpi in kpis: + # R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历 + if kpi_excluded(db_session, kpi, excluded): + skipped_demo += 1 + continue # 1. 差异预警：实际 vs 预算（统一逻辑） result = build_deviation_alert(db_session, kpi, period) if result["triggered"]: @@ -375,6 +394,8 @@ def run_deviation_check(db_session, period: str = None) -> int: new_count += 1 logger.info(f" 新增趋势预警 [{trend['level']}] {trend_alert_msg}") + if skipped_demo: + logger.info(" 差异预警跳过 demo 账套 KPI %s 项（R1，裁定 2026-09-19）", skipped_demo) db_session.commit() return new_count diff --git a/backend/scripts/alert_generator.py b/backend/scripts/alert_generator.py index 92fecb0d..14f967de 100644 --- a/backend/scripts/alert_generator.py +++ b/backend/scripts/alert_generator.py @@ -20,6 +20,11 @@ from app.utils.threshold_parser import eval_threshold, LEVEL_PRIORITY # 裁定2a（2026-09-17 财务口径）：比率类指标统一「百分比(%)」口径（取数×100）， # 未核对/疑似单位错配的项降级待核（不落预警、不推送） from app.utils.caliber_units import resolve_caliber +# 裁定 R1/R2/R3（2026-09-19 财务口径）：账套范围（demo 排除）/ 期间窗口 / 规整月期间 +from app.utils.alert_entity_scope import ( + ALERT_WINDOW_MONTHS, excluded_entity_ids, is_month_period, kpi_excluded, + period_in_window, window_lower_period, +) logger = logging.getLogger("cma.alert_gen") @@ -86,7 +91,7 @@ def _check_threshold(actual: float, threshold_expr: str, level: str) -> tuple: def run_alert_check(db_session, period: Optional[str] = None, dry_run: bool = False, - stats: Optional[dict] = None) -> int: + stats: Optional[dict] = None, now: Optional[datetime] = None) -> int: """检查所有KPI的实际值是否触发预警，返回（将）新生成的预警数 2026-09-17 财务口径裁定 2a/2b： @@ -95,12 +100,29 @@ def run_alert_check(db_session, period: Optional[str] = None, dry_run: bool = Fa · **单位（2a）**：判定前经 resolve_caliber() 归一到「百分比(%)」口径（取数×100， 与阈值 150/80 同量纲）；**未核对/疑似单位错配**的项降级待核 → 不落预警、不推送； · dry_run=True → 只计算不写库（抽查复核与放行前证据用）。 + + 2026-09-19 财务口径裁定 R1/R2/R3（`alert_entity_scope` 单一事实源）： + · **R1 账套**：`status='demo'` 的账套（演示账套）不参与遍历 → 跳过并计 + `skipped_demo_kpis`； + · **R2 期间窗口**：仅生成「当月往前 6 个自然月（含当月）」内的预警；更早期间跳过 + （`out_of_window`），与 2b 叠加 = max(各 KPI 最近可用期间, 当月-6月)； + · **R3 伪期间**：仅规整月期间 YYYY-MM 参与，H1/H2/FY 跳过（`skipped_pseudo_period`）。 + · now 参数仅用于口径复算/测试（缺省=系统当前时间）。 """ st = stats if stats is not None else {} + window_lower = window_lower_period(now) + excluded_eids = excluded_entity_ids(db_session) for k, v in (("period_source", "fixed" if period else "per_kpi_latest"), ("period_fixed", period), ("periods_used", {}), ("normalized", 0), ("normalized_kpis", []), ("dimension_suspect", 0), ("suspect_kpis", []), ("no_period", 0), - ("no_value", 0), ("planned", []), ("skipped_existing", 0)): + ("no_value", 0), ("planned", []), ("skipped_existing", 0), + # R1/R2/R3 口径证据 + ("excluded_demo_entities", sorted(excluded_eids)), + ("window_lower_period", window_lower), + ("window_months", ALERT_WINDOW_MONTHS), + ("skipped_demo_kpis", 0), ("demo_kpis", []), + ("skipped_pseudo_period", 0), ("pseudo_period_kpis", []), + ("out_of_window", 0), ("out_of_window_kpis", [])): st.setdefault(k, v) kpis = db_session.query(KPIDefinition).filter( @@ -109,6 +131,12 @@ def run_alert_check(db_session, period: Optional[str] = None, dry_run: bool = Fa new_count = 0 for kpi in kpis: + # R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历 + if kpi_excluded(db_session, kpi, excluded_eids): + st["skipped_demo_kpis"] += 1 + st["demo_kpis"].append("%s/e%s" % (kpi.kpi_code, kpi.entity_id or 1)) + continue + # 跳过无阈值的KPI thr = { "red": kpi.threshold_red, @@ -125,6 +153,21 @@ def run_alert_check(db_session, period: Optional[str] = None, dry_run: bool = Fa logger.warning(f" 跳过（无可用期间）: {kpi.kpi_name} 无任何 YYYY-MM 实测值") continue + # R3（裁定 2026-09-19）：仅规整月期间 YYYY-MM 参与，伪期间一律不参与 + if not is_month_period(p): + st["skipped_pseudo_period"] += 1 + st["pseudo_period_kpis"].append("%s[%s]" % (kpi.kpi_code, p)) + logger.warning(" 跳过（伪期间 R3）: %s 期间=%s 非 YYYY-MM" % (kpi.kpi_name, p)) + continue + + # R2（裁定 2026-09-19）：期间窗口 = 当月往前 6 个自然月（含当月），更早不新生成 + if str(p) < window_lower: + st["out_of_window"] += 1 + st["out_of_window_kpis"].append("%s[%s]" % (kpi.kpi_code, p)) + logger.info(" 跳过（期间超出窗口 R2，窗口下限 %s）: %s[%s]" % ( + window_lower, kpi.kpi_name, p)) + continue + # 获取该KPI该期间的最新值 latest = db_session.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, @@ -261,7 +304,8 @@ def run_rule_engine_checks(db_session) -> dict: AlertRule, run_forecast_deviation_check, run_customer_concentration_check, ) - stats = {"forecast_deviation": 0, "customer_concentration": 0} + stats = {"forecast_deviation": 0, "customer_concentration": 0, + "skipped_demo_entities": 0, "excluded_entities": sorted(excluded_entity_ids(db_session))} try: rows = db_session.query(AlertRule.entity_id).filter( AlertRule.enabled == 1, @@ -272,6 +316,11 @@ def run_rule_engine_checks(db_session) -> dict: return stats for (eid,) in rows: + # R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历 + if int(eid or 1) in set(stats["excluded_entities"]): + stats["skipped_demo_entities"] += 1 + logger.info(f" 规则引擎跳过（demo 账套 R1）: entity={eid}") + continue for key, fn in (("forecast_deviation", run_forecast_deviation_check), ("customer_concentration", run_customer_concentration_check)): try: @@ -285,24 +334,42 @@ def run_rule_engine_checks(db_session) -> dict: return stats -def generate_and_push(db_session, period: Optional[str] = None, dry_run: bool = False) -> dict: +def _enabled_channel_count(db_session) -> int: + """R4（裁定 2026-09-19）：推送出口计数 —— 生成/推送解耦，空通道只记录不报错。""" + from app.models import NotificationChannel + try: + return int(db_session.query(NotificationChannel).filter( + NotificationChannel.enabled == True # noqa: E712 - 与其他调用点一致 + ).count()) + except Exception as e: # pragma: no cover - 防御分支 + logger.warning(f"读取通知渠道失败（按无渠道处理）: {e}") + return 0 + + +def generate_and_push(db_session, period: Optional[str] = None, dry_run: bool = False, + now: Optional[datetime] = None) -> dict: """生成预警并推送，返回统计 2026-09-17 财务口径裁定 2a/2b： · **period=None** → 判定期间 = 各 KPI「最近可用期间」（裁定2b，不再硬编码当月）； 显式给出 period → 固定该期间（抽查复核用）； · **dry_run=True** → 零写入、零推送：返回将生成的预警清单与口径证据（放行前抽查复核）。 + + 2026-09-19 裁定 R1–R4： + · R1/R2/R3 在 run_alert_check 内落地（demo 账套排除 / 期间窗口 / 规整月期间）； + · **R4 生成与推送解耦**：`notification_channels` 为空时仅入库、不报错、不阻断； """ - cur_month = datetime.now().strftime("%Y-%m") + cur_month = (now or datetime.now()).strftime("%Y-%m") st = {} - logger.info("开始预警检查（当月=%s；判定期间=%s）..." % ( - cur_month, ("固定 %s" % period) if period else "各 KPI 最近可用期间（裁定2b）")) + logger.info("开始预警检查（当月=%s；判定期间=%s；窗口下限=%s；排除账套=%s）..." % ( + cur_month, ("固定 %s" % period) if period else "各 KPI 最近可用期间（裁定2b）", + window_lower_period(now), sorted(excluded_entity_ids(db_session)))) # 0. Bug1修复: 先清理历史green预警（dry-run 不写） cleaned = 0 if dry_run else cleanup_green_alerts(db_session) # 1. 生成预警（期间口径 + 单位口径均在此落地） - new_count = run_alert_check(db_session, period=period, dry_run=dry_run, stats=st) + new_count = run_alert_check(db_session, period=period, dry_run=dry_run, stats=st, now=now) logger.info(f"预警检查完成: {'将新增' if dry_run else '新增'} {new_count} 条") # 1a. 覆盖度自检（2026-09-16 项3；2026-09-17 裁定2b 过渡期保留并扩展「期间回落」明细）： @@ -338,19 +405,29 @@ def generate_and_push(db_session, period: Optional[str] = None, dry_run: bool = else run_rule_engine_checks(db_session) # 2. 推送（dry-run 零推送） + # R4（裁定 2026-09-19）：生成与推送解耦 —— notification_channels 为空时 + # **仅入库、不报错、不阻断**（推送出口缺失只是"未触达"，不等于"未生成"）。 + channels_enabled = _enabled_channel_count(db_session) + push_skipped_no_channel = channels_enabled == 0 pushed = 0 if new_count > 0 and not dry_run: - try: - from app.utils.notifier import push_pending_alerts - pushed = push_pending_alerts(db_session) - logger.info(f"推送完成: {pushed} 条") - except Exception as e: - logger.error(f"推送失败: {e}") + if push_skipped_no_channel: + logger.info("推送出口缺失（notification_channels 启用数=0）→ 仅入库、不推送、不阻断（裁定 R4）") + else: + try: + from app.utils.notifier import push_pending_alerts + pushed = push_pending_alerts(db_session) + logger.info(f"推送完成: {pushed} 条") + except Exception as e: + logger.error(f"推送失败（不影响已入库预警，生成/推送解耦 R4）: {e}") return {"period": cur_month, "dry_run": dry_run, "period_source": st.get("period_source"), "period_fixed": st.get("period_fixed"), "periods_used": st.get("periods_used", {}), "new_alerts": new_count, "pushed": pushed, + # R4：推送出口（生成/推送解耦证据） + "channels_enabled": channels_enabled, + "push_skipped_no_channel": push_skipped_no_channel, "cleaned_green": cleaned, "rule_engine": rule_stats, "coverage": coverage, # 裁定2a 口径证据 @@ -358,7 +435,14 @@ def generate_and_push(db_session, period: Optional[str] = None, dry_run: bool = "dimension_suspect": st.get("dimension_suspect", 0), "suspect_kpis": st.get("suspect_kpis", []), # 期间/取值覆盖明细 "no_period": st.get("no_period", 0), "no_value": st.get("no_value", 0), - "skipped_existing": st.get("skipped_existing", 0), "planned": st.get("planned", [])} + "skipped_existing": st.get("skipped_existing", 0), "planned": st.get("planned", []), + # 裁定 R1/R2/R3 口径证据 + "excluded_demo_entities": st.get("excluded_demo_entities", []), + "skipped_demo_kpis": st.get("skipped_demo_kpis", 0), + "window_lower_period": st.get("window_lower_period"), "window_months": st.get("window_months"), + "out_of_window": st.get("out_of_window", 0), "out_of_window_kpis": st.get("out_of_window_kpis", []), + "skipped_pseudo_period": st.get("skipped_pseudo_period", 0), + "pseudo_period_kpis": st.get("pseudo_period_kpis", [])} def coverage_note(result: dict) -> str: @@ -380,6 +464,18 @@ def coverage_note(result: dict) -> str: % (result["dimension_suspect"], (result.get("suspect_kpis") or [])[:3])) if result.get("normalized"): parts.append(" | 口径归一 ×100 %s 项" % result["normalized"]) + # 裁定 R1/R2/R3/R4（2026-09-19）证据 + if result.get("skipped_demo_kpis"): + parts.append(" | R1 排除 demo 账套 %s（跳过 KPI %s 项）" + % (result.get("excluded_demo_entities"), result["skipped_demo_kpis"])) + if result.get("out_of_window"): + parts.append(" | R2 窗口(下限 %s)外不新生成 %s 项" + % (result.get("window_lower_period"), result["out_of_window"])) + if result.get("skipped_pseudo_period"): + parts.append(" | R3 伪期间跳过 %s 项" % result["skipped_pseudo_period"]) + if result.get("push_skipped_no_channel"): + parts.append(" | R4 推送出口缺失(启用渠道 %s) → 仅入库、不阻断" + % result.get("channels_enabled")) return "".join(parts) diff --git a/backend/scripts/kpi_bep_deriver.py b/backend/scripts/kpi_bep_deriver.py index 19b76d2f..c1bc59ea 100644 --- a/backend/scripts/kpi_bep_deriver.py +++ b/backend/scripts/kpi_bep_deriver.py @@ -38,6 +38,8 @@ import re from sqlalchemy import func from app.database import get_session_local from app.models import KPIValue, KPIDefinition, VoucherDetail, KPIAlert +# 裁定 R1（2026-09-19 财务口径）：demo 账套不产生业务预警（派生值仍落库，仅跳过预警同步） +from app.utils.alert_entity_scope import is_excluded_entity logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger("cma.bep_deriver") @@ -320,7 +322,11 @@ def run_derive(db, entity_id: int, period: str, rebate_rate_pct: float, # 5. 同步预警: 仅对本派生KPI评估 threshold（green=正常不落 pending，yellow/red 落 pending）。 # 幂等规则：同一 KPIValue 已有同级别 pending → 保持不动（值未变不重建）； # 其它/旧级别 pending → resolve；最新派生行永远只留一条当前级别。 - for code, _ in TARGET_KPIS: + # R1（裁定 2026-09-19）：demo 账套不产生业务预警 → 整段预警同步跳过（派生值照落库）。 + target_sync = [] if is_excluded_entity(db, entity_id) else TARGET_KPIS + if not target_sync: + logger.info("demo 账套（entity=%s）跳过预警同步（R1，裁定 2026-09-19）", entity_id) + for code, _ in target_sync: kpi = kpis[code] kpi_value = value_rows.get(kpi.id) if not kpi_value: diff --git a/backend/tests/test_alert_caliber_r1r4_20260919.py b/backend/tests/test_alert_caliber_r1r4_20260919.py new file mode 100644 index 00000000..c2d06037 --- /dev/null +++ b/backend/tests/test_alert_caliber_r1r4_20260919.py @@ -0,0 +1,363 @@ +# -*- coding: utf-8 -*- +"""财务预警口径裁定 R1–R4 落地测试（2026-09-19） + +依据：`wecom-finance/reports/CMA_预警口径裁定_20260919.md`（任总 2026-09-18「按建议执行」） + · R1 —— `entities.status='demo'` 的账套不参与预警生成/遍历（demo 仅演示展示）； + · R2 —— 预警遍历期间 = 当月往前 6 个自然月（含当月），更早期间不再新生成； + · R3 —— 仅规整月期间 `YYYY-MM` 参与，`H1/H2/FY` 等伪期间一律不参与； + · R4 —— 生成与推送解耦：`notification_channels` 为空时仅入库、不报错、不阻断。 + +口径唯一事实源：`app/utils/alert_entity_scope.py`（本套件同时锁该模块的入口函数）。 +""" +from datetime import datetime + +import pytest +from sqlalchemy.orm import Session + +from tests.conftest import create_test_kpi +from app.models import Entity, KPIValue, KPIAlert, NotificationChannel +from app.utils.alert_entity_scope import ( + ALERT_EXCLUDED_ENTITY_STATUSES, ALERT_WINDOW_MONTHS, excluded_entity_ids, + is_excluded_entity, is_month_period, kpi_excluded, period_in_window, window_lower_period, +) +from scripts.alert_generator import generate_and_push, run_alert_check + +#: 固定「当月」= 2026-09 → 窗口 = 2026-04 ~ 2026-09（口径复算不依赖真实时钟） +NOW = datetime(2026, 9, 19, 10, 0) + +RED_EXPR = "<58.4" # 0<实测<10 比值形态不触发 dimension_suspect（见裁定2a 降级门） + + +def _ent(db, eid, status, name=None): + db.add(Entity(id=eid, name=name or ("账套%d" % eid), short_name="e%d" % eid, status=status)) + db.commit() + + +def _kpi_value(db, kpi, period, value, entity_id=1): + db.add(KPIValue(entity_id=entity_id, kpi_id=kpi.id, period=period, actual_value=value)) + db.commit() + + +def _demo_and_active(db): + _ent(db, 1, "active", "陕西酣客文化传媒") + _ent(db, 6, "demo", "演示企业（虚构数据）") + + +# ───────────────────────── R1：demo 账套排除 ───────────────────────── + +class TestR1DemoEntityExcluded: + def test_excluded_ids_only_demo(self, db: Session): + _ent(db, 1, "active") + _ent(db, 3, "demo") + _ent(db, 4, "demo") + _ent(db, 5, "demo") + _ent(db, 6, "demo") + assert excluded_entity_ids(db) == {3, 4, 5, 6} + assert is_excluded_entity(db, 6) is True + assert is_excluded_entity(db, 1) is False + assert is_excluded_entity(db, None) is False, "None → 默认账套 1（不被排除）" + assert ALERT_EXCLUDED_ENTITY_STATUSES == ("demo",) + + def test_demo_entity_missing_table_fail_open(self, db: Session, monkeypatch): + """口径读取失败时 fail-open（不阻断预警主流程），返回空集""" + import app.utils.alert_entity_scope as scope + + def _boom(*a, **kw): + raise RuntimeError("simulated db error") + + monkeypatch.setattr(db, "query", _boom) + assert scope.excluded_entity_ids(db) == set() + + def test_demo_kpi_not_traversed_active_kpi_still_is(self, db: Session): + """同一形态两份 KPI：demo 账套(e6) 跳过、active 账套(e1) 正常判定（正对照）""" + _demo_and_active(db) + k_demo = create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入", + entity_id=6, unit="%", threshold_red=RED_EXPR) + k_real = create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入", + entity_id=1, unit="%", threshold_red=RED_EXPR) + _kpi_value(db, k_demo, "2026-09", 14.45, entity_id=6) + _kpi_value(db, k_real, "2026-09", 14.45, entity_id=1) + + st = {} + n = run_alert_check(db, period=None, dry_run=True, stats=st, now=NOW) + + assert n == 1, "只有 active 账套的 KPI 参与判定" + assert st["skipped_demo_kpis"] == 1 + assert st["demo_kpis"] == ["DEMO_REVENUE/e6"] + assert st["excluded_demo_entities"] == [6] + assert [p["kpi"].split("/")[0] for p in st["planned"]] == ["F_REVENUE"] + assert kpi_excluded(db, k_demo) is True and kpi_excluded(db, k_real) is False + + def test_demo_kpi_writes_no_alert(self, db: Session): + """非 dry-run：demo KPI 不落库；active KPI 落库（前后计数差 =1）""" + _demo_and_active(db) + k_demo = create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入", + entity_id=6, unit="%", threshold_red=RED_EXPR) + k_real = create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入", + entity_id=1, unit="%", threshold_red=RED_EXPR) + _kpi_value(db, k_demo, "2026-09", 14.45, entity_id=6) + _kpi_value(db, k_real, "2026-09", 14.45, entity_id=1) + before = db.query(KPIAlert).count() + + st = {} + n = run_alert_check(db, dry_run=False, stats=st, now=NOW) + db.expire_all() + + rows = db.query(KPIAlert).all() + assert n == 1 and len(rows) == before + 1 + assert [r.entity_id for r in rows] == [1], "demo 账套零预警" + + def test_rule_engine_skips_demo_entity(self, db: Session, monkeypatch): + """规则引擎（按 AlertRule.entity_id 遍历）跳过 demo 账套""" + from app.api import alert_rules as ar + from scripts import alert_generator as ag + + ar.AlertRule.__table__.create(bind=db.get_bind(), checkfirst=True) + _demo_and_active(db) + called = [] + monkeypatch.setattr(ar, "run_forecast_deviation_check", + lambda db, entity_id: called.append(("fd", entity_id)) or {"generated": 0}) + monkeypatch.setattr(ar, "run_customer_concentration_check", + lambda db, entity_id: called.append(("cc", entity_id)) or {"generated": 0}) + for eid in (1, 6): + db.add(ar.AlertRule(entity_id=eid, kpi_id=1, rule_type="forecast_deviation", enabled=1)) + db.commit() + + stats = ag.run_rule_engine_checks(db) + + assert stats["excluded_entities"] == [6] + assert stats["skipped_demo_entities"] == 1 + assert sorted(e for _, e in called) == [1, 1], "demo 账套一条规则都不跑" + + def test_alert_rules_endpoints_skip_demo(self, db: Session): + """生成端点对 demo 账套直接返回 skipped（不起规则遍历）""" + from app.api.alert_rules import ( + AlertRule, run_all_alert_checks, run_forecast_alert_check, + run_forecast_deviation_check, run_customer_concentration_check, + ) + + AlertRule.__table__.create(bind=db.get_bind(), checkfirst=True) + _demo_and_active(db) + for fn in (run_all_alert_checks, run_forecast_alert_check, + run_forecast_deviation_check, run_customer_concentration_check): + r = fn(db=db, entity_id=6) + assert r.get("skipped") is True, fn.__name__ + assert r.get("reason") == "R1_demo_entity_excluded" + assert r.get("generated") == 0 + + def test_cash_alerts_skip_demo(self, db: Session): + from app.utils.cash_forecast_engine import check_cash_alerts + + _demo_and_active(db) + r = check_cash_alerts(db, entity_id=6) + assert r["skipped"] is True and r["reason"] == "R1_demo_entity_excluded" + assert db.query(KPIAlert).count() == 0 + + def test_deviation_check_excludes_demo_kpi(self, db: Session, monkeypatch): + """差异预警遍历（daily_sync 第4步）同样排除 demo 账套 KPI""" + import app.utils.deviation_engine as de + + _demo_and_active(db) + create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入", + entity_id=6, unit="%", threshold_red=RED_EXPR) + create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入", + entity_id=1, unit="%", threshold_red=RED_EXPR) + # 让所有 KPI 都「差异触发」，只观察遍历范围（不依赖预算数据形态） + monkeypatch.setattr(de, "build_deviation_alert", + lambda db_s, kpi, period, *a, **kw: { + "triggered": True, "kpi_alert_level": "red", "alert_type": "budget", + "suggestion": None, + "deviation": {"actual_value": 14.45, "budget_value": 100.0, + "deviation_rate": -85.6}}) + monkeypatch.setattr(de, "check_trend_anomaly", lambda *a, **kw: {"anomaly": False}) + + n = de.run_deviation_check(db, period="2026-08") + db.expire_all() + + rows = db.query(KPIAlert).all() + assert n == 1 and [r.entity_id for r in rows] == [1] + + +# ───────────────────────── R2：期间窗口（6 个自然月） ───────────────────────── + +class TestR2Window: + def test_window_lower_period_math(self): + assert ALERT_WINDOW_MONTHS == 6 + assert window_lower_period(NOW) == "2026-04" # 2026-09 往前 5 个月 + assert window_lower_period(datetime(2026, 9, 1)) == "2026-04" + assert window_lower_period(datetime(2026, 6, 30)) == "2026-01" + assert window_lower_period(datetime(2026, 1, 5)) == "2025-08" # 跨年 + assert window_lower_period(datetime(2026, 3, 31)) == "2025-10" + + def test_period_in_window(self): + assert period_in_window("2026-04", NOW) is True # 窗口下限（含） + assert period_in_window("2026-09", NOW) is True # 当月（含） + assert period_in_window("2026-03", NOW) is False # 窗口外 + assert period_in_window("2026-01", NOW) is False + assert period_in_window("2026-H1", NOW) is False # R3 + assert period_in_window(None, NOW) is False + + def test_out_of_window_kpi_not_generated(self, db: Session): + """博海形态：最新可用期间 2026-01/03（窗口外）→ 不再新生成""" + _demo_and_active(db) + for code, period in (("BH_SALES_1", "2026-03"), ("BH_SALES_5", "2026-01")): + k = create_test_kpi(db, kpi_code=code, kpi_name=code, entity_id=2, + unit="%", threshold_red=RED_EXPR) + _kpi_value(db, k, period, 63070.3, entity_id=2) + + st = {} + n = run_alert_check(db, period=None, dry_run=True, stats=st, now=NOW) + + assert n == 0 + assert st["out_of_window"] == 2 + assert sorted(st["out_of_window_kpis"]) == ["BH_SALES_1[2026-03]", "BH_SALES_5[2026-01]"] + assert st["window_lower_period"] == "2026-04" + + def test_window_boundary_generates_inside_only(self, db: Session): + """边界：窗口下限 2026-04 生成；2026-03 不生成""" + _demo_and_active(db) + k_in = create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入", entity_id=1, + unit="%", threshold_red=RED_EXPR) + k_out = create_test_kpi(db, kpi_code="F_COST_RATIO", kpi_name="费用率", entity_id=1, + unit="%", threshold_red=RED_EXPR) + _kpi_value(db, k_in, "2026-04", 14.45) + _kpi_value(db, k_out, "2026-03", 14.45) + + st = {} + n = run_alert_check(db, period=None, dry_run=True, stats=st, now=NOW) + + assert n == 1 + assert st["periods_used"] == {"2026-04": 1} + assert st["out_of_window"] == 1 + assert st["planned"][0]["kpi"].startswith("F_REVENUE") + + def test_no_new_alert_written_out_of_window(self, db: Session): + """非 dry-run 回查：窗口外 KPI 不落库（前后计数一致）""" + _demo_and_active(db) + k = create_test_kpi(db, kpi_code="BH_SALES_1", kpi_name="销售-陈艳", entity_id=2, + unit="%", threshold_red=RED_EXPR) + _kpi_value(db, k, "2026-01", 63070.3, entity_id=2) + before = db.query(KPIAlert).count() + + run_alert_check(db, dry_run=False, now=NOW) + db.expire_all() + + assert db.query(KPIAlert).count() == before == 0 + + +# ───────────────────────── R3：伪期间不参与 ───────────────────────── + +class TestR3PseudoPeriod: + def test_is_month_period(self): + assert is_month_period("2026-08") is True and is_month_period("2025-12") is True + for bad in ("2026-H1", "2026-H2", "2025-FY", "2025", "2026-13", None, ""): + assert is_month_period(bad) is False, bad + + def test_explicit_pseudo_period_skipped(self, db: Session): + """显式传入 2026-H1 → 伪期间跳过（不放行、不落库）""" + _demo_and_active(db) + k = create_test_kpi(db, kpi_code="F_GROSS_MARGIN", kpi_name="毛利率", + entity_id=1, unit="%", threshold_red="<15") + _kpi_value(db, k, "2026-08", 5.0) + before = db.query(KPIAlert).count() + + st = {} + n = run_alert_check(db, period="2026-H1", dry_run=False, stats=st, now=NOW) + db.expire_all() + + assert n == 0 and db.query(KPIAlert).count() == before + assert st["skipped_pseudo_period"] == 1 + assert st["pseudo_period_kpis"] == ["F_GROSS_MARGIN[2026-H1]"] + + def test_only_pseudo_period_means_no_period(self, db: Session): + """KPI 只有 2026-H1 数据 → 无可用期间（no_period，不静默）""" + _demo_and_active(db) + k = create_test_kpi(db, kpi_code="F_GROSS_MARGIN", kpi_name="毛利率", + entity_id=1, unit="%", threshold_red="<15") + _kpi_value(db, k, "2026-H1", 5.0) + + st = {} + n = run_alert_check(db, period=None, dry_run=True, stats=st, now=NOW) + + assert n == 0 and st["no_period"] == 1 and st["skipped_pseudo_period"] == 0 + + +# ───────────────────────── R4：生成与推送解耦 ───────────────────────── + +class TestR4PushDecoupled: + def _triggering_kpi(self, db): + _demo_and_active(db) + k = create_test_kpi(db, kpi_code="F_REVENUE", kpi_name="营业收入", entity_id=1, + unit="%", threshold_red=RED_EXPR) + _kpi_value(db, k, "2026-09", 14.45) + return k + + def test_empty_channels_only_store_no_error(self, db: Session): + """notification_channels=0 → 仅入库、不报错、不阻断（生成结果照常返回）""" + self._triggering_kpi(db) + assert db.query(NotificationChannel).count() == 0 + + r = generate_and_push(db, dry_run=False, now=NOW) + db.expire_all() + + assert r["new_alerts"] == 1, "预警照常生成入库" + assert r["channels_enabled"] == 0 and r["push_skipped_no_channel"] is True + assert r["pushed"] == 0 + assert db.query(KPIAlert).count() == 1 + assert db.query(KPIAlert).first().status == "pending" + + def test_push_failure_does_not_block_generation(self, db: Session, monkeypatch): + """有渠道但推送抛异常 → 生成结果不受影响（try/except 解耦）""" + import app.utils.notifier as notifier + + self._triggering_kpi(db) + db.add(NotificationChannel(name="企微群", channel_type="wecom", + config={"webhook_url": "https://example.invalid/hook"}, + enabled=True)) + db.commit() + + def _boom(*a, **kw): + raise RuntimeError("simulated push failure") + + monkeypatch.setattr(notifier, "push_pending_alerts", _boom) + + r = generate_and_push(db, dry_run=False, now=NOW) + assert r["new_alerts"] == 1 and r["pushed"] == 0 + assert r["channels_enabled"] == 1 and r["push_skipped_no_channel"] is False + + def test_push_called_when_channel_present(self, db: Session, monkeypatch): + import app.utils.notifier as notifier + + self._triggering_kpi(db) + db.add(NotificationChannel(name="企微群", channel_type="wecom", + config={"webhook_url": "https://example.invalid/hook"}, + enabled=True)) + db.commit() + monkeypatch.setattr(notifier, "push_pending_alerts", lambda db_: 1) + + r = generate_and_push(db, dry_run=False, now=NOW) + assert r["new_alerts"] == 1 and r["pushed"] == 1 and r["channels_enabled"] == 1 + + def test_dry_run_zero_write_and_zero_push(self, db: Session, monkeypatch): + """dry-run：零写入、零推送，且返回 R1/R2/R3 口径证据""" + import app.utils.notifier as notifier + + self._triggering_kpi(db) + k_old = create_test_kpi(db, kpi_code="BH_SALES_1", kpi_name="销售-陈艳", entity_id=2, + unit="%", threshold_red=RED_EXPR) + _kpi_value(db, k_old, "2026-01", 63070.3, entity_id=2) + create_test_kpi(db, kpi_code="DEMO_REVENUE", kpi_name="演示收入", entity_id=6, + unit="%", threshold_red=RED_EXPR) + called = [] + monkeypatch.setattr(notifier, "push_pending_alerts", lambda db_: called.append(1) or 9) + + before = db.query(KPIAlert).count() + r = generate_and_push(db, dry_run=True, now=NOW) + db.expire_all() + + assert db.query(KPIAlert).count() == before == 0, "dry-run 零写入" + assert called == [], "dry-run 零推送" + assert r["new_alerts"] == 1 and r["pushed"] == 0 + assert r["skipped_demo_kpis"] == 1 and r["out_of_window"] == 1 + assert r["window_lower_period"] == "2026-04" and r["excluded_demo_entities"] == [6] + assert r["push_skipped_no_channel"] is True