""" CMA管理报表中心 — 管理会计OS 非传统财务报表,聚焦管理决策分析 报表: 1. 管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润 2. 预算执行报告 — 各KPI预算vs实际vs差异率 3. KPI趋势报告 — 选定KPI的历史趋势 4. 四维度绩效评分卡 — BSC健康度雷达图 """ from fastapi import APIRouter, Depends, Query, HTTPException, Request import json import os from pydantic import BaseModel from sqlalchemy.orm import Session from sqlalchemy import func, or_, text from typing import Optional from datetime import datetime, date from app.database import get_db from app.auth_middleware import require_role, require_auth from app.models import KPIDefinition, KPIValue, BudgetPlan, StrategicMap, KPIAlert, User, Subject, ActionPlan, OperationLog, ReportHistory from app.utils.deviation_engine import calc_period_deviation, calc_period_diff, calc_deviation, get_budget_for_kpi from app.utils.alert_entity_scope import month_period_clause from app.utils.profitability_params import load_profitability_params from app.utils.dupont_basis import ( annualize_factors, available_months, build_window_meta, check_identity, compute_factors, latest_stock_month, load_dupont_params, months_between, prev_month, resolve_window, stock_balance, sum_flows, ) from app.utils.voucher_guard import apply_voucher_guard from app.services import report_reclass from app.deps import get_entity_id, assert_entity_slug_scope, resolve_session_slug import logging logger = logging.getLogger("cma.reports") router = APIRouter(prefix="/api/cma/reports", tags=["管理报表"], dependencies=[Depends(require_role("ceo", "finance", "business"))], ) # ============================================================ # 报表1: 管理利润表 # ============================================================ @router.get("/profit-summary") def get_profit_summary( period: str = Query(None, description="格式 YYYY-MM"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润""" if period is None: period = datetime.now().strftime("%Y-%m") # 从KPI数据中获取各利润要素 def get_val(code: str): kpi = db.query(KPIDefinition).filter( KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() return v.actual_value if v else None revenue = get_val("F_REVENUE") gross_profit_rate = get_val("F_PROFIT_RATE") net_profit_rate = get_val("F_NET_PROFIT_RATE") cost_ratio = get_val("F_COST_RATIO") # 计算利润要素 # 营收已知,用毛利率算毛利,用成本率算成本 gross_profit = round(revenue * (gross_profit_rate / 100), 2) if revenue and gross_profit_rate else None total_cost = round(revenue * (cost_ratio / 100), 2) if revenue and cost_ratio else None net_profit = round(revenue * (net_profit_rate / 100), 2) if revenue and net_profit_rate else None # 边际贡献 ≈ 毛利(简化模型) contribution_margin = gross_profit # 固定成本 ≈ 总成本 - 变动成本(假设变动成本=营收*50%) variable_cost = round(revenue * 0.50, 2) if revenue else None fixed_cost = round(total_cost - variable_cost, 2) if total_cost and variable_cost else None # 找上期做环比 prev_year, prev_month = period.split("-") py, pm = int(prev_year), int(prev_month) pm -= 1 if pm <= 0: pm += 12 py -= 1 prev_period = f"{py}-{pm:02d}" def get_prev_val(code: str): kpi = db.query(KPIDefinition).filter( KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == prev_period ).order_by(KPIValue.id.desc()).first() return v.actual_value if v else None prev_revenue = get_prev_val("F_REVENUE") prev_gross_profit_rate = get_prev_val("F_PROFIT_RATE") prev_net_profit_rate = get_prev_val("F_NET_PROFIT_RATE") prev_cost_ratio = get_prev_val("F_COST_RATIO") prev_gross_profit = round(prev_revenue * (prev_gross_profit_rate / 100), 2) if prev_revenue and prev_gross_profit_rate else None prev_total_cost = round(prev_revenue * (prev_cost_ratio / 100), 2) if prev_revenue and prev_cost_ratio else None prev_net_profit = round(prev_revenue * (prev_net_profit_rate / 100), 2) if prev_revenue and prev_net_profit_rate else None prev_contribution_margin = prev_gross_profit prev_variable_cost = round(prev_revenue * 0.50, 2) if prev_revenue else None prev_fixed_cost = round(prev_total_cost - prev_variable_cost, 2) if prev_total_cost and prev_variable_cost else None def calc_chg(cur, prev): if cur is not None and prev is not None and prev != 0: return round((cur - prev) / prev * 100, 2) return None items = [ { "name": "营业收入", "value": revenue, "prev_value": prev_revenue, "change_rate": calc_chg(revenue, prev_revenue), "ratio": 100.0, }, { "name": "减:变动成本", "value": variable_cost, "prev_value": prev_variable_cost, "change_rate": calc_chg(variable_cost, prev_variable_cost), "ratio": round(variable_cost / revenue * 100, 2) if variable_cost and revenue else None, }, { "name": "= 边际贡献", "value": contribution_margin, "prev_value": prev_contribution_margin, "change_rate": calc_chg(contribution_margin, prev_contribution_margin), "ratio": round(contribution_margin / revenue * 100, 2) if contribution_margin and revenue else None, "is_subtotal": True, }, { "name": "减:固定成本", "value": fixed_cost, "prev_value": prev_fixed_cost, "change_rate": calc_chg(fixed_cost, prev_fixed_cost), "ratio": round(fixed_cost / revenue * 100, 2) if fixed_cost and revenue else None, }, { "name": "= 息税前利润", "value": net_profit, "prev_value": prev_net_profit, "change_rate": calc_chg(net_profit, prev_net_profit), "ratio": round(net_profit / revenue * 100, 2) if net_profit and revenue else None, "is_total": True, }, ] return { "period": period, "prev_period": prev_period, "items": items, } # ============================================================ # 报表2: 预算执行报告 # ============================================================ @router.get("/budget-execution") def get_budget_execution( period: str = Query(None, description="格式 YYYY-MM"), dimension: Optional[str] = Query(None), alert_level: Optional[str] = Query(None), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """预算执行报告 — 各KPI预算vs实际vs差异率(账套隔离 2026-08-31)""" if period is None: period = datetime.now().strftime("%Y-%m") query = db.query(KPIDefinition).filter(KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id) if dimension: query = query.filter(KPIDefinition.dimension == dimension) kpis = query.order_by(KPIDefinition.dimension, KPIDefinition.kpi_code).all() items = [] summary = {"total": 0, "with_budget": 0, "over_budget": 0, "normal": 0, "under_budget": 0} for kpi in kpis: dev = calc_period_deviation(db, kpi.id, period) if dev.get("actual_value") is None and dev.get("budget_value") is None: continue # 跳过完全无数据的KPI summary["total"] += 1 if dev.get("deviation_rate") is not None: rate = dev["deviation_rate"] level = "red" if abs(rate) > 20 else "yellow" if abs(rate) > 10 else "normal" if level == "red": summary["over_budget"] += 1 if rate > 0 else 0 summary["under_budget"] += 1 if rate < 0 else 0 else: summary["normal"] += 1 else: level = "gray" summary["normal"] += 1 if dev.get("budget_value") is not None: summary["with_budget"] += 1 items.append({ "kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name, "dimension": kpi.dimension, "unit": kpi.unit, "actual_value": dev.get("actual_value"), "budget_value": dev.get("budget_value"), "deviation_amount": dev.get("deviation_amount"), "deviation_rate": dev.get("deviation_rate"), "is_over_budget": dev.get("is_over_budget"), "alert_level": level, }) # alert_level 过滤 if alert_level: items = [i for i in items if i["alert_level"] == alert_level] return {"period": period, "summary": summary, "items": items} # ============================================================ # 报表3: KPI趋势报告 # ============================================================ @router.get("/kpi-trends") def get_kpi_trends( kpi_id: Optional[int] = Query(None), dimension: Optional[str] = Query(None), months: int = Query(12, ge=3, le=36), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """KPI趋势报告 — 选定KPI的历史趋势线(账套隔离 2026-08-31)""" query = db.query(KPIDefinition).filter(KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id) if kpi_id: query = query.filter(KPIDefinition.id == kpi_id) if dimension: query = query.filter(KPIDefinition.dimension == dimension) kpis = query.order_by(KPIDefinition.dimension, KPIDefinition.kpi_code).all() results = [] for kpi in kpis: values = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, month_period_clause(KPIValue.period), ).order_by(KPIValue.period.desc()).limit(months).all() values.reverse() trend = [{"period": v.period, "value": v.actual_value} for v in values] vals = [v.actual_value for v in values if v.actual_value is not None] target = kpi.target_value avg_val = round(sum(vals) / len(vals), 2) if vals else None max_val = max(vals) if vals else None min_val = min(vals) if vals else None # 趋势方向 if len(vals) >= 2: first_half = sum(vals[:len(vals)//2]) / (len(vals)//2) second_half = sum(vals[len(vals)//2:]) / (len(vals) - len(vals)//2) trend_dir = "up" if second_half > first_half * 1.05 else "down" if second_half < first_half * 0.95 else "stable" else: trend_dir = "stable" results.append({ "kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name, "dimension": kpi.dimension, "unit": kpi.unit, "target_value": target, "trend": trend, "trend_dir": trend_dir, "avg": avg_val, "max": max_val, "min": min_val, }) return {"data": results} # ============================================================ # 报表4: 四维度绩效评分卡 # ============================================================ DIM_CONFIG = { "finance": {"name": "财务维度", "icon": "💰", "color": "#409eff"}, "customer": {"name": "客户维度", "icon": "🤝", "color": "#67c23a"}, "process": {"name": "内部流程", "icon": "⚙️", "color": "#e6a23c"}, "learning": {"name": "学习成长", "icon": "📚", "color": "#f56c6c"}, } @router.get("/bsc-scorecard") def get_bsc_scorecard( map_id: Optional[int] = Query(None), period: Optional[str] = Query(None), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """四维度绩效评分卡 — BSC健康度(账套隔离 2026-08-31)""" if period is None: period = datetime.now().strftime("%Y-%m") # 取最新的已发布地图(当前企业) map_query = db.query(StrategicMap).filter(StrategicMap.status == "published", StrategicMap.entity_id == entity_id) if map_id: map_query = map_query.filter(StrategicMap.id == map_id) sm = map_query.order_by(StrategicMap.updated_at.desc()).first() if not sm: # 没有已发布地图,按维度聚合KPI return _build_scorecard_from_kpis(db, period, entity_id) # 从战略地图维度数据构建评分卡 dims = sm.dimensions if isinstance(dims, str): import json dims = json.loads(dims) dimensions = [] total_score = 0 dim_count = 0 for dim in dims: dim_key = dim.get("key", "") config = DIM_CONFIG.get(dim_key, {"name": dim.get("name", dim_key), "icon": "📊", "color": "#999"}) objectives = dim.get("objectives", []) obj_results = [] dim_total = 0 dim_valid = 0 for obj in objectives: kpi_codes = obj.get("kpis", []) kpi_scores = [] for code in kpi_codes: kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).first() if not kpi: continue v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() if v and v.actual_value and kpi.target_value: ratio = v.actual_value / kpi.target_value score = min(round(ratio * 100, 1), 100) level = "green" if ratio >= 0.9 else "yellow" if ratio >= 0.7 else "red" kpi_scores.append({"code": code, "name": kpi.kpi_name, "actual": v.actual_value, "target": kpi.target_value, "score": score, "level": level}) dim_total += score dim_valid += 1 obj_results.append({ "name": obj.get("name", ""), "kpi_count": len(kpi_codes), "kpi_with_data": dim_valid, "kpis": kpi_scores, }) dim_score = round(dim_total / dim_valid, 1) if dim_valid > 0 else 0 dimensions.append({ "key": dim_key, "name": config["name"], "icon": config["icon"], "color": config["color"], "score": dim_score, "objectives": obj_results, }) total_score += dim_score dim_count += 1 overall = round(total_score / dim_count, 1) if dim_count > 0 else 0 return { "period": period, "map_id": sm.id, "map_title": sm.title, "overall_score": overall, "dimensions": dimensions, } def _build_scorecard_from_kpis(db: Session, period: str, entity_id: int = 1) -> dict: """没有战略地图时,直接按维度聚合KPI算分(账套隔离 2026-08-31)""" kpis = db.query(KPIDefinition).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).all() dims: dict = {} for kpi in kpis: dim = kpi.dimension or "other" if dim not in dims: dims[dim] = {"kpis": [], "total_score": 0, "valid": 0} v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() score = None level = "gray" if v and v.actual_value and kpi.target_value: ratio = v.actual_value / kpi.target_value score = min(round(ratio * 100, 1), 100) level = "green" if ratio >= 0.9 else "yellow" if ratio >= 0.7 else "red" dims[dim]["total_score"] += score dims[dim]["valid"] += 1 dims[dim]["kpis"].append({ "code": kpi.kpi_code, "name": kpi.kpi_name, "actual": v.actual_value if v else None, "target": kpi.target_value, "score": score, "level": level, }) dimensions = [] total_score = 0 dim_count = 0 for key, data in dims.items(): config = DIM_CONFIG.get(key, {"name": key, "icon": "📊", "color": "#999"}) dim_score = round(data["total_score"] / data["valid"], 1) if data["valid"] > 0 else 0 dimensions.append({ "key": key, "name": config["name"], "icon": config["icon"], "color": config["color"], "score": dim_score, "objectives": [{"name": "全部KPI", "kpis": data["kpis"], "kpi_count": len(data["kpis"]), "kpi_with_data": data["valid"]}], }) total_score += dim_score dim_count += 1 return { "period": period, "map_id": None, "map_title": None, "overall_score": round(total_score / dim_count, 1) if dim_count > 0 else 0, "dimensions": dimensions, } # ============================================================ # 利润表: 新30号准则五板块结构 (2027) # ============================================================ # 科目编码 → 新30号准则板块映射(PRD第118-145行) NEW_STANDARD_MAP = { # 经营类 "6001": "operating", # 主营业务收入 "6051": "operating", # 其他业务收入 "6401": "operating", # 主营业务成本 "6402": "operating", # 其他业务成本 "6601": "operating", # 销售费用 "6602": "operating", # 管理费用 "660204": "operating_rd", # 研发费用(从管理费剥离) "6603": "operating_fx", # 经营汇兑损益 "6701": "operating", # 经营资产减值损失 # 投资类 "6011": "investing", # 利息收入(银行存款) "6111": "investing", # 投资收益 "611101": "investing", # 股权投资 "670101": "investing", # 投资类资产减值 # 筹资类 "660301": "financing", # 利息支出(借款) "660302": "financing_fx", # 筹资汇兑损益 # 所得税 "6801": "tax", # 所得税费用 # 终止经营 "6901": "discontinued", # 终止经营损益 } # 板块 → 展示信息 BLOCK_INFO = { "operating": { "name": "一、经营类损益", "short_name": "经营类", "items": [ {"code": "6001", "name": "营业收入", "sign": 1}, {"code": "6051", "name": "其他业务收入", "sign": 1}, {"code": "6401", "name": "减:营业成本", "sign": -1}, {"code": "6402", "name": "减:其他业务成本", "sign": -1}, {"code": "6601", "name": "减:销售费用", "sign": -1}, {"code": "6602", "name": "减:管理费用", "sign": -1}, {"code": "660204", "name": "减:研发费用", "sign": -1}, {"code": "6603", "name": "经营汇兑损益", "sign": 1}, {"code": "6701", "name": "减:经营资产减值损失", "sign": -1}, ], "result_key": "operating_profit", "result_name": "经营利润", }, "investing": { "name": "二、投资类损益", "short_name": "投资类", "items": [ {"code": "6011", "name": "利息收入", "sign": 1}, {"code": "6111", "name": "投资收益", "sign": 1}, {"code": "670101", "name": "减:投资类资产减值", "sign": -1}, ], "result_key": "investing_profit", "result_name": "投资净收益", }, "financing": { "name": "三、筹资类损益", "short_name": "筹资类", "items": [ {"code": "660301", "name": "减:利息支出", "sign": -1}, {"code": "660302", "name": "筹资汇兑损益", "sign": 1}, ], "result_key": "financing_profit", "result_name": "筹资费用净额", }, "tax": { "name": "四、所得税费用", "short_name": "所得税", "items": [ {"code": "6801", "name": "减:所得税费用", "sign": -1}, ], "result_key": "tax_profit", "result_name": "所得税费用", }, "discontinued": { "name": "五、终止经营损益", "short_name": "终止经营", "items": [ {"code": "6901", "name": "终止经营损益", "sign": 1}, ], "result_key": "discontinued_profit", "result_name": "终止经营损益", }, } def _get_subject_amount(db: Session, code: str, period: str, entity_id: int = 1) -> Optional[float]: """从 subjects + kpi_values 获取科目金额数据(账套隔离 2026-08-31)""" # 尝试从KPI数据获取(KPI编码与科目编码映射) kpi_code_map = { "6001": "F_REVENUE", "6051": "F_REVENUE_OTHER", "6401": "F_COST", "6402": "F_COST_OTHER", "6601": "F_SELLING_EXP", "6602": "F_ADMIN_EXP", "660204": "F_RD_EXP", "6603": "F_FINANCE_EXP", "6701": "F_IMPAIRMENT_LOSS", "6011": "F_INTEREST_INCOME", "6111": "F_INVEST_INCOME", "611101": "F_INVEST_INCOME", "660301": "F_INTEREST_EXP", "660302": "F_FX_LOSS", "6801": "F_TAX_EXP", "6901": "F_DISCONTINUED", } # 1. 优先从 kpi_values 取 if code in kpi_code_map: kpi_code = kpi_code_map[code] kpi = db.query(KPIDefinition).filter( KPIDefinition.kpi_code == kpi_code, KPIDefinition.entity_id == entity_id).first() if kpi: v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() if v and v.actual_value is not None: return float(v.actual_value) # 2. 从 subjects + voucher_details 取(如果存在) try: from app.models import VoucherDetail q = db.query( func.sum(VoucherDetail.debit_amount - VoucherDetail.credit_amount) ).filter( VoucherDetail.subject_code == code, VoucherDetail.period == period, VoucherDetail.entity_id == entity_id, # 账套隔离(2026-09-14 补) ) # 作废/撤单标记过滤(2026-09-14 收口) result = apply_voucher_guard(db, q, VoucherDetail).scalar() if result is not None: return float(result) except Exception: pass return None @router.get("/profit-statement") def get_profit_statement( period: str = Query(None, description="格式 YYYY-MM"), format: str = Query("old", description="old/new/dual"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """利润表 — 支持旧格式、新30号准则五板块格式、双列对比(账套隔离 2026-08-31)""" if period is None: period = datetime.now().strftime("%Y-%m") if format == "old": # 旧30号准则格式(保留兼容) return get_profit_summary(period=period, db=db, entity_id=entity_id) if format == "dual": # 双列对比:旧准则 vs 新准则 old_data = get_profit_summary(period=period, db=db, entity_id=entity_id) new_data = _build_new_format_profit(db, period, entity_id) return { "period": period, "format": "dual", "title": f"利润表双列对比({period})", "old_format": old_data, "new_format": new_data, } # === 新30号准则:五板块结构 === return _build_new_format_profit(db, period, entity_id) def _build_new_format_profit(db: Session, period: str, entity_id: int = 1) -> dict: """构建新30号准则五板块利润表(含附注明细)(账套隔离 2026-08-31)""" blocks = [] total_net_profit = 0 all_items_have_data = True for block_key in ["operating", "investing", "financing", "tax", "discontinued"]: block_cfg = BLOCK_INFO[block_key] items = [] block_subtotal = 0 block_has_data = False for item_cfg in block_cfg["items"]: amount = _get_subject_amount(db, item_cfg["code"], period, entity_id) if amount is not None: effective = amount * item_cfg["sign"] block_subtotal += effective block_has_data = True items.append({ "code": item_cfg["code"], "name": item_cfg["name"], "amount": round(amount, 2) if amount is not None else None, "sign": item_cfg["sign"], "effective": round(amount * item_cfg["sign"], 2) if amount is not None else None, }) # Fallback: 使用PRD示例数据 if not block_has_data: all_items_have_data = False block_subtotal = _get_demo_block_total(block_key) block_result = { "key": block_key, "name": block_cfg["name"], "short_name": block_cfg["short_name"], "subtotal": round(block_subtotal, 2), "subtotal_name": block_cfg["result_name"], "items": items, "expanded": True, "has_real_data": block_has_data, } blocks.append(block_result) total_net_profit += block_subtotal # 附注明细(对外法定报表披露要求) notes = _build_profit_notes(db, period, blocks, total_net_profit, entity_id) # 合计行:净利润 = 一二三+四+五 return { "period": period, "format": "new", "title": f"利润表 — 新30号准则({period})", "blocks": blocks, "net_profit": round(total_net_profit, 2), "net_profit_name": "净利润", "all_items_have_data": all_items_have_data, "notes": notes, "prev_period": None, # TODO: P1追溯调整 } def _build_profit_notes(db: Session, period: str, blocks: list, net_profit: float, entity_id: int = 1) -> dict: """利润表附注明细 — 收入/费用/财务费用拆解 + 板块勾稽 + 关键比率(账套隔离 2026-08-31)""" def amt(code): return _get_subject_amount(db, code, period, entity_id) revenue_main = amt("6001") revenue_other = amt("6051") revenue_total = None if revenue_main is not None or revenue_other is not None: revenue_total = round((revenue_main or 0) + (revenue_other or 0), 2) revenue_breakdown = [ {"name": "主营业务收入", "code": "6001", "amount": round(revenue_main, 2) if revenue_main is not None else None}, {"name": "其他业务收入", "code": "6051", "amount": round(revenue_other, 2) if revenue_other is not None else None}, ] expense_items = [ {"name": "营业成本", "code": "6401", "amount": amt("6401")}, {"name": "其他业务成本", "code": "6402", "amount": amt("6402")}, {"name": "销售费用", "code": "6601", "amount": amt("6601")}, {"name": "管理费用", "code": "6602", "amount": amt("6602")}, {"name": "研发费用", "code": "660204", "amount": amt("660204")}, {"name": "经营资产减值损失", "code": "6701", "amount": amt("6701")}, ] expense_breakdown = [ {"name": e["name"], "code": e["code"], "amount": round(e["amount"], 2) if e["amount"] is not None else None} for e in expense_items ] finance_breakdown = [ {"name": "利息收入(投资类)", "code": "6011", "amount": amt("6011")}, {"name": "投资收益(投资类)", "code": "6111", "amount": amt("6111")}, {"name": "利息支出(筹资类)", "code": "660301", "amount": amt("660301")}, {"name": "经营汇兑损益", "code": "6603", "amount": amt("6603")}, {"name": "筹资汇兑损益", "code": "660302", "amount": amt("660302")}, ] finance_breakdown = [ {"name": f["name"], "code": f["code"], "amount": round(f["amount"], 2) if f["amount"] is not None else None} for f in finance_breakdown ] # 板块勾稽(净利润 = 五板块之和) block_reconciliation = [ {"name": b["name"], "key": b["key"], "amount": b["subtotal"], "result_name": b["subtotal_name"]} for b in blocks ] # 关键比率 key_ratios = [] if revenue_total: key_ratios.append({ "name": "毛利率", "value": round((revenue_total - (amt("6401") or 0) - (amt("6402") or 0)) / revenue_total * 100, 2) if (amt("6401") is not None or amt("6402") is not None) else None, }) key_ratios.append({ "name": "净利率", "value": round(net_profit / revenue_total * 100, 2), }) rd = amt("660204") if rd is not None: key_ratios.append({"name": "研发费用率", "value": round(rd / revenue_total * 100, 2)}) else: key_ratios.append({"name": "毛利率", "value": None}) key_ratios.append({"name": "净利率", "value": None}) return { "revenue_total": round(revenue_total, 2) if revenue_total is not None else None, "revenue_breakdown": revenue_breakdown, "expense_breakdown": expense_breakdown, "finance_breakdown": finance_breakdown, "block_reconciliation": block_reconciliation, "net_profit": round(net_profit, 2), "key_ratios": key_ratios, } # ============================================================ # MPM管理层指标计算器 (P2) # ============================================================ # MPM指标类型定义 MPM_INDICATOR_TYPES = { "ebitda": { "name": "EBITDA", "description": "息税折旧摊销前利润", "base_label": "净利润", "default_adjustments": [ {"code": "tax", "name": "加:所得税费用", "sign": 1, "checked": True}, {"code": "interest", "name": "加:利息支出", "sign": 1, "checked": True}, {"code": "depreciation", "name": "加:折旧与摊销", "sign": 1, "checked": True}, {"code": "impairment", "name": "加:资产减值损失", "sign": 1, "checked": False}, ], }, "adjusted_net_profit": { "name": "调整后净利润", "description": "剔除非经常性项目后的可持续净利润", "base_label": "净利润(准则)", "default_adjustments": [ {"code": "impairment", "name": "加:资产减值损失", "sign": 1, "checked": True}, {"code": "equity_incentive", "name": "加:股权激励费用", "sign": 1, "checked": True}, {"code": "ma_cost", "name": "加:并购相关费用", "sign": 1, "checked": False}, {"code": "nonrecurring_income", "name": "减:非经常性投资收益", "sign": -1, "checked": True}, {"code": "asset_disposal", "name": "减:资产处置收益", "sign": -1, "checked": False}, {"code": "government_grant", "name": "减:政府补助", "sign": -1, "checked": False}, ], }, "free_cash_flow": { "name": "自由现金流", "description": "经营现金流扣除资本支出后的可自由支配现金流", "base_label": "经营现金流", "default_adjustments": [ {"code": "capex", "name": "减:资本支出", "sign": -1, "checked": True}, {"code": "working_capital", "name": "减:营运资本增加", "sign": -1, "checked": True}, {"code": "maintenance_capex", "name": "减:维护性资本支出", "sign": -1, "checked": False}, {"code": "dividend", "name": "加:股息收入", "sign": 1, "checked": False}, ], }, "custom": { "name": "自定义指标", "description": "自定义管理层指标", "base_label": "净利润", "default_adjustments": [ {"code": "adjustment_1", "name": "调整项目1", "sign": 1, "checked": False, "amount": None}, {"code": "adjustment_2", "name": "调整项目2", "sign": -1, "checked": False, "amount": None}, {"code": "adjustment_3", "name": "调整项目3", "sign": 1, "checked": False, "amount": None}, ], }, } # 调整项默认金额(从利润表自动取值映射) ADJUSTMENT_VALUE_MAP = { "tax": {"code": "6801", "sign_inverse": True}, # 所得税费用,加回需取绝对值 "interest": {"code": "660301", "sign_inverse": True}, "impairment": {"code": "6701", "sign_inverse": False}, # 资产减值损失本身是费用 "equity_incentive": None, # 无映射,需用户输入 "ma_cost": None, "nonrecurring_income": {"code": "6111", "sign_inverse": False}, "asset_disposal": None, "government_grant": None, "capex": None, "working_capital": None, "maintenance_capex": None, "dividend": None, } class MpmCalculateRequest(BaseModel): indicator_type: str = "adjusted_net_profit" period: str = None adjustments: list[dict] = None # [{code, name, sign, checked, amount}] @router.post("/mpm-calculate") def mpm_calculate( req: MpmCalculateRequest, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """MPM管理层指标计算器 — 生成合规调节表(账套隔离 2026-08-31)""" if req.period is None: req.period = datetime.now().strftime("%Y-%m") indicator_cfg = MPM_INDICATOR_TYPES.get(req.indicator_type) if not indicator_cfg: raise HTTPException(status_code=400, detail=f"不支持的指标类型: {req.indicator_type}") # 获取基准值:净利润 net_profit = _calc_new_net_profit(db, req.period, entity_id) if net_profit is None: net_profit = 0 # 经营现金流(自由现金流的基准) operating_cf = _get_kpi_val(db, "F_OPERATING_CF", req.period, entity_id) # 确定基准值 if req.indicator_type == "free_cash_flow": base_value = operating_cf or net_profit # fallback base_label = "经营现金流" else: base_value = net_profit base_label = indicator_cfg["base_label"] # 获取调整项(来自请求或默认) adjustments = req.adjustments if req.adjustments else indicator_cfg["default_adjustments"] # 自动填充调整项金额 reconciliation_items = [] running_total = base_value # 第一步:基准值 reconciliation_items.append({ "step": 0, "code": "_base", "name": base_label, "sign": 1, "amount": round(base_value, 2), "effective": round(base_value, 2), "is_base": True, "running_total": round(base_value, 2), }) for adj in adjustments: code = adj.get("code", "") checked = adj.get("checked", False) sign = adj.get("sign", 1) name = adj.get("name", "") amount = adj.get("amount") # 尝试自动取值 if amount is None and checked: amount = _get_adjustment_value(db, code, req.period, entity_id) effective = round(amount * sign, 2) if amount is not None else None item = { "step": len(reconciliation_items), "code": code, "name": name, "sign": sign, "amount": round(amount, 2) if amount is not None else None, "effective": effective, "checked": checked, "is_base": False, "running_total": None, } if checked and effective is not None: running_total += effective item["running_total"] = round(running_total, 2) reconciliation_items.append(item) # 最终结果 final_value = round(running_total, 2) return { "indicator_type": req.indicator_type, "indicator_name": indicator_cfg["name"], "indicator_desc": indicator_cfg["description"], "period": req.period, "base_value": round(base_value, 2), "base_label": base_label, "final_value": final_value, "final_label": indicator_cfg["name"], "adjustment_count": sum(1 for a in adjustments if a.get("checked", False)), "total_adjustments": len(adjustments), "reconciliation_items": reconciliation_items, "has_real_data": net_profit != 0, } def _get_kpi_val(db: Session, code: str, period: str, entity_id: int = 1) -> Optional[float]: """从KPI定义+值获取数值(账套隔离 2026-08-31)""" kpi = db.query(KPIDefinition).filter( KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() return float(v.actual_value) if v and v.actual_value is not None else None def _calc_new_net_profit(db: Session, period: str, entity_id: int = 1) -> Optional[float]: """计算新30号准则下的净利润(账套隔离 2026-08-31)""" total = 0 has_data = False for block_key in ["operating", "investing", "financing", "tax", "discontinued"]: block_cfg = BLOCK_INFO[block_key] for item_cfg in block_cfg["items"]: amount = _get_subject_amount(db, item_cfg["code"], period, entity_id) if amount is not None: total += amount * item_cfg["sign"] has_data = True if not has_data: return None return round(total, 2) def _get_adjustment_value(db: Session, adj_code: str, period: str, entity_id: int = 1) -> Optional[float]: """获取调整项的自动取值(账套隔离 2026-08-31)""" mapping = ADJUSTMENT_VALUE_MAP.get(adj_code) if mapping is None: return None # 需要用户输入 code = mapping["code"] amount = _get_subject_amount(db, code, period, entity_id) if amount is None: return None # sign_inverse: 如果调整项是"加回"费用,费用本身的sign是负的(在利润表中是减项) # 但在MPM调节中,加回费用取绝对值 return abs(amount) if mapping.get("sign_inverse", False) else amount def _get_demo_block_total(block_key: str) -> float: """PRD示例数据 fallback""" demo = { "operating": -567883, "investing": 123456, "financing": -98765, "tax": -43210, "discontinued": 0, } return demo.get(block_key, 0) # ============================================================ # 追溯调整: 2026年数据按新30号准则重述 (P1) # ============================================================ @router.get("/restatement") def get_restatement( period: str = Query(None, description="格式 YYYY-MM"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """2026年数据按新准则重述 — 旧口径vs新口径双列对比,自动标记调整项(账套隔离 2026-08-31)""" if period is None: period = datetime.now().strftime("%Y-%m") # 旧口径数据 (传统利润表项目) def _old_kpi_val(code: str): kpi = db.query(KPIDefinition).filter( KPIDefinition.kpi_code == code, KPIDefinition.entity_id == entity_id).first() if not kpi: return None v = db.query(KPIValue).filter( KPIValue.kpi_id == kpi.id, KPIValue.period == period ).order_by(KPIValue.id.desc()).first() return float(v.actual_value) if v and v.actual_value is not None else None old_revenue = _old_kpi_val("F_REVENUE") old_revenue_other = _old_kpi_val("F_REVENUE_OTHER") old_cost = _old_kpi_val("F_COST") old_cost_other = _old_kpi_val("F_COST_OTHER") old_selling_exp = _old_kpi_val("F_SELLING_EXP") old_admin_exp = _old_kpi_val("F_ADMIN_EXP") old_finance_exp = _old_kpi_val("F_FINANCE_EXP") old_rd_exp = _old_kpi_val("F_RD_EXP") # 新口径数据 (从科目映射或kpi_values获取) new_revenue = _get_subject_amount(db, "6001", period, entity_id) new_revenue_other = _get_subject_amount(db, "6051", period, entity_id) new_cost = _get_subject_amount(db, "6401", period, entity_id) new_cost_other = _get_subject_amount(db, "6402", period, entity_id) new_selling = _get_subject_amount(db, "6601", period, entity_id) new_admin = _get_subject_amount(db, "6602", period, entity_id) new_rd = _get_subject_amount(db, "660204", period, entity_id) new_interest_income = _get_subject_amount(db, "6011", period, entity_id) new_interest_exp = _get_subject_amount(db, "660301", period, entity_id) new_fx = _get_subject_amount(db, "6603", period, entity_id) new_fx_financing = _get_subject_amount(db, "660302", period, entity_id) new_invest_income = _get_subject_amount(db, "6111", period, entity_id) new_impairment = _get_subject_amount(db, "6701", period, entity_id) new_invest_impairment = _get_subject_amount(db, "670101", period, entity_id) new_tax = _get_subject_amount(db, "6801", period, entity_id) new_discontinued = _get_subject_amount(db, "6901", period, entity_id) # 旧口径汇总计算 old_operating_items = [ {"name": "营业收入", "old_value": old_revenue, "category": "operating"}, {"name": "其他业务收入", "old_value": old_revenue_other, "category": "operating"}, {"name": "减:营业成本", "old_value": old_cost, "category": "operating"}, {"name": "减:其他业务成本", "old_value": old_cost_other, "category": "operating"}, {"name": "减:销售费用", "old_value": old_selling_exp, "category": "operating"}, {"name": "减:管理费用", "old_value": old_admin_exp, "category": "operating"}, {"name": "减:财务费用", "old_value": old_finance_exp, "category": "financing"}, ] # 新口径项目明细 new_items_map = [ # 经营类 {"name": "营业收入", "new_value": new_revenue, "category": "operating"}, {"name": "其他业务收入", "new_value": new_revenue_other, "category": "operating"}, {"name": "减:营业成本", "new_value": new_cost, "category": "operating"}, {"name": "减:其他业务成本", "new_value": new_cost_other, "category": "operating"}, {"name": "减:销售费用", "new_value": new_selling, "category": "operating"}, {"name": "减:管理费用(不含研发)", "new_value": new_admin, "category": "operating"}, {"name": "减:研发费用", "new_value": new_rd, "category": "operating"}, {"name": "减:经营资产减值损失", "new_value": new_impairment, "category": "operating"}, {"name": "经营汇兑损益", "new_value": new_fx, "category": "operating"}, # 投资类 {"name": "利息收入", "new_value": new_interest_income, "category": "investing"}, {"name": "投资收益", "new_value": new_invest_income, "category": "investing"}, {"name": "减:投资类资产减值", "new_value": new_invest_impairment, "category": "investing"}, # 筹资类 {"name": "减:利息支出", "new_value": new_interest_exp, "category": "financing"}, {"name": "筹资汇兑损益", "new_value": new_fx_financing, "category": "financing"}, # 所得税 {"name": "减:所得税费用", "new_value": new_tax, "category": "tax"}, # 终止经营 {"name": "终止经营损益", "new_value": new_discontinued, "category": "discontinued"}, ] # 构建对比行 items = [] adjusted_count = 0 new_count = 0 unchanged_count = 0 # 先处理旧口径中存在的项目,匹配新口径 old_new_mapping = { "营业收入": "营业收入", "其他业务收入": "其他业务收入", "减:营业成本": "减:营业成本", "减:其他业务成本": "减:其他业务成本", "减:销售费用": "减:销售费用", "减:管理费用": "减:管理费用(不含研发)", } for old_item in old_operating_items: name = old_item["name"] old_val = old_item["old_value"] new_name = old_new_mapping.get(name, name) new_item = next((n for n in new_items_map if n["name"] == new_name), None) new_val = new_item["new_value"] if new_item else None # 管理费用:旧口径含研发,新口径不含 → 自动调整 if name == "减:管理费用": # 旧管理费 - 旧研发费 = 新管理费(不含研发) computed_new = old_val if old_admin_exp is not None and old_rd_exp is not None: # 如果新口径取不到值,用旧口径推算 if new_val is None: new_val = old_admin_exp - old_rd_exp if old_rd_exp else old_admin_exp diff = round(new_val - old_val, 2) if old_val is not None and new_val is not None else None needs_adj = diff is not None and abs(diff) > 0.01 if needs_adj: adjusted_count += 1 items.append({ "item_name": name, "old_value": old_val, "new_value": new_val, "difference": diff, "needs_adjustment": needs_adj, "adjustment_reason": "研发费用剥离" if needs_adj else None, "category": old_item["category"], }) # 自动带上研发费用行 rd_old = None # 旧口径无单独研发费用 rd_new = new_rd or old_rd_exp if rd_new is not None: adjusted_count += 1 items.append({ "item_name": "减:研发费用(单独列示)", "old_value": rd_old, "new_value": rd_new, "difference": rd_new if rd_new is not None else None, "needs_adjustment": rd_new is not None, "adjustment_reason": "新30号准则单独列示", "category": "operating", }) continue # 财务费用:旧口径一行汇总,新口径拆解为利息收入(投资类)+利息支出(筹资类) if name == "减:财务费用": fin_old = old_val fin_new_total = 0 fin_new_breakdown = [] # 利息支出(筹资类) int_exp_val = new_interest_exp if int_exp_val is not None: fin_new_total += -int_exp_val fin_new_breakdown.append({ "item_name": "减:利息支出(筹资类)", "old_value": None, "new_value": int_exp_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "financing", }) adjusted_count += 1 # 筹资汇兑损益 fx_fin_val = new_fx_financing if fx_fin_val is not None: fin_new_total += fx_fin_val fin_new_breakdown.append({ "item_name": "筹资汇兑损益", "old_value": None, "new_value": fx_fin_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "financing", }) adjusted_count += 1 # 利息收入(投资类) int_inc_val = new_interest_income if int_inc_val is not None: fin_new_breakdown.append({ "item_name": "利息收入(投资类)", "old_value": None, "new_value": int_inc_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "investing", }) adjusted_count += 1 # 经营汇兑损益(经营类) fx_op_val = new_fx if fx_op_val is not None: fin_new_breakdown.append({ "item_name": "经营汇兑损益(经营类)", "old_value": None, "new_value": fx_op_val, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解", "category": "operating", }) adjusted_count += 1 items.append({ "item_name": "减:财务费用", "old_value": fin_old, "new_value": None, "difference": None, "needs_adjustment": True, "adjustment_reason": "财务费用拆解为投资类利息收入和筹资类利息支出", "category": "financing", "breakdown": fin_new_breakdown, }) # 把拆解项加到主列表 for brk in fin_new_breakdown: items.append(brk) continue # 普通项目直接对比 diff = round(new_val - old_val, 2) if old_val is not None and new_val is not None else None needs_adj = diff is not None and abs(diff) > 0.01 if needs_adj: adjusted_count += 1 else: unchanged_count += 1 items.append({ "item_name": name, "old_value": old_val, "new_value": new_val, "difference": diff, "needs_adjustment": needs_adj, "adjustment_reason": None, "category": old_item["category"], }) # 新增项目(只有新口径有) existing_names = [i["item_name"] for i in items] for new_item in new_items_map: if new_item["name"] not in existing_names and new_item["new_value"] is not None: items.append({ "item_name": new_item["name"], "old_value": None, "new_value": new_item["new_value"], "difference": None, "needs_adjustment": True, "adjustment_reason": "新30号准则新增项目", "category": new_item["category"], }) new_count += 1 adjusted_count += 1 # 计算旧口径净利润和新口径净利润 def _calc_net_profit_old(): """旧口径净利润 (简化)""" rev = old_revenue or 0 rev_other = old_revenue_other or 0 c = old_cost or 0 c_other = old_cost_other or 0 sell = old_selling_exp or 0 admin = old_admin_exp or 0 fin = old_finance_exp or 0 return rev + rev_other - c - c_other - sell - admin - fin def _calc_net_profit_new(): """新口径净利润 = 经营利润 + 投资净收益 + 筹资净费用 + 所得税 + 终止经营""" op_items = ["6001", "6051", "6401", "6402", "6601", "6602", "660204", "6603", "6701"] inv_items = ["6011", "6111", "670101"] fin_items = ["660301", "660302"] tax_items = ["6801"] dis_items = ["6901"] total = 0 for codes in [op_items, inv_items, fin_items, tax_items, dis_items]: for code in codes: amt = _get_subject_amount(db, code, period, entity_id) if amt is not None: # 根据BLOCK_INFO中的sign处理 for bk in BLOCK_INFO.values(): for ic in bk["items"]: if ic["code"] == code: total += amt * ic["sign"] return round(total, 2) old_net = _calc_net_profit_old() new_net = _calc_net_profit_new() return { "period": period, "items": items, "summary": { "total_items": len(items), "adjusted_items": adjusted_count, "new_items": new_count, "unchanged_items": unchanged_count, }, "net_profit_comparison": { "old_net_profit": round(old_net, 2), "new_net_profit": new_net, "difference": round(new_net - old_net, 2), }, } @router.get("/category-map") def get_category_map( db: Session = Depends(get_db), ): """返回科目→新30号准则板块映射 豁免多租户隔离(2026-08-31):Subject 为全局会计科目字典(无 entity_id 列), 返回的是科目分类映射常量,非企业业务数据,故不做 entity 过滤""" subjects_data = db.query(Subject).filter(Subject.is_active == 1).order_by(Subject.subject_code).all() map_list = [] for s in subjects_data: if s.new_standard_category: map_list.append({ "subject_code": s.subject_code, "subject_name": s.subject_name, "category": s.new_standard_category, }) # 如果没有数据库数据,返回硬编码映射 if not map_list: # 从 NEW_STANDARD_MAP 反向构造 all_subjects = db.query(Subject).filter(Subject.is_active == 1).all() subj_map = {s.subject_code: s.subject_name for s in all_subjects} for code, cat in NEW_STANDARD_MAP.items(): map_list.append({ "subject_code": code, "subject_name": subj_map.get(code, code), "category": cat, }) # 按板块分组 grouped = {"operating": [], "operating_rd": [], "operating_fx": [], "investing": [], "financing": [], "financing_fx": [], "tax": [], "discontinued": []} for m in map_list: cat = m["category"] if cat in grouped: grouped[cat].append(m) else: grouped.setdefault(cat, []).append(m) return { "mapping": NEW_STANDARD_MAP, "subjects": map_list, "grouped": grouped, "total": len(map_list), } # ============================================================ # 对外法定报表 — 新30号准则适配 (2027) # 利润表(五板块+附注) / 资产负债表(新准则科目分类) / 现金流量表(三活动) # ============================================================ # 资产负债表行项目: (编码列表[(code, sign)], 名称, 板块key, 新准则分类, 是否合计行) # 新准则分类: operating经营 / investing投资 / financing筹资 / equity权益 BALANCE_SHEET_SECTIONS = [ { "key": "current_assets", "name": "流动资产", "category_label": "经营资产", "lines": [ {"codes": [("1001", 1), ("1002", 1), ("1012", 1)], "name": "货币资金", "ns_category": "operating"}, {"codes": [("1101", 1)], "name": "交易性金融资产", "ns_category": "investing"}, {"codes": [("1122", 1)], "name": "应收账款", "ns_category": "operating"}, {"codes": [("1123", 1)], "name": "预付账款", "ns_category": "operating"}, {"codes": [("1131", 1)], "name": "应收股利", "ns_category": "investing"}, {"codes": [("1221", 1)], "name": "其他应收款", "ns_category": "operating"}, {"codes": [("1405", 1)], "name": "存货", "ns_category": "operating"}, ], }, { "key": "non_current_assets", "name": "非流动资产", "category_label": "投资资产", "lines": [ {"codes": [("1511", 1)], "name": "长期股权投资", "ns_category": "investing"}, {"codes": [("1501", 1)], "name": "持有至到期投资", "ns_category": "investing"}, {"codes": [("1601", 1), ("1602", -1)], "name": "固定资产净额", "ns_category": "operating"}, {"codes": [("1701", 1)], "name": "无形资产", "ns_category": "operating"}, # 长期待摊费用(1801):小企业准则项目池含该项,原模板漏列 → 致勾稽差额 128,758.24(2026-09-16 补) {"codes": [("1801", 1)], "name": "长期待摊费用", "ns_category": "operating"}, ], }, { "key": "current_liabilities", "name": "流动负债", "category_label": "经营负债", "lines": [ {"codes": [("2001", 1)], "name": "短期借款", "ns_category": "financing"}, {"codes": [("2202", 1)], "name": "应付账款", "ns_category": "operating"}, {"codes": [("2203", 1)], "name": "预收账款", "ns_category": "operating"}, {"codes": [("2211", 1)], "name": "应付职工薪酬", "ns_category": "operating"}, {"codes": [("2221", 1)], "name": "应交税费", "ns_category": "operating"}, {"codes": [("2241", 1)], "name": "其他应付款", "ns_category": "operating"}, ], }, { "key": "non_current_liabilities", "name": "非流动负债", "category_label": "筹资负债", "lines": [ {"codes": [("2501", 1)], "name": "长期借款", "ns_category": "financing"}, {"codes": [("2502", 1)], "name": "应付债券", "ns_category": "financing"}, ], }, { "key": "equity", "name": "所有者权益", "category_label": "所有者权益", # M7(规则表 v1.1 §三):权益行科目集 = {3001,3002,3101,3103,3104,3105} # (含 4001/4002/4101/4103/4104 别名)。账套实际记账用小企业准则 3xxx 科目, # 别名只参与取数、不参与行 key(避免打破按 key 的预算映射)。缺 3001 会漏实收资本 2,000,000。 "lines": [ {"codes": [("4001", 1)], "alias_codes": [("3001", 1)], "name": "实收资本", "ns_category": "equity"}, {"codes": [("4002", 1)], "alias_codes": [("3002", 1)], "name": "资本公积", "ns_category": "equity"}, {"codes": [("4101", 1)], "alias_codes": [("3101", 1)], "name": "盈余公积", "ns_category": "equity"}, # alias_codes:模板用 4103/4104,智享通凭证明细实际用 3103/3104(2026-09-16 补) {"codes": [("4103", 1), ("4104", 1)], "alias_codes": [("3103", 1), ("3104", 1), ("3105", 1)], "name": "未分配利润", "ns_category": "equity"}, ], }, ] # 资产负债表示例数据(博海科技, 期末/期初, 单位: 千元) BS_DEMO = { "1001|1002|1012": {"end": 850, "begin": 780}, "1101": {"end": 250, "begin": 220}, "1122": {"end": 1100, "begin": 1050}, "1123": {"end": 180, "begin": 160}, "1131": {"end": 120, "begin": 100}, "1221": {"end": 90, "begin": 80}, "1405": {"end": 930, "begin": 900}, "1511": {"end": 580, "begin": 550}, "1501": {"end": 200, "begin": 180}, "1601|1602": {"end": 1220, "begin": 1150}, "1701": {"end": 130, "begin": 120}, "2001": {"end": 1500, "begin": 1300}, "2202": {"end": 1050, "begin": 980}, "2203": {"end": 280, "begin": 250}, "2211": {"end": 140, "begin": 130}, "2221": {"end": 90, "begin": 85}, "2241": {"end": 60, "begin": 50}, "2501": {"end": 800, "begin": 750}, "2502": {"end": 270, "begin": 250}, "4001": {"end": 500, "begin": 500}, "4002": {"end": 180, "begin": 175}, "4103|4104": {"end": 780, "begin": 820}, } BS_CATEGORY_CN = { "operating": "经营类", "investing": "投资类", "financing": "筹资类", "equity": "权益类", } def _prev_period_str(period: str) -> str: """上一期间 YYYY-MM → YYYY-(MM-1)""" try: y, m = period.split("-") y, m = int(y), int(m) m -= 1 if m <= 0: m += 12 y -= 1 return f"{y}-{m:02d}" except Exception: return period def _get_bs_amount(db: Session, codes: list, period: str, entity_id: int = 1) -> Optional[float]: """资产负债表科目余额 — 优先凭证明细,无数据返回 None(账套隔离 2026-08-31) ⚠️ 保留给现金流量表/预编报表使用(精确匹配 + 单期口径);资产负债表本体走 `app/services/report_reclass.py`(前缀匹配 + 期末余额口径 + 取数层重分类)。""" total = 0.0 has_data = False try: from app.models import VoucherDetail for code, sign in codes: q = db.query( func.sum(VoucherDetail.debit_amount - VoucherDetail.credit_amount) ).filter( VoucherDetail.subject_code == code, VoucherDetail.period == period, VoucherDetail.entity_id == entity_id, # 账套隔离(2026-09-14 补) ) # 作废/撤单标记过滤(2026-09-14 收口) result = apply_voucher_guard(db, q, VoucherDetail).scalar() if result is not None: total += float(result) * sign has_data = True except Exception: pass return round(total, 2) if has_data else None def _bs_line_amount(db: Session, line: dict, period: str, column: str = "end", entity_id: int = 1) -> dict: """单行:凭证数据优先,否则示例数据(column: end期末 / begin期初)(账套隔离 2026-08-31)""" real = _get_bs_amount(db, line["codes"], period, entity_id) if real is not None: return {"value": real, "is_demo": False} key = "|".join(c for c, _ in line["codes"]) demo = BS_DEMO.get(key) if demo: return {"value": demo.get(column, demo["end"]), "is_demo": True} return {"value": None, "is_demo": True} # ── 资产负债表取数口径(2026-09-16 重分类落地) ── # end = 期末余额(累计至期间,含期初/结转)——资产负债表定义口径,**默认** # period = 本期发生额(旧报表口径,兼容回退;可用 CMA_BS_BASIS_DEFAULT / ?basis= 切换) _BS_BASIS_DEFAULT = os.getenv("CMA_BS_BASIS_DEFAULT", "end") def _bs_basis(basis) -> str: """口径解析:兼容内部直调(未传值时 FastAPI 的 Query 默认对象会漏进来 → 取默认口径)""" b = basis if isinstance(basis, str) else _BS_BASIS_DEFAULT b = (b or "end").strip().lower() return "period" if b == "period" else "end" def _bs_analysis(db: Session, entity_id: int, period: str, basis: str, include_unmapped: bool = True) -> dict: """取数层重分类分析(规则表 §二 R1~R12;调表不调账,不生成凭证)""" return report_reclass.analyze( db, entity_id, period, basis=basis, template_sections=BALANCE_SHEET_SECTIONS, include_unmapped=include_unmapped, ) @router.get("/balance-sheet") def get_balance_sheet( period: str = Query(None, description="格式 YYYY-MM"), basis: Optional[str] = Query(None, description="取数口径 end=期末余额(默认) / period=本期发生额"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """资产负债表 — 取数层重分类(R1~R12,调表不调账)+ 新30号准则科目分类(账套隔离 2026-08-31) · 余额方向:资产 Σ(借−贷) / 负债·权益 Σ(贷−借) → 正常余额一律为正数 · 往来·税费类反方向余额按规则转正(转入对方科目),报表不再出现负数项目 · 有凭证明细时**不再混入示例数据**(展示口径与落库口径同源) """ if period is None: period = datetime.now().strftime("%Y-%m") use_basis = _bs_basis(basis) prev_period = _prev_period_str(period) plan = _bs_analysis(db, entity_id, period, use_basis) has_real = bool(plan["has_voucher_data"]) # M2:期初未补 → 方向判定不可信 → 重分类结论只进底稿诊断,不落地到对外行项目 diag_only = bool(plan.get("direction_check_unreliable")) plan_prev = _bs_analysis(db, entity_id, prev_period, use_basis, include_unmapped=False) if has_real else None def _present_key(pl: Optional[dict]) -> str: """M2:期初未补(方向判定不可信)时,对外行项目用**原始余额**(不呈现"重分类转正")""" return "raw" if (pl or {}).get("direction_check_unreliable") else "after" def _line_value(pl: Optional[dict], key: str, col: str): if pl is None: return None return pl[_present_key(pl)].get(key) # 规则落位项目(模板未列报的目标项)→ 按板块挂载虚拟行 virtual: dict = {} for key, item in plan["items"].items(): if item["template"] or item.get("blocked"): continue end_v = plan[_present_key(plan)].get(key) begin_v = plan_prev[_present_key(plan_prev)].get(key) if plan_prev else None if not end_v and not begin_v: continue virtual.setdefault(item["section_key"], []).append({ "name": item["name"], "item_key": key, "ns_category": item.get("ns_category") or "operating", "ns_category_label": BS_CATEGORY_CN.get(item.get("ns_category") or "operating", ""), "end_value": round(float(end_v or 0), 2), "begin_value": round(float(begin_v), 2) if begin_v is not None else None, "is_demo": False, "is_reclass_line": True, "template_gap": True, "note": item.get("note", ""), }) sections = [] total_assets_end = total_assets_begin = 0 total_liab_end = total_liab_begin = 0 total_equity_end = total_equity_begin = 0 all_real = has_real for sec in BALANCE_SHEET_SECTIONS: lines = [] sec_end = sec_begin = 0.0 sec_real = False for line in sec["lines"]: key = "|".join(c for c, _ in line["codes"]) if has_real: end = {"value": _line_value(plan, key, "end"), "is_demo": False} begin = {"value": _line_value(plan_prev, key, "begin"), "is_demo": False} else: end = _bs_line_amount(db, line, period, column="end", entity_id=entity_id) begin = _bs_line_amount(db, line, prev_period, column="begin", entity_id=entity_id) if end["is_demo"] or begin["is_demo"]: all_real = False if end["value"] is not None: sec_end += end["value"] sec_real = True if begin["value"] is not None: sec_begin += begin["value"] raw_key = plan["raw"].get(key) diag_adj = None if raw_key is None else round( float(plan["after"].get(key) or 0) - float(raw_key), 2) lines.append({ "name": line["name"], "item_key": key, "ns_category": line["ns_category"], "ns_category_label": BS_CATEGORY_CN.get(line["ns_category"], line["ns_category"]), "end_value": round(end["value"], 2) if end["value"] is not None else None, "begin_value": round(begin["value"], 2) if begin["value"] is not None else None, "is_demo": end["is_demo"] or begin["is_demo"], "is_reclass_line": False, "template_gap": False, "raw_value": None if raw_key is None else round(float(raw_key), 2), # 对外调整额按呈现口径计算:期初未补(方向判定不可信)时不落地 → 0 "reclass_adjustment": None if raw_key is None else round( float(_line_value(plan, key, "end") or 0) - float(raw_key), 2), # 诊断用调整额(无论是否落地都给出,供底稿/审计追溯) "diagnostic_reclass_adjustment": diag_adj, }) # 规则落位项目(虚拟行) for vline in virtual.get(sec["key"], []): if vline["end_value"] is not None and sec["key"] in ( "current_assets", "non_current_assets", "current_liabilities", "non_current_liabilities"): sec_end += vline["end_value"] if vline["begin_value"] is not None and sec["key"] in ( "current_assets", "non_current_assets", "current_liabilities", "non_current_liabilities"): sec_begin += vline["begin_value"] lines.append(vline) # 归属汇总 if sec["key"] in ("current_assets", "non_current_assets"): total_assets_end += sec_end total_assets_begin += sec_begin elif sec["key"] in ("current_liabilities", "non_current_liabilities"): total_liab_end += sec_end total_liab_begin += sec_begin else: total_equity_end += sec_end total_equity_begin += sec_begin sections.append({ "key": sec["key"], "name": sec["name"], "category_label": sec["category_label"], "subtotal_end": round(sec_end, 2), "subtotal_begin": round(sec_begin, 2), "lines": lines, "has_real_data": sec_real, }) inv = plan["invariants"] if has_real: balanced = abs(total_assets_end - total_liab_end - total_equity_end) < 0.01 and \ abs(total_assets_begin - total_liab_begin - total_equity_begin) < 0.01 else: balanced = abs(total_assets_end - total_liab_end - total_equity_end) < 0.01 return { "period": period, "prev_period": prev_period, "basis": use_basis, "title": f"资产负债表 — 新30号准则({period})", "sections": sections, "totals": { "assets": {"end": round(total_assets_end, 2), "begin": round(total_assets_begin, 2)}, "liabilities": {"end": round(total_liab_end, 2), "begin": round(total_liab_begin, 2)}, "equity": {"end": round(total_equity_end, 2), "begin": round(total_equity_begin, 2)}, "liab_equity": {"end": round(total_liab_end + total_equity_end, 2), "begin": round(total_liab_begin + total_equity_begin, 2)}, "balanced": balanced, "identity_gap": round(total_assets_end - total_liab_end - total_equity_end, 2), }, # ── 重分类(调表不调账)── "reclass": { # M2:期初未补 → 方向判定不可信 → 重分类结论只作底稿诊断,不落地到对外行项目 "applied": bool(plan["entries"]) and not diag_only, "diagnostic_only": diag_only, "direction_check_unreliable": diag_only, "caliber_warning": plan.get("caliber_warning") or "", "opening": plan.get("opening"), "rule_version": plan["rule_version"], "basis": plan["basis"], "has_voucher_data": has_real, "entries": plan["entries"], "invariants": inv, "negatives": plan["negatives"], "template_gaps": plan["template_gaps"], "residual_reported": plan.get("residual_reported") or [], "unmapped_subjects": plan["unmapped_subjects"], "pl_unclosed": plan["pl_unclosed"], "working_paper_url": f"/api/cma/reports/balance-sheet/reclass-working-paper?period={period}&basis={use_basis}", }, "all_items_have_data": all_real, } @router.get("/balance-sheet/reclass-working-paper") def get_reclass_working_paper( period: str = Query(None, description="格式 YYYY-MM"), basis: Optional[str] = Query(None, description="取数口径 end=期末余额(默认) / period=本期发生额"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """重分类底稿(审计可追溯)— 源科目/余额/转正额/目标项 + 不变量 + 诊断(账套隔离)""" if period is None: period = datetime.now().strftime("%Y-%m") use_basis = _bs_basis(basis) plan = _bs_analysis(db, entity_id, period, use_basis) return { "period": period, "basis": use_basis, "entity_id": entity_id, "rule_version": plan["rule_version"], "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "accounting_method": "调表不调账(报表取数层重分类,不生成记账凭证)", # ── M1/M2:期初层 + 方向判定可信度(期初未补 → 只作诊断)── "direction_check_unreliable": plan.get("direction_check_unreliable", False), "caliber_warning": plan.get("caliber_warning") or "", "opening": plan.get("opening"), "diagnostic_only": bool(plan.get("direction_check_unreliable")), "rules": plan["rules"], "entries": plan["entries"], "invariants": plan["invariants"], "template_gaps": plan["template_gaps"], "residual_reported": plan.get("residual_reported") or [], "negatives": plan["negatives"], "unmapped_subjects": plan["unmapped_subjects"], "pl_unclosed": plan["pl_unclosed"], "totals_before": plan["totals_before"], "totals_after": plan["totals_after"], } @router.get("/balance-sheet/reclass-working-paper/export") def export_reclass_working_paper( period: str = Query(None, description="格式 YYYY-MM"), basis: Optional[str] = Query(None, description="取数口径 end=期末余额(默认) / period=本期发生额"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """重分类底稿导出(Excel)— 底稿明细 + 诊断 + 规则表(账套隔离)""" if period is None: period = datetime.now().strftime("%Y-%m") use_basis = _bs_basis(basis) plan = _bs_analysis(db, entity_id, period, use_basis) from io import BytesIO from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from fastapi.responses import StreamingResponse from urllib.parse import quote wb = Workbook() head_fill = PatternFill("solid", fgColor="305496") head_font = Font(color="FFFFFF", bold=True, size=11) def _sheet(ws, title, headers, rows): ws.cell(row=1, column=1, value=title).font = Font(bold=True, size=13, color="305496") ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=max(len(headers), 1)) for ci, h in enumerate(headers, start=1): c = ws.cell(row=2, column=ci, value=h) c.font = head_font c.fill = head_fill c.alignment = Alignment(horizontal="center") for ri, row in enumerate(rows, start=3): for ci, h in enumerate(headers, start=1): ws.cell(row=ri, column=ci, value=row.get(h)) for ci, h in enumerate(headers, start=1): ws.column_dimensions[ws.cell(row=2, column=ci).column_letter].width = max(12, min(46, len(str(h)) * 2 + 10)) e_rows = report_reclass.entry_rows(plan) warn = (" ⚠️ 期初未补,方向判定不可信(重分类结论仅诊断,不落地)" if plan.get("direction_check_unreliable") else "") ws1 = wb.active ws1.title = "重分类底稿" if e_rows: _sheet(ws1, f"资产负债表重分类底稿(调表不调账) {period} 账套 entity_id={entity_id} 口径basis={use_basis} {plan['rule_version']}{warn}", list(e_rows[0].keys()), e_rows) else: ws1.cell(row=1, column=1, value=f"本期({period})无触发的重分类规则{warn}").font = Font(bold=True) d_rows = report_reclass.diagnostic_rows(plan) ws2 = wb.create_sheet("不变量与诊断") _sheet(ws2, "重分类不变量 / 负数项目 / 模板待核 / 未列报科目", ["项目", "值", "说明"], d_rows) ws3 = wb.create_sheet("规则表") _sheet(ws3, f"CMA 资产负债表重分类规则表 {plan['rule_version']}", ["code", "source_code", "source_name", "source_side", "target_item", "target_key", "target_side", "status", "note"], plan["rules"]) buf = BytesIO() wb.save(buf) buf.seek(0) filename = f"资产负债表重分类底稿_{period}_entity{entity_id}.xlsx" return StreamingResponse( buf, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(filename)}"}, ) # 现金流量表行项目: (code, 名称, 板块, 方向, kpi_code可选) CASH_FLOW_LINES = [ # 经营活动 {"code": "CF01", "name": "销售商品、提供劳务收到的现金", "section": "operating", "sign": 1, "kpi_code": None}, {"code": "CF02", "name": "收到的税费返还", "section": "operating", "sign": 1, "kpi_code": None}, {"code": "CF03", "name": "收到其他与经营活动有关的现金", "section": "operating", "sign": 1, "kpi_code": None}, {"code": "CF04", "name": "购买商品、接受劳务支付的现金", "section": "operating", "sign": -1, "kpi_code": None}, {"code": "CF05", "name": "支付给职工以及为职工支付的现金", "section": "operating", "sign": -1, "kpi_code": None}, {"code": "CF06", "name": "支付的各项税费", "section": "operating", "sign": -1, "kpi_code": None}, {"code": "CF07", "name": "支付其他与经营活动有关的现金", "section": "operating", "sign": -1, "kpi_code": None}, # 投资活动 {"code": "CF08", "name": "收回投资收到的现金", "section": "investing", "sign": 1, "kpi_code": None}, {"code": "CF09", "name": "取得投资收益收到的现金", "section": "investing", "sign": 1, "kpi_code": None}, {"code": "CF10", "name": "处置固定资产、无形资产等收回的现金", "section": "investing", "sign": 1, "kpi_code": None}, {"code": "CF11", "name": "购建固定资产、无形资产等支付的现金", "section": "investing", "sign": -1, "kpi_code": None}, {"code": "CF12", "name": "投资支付的现金", "section": "investing", "sign": -1, "kpi_code": None}, # 筹资活动 {"code": "CF13", "name": "吸收投资收到的现金", "section": "financing", "sign": 1, "kpi_code": None}, {"code": "CF14", "name": "取得借款收到的现金", "section": "financing", "sign": 1, "kpi_code": None}, {"code": "CF15", "name": "偿还债务支付的现金", "section": "financing", "sign": -1, "kpi_code": None}, {"code": "CF16", "name": "分配股利、利润或偿付利息支付的现金", "section": "financing", "sign": -1, "kpi_code": None}, ] # 现金流量表示例数据(2026-06, 单位: 千元, 与利润表/KPI口径一致) CF_DEMO = { "CF01": 5200, "CF02": 0, "CF03": 120, "CF04": -3150, "CF05": -820, "CF06": -360, "CF07": -140, "CF08": 200, "CF09": 50, "CF10": 30, "CF11": -330, "CF12": -100, "CF13": 0, "CF14": 500, "CF15": -200, "CF16": -150, } CF_DEMO_FX = 0 # 汇率变动对现金的影响 CF_DEMO_BEGIN = 1200 # 期初现金及现金等价物余额 def _get_cf_amount(db: Session, line: dict, period: str, entity_id: int = 1) -> dict: """现金流量表行项目 — 优先KPI/凭证,否则示例数据(账套隔离 2026-08-31)""" # 经营净额行特殊处理:优先取 F_OP_CFLOW if line.get("kpi_code"): kpi_val = _get_kpi_val(db, line["kpi_code"], period, entity_id) if kpi_val is not None: return {"value": round(kpi_val, 2), "is_demo": False} real = _get_bs_amount(db, [(line["code"], line["sign"])], period, entity_id) if real is not None: return {"value": real, "is_demo": False} demo = CF_DEMO.get(line["code"]) if demo is not None: return {"value": float(demo), "is_demo": True} return {"value": None, "is_demo": True} @router.get("/cash-flow") def get_cash_flow_statement( period: str = Query(None, description="格式 YYYY-MM"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """现金流量表 — 经营/投资/筹资三活动(新30号准则直接法)(账套隔离 2026-08-31)""" if period is None: period = datetime.now().strftime("%Y-%m") section_cfg = [ {"key": "operating", "name": "一、经营活动产生的现金流量", "short": "经营活动"}, {"key": "investing", "name": "二、投资活动产生的现金流量", "short": "投资活动"}, {"key": "financing", "name": "三、筹资活动产生的现金流量", "short": "筹资活动"}, ] sections = [] net_by_section = {} all_real = True for sc in section_cfg: lines = [] subtotal = 0.0 sec_real = False for line in CASH_FLOW_LINES: if line["section"] != sc["key"]: continue v = _get_cf_amount(db, line, period, entity_id) if v["is_demo"]: all_real = False if v["value"] is not None: subtotal += v["value"] sec_real = True lines.append({ "code": line["code"], "name": line["name"], "value": round(v["value"], 2) if v["value"] is not None else None, "is_demo": v["is_demo"], }) net_by_section[sc["key"]] = round(subtotal, 2) sections.append({ "key": sc["key"], "name": sc["name"], "short": sc["short"], "net": round(subtotal, 2), "lines": lines, "has_real_data": sec_real, }) # 经营净额行优先取 KPI F_OP_CFLOW(真实数据优先) op_kpi = _get_kpi_val(db, "F_OP_CFLOW", period, entity_id) if op_kpi is not None: sections[0]["net"] = round(op_kpi, 2) net_by_section["operating"] = round(op_kpi, 2) fx = _get_kpi_val(db, "F_FX_LOSS", period, entity_id) if fx is None: fx = CF_DEMO_FX fx_demo = True else: fx_demo = False begin_cash = CF_DEMO_BEGIN net_increase = round(net_by_section.get("operating", 0) + net_by_section.get("investing", 0) + net_by_section.get("financing", 0) + float(fx or 0), 2) end_cash = round(begin_cash + net_increase, 2) return { "period": period, "title": f"现金流量表 — 新30号准则({period})", "sections": sections, "fx_effect": {"name": "四、汇率变动对现金及现金等价物的影响", "value": round(float(fx), 2), "is_demo": fx_demo}, "summary": { "net_increase": net_increase, "begin_cash": begin_cash, "end_cash": end_cash, }, "all_items_have_data": all_real, } @router.get("/statutory") def get_statutory_reports( period: str = Query(None, description="格式 YYYY-MM"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """对外法定报表(新30号准则)— 利润表+资产负债表+现金流量表 组合视图""" if period is None: period = datetime.now().strftime("%Y-%m") return { "period": period, "title": f"对外法定报表 — 新30号准则({period})", "profit": _build_new_format_profit(db, period, entity_id), "balance_sheet": get_balance_sheet(period=period, db=db, entity_id=entity_id), "cash_flow": get_cash_flow_statement(period=period, db=db, entity_id=entity_id), "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } @router.get("/statutory/export") def export_statutory_reports( period: str = Query(None, description="格式 YYYY-MM"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """导出对外法定报表(新30号准则)— Excel 三表合一""" if period is None: period = datetime.now().strftime("%Y-%m") data = get_statutory_reports(period=period, db=db, entity_id=entity_id) from io import BytesIO from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from fastapi.responses import StreamingResponse wb = Workbook() head_fill = PatternFill("solid", fgColor="305496") head_font = Font(color="FFFFFF", bold=True, size=12) section_fill = PatternFill("solid", fgColor="D9E1F2") section_font = Font(bold=True, size=11) total_fill = PatternFill("solid", fgColor="FCE4D6") total_font = Font(bold=True, size=11) thin = Side(style="thin", color="BFBFBF") border = Border(left=thin, right=thin, top=thin, bottom=thin) demo_font = Font(color="D46B08", size=9, italic=True) def style_header(ws, row, ncols, title): ws.merge_cells(start_row=row, start_column=1, end_row=row, end_column=ncols) c = ws.cell(row=row, column=1, value=title) c.font = Font(bold=True, size=14, color="305496") c.alignment = Alignment(horizontal="center", vertical="center") ws.row_dimensions[row].height = 26 def write_row(ws, r, values, font=None, fill=None, use_border=True): for ci, v in enumerate(values, start=1): c = ws.cell(row=r, column=ci, value=v) if font: c.font = font if fill: c.fill = fill if use_border: c.border = border return r + 1 # ── Sheet 1: 利润表(五板块+附注) ── ws1 = wb.active ws1.title = "利润表-新30号准则" style_header(ws1, 1, 4, f"利润表(新30号准则五板块) {period} 单位: 元") r = write_row(ws1, 2, ["板块", "项目", "科目编码", "本期金额"], head_font, head_fill) profit = data["profit"] for block in profit.get("blocks", []): r = write_row(ws1, r, [block["name"], block.get("subtotal_name", ""), "", block["subtotal"]], section_font, section_fill) for item in block.get("items", []): r = write_row(ws1, r, ["", item["name"], item["code"], item["effective"]]) r = write_row(ws1, r, ["合计", "净利润", "", profit.get("net_profit")], total_font, total_fill) # 附注明细 notes = profit.get("notes") or {} r += 1 r = write_row(ws1, r, ["附注一、收入构成", "", "", ""], section_font, section_fill) for n in notes.get("revenue_breakdown", []): r = write_row(ws1, r, ["", n["name"], n["code"], n["amount"]]) r = write_row(ws1, r, ["", "营业收入合计", "", notes.get("revenue_total")], total_font) r = write_row(ws1, r, ["附注二、费用构成", "", "", ""], section_font, section_fill) for n in notes.get("expense_breakdown", []): r = write_row(ws1, r, ["", n["name"], n["code"], n["amount"]]) r = write_row(ws1, r, ["附注三、财务费用拆解", "", "", ""], section_font, section_fill) for n in notes.get("finance_breakdown", []): r = write_row(ws1, r, ["", n["name"], n["code"], n["amount"]]) r = write_row(ws1, r, ["附注四、板块勾稽(净利润=五板块之和)", "", "", ""], section_font, section_fill) for n in notes.get("block_reconciliation", []): r = write_row(ws1, r, ["", n["name"], "", n["amount"]]) r = write_row(ws1, r, ["", "净利润", "", notes.get("net_profit")], total_font, total_fill) r = write_row(ws1, r, ["附注五、关键比率", "", "", ""], section_font, section_fill) for n in notes.get("key_ratios", []): r = write_row(ws1, r, ["", n["name"], "", n["value"]]) for col, w in zip("ABCD", [28, 40, 14, 18]): ws1.column_dimensions[col].width = w # ── Sheet 2: 资产负债表 ── ws2 = wb.create_sheet("资产负债表-新30号准则") bs = data["balance_sheet"] style_header(ws2, 1, 5, f"资产负债表(新30号准则科目分类) {period} 单位: 元") r = write_row(ws2, 2, ["项目", "新准则分类", "期末余额", "期初余额", "数据来源"], head_font, head_fill) for sec in bs.get("sections", []): r = write_row(ws2, r, [sec["name"], sec.get("category_label", ""), sec["subtotal_end"], sec["subtotal_begin"], "小计"], section_font, section_fill) for line in sec.get("lines", []): src = "示例" if line.get("is_demo") else "凭证" r = write_row(ws2, r, [line["name"], line.get("ns_category_label", ""), line["end_value"], line["begin_value"], src]) t = bs.get("totals", {}) r = write_row(ws2, r, ["资产总计", "", t["assets"]["end"], t["assets"]["begin"], ""], total_font, total_fill) r = write_row(ws2, r, ["负债合计", "", t["liabilities"]["end"], t["liabilities"]["begin"], ""], total_font, total_fill) r = write_row(ws2, r, ["所有者权益合计", "", t["equity"]["end"], t["equity"]["begin"], ""], total_font, total_fill) r = write_row(ws2, r, ["负债和所有者权益总计", "", t["liab_equity"]["end"], t["liab_equity"]["begin"], ""], total_font, total_fill) r = write_row(ws2, r, ["勾稽校验(资产=负债+权益)", "", "✓ 平衡" if t.get("balanced") else "✗ 不平", "", ""], total_font, total_fill) rc = bs.get("reclass") or {} if rc: inv = rc.get("invariants") or {} r = write_row(ws2, r, ["重分类(调表不调账)", rc.get("rule_version", ""), f"触发 {inv.get('entries_count', 0)} 条 / 转正额 {inv.get('reclass_total', 0)}", ("basis=end(期末余额口径)" if rc.get("basis") == "end" else f"basis={rc.get('basis')}(本期发生额口径)"), "不生成凭证"], section_font, section_fill) if inv: r = write_row(ws2, r, ["不变量:Δ资产=Δ负债 / 权益不变", "", "PASS" if inv.get("delta_assets_equals_delta_liabilities") and inv.get("equity_unchanged") else "FAIL", f"Δ资产={inv.get('assets_delta', 0)},Δ负债={inv.get('liabilities_delta', 0)}", ""]) for col, w in zip("ABCDE", [36, 16, 18, 18, 12]): ws2.column_dimensions[col].width = w # ── Sheet 3: 现金流量表 ── ws3 = wb.create_sheet("现金流量表-新30号准则") cf = data["cash_flow"] style_header(ws3, 1, 4, f"现金流量表(新30号准则三活动) {period} 单位: 元") r = write_row(ws3, 2, ["项目", "行次", "本期金额", "数据来源"], head_font, head_fill) for sec in cf.get("sections", []): r = write_row(ws3, r, [sec["name"], "", "", "小计"], section_font, section_fill) for line in sec.get("lines", []): src = "示例" if line.get("is_demo") else "凭证" r = write_row(ws3, r, [line["name"], line["code"], line["value"], src]) r = write_row(ws3, r, [f"{sec['name']}净额", "", sec["net"], ""], total_font, total_fill) fx = cf.get("fx_effect", {}) r = write_row(ws3, r, [fx.get("name", ""), "", fx.get("value"), "示例" if fx.get("is_demo") else "凭证"], section_font, section_fill) s = cf.get("summary", {}) r = write_row(ws3, r, ["现金及现金等价物净增加额", "", s.get("net_increase"), ""], total_font, total_fill) r = write_row(ws3, r, ["加:期初现金及现金等价物余额", "", s.get("begin_cash"), ""], total_font, total_fill) r = write_row(ws3, r, ["期末现金及现金等价物余额", "", s.get("end_cash"), ""], total_font, total_fill) for col, w in zip("ABCD", [46, 10, 18, 12]): ws3.column_dimensions[col].width = w # ── Sheet 4: 重分类底稿(调表不调账,审计可追溯) ── ws4 = wb.create_sheet("重分类底稿") style_header(ws4, 1, 12, f"资产负债表重分类底稿(调表不调账) {period} 单位: 元") e_rows = report_reclass.entry_rows(bs.get("reclass") and {"entries": bs["reclass"]["entries"]} or {"entries": []}) if e_rows: r = write_row(ws4, 2, list(e_rows[0].keys()), head_font, head_fill) for row in e_rows: r = write_row(ws4, r, list(row.values())) else: r = write_row(ws4, 2, ["本期无触发的重分类规则"], section_font, section_fill) r += 1 r = write_row(ws4, r, ["不变量与诊断", "", "", "", "", "", "", "", "", "", "", ""], section_font, section_fill) plan_like = { "entries": (bs.get("reclass") or {}).get("entries") or [], "invariants": (bs.get("reclass") or {}).get("invariants") or {}, "negatives": (bs.get("reclass") or {}).get("negatives") or {"before": [], "after": [], "after_in_scope": []}, "template_gaps": (bs.get("reclass") or {}).get("template_gaps") or [], "unmapped_subjects": (bs.get("reclass") or {}).get("unmapped_subjects") or [], # M1/M2:期初层 + 方向判定可信度(底稿 Sheet 同源暴露) "residual_reported": (bs.get("reclass") or {}).get("residual_reported") or [], "direction_check_unreliable": (bs.get("reclass") or {}).get("direction_check_unreliable", False), "caliber_warning": (bs.get("reclass") or {}).get("caliber_warning", ""), } for row in report_reclass.diagnostic_rows(plan_like): r = write_row(ws4, r, [row.get("项目"), row.get("值"), row.get("说明")]) for col, w in zip("ABC", [42, 20, 60]): ws4.column_dimensions[col].width = w buf = BytesIO() wb.save(buf) buf.seek(0) from urllib.parse import quote filename = f"对外法定报表_新30号准则_{period}.xlsx" return StreamingResponse( buf, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(filename)}"}, ) # ============================================================ # 杜邦分析 — 区间取值(流量累计 × 存量平均 × 可选年化 + 恒等式守护) # 口径裁定: 财务Bot 2026-10-03(任总 2026-10-03 09:49 CST DM「立」) # 任务件: finance-20261003-dupont-period-basis-design-fullstack.md # 单一事实源: app/utils/dupont_basis.py(区间解析 / 取数 / 纯计算 / 恒等式守护) # 取数源(§二.5 注): 库内 subject_balances(存量)+ voucher_details(流量); # **不用**旧脚本 register_dupont_kpis.py 的 ERP 直连近似口径(缺应收/存货) # ============================================================ DUPONT_SLUG_TO_ID = {"hanke": 1, "bohai": 2} DUPONT_KPI_CODES = ("F_ASSET_TOTAL", "F_EQUITY_TOTAL", "F_EQUITY_MULTIPLIER", "F_ASSET_TURNOVER", "F_ROE", "F_ROA") def _dupont_window_values(db: Session, entity_id: int, caliber: dict, start: str, end: str) -> dict: """按任务件 §二 取一个区间的原值(万元)与因子(未年化)。 · 流量(损益类) = 区间**累计发生额** · 存量(资产负债类)= 区间**平均余额** = (期初 + 期末) / 2,一级科目(`level=1`,防跨层双计) · 期初缺失 ⇒ **显式**回落期末(`basis_fallback=True` + 原因),**禁静默**(§二.6) """ equity_prefixes = list(caliber["equity_prefixes"]) asset_prefix = caliber["asset_prefix"] ta_end = stock_balance(db, entity_id, end, [asset_prefix], "net") eq_end = stock_balance(db, entity_id, end, equity_prefixes, "net_rev") if start == end: ta_begin, eq_begin = ta_end.get("begin"), eq_end.get("begin") begin_hint = f"{start} 期行内期初列 ob_*(月初)" else: ta_start = stock_balance(db, entity_id, start, [asset_prefix], "net") eq_start = stock_balance(db, entity_id, start, equity_prefixes, "net_rev") ta_begin = ta_start.get("begin") if ta_start.get("row_found") else None eq_begin = eq_start.get("begin") if eq_start.get("row_found") else None begin_hint = f"{start} 期行内期初列 ob_*(期初)" fallback, reason = False, None if ta_end.get("end") is None or eq_end.get("end") is None: fallback, reason = True, f"期末存量缺失({end} 期无 subject_balances 一级科目行)" elif ta_begin is None or eq_begin is None: fallback = True reason = f"期初存量缺失({begin_hint} 取不到) ⇒ 显式回落期末口径,非静默" if fallback: ta_avg, eq_avg = ta_end.get("end"), eq_end.get("end") else: ta_avg = (ta_begin + ta_end["end"]) / 2.0 eq_avg = (eq_begin + eq_end["end"]) / 2.0 revenue = sum_flows(db, entity_id, start, end, {"codes": caliber["revenue_codes"], "field": "credit", "carry_forward": 0}) if caliber["net_profit_mode"] == "carry_forward": # 酣客(旧准则):期末结转腿口径 —— 与已上线 verified 的月度 F_NET_PROFIT/F_ROE/F_ROA 同源 net_profit = sum_flows(db, entity_id, start, end, {"prefixes": equity_prefixes, "field": "credit_debit", "carry_forward": 1}) else: # 博海(新 30 号准则):损益类科目净额口径(该账套 voucher_details 无结转腿) net_profit = sum_flows(db, entity_id, start, end, {"prefixes": caliber["pnl_prefixes"], "field": "credit_debit", "carry_forward": 0}) def w(v): return None if v is None else round(v / 10000.0, 6) raw = { "net_profit": w(net_profit), "revenue": w(revenue), "total_assets": w(ta_avg), "equity": w(eq_avg), "total_assets_begin": w(ta_begin), "total_assets_end": w(ta_end.get("end")), "equity_begin": w(eq_begin), "equity_end": w(eq_end.get("end")), } factors = compute_factors(raw["net_profit"], raw["revenue"], raw["total_assets"], raw["equity"]) return {"raw": raw, "factors": factors, "basis": "ending" if fallback else "average", "basis_fallback": fallback, "basis_fallback_reason": reason, "net_profit_mode": caliber["net_profit_mode"], "caliber": {"asset_prefix": asset_prefix, "equity_prefixes": equity_prefixes, "revenue_codes": list(caliber["revenue_codes"]), "pnl_prefixes": list(caliber["pnl_prefixes"])}} def _dupont_prev_window(start: str, months: int) -> tuple: """同长度上一区间:[prev_start, prev_end]""" prev_end = prev_month(start) prev_start = start for _ in range(1, months): prev_start = prev_month(prev_start) prev_start = prev_month(prev_start) return prev_start, prev_end def _dupont_kpi_refs(db: Session, entity_id: int, period: str) -> dict: """同口径 KPI 层镜像(derived 单一事实源):F_ASSET_TOTAL/F_EQUITY_TOTAL/ F_EQUITY_MULTIPLIER/F_ASSET_TURNOVER/F_ROE/F_ROA 在该期间的最新行。""" defs = {d.kpi_code: d for d in db.query(KPIDefinition).filter( KPIDefinition.entity_id == entity_id, KPIDefinition.kpi_code.in_(list(DUPONT_KPI_CODES)), ).all()} ids = [d.id for d in defs.values()] refs: dict = {} if not ids: return refs rows = db.query(KPIValue).filter( KPIValue.kpi_id.in_(ids), KPIValue.period == period, KPIValue.actual_value.isnot(None), ).order_by(KPIValue.id.desc()).all() by_id = {d.id: d for d in defs.values()} for row in rows: kpi = by_id.get(row.kpi_id) if kpi is None or kpi.kpi_code in refs: continue refs[kpi.kpi_code] = {"kpi_id": kpi.id, "kpi_code": kpi.kpi_code, "kpi_name": kpi.kpi_name, "unit": kpi.unit, "value": row.actual_value, "period": row.period, "source_type": row.source_type, "data_status": row.data_status, "data_source_type": kpi.data_source_type} return refs def _dupont_factor_obj(name: str, label: str, desc: str, value, unit: str, scale: float, annualized: bool, win: dict, basis: str, raw: dict, status: str, assessment: str, display_dp: int) -> dict: return { "value": None if value is None else round(value, display_dp), "value_exact": None if value is None else round(float(value), 6), "unit": unit, "label": label, "desc": desc, "basis": basis, "period_start": win["period_start"], "period_end": win["period_end"], "annualized": bool(annualized), "scale_applied": round(scale, 6), "raw": raw, "status": status, "assessment": assessment, } @router.get("/dupont") def get_dupont_analysis( request: Request, entity: Optional[str] = Query(None, description="hanke/bohai(须与会话账套一致);缺省按会话账套"), period: Optional[str] = Query(None, description="区间维度:ytd(默认)/ttm/YYYY-MM/YYYY-Qn/YYYY-Hn/YYYY/custom"), start: Optional[str] = Query(None, description="自定义区间起(YYYY-MM,需配 end)"), end: Optional[str] = Query(None, description="自定义区间止(YYYY-MM)"), annualize: bool = Query(False, description="流量年化 ×12/区间月数(比率型因子缩放见 scale_applied)"), guard_probe: Optional[str] = Query(None, description="恒等式守护判别力自检(只读诊断):mismatch=故意错配区间"), db: Session = Depends(get_db), ): """杜邦分析 — ROE 三级拆解(区间取值设计,2026-10-03) 多租户口径(不变):分析对象 = 会话账套;显式 entity 与会话不一致 → 403; 无会话 token(Bot/存量脚本通道)→ 缺省 bohai。 取值规则(任务件 §二): 流量(损益类)= 区间累计发生额;存量(资产负债类)= 区间平均余额 (期初+期末)/2; 跨期比较 = 年化 = 流量 × 12 / 区间月数(净利率/权益乘数为同量纲之比 ⇒ 年化相消)。 返回必带口径元数据:`period_start` / `period_end` / `basis` / `annualized`(§二.4)。 恒等式(§二.3):|ROE − 净利率×周转率×权益乘数| ≤ 0.05 百分点;不成立 ⇒ 只返回 `identity_broken` + 三因子各自的区间元数据,**不返回数值**。 """ slug = assert_entity_slug_scope(request, entity) if slug is None: slug = resolve_session_slug(request, "bohai") entity_id = DUPONT_SLUG_TO_ID.get(slug) if entity_id is None: raise HTTPException(status_code=400, detail=f"不支持的实体 slug={slug}(支持 hanke/bohai)") caliber = load_dupont_params(db, entity_id) avail = available_months(db, entity_id) anchor = latest_stock_month(db, entity_id) win = resolve_window(period, start, end, avail, anchor=anchor) if win.get("error"): raise HTTPException(status_code=400, detail=win["error"]) name_row = db.execute( text("SELECT name FROM entities WHERE id = :eid"), {"eid": entity_id}).first() entity_name = (name_row[0] if name_row else None) or ("陕西酣客文化传媒" if slug == "hanke" else "陕西博海网络科技") suffix = {"ytd": "(YTD 年初至今)", "ttm": "(TTM 滚动 12 月)", "month": "(单月)", "quarter": "(季度)", "half": "(半年)", "year": "(全年)", "custom": "(自定义区间)"} label = win["label"] + suffix.get(win["period_key"], "") cur = _dupont_window_values(db, entity_id, caliber, win["period_start"], win["period_end"]) ann = annualize_factors(dict(cur["factors"]), win["months"], annualize) f_ann = ann["factors"] scale = ann["scale"] identity = check_identity(f_ann.get("roe"), f_ann.get("net_profit_margin"), f_ann.get("asset_turnover"), f_ann.get("equity_multiplier")) mismatch_note = None if (guard_probe or "").strip().lower() == "mismatch": # 守护判别力自检(只读):ROE 取**请求区间**,三因子取**另一区间**(多期请求 → 区间末单月; # 单月请求 → 该月 YTD),即人为制造任务件所指「同一数据两份结果」的**区间错配** # ⇒ 三因子乘积 ≠ ROE,应判 identity_broken(证明守护不是摆设)。 if win["period_start"] != win["period_end"]: alt_start = alt_end = win["period_end"] else: alt_start, alt_end = f"{win['period_start'][:4]}-01", win["period_start"] alt_win = _dupont_window_values(db, entity_id, caliber, alt_start, alt_end) alt_months = months_between(alt_start, alt_end) alt = annualize_factors(dict(alt_win["factors"]), alt_months, annualize)["factors"] identity = check_identity(f_ann.get("roe"), alt.get("net_profit_margin"), alt.get("asset_turnover"), alt.get("equity_multiplier")) identity["probe"] = "mismatch" identity["probe_roe_window"] = [win["period_start"], win["period_end"]] identity["probe_factor_window"] = [alt_start, alt_end] mismatch_note = (f"守护自检:ROE 取 [{win['period_start']}..{win['period_end']}]," f"三因子取 [{alt_start}..{alt_end}](故意错配区间)") meta = build_window_meta(label, win["period_start"], win["period_end"], win["months"], cur["basis"], annualize, cur["basis_fallback"], scale, cur["basis_fallback_reason"]) # 因子元数据(无论恒等式是否成立都随值返回 —— §二.3 便于定位漂移源) factors = { "net_profit_margin": _dupont_factor_obj( "net_profit_margin", "净利润率", "净利润 ÷ 营业收入(同区间流量之比)", f_ann.get("net_profit_margin"), "%", ann["per_factor_scale"]["net_profit_margin"], annualize, win, "cumulative", {"net_profit": cur["raw"]["net_profit"], "revenue": cur["raw"]["revenue"]}, ("🔴" if (f_ann.get("net_profit_margin") or 0) < 0 else ("🟡" if (f_ann.get("net_profit_margin") or 0) < 5 else "✅")), ("净利率为负,区间内亏损经营" if (f_ann.get("net_profit_margin") or 0) < 0 else "区间盈利水平"), 2), "asset_turnover": _dupont_factor_obj( "asset_turnover", "总资产周转率", "营业收入 ÷ 平均总资产(区间流量 ÷ 区间平均存量)", f_ann.get("asset_turnover"), "次", ann["per_factor_scale"]["asset_turnover"], annualize, win, "average", {"revenue": cur["raw"]["revenue"], "total_assets": cur["raw"]["total_assets"]}, "🟡" if (f_ann.get("asset_turnover") or 0) < 1 else "✅", "单位资产创造的营业收入(区间口径)", 4), "equity_multiplier": _dupont_factor_obj( "equity_multiplier", "权益乘数", "平均总资产 ÷ 平均净资产(同区间存量之比)", f_ann.get("equity_multiplier"), "倍", ann["per_factor_scale"]["equity_multiplier"], annualize, win, "average", {"total_assets": cur["raw"]["total_assets"], "equity": cur["raw"]["equity"]}, "✅" if (f_ann.get("equity_multiplier") or 0) < 3 else "🟡", "财务杠杆水平(区间平均口径)", 4), } if identity.get("status") == "identity_broken": return { "entity": slug, "entity_id": entity_id, "entity_name": entity_name, "identity_broken": True, **meta, "months_label": f"{win['months']}个月", "identity": identity, "guard_probe_note": mismatch_note, "factors": factors, "raw_data": cur["raw"], "caliber": cur["caliber"], "kpi_refs": _dupont_kpi_refs(db, entity_id, win["period_end"]), "message": ("恒等式不成立:|ROE − 净利率×周转率×权益乘数| > 容差 —— 三因子区间口径错配," "不返回数值,请按 identity / factors[*].period_start|basis 定位漂移源"), "readonly": True, "data_source": "subject_balances + voucher_details", } # 上一同长度区间(仅用于趋势对比;越出可用月域 ⇒ None,不静默编数) prev_roe = None ps, pe = _dupont_prev_window(win["period_start"], win["months"]) if avail and ps >= avail[0]: prev = _dupont_window_values(db, entity_id, caliber, ps, pe) prev_roe = prev["factors"]["roe"] roe_exact = f_ann.get("roe") roe_change = None if (roe_exact is None or prev_roe is None) else round(roe_exact - prev_roe, 4) def _fmt(v): # 存量/流量缺失时 raw 值为 None(如博海存量仅到区间外)⇒ 显式「-」,不得 None 格式化 return "-" if v is None else f"{v:.2f}" detail = (f"{label}口径(流量累计 {win['period_start']}~{win['period_end']}," f"存量{'平均' if cur['basis'] == 'average' else '期末回落'}" f"{',年化 ×12/' + str(win['months']) if annualize else ',非年化'}):" f"净利润 {_fmt(cur['raw']['net_profit'])}万 / 营业收入 {_fmt(cur['raw']['revenue'])}万 / " f"平均总资产 {_fmt(cur['raw']['total_assets'])}万 / 平均净资产 {_fmt(cur['raw']['equity'])}万;" f"ROE = {factors['net_profit_margin']['value']}% × {factors['asset_turnover']['value']} × " f"{factors['equity_multiplier']['value']} = {None if roe_exact is None else round(roe_exact, 2)}%" f"(恒等式 {'通过' if identity.get('status') == 'pass' else identity.get('status')}," f"偏差 {identity.get('diff')} ≤ 容差 {identity.get('tolerance')})") if cur["basis_fallback"]: detail += f"|⚠️ {cur['basis_fallback_reason']}(basis_fallback=true)" return { "entity": slug, "entity_id": entity_id, "entity_name": entity_name, "identity_broken": False, **meta, "months_label": f"{win['months']}个月", "flow_basis": "cumulative", "period_key": win["period_key"], "clip_note": win.get("clip_note"), "roe": None if roe_exact is None else round(roe_exact, 2), "roe_exact": roe_exact, "roe_change": roe_change, "roe_trend": (None if roe_change is None else ("up" if roe_change > 0 else ("down" if roe_change < 0 else "stable"))), "prev_roe": None if prev_roe is None else round(prev_roe, 2), "factors": factors, "raw_data": cur["raw"], "identity": identity, "caliber": cur["caliber"], "params_source": caliber.get("_params_source"), "net_profit_mode": cur["net_profit_mode"], "kpi_refs": _dupont_kpi_refs(db, entity_id, win["period_end"]), "insight": { "improvement": "提升毛利率、控制销售折扣/现金折扣支出,而非加杠杆", "detail": detail, }, "readonly": True, "data_source": "subject_balances + voucher_details", "data_source_note": ("取数源 = CMA 库内 subject_balances(存量,level=1)+ voucher_details(流量," "carry_forward=0);不用旧脚本 register_dupont_kpis.py 的 ERP 直连近似口径"), } # ============================================================ # 价值管理区 — 真 ROI / ROA / EVA / RI(只读) # 口径: 财务Bot 2026-10-02 裁定(任总 11:59 CST DM「立;但①要补真ROI」) # ============================================================ @router.get("/value-management") def get_value_management( period: Optional[str] = Query(None, description="期间 YYYY-MM;缺省取该账套最新实测月"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """价值管理区 — 真 ROI / 总资产报酬率(ROA) / EVA / 剩余收益(RI)。 单一事实源:**只读 `kpi_values` 里 `source_type='derived'` 的派生行**(由 `scripts/kpi_profitability_deriver.py` 落库),本端点**不复算**——避免「同一指标两份结果」。 参数由 `app/utils/profitability_params.py` 从 `kpi_definitions.data_source_config.params` 读取。 口径要点:YTD 累计(非年化);投入资本 = 有息负债(2001+2501) + 所有者权益(3xxx 一级科目); 厂补 530103 / 渠补 571103 计入经营(开关可回旧口径)。 """ codes = ("F_ROI", "F_ROA", "F_EVA", "F_RI", "F_ROE") defs = {d.kpi_code: d for d in db.query(KPIDefinition).filter( KPIDefinition.entity_id == entity_id, KPIDefinition.kpi_code.in_(codes), ).all()} ids = [d.id for d in defs.values()] if period is None and ids: latest = db.query(KPIValue.period).filter( KPIValue.kpi_id.in_(ids), KPIValue.actual_value.isnot(None), month_period_clause(KPIValue.period), ).order_by(KPIValue.period.desc()).first() period = latest[0] if latest else None rows = [] if ids and period: rows = db.query(KPIValue).filter( KPIValue.kpi_id.in_(ids), KPIValue.period == period, KPIValue.actual_value.isnot(None), ).order_by(KPIValue.id.desc()).all() metric = {} for row in rows: # id desc ⇒ 每个 kpi 取最新一条(同指标多来源时显式暴露,不去重掩盖) kpi = next((d for d in defs.values() if d.id == row.kpi_id), None) if kpi is None or kpi.kpi_code in metric: continue metric[kpi.kpi_code] = { "value": row.actual_value, "unit": kpi.unit, "kpi_name": kpi.kpi_name, "source_type": row.source_type, "data_status": row.data_status, "source_batch": row.source_batch, "remark": row.remark, "formula_desc": kpi.formula_desc, } params = load_profitability_params(db, entity_id) return { "entity_id": entity_id, "period": period, "roa": metric.get("F_ROA"), "roi": metric.get("F_ROI"), "eva": metric.get("F_EVA"), "ri": metric.get("F_RI"), "roe": metric.get("F_ROE"), "params": { "tax_rate": params["tax_rate"], "wacc": params["wacc"], "min_return_rate": params["min_return_rate"], "include_factory_subsidy": params["include_factory_subsidy"], "include_channel_subsidy": params["include_channel_subsidy"], "include_sales_discount": params["include_sales_discount"], "annualize": params["annualize"], "capital_basis": params["capital_basis"], }, "readonly": True, "data_source": "derived", } # ============================================================ # 自动报告生成 — ChatBI优化P1 # 支持周报/月报/专项报告,定时/事件/手动触发 # ============================================================ WEEKDAY_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] DIM_CN = {"finance": "财务维度", "customer": "客户维度", "process": "内部流程", "learning": "学习成长"} ALERT_LEVEL_CN = {"red": "🔴 紧急", "yellow": "🟡 预警", "green": "🟢 正常"} def _get_current_period(report_type: str) -> str: """根据报告类型自动计算当前期间""" now = datetime.now() if report_type == "weekly": iso = now.isocalendar() return f"{iso[0]}-W{iso[1]:02d}" elif report_type == "monthly": return now.strftime("%Y-%m") elif report_type == "special": return now.strftime("%Y-%m") return now.strftime("%Y-%m") def _calc_week_range(period: str) -> tuple: """周期间 → 起止日期""" import datetime as dt year, week = period.split("-W") year, week = int(year), int(week) # ISO week: week 1 is the week containing Jan 4 jan4 = dt.date(year, 1, 4) start_of_week1 = jan4 - dt.timedelta(days=jan4.isoweekday() - 1) monday = start_of_week1 + dt.timedelta(weeks=week - 1) sunday = monday + dt.timedelta(days=6) return monday.strftime("%Y-%m-%d"), sunday.strftime("%Y-%m-%d") def _get_month_period_prefix(period: str) -> str: """YYYY-MM 的前期""" y, m = period.split("-") y, m = int(y), int(m) m -= 1 if m <= 0: m += 12 y -= 1 return f"{y}-{m:02d}" def _fetch_kpi_data(db: Session, entity_id: int = 1) -> list: """获取当前企业所有活跃KPI的当前值、目标值、维度、预警(账套隔离 2026-08-31)""" kpis = db.query(KPIDefinition).filter( KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id).all() result = [] for k in kpis: latest = db.query(KPIValue).filter( KPIValue.kpi_id == k.id, KPIValue.actual_value.isnot(None), month_period_clause(KPIValue.period), ).order_by(KPIValue.period.desc()).first() alerts = db.query(KPIAlert).filter( KPIAlert.entity_id == entity_id, KPIAlert.kpi_id == k.id, KPIAlert.status == "pending", ).order_by(KPIAlert.created_at.desc()).all() result.append({ "kpi_id": k.id, "kpi_code": k.kpi_code, "kpi_name": k.kpi_name, "dimension": k.dimension, "category": k.category, "unit": k.unit, "target_value": k.target_value, "current_value": latest.actual_value if latest else None, "current_period": latest.period if latest else None, "frequency": k.frequency, "alerts": [ {"level": a.alert_level, "message": a.alert_message} for a in alerts[:3] ], }) return result def _build_weekly_report(db: Session, period: str, entity_id: int = 1) -> dict: """生成周报(账套隔离 2026-08-31)""" kpis = _fetch_kpi_data(db, entity_id) monday, sunday = _calc_week_range(period) now_str = datetime.now().strftime("%Y-%m-%d %H:%M") # 最近7天新增的预警 from datetime import timedelta seven_days_ago = datetime.now() - timedelta(days=7) recent_alerts = db.query(KPIAlert).filter( KPIAlert.entity_id == entity_id, KPIAlert.created_at >= seven_days_ago, KPIAlert.status == "pending", ).order_by(KPIAlert.created_at.desc()).all() # 按维度分组统计 dim_stats = {} for k in kpis: d = k.get("dimension") or "other" if d not in dim_stats: dim_stats[d] = {"total": 0, "with_data": 0, "alert_count": 0} dim_stats[d]["total"] += 1 if k["current_value"] is not None: dim_stats[d]["with_data"] += 1 if k["alerts"]: dim_stats[d]["alert_count"] += len(k["alerts"]) # KPI变动(取有环比数据的) changes = [] for k in kpis: if k["current_value"] is None: continue prev_period = _get_month_period_prefix(k["current_period"]) if k["current_period"] else None if prev_period: prev_val = db.query(KPIValue).filter( KPIValue.kpi_id == k["kpi_id"], KPIValue.period == prev_period, ).first() if prev_val and prev_val.actual_value: diff = round(k["current_value"] - prev_val.actual_value, 2) rate = round(diff / prev_val.actual_value * 100, 2) if prev_val.actual_value != 0 else None changes.append({ **k, "prev_value": prev_val.actual_value, "change": diff, "change_rate": rate, }) changes.sort(key=lambda x: abs(x.get("change_rate") or 0), reverse=True) top_changes = changes[:8] # ── 构建 Markdown ── md_lines = [ f"📊 **经营分析周报**", f"📅 {monday} ~ {sunday} | 生成时间:{now_str}", "", "---", "", "## 一、本周概览", f"• 监控KPI:{len(kpis)} 个 | 有数据:{sum(1 for k in kpis if k['current_value'] is not None)} 个", f"• 待处理预警:{len(recent_alerts)} 条", ] # 按维度展示 for dim_key, dim_label in [("finance", "💰 财务"), ("customer", "🤝 客户"), ("process", "⚙️ 流程"), ("learning", "📚 学习成长")]: s = dim_stats.get(dim_key) if s: md_lines.append(f" - {dim_label}:{s['total']}个KPI | {s['with_data']}个有数据 | {s['alert_count']}条预警") md_lines.extend([ "", "## 二、关键KPI变动 TOP8", ]) for c in top_changes: direction = "📈" if (c.get("change_rate") or 0) > 0 else "📉" rate_str = f"{c['change_rate']:+.1f}%" if c.get("change_rate") is not None else "-" md_lines.append( f" {direction} **{c['kpi_name']}**:{c['current_value']}{c['unit']} " f"(上期{c.get('prev_value', '-')},变动{rate_str})" ) if recent_alerts: md_lines.extend([ "", "## 三、本周预警", ]) for a in recent_alerts[:10]: kpi = next((k for k in kpis if k["kpi_id"] == a.kpi_id), None) kpi_name = kpi["kpi_name"] if kpi else f"KPI#{a.kpi_id}" md_lines.append(f" {ALERT_LEVEL_CN.get(a.alert_level, '⚠️')} {kpi_name}:{a.alert_message}") md_lines.extend([ "", "## 四、改进行动", ]) actions = db.query(ActionPlan).filter( ActionPlan.entity_id == entity_id, ActionPlan.status.in_(["pending", "in_progress"]), ).order_by(ActionPlan.created_at.desc()).limit(5).all() if actions: for a in actions: bar = "▓" * (a.progress // 10) + "░" * (10 - a.progress // 10) md_lines.append(f" • {bar} {a.title}({a.progress}%)- {a.assignee or '未分配'}") else: md_lines.append(" (暂无进行中的改善行动)") md_lines.extend([ "", "---", f"💡 发送「分析报告」可重新生成", ]) markdown = "\n".join(md_lines) # ── 构建 JSON ── json_data = { "report_type": "weekly", "period": period, "date_range": {"start": monday, "end": sunday}, "generated_at": now_str, "overview": { "total_kpis": len(kpis), "kpis_with_data": sum(1 for k in kpis if k["current_value"] is not None), "pending_alerts": len(recent_alerts), }, "dimensions": {dk: { "label": DIM_CN.get(dk, dk), "kpi_count": ds["total"], "with_data": ds["with_data"], "alert_count": ds["alert_count"], } for dk, ds in dim_stats.items()}, "top_changes": [ { "kpi_code": c["kpi_code"], "kpi_name": c["kpi_name"], "current_value": c["current_value"], "prev_value": c.get("prev_value"), "change": c.get("change"), "change_rate": c.get("change_rate"), "unit": c["unit"], } for c in top_changes ], "alerts": [ { "kpi_id": a.kpi_id, "alert_level": a.alert_level, "alert_message": a.alert_message, } for a in recent_alerts[:10] ], } return {"markdown": markdown, "json": json_data, "title": f"经营分析周报 {monday}~{sunday}"} def _build_monthly_report(db: Session, period: str, entity_id: int = 1) -> dict: """生成月报(账套隔离 2026-08-31)""" kpis = _fetch_kpi_data(db, entity_id) now_str = datetime.now().strftime("%Y-%m-%d %H:%M") prev_period = _get_month_period_prefix(period) # 预算执行数据 budget_items = [] for k in kpis: dev = calc_period_deviation(db, k["kpi_id"], period) if dev.get("actual_value") is not None or dev.get("budget_value") is not None: budget_items.append({ "kpi_name": k["kpi_name"], "kpi_code": k["kpi_code"], "dimension": k["dimension"], "actual": dev.get("actual_value"), "budget": dev.get("budget_value"), "deviation_rate": dev.get("deviation_rate"), "unit": k["unit"], }) # 同比/环比 comparisons = [] for k in kpis[:20]: if k["current_value"] is None: continue mom = calc_period_diff(db, k["kpi_id"], period, "mom") yoy = calc_period_diff(db, k["kpi_id"], period, "yoy") comparisons.append({ "kpi_name": k["kpi_name"], "kpi_code": k["kpi_code"], "current": k["current_value"], "unit": k["unit"], "mom_rate": mom.get("diff_rate"), "yoy_rate": yoy.get("diff_rate"), }) # 预警汇总 pending_alerts = db.query(KPIAlert).filter( KPIAlert.entity_id == entity_id, KPIAlert.status == "pending", ).all() red_count = sum(1 for a in pending_alerts if a.alert_level == "red") yellow_count = sum(1 for a in pending_alerts if a.alert_level == "yellow") # 各维度达成情况 dim_summary = {} for k in kpis: d = k.get("dimension") or "other" if d not in dim_summary: dim_summary[d] = {"total": 0, "achieved": 0, "warning": 0, "failed": 0} dim_summary[d]["total"] += 1 if k["current_value"] is not None and k["target_value"]: ratio = k["current_value"] / k["target_value"] if ratio >= 0.9: dim_summary[d]["achieved"] += 1 elif ratio >= 0.7: dim_summary[d]["warning"] += 1 else: dim_summary[d]["failed"] += 1 # 改善行动 actions = db.query(ActionPlan).filter(ActionPlan.entity_id == entity_id).order_by(ActionPlan.created_at.desc()).limit(5).all() # ── 生成 Markdown ── md_lines = [ f"📊 **经营分析月报**", f"📅 {period} | 生成时间:{now_str}", "", "---", "", "## 一、月度总览", f"• 监控KPI:{len(kpis)} 个", f"• 预警状态:🔴 {red_count}条紧急 | 🟡 {yellow_count}条预警", "", "## 二、四维度达成情况", ] for dk in ["finance", "customer", "process", "learning"]: ds = dim_summary.get(dk) if ds: label = DIM_CN.get(dk, dk) total = ds["total"] achieved = ds["achieved"] pct = round(achieved / total * 100, 1) if total > 0 else 0 bar_len = 10 filled = int(pct / 10) bar = "▓" * filled + "░" * (bar_len - filled) md_lines.append(f"• {label}:{bar} {pct}%({achieved}/{total}达标)") md_lines.extend([ "", "## 三、预算执行 TOP异常", ]) budget_with_dev = [b for b in budget_items if b.get("deviation_rate") is not None] budget_with_dev.sort(key=lambda x: abs(x["deviation_rate"]), reverse=True) for b in budget_with_dev[:8]: direction = "🔴" if (b["deviation_rate"] or 0) > 0 else "🟢" md_lines.append( f" {direction} **{b['kpi_name']}**:实际{b['actual']}{b['unit']} " f"vs 预算{b['budget']}{b['unit']}(差异率{b['deviation_rate']:+.1f}%)" ) md_lines.extend([ "", "## 四、同比/环比分析", ]) for c in comparisons[:8]: mom_str = f"环比{c.get('mom_rate'):+.1f}%" if c.get("mom_rate") is not None else "环比N/A" yoy_str = f"同比{c.get('yoy_rate'):+.1f}%" if c.get("yoy_rate") is not None else "同比N/A" md_lines.append(f" • **{c['kpi_name']}**:{c['current']}{c['unit']} | {mom_str} | {yoy_str}") md_lines.extend([ "", "## 五、改善行动进展", ]) if actions: for a in actions: bar = "▓" * (a.progress // 10) + "░" * (10 - a.progress // 10) status_cn = {"pending": "待开始", "in_progress": "进行中", "completed": "已完成"}.get(a.status, a.status) md_lines.append(f" • {bar} {a.title}({a.progress}%)- {status_cn}") else: md_lines.append(" (暂无改善行动)") md_lines.extend([ "", "---", f"💡 发送「生成{period}经营报告」可重新生成", ]) markdown = "\n".join(md_lines) # ── 生成 JSON ── json_data = { "report_type": "monthly", "period": period, "generated_at": now_str, "overview": { "total_kpis": len(kpis), "red_alerts": red_count, "yellow_alerts": yellow_count, }, "dimensions": {dk: { "label": DIM_CN.get(dk, dk), "total": ds["total"], "achieved": ds["achieved"], "achievement_rate": round(ds["achieved"] / ds["total"] * 100, 1) if ds["total"] > 0 else 0, } for dk, ds in dim_summary.items()}, "budget_execution": [ { "kpi_code": b["kpi_code"], "kpi_name": b["kpi_name"], "actual": b.get("actual"), "budget": b.get("budget"), "deviation_rate": b.get("deviation_rate"), "unit": b["unit"], } for b in budget_with_dev[:15] ], "comparisons": [ { "kpi_code": c["kpi_code"], "kpi_name": c["kpi_name"], "current": c["current"], "mom_rate": c.get("mom_rate"), "yoy_rate": c.get("yoy_rate"), } for c in comparisons[:15] ], } return {"markdown": markdown, "json": json_data, "title": f"经营分析月报 {period}"} def _build_special_report(db: Session, period: str, alert_ref: str = None, entity_id: int = 1) -> dict: """生成专项分析报告 — 聚焦KPI异常(账套隔离 2026-08-31)""" kpis = _fetch_kpi_data(db, entity_id) now_str = datetime.now().strftime("%Y-%m-%d %H:%M") # 按偏差率排序(当前值/目标值) scored = [] for k in kpis: if k["current_value"] is not None and k["target_value"] and k["target_value"] > 0: ratio = k["current_value"] / k["target_value"] deviation = round((ratio - 1) * 100, 2) scored.append({**k, "achievement_ratio": ratio, "deviation_pct": deviation}) scored.sort(key=lambda x: abs(x["deviation_pct"]), reverse=True) top_issues = scored[:10] worst_issues = [s for s in scored if s["deviation_pct"] < 0][:5] best_issues = [s for s in scored if s["deviation_pct"] > 0][:3] # 如果有预警引用,聚焦该预警关联的KPI focus_kpi_name = None if alert_ref: alert = db.query(KPIAlert).filter(KPIAlert.id == int(alert_ref)).first() if alert_ref.isdigit() else None if alert: target_kpi = next((k for k in kpis if k["kpi_id"] == alert.kpi_id), None) if target_kpi: focus_kpi_name = target_kpi["kpi_name"] # 维度分布 dim_issues = {} for s in scored: d = s.get("dimension") or "other" if d not in dim_issues: dim_issues[d] = {"on_track": 0, "at_risk": 0, "critical": 0} if s["achievement_ratio"] >= 0.9: dim_issues[d]["on_track"] += 1 elif s["achievement_ratio"] >= 0.7: dim_issues[d]["at_risk"] += 1 else: dim_issues[d]["critical"] += 1 # ── Markdown ── md_lines = [ f"📊 **经营分析专项报告**", f"📅 {period} | 生成时间:{now_str}", ] if focus_kpi_name: md_lines.append(f"🎯 触发事件:{focus_kpi_name} 异常预警") md_lines.extend([ "", "---", "", "## 一、风险总览", ]) for dk in ["finance", "customer", "process", "learning"]: d = dim_issues.get(dk) if d: label = DIM_CN.get(dk, dk) md_lines.append( f"• {label}:{d['on_track']}正常 / {d['at_risk']}预警 / {d['critical']}危险" ) md_lines.extend([ "", "## 二、风险KPI TOP 5(严重未达标)", ]) for w in worst_issues: md_lines.append( f" 🔴 **{w['kpi_name']}**:实际{w['current_value']}{w['unit']} " f"vs 目标{w['target_value']}{w['unit']}(达成率{w['achievement_ratio']*100:.1f}%)" ) for a in w.get("alerts", []): md_lines.append(f" ⚠️ {a['message']}") md_lines.extend([ "", "## 三、待处理预警详情", ]) pending_alerts = db.query(KPIAlert).filter( KPIAlert.status == "pending", ).order_by(KPIAlert.created_at.desc()).limit(10).all() if pending_alerts: for a in pending_alerts: target_kpi = next((k for k in kpis if k["kpi_id"] == a.kpi_id), None) name = target_kpi["kpi_name"] if target_kpi else f"KPI#{a.kpi_id}" md_lines.append(f" {ALERT_LEVEL_CN.get(a.alert_level, '⚠️')} {name}:{a.alert_message}") else: md_lines.append(" ✅ 无待处理预警") md_lines.extend([ "", "## 四、改善建议", ]) for w in worst_issues: if w["target_value"] and w["current_value"]: gap = round(w["target_value"] - w["current_value"], 2) md_lines.append(f" • **{w['kpi_name']}**:缺口{gap}{w['unit']},需提升至{w['target_value']}{w['unit']}才能达标") md_lines.extend([ "", "## 五、表现优秀KPI", ]) for b in best_issues: md_lines.append(f" 🟢 **{b['kpi_name']}**:{b['current_value']}{b['unit']},超目标{b['deviation_pct']:+.1f}%") md_lines.extend([ "", "---", "💡 如有疑问,请回复「分析详情」获取更细颗粒度的数据", ]) markdown = "\n".join(md_lines) # ── JSON ── json_data = { "report_type": "special", "period": period, "generated_at": now_str, "focus_kpi": focus_kpi_name, "alert_ref": alert_ref, "risk_summary": {dk: { "label": DIM_CN.get(dk, dk), "on_track": dim_issues.get(dk, {}).get("on_track", 0), "at_risk": dim_issues.get(dk, {}).get("at_risk", 0), "critical": dim_issues.get(dk, {}).get("critical", 0), } for dk in ["finance", "customer", "process", "learning"]}, "worst_kpis": [ { "kpi_code": w["kpi_code"], "kpi_name": w["kpi_name"], "current_value": w["current_value"], "target_value": w["target_value"], "achievement_ratio": round(w["achievement_ratio"], 4), "gap": round(w["target_value"] - w["current_value"], 2) if w["target_value"] and w["current_value"] else None, "unit": w["unit"], } for w in worst_issues ], "best_kpis": [ { "kpi_code": b["kpi_code"], "kpi_name": b["kpi_name"], "current_value": b["current_value"], "target_value": b["target_value"], "achievement_ratio": round(b["achievement_ratio"], 4), "unit": b["unit"], } for b in best_issues ], "alerts": [ { "kpi_id": a.kpi_id, "alert_level": a.alert_level, "alert_message": a.alert_message, } for a in pending_alerts[:10] ], } return {"markdown": markdown, "json": json_data, "title": f"经营分析专项报告 {period}"} # ============================================================ # POST /api/cma/reports/generate — 自动报告生成主入口 # ============================================================ class GenerateReportRequest(BaseModel): report_type: str = "monthly" # weekly / monthly / special period: Optional[str] = None # 自动计算 if None trigger_type: str = "manual" # manual / scheduled / event alert_ref: Optional[str] = None # 事件触发时的预警ID @router.post("/generate", response_model=None) def generate_report( req: GenerateReportRequest, db: Session = Depends(get_db), current_user=Depends(require_auth), entity_id: int = Depends(get_entity_id), ): """生成经营分析报告(周报/月报/专项),返回markdown+JSON 触发方式: - POST ?trigger_type=manual (用户主动触发) - POST ?trigger_type=scheduled (定时任务触发) - POST ?trigger_type=event&alert_ref=123 (KPI异常事件触发) """ # 校验报告类型 if req.report_type not in ("weekly", "monthly", "special"): raise HTTPException(400, f"不支持的报告类型: {req.report_type},可选: weekly/monthly/special") # 确定期间 period = req.period or _get_current_period(req.report_type) # 生成报告 builders = { "weekly": lambda db, period: _build_weekly_report(db, period, entity_id), "monthly": lambda db, period: _build_monthly_report(db, period, entity_id), "special": lambda db, period: _build_special_report(db, period, alert_ref=req.alert_ref, entity_id=entity_id), } builder = builders[req.report_type] try: report_data = builder(db, period) except Exception as e: logger.error(f"报告生成异常: {e}", exc_info=True) raise HTTPException(500, f"报告生成失败: {str(e)}") # 保存到数据库(账套隔离 2026-08-31) record = ReportHistory( entity_id=entity_id, report_type=req.report_type, period=period, title=report_data["title"], markdown_content=report_data["markdown"], json_content=report_data["json"], status="generated", trigger_type=req.trigger_type, alert_ref=req.alert_ref, ) db.add(record) db.flush() # 记录操作日志 log = OperationLog( action="generate_report", target_type="report", target_id=record.id, detail=json.dumps({ "entity_id": entity_id, # 2026-09-23 ④:审计带账套(=会话账套) "report_type": req.report_type, "period": period, "trigger_type": req.trigger_type, }, ensure_ascii=False), ) db.add(log) db.commit() db.refresh(record) return { "id": record.id, "report_type": req.report_type, "period": period, "title": report_data["title"], "generated_at": record.created_at.strftime("%Y-%m-%d %H:%M:%S") if record.created_at else datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "markdown": report_data["markdown"], "json": report_data["json"], } @router.get("/history") def list_report_history( report_type: Optional[str] = Query(None), limit: int = Query(20, ge=1, le=100), db: Session = Depends(get_db), current_user=Depends(require_auth), entity_id: int = Depends(get_entity_id), ): """查看报告生成历史(账套隔离 2026-08-31)""" query = db.query(ReportHistory).filter(ReportHistory.entity_id == entity_id).order_by(ReportHistory.created_at.desc()) if report_type: query = query.filter(ReportHistory.report_type == report_type) records = query.limit(limit).all() return { "total": len(records), "data": [ { "id": r.id, "report_type": r.report_type, "period": r.period, "title": r.title, "status": r.status, "trigger_type": r.trigger_type, "created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None, } for r in records ], } @router.get("/history/{report_id}") def get_report_detail( report_id: int, db: Session = Depends(get_db), current_user=Depends(require_auth), entity_id: int = Depends(get_entity_id), ): """获取单条报告详情(含完整markdown内容,账套隔离 2026-08-31)""" r = db.query(ReportHistory).filter(ReportHistory.id == report_id, ReportHistory.entity_id == entity_id).first() if not r: raise HTTPException(404, "报告不存在") return { "id": r.id, "report_type": r.report_type, "period": r.period, "title": r.title, "status": r.status, "trigger_type": r.trigger_type, "alert_ref": r.alert_ref, "markdown": r.markdown_content, "json": r.json_content, "created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None, } # ============================================================ # 预编报表(预算版三张报表)— P2 2026-08-30 # 用 budget_plans 预算数据 + 现有报表模板,生成预算版 # 利润表 / 资产负债表 / 现金流量表,供高层拍板预算方案。 # 差异口径与 budget-execution 一致(calc_deviation: 实际-预算)。 # 无预算映射的行 has_budget=false 显式标注,不静默丢弃、不塞 demo 数据。 # ============================================================ # 利润表科目编码 → KPI 编码(与 _get_subject_amount 内 kpi_code_map 一致) PROFIT_SUBJECT_KPI_MAP = { "6001": "F_REVENUE", "6051": "F_REVENUE_OTHER", "6401": "F_COST", "6402": "F_COST_OTHER", "6601": "F_SELLING_EXP", "6602": "F_ADMIN_EXP", "660204": "F_RD_EXP", "6603": "F_FINANCE_EXP", "6701": "F_IMPAIRMENT_LOSS", "6011": "F_INTEREST_INCOME", "6111": "F_INVEST_INCOME", "611101": "F_INVEST_INCOME", "660301": "F_INTEREST_EXP", "660302": "F_FX_LOSS", "6801": "F_TAX_EXP", "6901": "F_DISCONTINUED", } # 资产负债表行项目(科目组合 key 以 "|" 连接,与 _bs_line_amount 一致)→ KPI 映射 # ratio_kpi=true 表示该KPI为比率/天数型,预算值与金额不可直接比较,需单独展示 BALANCE_SHEET_PROFORMA_MAP = { "1001|1002|1012": { "kpi_code": "F_OP_CFLOW", "ratio_kpi": False, "note": "货币资金以经营性现金流预算近似(无直接科目预算)", }, "1122": { "kpi_code": "F_AR_DAYS", "ratio_kpi": True, "note": "比率型KPI(应收账款周转天数),预算为天数指标,与金额不可直接比较,需单独展示", }, } # 现金流量表行项目 → KPI 映射(CF行无直接预算,用金额KPI近似;经营净额走 F_OP_CFLOW) CASH_FLOW_PROFORMA_MAP = { "CF01": {"kpi_code": "F_REVENUE", "note": "销售商品收到的现金以营业收入预算近似"}, "CF04": {"kpi_code": "F_COST", "note": "购买商品支付的现金以营业成本预算近似"}, "CF05": {"kpi_code": "F_ADMIN_EXP", "note": "支付给职工的现金以管理费用预算近似"}, "CF06": {"kpi_code": "F_TAX_EXP", "note": "支付的各项税费以所得税费用预算近似"}, } def _find_kpi_by_code(db: Session, kpi_code: Optional[str], entity_id: int): """按 entity + kpi_code 查 KPI(proforma 专用,带租户隔离,兼容历史 NULL entity 行)""" if not kpi_code: return None return db.query(KPIDefinition).filter( or_(KPIDefinition.entity_id == entity_id, KPIDefinition.entity_id.is_(None)), KPIDefinition.kpi_code == kpi_code, ).first() def _proforma_budget(db: Session, kpi_id: int, period: str, version: Optional[str] = None, entity_id: int = 1): """预编报表预算取数:budget_plan → target_split → none 与 calc_period_deviation 口径一致(无预算时用 KPI 目标值按月分摊)。 返回 (budget_value, budget_source, budget_version) """ budget = get_budget_for_kpi(db, kpi_id, period, version) if budget is not None: query = db.query(BudgetPlan).filter( BudgetPlan.kpi_id == kpi_id, BudgetPlan.period == period, BudgetPlan.status == "active", ) if version: query = query.filter(BudgetPlan.version == version) plan = query.order_by(BudgetPlan.updated_at.desc()).first() return round(float(budget), 2), "budget_plan", (plan.version if plan else version) kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() if kpi: try: month = int(period.split("-")[1]) except Exception: month = 1 target = kpi.target_value if target and target > 0 and kpi.frequency == "monthly": return round(float(target) / 12, 2), "target_split", None return None, "none", None def _proforma_deviation(actual: Optional[float], budget: Optional[float], ratio_kpi: bool = False) -> dict: """差异三列 — 口径与 calc_period_deviation 一致(实际-预算,实际为空不计算); 比率型KPI不计算金额差异""" if ratio_kpi or actual is None: return {"deviation_amount": None, "deviation_rate": None, "is_over_budget": None} return calc_deviation(actual, budget) def _proforma_cf_actual(db: Session, line: dict, period: str, entity_id: int = 1) -> Optional[float]: """现金流量表行项目实际值 — 真实数据优先(KPI → 凭证),不塞 demo 数据(账套隔离 2026-08-31)""" if line.get("kpi_code"): v = _get_kpi_val(db, line["kpi_code"], period) if v is not None: return float(v) v = _get_bs_amount(db, [(line["code"], line["sign"])], period) if v is not None: return float(v) return None def _proforma_versions(found_versions: set): """budget_version 输出:单一版本→字符串,多版本→列表,无→None""" if not found_versions: return None vs = sorted(found_versions) return vs[0] if len(vs) == 1 else vs @router.get("/proforma/profit-statement") def get_proforma_profit_statement( period: str = Query(None, description="格式 YYYY-MM"), version: Optional[str] = Query(None, description="预算版本,默认取最新active"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """预算版利润表 — 新30号准则五板块结构,每行叠加 预算值/实际值/差异""" if period is None: period = datetime.now().strftime("%Y-%m") blocks = [] net_actual = net_budget = 0.0 found_versions = set() for block_key in ["operating", "investing", "financing", "tax", "discontinued"]: block_cfg = BLOCK_INFO[block_key] items = [] block_actual = block_budget = 0.0 block_has_budget = False for item_cfg in block_cfg["items"]: code = item_cfg["code"] kpi_code = PROFIT_SUBJECT_KPI_MAP.get(code) actual = _get_subject_amount(db, code, period, entity_id) budget, source, ver = None, "none", None kpi = _find_kpi_by_code(db, kpi_code, entity_id) if kpi_code else None if kpi: budget, source, ver = _proforma_budget(db, kpi.id, period, version) has_budget = budget is not None if has_budget: block_has_budget = True if ver: found_versions.add(ver) if actual is not None: block_actual += actual * item_cfg["sign"] if budget is not None: block_budget += budget * item_cfg["sign"] dev = _proforma_deviation(actual, budget) items.append({ "code": code, "name": item_cfg["name"], "sign": item_cfg["sign"], "actual_value": round(actual, 2) if actual is not None else None, "budget_value": budget, "deviation_amount": dev.get("deviation_amount"), "deviation_rate": dev.get("deviation_rate"), "has_budget": has_budget, "mapped_kpi_code": kpi_code, "budget_source": source, "ratio_kpi": False, "note": None, }) blocks.append({ "key": block_key, "name": block_cfg["name"], "short_name": block_cfg["short_name"], "subtotal_actual": round(block_actual, 2), "subtotal_budget": round(block_budget, 2), "subtotal_name": block_cfg["result_name"], "has_budget": block_has_budget, "items": items, }) net_actual += block_actual net_budget += block_budget return { "period": period, "budget_version": _proforma_versions(found_versions), "requested_version": version, "title": f"预算版利润表 — 新30号准则({period})", "blocks": blocks, "net_profit_actual": round(net_actual, 2), "net_profit_budget": round(net_budget, 2), "budget_source_hint": "budget_plan=预算方案 / target_split=KPI目标值按月分摊 / none=无预算", } @router.get("/proforma/balance-sheet") def get_proforma_balance_sheet( period: str = Query(None, description="格式 YYYY-MM"), version: Optional[str] = Query(None, description="预算版本,默认取最新active"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """预算版资产负债表 — 复用 BALANCE_SHEET_SECTIONS,每行叠加 预算值/实际值/差异""" if period is None: period = datetime.now().strftime("%Y-%m") sections = [] found_versions = set() for sec in BALANCE_SHEET_SECTIONS: lines = [] sec_actual = sec_budget = 0.0 sec_has_budget = False for line in sec["lines"]: key = "|".join(c for c, _ in line["codes"]) map_cfg = BALANCE_SHEET_PROFORMA_MAP.get(key) or {} kpi_code = map_cfg.get("kpi_code") ratio_kpi = map_cfg.get("ratio_kpi", False) note = map_cfg.get("note") # 实际值:真实凭证数据(预编报表不塞 demo 示例数据) actual = _get_bs_amount(db, line["codes"], period) budget, source, ver = None, "none", None kpi = _find_kpi_by_code(db, kpi_code, entity_id) if kpi_code else None if kpi: budget, source, ver = _proforma_budget(db, kpi.id, period, version) has_budget = budget is not None if has_budget: sec_has_budget = True if ver: found_versions.add(ver) if actual is not None: sec_actual += actual if budget is not None and not ratio_kpi: sec_budget += budget dev = _proforma_deviation(actual, budget, ratio_kpi) lines.append({ "name": line["name"], "ns_category": line["ns_category"], "ns_category_label": BS_CATEGORY_CN.get(line["ns_category"], line["ns_category"]), "actual_value": round(actual, 2) if actual is not None else None, "budget_value": budget, "deviation_amount": dev.get("deviation_amount"), "deviation_rate": dev.get("deviation_rate"), "has_budget": has_budget, "mapped_kpi_code": kpi_code, "ratio_kpi": ratio_kpi, "budget_source": source, "note": note, }) sections.append({ "key": sec["key"], "name": sec["name"], "category_label": sec["category_label"], "subtotal_actual": round(sec_actual, 2), "subtotal_budget": round(sec_budget, 2), "has_budget": sec_has_budget, "lines": lines, }) return { "period": period, "budget_version": _proforma_versions(found_versions), "requested_version": version, "title": f"预算版资产负债表({period})", "sections": sections, "budget_source_hint": "budget_plan=预算方案 / target_split=KPI目标值按月分摊 / none=无预算", } @router.get("/proforma/cash-flow") def get_proforma_cash_flow( period: str = Query(None, description="格式 YYYY-MM"), version: Optional[str] = Query(None, description="预算版本,默认取最新active"), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """预算版现金流量表 — 复用 CASH_FLOW_LINES,每行叠加 预算值/实际值/差异""" if period is None: period = datetime.now().strftime("%Y-%m") section_cfg = [ {"key": "operating", "name": "一、经营活动产生的现金流量", "short": "经营活动"}, {"key": "investing", "name": "二、投资活动产生的现金流量", "short": "投资活动"}, {"key": "financing", "name": "三、筹资活动产生的现金流量", "short": "筹资活动"}, ] sections = [] found_versions = set() for sc in section_cfg: lines = [] subtotal_actual = subtotal_budget = 0.0 sec_has_budget = False for line in CASH_FLOW_LINES: if line["section"] != sc["key"]: continue map_cfg = CASH_FLOW_PROFORMA_MAP.get(line["code"]) or {} kpi_code = map_cfg.get("kpi_code") note = map_cfg.get("note") actual = _proforma_cf_actual(db, line, period) budget, source, ver = None, "none", None kpi = _find_kpi_by_code(db, kpi_code, entity_id) if kpi_code else None if kpi: budget, source, ver = _proforma_budget(db, kpi.id, period, version) has_budget = budget is not None if has_budget: sec_has_budget = True if ver: found_versions.add(ver) if actual is not None: subtotal_actual += actual if budget is not None: subtotal_budget += budget dev = _proforma_deviation(actual, budget) lines.append({ "code": line["code"], "name": line["name"], "actual_value": round(actual, 2) if actual is not None else None, "budget_value": budget, "deviation_amount": dev.get("deviation_amount"), "deviation_rate": dev.get("deviation_rate"), "has_budget": has_budget, "mapped_kpi_code": kpi_code, "ratio_kpi": False, "budget_source": source, "note": note, }) # 经营净额:优先取 F_OP_CFLOW(真实),预算取 F_OP_CFLOW 预算 net_actual = subtotal_actual net_budget = subtotal_budget if sc["key"] == "operating": op_kpi = _find_kpi_by_code(db, "F_OP_CFLOW", entity_id) if op_kpi: op_actual = _get_kpi_val(db, "F_OP_CFLOW", period, entity_id) if op_actual is not None: net_actual = round(float(op_actual), 2) op_budget, op_source, op_ver = _proforma_budget(db, op_kpi.id, period, version) if op_budget is not None: net_budget = op_budget sec_has_budget = True if op_ver: found_versions.add(op_ver) sections.append({ "key": sc["key"], "name": sc["name"], "short": sc["short"], "net_actual": round(net_actual, 2), "net_budget": round(net_budget, 2), "has_budget": sec_has_budget, "lines": lines, }) net_increase_actual = round(sum(s["net_actual"] for s in sections), 2) net_increase_budget = round(sum(s["net_budget"] for s in sections), 2) return { "period": period, "budget_version": _proforma_versions(found_versions), "requested_version": version, "title": f"预算版现金流量表({period})", "sections": sections, "summary": { "net_increase_actual": net_increase_actual, "net_increase_budget": net_increase_budget, }, "budget_source_hint": "budget_plan=预算方案 / target_split=KPI目标值按月分摊 / none=无预算", }