Hermes CI Fix
3dddd36866
包含前后端完整代码： - 前端：Vue3+Vite+ElementPlus - 后端：FastAPI+SQLAlchemy - 模块：驾驶舱/KPI/战略地图/预警/预算/成本/预测/改善行动 - 当前版本：v1.0.0
283 lines
10 KiB
Python
283 lines
10 KiB
Python
"""成本分析引擎 — 管理会计OS
|
|
标准成本vs实际成本差异分析（量差/价差/效率差异）
|
|
ABC作业成本法分配
|
|
"""
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict
|
|
from app.database import get_session_local
|
|
from app.models import StandardCost, ActualCost, AbcActivity, AbcAllocation, KPIDefinition, KPIValue
|
|
|
|
logger = logging.getLogger("cma.cost")
|
|
|
|
ERP_API = "http://127.0.0.1:8300"
|
|
ERP_KEY = "erp-gateway-key-bhwl-2026"
|
|
|
|
|
|
# ============================================================
|
|
# 差异计算
|
|
# ============================================================
|
|
|
|
def calc_variance(standard_qty: float, actual_qty: float,
|
|
standard_price: float, actual_price: float) -> dict:
|
|
"""计算量差和价差
|
|
|
|
量差 = (实际用量 - 标准用量) × 标准价格
|
|
价差 = (实际价格 - 标准价格) × 实际用量
|
|
总差异 = 量差 + 价差
|
|
"""
|
|
qty_variance = round((actual_qty - standard_qty) * standard_price, 2)
|
|
price_variance = round((actual_price - standard_price) * actual_qty, 2)
|
|
total_variance = round(qty_variance + price_variance, 2)
|
|
return {
|
|
"qty_variance": qty_variance, # 量差
|
|
"price_variance": price_variance, # 价差
|
|
"total_variance": total_variance, # 总差异
|
|
"standard_cost": round(standard_qty * standard_price, 2),
|
|
"actual_cost": round(actual_qty * actual_price, 2),
|
|
}
|
|
|
|
|
|
def calc_efficiency_variance(standard_hours: float, actual_hours: float,
|
|
standard_rate: float) -> dict:
|
|
"""计算效率差异（人工/制造费用）
|
|
|
|
效率差异 = (实际工时 - 标准工时) × 标准分配率
|
|
分配率差异 = (实际分配率 - 标准分配率) × 实际工时
|
|
"""
|
|
eff = round((actual_hours - standard_hours) * standard_rate, 2)
|
|
# 假设实际分配率从外面传入
|
|
return {
|
|
"efficiency_variance": eff,
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# 产品级差异分析
|
|
# ============================================================
|
|
|
|
def calc_product_variance(product_code: str, period: str) -> dict:
|
|
"""计算指定产品在指定期间的成本差异"""
|
|
db = get_session_local()()
|
|
try:
|
|
standards = db.query(StandardCost).filter(
|
|
StandardCost.product_code == product_code,
|
|
StandardCost.status == "active",
|
|
).all()
|
|
actuals = db.query(ActualCost).filter(
|
|
ActualCost.product_code == product_code,
|
|
ActualCost.period == period,
|
|
).all()
|
|
|
|
if not standards or not actuals:
|
|
return {"product_code": product_code, "error": "标准成本或实际成本数据不足", "items": [], "summary": {}}
|
|
|
|
# 按 cost_type 分组
|
|
cost_types = set()
|
|
for s in standards: cost_types.add(s.cost_type)
|
|
for a in actuals: cost_types.add(a.cost_type)
|
|
|
|
items = []
|
|
total_std = 0
|
|
total_act = 0
|
|
total_qty_var = 0
|
|
total_price_var = 0
|
|
|
|
for ct in sorted(cost_types):
|
|
std_items = [s for s in standards if s.cost_type == ct]
|
|
act_items = [a for a in actuals if a.cost_type == ct]
|
|
|
|
if std_items and act_items:
|
|
s = std_items[0]
|
|
a = act_items[0]
|
|
var = calc_variance(s.standard_quantity, a.actual_quantity,
|
|
s.standard_price, a.actual_price)
|
|
items.append({
|
|
"cost_type": ct,
|
|
"item_name": s.item_name,
|
|
"standard_quantity": s.standard_quantity,
|
|
"actual_quantity": a.actual_quantity,
|
|
"standard_price": s.standard_price,
|
|
"actual_price": a.actual_price,
|
|
"standard_cost": var["standard_cost"],
|
|
"actual_cost": var["actual_cost"],
|
|
"qty_variance": var["qty_variance"],
|
|
"price_variance": var["price_variance"],
|
|
"total_variance": var["total_variance"],
|
|
})
|
|
total_std += var["standard_cost"]
|
|
total_act += var["actual_cost"]
|
|
total_qty_var += var["qty_variance"]
|
|
total_price_var += var["price_variance"]
|
|
|
|
return {
|
|
"product_code": product_code,
|
|
"period": period,
|
|
"items": items,
|
|
"summary": {
|
|
"total_standard_cost": round(total_std, 2),
|
|
"total_actual_cost": round(total_act, 2),
|
|
"total_variance": round(total_act - total_std, 2),
|
|
"total_qty_variance": round(total_qty_var, 2),
|
|
"total_price_variance": round(total_price_var, 2),
|
|
"variance_rate": round((total_act - total_std) / total_std * 100, 2) if total_std else 0,
|
|
}
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ============================================================
|
|
# ABC 作业成本分配
|
|
# ============================================================
|
|
|
|
def calc_driver_rate(activity_id: int) -> dict:
|
|
"""计算作业动因分配率 = 总成本 / 动因总量"""
|
|
db = get_session_local()()
|
|
try:
|
|
act = db.query(AbcActivity).filter(AbcActivity.id == activity_id).first()
|
|
if not act or not act.driver_volume:
|
|
return {"error": "作业中心不存在或动因总量为0"}
|
|
rate = round(act.total_cost / act.driver_volume, 4) if act.driver_volume > 0 else 0
|
|
act.driver_rate = rate
|
|
db.commit()
|
|
return {
|
|
"activity_code": act.activity_code,
|
|
"activity_name": act.activity_name,
|
|
"total_cost": act.total_cost,
|
|
"driver_volume": act.driver_volume,
|
|
"driver_rate": rate,
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def allocate_cost(activity_id: int, period: str, product_code: str,
|
|
product_name: str, driver_consumed: float) -> dict:
|
|
"""按动因分配成本到产品"""
|
|
db = get_session_local()()
|
|
try:
|
|
act = db.query(AbcActivity).filter(AbcActivity.id == activity_id).first()
|
|
if not act or act.driver_rate == 0:
|
|
# 自动计算分配率
|
|
if act and act.driver_volume > 0:
|
|
act.driver_rate = round(act.total_cost / act.driver_volume, 4)
|
|
db.commit()
|
|
if not act or act.driver_rate == 0:
|
|
return {"error": "分配率未设置"}
|
|
allocated = round(driver_consumed * act.driver_rate, 2)
|
|
alloc = AbcAllocation(
|
|
period=period,
|
|
activity_id=activity_id,
|
|
product_code=product_code,
|
|
product_name=product_name,
|
|
driver_consumed=driver_consumed,
|
|
allocated_cost=allocated,
|
|
)
|
|
db.add(alloc)
|
|
db.commit()
|
|
return {
|
|
"activity_code": act.activity_code,
|
|
"product_code": product_code,
|
|
"driver_consumed": driver_consumed,
|
|
"driver_rate": act.driver_rate,
|
|
"allocated_cost": allocated,
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ============================================================
|
|
# 成本总览数据
|
|
# ============================================================
|
|
|
|
def get_cost_overview(period: str) -> dict:
|
|
"""获取成本总览数据（总成本、结构占比、趋势）"""
|
|
db = get_session_local()()
|
|
try:
|
|
# 从实际成本表汇总
|
|
actual_costs = db.query(ActualCost).filter(ActualCost.period == period).all()
|
|
total_cost = sum(a.actual_cost for a in actual_costs)
|
|
|
|
# 按成本类型分组
|
|
by_type: Dict[str, float] = {}
|
|
for a in actual_costs:
|
|
by_type[a.cost_type] = by_type.get(a.cost_type, 0) + a.actual_cost
|
|
|
|
structure = [
|
|
{"cost_type": k, "amount": round(v, 2), "ratio": round(v / total_cost * 100, 1) if total_cost else 0}
|
|
for k, v in sorted(by_type.items())
|
|
]
|
|
|
|
# 从ERP获取成本数据做补充
|
|
import httpx
|
|
try:
|
|
resp = httpx.get(f"{ERP_API}/api/v1/query", params={"table": "BalanceInfo", "limit": 100},
|
|
headers={"X-API-Key": ERP_KEY}, timeout=10)
|
|
balance_data = resp.json().get("data", [])
|
|
erp_cost = 0
|
|
for b in balance_data:
|
|
if b.get("Act_ID") == 5: # 营业成本
|
|
erp_cost += float(b.get("Act_Hap", 0) or 0) + float(b.get("Act_Tot", 0) or 0)
|
|
except Exception:
|
|
erp_cost = 0
|
|
|
|
# 历史趋势（近6个月）
|
|
from sqlalchemy import text
|
|
year = period[:4]
|
|
months_texts = []
|
|
try:
|
|
m = int(period.split("-")[1])
|
|
for i in range(6):
|
|
pm = m - i
|
|
py = int(year)
|
|
while pm <= 0:
|
|
pm += 12
|
|
py -= 1
|
|
months_texts.append(f"{py}-{pm:02d}")
|
|
|
|
trend = []
|
|
for p in reversed(months_texts):
|
|
costs = db.query(ActualCost).filter(ActualCost.period == p).all()
|
|
total = round(sum(c.actual_cost for c in costs), 2)
|
|
trend.append({"period": p, "total_cost": total})
|
|
except Exception:
|
|
trend = []
|
|
|
|
return {
|
|
"period": period,
|
|
"total_cost": round(total_cost + erp_cost, 2),
|
|
"erp_cost": round(erp_cost, 2),
|
|
"manual_cost": round(total_cost, 2),
|
|
"structure": structure,
|
|
"trend": trend,
|
|
}
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def get_cost_breakdown(product_code: str, period: str) -> dict:
|
|
"""获取成本构成（料/工/费占比）"""
|
|
db = get_session_local()()
|
|
try:
|
|
actuals = db.query(ActualCost).filter(
|
|
ActualCost.product_code == product_code,
|
|
ActualCost.period == period,
|
|
).all()
|
|
|
|
material = sum(a.actual_cost for a in actuals if a.cost_type == "material")
|
|
labor = sum(a.actual_cost for a in actuals if a.cost_type == "labor")
|
|
overhead = sum(a.actual_cost for a in actuals if a.cost_type == "overhead")
|
|
total = material + labor + overhead
|
|
|
|
return {
|
|
"product_code": product_code,
|
|
"period": period,
|
|
"material": {"amount": round(material, 2), "ratio": round(material / total * 100, 1) if total else 0},
|
|
"labor": {"amount": round(labor, 2), "ratio": round(labor / total * 100, 1) if total else 0},
|
|
"overhead": {"amount": round(overhead, 2), "ratio": round(overhead / total * 100, 1) if total else 0},
|
|
"total": round(total, 2),
|
|
}
|
|
finally:
|
|
db.close()
|