Hermes CI Fix
4fe956fcf9
ci/woodpecker/push/main Pipeline failed
Root cause: resolve_window() matched quarter/half with p[4] == "q" / "h"
(always '-' for "2026-q2"/"2026-h1"), and the branches sat after the
single-month branch -> dead code. Real input fell into the month branch,
so months_between() raised ValueError on int('q2') -> HTTP 500 for all
18 quarter/half options in the period selector.
Fix: judge on p[5] in ("q","h") with a digit guard on p[6], evaluated
before the single-month branch; month branch tightened to p[5:].isdigit().
Existing error-branch wording unchanged.
Tests: backend/tests/test_dupont_period_options.py (18 defect options,
case equivalence, month/year/ytd/ttm/custom non-regression, malformed
input returns error dict instead of raising).
392 lines
19 KiB
Python
392 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""杜邦分析「区间取值」口径单一事实源 —— 流量累计 × 存量平均 × 可选年化 + 恒等式守护
|
||
|
||
依据（唯一有效出处）：
|
||
· 任务件 `finance-20261003-dupont-period-basis-design-fullstack.md`
|
||
· 放行出处：任总 2026-10-03 09:49 CST DM「立」（承 09:31「杜邦分析如何进行区间取值设计」）
|
||
· 财务口径裁定见任务件 §二（本模块逐条落地，不另立口径）
|
||
|
||
核心原则（§二 开篇）：
|
||
**区间一旦选定，流量（分子）与存量平均（分母）必须同区间；且三因子必须同区间——杜邦恒等式才成立。**
|
||
|
||
取值规则（§二.1，硬性）：
|
||
· 损益类（流量） → 区间**累计发生额**（期间 [start, end] 内净额）
|
||
· 资产负债类（存量）→ 区间**平均余额** = (期初 + 期末) / 2
|
||
· 跨期比较 → **年化** = 流量 × 12 / 区间月数（比率型因子见 `annualize_factors` 说明）
|
||
|
||
fail-closed（§二.6）：
|
||
期初余额缺失 ⇒ **显式**回落期末，并在返回值标注 `basis_fallback=true`；**禁静默**。
|
||
|
||
单一事实源（§二.5）：
|
||
总资产 / 净资产 / 权益乘数 不再由前端拼 `raw_data`，而是由
|
||
`scripts/kpi_dupont_deriver.py` 统一派生为 `F_ASSET_TOTAL` / `F_EQUITY_TOTAL` /
|
||
`F_EQUITY_MULTIPLIER`（`source_type='derived'`）；本模块是其**唯一读取入口**（派生器与
|
||
只读端点共用），口径参数落 `kpi_definitions.data_source_config.params`。
|
||
|
||
⚠️ 取数源（§二.5 注）：**不用**旧脚本 `register_dupont_kpis.py` 的 ERP 直连近似口径
|
||
（那套只有「现金银行 + 固定资产」，缺应收/存货）；一律走 CMA 库内
|
||
`subject_balances`（存量）+ `voucher_details`（流量）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import text
|
||
|
||
#: 恒等式容差（百分点）——§二.3 建议 0.05
|
||
IDENTITY_TOLERANCE = 0.05
|
||
|
||
#: 支持的区间维度（前端选择器 §二.2）
|
||
PERIOD_PRESETS = ("ytd", "ttm", "month", "quarter", "half", "year", "custom")
|
||
|
||
|
||
# ────────────────────────── 账套口径（单一事实源） ──────────────────────────
|
||
|
||
#: 各账套会计科目体系差异（实测 2026-10-03）：
|
||
#: · entity 1（酣客）：旧准则科目 —— 收入 5001，权益 3xxx，资产 1xxx，负债 2xxx；
|
||
#: 净利润有期末结转腿（`carry_forward=1` 的 3xxx 行），与已上线 verified 的月度
|
||
#: F_NET_PROFIT/F_ROE/F_ROA 同口径（实测 2026-08：F_NET_PROFIT=-10.46、F_ROE=-0.95、
|
||
#: F_ROA=-0.57 与本模块输出逐项一致）。
|
||
#: · entity 2（博海）：新 30 号准则科目 —— 收入 6001/6051/6052，权益 4xxx，费用 6xxx；
|
||
#: `voucher_details` 无任何 `carry_forward=1` 行 ⇒ 净利润走损益类净额口径。
|
||
CALIBER_DEFAULTS: Dict[str, Any] = {
|
||
"asset_prefix": "1",
|
||
"liability_prefix": "2",
|
||
"1": { # entity_id=1 酣客
|
||
"equity_prefixes": ["3"],
|
||
"revenue_codes": ["5001"],
|
||
"pnl_prefixes": ["5", "6"],
|
||
"net_profit_mode": "carry_forward", # Σ(贷−借) over carry_forward=1 且 权益前缀
|
||
},
|
||
"2": { # entity_id=2 博海
|
||
"equity_prefixes": ["4"],
|
||
"revenue_codes": ["6001", "6051", "6052"],
|
||
"pnl_prefixes": ["6"],
|
||
"net_profit_mode": "pnl_net", # Σ(贷−借) over carry_forward=0 且 损益前缀
|
||
},
|
||
#: 未登记账套的保守兜底（新准则形态）
|
||
"_fallback": {
|
||
"equity_prefixes": ["4"],
|
||
"revenue_codes": ["6001", "6051", "6052"],
|
||
"pnl_prefixes": ["6"],
|
||
"net_profit_mode": "pnl_net",
|
||
},
|
||
}
|
||
|
||
#: 参数读取入口（可被 `kpi_definitions.data_source_config.params` 覆盖；禁散落字面量）
|
||
PARAM_KPI_CODE = "F_ASSET_TOTAL"
|
||
|
||
|
||
def caliber_for(entity_id: int) -> Dict[str, Any]:
|
||
"""账套科目体系口径（内置默认；可供 DB 覆盖的键：equity_prefixes/revenue_codes/net_profit_mode）"""
|
||
ent = CALIBER_DEFAULTS.get(str(entity_id)) or CALIBER_DEFAULTS["_fallback"]
|
||
return {
|
||
"asset_prefix": CALIBER_DEFAULTS["asset_prefix"],
|
||
"liability_prefix": CALIBER_DEFAULTS["liability_prefix"],
|
||
"equity_prefixes": list(ent["equity_prefixes"]),
|
||
"revenue_codes": list(ent["revenue_codes"]),
|
||
"pnl_prefixes": list(ent["pnl_prefixes"]),
|
||
"net_profit_mode": ent["net_profit_mode"],
|
||
}
|
||
|
||
|
||
def load_dupont_params(db, entity_id: int) -> Dict[str, Any]:
|
||
"""读取账套口径参数：`kpi_definitions.data_source_config.params` → 内置默认兜底。
|
||
|
||
兼容形态：`{"params": {...}}` / 裸 `{...}` / JSON 字符串 / None。读取失败绝不抛错
|
||
（口径回落默认，端点不因参数缺件 500）。
|
||
"""
|
||
caliber = caliber_for(entity_id)
|
||
try:
|
||
row = db.execute(text(
|
||
"SELECT data_source_config FROM kpi_definitions "
|
||
"WHERE entity_id = :eid AND kpi_code = :code LIMIT 1"
|
||
), {"eid": entity_id, "code": PARAM_KPI_CODE}).first()
|
||
except Exception: # pragma: no cover - 表缺失等极端情形
|
||
row = None
|
||
raw = row[0] if row else None
|
||
if isinstance(raw, (bytes, bytearray)):
|
||
raw = raw.decode("utf-8", "replace")
|
||
if isinstance(raw, str):
|
||
try:
|
||
raw = json.loads(raw)
|
||
except Exception:
|
||
raw = None
|
||
if isinstance(raw, dict):
|
||
params = raw.get("params") if isinstance(raw.get("params"), dict) else raw
|
||
for key in ("equity_prefixes", "revenue_codes", "pnl_prefixes", "net_profit_mode",
|
||
"asset_prefix", "liability_prefix"):
|
||
if key in params:
|
||
caliber[key] = params[key]
|
||
caliber["_params_source"] = "kpi_definitions.data_source_config.params"
|
||
else:
|
||
caliber["_params_source"] = "builtin-default"
|
||
return caliber
|
||
|
||
|
||
# ────────────────────────── 区间窗口 ──────────────────────────
|
||
|
||
def prev_month(period: str) -> str:
|
||
y, m = int(period[:4]), int(period[5:7])
|
||
if m == 1:
|
||
return f"{y - 1}-12"
|
||
return f"{y}-{m - 1:02d}"
|
||
|
||
|
||
def month_index(period: str) -> int:
|
||
return int(period[5:7])
|
||
|
||
|
||
def months_between(start: str, end: str) -> int:
|
||
"""闭区间 [start, end] 的月数（含首尾）。"""
|
||
if not start or not end:
|
||
return 0
|
||
return (int(end[:4]) - int(start[:4])) * 12 + (int(end[5:7]) - int(start[5:7])) + 1
|
||
|
||
|
||
def _fmt_cn(start: str, end: str) -> str:
|
||
if start == end:
|
||
return f"{start[:4]}年{int(start[5:7])}月"
|
||
return f"{start[:4]}年{int(start[5:7])}-{int(end[5:7])}月"
|
||
|
||
|
||
def resolve_window(period: Optional[str], start: Optional[str], end: Optional[str],
|
||
available: List[str], preset: Optional[str] = None,
|
||
today: Optional[str] = None, anchor: Optional[str] = None) -> Dict[str, Any]:
|
||
"""区间维度 → [period_start, period_end]（闭区间，YYYY-MM）。
|
||
|
||
§二.2 区间维度：月 / 季 / 半年 / 年 / YTD（默认）/ TTM / 自定义起止。
|
||
可用月域 `available`（升序 YYYY-MM 列表）用于：① 缺省锚定；② 区间裁剪。
|
||
`anchor` = 默认区间的锚月（端点传入「存量最新月」`latest_stock_month`，缺省取 available 末月）。
|
||
|
||
⚠️ 裁剪纪律：区间上界**只裁不扩**——超出可用月域即裁到可用域（并在 `clip_note` 留痕），
|
||
绝不因裁剪而静默改变用户已选下界（下界越界则显式 400，见 `validate`）。
|
||
"""
|
||
avail = sorted(available or [])
|
||
latest = avail[-1] if avail else None
|
||
today = today or datetime.now().strftime("%Y-%m")
|
||
anchor = min(anchor or latest or today, today)
|
||
if latest:
|
||
anchor = min(anchor, latest)
|
||
|
||
def clip_end(e: str) -> Tuple[str, Optional[str]]:
|
||
if latest and e > latest:
|
||
return latest, f"上界 {e} 超出可用月域（最新 {latest}），已裁至可用域"
|
||
return e, None
|
||
|
||
p = (period or "").strip().lower()
|
||
if preset:
|
||
p = preset.strip().lower()
|
||
note = None
|
||
if not p or p in ("ytd", "default"):
|
||
if not anchor:
|
||
return {"error": "无可用数据月域，无法解析默认 YTD 区间"}
|
||
s, e = f"{anchor[:4]}-01", anchor
|
||
key = "ytd"
|
||
elif p in ("ttm",):
|
||
if not anchor:
|
||
return {"error": "无可用数据月域，无法解析 TTM 区间"}
|
||
y, m = int(anchor[:4]), int(anchor[5:7])
|
||
s = f"{y - 1}-{m:02d}" if m != 12 else f"{y}-01"
|
||
e = anchor
|
||
key = "ttm"
|
||
elif p in ("custom",) or start or end:
|
||
if not (start and end):
|
||
return {"error": "自定义区间须同时给出 start 与 end（YYYY-MM）"}
|
||
s, e = start, end
|
||
key = "custom"
|
||
elif len(p) == 7 and p[5] == "q" and p[6].isdigit(): # 2026-q3
|
||
y, q = int(p[:4]), int(p[6])
|
||
s, e = f"{y}-{(q - 1) * 3 + 1:02d}", f"{y}-{q * 3:02d}"
|
||
key = "quarter"
|
||
elif len(p) == 7 and p[5] == "h" and p[6].isdigit(): # 2026-h1
|
||
y, h = int(p[:4]), int(p[6])
|
||
s, e = f"{y}-{(h - 1) * 6 + 1:02d}", f"{y}-{h * 6:02d}"
|
||
key = "half"
|
||
elif len(p) == 7 and p[5:].isdigit(): # 2026-08 → 单月
|
||
s = e = p
|
||
key = "month"
|
||
elif len(p) == 4 and p.isdigit(): # 2026 → 全年
|
||
s, e = f"{p}-01", f"{p}-12"
|
||
key = "year"
|
||
else:
|
||
return {"error": f"无法识别的区间参数 period={period!r}（支持 ytd/ttm/YYYY-MM/YYYY-Qn/YYYY-Hn/YYYY/custom）"}
|
||
|
||
e, cn = clip_end(e)
|
||
note = cn
|
||
months = months_between(s, e)
|
||
return {
|
||
"period_key": key, "period_start": s, "period_end": e, "months": months,
|
||
"label": _fmt_cn(s, e), "clip_note": note,
|
||
"available_start": avail[0] if avail else None, "available_end": latest,
|
||
}
|
||
|
||
|
||
# ────────────────────────── 取数（流量 / 存量） ──────────────────────────
|
||
|
||
def _flow_expr(field: str) -> str:
|
||
"""流量表达式。命名显式化（避免与存量 net 的借−贷/贷−借混淆）：
|
||
|
||
· `credit` → 贷方发生额
|
||
· `debit` → 借方发生额
|
||
· `credit_debit` → 贷−借（收入类、结转腿自然方向）
|
||
· `debit_credit` → 借−贷（成本费用类自然方向）
|
||
"""
|
||
expr = {"credit": "credit_amount", "debit": "debit_amount",
|
||
"credit_debit": "(credit_amount - debit_amount)",
|
||
"debit_credit": "(debit_amount - credit_amount)"}.get(field)
|
||
if expr is None:
|
||
raise ValueError(f"未知流量字段 {field!r}（credit/debit/credit_debit/debit_credit）")
|
||
return expr
|
||
|
||
|
||
def sum_flows(db, entity_id: int, start: str, end: str, spec: Dict[str, Any]) -> float:
|
||
"""流量累计（元）。spec: {codes:[...]|prefixes:[...], field: 见 _flow_expr, carry_forward: 0|1}"""
|
||
parts, args = [], {"eid": entity_id, "s": start, "e": end}
|
||
where = ["entity_id = :eid", "period >= :s", "period <= :e"]
|
||
cf = spec.get("carry_forward")
|
||
if cf is not None:
|
||
where.append("carry_forward = :cf")
|
||
args["cf"] = int(cf)
|
||
for i, code in enumerate(spec.get("codes") or []):
|
||
parts.append(f"subject_code = :c{i}")
|
||
args[f"c{i}"] = code
|
||
for i, pfx in enumerate(spec.get("prefixes") or []):
|
||
parts.append(f"subject_code LIKE :p{i}")
|
||
args[f"p{i}"] = f"{pfx}%"
|
||
if parts:
|
||
where.append("(" + " OR ".join(parts) + ")")
|
||
sql = (f"SELECT COALESCE(SUM({_flow_expr(spec.get('field', 'credit_debit'))}), 0) "
|
||
f"FROM voucher_details WHERE " + " AND ".join(where))
|
||
return float(db.execute(text(sql), args).scalar() or 0.0)
|
||
|
||
|
||
def _bal_expr(field: str) -> str:
|
||
return {"cb": "(cb_dr - cb_cr)", "cb_rev": "(cb_cr - cb_dr)",
|
||
"ob": "(ob_dr - ob_cr)", "ob_rev": "(ob_cr - ob_dr)"}[field]
|
||
|
||
|
||
def stock_balance(db, entity_id: int, period: str, prefixes: List[str],
|
||
side: str = "net") -> Dict[str, Optional[float]]:
|
||
"""存量余额（元）：期初 `ob_*` / 期末 `cb_*`，一级科目（`level=1`，防跨层双计）。
|
||
|
||
side: 'net' = 借−贷（资产用）/ 'net_rev' = 贷−借（权益、负债用）。
|
||
返回 {"begin": 元|None, "end": 元|None, "row_found": bool}
|
||
ob 字段: net→ob / net_rev→ob_rev；cb 同理。
|
||
"""
|
||
ob_f = "ob" if side == "net" else "ob_rev"
|
||
cb_f = "cb" if side == "net" else "cb_rev"
|
||
pfx_sql, args = [], {"eid": entity_id, "period": period}
|
||
for i, pfx in enumerate(prefixes):
|
||
pfx_sql.append(f"subject_code LIKE :p{i}")
|
||
args[f"p{i}"] = f"{pfx}%"
|
||
sql = (f"SELECT COALESCE(SUM({_bal_expr(ob_f)}), 0), COALESCE(SUM({_bal_expr(cb_f)}), 0), COUNT(*) "
|
||
f"FROM subject_balances WHERE entity_id = :eid AND period = :period AND `level` = 1 "
|
||
f"AND (" + " OR ".join(pfx_sql) + ")")
|
||
row = db.execute(text(sql), args).first()
|
||
if not row or not row[2]:
|
||
return {"begin": None, "end": None, "row_found": False}
|
||
return {"begin": float(row[0] or 0.0), "end": float(row[1] or 0.0), "row_found": True}
|
||
|
||
|
||
def available_months(db, entity_id: int) -> List[str]:
|
||
"""可用月域 = `voucher_details` 与 `subject_balances` 期间并集（升序）。
|
||
|
||
⚠️ 两次单表查询后 Python 合并，**不用 SQL `UNION`** —— 两表 `period` 列 collation
|
||
不同，MySQL 直接 UNION 会报 `1271 Illegal mix of collations`（2026-10-03 实测）。
|
||
"""
|
||
months: set = set()
|
||
for tbl in ("voucher_details", "subject_balances"):
|
||
rows = db.execute(text(
|
||
f"SELECT DISTINCT period FROM {tbl} WHERE entity_id = :eid"
|
||
), {"eid": entity_id}).fetchall()
|
||
months.update(str(r[0]) for r in rows if r[0])
|
||
return sorted(months)
|
||
|
||
|
||
def latest_stock_month(db, entity_id: int) -> Optional[str]:
|
||
"""最后一个**存量可取**的期间（subject_balances 有该期一级科目行）。
|
||
|
||
默认区间锚点必须取它而非流量最新月：两账套刻度不同（实测 entity2 流量到 2026-09、
|
||
存量只到 2026-08），若按流量最新月锚定，默认视图会落到「期末存量缺失」的区间上
|
||
（fail-closed ⇒ 全空），属可用性缺陷而非诚实的 fail-closed。
|
||
"""
|
||
row = db.execute(text(
|
||
"SELECT MAX(period) FROM subject_balances WHERE entity_id = :eid AND `level` = 1"
|
||
), {"eid": entity_id}).scalar()
|
||
return str(row) if row else None
|
||
|
||
|
||
# ────────────────────────── 纯计算（无 DB，便于单测） ──────────────────────────
|
||
|
||
def annualize_factors(factors: Dict[str, Optional[float]], months: int,
|
||
annualize: bool) -> Dict[str, Any]:
|
||
"""年化（§二.1）：**流量** × 12 / 区间月数。
|
||
|
||
比率型因子的年化是非平凡的，本函数显式给出每因子的实际缩放倍数，避免「看着像没生效」：
|
||
|
||
· 净利率 = 净利润/营业收入 —— 分子分母同为流量，×k 相消 ⇒ scale=1.0（**值不变**）
|
||
· 总资产周转率 = 营业收入/平均总资产 —— 仅分子年化 ⇒ scale=k
|
||
· 权益乘数 = 平均总资产/平均净资产 —— 分子分母同为存量 ⇒ scale=1.0（**值不变**）
|
||
· ROE = 净利润/平均净资产 —— 仅分子年化 ⇒ scale=k
|
||
|
||
这也正是恒等式在年化下继续成立的原因：
|
||
k·净利润/平均净资产 ≡ 净利率 × (k·周转率) × 权益乘数。
|
||
"""
|
||
k = (12.0 / months) if (annualize and months) else 1.0
|
||
scales = {"net_profit_margin": 1.0, "asset_turnover": k, "equity_multiplier": 1.0,
|
||
"roe": k}
|
||
out: Dict[str, Any] = {"scale": k, "factors": {}, "per_factor_scale": scales}
|
||
for name, val in (factors or {}).items():
|
||
s = scales.get(name, 1.0)
|
||
out["factors"][name] = None if val is None else val * s
|
||
return out
|
||
|
||
|
||
def compute_factors(net_profit_w: Optional[float], revenue_w: Optional[float],
|
||
total_assets_w: Optional[float], equity_w: Optional[float]) -> Dict[str, Optional[float]]:
|
||
"""三因子 + ROE（万元口径；ROE/净利率为百分点）。任一分母为 0/缺失 ⇒ 该因子 None（fail-closed）。"""
|
||
out = {"net_profit_margin": None, "asset_turnover": None,
|
||
"equity_multiplier": None, "roe": None}
|
||
if revenue_w:
|
||
out["net_profit_margin"] = net_profit_w / revenue_w * 100.0 if net_profit_w is not None else None
|
||
if total_assets_w:
|
||
out["asset_turnover"] = revenue_w / total_assets_w if revenue_w is not None else None
|
||
if equity_w:
|
||
out["equity_multiplier"] = total_assets_w / equity_w
|
||
if net_profit_w is not None:
|
||
out["roe"] = net_profit_w / equity_w * 100.0
|
||
return {k: (None if v is None else v) for k, v in out.items()}
|
||
|
||
|
||
def check_identity(roe: Optional[float], net_profit_margin: Optional[float],
|
||
asset_turnover: Optional[float], equity_multiplier: Optional[float],
|
||
tolerance: float = IDENTITY_TOLERANCE) -> Dict[str, Any]:
|
||
"""恒等式守护（§二.3）：| ROE − 净利润率 × 总资产周转率 × 权益乘数 | ≤ 容差（百分点）。
|
||
|
||
不成立 ⇒ `status='identity_broken'`，调用方**不得返回数值**（只返回三因子各自的区间元数据）。
|
||
"""
|
||
if None in (roe, net_profit_margin, asset_turnover, equity_multiplier):
|
||
return {"checked": False, "status": "unavailable", "tolerance": tolerance,
|
||
"reason": "三因子或 ROE 不可得（分母缺失），恒等式免检"}
|
||
product = net_profit_margin / 100.0 * asset_turnover * equity_multiplier * 100.0
|
||
diff = abs(roe - product)
|
||
return {"checked": True, "status": "pass" if diff <= tolerance else "identity_broken",
|
||
"roe": round(roe, 6), "product": round(product, 6), "diff": round(diff, 6),
|
||
"tolerance": tolerance}
|
||
|
||
|
||
def build_window_meta(label: str, start: str, end: str, months: int,
|
||
basis: str, annualized: bool, fallback: bool,
|
||
annualize_scale: float = 1.0,
|
||
fallback_reason: Optional[str] = None) -> Dict[str, Any]:
|
||
"""§二.4 口径元数据（随值返回）：period_start / period_end / basis / annualized 四项。"""
|
||
meta = {"period": label, "period_start": start, "period_end": end, "months": months,
|
||
"basis": basis, "annualized": bool(annualized), "basis_fallback": bool(fallback),
|
||
"annualize_scale": annualize_scale}
|
||
if fallback_reason:
|
||
meta["basis_fallback_reason"] = fallback_reason
|
||
return meta
|