sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init: 管理会计OS初始代码

Browse Source 
包含前后端完整代码：
- 前端：Vue3+Vite+ElementPlus
- 后端：FastAPI+SQLAlchemy
- 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动
- 当前版本：v1.0.0
This commit is contained in:
Hermes CI Fix
2026-05-28 17:32:22 +08:00
commit 3dddd36866
142 changed files with 18533 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/utils/predict_engine.py
+260
View File
@@ -0,0 +1,260 @@
"""预测模拟引擎 — 管理会计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],
}