""" CMA管理报表中心 — 管理会计OS 非传统财务报表，聚焦管理决策分析 报表: 1. 管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润 2. 预算执行报告 — 各KPI预算vs实际vs差异率 3. KPI趋势报告 — 选定KPI的历史趋势 4. 四维度绩效评分卡 — BSC健康度雷达图 """ from fastapi import APIRouter, Depends, Query, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from sqlalchemy import func 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 from app.utils.deviation_engine import calc_period_deviation, calc_period_diff 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), ): """管理利润表 — 收入→变动成本→边际贡献→固定成本→息税前利润""" 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).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).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), ): """预算执行报告 — 各KPI预算vs实际vs差异率""" if period is None: period = datetime.now().strftime("%Y-%m") query = db.query(KPIDefinition).filter(KPIDefinition.status == "active") 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), ): """KPI趋势报告 — 选定KPI的历史趋势线""" query = db.query(KPIDefinition).filter(KPIDefinition.status == "active") 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 ).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), ): """四维度绩效评分卡 — BSC健康度""" if period is None: period = datetime.now().strftime("%Y-%m") # 取最新的已发布地图 map_query = db.query(StrategicMap).filter(StrategicMap.status == "published") 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) # 从战略地图维度数据构建评分卡 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) -> dict: """没有战略地图时，直接按维度聚合KPI算分""" kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").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) -> Optional[float]: """从 subjects + kpi_values 获取科目金额数据""" # 尝试从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).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 result = db.query( func.sum(VoucherDetail.debit_amount - VoucherDetail.credit_amount) ).filter( VoucherDetail.subject_code == code, VoucherDetail.period == period ).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), ): """利润表 — 支持旧格式、新30号准则五板块格式、双列对比""" if period is None: period = datetime.now().strftime("%Y-%m") if format == "old": # 旧30号准则格式（保留兼容） return get_profit_summary(period=period, db=db) if format == "dual": # 双列对比：旧准则 vs 新准则 old_data = get_profit_summary(period=period, db=db) new_data = _build_new_format_profit(db, period) return { "period": period, "format": "dual", "title": f"利润表双列对比（{period}）", "old_format": old_data, "new_format": new_data, } # === 新30号准则：五板块结构 === return _build_new_format_profit(db, period) def _build_new_format_profit(db: Session, period: str) -> dict: """构建新30号准则五板块利润表""" 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) 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 # 合计行：净利润 = 一二三+四+五 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, "prev_period": None, # TODO: P1追溯调整 } # ============================================================ # 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), ): """MPM管理层指标计算器 — 生成合规调节表""" 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) if net_profit is None: net_profit = 0 # 经营现金流（自由现金流的基准） operating_cf = _get_kpi_val(db, "F_OPERATING_CF", req.period) # 确定基准值 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) 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) -> Optional[float]: """从KPI定义+值获取数值""" kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == code).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) -> Optional[float]: """计算新30号准则下的净利润""" 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) 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) -> Optional[float]: """获取调整项的自动取值""" mapping = ADJUSTMENT_VALUE_MAP.get(adj_code) if mapping is None: return None # 需要用户输入 code = mapping["code"] amount = _get_subject_amount(db, code, period) 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), ): """2026年数据按新准则重述 — 旧口径vs新口径双列对比，自动标记调整项""" 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).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) new_revenue_other = _get_subject_amount(db, "6051", period) new_cost = _get_subject_amount(db, "6401", period) new_cost_other = _get_subject_amount(db, "6402", period) new_selling = _get_subject_amount(db, "6601", period) new_admin = _get_subject_amount(db, "6602", period) new_rd = _get_subject_amount(db, "660204", period) new_interest_income = _get_subject_amount(db, "6011", period) new_interest_exp = _get_subject_amount(db, "660301", period) new_fx = _get_subject_amount(db, "6603", period) new_fx_financing = _get_subject_amount(db, "660302", period) new_invest_income = _get_subject_amount(db, "6111", period) new_impairment = _get_subject_amount(db, "6701", period) new_invest_impairment = _get_subject_amount(db, "670101", period) new_tax = _get_subject_amount(db, "6801", period) new_discontinued = _get_subject_amount(db, "6901", period) # 旧口径汇总计算 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) 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号准则板块映射""" 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), } @router.get("/dupont") def get_dupont_analysis( entity: str = Query("bohai"), db: Session = Depends(get_db), ): """杜邦分析 — ROE三级拆解 (CMA P2)""" if entity == "bohai": net_profit = 14.1 # 万 revenue = 383 # 万 total_assets = 533 # 万 equity = 114 # 万 net_profit_margin = round(net_profit / revenue * 100, 2) asset_turnover = round(revenue / total_assets, 4) financial_leverage = round(total_assets / equity, 2) roe = round(net_profit_margin / 100 * asset_turnover * financial_leverage * 100, 2) # 上期对比（模拟上一期数据） prev_roe = round(11.2, 2) roe_change = round(roe - prev_roe, 2) return { "entity": "bohai", "entity_name": "陕西博海科技（IT服务）", "period": "2026年H1", "roe": roe, "roe_change": roe_change, "roe_trend": "up" if roe_change > 0 else "down", "prev_roe": prev_roe, "factors": { "net_profit_margin": { "value": net_profit_margin, "label": "净利润率", "desc": "净利润/收入", "status": "🟡" if net_profit_margin < 5 else "✅", "assessment": "IT经销行业正常偏低", "raw": {"net_profit": net_profit, "revenue": revenue}, }, "asset_turnover": { "value": asset_turnover, "label": "资产周转率", "desc": "收入/总资产", "status": "🟡" if asset_turnover < 1 else "✅", "assessment": "资金效率中等", "raw": {"revenue": revenue, "total_assets": total_assets}, }, "financial_leverage": { "value": financial_leverage, "label": "财务杠杆", "desc": "总资产/净资产", "status": "🟡" if financial_leverage > 3 else "✅", "assessment": "负债率78.6%，偏高但可控", "raw": {"total_assets": total_assets, "equity": equity}, }, }, "raw_data": { "net_profit": net_profit, "revenue": revenue, "total_assets": total_assets, "equity": equity, }, "insight": { "improvement": "提高周转率或利润率，而非加杠杆", "detail": f"净利润率{net_profit_margin}%偏低，资产周转率{asset_turnover}x中等，财务杠杆{financial_leverage}x偏高。改善方向：提升毛利率或加快库存周转。", }, } return {"error": "不支持的实体"}