From 1613f7f96187f8d5d6429444a75f44a2e2e92171 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Sun, 4 Oct 2026 09:42:40 +0800 Subject: [PATCH] fix(cpa): sales bill-type whitelist for customer revenue caliber (P0) Customer revenue/concentration now filters bill_type IN (1,4,14) (XSD/XSTHD/XSHHD). Non-sales docs (JHD purchase, YFKD prepay, YSKD accept, QTRK, SKD, YBFYD) carry SUPPLIER unit_id and were polluting 'customer concentration' (old Top1/Top2 were our biggest suppliers; root cause of 6.04x base gap). Unify sign filter to (sum_money+sum_tax_money)>0 (equivalent to money_only on current data, diff 0 rows) and correct the withdrawn 407-row/17.2% wording in cpa_caliber docstring. Deriver: anchors from cpa_caliber single source; new --concentration overwrite and --only concentration (minimal 9-row write face) with pre-overwrite backup. Tests: fixture gains bill_type/sum_tax_money; +5 whitelist cases (44 passed targeted). --- backend/app/api/alert_rules.py | 18 ++- backend/app/utils/cpa_caliber.py | 120 ++++++++++----- .../kpi_customer_profitability_deriver.py | 145 ++++++++++++++---- .../tests/test_customer_concentration_rule.py | 94 +++++++++++- 4 files changed, 298 insertions(+), 79 deletions(-) diff --git a/backend/app/api/alert_rules.py b/backend/app/api/alert_rules.py index 68862729..f3fd47d6 100644 --- a/backend/app/api/alert_rules.py +++ b/backend/app/api/alert_rules.py @@ -812,7 +812,7 @@ def run_forecast_deviation_check( # ============================================================ def _customer_revenue_shares(db: Session, entity_id: int, lookback_months: int = 12, top_n: int = 1): - """聚合客户（往来单位）收入占比 — 数据源 zxt_masterbill（智享通销售单据, unit_id=客户/往来单位） + """聚合客户（往来单位）收入占比 — 数据源 zxt_masterbill（智享通**销售类**单据, unit_id=客户） 返回 (shares, total, period_range)： shares = [(unit_id, amount, pct), ...] 按金额倒序取前 top_n 名 @@ -820,9 +820,17 @@ def _customer_revenue_shares(db: Session, entity_id: int, lookback_months: int = period_range = (start_date, end_date) ISO 字符串 无可用数据时返回 ([], 0.0, None)。 - **口径单一事实源**：SQL 已抽到 `app/utils/cpa_caliber.py`（CPA 派生器共用同一实现， - 防「同一客户收入两份结果」）。行为与抽函数前逐字一致（同过滤、同排序、同四舍五入）。 - 用独立连接执行，避免表不存在/方言差异污染调用方 session 事务。 + **口径单一事实源**：SQL 在 `app/utils/cpa_caliber.py`（CPA 派生器共用同一实现， + 防「同一客户收入两份结果」）。用独立连接执行，避免表不存在/方言差异污染调用方 session 事务。 + + 2026-10-04 P0 口径纠错（任务 finance-20261004T0905-cpa-revenue-caliber-p0）： + ① 增列**单据类型白名单** `bill_type IN (1,4,14)`（销售单/退货/换货）——非销售单据 + （进货/验收/其它入库/预付款/收款/费用）的 `unit_id` 是**供应商**，混入会把 + 「客户集中度」度量成「往来单位集中度」（旧口径 Top1/Top2 恰是最大供应商）； + ② 取舍口径统一为 `(sum_money + sum_tax_money) > 0`（原 `money_only` 与之在本库 + 当前数据上**逐行同集、差集 0 行**，故行为**零变化**；原「17.2% 低估」表述 + 经财务复验证否，见 `cpa_caliber` 模块 docstring）。 + 预警阈值 50% 在新口径下仍不触发（Top1 22.52% < 50%）⇒ 阈值与分级不漂移。 """ from datetime import date, timedelta from app.database import get_engine @@ -831,7 +839,7 @@ def _customer_revenue_shares(db: Session, entity_id: int, lookback_months: int = end = date.today() start = end - timedelta(days=max(1, int(lookback_months)) * 30) shares, total, rng = cpa_caliber.fetch_revenue_rows( - get_engine(), entity_id, start, end, top_n, sign_basis="money_only") + get_engine(), entity_id, start, end, top_n) if not shares: return ([], 0.0, None) return (shares, total, rng) diff --git a/backend/app/utils/cpa_caliber.py b/backend/app/utils/cpa_caliber.py index 545aca40..edeabf21 100644 --- a/backend/app/utils/cpa_caliber.py +++ b/backend/app/utils/cpa_caliber.py @@ -1,4 +1,8 @@ -"""CPA（客户获利性）取数口径 —— 单一事实源（2026-10-04，任务 finance-20261003-cpa-customer-profitability-mvp-fullstack）。 +"""CPA（客户获利性）取数口径 —— 单一事实源（2026-10-04，任务 finance-20261003-cpa-customer-profitability-mvp-fullstack。 + +2026-10-04 P0 口径纠错（任务 finance-20261004T0905-cpa-revenue-caliber-p0，依据财务独立复验件 +`finance-20261004-cpa-mvp-independent-verify` §5.3）：增列**单据类型白名单**（只取销售类）， +避免把采购/资金/库存/费用单据的**供应商**当成客户。 为什么有本模块 -------------- @@ -9,15 +13,34 @@ 基座 ---- -`zxt_masterbill`（智享通销售单据，仅 entity1 有数据：15,755 行 / 284 个往来单位 / +`zxt_masterbill`（智享通单据流，仅 entity1 有数据：15,755 行 / 284 个往来单位 / 2021-03 ~ 2026-09-30）。**无** unit 名称列 ⇒ 客户标识只用 `unit_id`。 口径（勿擅改） -------------- -* `C_CUST_REVENUE` / 快照明细 `revenue` = Σ`zxt_masterbill.sum_money`（按 `unit_id`、期内、**含税**原值）。 - 过滤：`entity_id=? AND unit_id>0 AND sum_money>0 AND bill_date NOT NULL`（与预警路径逐字一致）。 - 这组过滤下的对拍锚点（2026-01~09）：73 客户 / 4,883.52 万元 / Top1=25.3835% / - Top2=23.4905% / Top4 累计=63.3249% / Top10 累计=81.7675%。 +* **单据类型白名单 = 销售类 `bill_type IN (1, 4, 14)`**（1=XSD 销售单 / 4=XSTHD 销售退货 / + 14=XSHHD 销售换货；等价判据 `bill_sn LIKE 'XSD%'/'XSTH%'/'XSHH%'` 实测逐行同集）。 + 非销售单据一律排除：0=JHD 进货 / 128=YSKD 验收 / 17=QTRK 其它入库 / 129=YFKD 预付款 / + 101=SKD 收款 / 103=YBFYD 费用 / 104=QTSRD 其它收入 / 18=QTCK 其它出库。 + **依据**：这些单据的 `unit_id` 是**供应商**（`abst` 实读「向【酣客酒业（广州）】进货」「预付款单」）， + 混入会把「客户集中度」变成「往来单位（含供应商）集中度」——旧口径 Top1/Top2 恰是本公司最大供应商， + 也是「4,883.52 万 ÷ 账套 5001 808.80 万 = 6.04× 基座差」的真因。 + ⚠️ 该纠错**改变对外数字**（户数 73→65、Top1 25.3835%→22.5150%、Top2 归属由供应商改真实客户）； + 属**事实性纠错**（财务定死 + 任总知会），非选型。 +* `C_CUST_REVENUE` / 快照明细 `revenue` = Σ`zxt_masterbill.sum_money`（按 `unit_id`、期内、**不含税原值**）。 + 过滤：`entity_id=? AND unit_id>0 AND bill_type IN (1,4,14) AND (sum_money+sum_tax_money)>0 + AND bill_date IS NOT NULL`。 +* **取舍口径（符号过滤）**：`(sum_money + sum_tax_money) > 0`（`tax_incl`）与 `sum_money > 0` + （`money_only`）在本库当前数据上**完全等价**（2026-01~09 双向差集 **0 行**；全表亦 0 行）。 + 原报告「407 行红冲（含税为正、金额为负）⇒ 预警侧低估客户收入 17.2%」经财务复验**证否并作废**： + 407 行 = 1,651(无过滤全行) − 1,244(符号过滤)，由 **302 行零额 + 105 行负额**组成（负额两式同样排除， + Σ`sum_money`=−841.22 万 / Σ含税=−1,717.97 万）；所谓「1,651 行 / 78 户 / 4,042.31 万」实为 + **无符号过滤（全行）**的结果（原报告**标签错置**）；「17.23%」= 4,883.52÷4,042.31−1， + 是「加符号过滤 vs **不滤**」之差（方向为**高估**），**不是**两路径之差。 + ⇒ 全体系**统一书面口径** `(sum_money + sum_tax_money) > 0`（预警与 CPA 同源同式，差集 0 ⇒ 行为零变化）。 +* 对拍锚点（2026-01~09，销售类白名单口径）：**65 户 / 1,207.74 万元 / Top1=22.5150%(unit152) / + Top2 单户=13.9734%(unit155) / Top4 累计=46.9407% / Top10 累计=69.7511%**（见 `ANCHORS_2026_01_09`）。 + 与账套 5001 逐月同向同量级（01: 229.02 vs 180.87；07: 81.55 vs 81.08；08: 15.83 vs 14.45）⇒ 可勾稽。 * `C_CUST_CONCENTRATION`（**沿用** `kpi_definitions.formula`）= 前5大客户收入 ÷ 期内客户收入合计 ×100。 （任务书 §2.1 写的 Top1 与现网定义「前5大」不一致 ⇒ 见报告「口径冲突登记」； 两者本模块都算：`top_n=1` / `top_n=5`。） @@ -36,26 +59,40 @@ from typing import Dict, Iterable, List, Optional, Sequence, Tuple # 口径常量（单一事实源） # --------------------------------------------------------------------------- +# 销售类单据白名单（P0 纠错主体）：1=销售单 XSD / 4=销售退货 XSTHD / 14=销售换货 XSHHD +SALES_BILL_TYPES: Tuple[int, ...] = (1, 4, 14) +BILL_SCOPE_FILTER = { + "sales": f"bill_type IN ({', '.join(str(t) for t in SALES_BILL_TYPES)})", + # 'all' 仅用于对拍/回归（复现旧口径与财务复验的全单据数字），**不得**用于对外指标口径。 + "all": "1 = 1", +} +DEFAULT_BILL_SCOPE = "sales" + REVENUE_FILTER = ( "entity_id = :eid AND unit_id IS NOT NULL AND unit_id > 0 " - "AND {sign} AND bill_date IS NOT NULL" + "AND {bill_type} AND {sign} AND bill_date IS NOT NULL" ) -# 金额与取舍口径（**2026-10-04 侦查实证**，逐项复现任务验收锚点） +# 金额与取舍口径（**2026-10-04 财务复验更正**） # 度量 = sum_money（不含税原值；与任务 §2.1 字面一致） # 取舍 = 两种「净额符号」过滤器，决定**哪些行计入**： -# 'tax_incl' = (sum_money + sum_tax_money) > 0 ← CPA 采用：唯此口径能逐项复现验收锚点 -# 2026-01~09：1,244 行 / 73 户 / 4,883.52 万元 / Top1 unit146 25.3835% / -# 次席 23.4905% / top4 累计 63.3249% / top10 累计 81.7675%（六项全中）。 -# 与 'money_only' 的差集 = 407 行红冲/退货类（sum_money<0 但含税为正，合计 −841.19 万）。 -# 'money_only'= sum_money > 0 ← 客户集中度预警 `_customer_revenue_shares` 既有口径， -# 为不改变既有预警阈值行为而**原样保留**（1,651 行 / 78 户 / 4,042.31 万元）。 -# ⚠️ 取舍口径并存属**在途分歧**：预警侧若沿用 money_only，会系统性低估客户收入（−17.2%）。 -# 已登记 ASK 交财务裁定；未裁定前不擅自改预警侧口径（改动=预警阈值漂移）。 +# 'tax_incl' = (sum_money + sum_tax_money) > 0 ← **全体系统一书面口径**（预警 + CPA 同源） +# 'money_only'= sum_money > 0 ← 保留作回归位；与本库当前数据**逐行同集**（差集 0 行） +# 2026-01~09 销售类白名单下（两式同值）：624 行 / 65 户 / 1,207.74 万元 / +# Top1 unit152 22.5150% / Top2 unit155 13.9734% / Top4 累计 46.9407% / Top10 累计 69.7511%。 MEASURE_SQL = "sum_money" SIGN_FILTER = {"tax_incl": "(sum_money + sum_tax_money) > 0", "money_only": "sum_money > 0"} DEFAULT_SIGN_BASIS = "tax_incl" +# 2026-01~09 对拍锚点（销售类白名单口径；派生器 `--anchor` 与回归共用，禁各写一份） +ANCHORS_2026_01_09 = { + "units": 65, "total_wan": 1207.74, + "top1": 22.5150, "top1_unit": 152, + "top2": 13.9734, "top2_unit": 155, + "top4_cum": 46.9407, "top10_cum": 69.7511, +} +ANCHOR_TOL = 0.05 + REVENUE_SQL_WINDOW = """ SELECT unit_id, SUM({money}) AS amount FROM zxt_masterbill @@ -75,26 +112,35 @@ REVENUE_SQL_MONTHLY = """ """ -def _sql_window(sign_basis: str = DEFAULT_SIGN_BASIS) -> str: - return REVENUE_SQL_WINDOW.format( - money=MEASURE_SQL, filter=REVENUE_FILTER.format(sign=SIGN_FILTER[sign_basis])) +def _filter_sql(sign_basis: str, bill_scope: str) -> str: + if sign_basis not in SIGN_FILTER: + raise ValueError(f"unknown sign_basis: {sign_basis!r} (expect {sorted(SIGN_FILTER)})") + if bill_scope not in BILL_SCOPE_FILTER: + raise ValueError(f"unknown bill_scope: {bill_scope!r} (expect {sorted(BILL_SCOPE_FILTER)})") + return REVENUE_FILTER.format( + bill_type=BILL_SCOPE_FILTER[bill_scope], sign=SIGN_FILTER[sign_basis]) -def _sql_monthly(sign_basis: str = DEFAULT_SIGN_BASIS) -> str: - return REVENUE_SQL_MONTHLY.format( - money=MEASURE_SQL, filter=REVENUE_FILTER.format(sign=SIGN_FILTER[sign_basis])) +def _sql_window(sign_basis: str = DEFAULT_SIGN_BASIS, bill_scope: str = DEFAULT_BILL_SCOPE) -> str: + return REVENUE_SQL_WINDOW.format(money=MEASURE_SQL, filter=_filter_sql(sign_basis, bill_scope)) + + +def _sql_monthly(sign_basis: str = DEFAULT_SIGN_BASIS, bill_scope: str = DEFAULT_BILL_SCOPE) -> str: + return REVENUE_SQL_MONTHLY.format(money=MEASURE_SQL, filter=_filter_sql(sign_basis, bill_scope)) CONCENTRATION_TOP_N_DEFAULT = 5 # 现网 kpi_definitions.formula = 前5大客户 FAIL_CLOSED_REASONS = { "C_CUST_COGS_ALLOC": ( - "销货成本分摊 = 客户收入 × 公司整体销货成本率（5401÷5001）。" - "**基数口径未决**：客户收入基座为 zxt_masterbill 含税单据流（2026-01~09 = 4,883.52 万），" - "而 5401/5001 来自凭证账（同期主营业务收入量级 ~900 万），两者相差约 5.4×；" - "直接相乘产出无经济意义的数。且 handoff 要求与「统一后的 CVP 变动成本口径」一致，" - "而 CVP 口径统一单（finance-20261003-cvp-fix-and-bohai-coverage-fullstack 第⑤项）尚在途。" - "⇒ 待财务裁定基数后方可落值。" + "销货成本分摊 = 客户收入 × 公司整体销货成本率（5401÷5001）——**乘法式已废止**" + "（财务定死 2026-10-04：跨基座相乘无经济意义）。" + "**基数口径待任总一字定**：客户收入基座为 zxt_masterbill **销售类**单据流（2026-01~09 = 1,207.74 万），" + "而 5401/5001 来自凭证账（同期 5001 = 808.80 万、5401 = 811.76 万，成本率 100.37%）；" + "乘法式会凭空放大 6 倍（旧全单据口径 4,883.52 ÷ 808.80 = 6.04× 基座差）。" + "财务推荐「成本池 × 客户份额」分置式：`Σ5401(期间,carry_forward=0) × [客户 i 收入 ÷ 期内客户收入合计]`" + "（份额分母须与本模块口径同一事实源）；选项 2 = zxt 实际单据成本；选项 3 = 维持 NULL。" + "⇒ 裁定前该列**恒 NULL**，禁 0、禁硬摊。" ), "C_CUST_CHAN_SUB": ( "客户专属渠道补贴须先按客户/渠道打标（571103 目前为总额、不可拆分，见任务 §2.2）；" @@ -113,11 +159,12 @@ WRITABLE_SNAPSHOT_COLUMNS = ("revenue",) # cogs_alloc/chan_sub/contrib/contrib_ # --------------------------------------------------------------------------- def fetch_revenue_rows(engine, entity_id: int, start: date, end: date, top_n: int = 1, - sign_basis: str = DEFAULT_SIGN_BASIS): + sign_basis: str = DEFAULT_SIGN_BASIS, bill_scope: str = DEFAULT_BILL_SCOPE): """窗口内按客户聚合收入（元），按金额倒序。 - sign_basis: 'tax_incl'=按含税净额取符号（CPA 默认，复现验收锚点）/ 'money_only'=按 sum_money 取符号 - （客户集中度预警既有行为，原样保留） + sign_basis: 'tax_incl'=按含税净额取符号（**统一书面口径**）/ 'money_only'=按 sum_money 取符号 + （回归位；与本库当前数据逐行同集） + bill_scope: 'sales'=销售类单据白名单（默认，P0 纠错）/ 'all'=全单据（仅对拍用，不得对外） 返回 (shares, total, period_range)： shares = [(unit_id, amount, pct), ...] 取前 top_n @@ -130,7 +177,8 @@ def fetch_revenue_rows(engine, entity_id: int, start: date, end: date, top_n: in try: with engine.connect() as conn: rows = conn.execute( - text(_sql_window(sign_basis)), {"eid": entity_id, "start": start, "end": end} + text(_sql_window(sign_basis, bill_scope)), + {"eid": entity_id, "start": start, "end": end}, ).fetchall() except Exception: return ([], 0.0, None) @@ -148,7 +196,8 @@ def fetch_revenue_rows(engine, entity_id: int, start: date, end: date, top_n: in return (shares, total, (start.isoformat(), end.isoformat())) -def fetch_monthly_revenue(engine, entity_id: int, sign_basis: str = DEFAULT_SIGN_BASIS) -> Dict[str, List[Tuple[int, float]]]: +def fetch_monthly_revenue(engine, entity_id: int, sign_basis: str = DEFAULT_SIGN_BASIS, + bill_scope: str = DEFAULT_BILL_SCOPE) -> Dict[str, List[Tuple[int, float]]]: """按自然月聚合的客户收入（元）。 返回 {period: [(unit_id, amount), ...] 按金额倒序}；period 形如 'YYYY-MM'。 @@ -158,7 +207,8 @@ def fetch_monthly_revenue(engine, entity_id: int, sign_basis: str = DEFAULT_SIGN out: Dict[str, List[Tuple[int, float]]] = {} with engine.connect() as conn: - rows = conn.execute(text(_sql_monthly(sign_basis)), {"eid": entity_id}).fetchall() + rows = conn.execute(text(_sql_monthly(sign_basis, bill_scope)), + {"eid": entity_id}).fetchall() for period, unit_id, amount in rows: out.setdefault(str(period), []).append((int(unit_id), float(amount or 0))) for p in out: @@ -188,7 +238,7 @@ def to_wan(yuan: float) -> float: def active_customer_count(pairs: Sequence[Tuple[int, float]]) -> int: - """期内活跃客户数（按 unit_id 去重，含极小金额客户——与对拍锚点 73 口径一致）。""" + """期内活跃客户数（按 unit_id 去重，含极小金额客户——与对拍锚点 65 口径一致）。""" return len({u for u, _ in pairs}) diff --git a/backend/scripts/kpi_customer_profitability_deriver.py b/backend/scripts/kpi_customer_profitability_deriver.py index 0358473e..535f05b2 100644 --- a/backend/scripts/kpi_customer_profitability_deriver.py +++ b/backend/scripts/kpi_customer_profitability_deriver.py @@ -8,8 +8,19 @@ 2) `kpi_values` 汇总口径（**只存汇总，避免 284 客户 × 期数污染 KPI 宽表**，任务 §2.1）： * `C_CUST_REVENUE` 期内客户收入合计（万元） * `C_CUST_COUNT` 期内活跃客户数（个） - * `C_CUST_CONCENTRATION` **补值**：仅对**库内尚无该期行**的期次写入（现网定义 = 前5大客户 - 收入 ÷ 期内合计 ×100；已有 2026-01~09 九期**原样保留、不覆盖**） + * `C_CUST_CONCENTRATION` 现网定义 = 前5大客户收入 ÷ 期内合计 ×100。 + `--concentration`（默认 `fill-gaps`）控制**已有期次**的处置： + * `fill-gaps` = 仅补库内尚无该期行的期次（2026-01~09 九期原样保留） + * `overwrite` = **重算并覆盖**指定期次（P0 口径纠错后重算九期用； + 属生产写，须走放行） + +口径（2026-10-04 P0 纠错后） + 单据类型白名单 = **销售类 `bill_type IN (1,4,14)`**（XSD 销售单 / XSTHD 销售退货 / + XSHHD 销售换货）；非销售单据（进货/验收/入库/预付款/收款/费用）的 unit_id 是**供应商**， + 不得计入客户收入。取数口径单一事实源 = `app/utils/cpa_caliber.py`（与 + `alert_rules._customer_revenue_shares` 同一实现）。 + 2026-01~09 对拍锚点：65 户 / 1,207.74 万 / Top1 22.5150%(u152) / Top2 13.9734%(u155) / + Top4 46.9407% / Top10 69.7511%（`cpa_caliber.ANCHORS_2026_01_09`）。 fail-closed（本器**不写**以下列/值；理由见 `app/utils/cpa_caliber.py::FAIL_CLOSED_REASONS`） `cogs_alloc` / `chan_sub` / `contrib` / `contrib_rate` / `C_CUST_COGS_ALLOC` / @@ -53,9 +64,10 @@ FAIL_CLOSED_NOTE = ( "cogs_alloc/chan_sub/contrib/contrib_rate = P1 fail-closed（未裁定口径/未打标），禁静默 0 或分摊值" ) -ANCHOR_2026_01_09 = {"units": 73, "total_wan": 4883.52, "top1": 25.3835, "top2": 23.4905, - "top4_cum": 63.3249, "top10_cum": 81.7675} -ANCHOR_TOL = 0.1 +# 对拍锚点**单一事实源** = `cpa_caliber.ANCHORS_2026_01_09`（销售类白名单口径，2026-10-04 P0 纠错后） +def _anchors(): + cal = _load_caliber() + return cal.ANCHORS_2026_01_09, cal.ANCHOR_TOL def _engine(args): @@ -109,6 +121,30 @@ def _load_caliber(): return cpa_caliber +def _dump_prior_concentration(engine, entity_id, kpi_id, ts): + """**覆盖前备份**现网 `C_CUST_CONCENTRATION` 全部行（回退点），落 `scripts/backups/`。 + + 返回备份文件绝对路径（写进结果 JSON，供审批串/回退引用）。 + """ + import json as _json + from sqlalchemy import text + cols = ("id", "period", "actual_value", "data_status", "source_type", "source_batch", + "remark", "entity_id") + with engine.connect() as conn: + rows = conn.execute(text( + "SELECT id, period, actual_value, data_status, source_type, source_batch, remark, entity_id " + "FROM kpi_values WHERE entity_id=:e AND kpi_id=:k ORDER BY period, id"), + {"e": entity_id, "k": kpi_id}).fetchall() + d = os.path.join(HERE, "backups") + os.makedirs(d, exist_ok=True) + path = os.path.join(d, "cpa-concentration-prior-%s.json" % ts.replace(":", "").replace(" ", "-")) + with open(path, "w", encoding="utf-8") as f: + _json.dump({"ts": ts, "entity_id": entity_id, "kpi_id": kpi_id, + "rows": [dict(zip(cols, r)) for r in rows]}, f, ensure_ascii=False, + indent=2, default=str) + return path + + def compute(engine, dialect, entity_id): """只读计算：{period: {unit_id: amount}} + 汇总。""" cal = _load_caliber() @@ -138,15 +174,24 @@ def anchor_check(summary): return pairs -def run(engine, dialect, entity_id, batch, apply=False, verify=False, since=None, until=None): +def run(engine, dialect, entity_id, batch, apply=False, verify=False, since=None, until=None, + concentration="fill-gaps", only="all"): cal = _load_caliber() from sqlalchemy import text - result = {"entity_id": entity_id, "batch": batch, "apply": apply, "verify": verify} + result = {"entity_id": entity_id, "batch": batch, "apply": apply, "verify": verify, + "concentration_mode": concentration, "only": only} + if concentration not in ("fill-gaps", "overwrite"): + raise SystemExit("FAIL-CLOSED: --concentration 仅支持 fill-gaps / overwrite") + if only not in ("all", "concentration"): + raise SystemExit("FAIL-CLOSED: --only 仅支持 all / concentration") + if apply and only == "concentration" and concentration != "overwrite": + raise SystemExit("FAIL-CLOSED: --only concentration 必须配 --concentration overwrite（否则九期被跳过=空写）") with engine.connect() as conn: exists = _table_exists(conn, dialect) - if apply and not exists: + if apply and only != "concentration" and not exists: raise SystemExit("FAIL-CLOSED: 目标表不存在，请先跑 migrate_cpa_snapshot_20261004.py --apply") + rows_before = None if exists: rows_before = conn.execute(text(f"SELECT COUNT(*) FROM {SNAPSHOT_TABLE} WHERE entity_id=:e"), @@ -174,21 +219,27 @@ def run(engine, dialect, entity_id, batch, apply=False, verify=False, since=None snap_rows = [] kpi_rows = [] + conc_written_periods = [] for p in periods: pairs = monthly[p] total_yuan = sum(a for _, a in pairs) - for idx, (uid, amount) in enumerate(pairs, start=1): - snap_rows.append({ - "entity_id": entity_id, "period": p, "unit_id": uid, - "revenue": cal.to_wan(amount), "rank_no": idx, - "calib_status": "revenue_only", "fail_closed_reason": FAIL_CLOSED_NOTE, - }) - kpi_rows.append(("C_CUST_REVENUE", p, cal.to_wan(total_yuan))) - kpi_rows.append(("C_CUST_COUNT", p, float(cal.active_customer_count(pairs)))) - if p not in existing_conc_periods: + if only != "concentration": + for idx, (uid, amount) in enumerate(pairs, start=1): + snap_rows.append({ + "entity_id": entity_id, "period": p, "unit_id": uid, + "revenue": cal.to_wan(amount), "rank_no": idx, + "calib_status": "revenue_only", "fail_closed_reason": FAIL_CLOSED_NOTE, + }) + kpi_rows.append(("C_CUST_REVENUE", p, cal.to_wan(total_yuan))) + kpi_rows.append(("C_CUST_COUNT", p, float(cal.active_customer_count(pairs)))) + if concentration == "overwrite" or p not in existing_conc_periods: kpi_rows.append(("C_CUST_CONCENTRATION", p, cal.concentration_pct(pairs, 5))) + if p in existing_conc_periods: + conc_written_periods.append(p) + result["only"] = only result["snapshot_rows"] = len(snap_rows) result["kpi_rows"] = len(kpi_rows) + result["concentration_overwritten_periods"] = sorted(conc_written_periods) result["kpi_rows_skipped_no_definition"] = [c for c, i in kpi_ids.items() if i is None] if not apply: @@ -204,11 +255,17 @@ def run(engine, dialect, entity_id, batch, apply=False, verify=False, since=None result["grants_ok"] = ok if not ok: raise SystemExit("FAIL-CLOSED: 应用账号缺 DML 授权（kpi_values / 目标表）。原始 SHOW GRANTS:\n" + grants) - missing = [c for c, i in kpi_ids.items() if i is None] + need = (["C_CUST_CONCENTRATION"] if only == "concentration" + else ["C_CUST_REVENUE", "C_CUST_COUNT", "C_CUST_CONCENTRATION"]) + missing = [c for c in need if kpi_ids.get(c) is None] if missing: raise SystemExit("FAIL-CLOSED: KPI 定义缺失 %s —— 请先执行 scripts/sql/cpa_kpi_defs_20261004.sql" % missing) now_ts = _dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + if concentration == "overwrite" and kpi_ids.get("C_CUST_CONCENTRATION"): + result["prior_backup"] = _dump_prior_concentration(engine, entity_id, + kpi_ids["C_CUST_CONCENTRATION"], now_ts) + inserted_snap = updated_snap = 0 kpi_inserted = kpi_updated = 0 with engine.begin() as conn: @@ -236,10 +293,14 @@ def run(engine, dialect, entity_id, batch, apply=False, verify=False, since=None "SELECT id, actual_value, data_status FROM kpi_values " "WHERE entity_id=:e AND kpi_id=:k AND period=:p ORDER BY id DESC LIMIT 1"), {"e": entity_id, "k": kid, "p": period}).fetchone() - remark = {"C_CUST_REVENUE": "口径=Σzxt_masterbill.sum_money（期内、unit_id>0、含税净额>0 取舍）", - "C_CUST_COUNT": "口径=期内 zxt_masterbill 活跃往来单位数（unit_id 去重）", - "C_CUST_CONCENTRATION": "口径=前5大客户收入/期内客户收入合计×100%（补值期次；" - "2026-01~09 现网值原样保留）"}[code] + conc_note = ("重算覆盖期次（P0 销售类白名单口径）" + if concentration == "overwrite" and period in existing_conc_periods + else "补值期次（库内原无该期行）") + remark = {"C_CUST_REVENUE": "口径=Σzxt_masterbill.sum_money（销售类 bill_type IN (1,4,14)、" + "期内、unit_id>0、符号过滤(sum_money+sum_tax_money)>0）", + "C_CUST_COUNT": "口径=期内 zxt_masterbill 销售类单据活跃客户数（unit_id 去重）", + "C_CUST_CONCENTRATION": f"口径=前5大客户收入/期内客户收入合计×100%（{conc_note}）"}[code] + if row: # 幂等：值不变则零写入；不降级 data_status if abs(float(row[1] or 0) - float(value)) < 1e-9 and row[2]: @@ -300,26 +361,36 @@ def _verify(engine, dialect, entity_id, monthly, periods, kpi_ids, existing_conc def anchor_assert(engine, entity_id): - """2026-01~09 窗口对拍锚点（任务验收标准逐项）。""" - from sqlalchemy import text + """2026-01~09 窗口对拍锚点（任务验收标准逐项）。 + + 取数**只走** `cpa_caliber` 单一事实源（不另写第二份聚合 SQL）：逐月客户对按月求和 + = 窗口客户合计（每行只属一个 unit，故月度求和与窗口 GROUP BY 等价）。 + """ cal = _load_caliber() - with engine.connect() as conn: - rows = conn.execute(text( - "SELECT unit_id, SUM(sum_money) AS amount FROM zxt_masterbill WHERE entity_id=:e " - "AND unit_id IS NOT NULL AND unit_id>0 AND (sum_money + sum_tax_money) > 0 " - "AND bill_date>='2026-01-01' AND bill_date<='2026-09-30' GROUP BY unit_id ORDER BY amount DESC"), - {"e": entity_id}).fetchall() - pairs = [(int(r[0]), float(r[1])) for r in rows] + anchors, tol = _anchors() + monthly = cal.fetch_monthly_revenue(engine, entity_id) + by_unit = {} + for p, pairs in monthly.items(): + if "2026-01" <= p <= "2026-09": + for u, a in pairs: + by_unit[u] = by_unit.get(u, 0.0) + a + pairs = sorted(by_unit.items(), key=lambda x: x[1], reverse=True) total = sum(a for _, a in pairs) cum = lambda n: round(sum(a for _, a in pairs[:n]) / total * 100, 4) if total else None got = {"units": len(pairs), "total_wan": round(total / 10000, 2), "top1": round(pairs[0][1] / total * 100, 4) if pairs else None, + "top1_unit": pairs[0][0] if pairs else None, "top2": round(pairs[1][1] / total * 100, 4) if len(pairs) > 1 else None, + "top2_unit": pairs[1][0] if len(pairs) > 1 else None, "top4_cum": cum(4), "top10_cum": cum(10)} checks = {} - for k, exp in ANCHOR_2026_01_09.items(): + for k, exp in anchors.items(): v = got.get(k) - checks[k] = {"expected": exp, "got": v, "pass": (v is not None and abs(float(v) - exp) <= ANCHOR_TOL)} + if isinstance(exp, int) and not isinstance(exp, bool) and k.endswith("_unit"): + checks[k] = {"expected": exp, "got": v, "pass": (v == exp)} + continue + checks[k] = {"expected": exp, "got": v, + "pass": (v is not None and abs(float(v) - float(exp)) <= tol)} return {"anchor": got, "checks": checks, "ALL_PASS": all(c["pass"] for c in checks.values())} @@ -333,6 +404,11 @@ def main(): ap.add_argument("--entity", type=int, default=1) ap.add_argument("--db-url") ap.add_argument("--batch", default=DEFAULT_BATCH) + ap.add_argument("--concentration", choices=("fill-gaps", "overwrite"), default="fill-gaps", + help="已有期次处置：fill-gaps=只补缺期（默认）/ overwrite=重算并覆盖（P0 纠错后重算九期用）") + ap.add_argument("--only", choices=("all", "concentration"), default="all", + help="写面：all=快照+三族汇总（默认）/ concentration=**仅**写 C_CUST_CONCENTRATION 期次" + "（P0 九期重算最小写面；需配 --concentration overwrite）") ap.add_argument("--since") ap.add_argument("--until") ap.add_argument("--json", action="store_true") @@ -347,7 +423,8 @@ def main(): return 0 if res["ALL_PASS"] else 1 res = run(engine, dialect, args.entity, args.batch, apply=args.apply, - verify=args.verify, since=args.since, until=args.until) + verify=args.verify, since=args.since, until=args.until, + concentration=args.concentration, only=args.only) print(json.dumps(res, ensure_ascii=False, indent=2)) return 0 diff --git a/backend/tests/test_customer_concentration_rule.py b/backend/tests/test_customer_concentration_rule.py index c1729251..b55a928d 100644 --- a/backend/tests/test_customer_concentration_rule.py +++ b/backend/tests/test_customer_concentration_rule.py @@ -6,6 +6,8 @@ 2. _customer_revenue_shares 客户收入聚合（zxt_masterbill 数据源） 3. _check_customer_concentration 双数据源（客户收入 / KPI 口径）+ 黄红灯分级 4. check-all 与 /run-customer-concentration 触发链路 + 去重 + 5. **销售类单据白名单**（P0 纠错 2026-10-04）：bill_type IN (1,4,14) 生效、 + 非销售单据（供应商）被排除、预警路径与 CPA 口径同源同式 """ import pytest from sqlalchemy import text as sa_text @@ -37,21 +39,37 @@ def _isolate_zxt_masterbill(): def _seed_masterbill(db: Session, rows, entity_id: int = 1): - """在测试库建 zxt_masterbill 并写入 (unit_id, amount, bill_date) 测试数据""" + """在测试库建 zxt_masterbill 并写入 + (unit_id, amount, bill_date[, bill_type[, sum_tax_money]]) 测试数据 + + 2026-10-04 P0 口径纠错后聚合过滤含**销售类白名单** `bill_type IN (1,4,14)` + 与**统一符号口径** `(sum_money + sum_tax_money) > 0` ⇒ 夹具必须带 `bill_type` / + `sum_tax_money` 两列（旧夹具只有 sum_money，改用例夹具属为适配新契约的**显式改写**）： + * bill_type 默认 1 = 销售单 XSD；逐行可指定应被排除的非销售单据 + （0=JHD 进货 / 129=YFKD 预付款 / 17=QTRK 其它入库 / 103=YBFYD 费用 / 128=YSKD 验收 + —— 这些行的 unit_id 是供应商）。 + * sum_tax_money 默认 0.0。 + """ db.execute(sa_text(""" CREATE TABLE IF NOT EXISTS zxt_masterbill ( id INTEGER PRIMARY KEY, entity_id INTEGER, unit_id INTEGER, + bill_type INTEGER, sum_money FLOAT, + sum_tax_money FLOAT, bill_date DATE ) """)) - for i, (unit_id, amount, bill_date) in enumerate(rows, start=1): + for i, row in enumerate(rows, start=1): + unit_id, amount, bill_date = row[0], row[1], row[2] + bill_type = row[3] if len(row) > 3 else 1 + tax = row[4] if len(row) > 4 else 0.0 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}) + "INSERT INTO zxt_masterbill (id, entity_id, unit_id, bill_type, sum_money, " + "sum_tax_money, bill_date) VALUES (:id, :eid, :uid, :bt, :amt, :tax, :bd)" + ), {"id": i, "eid": entity_id, "uid": unit_id, "bt": bill_type, "amt": amount, + "tax": tax, "bd": bill_date}) db.commit() @@ -140,6 +158,72 @@ class TestCustomerRevenueAggregation: assert shares[0][0] == 101 and shares[0][2] == 60.0 +class TestSalesBillTypeWhitelist: + """P0 口径纠错（2026-10-04）：单据类型白名单（销售类 bill_type IN (1,4,14)） + + 背景：非销售单据（进货/预付款/验收/入库/收款/费用）的 unit_id 是**供应商** + （abst 实读「向【酣客酒业（广州）】进货」「预付款单」），混入会把 + 「客户集中度」度量成「往来单位（含供应商）集中度」。 + """ + + def test_purchase_and_prepay_rows_excluded(self, db: Session): + _seed_masterbill(db, [ + (152, 600.0, "2026-08-01", 1), # XSD 销售单 → 计入 + (193, 300.0, "2026-08-02", 4), # XSTHD 销售退货 → 计入（销售类） + (146, 9000.0, "2026-08-03", 0), # JHD 进货（供应商）→ 排除 + (146, 8000.0, "2026-08-04", 129), # YFKD 预付款（供应商）→ 排除 + (474, 7000.0, "2026-08-05", 17), # QTRK 其它入库（供应商）→ 排除 + (101, 5000.0, "2026-08-06", 103), # YBFYD 一般费用单 → 排除 + (102, 4000.0, "2026-08-07", 128), # YSKD 验收单 → 排除 + ]) + shares, total, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=2) + assert total == 900.0 + assert [s[0] for s in shares] == [152, 193] + assert shares[0][2] == round(600.0 / 900.0 * 100, 2) + + def test_sales_whitelist_in_generated_sql(self): + from app.utils import cpa_caliber + sql = cpa_caliber._sql_window() + assert "bill_type IN (1, 4, 14)" in sql # 白名单生效 + assert "(sum_money + sum_tax_money) > 0" in sql # 统一书面口径 + assert sql == cpa_caliber._sql_window("tax_incl", "sales") + all_sql = cpa_caliber._sql_window(bill_scope="all") + assert "bill_type" not in all_sql # 对拍用全单据口径不带白名单 + + def test_unknown_scope_and_sign_rejected(self): + from app.utils import cpa_caliber + with pytest.raises(ValueError): + cpa_caliber._sql_window(bill_scope="bogus") + with pytest.raises(ValueError): + cpa_caliber._sql_window(sign_basis="bogus") + + def test_sign_caliber_semantics_on_negative_row(self, db: Session): + """负额行（sum_money<0、含税>0）：tax_incl 计入 / money_only 排除 ⇒ 两式**非恒等**。 + + 本库当前数据**不存在**此类行（2026-01~09 双向差集 0 行；全表亦 0 行）， + 故「两口径等价」是**数据事实**（财务复验 §5.1），不是恒等式——本用例固化语义， + 防后人据「等价」误删符号过滤。 + """ + from app.database import get_engine + from app.utils import cpa_caliber + _seed_masterbill(db, [(999, -100.0, "2026-08-03", 1, 130.0)]) + incl = cpa_caliber.fetch_monthly_revenue(get_engine(), 1, sign_basis="tax_incl") + money = cpa_caliber.fetch_monthly_revenue(get_engine(), 1, sign_basis="money_only") + assert incl.get("2026-08") == [(999, -100.0)] # 含税净额 +30 > 0 ⇒ 计入 + assert "2026-08" not in money # sum_money -100 ⇒ 排除 + + def test_alert_path_shares_single_caliber_with_cpa(self, db: Session): + """预警路径与 CPA 派生器同源同式（防「同一客户收入两份结果」）""" + from datetime import date, timedelta + from app.database import get_engine + from app.utils import cpa_caliber + _seed_masterbill(db, [(152, 600.0, "2026-08-01", 1), (146, 9000.0, "2026-08-03", 0)]) + shares, total, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=5) + cpa = cpa_caliber.fetch_revenue_rows( + get_engine(), 1, date.today() - timedelta(days=360), date.today(), 5) + assert shares == cpa[0] and total == cpa[1] == 600.0 + + class TestConcentrationChecker: """占比判定：阈值 / 黄红灯 / 双数据源"""