Hermes CI Fix
3dddd36866
包含前后端完整代码： - 前端：Vue3+Vite+ElementPlus - 后端：FastAPI+SQLAlchemy - 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动 - 当前版本：v1.0.0
261 lines
9.1 KiB
Python
261 lines
9.1 KiB
Python
"""预测模拟引擎 — 管理会计OS
|
|
CVP本量利分析、投资决策(NPV/IRR)、敏感性分析、情景模拟
|
|
"""
|
|
import math
|
|
import logging
|
|
from typing import List, Dict, Optional, Tuple
|
|
|
|
logger = logging.getLogger("cma.predict")
|
|
|
|
|
|
# ============================================================
|
|
# CVP 本量利分析
|
|
# ============================================================
|
|
|
|
def cvp_analysis(
|
|
unit_price: float, # 单价
|
|
unit_variable_cost: float, # 单位变动成本
|
|
fixed_cost: float, # 固定成本
|
|
target_profit: float = None, # 目标利润（可选）
|
|
actual_volume: float = None, # 实际销量（可选）
|
|
) -> dict:
|
|
"""CVP本量利分析
|
|
|
|
返回：盈亏平衡点、安全边际、目标利润所需销量
|
|
"""
|
|
if unit_price <= unit_variable_cost:
|
|
return {"error": "单价必须大于单位变动成本"}
|
|
|
|
contribution_margin = unit_price - unit_variable_cost # 单位边际贡献
|
|
contribution_ratio = round(contribution_margin / unit_price * 100, 2) # 边际贡献率
|
|
|
|
# 盈亏平衡点（保本点）
|
|
bep_units = round(fixed_cost / contribution_margin, 2) # 保本销量
|
|
bep_revenue = round(bep_units * unit_price, 2) # 保本销售额
|
|
|
|
result = {
|
|
"unit_price": unit_price,
|
|
"unit_variable_cost": unit_variable_cost,
|
|
"fixed_cost": fixed_cost,
|
|
"contribution_margin": round(contribution_margin, 2),
|
|
"contribution_ratio": contribution_ratio,
|
|
"bep_units": bep_units,
|
|
"bep_revenue": bep_revenue,
|
|
}
|
|
|
|
# 安全边际
|
|
if actual_volume is not None:
|
|
safety_margin_units = actual_volume - bep_units
|
|
safety_margin_ratio = round(safety_margin_units / actual_volume * 100, 2) if actual_volume > 0 else 0
|
|
actual_profit = round((unit_price - unit_variable_cost) * actual_volume - fixed_cost, 2)
|
|
result["safety_margin_units"] = round(safety_margin_units, 2)
|
|
result["safety_margin_revenue"] = round(safety_margin_units * unit_price, 2)
|
|
result["safety_margin_ratio"] = safety_margin_ratio
|
|
result["actual_profit"] = actual_profit
|
|
|
|
# 目标利润
|
|
if target_profit is not None:
|
|
target_units = round((fixed_cost + target_profit) / contribution_margin, 2)
|
|
target_revenue = round(target_units * unit_price, 2)
|
|
result["target_profit"] = target_profit
|
|
result["target_units"] = target_units
|
|
result["target_revenue"] = target_revenue
|
|
|
|
return result
|
|
|
|
|
|
# ============================================================
|
|
# 投资决策模型
|
|
# ============================================================
|
|
|
|
def npv(initial_investment: float, cash_flows: List[float], discount_rate: float) -> dict:
|
|
"""计算净现值 NPV = Σ CFt / (1+r)^t - I0"""
|
|
if not cash_flows:
|
|
return {"error": "现金流列表不能为空"}
|
|
r = discount_rate / 100
|
|
pv = 0
|
|
for t, cf in enumerate(cash_flows, 1):
|
|
pv += cf / ((1 + r) ** t)
|
|
npv_value = round(pv - initial_investment, 2)
|
|
|
|
# 盈利能力指数 PI = PV / I0
|
|
pi = round(pv / initial_investment, 4) if initial_investment > 0 else 0
|
|
|
|
return {
|
|
"initial_investment": initial_investment,
|
|
"discount_rate": discount_rate,
|
|
"pv_of_cash_flows": round(pv, 2),
|
|
"npv": npv_value,
|
|
"profitability_index": pi,
|
|
"is_viable": npv_value > 0,
|
|
}
|
|
|
|
|
|
def irr(initial_investment: float, cash_flows: List[float], max_iter: int = 1000, tolerance: float = 1e-6) -> dict:
|
|
"""计算内部收益率 IRR（迭代法）"""
|
|
if not cash_flows:
|
|
return {"error": "现金流列表不能为空"}
|
|
|
|
# 确保现金流总和 > 初始投资（否则 IRR 可能为负）
|
|
total_cf = sum(cash_flows)
|
|
if total_cf <= initial_investment:
|
|
# 用牛顿法尝试求负IRR
|
|
pass
|
|
|
|
def _npv_at(rate: float) -> float:
|
|
return sum(cf / ((1 + rate) ** (t + 1)) for t, cf in enumerate(cash_flows)) - initial_investment
|
|
|
|
# 牛顿法求根
|
|
rate = 0.1 # 初始猜测 10%
|
|
for _ in range(max_iter):
|
|
f = _npv_at(rate)
|
|
if abs(f) < tolerance:
|
|
break
|
|
# 导数近似
|
|
h = 1e-4
|
|
df = (_npv_at(rate + h) - _npv_at(rate - h)) / (2 * h)
|
|
if abs(df) < tolerance:
|
|
break
|
|
rate -= f / df
|
|
if rate < -0.99: # IRR 不能低于 -99%
|
|
rate = -0.99
|
|
break
|
|
|
|
irr_value = round(rate * 100, 2)
|
|
|
|
# 回收期
|
|
cumulative = 0
|
|
payback_period = None
|
|
for t, cf in enumerate(cash_flows, 1):
|
|
cumulative += cf
|
|
if cumulative >= initial_investment:
|
|
payback_period = t
|
|
break
|
|
|
|
# 动态回收期（折现）
|
|
r = irr_value / 100 if irr_value > 0 else 0.1
|
|
discounted_cumulative = 0
|
|
discounted_payback = None
|
|
for t, cf in enumerate(cash_flows, 1):
|
|
discounted_cumulative += cf / ((1 + r) ** t)
|
|
if discounted_cumulative >= initial_investment:
|
|
discounted_payback = t
|
|
break
|
|
|
|
return {
|
|
"initial_investment": initial_investment,
|
|
"cash_flows": cash_flows,
|
|
"irr": irr_value,
|
|
"payback_period": payback_period, # 静态回收期（年）
|
|
"discounted_payback_period": discounted_payback, # 动态回收期
|
|
"is_viable": irr_value > 0,
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# 敏感性分析
|
|
# ============================================================
|
|
|
|
def sensitivity_analysis(
|
|
base_revenue: float, # 基准收入
|
|
base_cost: float, # 基准成本
|
|
base_profit: float = None, # 基准利润（若为None则自动 = 收入-成本）
|
|
step: float = 5, # 步长 %
|
|
max_step: float = 20, # 最大变动 %
|
|
) -> dict:
|
|
"""单因素敏感性分析
|
|
|
|
分析销量、单价、成本变动对利润的影响
|
|
"""
|
|
if base_profit is None:
|
|
base_profit = base_revenue - base_cost
|
|
|
|
factors = []
|
|
steps = [s for s in range(-max_step, max_step + 1, step)] or [0]
|
|
|
|
for pct in steps:
|
|
factor = pct / 100
|
|
|
|
# 收入变动（销量变动）
|
|
revenue_change_profit = base_profit * (1 + factor)
|
|
rev_sensitivity = round((revenue_change_profit - base_profit) / base_profit * 100, 2) if base_profit else 0
|
|
|
|
# 成本变动
|
|
cost_change_profit = base_profit - base_cost * factor
|
|
cost_sensitivity = round((cost_change_profit - base_profit) / base_profit * 100, 2) if base_profit else 0
|
|
|
|
# 同时变动（收入+5%同时成本+5%）
|
|
both_profit = (base_revenue * (1 + factor)) - (base_cost * (1 + factor))
|
|
both_sensitivity = round((both_profit - base_profit) / base_profit * 100, 2) if base_profit else 0
|
|
|
|
factors.append({
|
|
"change_pct": pct,
|
|
"revenue_change_profit": round(revenue_change_profit, 2),
|
|
"revenue_sensitivity": rev_sensitivity,
|
|
"cost_change_profit": round(cost_change_profit, 2),
|
|
"cost_sensitivity": cost_sensitivity,
|
|
"both_change_profit": round(both_profit, 2),
|
|
"both_sensitivity": both_sensitivity,
|
|
})
|
|
|
|
return {
|
|
"base_revenue": base_revenue,
|
|
"base_cost": base_cost,
|
|
"base_profit": round(base_profit, 2),
|
|
"step": step,
|
|
"max_step": max_step,
|
|
"factors": factors,
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# 情景模拟
|
|
# ============================================================
|
|
|
|
def scenario_analysis(
|
|
optimistic: dict, # {"revenue": 130, "cost": 90}
|
|
pessimistic: dict, # {"revenue": 80, "cost": 110}
|
|
base: dict, # {"revenue": 100, "cost": 100}
|
|
) -> dict:
|
|
"""三情景模拟（乐观/中性/悲观）
|
|
|
|
每个情景包含 revenue(收入) 和 cost(成本)
|
|
计算各情景下的利润和偏差
|
|
"""
|
|
scenarios = []
|
|
for label, data in [("乐观", optimistic), ("中性", base), ("悲观", pessimistic)]:
|
|
revenue = data.get("revenue", 0)
|
|
cost = data.get("cost", 0)
|
|
profit = round(revenue - cost, 2)
|
|
scenarios.append({
|
|
"scenario": label,
|
|
"revenue": revenue,
|
|
"cost": cost,
|
|
"profit": profit,
|
|
"profit_margin": round(profit / revenue * 100, 2) if revenue else 0,
|
|
})
|
|
|
|
base_profit = scenarios[1]["profit"] # 中性情景利润
|
|
for s in scenarios:
|
|
if base_profit:
|
|
s["deviation_from_base"] = round(s["profit"] - base_profit, 2)
|
|
s["deviation_pct"] = round((s["profit"] - base_profit) / base_profit * 100, 2)
|
|
else:
|
|
s["deviation_from_base"] = s["profit"]
|
|
s["deviation_pct"] = 0
|
|
|
|
# 最好/最坏/期望值（假设各1/3概率）
|
|
expected_profit = round(
|
|
(scenarios[0]["profit"] + scenarios[1]["profit"] + scenarios[2]["profit"]) / 3, 2
|
|
)
|
|
variance = sum((s["profit"] - expected_profit) ** 2 for s in scenarios) / 3
|
|
std_dev = round(math.sqrt(variance), 2)
|
|
|
|
return {
|
|
"scenarios": scenarios,
|
|
"expected_profit": expected_profit,
|
|
"std_deviation": std_dev,
|
|
"best_case": scenarios[0],
|
|
"worst_case": scenarios[2],
|
|
}
|