feat(alerts): 新增客户集中度预警规则 customer_concentration + 预测偏差规则启用/联动每日预警生成

- alert_rules: 新增 rule_type=customer_concentration(单一/前N客户收入占比超阈值触发,黄=超阈值/红=阈值x倍数)
  数据源 zxt_masterbill 客户(往来单位)收入聚合,unit_id<=0 占位剔除;无客户数据可回退 KPI 口径
  复用现有 KPIAlert 生成链路:check-all 内联分支 + 专用端点 POST /run-customer-concentration
- alert_generator(每日9:00): 新增规则引擎步骤,复用上述检查函数跑 forecast_deviation/customer_concentration
- AlertList.vue: 规则类型筛选/编辑项 + 参数表单 + 客户集中度检查按钮
- 测试: tests/test_customer_concentration_rule.py 17 条(规则注册/聚合/双数据源/黄红灯/去重/check-all)
- 验证脚本: scripts/verify_forecast_deviation_rules_e2e.py / verify_customer_concentration_rule_e2e.py
This commit is contained in:
Hermes CI Fix
2026-09-10 09:56:50 +08:00
parent f8688713db
commit 9822fbd9ba
7 changed files with 866 additions and 8 deletions
+190 -3
View File
@@ -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}
+42 -2
View File
@@ -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()
@@ -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())
@@ -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())
@@ -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
+2
View File
@@ -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'),
}
+41 -3
View File
@@ -145,6 +145,8 @@
<el-option label="动态阈值" value="dynamic" />
<el-option label="上升趋势" value="trend_up" />
<el-option label="下降趋势" value="trend_down" />
<el-option label="预测偏差" value="forecast_deviation" />
<el-option label="客户集中度" value="customer_concentration" />
</el-select>
<el-select v-model="ruleEnabledFilter" placeholder="状态" clearable size="small" style="width:100px">
<el-option label="已启用" :value="1" />
@@ -152,6 +154,7 @@
</el-select>
<el-button type="primary" size="small" @click="loadRules">查询</el-button>
<el-button size="small" @click="checkForecastAlerts" :loading="fcLoading">🔮 检查预测值</el-button>
<el-button size="small" @click="checkConcentrationAlerts" :loading="ccLoading">🏦 检查客户集中度</el-button>
<el-button size="small" @click="generateAlertSuggestions" :loading="sgLoading">💡 生成情景建议</el-button>
</div>
@@ -268,6 +271,8 @@
<el-option label="动态阈值" value="dynamic" />
<el-option label="上升趋势预警" value="trend_up" />
<el-option label="下降趋势预警" value="trend_down" />
<el-option label="预测偏差预警" value="forecast_deviation" />
<el-option label="客户集中度预警" value="customer_concentration" />
</el-select>
</el-form-item>
<el-form-item label="触发条件">
@@ -298,6 +303,26 @@
<span style="color:#909399;font-size:12px;margin-left:8px;">超过此百分比触发预警</span>
</el-form-item>
</template>
<template v-if="ruleForm.rule_type === 'forecast_deviation'">
<el-form-item label="偏差阈值(%)">
<el-input-number v-model="ruleForm.params.threshold_pct" :min="1" :max="100" />
<span style="color:#909399;font-size:12px;margin-left:8px;">预测值与该期实际值偏差超过此比例触发(红灯=2倍)</span>
</el-form-item>
</template>
<template v-if="ruleForm.rule_type === 'customer_concentration'">
<el-form-item label="占比阈值(%)">
<el-input-number v-model="ruleForm.params.threshold_pct" :min="1" :max="100" />
<span style="color:#909399;font-size:12px;margin-left:8px;">单一客户收入占比超过此比例触发(默认50)</span>
</el-form-item>
<el-form-item label="红灯倍数">
<el-input-number v-model="ruleForm.params.red_ratio" :min="1" :max="3" :step="0.1" />
<span style="color:#909399;font-size:12px;margin-left:8px;">占比 ≥ 阈值×倍数 → 红灯(默认1.5)</span>
</el-form-item>
<el-form-item label="前N客户">
<el-input-number v-model="ruleForm.params.top_n" :min="1" :max="10" />
<span style="color:#909399;font-size:12px;margin-left:8px;">默认1=单一客户占比</span>
</el-form-item>
</template>
</el-form>
<template #footer>
<el-button @click="showRuleForm=false">取消</el-button>
@@ -383,7 +408,7 @@ const ruleForm = ref({
rule_type: 'static',
trigger_on: 'actual',
enabled: 1,
params: { green: '', yellow: '', red: '', sensitivity: 1.0, threshold_pct: 10 },
params: { green: '', yellow: '', red: '', sensitivity: 1.0, threshold_pct: 10, red_ratio: 1.5, top_n: 1 },
})
// ── 改善计划 ──
@@ -416,10 +441,10 @@ function formatTime(t: string): string {
}
function ruleTypeLabel(type: string): string {
return { static: '静态阈值', dynamic: '动态阈值', trend_up: '上升趋势', trend_down: '下降趋势' }[type] || type
return { static: '静态阈值', dynamic: '动态阈值', trend_up: '上升趋势', trend_down: '下降趋势', forecast_deviation: '预测偏差', customer_concentration: '客户集中度' }[type] || type
}
function ruleTypeTag(type: string): string {
return { static: 'primary', dynamic: 'success', trend_up: 'warning', trend_down: 'danger' }[type] || 'info'
return { static: 'primary', dynamic: 'success', trend_up: 'warning', trend_down: 'danger', forecast_deviation: 'warning', customer_concentration: 'danger' }[type] || 'info'
}
function formatRuleParams(params: any, type: string): string {
@@ -428,6 +453,8 @@ function formatRuleParams(params: any, type: string): string {
if (type === 'static') return `绿${params.green||'-'} / 黄${params.yellow||'-'} / 红${params.red||'-'}`
if (type === 'dynamic') return `敏感度=${params.sensitivity||1.0}`
if (type === 'trend_up' || type === 'trend_down') return `阈值>${params.threshold_pct||10}%`
if (type === 'forecast_deviation') return `偏差>${params.threshold_pct||15}%`
if (type === 'customer_concentration') return `单一客户占比>${params.threshold_pct||50}%(前${params.top_n||1}客户)`
return JSON.stringify(params)
}
@@ -549,6 +576,16 @@ async function checkForecastAlerts() {
fcLoading.value = false
}
async function checkConcentrationAlerts() {
ccLoading.value = true
try {
const r: any = await alertRulesApi.checkConcentration()
ElMessage.success(r.message || '客户集中度检查完成')
loadAlerts()
} catch (e) { ElMessage.error('检查失败') }
ccLoading.value = false
}
async function generateAlertSuggestions() {
sgLoading.value = true
try {
@@ -561,6 +598,7 @@ async function generateAlertSuggestions() {
const fcLoading = ref(false)
const sgLoading = ref(false)
const ccLoading = ref(false)
async function calcDynamic() {
calcLoading.value = true