74 lines
3.8 KiB
Python
74 lines
3.8 KiB
Python
"""Normalize peer data cards without assuming an upstream/downstream dependency."""
|
|
import math
|
|
import re
|
|
from decimal import Decimal, InvalidOperation
|
|
from datetime import date
|
|
|
|
MONEY_FIELDS = {"facility.amount", "financial.total_assets", "financial.total_liabilities", "financial.operating_cashflow", "lease.total_value", "insurance.amount", "deposit.amount", "guarantee.amount"}
|
|
SCALE = {"元": Decimal("0.0001"), "万元": Decimal(1), "亿元": Decimal(10000)}
|
|
|
|
|
|
def period_key(value):
|
|
text = str(value or "")
|
|
if re.fullmatch(r"\d{4}", text):
|
|
return date(int(text), 12, 31)
|
|
if re.fullmatch(r"\d{4}-\d{2}", text):
|
|
return date.fromisoformat(text + "-01")
|
|
return date.fromisoformat(text)
|
|
|
|
|
|
def align(card, rows, project_id, requested_period=None):
|
|
config = card.get("alignment", {})
|
|
project_field = config.get("project_field", "project_id")
|
|
period_field = config.get("period_field", "")
|
|
scope_field = config.get("scope_field", "")
|
|
selection = config.get("selection", "single")
|
|
if not rows:
|
|
raise ValueError("未取得项目记录，不得解释为无风险")
|
|
if not project_field or any(str(row.get(project_field, "")) != project_id for row in rows):
|
|
raise ValueError("返回记录的项目标识缺失或不一致，请检查查询和项目关联字段")
|
|
records = []
|
|
for row in rows:
|
|
period = row.get(period_field) if period_field else card.get("period")
|
|
scope = row.get(scope_field) if scope_field else card.get("scope")
|
|
mapped, units = {}, {}
|
|
for field, source in card.get("mapping", {}).items():
|
|
value = row.get(source)
|
|
unit = card.get("field_units", {}).get(field, card.get("unit") if field in MONEY_FIELDS else None)
|
|
if value is not None and unit:
|
|
if unit not in SCALE:
|
|
raise ValueError(f"{field} 的金额单位不支持：{unit}")
|
|
try:
|
|
value = float(Decimal(str(value)) * SCALE[unit])
|
|
if not math.isfinite(value): raise ValueError("非有限值")
|
|
except (ValueError, InvalidOperation):
|
|
raise ValueError(f"{field} 金额不是有效数值")
|
|
units[field] = "万元"
|
|
mapped[field] = value
|
|
records.append({"period": period, "scope": scope, "values": mapped, "units": units, "raw": row})
|
|
target_scope = config.get("target_scope", "")
|
|
candidates = [r for r in records if not target_scope or r["scope"] == target_scope]
|
|
if not candidates:
|
|
raise ValueError("未取得指定财务口径的记录")
|
|
if len({r["scope"] for r in candidates}) > 1:
|
|
raise ValueError("返回多个财务口径，请明确目标口径，不能混合计算")
|
|
if selection in ("latest", "period", "all"):
|
|
try:
|
|
for r in candidates: period_key(r["period"])
|
|
except (ValueError, TypeError):
|
|
raise ValueError("多期记录缺少有效期间，应使用 YYYY、YYYY-MM 或 YYYY-MM-DD")
|
|
keys = [(r["period"], r["scope"]) for r in candidates]
|
|
if len(set(keys)) != len(keys):
|
|
raise ValueError("同一期间与口径存在重复记录，可能由多表关联放大行数")
|
|
if (requested_period and period_field) or selection == "period":
|
|
target = requested_period or config.get("target_period")
|
|
candidates = [r for r in candidates if str(r["period"]) == str(target)]
|
|
elif selection == "latest":
|
|
newest = max(period_key(r["period"]) for r in candidates)
|
|
candidates = [r for r in candidates if period_key(r["period"]) == newest]
|
|
if selection == "all" and not (requested_period and period_field):
|
|
return records, None
|
|
if len(candidates) != 1:
|
|
raise ValueError(f"取得 {len(candidates)} 条候选记录，请指定期间或明确唯一粒度")
|
|
return records, candidates[0]
|