wecom-fullstack
52b693ddbb
- deps.get_entity_id: Bot 通道（带 X-BOT-KEY）无显式账套 → 400（旧为兜底 entity 1=酣客）； 无 key 的存量公开/用户端点行为不变（仍 401/兜底 1） - 新增 deps.require_bot_entity_id()：写端点缺省账套一处收口 → 400（406/422 同族） 接入 bot_bridge(×2) / bot_bridge_v2 / bot_suggestions / cash_vouchers_json，并同步文档口径 - deviation_engine.run_deviation_check: period=None 不再硬编码当月，逐 KPI 回落 「该 KPI 自己的最近可用期间」（裁定2b，单一实现收口到 alert_entity_scope）；（当月无 kpi_values ⇒ 差异/趋势入口永久 0 新增）；R2/R3 窗口判定改为逐 KPI；显式 period 语义不变 - alert_generator.resolve_latest_period 委托同一实现（消除期间口径两份） - 测试：新增 tests/test_ask15_17_engine_and_entity_required_20260925.py（12 例）； 4 个既有测试文件把「旧兜底账套 1」显式化（断言未改）
526 lines
22 KiB
Python
526 lines
22 KiB
Python
"""差异预警引擎 — 管理会计OS
|
||
实际 vs 预算/目标对比，超阈值自动推送预警
|
||
|
||
功能:
|
||
1. 实际 vs 预算差异计算（差异额/差异率）
|
||
2. 同比/环比差异计算
|
||
3. 趋势异常检测（连续N期下滑/上升）
|
||
4. 差异预警触发（集成到现有预警系统）
|
||
"""
|
||
import logging
|
||
import json
|
||
from datetime import datetime
|
||
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,
|
||
guard_alert_period, # R5-补1（2026-09-19）：期间可归期 + 关联 kpi_values
|
||
resolve_latest_available_period, # 裁定2b（2026-09-25 ASK-15）：期间回落单一实现
|
||
)
|
||
|
||
logger = logging.getLogger("cma.deviation")
|
||
|
||
|
||
# 越高越好型KPI默认列表（P2-⑤ 2026-08-28: 提为 system_configs 可配置）
|
||
DEFAULT_HIGHER_BETTER = [
|
||
"SALES_TOTAL", "CUSTOMER_COUNT", "SALES_PROFIT_RATE",
|
||
"RECEIVABLE_TURNOVER", "TURNOVER_RATE",
|
||
"CUSTOMER_SATISFACTION", "ORDER_DELIVERY_RATE",
|
||
]
|
||
CONFIG_KEY_HIGHER_BETTER = "kpi_alert_higher_better"
|
||
|
||
|
||
def higher_better_key(entity_id: int) -> str:
|
||
"""账套维度的方向配置键名：`kpi_alert_higher_better.<entity_id>`（零 DDL）
|
||
|
||
entity_id 必传：`None` → ValueError（防「漏传 → 落默认账套」重演污染）。
|
||
"""
|
||
if entity_id is None:
|
||
raise ValueError(
|
||
"higher_better_key 必须传 entity_id（账套隔离，2026-09-23 "
|
||
"dama-entity-isolation-sweep-20260923）"
|
||
)
|
||
return f"{CONFIG_KEY_HIGHER_BETTER}.{int(entity_id)}"
|
||
|
||
|
||
def get_higher_better_codes(db, entity_id: int) -> list:
|
||
"""读取**本账套**越高越好型KPI编码列表（entity_id 必传，漏传 TypeError）
|
||
|
||
账套隔离（2026-09-23 dama-entity-isolation-sweep-20260923 §一①，零 DDL）：
|
||
· 只读 `kpi_alert_higher_better.<entity_id>`，**不做「回退旧全局键」**——
|
||
回退分支本身就是跨账套借用通路（与 `cash.current_balance` 同源缺陷）。
|
||
· 本账套未配置 → `DEFAULT_HIGHER_BETTER`（业务兜底：方向判定的内置默认表；
|
||
与拆键前「全局键无行」时的行为逐字一致，故对现存账套零行为差异）。
|
||
· 旧全局键 `kpi_alert_higher_better` 原样保留作回退杠杆
|
||
（搬运脚本 `--down` 只删新键，回退老码即可继续读旧键）。
|
||
"""
|
||
cfg = db.query(SystemConfig).filter(
|
||
SystemConfig.config_key == higher_better_key(entity_id)
|
||
).first()
|
||
if cfg and cfg.config_value:
|
||
try:
|
||
codes = json.loads(cfg.config_value)
|
||
if isinstance(codes, list):
|
||
return [str(c) for c in codes]
|
||
except Exception:
|
||
logger.warning("system_configs[%s] 解析失败, 回落默认", higher_better_key(entity_id))
|
||
return list(DEFAULT_HIGHER_BETTER)
|
||
|
||
|
||
# ============================================================
|
||
# 差异计算
|
||
# ============================================================
|
||
|
||
def calc_deviation(actual: float, budget: float) -> dict:
|
||
"""计算差异额和差异率"""
|
||
if budget is None or budget == 0:
|
||
return {
|
||
"deviation_amount": None,
|
||
"deviation_rate": None,
|
||
"is_over_budget": None,
|
||
}
|
||
amount = round(actual - budget, 2)
|
||
rate = round(amount / budget * 100, 2)
|
||
return {
|
||
"deviation_amount": amount,
|
||
"deviation_rate": rate,
|
||
"is_over_budget": amount > 0,
|
||
}
|
||
|
||
|
||
def get_budget_for_kpi(db, kpi_id: int, period: str, version: str = None) -> Optional[float]:
|
||
"""获取指定KPI在指定期间的预算值"""
|
||
query = db.query(BudgetPlan).filter(
|
||
BudgetPlan.kpi_id == kpi_id,
|
||
BudgetPlan.period == period,
|
||
BudgetPlan.status == "active",
|
||
)
|
||
if version:
|
||
query = query.filter(BudgetPlan.version == version)
|
||
plan = query.order_by(BudgetPlan.updated_at.desc()).first()
|
||
return plan.budget_value if plan else None
|
||
|
||
|
||
def get_kpi_value_row(db, kpi_id: int, period: str):
|
||
"""取 (kpi_id, period) 的 KPIValue 行（取行口径唯一来源）。
|
||
|
||
与 get_actual_for_kpi 同源：预警行关联的 kpi_values 行 = 数值来源那一行。
|
||
"""
|
||
return db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi_id,
|
||
KPIValue.period == period,
|
||
).order_by(KPIValue.calculated_at.desc()).first()
|
||
|
||
|
||
def get_actual_for_kpi(db, kpi_id: int, period: str) -> Optional[float]:
|
||
"""获取指定KPI在指定期间的实际值"""
|
||
row = get_kpi_value_row(db, kpi_id, period)
|
||
return row.actual_value if row else None
|
||
|
||
|
||
def resolve_alert_kpi_value(db, kpi, period: str):
|
||
"""R5-补1（财务裁定 2026-09-19）：解析差异/趋势预警行必须关联的 `kpi_values` 行。
|
||
|
||
裁定原文：「此类预警行必须在落库时写入 `kpi_value_id`，指向 **同 `entity_id` +
|
||
同 `kpi_id` + 文案所示规整月期间** 的 `kpi_values` 行。文案期间不可解析为 `YYYY-MM`，
|
||
或该 (kpi_id, period) 无 `kpi_values` 行 → 按 R5 `guard_alert_period` 跳过不入库
|
||
（不得以 `kpi_value_id=NULL` 落行——『文案里有期间』不构成可归期）。」
|
||
|
||
- 账套约束由 `kpi_id` 传递：KPIDefinition 归属单一账套，故 kpi_id 已唯一确定 entity_id；
|
||
取行口径与 `get_kpi_value_row`（= `get_actual_for_kpi` 的来源行）完全一致，
|
||
保证「预警关联的那一行」就是该预警数值的来源行，不会关联到别的账套/别的期间。
|
||
- 返回 `(kpi_value_row | None, skip_reason | None)`；跳过原因供调用点计数与日志。
|
||
"""
|
||
if not guard_alert_period(period, source="deviation_engine.resolve_alert_kpi_value",
|
||
detail="kpi=%s" % getattr(kpi, "kpi_code", "")):
|
||
return None, "R5_non_attributable_period"
|
||
row = get_kpi_value_row(db, kpi.id, period)
|
||
if row is None:
|
||
logger.info(
|
||
"R5-补1 期间关联：跳过不入库（该 (kpi, period) 无 kpi_values 行，期间不可派生）"
|
||
" kpi=%s period=%s", getattr(kpi, "kpi_code", ""), period)
|
||
return None, "no_kpi_value_row"
|
||
return row, None
|
||
|
||
|
||
def calc_period_deviation(db, kpi_id: int, period: str) -> dict:
|
||
"""单KPI单期的差异计算"""
|
||
actual = get_actual_for_kpi(db, kpi_id, period)
|
||
budget = get_budget_for_kpi(db, kpi_id, period)
|
||
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
kpi_code = kpi.kpi_code if kpi else "unknown"
|
||
|
||
# 如果没预算值，用 target_value 作为替代
|
||
if budget is None and kpi:
|
||
# 尝试把年度目标按月均分
|
||
month = int(period.split("-")[1])
|
||
target = kpi.target_value
|
||
if target and target > 0 and kpi.frequency == "monthly":
|
||
budget = round(target / 12, 2)
|
||
|
||
deviation = calc_deviation(actual, budget) if actual is not None else None
|
||
|
||
result = {
|
||
"kpi_id": kpi_id,
|
||
"kpi_code": kpi_code,
|
||
"period": period,
|
||
"actual_value": actual,
|
||
"budget_value": budget,
|
||
}
|
||
if deviation:
|
||
result.update(deviation)
|
||
return result
|
||
|
||
|
||
# ============================================================
|
||
# 同比/环比差异
|
||
# ============================================================
|
||
|
||
def calc_period_diff(db, kpi_id: int, current_period: str, diff_type: str = "yoy") -> dict:
|
||
"""计算同比（上年同期）或环比（上期）差异"""
|
||
year, month = current_period.split("-")
|
||
y, m = int(year), int(month)
|
||
|
||
if diff_type == "yoy":
|
||
# 同比：上年同期
|
||
prev_period = f"{y-1}-{m:02d}"
|
||
label = "同比"
|
||
elif diff_type == "mom":
|
||
# 环比：上个月
|
||
prev_m = m - 1
|
||
prev_y = y
|
||
if prev_m <= 0:
|
||
prev_m += 12
|
||
prev_y -= 1
|
||
prev_period = f"{prev_y}-{prev_m:02d}"
|
||
label = "环比"
|
||
else:
|
||
return {"error": f"未知比较类型: {diff_type}"}
|
||
|
||
current = get_actual_for_kpi(db, kpi_id, current_period)
|
||
previous = get_actual_for_kpi(db, kpi_id, prev_period)
|
||
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
kpi_code = kpi.kpi_code if kpi else "unknown"
|
||
|
||
if current is None or previous is None or previous == 0:
|
||
return {
|
||
"kpi_id": kpi_id,
|
||
"kpi_code": kpi_code,
|
||
"type": diff_type,
|
||
"label": label,
|
||
"current_period": current_period,
|
||
"prev_period": prev_period,
|
||
"current_value": current,
|
||
"prev_value": previous,
|
||
"diff_amount": None,
|
||
"diff_rate": None,
|
||
}
|
||
|
||
diff_amount = round(current - previous, 2)
|
||
diff_rate = round(diff_amount / previous * 100, 2)
|
||
return {
|
||
"kpi_id": kpi_id,
|
||
"kpi_code": kpi_code,
|
||
"type": diff_type,
|
||
"label": label,
|
||
"current_period": current_period,
|
||
"prev_period": prev_period,
|
||
"current_value": current,
|
||
"prev_value": previous,
|
||
"diff_amount": diff_amount,
|
||
"diff_rate": diff_rate,
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# 趋势检测
|
||
# ============================================================
|
||
|
||
def check_trend_anomaly(db, kpi_id: int, period: str, consecutive: int = 3) -> dict:
|
||
"""检测连续N期下滑或上升的趋势异常"""
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||
if not kpi:
|
||
return {"anomaly": False}
|
||
|
||
# 获取包括当前期在内的近期数据
|
||
year, month = period.split("-")
|
||
y, m = int(year), int(month)
|
||
|
||
values = []
|
||
for i in range(consecutive + 2): # 多取2期做参考
|
||
p = f"{y}-{m:02d}"
|
||
v = get_actual_for_kpi(db, kpi_id, p)
|
||
if v is not None:
|
||
values.append({"period": p, "value": v})
|
||
m -= 1
|
||
if m <= 0:
|
||
m += 12
|
||
y -= 1
|
||
|
||
values.reverse() # 按时间正序
|
||
if len(values) < consecutive:
|
||
return {"anomaly": False, "reason": "数据不足"}
|
||
|
||
last_n = values[-consecutive:]
|
||
all_decreasing = all(last_n[i]["value"] > last_n[i + 1]["value"] for i in range(len(last_n) - 1))
|
||
all_increasing = all(last_n[i]["value"] < last_n[i + 1]["value"] for i in range(len(last_n) - 1))
|
||
|
||
if all_decreasing:
|
||
return {
|
||
"anomaly": True,
|
||
"type": "continuous_decline",
|
||
"level": "yellow" if consecutive >= 3 else "green",
|
||
"periods": [v["period"] for v in last_n],
|
||
"values": [v["value"] for v in last_n],
|
||
"message": f"{kpi.kpi_name} 连续{consecutive}期下滑",
|
||
}
|
||
if all_increasing:
|
||
return {
|
||
"anomaly": True,
|
||
"type": "continuous_rise",
|
||
"level": "yellow" if consecutive >= 3 else "green",
|
||
"periods": [v["period"] for v in last_n],
|
||
"values": [v["value"] for v in last_n],
|
||
"message": f"{kpi.kpi_name} 连续{consecutive}期上升（可能过热）",
|
||
}
|
||
|
||
return {"anomaly": False}
|
||
|
||
|
||
# ============================================================
|
||
# 差异预警触发
|
||
# ============================================================
|
||
|
||
def build_deviation_alert(db, kpi, period: str, entity_id: int = 1, min_rate: float = 10.0) -> dict:
|
||
"""统一告警构建 — 双出口共享一套逻辑 (P2-⑤ 2026-08-28)
|
||
|
||
预算告警(budget_deviation_alerts) 与 KPIAlert 都调用本函数，差异仅级别映射：
|
||
- budget 出口: warning/critical @ 20/50
|
||
- KPIAlert 出口: yellow/red @ 10/30
|
||
归因(P1-③): attribution 拆解 + 场景建议 由 alert_attribution 组装。
|
||
|
||
返回:
|
||
triggered: bool 是否触发
|
||
level: budget出口级别 warning/critical
|
||
kpi_alert_level: KPIAlert出口级别 yellow/red
|
||
deviation: calc_period_deviation 结果
|
||
suggestion: 模板建议文案
|
||
alert_type: 归因场景类型
|
||
attribution: 归因JSON dict
|
||
scenario_id: 场景建议ID
|
||
"""
|
||
from app.utils.alert_attribution import build_attribution, match_scenario
|
||
|
||
deviation = calc_period_deviation(db, kpi.id, period)
|
||
if deviation.get("deviation_rate") is None:
|
||
return {"triggered": False}
|
||
|
||
rate = abs(deviation["deviation_rate"])
|
||
actual = deviation.get("actual_value")
|
||
budget = deviation.get("budget_value")
|
||
|
||
# 方向性：越高越好型（配置化，本账套键 system_configs.kpi_alert_higher_better.<entity_id>；
|
||
# 2026-09-23 拆账套：entity_id 由调用方必传，不再读全局唯一键）
|
||
higher_better = kpi.kpi_code in get_higher_better_codes(db, entity_id)
|
||
|
||
if higher_better:
|
||
# 实际低于预算才是问题
|
||
if not (actual is not None and budget is not None and actual < budget and rate >= min_rate):
|
||
return {"triggered": False}
|
||
suggestion = (
|
||
f"实际值低于预算 {rate}%，建议分析业务量未达预期的原因（子KPI拆解见归因），"
|
||
f"制定增量获客或转化提升计划"
|
||
)
|
||
else:
|
||
# 实际高于预算才是问题（成本型）
|
||
if not (actual is not None and budget is not None and actual > budget and rate >= min_rate):
|
||
return {"triggered": False}
|
||
suggestion = (
|
||
f"实际值超出预算 {rate}%，建议核查超支原因（科目明细拆解见归因）并采取控制措施"
|
||
)
|
||
|
||
# 级别映射（双出口）
|
||
budget_level = "critical" if rate > 50 else "warning"
|
||
kpi_alert_level = "red" if rate >= 30 else "yellow"
|
||
|
||
# 归因组装 (P1-③)
|
||
alert_type = None
|
||
attribution = None
|
||
scenario_id = None
|
||
try:
|
||
attribution, alert_type = build_attribution(db, kpi.id, period)
|
||
scenario = match_scenario(db, alert_type)
|
||
if scenario:
|
||
scenario_id = scenario["scenario_id"]
|
||
except Exception as e: # 归因失败不阻断告警主流程
|
||
logger.warning("归因组装失败 kpi=%s: %s", kpi.kpi_code, e)
|
||
|
||
return {
|
||
"triggered": True,
|
||
"level": budget_level,
|
||
"kpi_alert_level": kpi_alert_level,
|
||
"deviation": deviation,
|
||
"suggestion": suggestion,
|
||
"alert_type": alert_type,
|
||
"attribution": attribution,
|
||
"scenario_id": scenario_id,
|
||
}
|
||
|
||
|
||
def run_deviation_check(db_session, period: str = None) -> int:
|
||
"""运行差异预警检查，返回新增预警数（统一走 build_deviation_alert，P2-⑤）
|
||
|
||
2026-09-19 财务口径裁定 R1/R2/R3（`alert_entity_scope`）：
|
||
· R1 —— `status='demo'` 的账套不参与遍历（demo 仅演示，不产生业务预警）；
|
||
· R2/R3 —— 期间须为规整月 `YYYY-MM` 且落在窗口（当月往前 6 个自然月，含当月）内。
|
||
· R5-补1 —— 差异/趋势分支**必须落 `kpi_value_id` 关联**（同 kpi 同期间的 `kpi_values` 行）；
|
||
文案期间不可解析为 `YYYY-MM` 或该 (kpi, period) 无 `kpi_values` 行 → 跳过不入库；
|
||
趋势类以「计算基准期（趋势末端当期）」为关联期间；差异类去重键
|
||
= `(entity_id, kpi_id, kpi_value_id, alert_type)`。
|
||
|
||
2026-09-25（ASK-20260924-15，任总 2026-09-24 16:40「放行 15」）——**期间口径修复**：
|
||
· **period=None → 每个 KPI 回落「自己的最近可用期间」**（裁定2b，
|
||
`alert_entity_scope.resolve_latest_available_period`），与 `alert_generator.run_alert_check`
|
||
同源；**不再硬编码 `datetime.now()`（当月）**。
|
||
缺陷实证：当月（2026-09）尚无 `kpi_values` ⇒ 差异分支取不到实际值、趋势末端期落在
|
||
空月 ⇒ 本入口**永久 0 新增**（这正是「当期为当月 → 0 新增」的缺陷形态；同一缺陷
|
||
2026-09-17 只在 alert_generator 侧修过，本入口漏改）。
|
||
· 显式传 period → 固定该期间（原语义不变：抽查复核/测试用）。
|
||
· R2/R3 窗口判定**逐 KPI**做（回落期间各不相同，不能再用一次全局判定）。
|
||
"""
|
||
fixed_period = period # None = 逐 KPI 回落（裁定2b）；有值 = 固定期间（原语义）
|
||
excluded = excluded_entity_ids(db_session)
|
||
kpis = db_session.query(KPIDefinition).filter(
|
||
KPIDefinition.status == "active"
|
||
).all()
|
||
|
||
new_count = 0
|
||
skipped_demo = 0
|
||
skipped_period_linkage = 0
|
||
skipped_linkage_kpis = []
|
||
skipped_no_period = 0
|
||
skipped_out_of_window = 0
|
||
periods_used = {}
|
||
for kpi in kpis:
|
||
# R1（裁定 2026-09-19）：demo 演示账套不参与预警遍历
|
||
if kpi_excluded(db_session, kpi, excluded):
|
||
skipped_demo += 1
|
||
continue
|
||
|
||
# 期间口径（裁定2b，2026-09-25）：显式 → 固定；缺省 → 该 KPI 自己的最近可用期间
|
||
p = fixed_period or resolve_latest_available_period(db_session, kpi.id)
|
||
if not p:
|
||
skipped_no_period += 1
|
||
logger.warning(" 差异/趋势预警跳过（无可用期间）: %s", kpi.kpi_name)
|
||
continue
|
||
# R2/R3（逐 KPI）：期间须为规整月且落在窗口（当月往前 6 个自然月，含当月）
|
||
if not period_in_window(p):
|
||
skipped_out_of_window += 1
|
||
logger.info(" 差异/趋势预警跳过（R2/R3，窗口下限 %s）: %s[%s]",
|
||
window_lower_period(), kpi.kpi_name, p)
|
||
continue
|
||
periods_used[p] = periods_used.get(p, 0) + 1
|
||
|
||
# 1. 差异预警：实际 vs 预算（统一逻辑）
|
||
result = build_deviation_alert(db_session, kpi, p)
|
||
if result["triggered"]:
|
||
deviation = result["deviation"]
|
||
alert_msg = (
|
||
f"{kpi.kpi_name}[{p}] 差异预警: 实际{deviation['actual_value']} "
|
||
f"vs 预算{deviation['budget_value']}，"
|
||
f"差异率{deviation['deviation_rate']}%"
|
||
)
|
||
alert_type = result["alert_type"] or "actual"
|
||
|
||
# R5-补1（裁定 2026-09-19）：期间可归期 + 必须落 kpi_value_id 关联
|
||
#（文案里的期间不构成可归期；无 kpi_values 行 → 跳过不入库）
|
||
linked_kv, _skip_reason = resolve_alert_kpi_value(db_session, kpi, p)
|
||
if linked_kv is None:
|
||
skipped_period_linkage += 1
|
||
skipped_linkage_kpis.append("%s[%s]" % (kpi.kpi_code, p))
|
||
else:
|
||
# 去重键（R5-补1 变更）：(entity_id, kpi_id, kpi_value_id, alert_type)
|
||
#（原按 `alert_message contains period` 判断，改关联后按可归期目标去重）
|
||
existing = db_session.query(KPIAlert).filter(
|
||
KPIAlert.entity_id == (kpi.entity_id or 1),
|
||
KPIAlert.kpi_id == kpi.id,
|
||
KPIAlert.kpi_value_id == linked_kv.id,
|
||
KPIAlert.alert_type == alert_type,
|
||
KPIAlert.status.in_(["pending", "processing"]),
|
||
).first()
|
||
|
||
if not existing:
|
||
alert = KPIAlert(
|
||
entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属（原缺省=1 串账套）
|
||
kpi_id=kpi.id,
|
||
kpi_value_id=linked_kv.id, # R5-补1: 期间经该行可派生（可归期）
|
||
alert_level=result["kpi_alert_level"],
|
||
alert_message=f"[差异预警] {alert_msg}",
|
||
alert_type=alert_type,
|
||
suggestion=result["suggestion"],
|
||
status="pending",
|
||
)
|
||
db_session.add(alert)
|
||
new_count += 1
|
||
logger.info(f" 新增差异预警 [{result['kpi_alert_level']}] {kpi.kpi_name}: 差异率{deviation['deviation_rate']}%")
|
||
|
||
# 2. 趋势异常检测（每期检查连续3期；末端期 = 本 KPI 的回落期间 p）
|
||
trend = check_trend_anomaly(db_session, kpi.id, p, consecutive=3)
|
||
if trend.get("anomaly") and trend.get("level") in ("yellow", "red"):
|
||
trend_alert_msg = trend["message"]
|
||
# R5-补1：趋势类以「计算基准期（趋势末端当期）」为关联期间；该期无行 → 跳过
|
||
trend_periods = trend.get("periods") or []
|
||
base_period = trend_periods[-1] if trend_periods else p
|
||
linked_kv, _skip_reason = resolve_alert_kpi_value(db_session, kpi, base_period)
|
||
if linked_kv is None:
|
||
skipped_period_linkage += 1
|
||
skipped_linkage_kpis.append("%s[%s]" % (kpi.kpi_code, base_period))
|
||
else:
|
||
existing_trend = db_session.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == kpi.id,
|
||
KPIAlert.alert_message.contains("[趋势预警]"),
|
||
KPIAlert.status.in_(["pending", "processing"]),
|
||
).first()
|
||
|
||
if not existing_trend:
|
||
alert = KPIAlert(
|
||
entity_id=kpi.entity_id or 1, # 2026-09-13 F1: 显式写归属
|
||
kpi_id=kpi.id,
|
||
kpi_value_id=linked_kv.id, # R5-补1: 趋势末端当期关联
|
||
alert_level=trend["level"],
|
||
alert_message=f"[趋势预警] {trend_alert_msg}",
|
||
status="pending",
|
||
)
|
||
db_session.add(alert)
|
||
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)
|
||
logger.info(" 差异/趋势预警期间口径: %s；判定期间分布=%s；跳过（无可用期间）%s 项 / （R2-R3 窗口外）%s 项 / （R5-补1 无 kpi_values 行）%s 项",
|
||
("固定 %s" % fixed_period) if fixed_period else "各 KPI 最近可用期间（裁定2b）",
|
||
periods_used, skipped_no_period, skipped_out_of_window, skipped_period_linkage)
|
||
if skipped_period_linkage:
|
||
logger.info(" 差异/趋势预警跳过（R5-补1 期间不可归期或无 kpi_values 行）KPI %s 项: %s",
|
||
skipped_period_linkage, skipped_linkage_kpis)
|
||
db_session.commit()
|
||
return new_count
|
||
|
||
|
||
if __name__ == "__main__":
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
)
|
||
db = get_session_local()()
|
||
try:
|
||
n = run_deviation_check(db)
|
||
print(f"差异预警检查完成: 新增 {n} 条")
|
||
finally:
|
||
db.close()
|