diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index 539d78ba..4b9350e6 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -17,6 +17,16 @@ from app.models import KPIDefinition, KPIValue, KPIAlert, OperationLog 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） +) + # ============================================================ # AlertRule 模型 # ============================================================ @@ -111,7 +121,7 @@ def create_alert_rule(data: dict, db: Session = Depends(get_db), user=Depends(re raise HTTPException(404, "KPI不存在") if kpi.entity_id != entity_id: raise HTTPException(404, "KPI不存在") - if rule_type not in ("static", "dynamic", "trend_up", "trend_down", "forecast_deviation"): # 升级2b: 预测偏差 + if rule_type not in VALID_RULE_TYPES: # 升级2b: 预测偏差 / P2-20260910: 客户集中度 raise HTTPException(400, f"不支持的规则类型: {rule_type}") rule = AlertRule( @@ -171,6 +181,8 @@ def batch_create_rules(data: dict, db: Session = Depends(get_db), entity_id: int 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 kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() if not kpi or kpi.entity_id != entity_id: continue @@ -263,7 +275,35 @@ def run_all_alert_checks(db: Session = Depends(get_db), entity_id: int = Depends 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 + # 获取最新值 latest_value = value_cache.get(rule.kpi_id) if latest_value is None: @@ -278,7 +318,6 @@ def run_all_alert_checks(db: Session = Depends(get_db), entity_id: int = Depends 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 @@ -718,3 +757,151 @@ def run_forecast_deviation_check( 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)。 + 用独立连接执行，避免表不存在/方言差异污染调用方 session 事务。 + """ + from datetime import date, timedelta + from app.database import get_engine + + end = date.today() + start = end - timedelta(days=max(1, int(lookback_months)) * 30) + sql = text(""" + SELECT unit_id, SUM(sum_money) AS amount + FROM zxt_masterbill + WHERE entity_id = :eid + AND unit_id IS NOT NULL AND unit_id > 0 + AND sum_money IS NOT NULL AND sum_money > 0 + AND bill_date IS NOT NULL + AND bill_date >= :start AND bill_date <= :end + GROUP BY unit_id + ORDER BY amount DESC + """) + try: + with get_engine().connect() as conn: + rows = conn.execute(sql, {"eid": entity_id, "start": start, "end": end}).fetchall() + except Exception as e: + logger.error(f"客户收入聚合失败 entity_id={entity_id}: {e}") + return ([], 0.0, None) + + if not rows: + return ([], 0.0, None) + total = float(sum(float(r[1] or 0) for r in rows)) + if total <= 0: + return ([], 0.0, None) + n = max(1, int(top_n)) + shares = [(int(r[0]), float(r[1] or 0), round(float(r[1] or 0) / total * 100, 2)) for r in rows[:n]] + return (shares, total, (start.isoformat(), end.isoformat())) + + +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"): + latest = db.query(KPIValue).filter( + KPIValue.kpi_id == kpi_id, + KPIValue.actual_value.isnot(None), + ).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 预警（去重）。""" + 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} diff --git a/backend/scripts/alert_generator.py b/backend/scripts/alert_generator.py index 7262123a..0e6532c8 100644 --- a/backend/scripts/alert_generator.py +++ b/backend/scripts/alert_generator.py @@ -190,6 +190,41 @@ def run_alert_check(db_session, period: str = None) -> int: return new_count +def run_rule_engine_checks(db_session) -> dict: + """预警规则引擎（rule_type 驱动）— 预测偏差 + 客户集中度（P2 2026-09-10） + + 复用 alert_rules 的检查端点实现（FastAPI 路由函数=带 Depends 默认值的普通函数， + 显式传参即可复用，零逻辑复制），按存在规则的账套逐个执行。 + 任何单点失败只记录日志，不影响静态阈值主流程。 + """ + from app.api.alert_rules import ( + AlertRule, run_forecast_deviation_check, run_customer_concentration_check, + ) + + stats = {"forecast_deviation": 0, "customer_concentration": 0} + try: + rows = db_session.query(AlertRule.entity_id).filter( + AlertRule.enabled == 1, + AlertRule.rule_type.in_(("forecast_deviation", "customer_concentration")), + ).distinct().all() + except Exception as e: + logger.error(f"规则引擎检查跳过（规则查询失败）: {e}") + return stats + + for (eid,) in rows: + for key, fn in (("forecast_deviation", run_forecast_deviation_check), + ("customer_concentration", run_customer_concentration_check)): + try: + r = fn(db=db_session, entity_id=eid) + n = int(r.get("generated", 0) or 0) + stats[key] += n + if n: + logger.info(f" 规则预警 entity={eid} {key}: {n}条") + except Exception as e: + logger.error(f"规则检查失败 entity={eid} {key}: {e}") + return stats + + def generate_and_push(db_session) -> dict: """生成预警并推送，返回统计""" period = datetime.now().strftime("%Y-%m") @@ -202,6 +237,9 @@ def generate_and_push(db_session) -> dict: new_count = run_alert_check(db_session, period) logger.info(f"预警检查完成: 新增 {new_count} 条") + # 1b. 规则引擎检查（预测偏差 / 客户集中度，P2 2026-09-10） + rule_stats = run_rule_engine_checks(db_session) + # 2. 推送 pushed = 0 if new_count > 0: @@ -212,7 +250,8 @@ def generate_and_push(db_session) -> dict: except Exception as e: logger.error(f"推送失败: {e}") - return {"period": period, "new_alerts": new_count, "pushed": pushed, "cleaned_green": cleaned} + return {"period": period, "new_alerts": new_count, "pushed": pushed, + "cleaned_green": cleaned, "rule_engine": rule_stats} if __name__ == "__main__": @@ -224,6 +263,7 @@ if __name__ == "__main__": db = get_session_local()() try: result = generate_and_push(db) - print(f"预警检查: {result['new_alerts']}条新预警 / {result['pushed']}条已推送") + print(f"预警检查: {result['new_alerts']}条新预警 / {result['pushed']}条已推送" + f" / 规则引擎{result['rule_engine']}") finally: db.close() diff --git a/backend/scripts/verify_customer_concentration_rule_e2e.py b/backend/scripts/verify_customer_concentration_rule_e2e.py new file mode 100644 index 00000000..425cda58 --- /dev/null +++ b/backend/scripts/verify_customer_concentration_rule_e2e.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""任务B 交付验收脚本 — customer_concentration 客户集中度预警（生产 E2E） + +做四件事（可复跑）： + 1. 为 entity1 配置常驻客户集中度规则（KPI=C_CUST_CONCENTRATION, source=auto → zxt 客户收入） + 2. 打印真实数据下的单一客户占比（zxt_masterbill 近12个月）作为基线 + 3. 触发测试：建带标记的临时规则（source=kpi）+ 带标记的临时 KPIValue（占比 68%） + → 调 POST /api/cma/alert-rules/run-customer-concentration → 断言 pending 预警生成 + 4. 清理临时规则/KPIValue/预警，回查归零 + +用法：cd /root/cma-management/backend && venv/bin/python3 scripts/verify_customer_concentration_rule_e2e.py +""" +import json +import sys +import urllib.request +import urllib.error + +BASE = "http://127.0.0.1:8010" +ADMIN = {"username": "admin", "password": "admin123"} +MARK = "【系统验证临时数据-验证后删除】" +TEMP_TAG = "e2e-verify-temp" +RULE_TYPE = "customer_concentration" +KPI_ID = 23 # C_CUST_CONCENTRATION 客户集中度 (entity1) + + +def http(method, path, token=None, body=None, timeout=60): + req = urllib.request.Request( + BASE + path, + data=json.dumps(body).encode() if body is not None else None, + method=method, + ) + req.add_header("Content-Type", "application/json") + if token: + req.add_header("Authorization", f"Bearer {token}") + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + return r.status, json.loads(r.read().decode() or "{}") + except urllib.error.HTTPError as e: + return e.code, {"detail": e.read().decode()[:300]} + + +def login(entity_id=1): + st, r = http("POST", "/api/cma/auth/login", body={**ADMIN, "entity_id": entity_id}) + assert st == 200, f"登录失败: {st} {r}" + return r["token"] + + +def main(): + sys.path.insert(0, ".") + from app.database import get_session_local + from app.models import KPIValue, KPIAlert, KPIDefinition + from app.api.alert_rules import AlertRule, _customer_revenue_shares + + token = login(1) + db = get_session_local()() + created_ids = [] + try: + # ── 1+2. 常驻规则 + 真实基线 ────────────────────────────── + print("=== 1. 常驻客户集中度规则（entity1）===") + st, rules = http("GET", f"/api/cma/alert-rules?rule_type={RULE_TYPE}", token=token) + assert st == 200, f"列规则失败: {st} {rules}" + perm = [r for r in rules.get("data", []) if r["kpi_id"] == KPI_ID and r["entity_id"] == 1 + and not (r.get("params") or {}).get("_verify")] + if perm: + print(f" 已存在 rule_id={perm[0]['id']}（幂等跳过）") + else: + st, r = http("POST", "/api/cma/alert-rules", token=token, body={ + "kpi_id": KPI_ID, + "rule_type": RULE_TYPE, + "trigger_on": "actual", + "enabled": 1, + "params": {"threshold_pct": 50, "red_ratio": 1.5, "top_n": 1, "source": "auto"}, + }) + assert st == 200, f"建规则失败: {st} {r}" + print(f" 新建 rule_id={r['data']['id']} params={r['data']['params']}") + + print("=== 2. 真实数据基线（zxt_masterbill 近12个月, entity1）===") + db.rollback() + shares, total, rng = _customer_revenue_shares(db, 1, lookback_months=12, top_n=5) + assert total > 0, "entity1 无客户收入数据 → 基线取数失败" + print(f" 客户收入合计={total:,.0f}元 ({rng[0]}~{rng[1]}), 客户数={len(shares)}") + for uid, amt, pct in shares: + print(f" 客户#{uid}: {amt:,.0f}元 / {pct}%") + top1 = shares[0][2] + print(f" → 单一客户占比 {top1}%（阈值50% → {'触发' if top1 > 50 else '未触发'}）") + + # ── 3. 触发测试（临时规则 + 临时KPI值）──────────────────── + print("=== 3. 触发测试（构造占比超阈值场景）===") + st, r = http("POST", "/api/cma/alert-rules", token=token, body={ + "kpi_id": KPI_ID, + "rule_type": RULE_TYPE, + "trigger_on": "actual", + "enabled": 1, + "params": {"threshold_pct": 50, "red_ratio": 1.5, "top_n": 1, + "source": "kpi", "_verify": TEMP_TAG}, + }) + assert st == 200, f"建临时规则失败: {st} {r}" + temp_rule_id = r["data"]["id"] + created_ids.append(temp_rule_id) + print(f" 临时规则 rule_id={temp_rule_id} (source=kpi, 阈值50%)") + + db.rollback() + temp = KPIValue(kpi_id=KPI_ID, entity_id=1, period="2026-08", actual_value=68.0, + source_type="manual", data_status="pending", remark=MARK) + db.add(temp) + db.commit() + db.refresh(temp) + temp_kv_id = temp.id + print(f" 临时KPIValue id={temp_kv_id} 占比=68.0%") + + st, r = http("POST", "/api/cma/alert-rules/run-customer-concentration", token=token) + assert st == 200, f"触发失败: {st} {r}" + print(f" run-customer-concentration → {r}") + + db.rollback() + alerts = db.query(KPIAlert).filter( + KPIAlert.entity_id == 1, KPIAlert.kpi_id == KPI_ID, + KPIAlert.alert_type == "concentration", KPIAlert.status == "pending", + ).order_by(KPIAlert.id.desc()).all() + assert alerts, "未生成客户集中度预警 → 触发测试 FAIL" + a = alerts[0] + alert_id = a.id + print(f" ✅ 预警生成: id={a.id} level={a.alert_level} type={a.alert_type}") + print(f" message={a.alert_message}") + assert a.alert_level == "yellow", f"68%应为黄灯(50%~75%), 实际={a.alert_level}" + + # ── 4. 清理 ──────────────────────────────────────────── + print("=== 4. 清理临时数据（生产不留假数据）===") + db.rollback() + for obj in (db.query(KPIAlert).filter(KPIAlert.id == alert_id).first(), + db.query(KPIValue).filter(KPIValue.id == temp_kv_id).first(), + db.query(AlertRule).filter(AlertRule.id == temp_rule_id).first()): + if obj: + db.delete(obj) + db.commit() + left = { + "alert": db.query(KPIAlert).filter(KPIAlert.id == alert_id).count(), + "kpi_value": db.query(KPIValue).filter(KPIValue.remark == MARK).count(), + "temp_rule": db.query(AlertRule).filter(AlertRule.id == temp_rule_id).count(), + } + print(f" 回查: {left}") + assert all(v == 0 for v in left.values()) + + # ── 5. 终态 ──────────────────────────────────────────── + print("=== 5. 常驻规则终态 ===") + st, rules = http("GET", f"/api/cma/alert-rules?rule_type={RULE_TYPE}", token=token) + for r in rules.get("data", []): + kpi = db.query(KPIDefinition).filter(KPIDefinition.id == r["kpi_id"]).first() + print(f" rule_id={r['id']} entity={r['entity_id']} kpi={kpi.kpi_code if kpi else r['kpi_id']}" + f" enabled={r['enabled']} params={r['params']}") + finally: + db.close() + + print("\nALL PASS") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/scripts/verify_forecast_deviation_rules_e2e.py b/backend/scripts/verify_forecast_deviation_rules_e2e.py new file mode 100644 index 00000000..c479a807 --- /dev/null +++ b/backend/scripts/verify_forecast_deviation_rules_e2e.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""任务A 交付验收脚本 — forecast_deviation 预测偏差规则（酣客entity1 / 博海entity2） + +做四件事（可复跑，幂等）： + 1. 经生产 API（管理员 JWT）为 entity1 / entity2 各建 1 条 forecast_deviation 规则 + 2. 回查 API 列表 + 直查 DB 实证规则存在（含 rule_id / kpi / params） + 3. 触发测试：给 entity1 规则构造一期「预测偏差超阈值」场景（带标记的临时 KPIValue） + → 调 POST /api/cma/alert-rules/run-forecast-deviation → 断言 pending 预警生成 + 4. 清理：删除临时 KPIValue 与该次生成的预警，回查归零（生产不留假数据） + +用法：cd /root/cma-management/backend && venv/bin/python3 scripts/verify_forecast_deviation_rules_e2e.py +""" +import json +import sys +import urllib.request +import urllib.error + +BASE = "http://127.0.0.1:8010" +ADMIN = {"username": "admin", "password": "admin123"} +MARK = "【系统验证临时数据-验证后删除】" + +# 规则目标：酣客 → 应收账款周转天数(F_AR_DAYS, id=6，有 2026-09 预测日志) +# 博海 → 营收(BH_REVENUE, id=421) +TARGETS = [ + {"entity_id": 1, "kpi_id": 6, "kpi_label": "F_AR_DAYS"}, + {"entity_id": 2, "kpi_id": 421, "kpi_label": "BH_REVENUE"}, +] +RULE_TYPE = "forecast_deviation" +PARAMS = {"threshold_pct": 15, "red_ratio": 2} # >15% 黄，>30% 红（实现按 threshold×2 定红） + + +def http(method, path, token=None, body=None, timeout=30): + url = BASE + path + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request(url, data=data, method=method) + req.add_header("Content-Type", "application/json") + if token: + req.add_header("Authorization", f"Bearer {token}") + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + return r.status, json.loads(r.read().decode() or "{}") + except urllib.error.HTTPError as e: + return e.code, {"detail": e.read().decode()[:300]} + + +def login(entity_id): + st, r = http("POST", "/api/cma/auth/login", body={**ADMIN, "entity_id": entity_id}) + assert st == 200, f"登录 entity{entity_id} 失败: {st} {r}" + return r["token"], r.get("entity_name") + + +def main(): + ok = True + + # ── 1+2. 建规则 ──────────────────────────────────────────── + print("=== 1. 创建 forecast_deviation 规则 ===") + created = {} + for t in TARGETS: + eid = t["entity_id"] + token, ename = login(eid) + st, rules = http("GET", f"/api/cma/alert-rules?rule_type={RULE_TYPE}", token=token) + assert st == 200, f"列规则失败: {st} {rules}" + exist = [r for r in rules.get("data", []) if r["kpi_id"] == t["kpi_id"] and r["rule_type"] == RULE_TYPE] + if exist: + rid = exist[0]["id"] + print(f" entity{eid}({ename}) {t['kpi_label']}: 已存在 rule_id={rid}（幂等跳过）") + else: + st, r = http("POST", "/api/cma/alert-rules", token=token, body={ + "kpi_id": t["kpi_id"], + "rule_type": RULE_TYPE, + "trigger_on": "actual", + "enabled": 1, + "params": PARAMS, + }) + assert st == 200, f"建规则失败 entity{eid}: {st} {r}" + rid = r["data"]["id"] + print(f" entity{eid}({ename}) {t['kpi_label']}: 新建 rule_id={rid} params={r['data']['params']}") + created[eid] = rid + + # ── 3. DB 回查 ───────────────────────────────────────────── + print("=== 2. DB 回查实证 ===") + sys.path.insert(0, ".") + from app.database import get_session_local + from app.models import KPIDefinition, KPIValue, KPIAlert, KpiForecastLog + from app.api.alert_rules import AlertRule + db = get_session_local()() + try: + for eid, rid in created.items(): + row = db.query(AlertRule).filter(AlertRule.id == rid).first() + kpi = db.query(KPIDefinition).filter(KPIDefinition.id == row.kpi_id).first() + print(f" rule_id={row.id} entity_id={row.entity_id} kpi_id={row.kpi_id}" + f"({kpi.kpi_code if kpi else '?'}) type={row.rule_type} enabled={row.enabled} params={row.params}") + assert row.entity_id == eid and row.rule_type == RULE_TYPE + assert row.enabled == 1 + + # ── 4. 触发测试（entity1）───────────────────────────── + print("=== 3. 触发测试（构造偏差场景, entity1）===") + t1 = TARGETS[0] + rid = created[1] + rule = db.query(AlertRule).filter(AlertRule.id == rid).first() + log = (db.query(KpiForecastLog) + .filter(KpiForecastLog.entity_id == 1, KpiForecastLog.kpi_id == rule.kpi_id) + .order_by(KpiForecastLog.created_at.desc()).first()) + forecast = float(log.forecast_value) + # 构造：实际值 = 预测值的一半 → 偏差 100% > 15% → red（>30%） + actual = round(forecast / 2, 2) + temp = KPIValue(kpi_id=rule.kpi_id, entity_id=1, period=log.period, + actual_value=actual, source_type="manual", data_status="pending", remark=MARK) + db.add(temp) + db.commit() + db.refresh(temp) + print(f" 临时KPIValue id={temp.id} period={log.period} 预测={forecast} 实际={actual}") + + token1, _ = login(1) + st, r = http("POST", "/api/cma/alert-rules/run-forecast-deviation", token=token1, timeout=60) + assert st == 200, f"触发失败: {st} {r}" + print(f" run-forecast-deviation → {r}") + + db.expire_all() + # MySQL REPEATABLE READ：API 在另一进程提交，本 session 需结束旧事务快照才能读到 + db.rollback() + alerts = db.query(KPIAlert).filter( + KPIAlert.entity_id == 1, KPIAlert.kpi_id == rule.kpi_id, + KPIAlert.alert_type == "forecast", KPIAlert.status == "pending", + ).order_by(KPIAlert.id.desc()).all() + assert alerts, "未生成预测偏差预警 → 触发测试 FAIL" + a = alerts[0] + alert_id = a.id + print(f" ✅ 预警生成: id={a.id} level={a.alert_level} type={a.alert_type}") + print(f" message={a.alert_message}") + assert a.alert_level == "red", f"偏差100%应为红灯, 实际={a.alert_level}" + + # ── 5. 清理 ──────────────────────────────────────────── + print("=== 4. 清理临时数据（生产不留假数据）===") + temp_id = temp.id + db.rollback() + a_row = db.query(KPIAlert).filter(KPIAlert.id == alert_id).first() + temp_row = db.query(KPIValue).filter(KPIValue.id == temp_id).first() + if a_row: + db.delete(a_row) + if temp_row: + db.delete(temp_row) + db.commit() + left_v = db.query(KPIValue).filter(KPIValue.remark == MARK).count() + left_a = db.query(KPIAlert).filter(KPIAlert.id == alert_id).count() + print(f" 回查: 临时KPIValue剩余={left_v} / 本次预警剩余={left_a}") + assert left_v == 0 and left_a == 0 + + print("=== 5. 规则终态（交付物保留）===") + for eid, rid in created.items(): + row = db.query(AlertRule).filter(AlertRule.id == rid).first() + print(f" entity{eid}: rule_id={row.id} kpi_id={row.kpi_id} type={row.rule_type} enabled={row.enabled}") + finally: + db.close() + + print("\nALL PASS" if ok else "\nFAILED") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_customer_concentration_rule.py b/backend/tests/test_customer_concentration_rule.py new file mode 100644 index 00000000..c1729251 --- /dev/null +++ b/backend/tests/test_customer_concentration_rule.py @@ -0,0 +1,271 @@ +""" +客户集中度预警规则测试（P2 2026-09-10） + +覆盖： + 1. 规则类型 customer_concentration 创建/校验（含非法类型拒绝） + 2. _customer_revenue_shares 客户收入聚合（zxt_masterbill 数据源） + 3. _check_customer_concentration 双数据源（客户收入 / KPI 口径）+ 黄红灯分级 + 4. check-all 与 /run-customer-concentration 触发链路 + 去重 +""" +import pytest +from sqlalchemy import text as sa_text +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi +from app.models import KPIAlert, KPIValue +from app.api.alert_rules import _check_customer_concentration, _customer_revenue_shares + + +@pytest.fixture(autouse=True) +def _isolate_zxt_masterbill(): + """zxt_masterbill 是原生表（不在 Base.metadata 里），conftest 的 drop_all 清不掉它， + 这里在每个用例前后显式丢弃，保证用例间数据隔离。""" + from app.database import get_engine + + def _drop(): + try: + with get_engine().connect() as conn: + conn.execute(sa_text("DROP TABLE IF EXISTS zxt_masterbill")) + conn.commit() + except Exception: + pass + + _drop() + yield + _drop() + + +def _seed_masterbill(db: Session, rows, entity_id: int = 1): + """在测试库建 zxt_masterbill 并写入 (unit_id, amount, bill_date) 测试数据""" + db.execute(sa_text(""" + CREATE TABLE IF NOT EXISTS zxt_masterbill ( + id INTEGER PRIMARY KEY, + entity_id INTEGER, + unit_id INTEGER, + sum_money FLOAT, + bill_date DATE + ) + """)) + for i, (unit_id, amount, bill_date) in enumerate(rows, start=1): + db.execute(sa_text( + "INSERT INTO zxt_masterbill (id, entity_id, unit_id, sum_money, bill_date) " + "VALUES (:id, :eid, :uid, :amt, :bd)" + ), {"id": i, "eid": entity_id, "uid": unit_id, "amt": amount, "bd": bill_date}) + db.commit() + + +class TestRuleTypeRegistration: + """规则类型注册与校验""" + + def test_create_customer_concentration_rule(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + kpi = create_test_kpi(db, kpi_code="C_CONC_T1", kpi_name="客户集中度", dimension="customer") + + resp = client.post("/api/cma/alert-rules", headers=auth_header(token), json={ + "kpi_id": kpi.id, + "rule_type": "customer_concentration", + "trigger_on": "actual", + "params": {"threshold_pct": 50}, + }) + assert resp.status_code == 200, resp.text + data = resp.json()["data"] + assert data["rule_type"] == "customer_concentration" + assert data["entity_id"] == 1 + + def test_reject_unknown_rule_type(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + kpi = create_test_kpi(db, kpi_code="C_CONC_T2") + + resp = client.post("/api/cma/alert-rules", headers=auth_header(token), json={ + "kpi_id": kpi.id, + "rule_type": "no_such_type", + }) + assert resp.status_code == 400 + + def test_batch_create_skips_unknown_type(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + kpi = create_test_kpi(db, kpi_code="C_CONC_T3") + + resp = client.post("/api/cma/alert-rules/batch", headers=auth_header(token), json={ + "rules": [ + {"kpi_id": kpi.id, "rule_type": "customer_concentration", "params": {"threshold_pct": 50}}, + {"kpi_id": kpi.id, "rule_type": "bogus_type"}, + ] + }) + assert resp.status_code == 200 + assert resp.json()["created"] == 1 + + +class TestCustomerRevenueAggregation: + """客户收入聚合（zxt_masterbill）""" + + def test_shares_top1_and_top2(self, db: Session): + _seed_masterbill(db, [ + (101, 600.0, "2026-08-01"), + (102, 300.0, "2026-08-02"), + (103, 100.0, "2026-08-03"), + ]) + shares, total, rng = _customer_revenue_shares(db, 1, lookback_months=12, top_n=1) + assert total == 1000.0 + assert len(shares) == 1 + assert shares[0][0] == 101 + assert shares[0][2] == 60.0 + + shares2, total2, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=2) + assert total2 == 1000.0 + assert round(sum(s[2] for s in shares2), 2) == 90.0 + + def test_no_data_returns_empty(self, db: Session): + shares, total, rng = _customer_revenue_shares(db, 1) + assert shares == [] and total == 0.0 and rng is None + + def test_entity_isolation(self, db: Session): + _seed_masterbill(db, [(201, 500.0, "2026-08-01")], entity_id=2) + shares, total, _ = _customer_revenue_shares(db, 1) + assert total == 0.0 and shares == [] + + def test_placeholder_unit_excluded(self, db: Session): + """unit_id<=0 是占位（无客户/散客），不计入集中度分母""" + _seed_masterbill(db, [ + (-1, 9000.0, "2026-08-01"), + (101, 600.0, "2026-08-02"), + (102, 400.0, "2026-08-03"), + ]) + shares, total, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=1) + assert total == 1000.0 + assert shares[0][0] == 101 and shares[0][2] == 60.0 + + +class TestConcentrationChecker: + """占比判定：阈值 / 黄红灯 / 双数据源""" + + def test_kpi_source_yellow(self, db: Session): + create_test_user(db) + kpi = create_test_kpi(db, kpi_code="C_CONC_10", kpi_name="客户集中度", dimension="customer") + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=68.0, entity_id=1)) + db.commit() + + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1) + assert level == "yellow" + assert "68.0%" in msg + + def test_kpi_source_red(self, db: Session): + create_test_user(db) + kpi = create_test_kpi(db, kpi_code="C_CONC_11", kpi_name="客户集中度", dimension="customer") + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=85.0, entity_id=1)) + db.commit() + + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1) + assert level == "red" + + def test_below_threshold_no_alert(self, db: Session): + create_test_user(db) + kpi = create_test_kpi(db, kpi_code="C_CONC_12", kpi_name="客户集中度", dimension="customer") + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=40.0, entity_id=1)) + db.commit() + + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1) + assert level is None and msg is None + + def test_no_data_no_alert(self, db: Session): + create_test_user(db) + kpi = create_test_kpi(db, kpi_code="C_CONC_13", kpi_name="客户集中度", dimension="customer") + + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1) + assert level is None and msg is None + + def test_masterbill_source_with_kpi_fallback_disabled(self, db: Session): + create_test_user(db) + kpi = create_test_kpi(db, kpi_code="C_CONC_14", kpi_name="客户集中度", dimension="customer") + # KPI 值虽高，但显式指定 source=zxt_masterbill 且无客户收入数据 → 不触发 + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=90.0, entity_id=1)) + db.commit() + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "zxt_masterbill"}, db, 1) + assert level is None + + _seed_masterbill(db, [(101, 600.0, "2026-08-01"), (102, 400.0, "2026-08-05")]) + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "zxt_masterbill"}, db, 1) + assert level == "yellow" + assert "单一客户收入占比=60.0%" in msg + + def test_auto_prefers_masterbill_then_kpi(self, db: Session): + create_test_user(db) + kpi = create_test_kpi(db, kpi_code="C_CONC_15", kpi_name="客户集中度", dimension="customer") + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=70.0, entity_id=1)) + db.commit() + + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50}, db, 1) + assert level == "yellow" + assert "KPI值=70.0%" in msg # 无客户收入数据 → 回退KPI口径 + + _seed_masterbill(db, [(101, 900.0, "2026-08-01"), (102, 100.0, "2026-08-05")]) + level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50}, db, 1) + assert level == "red" # 90% ≥ 50×1.5 + assert "单一客户收入占比=90.0%" in msg + + +class TestConcentrationTriggerChain: + """触发链路：check-all / 专用端点 + 去重""" + + def _mk_rule(self, client, db, token, kpi, params=None): + return client.post("/api/cma/alert-rules", headers=auth_header(token), json={ + "kpi_id": kpi.id, + "rule_type": "customer_concentration", + "params": params or {"threshold_pct": 50, "source": "kpi"}, + }) + + def test_run_endpoint_generates_alert(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + kpi = create_test_kpi(db, kpi_code="C_CONC_20", kpi_name="客户集中度", dimension="customer") + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=72.0, entity_id=1)) + db.commit() + assert self._mk_rule(client, db, token, kpi).status_code == 200 + + resp = client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token)) + assert resp.status_code == 200, resp.text + assert resp.json()["generated"] == 1 + + alerts = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all() + assert len(alerts) == 1 + assert alerts[0].alert_level == "yellow" + assert alerts[0].entity_id == 1 + assert "客户集中度" in alerts[0].alert_message + + def test_run_endpoint_dedupes(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + kpi = create_test_kpi(db, kpi_code="C_CONC_21", kpi_name="客户集中度", dimension="customer") + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=72.0, entity_id=1)) + db.commit() + self._mk_rule(client, db, token, kpi) + + client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token)) + client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token)) + alerts = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all() + assert len(alerts) == 1 # 第二次是更新而非新增 + + def test_check_all_includes_concentration(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + kpi = create_test_kpi(db, kpi_code="C_CONC_22", kpi_name="客户集中度", dimension="customer") + db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=80.0, entity_id=1)) + db.commit() + self._mk_rule(client, db, token, kpi) + + resp = client.post("/api/cma/alert-rules/check-all", headers=auth_header(token)) + assert resp.status_code == 200 + alerts = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all() + assert len(alerts) == 1 + assert alerts[0].alert_level == "red" + + def test_run_endpoint_without_rule(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + resp = client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token)) + assert resp.status_code == 200 + assert resp.json()["generated"] == 0 diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 76091cc5..1ddc07fd 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -172,6 +172,8 @@ export const alertRulesApi = { dynamicThresholds: (params?: any) => api.get('/alert-rules/dynamic-thresholds', { params }), // AI事前预警 checkForecast: () => api.post('/alert-rules/check-forecast'), + checkForecastDeviation: () => api.post('/alert-rules/run-forecast-deviation'), + checkConcentration: () => api.post('/alert-rules/run-customer-concentration'), generateSuggestions: () => api.post('/alert-rules/generate-suggestions'), } diff --git a/frontend/src/views/AlertList.vue b/frontend/src/views/AlertList.vue index 02d91163..91f1d942 100644 --- a/frontend/src/views/AlertList.vue +++ b/frontend/src/views/AlertList.vue @@ -145,6 +145,8 @@ + + @@ -152,6 +154,7 @@ 查询 🔮 检查预测值 + 🏦 检查客户集中度 💡 生成情景建议 @@ -268,6 +271,8 @@ + + @@ -298,6 +303,26 @@ 超过此百分比触发预警 + +