"""多情景预测模拟引擎 — 管理会计OS P2-1 从战略地图KPI输入变量出发，按类别映射到财务影响， 输出乐观/基准/保守三情景数值+曲线数据 """ import math import logging from typing import List, Dict, Optional from datetime import datetime logger = logging.getLogger("cma.scenario") # KPI类别 → 财务影响映射系数 # 降本类: 每变化1% → 成本节省系数 # 增收类: 每变化1% → 收入增长系数 KPI_CATEGORY_MAP = { # 增收类 "revenue_growth": {"type": "revenue", "factor": 0.8, "desc": "收入增长"}, "customer_scale": {"type": "revenue", "factor": 0.6, "desc": "客户规模→收入"}, # 降本类 "cost_control": {"type": "cost", "factor": -0.7, "desc": "成本节约"}, "asset_efficiency": {"type": "cost", "factor": -0.3, "desc": "资产效率→成本"}, # 利润类 "profitability": {"type": "profit", "factor": 0.5, "desc": "直接利润影响"}, # 现金流类 "cash_risk": {"type": "cash", "factor": 0.4, "desc": "现金流影响"}, # 客户类→收入 "customer_satisfaction": {"type": "revenue", "factor": 0.3, "desc": "满意度→收入"}, "customer_concentration": {"type": "revenue", "factor": -0.2, "desc": "集中度→风险"}, # 流程类→成本 "delivery_quality": {"type": "cost", "factor": -0.3, "desc": "交付质量→成本"}, "supply_chain": {"type": "cost", "factor": -0.2, "desc": "供应链→成本"}, # 学习类→长期收入 "talent_pipeline": {"type": "revenue", "factor": 0.15, "desc": "人才→收入"}, "employee_engagement": {"type": "cost", "factor": -0.1, "desc": "敬业度→成本"}, "innovation": {"type": "revenue", "factor": 0.2, "desc": "创新→收入"}, } # 默认基准财务数据（万元/月） DEFAULT_BASE_REVENUE = 1000.0 # 基准收入 DEFAULT_BASE_COST = 700.0 # 基准成本 DEFAULT_BASE_PROFIT = 300.0 # 基准利润 def calculate_scenario( variables: List[Dict], scenario_type: str = "base", # "optimistic" / "base" / "pessimistic" base_revenue: float = DEFAULT_BASE_REVENUE, base_cost: float = DEFAULT_BASE_COST, ) -> Dict: """根据KPI变量列表和三情景系数计算财务影响 Args: variables: [{"kpi_code", "kpi_name", "category", "value", "step_optimistic", "step_base", "step_pessimistic"}, ...] scenario_type: 情景类型 base_revenue: 基准收入 base_cost: 基准成本 Returns: {revenue, cost, profit, profit_margin, kpi_impacts, details} """ step_key = { "optimistic": "step_optimistic", "base": "step_base", "pessimistic": "step_pessimistic", }.get(scenario_type, "step_base") total_revenue_impact = 0.0 total_cost_impact = 0.0 total_profit_impact = 0.0 total_cash_impact = 0.0 base_profit_val = base_revenue - base_cost details = [] for var in variables: kpi_code = var.get("kpi_code", "") kpi_name = var.get("kpi_name", "") category = var.get("category", "") current_value = var.get("value", 0) step_value = var.get(step_key, 0) # 变化百分比 (当前值变化 / 当前值) if current_value and current_value != 0: change_pct = step_value / abs(current_value) * 100 else: change_pct = 0 # 查找类别映射 mapping = KPI_CATEGORY_MAP.get(category, {"type": "revenue", "factor": 0.5, "desc": "通用影响"}) impact_type = mapping["type"] factor = mapping["factor"] impact_desc = mapping["desc"] # 计算财务影响 = 变化率 × 系数 × 基准值 financial_impact = change_pct / 100 * factor if impact_type == "revenue": impact_amount = financial_impact * base_revenue total_revenue_impact += impact_amount elif impact_type == "cost": impact_amount = financial_impact * base_cost total_cost_impact += impact_amount elif impact_type == "profit": impact_amount = financial_impact * base_profit_val total_profit_impact += impact_amount elif impact_type == "cash": impact_amount = financial_impact * base_profit_val total_cash_impact += impact_amount else: impact_amount = 0 details.append({ "kpi_code": kpi_code, "kpi_name": kpi_name, "category": category, "current_value": current_value, "scenario_value": step_value, "change_pct": round(change_pct, 2), "impact_type": impact_type, "impact_desc": impact_desc, "impact_amount": round(impact_amount, 2), }) # 合成最终财务数据 final_revenue = base_revenue + total_revenue_impact final_cost = base_cost + total_cost_impact # 重新计算利润（考虑所有影响） final_profit = (final_revenue - final_cost) + total_profit_impact profit_margin = round(final_profit / final_revenue * 100, 2) if final_revenue else 0 return { "scenario_type": scenario_type, "base_revenue": base_revenue, "base_cost": base_cost, "base_profit": base_revenue - base_cost, "revenue": round(final_revenue, 2), "cost": round(final_cost, 2), "profit": round(final_profit, 2), "profit_margin": profit_margin, "revenue_impact": round(total_revenue_impact, 2), "cost_impact": round(total_cost_impact, 2), "profit_impact": round(total_profit_impact, 2), "cash_impact": round(total_cash_impact, 2), "kpi_impacts": details, } def run_three_scenarios( variables: List[Dict], base_revenue: float = DEFAULT_BASE_REVENUE, base_cost: float = DEFAULT_BASE_COST, months: int = 12, ) -> Dict: """运行三情景模拟，生成曲线数据 Args: variables: KPI变量列表，每个包含step_optimistic/step_base/step_pessimistic base_revenue: 基准月度收入 base_cost: 基准月度成本 months: 预测月数 Returns: {scenarios: [...], chart_data: {months, optimistic, base, pessimistic}, summary} """ optimistic = calculate_scenario(variables, "optimistic", base_revenue, base_cost) base = calculate_scenario(variables, "base", base_revenue, base_cost) pessimistic = calculate_scenario(variables, "pessimistic", base_revenue, base_cost) # 生成月度曲线数据（按月线性趋近情景值） start_revenue = base_revenue start_cost = base_cost start_profit = base_revenue - base_cost chart_data = { "months": [], "optimistic": {"revenue": [], "cost": [], "profit": []}, "base": {"revenue": [], "cost": [], "profit": []}, "pessimistic": {"revenue": [], "cost": [], "profit": []}, } for m in range(1, months + 1): progress = m / months # 从0到1线性趋近 label = f"第{m}月" if months <= 12 else f"M{m}" for scenario_type, scenario_data in [ ("optimistic", optimistic), ("base", base), ("pessimistic", pessimistic), ]: rev = start_revenue + (scenario_data["revenue"] - start_revenue) * progress cst = start_cost + (scenario_data["cost"] - start_cost) * progress prf = start_profit + (scenario_data["profit"] - start_profit) * progress chart_data[scenario_type]["revenue"].append(round(rev, 2)) chart_data[scenario_type]["cost"].append(round(cst, 2)) chart_data[scenario_type]["profit"].append(round(prf, 2)) chart_data["months"].append(label) # 汇总 base_profit_val = base["profit"] scenarios_list = [] for label, data in [ ("乐观", optimistic), ("基准", base), ("保守", pessimistic), ]: deviation = data["profit"] - base_profit_val scenarios_list.append({ "scenario": label, "revenue": data["revenue"], "cost": data["cost"], "profit": data["profit"], "profit_margin": data["profit_margin"], "deviation_from_base": round(deviation, 2), "deviation_pct": round(deviation / base_profit_val * 100, 2) if base_profit_val else 0, "kpi_impacts": data["kpi_impacts"], }) summary = { "expected_profit": round((optimistic["profit"] + base["profit"] + pessimistic["profit"]) / 3, 2), "best_profit": optimistic["profit"], "worst_profit": pessimistic["profit"], "base_profit": base["profit"], "variance": round( ((optimistic["profit"] - base["profit"]) ** 2 + (base["profit"] - base["profit"]) ** 2 + (pessimistic["profit"] - base["profit"]) ** 2) / 3, 2 ), "base_revenue": base_revenue, "base_cost": base_cost, "months": months, } return { "scenarios": scenarios_list, "chart_data": chart_data, "summary": summary, }