237 lines
8.9 KiB
Python
237 lines
8.9 KiB
Python
"""多情景预测模拟引擎 — 管理会计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,
|
|
}
|