fix(cma): cashflow statements importer + auto_verify/alert_generator observability

- add backend/scripts/import_cashflow_statements.py: load zxt cashflow stat (CNY) into cashflow_statements in 10k CNY, dry-run default, --apply gated

- auto_verify_cron: per-reason skip counters + legacy-condition rule now resolves kpi_code via plan.kpi_id (was unreachable) + --dry-run

- alert_generator: coverage self-check warns when current-month KPI values are absent (0 alerts != no trigger)

task: finance-20260916-cma-health-gaps-3items (items 1/2/3)
This commit is contained in:
Hermes CI Fix
2026-09-16 23:46:48 +08:00
parent 27d3e1cd73
commit 9b3a131070
3 changed files with 275 additions and 21 deletions
+47 -2
View File
@@ -217,6 +217,38 @@ def generate_and_push(db_session) -> dict:
new_count = run_alert_check(db_session, period)
logger.info(f"预警检查完成: 新增 {new_count} 条")
# 1a. 覆盖度自检(2026-09-16 新增,任务 finance-20260916-cma-health-gaps-3items · 项3):
# run_alert_check 的期间口径 = 当月(period=current month),若当月尚无任何 KPIValue,
# 则「0 条新预警」并不代表"确无阈值触发",而是"本次检查未覆盖任何数据"——
# 该区别原实现不可见(日志与汇总行都只说 0 条),导致连续多日 0 新增无法归因。
try:
kpis_with_values = 0
active_kpis = db_session.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
for k in active_kpis:
if not any((k.threshold_red, k.threshold_yellow, k.threshold_green)):
continue
hit = db_session.query(KPIValue).filter(
KPIValue.kpi_id == k.id, KPIValue.period == period,
KPIValue.actual_value.isnot(None),
).first()
if hit:
kpis_with_values += 1
coverage = {"period": period, "kpis_with_threshold_and_value": kpis_with_values}
if kpis_with_values == 0:
latest_period = db_session.query(KPIValue.period).filter(
KPIValue.actual_value.isnot(None),
KPIValue.period.op("REGEXP")("^[0-9]{4}-[0-9]{2}$"), # 只取 YYYY-MM 月期间(排除 2026-H1/H2 伪期间)
).order_by(KPIValue.period.desc()).first()
logger.warning(
f"覆盖度告警: 当月({period})无任何 KPI 实际值 → 本次预警检查未覆盖任何 KPI,"
f"「0 条新预警」= 无数据可判,非「确无阈值触发」。库内最新期间="
f"{latest_period[0] if latest_period else 'N/A'}"
)
coverage["latest_period_in_db"] = latest_period[0] if latest_period else None
except Exception as e:
logger.error(f"覆盖度自检失败: {e}")
coverage = {"error": str(e)}
# 1b. 规则引擎检查(预测偏差 / 客户集中度,P2 2026-09-10)
rule_stats = run_rule_engine_checks(db_session)
@@ -231,7 +263,20 @@ def generate_and_push(db_session) -> dict:
logger.error(f"推送失败: {e}")
return {"period": period, "new_alerts": new_count, "pushed": pushed,
"cleaned_green": cleaned, "rule_engine": rule_stats}
"cleaned_green": cleaned, "rule_engine": rule_stats,
"coverage": coverage}
def coverage_note(result: dict) -> str:
"""汇总行追加提示:当月无 KPI 值时显式说明「0 条 = 无数据可判」而非「确无触发」
2026-09-16 新增(项3)。独立成函数便于回查(无需触发真实写路径即可断言文案)。
"""
cov = result.get("coverage") or {}
if cov.get("kpis_with_threshold_and_value") == 0:
return (f" | ⚠️ 当月({result['period']})无KPI值: 本次检查未覆盖任何KPI"
f"(库内最新月期间 {cov.get('latest_period_in_db')})→ 0 条不等于确无触发")
return ""
if __name__ == "__main__":
@@ -244,6 +289,6 @@ if __name__ == "__main__":
try:
result = generate_and_push(db)
print(f"预警检查: {result['new_alerts']}条新预警 / {result['pushed']}条已推送"
f" / 规则引擎{result['rule_engine']}")
f" / 规则引擎{result['rule_engine']}{coverage_note(result)}")
finally:
db.close()
+65 -19
View File
@@ -27,7 +27,19 @@ def evaluate_condition(condition, actual, target):
return func(actual, target)
def run_auto_verify():
def run_auto_verify(dry_run=False):
"""扫描到期pending计划,按 auto_verify_rule 自动验证执行效果。
2026-09-16 修复(任务 finance-20260916-cma-health-gaps-3items · 项2):
① 统计口径透明化——原汇总行「共处理N个到期计划」把「扫描到但全部 continue 跳过」的
计划也算作"处理",导致连续 36 天「验证通过0个」无法区分空转与失效。现输出
逐原因计数(无规则 / 无KPI代码 / KPI定义未匹配 / 无KPI值 / 实际评估)。
② 旧表达式规则可达——原 `if not kpi_code: continue` 位于 evaluate_condition 的
旧表达式兼容分支之前,使 rule={"condition":"value < 75"}(无 kpi_code,仅挂
plan.kpi_id)永远无法验证,该兼容分支成死代码。现先用 plan.kpi_id 反查
kpi_code 再判定(只用于本轮取数,不回写 DB)。
③ 新增 --dry-run:只读复算、不写库(供巡检/复核留痕用)。
"""
db = next(get_db())
try:
now = datetime.now()
@@ -41,14 +53,26 @@ def run_auto_verify():
verified_count = 0
escalated_count = 0
messages = []
# 跳过原因计数(2026-09-16 新增,用于区分「确无对象」与「判据失效」)
skip = {"no_rule": 0, "no_kpi_code": 0, "kpi_def_missing": 0, "no_kpi_value": 0}
evaluated = 0
for plan in plans:
rule = plan.auto_verify_rule
if not rule or not isinstance(rule, dict):
skip["no_rule"] += 1
continue
kpi_code = rule.get("kpi_code")
if not kpi_code and plan.kpi_id:
# 旧表达式规则(只有 condition,无 kpi_code)→ 用计划关联 KPI 反查
link_kpi = db.query(KPIDefinition).filter(KPIDefinition.id == plan.kpi_id).first()
if link_kpi and link_kpi.kpi_code:
kpi_code = link_kpi.kpi_code
rule = dict(rule)
rule["kpi_code"] = kpi_code
if not kpi_code:
skip["no_kpi_code"] += 1
continue
# 查最新KPI值
@@ -63,6 +87,7 @@ def run_auto_verify():
q = q.filter(KPIDefinition.entity_id == entity_id)
kpi = q.order_by(KPIDefinition.id.desc()).first()
if not kpi:
skip["kpi_def_missing"] += 1
continue
# 缺陷3残留修复(对齐 verify.py):KPIValue 按 period <= 当前月过滤,跨月验证不取未来期间
period_limit = now.strftime("%Y-%m")
@@ -72,6 +97,7 @@ def run_auto_verify():
KPIValue.period <= period_limit,
).order_by(KPIValue.period.desc(), KPIValue.calculated_at.desc(), KPIValue.id.desc()).first()
if not latest or latest.actual_value is None:
skip["no_kpi_value"] += 1
continue
actual = latest.actual_value
@@ -80,43 +106,58 @@ def run_auto_verify():
retry_max = rule.get("retry_max", 3)
passed = evaluate_condition(condition, actual, target)
evaluated += 1
plan.kpi_current_before = rule.get("baseline_value")
plan.kpi_current_after = actual
if not dry_run:
plan.kpi_current_before = rule.get("baseline_value")
plan.kpi_current_after = actual
if passed:
# 缺陷1残留修复(对齐 verify.py):plan 已处于"验证通过"状态(passed + verified_at 非空)则跳过 OKR 累加,防重复累加
already_verified = bool(plan.verify_status == "passed" and plan.verified_at is not None)
plan.verify_status = "passed"
plan.verified_at = now
plan.status = "completed" # 缺陷4残留修复:"done" 不在枚举(pending/in_progress/completed/cancelled),改 completed
plan.progress = 100
# 阶段3: OKR progress更新
update_okr_progress(db, plan, +15, already_verified=already_verified)
if not dry_run:
plan.verify_status = "passed"
plan.verified_at = now
plan.status = "completed" # 缺陷4残留修复:"done" 不在枚举(pending/in_progress/completed/cancelled),改 completed
plan.progress = 100
# 阶段3: OKR progress更新
update_okr_progress(db, plan, +15, already_verified=already_verified)
messages.append(f"✅ 行动计划#{plan.id}验证通过: {plan.title} ({kpi_code}: {rule.get('baseline_value')}→{actual})")
verified_count += 1
else:
plan.verify_attempts = (plan.verify_attempts or 0) + 1
if plan.verify_attempts >= retry_max:
plan.verify_status = "escalated"
attempts = (plan.verify_attempts or 0) + (0 if dry_run else 1)
if not dry_run:
plan.verify_attempts = attempts
if attempts >= retry_max:
if not dry_run:
plan.verify_status = "escalated"
messages.append(f"🚨 行动计划#{plan.id}连续{retry_max}次验证失败,升级人工: {plan.title}")
escalated_count += 1
else:
plan.verify_status = "retrying"
messages.append(f"⚠️ 行动计划#{plan.id}验证失败(第{plan.verify_attempts}次),待重试: {plan.title}")
if not dry_run:
plan.verify_status = "retrying"
messages.append(f"⚠️ 行动计划#{plan.id}验证失败(第{attempts}次),待重试: {plan.title}")
db.commit()
if not dry_run:
db.commit()
# 汇总通知
if messages:
if messages and not dry_run:
summary = "\n".join(messages)
try:
send_wecom_message(f"📋 行动计划自动验证日报\n{summary}", title="Auto-Verify")
except Exception:
pass
print(f"[{now.strftime('%Y-%m-%d %H:%M')}] 验证通过{verified_count}个,升级{escalated_count}个,共处理{len(plans)}个到期计划")
return {"verified": verified_count, "escalated": escalated_count, "total": len(plans)}
# 汇总行:保留「验证通过N个」前缀(巡检脚本 cma-daily-healthcheck.py 按该字面量取数),
# 追加逐原因明细 —— 使「0 通过」可区分为「无验证对象」还是「判据失效」
print(f"[{now.strftime('%Y-%m-%d %H:%M')}] 验证通过{verified_count}个,升级{escalated_count}个,"
f"共处理{len(plans)}个到期计划 | 实际评估{evaluated}个;跳过明细: "
f"无规则{skip['no_rule']}个/无KPI代码{skip['no_kpi_code']}个/"
f"KPI定义未匹配{skip['kpi_def_missing']}个/无KPI值{skip['no_kpi_value']}个"
+ ("(dry-run 未写库)" if dry_run else ""))
return {"verified": verified_count, "escalated": escalated_count, "total": len(plans),
"evaluated": evaluated, "skip": skip, "dry_run": dry_run}
finally:
db.close()
@@ -142,4 +183,9 @@ def update_okr_progress(db, plan, delta, already_verified=False):
if __name__ == "__main__":
run_auto_verify()
import sys as _sys
_dry = "--dry-run" in _sys.argv
res = run_auto_verify(dry_run=_dry)
if _dry:
import json as _json
print(_json.dumps(res, ensure_ascii=False))
@@ -0,0 +1,163 @@
#!/usr/bin/env python3
"""现金流量表(三大活动净额)入库器 —— 酣客(智享通) 现金流量统计表 → cashflow_statements
背景(任务 finance-20260916-cma-health-gaps-3items · 项1):
cashflow_statements 表 0 行 = **无导入链路**(此前只有 API 读/写端点,无任何 loader;
上游 智享通 数据虽已落盘 /root/WB_WS/_data/zxt-hanke/cashflow/<period>/,但从未入库)。
数据源(权威口径 = 智享通「现金流量统计表」项目级汇总,单位:元):
/root/WB_WS/_data/zxt-hanke/cashflow/<YYYY.MM>/cashflow_stat.csv
列:分类,编码,项目名称,方向,金额 分类∈{经营活动,投资活动,筹资活动}
兼容 2026.08 的另存名 cashflow_stat_<YYYY.MM>.csv
口径:
- 经营/投资/筹资 净额 = Σ(流入) − Σ(流出),单位 元 → 万元(÷10000,保留 2 位)
- 某活动无项目行 → 记 0.00(表列为 NOT NULL;同时写 remark 说明"无该项目=0")
- source='official'(官方报表口径);期初/期末现金 留空(主表采集缺陷,见报告 §项1-②)
- 同 (entity_id, period) 重复执行 = 覆盖更新(幂等)
用法:
python3 scripts/import_cashflow_statements.py # dry-run(默认,只算不写)
python3 scripts/import_cashflow_statements.py --period 2026-07 # 单期 dry-run
python3 scripts/import_cashflow_statements.py --apply # 落库(需放行)
"""
import argparse
import csv
import glob
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.database import get_session_local
from app.models import CashFlowStructureValue
from app.api.cashflow_structure import classify_enterprise
ZKT_BASE = "/root/WB_WS/_data/zxt-hanke/cashflow"
CLASSES = ("经营活动", "投资活动", "筹资活动")
def _read_stat_file(path):
"""读项目级汇总 CSV → [(分类, 编码, 项目名称, 方向, 金额元)]"""
rows = []
with open(path, encoding="utf-8-sig") as fh:
for r in csv.reader(fh):
if len(r) < 5 or not r[0].strip() or r[1].strip() == "编码":
continue
try:
amt = float(r[4].strip().replace(",", ""))
except ValueError:
continue
rows.append((r[0].strip(), r[1].strip(), r[2].strip(), r[3].strip(), amt))
return rows
def _nets(rows):
agg = {c: {"in": 0.0, "out": 0.0} for c in CLASSES}
for cls, _code, _name, direction, amt in rows:
if cls not in agg:
continue
agg[cls]["in" if "流入" in direction else "out"] += amt
return {c: round(v["in"] - v["out"], 2) for c, v in agg.items()}, agg
def collect(period_filter=None):
"""扫描盘上各期 → [payload dict](万元,未写库)"""
payloads = []
for d in sorted(glob.glob(os.path.join(ZKT_BASE, "2026.*"))):
period = os.path.basename(d).replace(".", "-")
if period_filter and period != period_filter:
continue
cands = sorted(glob.glob(os.path.join(d, "cashflow_stat*.csv")))
# 优先取当期明细名(cashflow_stat_<period>.csv),其次通用名
src = None
for c in cands:
if os.path.basename(c) != "cashflow_stat.csv":
src = c
src = src or (cands[0] if cands else None)
if not src:
payloads.append({"period": period, "status": "SKIP_no_source_file", "dir": d})
continue
rows = _read_stat_file(src)
if not rows:
payloads.append({"period": period, "status": "SKIP_empty",
"src": src, "rows": 0})
continue
nets_yuan, agg = _nets(rows)
missing = [c for c in CLASSES if not agg[c]["in"] and not agg[c]["out"]]
remark = ("源: 智享通现金流量统计表 %s(单位元→万元 /10000);项目行 %d 条;缺项目活动: %s"
% (os.path.basename(src), len(rows), "、".join(missing) or "无"))
payloads.append({
"period": period,
"status": "READY",
"src": src,
"rows": len(rows),
"nets_yuan": nets_yuan,
"nets_wan": {k: round(v / 10000.0, 2) for k, v in nets_yuan.items()},
"missing_classes": missing,
"remark": remark,
})
return payloads
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--entity-id", type=int, default=1,
help="账套:1=酣客(白酒, 智享通通道) 2=博海(IT, 无凭证明细)")
ap.add_argument("--period", default=None, help="仅处理某期 YYYY-MM")
ap.add_argument("--apply", action="store_true", help="真正写库(缺省=dry-run 只算不写)")
args = ap.parse_args()
payloads = collect(args.period)
if not payloads:
print("未发现任何期数据源(%s/2026.*)" % ZKT_BASE)
return 2
mode = "APPLY" if args.apply else "DRY-RUN"
print("=== 现金流量表入库器 %s | entity_id=%d ===" % (mode, args.entity_id))
db = get_session_local()() if args.apply else None
written = 0
try:
for p in payloads:
if p["status"] != "READY":
print(" %-8s %s (%s)" % (p["period"], p["status"], p.get("src") or p.get("dir")))
continue
n = p["nets_wan"]
cls = classify_enterprise(n["经营活动"], n["投资活动"], n["筹资活动"])
print(" %-8s O=%8.2f I=%8.2f F=%8.2f 万 | 判定=%s | 源=%s"
% (p["period"], n["经营活动"], n["投资活动"], n["筹资活动"],
("类型%d %s" % (cls["type_no"], cls["name"])) if cls.get("classified")
else ("未判定: " + cls.get("reason", "")[:28]),
os.path.basename(p["src"])))
if not args.apply:
continue
rec = db.query(CashFlowStructureValue).filter(
CashFlowStructureValue.entity_id == args.entity_id,
CashFlowStructureValue.period == p["period"],
).first()
created = rec is None
if rec is None:
rec = CashFlowStructureValue(entity_id=args.entity_id, period=p["period"])
db.add(rec)
rec.operating_net = n["经营活动"]
rec.investing_net = n["投资活动"]
rec.financing_net = n["筹资活动"]
rec.source = "official"
rec.remark = p["remark"]
db.flush()
written += 1
print(" → %s (id=%s)" % ("入库" if created else "更新", rec.id))
if args.apply:
db.commit()
print("=== 已提交 %d 期 ===" % written)
else:
print("=== dry-run 结束:未写库(%d 期可入库,加 --apply 落库)===" % len(
[p for p in payloads if p["status"] == "READY"]))
finally:
if db is not None:
db.close()
return 0
if __name__ == "__main__":
sys.exit(main())