Hermes CI Fix
3beec14d96
R1: exclude status=demo entities from alert traversal (alert_generator / deviation_engine / kpi_bep_deriver / alert_rules / cash engine) R2: only generate alerts within the last 6 calendar months (incl. current) R3: only regular month periods (YYYY-MM) participate R4: decouple generation from push - empty notification_channels never blocks Single source of truth: app/utils/alert_entity_scope.py
417 lines
20 KiB
Python
417 lines
20 KiB
Python
"""保本点(盈亏平衡)派生KPI执行器 — 管理会计OS (cma-bep-kpi 2026-09-08)
|
||
|
||
按财务Bot权威口径（陕西酣客_保本点分析_2026年1-7月.md）自动计算并落库五个派生KPI：
|
||
F_BEP_REVENUE 保本销售额(万元) = 月均管理费 ÷ 边际贡献率 (账面口径,含折旧)
|
||
F_BEP_COVER 保本覆盖率(%) = 月均销售 ÷ 保本销售额 × 100 (正向: >=100绿 >=85黄 <85红)
|
||
F_BEP_GAP 保本缺口(%) = (保本销售额-月均销售) ÷ 保本销售额 × 100 (反向: <=0绿 <=15黄 >15红)
|
||
F_BEP_REVENUE_CASH 现金保本销售额(万元) = (月均管理费-月均折旧) ÷ 边际贡献率 (现金口径,剔除折旧560213,中性无阈值)
|
||
F_BEP_COVER_CASH 现金保本覆盖率(%) = 月均销售 ÷ 现金保本销售额 × 100 (正向: >=100绿 >=85黄 <85红)
|
||
|
||
口径（与财务Bot报告一致）：
|
||
- 销售收入: voucher_details.subject_code='5001'(主营业务收入) 贷方合计，当年1月至目标月 YTD 月均
|
||
- 管理费: subject_code LIKE '5602%'(管理费用，含560213折旧) 借方合计，YTD 月均 → 账面口径
|
||
- 折旧: subject_code = '560213'(管理费用-折旧叶子) 借方合计，YTD 月均 → 现金口径剔除项(eq精确匹配,防5602130误含)
|
||
- 渠补率: 厂家政策定死 84.61% (571103渠补支出/5001销售, 1-7月累计凭证验证≈84.0%~84.6%)
|
||
- 边际贡献率 = 1 - 渠补率 = 15.39%
|
||
- 目标月: --period 显式指定；默认自动选最近「已结账」完整月（当月跳过）。
|
||
结账信号优先：DB 损益结转凭证(carry_forward=1 & subject_code='3103') → check.md 文件；
|
||
两者皆无时才退回旧「销售额阈值」兜底（销售低 ≠ 未结账，仅保底用，日志标注依据）
|
||
- 单位: voucher 金额为元，聚合后 /10000 → 万元
|
||
- 幂等: 同 kpi_id+period+source_type='derived' 存在则更新，可安全重跑
|
||
- 只写 source_type='derived' 行；不触碰 auto_collect/manual/excel/ledger 行；不改 collector 逻辑
|
||
|
||
用法:
|
||
venv/bin/python3 scripts/kpi_bep_deriver.py # 自动选目标月
|
||
venv/bin/python3 scripts/kpi_bep_deriver.py --period 2026-07 # 指定期间(实跑验证用)
|
||
venv/bin/python3 scripts/kpi_bep_deriver.py --entity 1 --rebate-rate 84.61
|
||
venv/bin/python3 scripts/kpi_bep_deriver.py --dry-run # 只计算不落库
|
||
"""
|
||
import argparse
|
||
import logging
|
||
import os
|
||
import sys
|
||
from datetime import datetime, date
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
import re
|
||
from sqlalchemy import func
|
||
from app.database import get_session_local
|
||
from app.models import KPIValue, KPIDefinition, VoucherDetail, KPIAlert
|
||
# 裁定 R1（2026-09-19 财务口径）：demo 账套不产生业务预警（派生值仍落库，仅跳过预警同步）
|
||
from app.utils.alert_entity_scope import is_excluded_entity
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
logger = logging.getLogger("cma.bep_deriver")
|
||
|
||
# 厂家政策渠补率（定死不可谈判，财务Bot报告权威值 2026-09-08）
|
||
POLICY_REBATE_RATE_PCT = 84.61
|
||
|
||
# 派生KPI 编码 → (kpi_code, 名称)
|
||
TARGET_KPIS = [
|
||
("F_BEP_REVENUE", "保本销售额"),
|
||
("F_BEP_COVER", "保本覆盖率"),
|
||
("F_BEP_GAP", "保本缺口"),
|
||
("F_BEP_REVENUE_CASH", "现金口径保本销售额"),
|
||
("F_BEP_COVER_CASH", "现金口径保本覆盖率"),
|
||
]
|
||
|
||
# 现金口径 KPI 编码（计算路径独立于账面，remark 文案用现金口径）
|
||
CASH_CODES = {"F_BEP_REVENUE_CASH", "F_BEP_COVER_CASH"}
|
||
|
||
SALES_CODE = "5001" # 主营业务收入
|
||
ADMIN_PREFIX = "5602" # 管理费用（含折旧叶子560213）
|
||
DEP_SUBJECT = "560213" # 管理费用-折旧（现金口径剔除项，eq 精确匹配防 5602130 误含）
|
||
REBATE_CODE = "571103" # 渠补/捐赠支出（校验用，不直接进公式）
|
||
CARRY_SUBJECT = "3103" # 本年利润（carry_forward=1 且该科目 = 该期损益结转凭证）
|
||
|
||
# 智享通落盘目录（与 app/api/data.py 的 ZXT_LANDING_DIR 一致）——结账信号次选来源
|
||
ZXT_LANDING_DIR = "/root/WB_WS/_data/zxt-hanke"
|
||
|
||
|
||
def _month_range(period: str):
|
||
"""返回该期间当年1月1日到月末的日期边界（用于 voucher_date 过滤兜底）"""
|
||
y = int(period[:4])
|
||
m = int(period[5:7])
|
||
start = date(y, 1, 1)
|
||
# 月末
|
||
if m == 12:
|
||
end = date(y, 12, 31)
|
||
else:
|
||
end = date(y, m + 1, 1).replace(day=1) # 次月1日
|
||
from datetime import timedelta
|
||
end = end - timedelta(days=1)
|
||
return start, end
|
||
|
||
|
||
def _sum_field(db, entity_id: int, period_from: str, period_to: str,
|
||
field: str, subject_filter=None) -> float:
|
||
"""聚合 voucher_details 金额（元）。subject_filter: ('eq', code) / ('like', prefix)"""
|
||
q = db.query(func.sum(getattr(VoucherDetail, field))).filter(
|
||
VoucherDetail.entity_id == entity_id,
|
||
VoucherDetail.carry_forward == 0,
|
||
VoucherDetail.period >= period_from,
|
||
VoucherDetail.period <= period_to,
|
||
)
|
||
if subject_filter:
|
||
kind, code = subject_filter
|
||
if kind == "eq":
|
||
q = q.filter(VoucherDetail.subject_code == code)
|
||
elif kind == "like":
|
||
q = q.filter(VoucherDetail.subject_code.like(code + "%"))
|
||
val = q.scalar()
|
||
return float(val or 0.0)
|
||
|
||
|
||
def _month_sales(db, entity_id: int, period: str) -> float:
|
||
"""单月销售收入（万元）——用于目标月完整度判断"""
|
||
return _sum_field(db, entity_id, period, period, "credit_amount",
|
||
("eq", SALES_CODE)) / 10000.0
|
||
|
||
|
||
def _is_period_closed(db, entity_id: int, period: str) -> tuple:
|
||
"""判断该期间是否已结账（损益结转已完成）。返回 (bool, 依据说明)。
|
||
|
||
主信号（DB）: voucher_details 存在 carry_forward=1 且 subject_code='3103'(本年利润) 的行
|
||
= 该期损益结转凭证已生成
|
||
次信号（文件）: ZXT_LANDING_DIR/{YYYY.MM}/check.md 同时含「本期: PASS」与「损益结转: PASS」
|
||
均不成立 → (False, "")。任何读取异常均吞掉，不影响主流程。
|
||
"""
|
||
# 主信号：DB 损益结转凭证
|
||
try:
|
||
hit = db.query(VoucherDetail.id).filter(
|
||
VoucherDetail.entity_id == entity_id,
|
||
VoucherDetail.carry_forward == 1,
|
||
VoucherDetail.subject_code == CARRY_SUBJECT,
|
||
VoucherDetail.period == period,
|
||
).first()
|
||
if hit:
|
||
return True, "DB损益结转"
|
||
except Exception as e: # 查询异常不阻断，降级到文件信号
|
||
logger.warning("结账信号(DB)查询失败 period=%s: %s", period, e)
|
||
|
||
# 次信号：智享通落盘 check.md
|
||
try:
|
||
path = os.path.join(ZXT_LANDING_DIR, period.replace("-", "."), "check.md")
|
||
if os.path.exists(path):
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
txt = f.read()
|
||
if "本期: PASS" in txt and "损益结转: PASS" in txt:
|
||
return True, "check.md"
|
||
except Exception:
|
||
pass # 文件缺失/读取失败一律忽略
|
||
|
||
return False, ""
|
||
|
||
|
||
def _pick_target_period(db, entity_id: int, now: date) -> str:
|
||
"""自动选择目标月：跳过当月。
|
||
|
||
判定顺序（2026-09-12 修订）:
|
||
1. 结账信号优先 —— 按 desc 返回第一个 _is_period_closed 为真的月；
|
||
结账成立时不再用销售额做任何筛选（销售低 ≠ 未结账，历史缺陷根因）
|
||
2. 所有候选月均无结账信号 → 退回旧「销售额 < 近6月均×30%」兜底（日志标注『依据销售额兜底』）
|
||
"""
|
||
# 收集已有数据月份（5001 贷方）
|
||
rows = db.query(VoucherDetail.period).filter(
|
||
VoucherDetail.entity_id == entity_id,
|
||
VoucherDetail.period.like("%-__"),
|
||
).distinct().order_by(VoucherDetail.period.desc()).all()
|
||
periods = [r[0] for r in rows if r[0] and len(r[0]) == 7]
|
||
if not periods:
|
||
logger.warning("voucher_details 无 period 数据，无法自动选月")
|
||
return now.strftime("%Y-%m")
|
||
|
||
# 过滤掉未来/当月（当月未结束数据不全）
|
||
cur = now.strftime("%Y-%m")
|
||
candidates = [p for p in periods if p < cur]
|
||
if not candidates:
|
||
candidates = [p for p in periods if p <= cur]
|
||
if not candidates:
|
||
candidates = periods
|
||
|
||
# 1) 结账信号优先（DB损益结转 → check.md），命中即采用
|
||
for p in candidates:
|
||
closed, reason = _is_period_closed(db, entity_id, p)
|
||
if closed:
|
||
logger.info("目标月 %s 判定已结账（依据：%s），采用", p, reason)
|
||
return p
|
||
|
||
# 2) 兜底：无任何结账信号时，退回旧销售额阈值逻辑
|
||
logger.warning("候选月 %s 均无结账信号(DB损益结转/check.md)，退回销售额阈值兜底",
|
||
", ".join(candidates[:6]))
|
||
|
||
# 参考月均：候选月中最近 6 个不同月的月销售
|
||
ref_months = []
|
||
for p in periods:
|
||
if p >= cur:
|
||
continue
|
||
s = _month_sales(db, entity_id, p)
|
||
ref_months.append((p, s))
|
||
if len(ref_months) >= 6:
|
||
break
|
||
ref_avg = (sum(s for _, s in ref_months) / len(ref_months)) if ref_months else None
|
||
|
||
for p in candidates:
|
||
s = _month_sales(db, entity_id, p)
|
||
if s <= 0:
|
||
continue # 该月无销售，往前找
|
||
if ref_avg and s < ref_avg * 0.3:
|
||
logger.info("目标月 %s 销售 %.1f万 < 参考月均 %.1f万*30%%，依据销售额兜底判定未结账，回退",
|
||
p, s, ref_avg)
|
||
continue
|
||
logger.info("目标月 %s 依据销售额兜底采用", p)
|
||
return p
|
||
return candidates[0]
|
||
|
||
|
||
def _eval_static_threshold(value: float, kpi) -> tuple:
|
||
"""按 KPI 定义 threshold_green/yellow/red 判定级别（绿→黄→红，命中即止）。
|
||
与 alert_generator._check_threshold 语义一致。green=正常不落预警。"""
|
||
for level, field in (("green", "threshold_green"), ("yellow", "threshold_yellow"),
|
||
("red", "threshold_red")):
|
||
expr = getattr(kpi, field, None)
|
||
if not expr:
|
||
continue
|
||
m = re.match(r"(>=|<=|>|<|=|!=)\s*([\d.]+)", str(expr).strip())
|
||
if not m:
|
||
continue
|
||
op, val = m.group(1), float(m.group(2))
|
||
hit = {
|
||
">=": value >= val, "<=": value <= val,
|
||
">": value > val, "<": value < val,
|
||
"=": value == val, "!=": value != val,
|
||
}[op]
|
||
if hit:
|
||
level_names = {"green": "正常", "yellow": "预警", "red": "紧急"}
|
||
msg = (f"KPI当前值 {value:.2f}，触发{level_names[level]}阈值 ({expr})")
|
||
return level, msg
|
||
return None, None
|
||
|
||
|
||
def run_derive(db, entity_id: int, period: str, rebate_rate_pct: float,
|
||
write_kpi: bool = True) -> dict:
|
||
"""计算并落库五个派生KPI（账面3条+现金2条）。返回结果摘要。"""
|
||
kpis = {k.kpi_code: k for k in db.query(KPIDefinition).filter(
|
||
KPIDefinition.entity_id == entity_id,
|
||
KPIDefinition.kpi_code.in_([c for c, _ in TARGET_KPIS]),
|
||
KPIDefinition.status == "active",
|
||
).all()}
|
||
missing = [c for c, _ in TARGET_KPIS if c not in kpis]
|
||
if missing:
|
||
raise RuntimeError(f"entity={entity_id} 缺少派生KPI定义: {missing}")
|
||
|
||
year = period[:4]
|
||
period_from = f"{year}-01"
|
||
period_to = period
|
||
months = int(period[5:7]) # YTD 月数（1月=1 ... 7月=7）
|
||
|
||
# 1. 取数（元 → 万元）
|
||
sales_sum_w = _sum_field(db, entity_id, period_from, period_to, "credit_amount", ("eq", SALES_CODE)) / 10000.0
|
||
admin_sum_w = _sum_field(db, entity_id, period_from, period_to, "debit_amount", ("like", ADMIN_PREFIX)) / 10000.0
|
||
rebate_sum_w = _sum_field(db, entity_id, period_from, period_to, "debit_amount", ("eq", REBATE_CODE)) / 10000.0
|
||
dep_sum_w = _sum_field(db, entity_id, period_from, period_to, "debit_amount", ("eq", DEP_SUBJECT)) / 10000.0
|
||
|
||
sales_avg = sales_sum_w / months
|
||
admin_avg = admin_sum_w / months
|
||
dep_avg = dep_sum_w / months
|
||
cash_admin_avg = admin_avg - dep_avg # 现金口径固定成本（剔除折旧纸面费用）
|
||
|
||
# 2. 渠补率：厂家政策值优先，日志附凭证累计校验
|
||
calc_rebate = (rebate_sum_w / sales_sum_w * 100.0) if sales_sum_w > 0 else 0.0
|
||
if rebate_rate_pct is None:
|
||
rebate_rate_pct = POLICY_REBATE_RATE_PCT
|
||
cm_pct = 100.0 - rebate_rate_pct # 边际贡献率(%)
|
||
|
||
# 3. 派生计算（万元 / %）——账面3条路径不变；现金口径在账面基础上剔除折旧
|
||
bep = admin_avg / (cm_pct / 100.0) if cm_pct > 0 else 0.0
|
||
cover = (sales_avg / bep * 100.0) if bep > 0 else 0.0
|
||
gap = ((bep - sales_avg) / bep * 100.0) if bep > 0 else 0.0
|
||
cash_bep = cash_admin_avg / (cm_pct / 100.0) if cm_pct > 0 else 0.0
|
||
cash_cover = (sales_avg / cash_bep * 100.0) if cash_bep > 0 else 0.0
|
||
|
||
values = {
|
||
"F_BEP_REVENUE": round(bep, 1),
|
||
"F_BEP_COVER": round(cover, 1),
|
||
"F_BEP_GAP": round(gap, 1),
|
||
"F_BEP_REVENUE_CASH": round(cash_bep, 1),
|
||
"F_BEP_COVER_CASH": round(cash_cover, 1),
|
||
}
|
||
|
||
# 4. upsert kpi_values (source_type='derived'，幂等)
|
||
batch = f"bep-{datetime.now().strftime('%Y%m%d%H%M%S')}"
|
||
written = {}
|
||
value_rows = {}
|
||
for code, _ in TARGET_KPIS:
|
||
kpi = kpis[code]
|
||
if code in CASH_CODES:
|
||
remark = (f"现金口径: 管理费月均{admin_avg:.2f}万-折旧月均{dep_avg:.2f}万"
|
||
f"=固定现金{cash_admin_avg:.2f}万 ÷ 边际贡献率{cm_pct:.2f}% → 保本{cash_bep:.2f}万")
|
||
else:
|
||
remark = f"保本派生器: 销售月均{sales_avg:.1f}万/管理费月均{admin_avg:.1f}万/渠补率{rebate_rate_pct:.2f}%"
|
||
existing = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi.id,
|
||
KPIValue.period == period,
|
||
KPIValue.source_type == "derived",
|
||
).first()
|
||
if write_kpi:
|
||
is_new = existing is None
|
||
if existing:
|
||
existing.actual_value = values[code]
|
||
existing.source_batch = batch
|
||
existing.remark = remark
|
||
existing.data_status = "pending"
|
||
else:
|
||
existing = KPIValue(
|
||
entity_id=entity_id,
|
||
kpi_id=kpi.id,
|
||
period=period,
|
||
actual_value=values[code],
|
||
source_type="derived",
|
||
source_batch=batch,
|
||
data_status="pending",
|
||
remark=remark,
|
||
)
|
||
db.add(existing)
|
||
db.flush()
|
||
value_rows[kpi.id] = existing
|
||
written[code] = "insert" if is_new else "upsert"
|
||
else:
|
||
written[code] = "dry-run"
|
||
if write_kpi:
|
||
# 5. 同步预警: 仅对本派生KPI评估 threshold（green=正常不落 pending，yellow/red 落 pending）。
|
||
# 幂等规则：同一 KPIValue 已有同级别 pending → 保持不动（值未变不重建）；
|
||
# 其它/旧级别 pending → resolve；最新派生行永远只留一条当前级别。
|
||
# R1（裁定 2026-09-19）：demo 账套不产生业务预警 → 整段预警同步跳过（派生值照落库）。
|
||
target_sync = [] if is_excluded_entity(db, entity_id) else TARGET_KPIS
|
||
if not target_sync:
|
||
logger.info("demo 账套（entity=%s）跳过预警同步（R1，裁定 2026-09-19）", entity_id)
|
||
for code, _ in target_sync:
|
||
kpi = kpis[code]
|
||
kpi_value = value_rows.get(kpi.id)
|
||
if not kpi_value:
|
||
continue
|
||
level, msg = _eval_static_threshold(kpi_value.actual_value, kpi)
|
||
all_pending = db.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == kpi.id,
|
||
KPIAlert.status == "pending",
|
||
).all()
|
||
same_value = [a for a in all_pending if a.kpi_value_id == kpi_value.id]
|
||
if level in ("yellow", "red") and any(a.alert_level == level for a in same_value):
|
||
# 当前值级别未变 → 只清其它旧值预警，不重建
|
||
for a in all_pending:
|
||
if a.kpi_value_id != kpi_value.id:
|
||
a.status = "resolved"
|
||
a.resolution = "保本派生器: 派生值更新，旧预警自动解除"
|
||
a.resolved_at = datetime.now()
|
||
continue
|
||
for a in all_pending:
|
||
a.status = "resolved"
|
||
a.resolution = "保本派生器: 派生值更新/级别变更，旧预警自动解除"
|
||
a.resolved_at = datetime.now()
|
||
if level in ("yellow", "red"):
|
||
db.add(KPIAlert(
|
||
entity_id=entity_id,
|
||
kpi_id=kpi.id,
|
||
kpi_value_id=kpi_value.id,
|
||
alert_level=level,
|
||
alert_message=f"{kpi.kpi_name}[{period}] {msg}",
|
||
status="pending",
|
||
))
|
||
db.commit()
|
||
|
||
return {
|
||
"entity_id": entity_id,
|
||
"period": period,
|
||
"months": months,
|
||
"sales_sum_w": round(sales_sum_w, 2),
|
||
"sales_avg": round(sales_avg, 2),
|
||
"admin_sum_w": round(admin_sum_w, 2),
|
||
"admin_avg": round(admin_avg, 2),
|
||
"dep_sum_w": round(dep_sum_w, 2),
|
||
"dep_avg": round(dep_avg, 2),
|
||
"cash_admin_avg": round(cash_admin_avg, 2),
|
||
"rebate_sum_w": round(rebate_sum_w, 2),
|
||
"rebate_policy_pct": rebate_rate_pct,
|
||
"rebate_calc_pct": round(calc_rebate, 2),
|
||
"cm_pct": round(cm_pct, 2),
|
||
"values": values,
|
||
"written": written,
|
||
}
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="保本点派生KPI执行器")
|
||
ap.add_argument("--period", default=None, help="目标期间 YYYY-MM（默认自动选最近完整月）")
|
||
ap.add_argument("--entity", type=int, default=1, help="entity_id（默认1=陕西酣客）")
|
||
ap.add_argument("--rebate-rate", type=float, default=None,
|
||
help="渠补率%覆盖（默认厂家政策84.61）")
|
||
ap.add_argument("--dry-run", action="store_true", help="只计算不落库")
|
||
args = ap.parse_args()
|
||
|
||
db = get_session_local()()
|
||
try:
|
||
period = args.period or _pick_target_period(db, args.entity, date.today())
|
||
r = run_derive(db, args.entity, period, args.rebate_rate,
|
||
write_kpi=not args.dry_run)
|
||
print(f"保本点派生计算完成 entity={r['entity_id']} period={r['period']} "
|
||
f"(YTD {r['months']}个月) mode={'dry-run' if args.dry_run else 'write'}")
|
||
print(f" 销售收入累计 {r['sales_sum_w']:.2f}万 月均 {r['sales_avg']:.2f}万")
|
||
print(f" 管理费累计 {r['admin_sum_w']:.2f}万 月均 {r['admin_avg']:.2f}万 (含折旧,账面口径)")
|
||
print(f" 折旧累计 {r['dep_sum_w']:.2f}万 月均 {r['dep_avg']:.2f}万 (560213) | "
|
||
f"现金固定成本月均 {r['cash_admin_avg']:.2f}万 (管理费-折旧)")
|
||
print(f" 渠补支出累计 {r['rebate_sum_w']:.2f}万 | 渠补率: 政策{r['rebate_policy_pct']:.2f}% "
|
||
f"(凭证累计校验 {r['rebate_calc_pct']:.2f}%) | 边际贡献率 {r['cm_pct']:.2f}%")
|
||
print(f" F_BEP_REVENUE 保本销售额 = {r['values']['F_BEP_REVENUE']} 万元")
|
||
print(f" F_BEP_COVER 保本覆盖率 = {r['values']['F_BEP_COVER']} %")
|
||
print(f" F_BEP_GAP 保本缺口 = {r['values']['F_BEP_GAP']} %")
|
||
print(f" F_BEP_REVENUE_CASH 现金保本销售额 = {r['values']['F_BEP_REVENUE_CASH']} 万元")
|
||
print(f" F_BEP_COVER_CASH 现金保本覆盖率 = {r['values']['F_BEP_COVER_CASH']} %")
|
||
print(f" 落库: {r['written']}")
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|