feat(kpi): CMA保本点派生KPI F_BEP_REVENUE/COVER/GAP + 每日派生执行器 + 红黄绿静态阈值判定

- kpi_definitions entity1 新增3条派生KPI(456/457/458, formula/threshold/unit齐全) - 新增 scripts/kpi_bep_deriver.py: 从 voucher_details 按财务口径聚合 (5001销售收入/5602管理费 YTD月均, 渠补率政策84.61%) → 计算保本销售额/ 覆盖率/缺口 → upsert kpi_values source_type='derived' 幂等 + 同步yellow/red预警 - cron 06:35 注册(紧随 collector 06:30); 自动选最近完整月(8月未结账自动回退7月) - kpis.py get_kpi: F_BEP_* 派生KPI按静态阈值文本判红黄绿(66.9→黄), 不走target比率 - 阈值经财务Bot 2026-09-08裁决: COVER >=100绿 50~100黄 <50红; GAP <=0绿 <=50黄 >50红
2026-09-08 13:43:20 +08:00
parent ebe0962c09
commit 1a5356a2d1
4 changed files with 418 additions and 16 deletions
@@ -0,0 +1,322 @@
+"""保本点(盈亏平衡)派生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红)
+
+口径（与财务Bot报告一致）：
+- 销售收入: voucher_details.subject_code='5001'(主营业务收入) 贷方合计，当年1月至目标月 YTD 月均
+- 管理费:   subject_code LIKE '5602%'(管理费用，含560213折旧) 借方合计，YTD 月均 → 账面口径
+- 渠补率:   厂家政策定死 84.61% (571103渠补支出/5001销售, 1-7月累计凭证验证≈84.0%~84.6%)
+- 边际贡献率 = 1 - 渠补率 = 15.39%
+- 目标月:   --period 显式指定；默认自动选最近完整月（当月跳过；最近月销售额<近6月均30%视为未结账回退）
+- 单位:     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
+
+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", "保本缺口"),
+]
+
+SALES_CODE = "5001"          # 主营业务收入
+ADMIN_PREFIX = "5602"        # 管理费用（含折旧叶子560213）
+REBATE_CODE = "571103"       # 渠补/捐赠支出（校验用，不直接进公式）
+
+
+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 _pick_target_period(db, entity_id: int, now: date) -> str:
+    """自动选择目标月：跳过当月；最近月销售额<近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
+
+    # 参考月均：候选月中最近 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
+        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。返回结果摘要。"""
+    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
+
+    sales_avg = sales_sum_w / months
+    admin_avg = admin_sum_w / months
+
+    # 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. 派生计算（万元 / %）
+    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
+
+    values = {
+        "F_BEP_REVENUE": round(bep, 1),
+        "F_BEP_COVER": round(cover, 1),
+        "F_BEP_GAP": round(gap, 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]
+        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 = f"保本派生器: 销售月均{sales_avg:.1f}万/管理费月均{admin_avg:.1f}万/渠补率{rebate_rate_pct:.2f}%"
+                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=f"保本派生器: 销售月均{sales_avg:.1f}万/管理费月均{admin_avg:.1f}万/渠补率{rebate_rate_pct:.2f}%",
+                )
+                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；最新派生行永远只留一条当前级别。
+        for code, _ in TARGET_KPIS:
+            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),
+        "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['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"  落库: {r['written']}")
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    main()