Hermes CI Fix
3dddd36866
包含前后端完整代码： - 前端：Vue3+Vite+ElementPlus - 后端：FastAPI+SQLAlchemy - 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动 - 当前版本：v1.0.0
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""预测模拟API — 管理会计OS"""
|
|
import logging
|
|
from fastapi import APIRouter, HTTPException
|
|
from app.utils.predict_engine import (
|
|
cvp_analysis, npv, irr,
|
|
sensitivity_analysis, scenario_analysis,
|
|
)
|
|
|
|
logger = logging.getLogger("cma.predict")
|
|
router = APIRouter(prefix="/api/cma/predict", tags=["预测模拟"])
|
|
|
|
|
|
@router.post("/cvp")
|
|
def api_cvp_analysis(data: dict):
|
|
"""CVP本量利分析"""
|
|
try:
|
|
result = cvp_analysis(
|
|
unit_price=float(data.get("unit_price", 0)),
|
|
unit_variable_cost=float(data.get("unit_variable_cost", 0)),
|
|
fixed_cost=float(data.get("fixed_cost", 0)),
|
|
target_profit=float(data["target_profit"]) if data.get("target_profit") else None,
|
|
actual_volume=float(data["actual_volume"]) if data.get("actual_volume") else None,
|
|
)
|
|
return result
|
|
except Exception as e:
|
|
raise HTTPException(400, f"CVP计算失败: {str(e)}")
|
|
|
|
|
|
@router.post("/investment")
|
|
def api_investment_analysis(data: dict):
|
|
"""投资决策分析（NPV/IRR/回收期）"""
|
|
try:
|
|
initial = float(data.get("initial_investment", 0))
|
|
rate = float(data.get("discount_rate", 10))
|
|
cash_flows = [float(cf) for cf in data.get("cash_flows", [])]
|
|
|
|
if not cash_flows:
|
|
raise HTTPException(400, "现金流列表不能为空")
|
|
|
|
npv_result = npv(initial, cash_flows, rate)
|
|
irr_result = irr(initial, cash_flows)
|
|
|
|
return {
|
|
"npv_analysis": npv_result,
|
|
"irr_analysis": irr_result,
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(400, f"投资决策计算失败: {str(e)}")
|
|
|
|
|
|
@router.post("/sensitivity")
|
|
def api_sensitivity_analysis(data: dict):
|
|
"""敏感性分析"""
|
|
try:
|
|
result = sensitivity_analysis(
|
|
base_revenue=float(data.get("base_revenue", 0)),
|
|
base_cost=float(data.get("base_cost", 0)),
|
|
base_profit=float(data["base_profit"]) if data.get("base_profit") else None,
|
|
step=int(data.get("step", 5)),
|
|
max_step=int(data.get("max_step", 20)),
|
|
)
|
|
return result
|
|
except Exception as e:
|
|
raise HTTPException(400, f"敏感性分析失败: {str(e)}")
|
|
|
|
|
|
@router.post("/scenario")
|
|
def api_scenario_analysis(data: dict):
|
|
"""情景模拟"""
|
|
try:
|
|
optimistic = data.get("optimistic", {})
|
|
pessimistic = data.get("pessimistic", {})
|
|
base = data.get("base", {})
|
|
|
|
if not all([optimistic, pessimistic, base]):
|
|
raise HTTPException(400, "需要提供乐观/中性/悲观三个情景的参数")
|
|
|
|
result = scenario_analysis(
|
|
optimistic={
|
|
"revenue": float(optimistic.get("revenue", 0)),
|
|
"cost": float(optimistic.get("cost", 0)),
|
|
},
|
|
pessimistic={
|
|
"revenue": float(pessimistic.get("revenue", 0)),
|
|
"cost": float(pessimistic.get("cost", 0)),
|
|
},
|
|
base={
|
|
"revenue": float(base.get("revenue", 0)),
|
|
"cost": float(base.get("cost", 0)),
|
|
},
|
|
)
|
|
return result
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(400, f"情景模拟失败: {str(e)}")
|