Hermes CI Fix
5e0b09e6b6
- 表名: ORM+生产库 RENAME cash_flow_structure_values → cashflow_statements (DoD SHOW TABLES LIKE 字面验收)
- 兼容端点: /api/cma/cashflow/statements(GET/POST)+analysis 注册 main.py (DoD 验收路径)
- 菜单: 移入 🟡C 监控与评价 组(与杜邦分析/管理报表平级), roles 收敛 ceo/finance (前端 ROLE_ROUTES/MENU_ITEMS/router meta + 后端 require_role 四处同步)
- 术语: docstring 改写, grep 现金流肖像=0
- 验收脚本: backend/scripts/verify_cashflow_structure.py (14项断言全PASS: type6→type1双向/瀑布5步/alias路径/权限/清理)
368 lines
16 KiB
Python
368 lines
16 KiB
Python
"""
|
||
现金流量表结构分析 — 管理会计OS (CMA P2 · 财务报表分析)
|
||
规范命名：现金流量表结构分析（本模块不采用民间俗称，术语以任务单为准）
|
||
|
||
核心方法：经营(O)/投资(I)/筹资(F) 三大活动净额正负组合 → 8类企业定性分析
|
||
O 造血能力 / I 成长代价 / F 输血管道
|
||
长期黄金法则：经营现金流必须为正，且应大于净利润（利润质量）。
|
||
|
||
数据模型：cashflow_statements — 官方现金流量表三大活动净额按期间注入
|
||
- 未注入期间：无判定（不拿演示/示例数据冒充真实类型）
|
||
- 官方净额入库后类型判定自动生效（博海2026官方表注入路径）
|
||
"""
|
||
from fastapi import APIRouter, Depends, Query, HTTPException
|
||
from pydantic import BaseModel
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import func
|
||
from typing import Optional, List
|
||
from datetime import datetime
|
||
|
||
from app.database import get_db
|
||
from app.auth_middleware import require_role
|
||
from app.deps import get_entity_id
|
||
from app.models import CashFlowStructureValue
|
||
|
||
router = APIRouter(
|
||
prefix="/api/cma/cashflow-structure",
|
||
tags=["现金流量表结构分析"],
|
||
dependencies=[Depends(require_role("ceo", "finance"))],
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# 八类企业定性分析规则表（O/I/F 符号组合 → 类型）
|
||
# ============================================================
|
||
CASHFLOW_TYPE_RULES = [
|
||
{
|
||
"type_no": 1, "o_sign": "+", "i_sign": "-", "f_sign": "-",
|
||
"name": "成熟健康型", "health_emoji": "🟢", "health_label": "最优",
|
||
"lifecycle": "成熟期",
|
||
"profile": "主业盈利，自给自足；仍在投入扩张且有余力还债/分红。最健康形态（如成熟消费龙头）。",
|
||
"risk": "注意扩张效率与分红可持续性；经营现金流应大于净利润，否则利润质量存疑。",
|
||
},
|
||
{
|
||
"type_no": 2, "o_sign": "+", "i_sign": "+", "f_sign": "-",
|
||
"name": "成熟收割型", "health_emoji": "🟢", "health_label": "良好",
|
||
"lifecycle": "成熟期后期/集团收割期",
|
||
"profile": "主业赚钱+过去投资进入回报期（处置/收益回笼）+还在还债分红。",
|
||
"risk": "若投资净流入来自大量处置核心资产，需警惕「卖资产过日子」。",
|
||
},
|
||
{
|
||
"type_no": 3, "o_sign": "+", "i_sign": "-", "f_sign": "+",
|
||
"name": "高速扩张型", "health_emoji": "🟡", "health_label": "看增速",
|
||
"lifecycle": "成长期",
|
||
"profile": "主业能赚钱，但扩张更快、现金不够花，需要融资支撑资本开支。成长质量取决于经营造血增速能否追上扩张。",
|
||
"risk": "扩张节奏失控 → 现金消耗加速 → 滑向类型6/7。",
|
||
},
|
||
{
|
||
"type_no": 4, "o_sign": "+", "i_sign": "+", "f_sign": "+",
|
||
"name": "全面回笼型", "health_emoji": "🟡", "health_label": "需辨析",
|
||
"lifecycle": "重组/收缩盘点期",
|
||
"profile": "三类活动都进钱，少见。多为经营好转+资产处置+新增融资并举（可能正在重组）。",
|
||
"risk": "需具体辨析筹资来源（股权/债权）与投资流入是否靠卖资产。",
|
||
},
|
||
{
|
||
"type_no": 5, "o_sign": "-", "i_sign": "+", "f_sign": "-",
|
||
"name": "衰退变卖型", "health_emoji": "🔴", "health_label": "高危",
|
||
"lifecycle": "衰退期/清盘前奏",
|
||
"profile": "主业亏损，靠卖资产/处置投资回笼现金，同时还在还债。濒危信号。",
|
||
"risk": "资产卖完即无以为继；除非经营反转否则难持续。",
|
||
},
|
||
{
|
||
"type_no": 6, "o_sign": "-", "i_sign": "-", "f_sign": "+",
|
||
"name": "初创烧钱型", "health_emoji": "🟠", "health_label": "高风险",
|
||
"lifecycle": "初创期/高增长投入期",
|
||
"profile": "主业烧钱+持续投入，全靠融资吊命。常见于初创/高增长科技企业；也是庞氏结构的表面特征。",
|
||
"risk": "融资链条断裂即死；须区分「真成长」（收入/用户高速增长）与「假烧钱」。",
|
||
},
|
||
{
|
||
"type_no": 7, "o_sign": "-", "i_sign": "+", "f_sign": "+",
|
||
"name": "输血生存型", "health_emoji": "🔴", "health_label": "高危",
|
||
"lifecycle": "挣扎期",
|
||
"profile": "经营亏损，靠卖资产+借款输血双管齐下维持。压力极大。",
|
||
"risk": "短期可活，长期必须扭转经营现金流，否则资金链断裂。",
|
||
},
|
||
{
|
||
"type_no": 8, "o_sign": "-", "i_sign": "-", "f_sign": "-",
|
||
"name": "全面失血型", "health_emoji": "🔴", "health_label": "极危",
|
||
"lifecycle": "深度危机期",
|
||
"profile": "经营/投资/筹资三个引擎全熄火，现金只出不进。最危险状态。",
|
||
"risk": "极可能需要紧急债务重组/股东注资/断臂求生。",
|
||
},
|
||
]
|
||
|
||
# 类型 → 规则 快速索引
|
||
TYPE_RULE_MAP = {r["type_no"]: r for r in CASHFLOW_TYPE_RULES}
|
||
|
||
|
||
def _sign(value: Optional[float]) -> Optional[str]:
|
||
"""净额符号：>0 → +，<0 → -，==0 → '0'（无法落入8类组合），None → None"""
|
||
if value is None:
|
||
return None
|
||
if value > 0:
|
||
return "+"
|
||
if value < 0:
|
||
return "-"
|
||
return "0"
|
||
|
||
|
||
def classify_enterprise(operating_net: Optional[float],
|
||
investing_net: Optional[float],
|
||
financing_net: Optional[float]) -> dict:
|
||
"""O/I/F 三净额符号 → 8类企业定性判定
|
||
|
||
返回:
|
||
{classified, type_no, name, health_emoji, health_label, risk, profile, lifecycle, o_sign, i_sign, f_sign, reason}
|
||
classified=False 时 reason 说明未判定的原因（任一净额缺失或为0）
|
||
"""
|
||
o, i, f = _sign(operating_net), _sign(investing_net), _sign(financing_net)
|
||
if o is None or i is None or f is None:
|
||
missing = [name for sign, name in [(o, "经营"), (i, "投资"), (f, "筹资")] if sign is None]
|
||
return {"classified": False, "reason": "以下活动净额缺失：" + "、".join(missing) + "（需官方现金流量表数据）"}
|
||
if "0" in (o, i, f):
|
||
zero = [name for sign, name in [(o, "经营"), (i, "投资"), (f, "筹资")] if sign == "0"]
|
||
return {"classified": False, "reason": "以下活动净额为0（恰好为0无法落入8类正负组合）：" + "、".join(zero) + "，请按业务实质人工判定"}
|
||
|
||
for rule in CASHFLOW_TYPE_RULES:
|
||
if rule["o_sign"] == o and rule["i_sign"] == i and rule["f_sign"] == f:
|
||
return {
|
||
"classified": True,
|
||
"type_no": rule["type_no"],
|
||
"name": rule["name"],
|
||
"health_emoji": rule["health_emoji"],
|
||
"health_label": rule["health_label"],
|
||
"lifecycle": rule["lifecycle"],
|
||
"profile": rule["profile"],
|
||
"risk": rule["risk"],
|
||
"o_sign": o, "i_sign": i, "f_sign": f,
|
||
"reason": "经营(%s) 投资(%s) 筹资(%s) → 类型%d %s" % (o, i, f, rule["type_no"], rule["name"]),
|
||
}
|
||
return {"classified": False, "reason": "未知组合，请检查数据"}
|
||
|
||
|
||
class CashFlowValueIn(BaseModel):
|
||
"""三大活动净额录入（官方现金流量表 / 人工）"""
|
||
period: str
|
||
operating_net: float
|
||
investing_net: float
|
||
financing_net: float
|
||
begin_cash: Optional[float] = None
|
||
end_cash: Optional[float] = None
|
||
source: str = "official"
|
||
remark: Optional[str] = None
|
||
|
||
|
||
def _validate_period(period: str) -> str:
|
||
"""期间格式校验 YYYY-MM"""
|
||
period = (period or "").strip()
|
||
try:
|
||
datetime.strptime(period, "%Y-%m")
|
||
except Exception:
|
||
raise HTTPException(status_code=422, detail="期间格式须为 YYYY-MM，如 2026-08")
|
||
return period
|
||
|
||
|
||
# ============================================================
|
||
# 接口
|
||
# ============================================================
|
||
|
||
@router.get("/rules")
|
||
def get_rules():
|
||
"""八类企业定性判定规则表（O/I/F 符号组合速查）"""
|
||
return {"rules": CASHFLOW_TYPE_RULES}
|
||
|
||
|
||
@router.get("/periods")
|
||
def list_periods(
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""已注入净额的期间列表（倒序，供按期间查询）"""
|
||
rows = db.query(CashFlowStructureValue.period).filter(
|
||
CashFlowStructureValue.entity_id == entity_id
|
||
).order_by(CashFlowStructureValue.period.desc()).all()
|
||
return {"periods": [r[0] for r in rows]}
|
||
|
||
|
||
@router.get("/values")
|
||
def list_values(
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""已注入的三大活动净额记录（按期间倒序）"""
|
||
rows = db.query(CashFlowStructureValue).filter(
|
||
CashFlowStructureValue.entity_id == entity_id
|
||
).order_by(CashFlowStructureValue.period.desc()).all()
|
||
return {"items": [
|
||
{
|
||
"id": r.id, "period": r.period,
|
||
"operating_net": r.operating_net,
|
||
"investing_net": r.investing_net,
|
||
"financing_net": r.financing_net,
|
||
"begin_cash": r.begin_cash, "end_cash": r.end_cash,
|
||
"source": r.source, "remark": r.remark,
|
||
"updated_at": r.updated_at.strftime("%Y-%m-%d %H:%M") if r.updated_at else None,
|
||
} for r in rows
|
||
]}
|
||
|
||
|
||
@router.get("/analysis")
|
||
def get_analysis(
|
||
period: str = Query(None, description="格式 YYYY-MM，缺省取最近已注入期间"),
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""现金流量表结构分析 — 三净额 → 八类企业判定 + 瀑布图数据"""
|
||
if period:
|
||
period = _validate_period(period)
|
||
|
||
# 1. 取官方/人工注入的净额（唯一权威判定数据）
|
||
q = db.query(CashFlowStructureValue).filter(
|
||
CashFlowStructureValue.entity_id == entity_id
|
||
)
|
||
if period:
|
||
rec = q.filter(CashFlowStructureValue.period == period).first()
|
||
else:
|
||
rec = q.order_by(CashFlowStructureValue.period.desc()).first()
|
||
if rec:
|
||
period = rec.period
|
||
|
||
if rec is None:
|
||
return {
|
||
"period": period or datetime.now().strftime("%Y-%m"),
|
||
"data_available": False,
|
||
"message": "该期间暂无官方现金流量表净额数据，请先录入三大活动净额（官方现金流量表注入后自动生效）。",
|
||
"nets": None,
|
||
"classification": None,
|
||
"waterfall": None,
|
||
}
|
||
|
||
nets = {
|
||
"operating": {"key": "operating", "name": "经营活动", "value": rec.operating_net,
|
||
"sign": _sign(rec.operating_net), "unit": "万元"},
|
||
"investing": {"key": "investing", "name": "投资活动", "value": rec.investing_net,
|
||
"sign": _sign(rec.investing_net), "unit": "万元"},
|
||
"financing": {"key": "financing", "name": "筹资活动", "value": rec.financing_net,
|
||
"sign": _sign(rec.financing_net), "unit": "万元"},
|
||
}
|
||
|
||
classification = classify_enterprise(rec.operating_net, rec.investing_net, rec.financing_net)
|
||
|
||
# 2. 瀑布图：期初现金 → +经营 → +投资 → +筹资 → 期末现金
|
||
waterfall = None
|
||
if rec.begin_cash is not None or rec.end_cash is not None:
|
||
begin = rec.begin_cash
|
||
end = rec.end_cash
|
||
steps = []
|
||
cum = begin if begin is not None else 0.0
|
||
if begin is not None:
|
||
steps.append({"step": "期初现金", "value": begin, "type": "begin"})
|
||
for net_key, net_name in [("operating", "经营活动"), ("investing", "投资活动"), ("financing", "筹资活动")]:
|
||
val = getattr(rec, net_key + "_net") or 0.0
|
||
cum += val
|
||
steps.append({"step": net_name, "value": round(val, 2), "type": "flow"})
|
||
if end is not None:
|
||
steps.append({"step": "期末现金", "value": end, "type": "end"})
|
||
elif begin is not None:
|
||
end = round(cum, 2)
|
||
steps.append({"step": "期末现金(推算)", "value": end, "type": "end"})
|
||
waterfall = {
|
||
"begin_cash": begin, "end_cash": end,
|
||
"net_increase": round((end - begin), 2) if (begin is not None and end is not None) else None,
|
||
"steps": steps,
|
||
}
|
||
|
||
return {
|
||
"period": rec.period,
|
||
"data_available": True,
|
||
"source": rec.source,
|
||
"remark": rec.remark,
|
||
"nets": nets,
|
||
"classification": classification,
|
||
"waterfall": waterfall,
|
||
}
|
||
|
||
|
||
@router.post("/values", dependencies=[Depends(require_role("ceo", "finance"))])
|
||
def upsert_value(
|
||
payload: CashFlowValueIn,
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""录入/更新某期间三大活动净额（官方现金流量表注入；同期间重复提交=覆盖更新）"""
|
||
period = _validate_period(payload.period)
|
||
if payload.source not in ("official", "manual"):
|
||
raise HTTPException(status_code=422, detail="source 仅支持 official(官方报表)/manual(人工录入)")
|
||
|
||
rec = db.query(CashFlowStructureValue).filter(
|
||
CashFlowStructureValue.entity_id == entity_id,
|
||
CashFlowStructureValue.period == period,
|
||
).first()
|
||
created = rec is None
|
||
if rec is None:
|
||
rec = CashFlowStructureValue(entity_id=entity_id, period=period)
|
||
db.add(rec)
|
||
|
||
rec.operating_net = round(payload.operating_net, 2)
|
||
rec.investing_net = round(payload.investing_net, 2)
|
||
rec.financing_net = round(payload.financing_net, 2)
|
||
rec.begin_cash = round(payload.begin_cash, 2) if payload.begin_cash is not None else rec.begin_cash
|
||
rec.end_cash = round(payload.end_cash, 2) if payload.end_cash is not None else rec.end_cash
|
||
rec.source = payload.source
|
||
rec.remark = payload.remark
|
||
db.commit()
|
||
|
||
classification = classify_enterprise(rec.operating_net, rec.investing_net, rec.financing_net)
|
||
return {
|
||
"ok": True,
|
||
"id": rec.id,
|
||
"period": rec.period,
|
||
"classification": classification,
|
||
"message": "三大活动净额已%s" % ("入库" if created else "更新"),
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# 兼容别名端点 — 任务派发单 DoD 验收路径 /api/cma/cashflow/*
|
||
# (规范实现主路径为 /api/cma/cashflow-structure/*；两路径功能一致)
|
||
# ============================================================
|
||
compat_router = APIRouter(
|
||
prefix="/api/cma/cashflow",
|
||
tags=["现金流量表结构分析(兼容路径)"],
|
||
dependencies=[Depends(require_role("ceo", "finance"))],
|
||
)
|
||
|
||
|
||
@compat_router.get("/statements")
|
||
def compat_list_statements(
|
||
period: Optional[str] = Query(None, description="期间 YYYY-MM，可选"),
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""[兼容] 查询已录入三大活动净额（同 /api/cma/cashflow-structure/values）"""
|
||
items = list_values(db=db, entity_id=entity_id)["items"]
|
||
if period:
|
||
period = _validate_period(period)
|
||
items = [x for x in items if x["period"] == period]
|
||
return {"items": items}
|
||
|
||
|
||
@compat_router.post("/statements", dependencies=[Depends(require_role("ceo", "finance"))])
|
||
def compat_save_statement(
|
||
payload: CashFlowValueIn,
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""[兼容] 保存/更新某期间三大活动净额（同 POST /api/cma/cashflow-structure/values）"""
|
||
return upsert_value(payload=payload, db=db, entity_id=entity_id)
|
||
|
||
|
||
@compat_router.get("/analysis")
|
||
def compat_get_analysis(
|
||
period: str = Query(None, description="格式 YYYY-MM，缺省取最近已注入期间"),
|
||
db: Session = Depends(get_db),
|
||
entity_id: int = Depends(get_entity_id),
|
||
):
|
||
"""[兼容] 结构分析+八类判定+瀑布数据（同 /api/cma/cashflow-structure/analysis）"""
|
||
return get_analysis(period=period, db=db, entity_id=entity_id)
|