feat(cashflow): 现金流量表结构分析——三大活动净额八类企业定性判定 (P1)
- 后端: cashflow_structure.py 模块 (GET rules/periods/values/analysis + POST values upsert) - 判定: O/I/F 三净额符号 → 8类企业定性 (CMA P2 现金流量表分析, 无'现金流肖像'字样) - 数据: cash_flow_structure_values 表 (entity_id+period 唯一), 官方现金流量表净额按期间注入 - 前端: CashFlowStructure.vue 判定横幅/瀑布图/规则表/入库记录 + MyDialog 录入弹窗 - 菜单: 财务报表分析 → 现金流量表分析 (路由 /cashflow-structure, 角色 ceo/finance/it) - 测试: tests/test_cashflow_structure.py 7 条通过 (规则8类/判定/覆盖/博海边界6-7/非法期间)
This commit is contained in:
Hermes CI Fix
@@ -0,0 +1,322 @@
|
||||
"""
|
||||
现金流量表结构分析 — 管理会计OS (CMA P2 · 财务报表分析)
|
||||
规范命名：现金流量表结构分析（不使用"现金流肖像"等民间说法）
|
||||
|
||||
核心方法：经营(O)/投资(I)/筹资(F) 三大活动净额正负组合 → 8类企业定性分析
|
||||
O 造血能力 / I 成长代价 / F 输血管道
|
||||
长期黄金法则：经营现金流必须为正，且应大于净利润（利润质量）。
|
||||
|
||||
数据模型：cash_flow_structure_values — 官方现金流量表三大活动净额按期间注入
|
||||
- 未注入期间：无判定（不拿演示/示例数据冒充真实类型）
|
||||
- 官方净额入库后类型判定自动生效（博海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", "it"))],
|
||||
)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 八类企业定性分析规则表（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 "更新"),
|
||||
}
|
||||
+2
-1
@@ -5,7 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse
|
||||
from dotenv import load_dotenv
|
||||
from app.database import init_db
|
||||
from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify
|
||||
from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure
|
||||
from app.utils.cache import clear_all as clear_cache, delete as delete_cache
|
||||
from scripts.erp_sync import run_sync as run_erp_sync
|
||||
from app.auth_middleware import require_auth
|
||||
@@ -59,6 +59,7 @@ app.include_router(value_sources.router)
|
||||
app.include_router(zero_based.router)
|
||||
app.include_router(derivation_rules.router)
|
||||
app.include_router(cash_classify.router)
|
||||
app.include_router(cashflow_structure.router)
|
||||
app.include_router(cost.router)
|
||||
app.include_router(predict.router)
|
||||
app.include_router(growth_quality.router)
|
||||
|
||||
@@ -940,3 +940,24 @@ class AISuggestion(Base):
|
||||
apply_detail = Column(JSON, nullable=True, comment="应用结果明细: [{target_type,target_id,action,before,after}]")
|
||||
created_at = Column(DateTime, server_default=func.now())
|
||||
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
|
||||
|
||||
|
||||
class CashFlowStructureValue(Base):
|
||||
"""现金流量表结构分析 — 三大活动净额 (P1 2026-09-03 八类企业定性分析)
|
||||
官方现金流量表三大活动净额按期间注入，支撑 O/I/F 符号组合 → 8类企业判定
|
||||
单位统一为 万元（与 BH_CASH_BALANCE / ledger 口径一致）
|
||||
"""
|
||||
__tablename__ = "cash_flow_structure_values"
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
entity_id = Column(Integer, nullable=False, default=1, comment="企业ID（账套，多租户隔离）")
|
||||
period = Column(String(20), nullable=False, comment="期间 YYYY-MM")
|
||||
operating_net = Column(Float, nullable=False, comment="经营活动现金流量净额（万元）")
|
||||
investing_net = Column(Float, nullable=False, comment="投资活动现金流量净额（万元）")
|
||||
financing_net = Column(Float, nullable=False, comment="筹资活动现金流量净额（万元）")
|
||||
begin_cash = Column(Float, nullable=True, comment="期初现金及现金等价物余额（万元）")
|
||||
end_cash = Column(Float, nullable=True, comment="期末现金及现金等价物余额（万元）")
|
||||
source = Column(String(50), default="official", comment="official官方报表/manual人工录入")
|
||||
remark = Column(String(500), nullable=True, comment="备注/口径说明")
|
||||
created_at = Column(DateTime, server_default=func.now())
|
||||
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
|
||||
__table_args__ = (UniqueConstraint("entity_id", "period", name="uq_cf_struct_entity_period"),)
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
"""现金流量表结构分析模块测试 — 八类企业定性判定 (P1 2026-09-03)
|
||||
|
||||
覆盖 cashflow_structure.py:
|
||||
GET /api/cma/cashflow-structure/rules 八类规则表
|
||||
GET /api/cma/cashflow-structure/periods 已注入期间
|
||||
GET /api/cma/cashflow-structure/analysis 结构分析+类型判定+瀑布数据
|
||||
POST /api/cma/cashflow-structure/values 三大活动净额注入(upsert)
|
||||
"""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||||
|
||||
BASE = "/api/cma/cashflow-structure"
|
||||
|
||||
|
||||
class TestRules:
|
||||
def test_rules_8_types(self, client: TestClient, db: Session):
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
resp = client.get(f"{BASE}/rules", headers=auth_header(token))
|
||||
assert resp.status_code == 200
|
||||
rules = resp.json()["rules"]
|
||||
assert len(rules) == 8
|
||||
# 判定组合必须恰好 8 种且无重复
|
||||
combos = [(r["o_sign"], r["i_sign"], r["f_sign"]) for r in rules]
|
||||
assert len(set(combos)) == 8
|
||||
# 规范命名：不含民间叫法
|
||||
assert all("现金流肖像" not in r["name"] for r in rules)
|
||||
|
||||
def test_rules_unauthorized(self, client: TestClient):
|
||||
resp = client.get(f"{BASE}/rules")
|
||||
assert resp.status_code in (401, 403)
|
||||
|
||||
|
||||
class TestStructureAnalysis:
|
||||
def _post_value(self, client, token, **kw):
|
||||
payload = {
|
||||
"period": "2026-08",
|
||||
"operating_net": -10.0,
|
||||
"investing_net": 3.0,
|
||||
"financing_net": 12.0, # - + + → 类型7 输血生存型
|
||||
"source": "official",
|
||||
"remark": "pytest 临时数据",
|
||||
}
|
||||
payload.update(kw)
|
||||
return client.post(f"{BASE}/values", headers=auth_header(token), json=payload)
|
||||
|
||||
def test_empty_analysis(self, client: TestClient, db: Session):
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
resp = client.get(f"{BASE}/analysis", headers=auth_header(token))
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["data_available"] is False
|
||||
assert body["classification"] is None
|
||||
|
||||
def test_upsert_and_classify_type7(self, client: TestClient, db: Session):
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
resp = self._post_value(client, token)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["ok"] is True
|
||||
assert body["classification"]["classified"] is True
|
||||
assert body["classification"]["type_no"] == 7
|
||||
assert body["classification"]["name"] == "输血生存型"
|
||||
|
||||
# 分析接口自动命中
|
||||
r2 = client.get(f"{BASE}/analysis", headers=auth_header(token), params={"period": "2026-08"})
|
||||
assert r2.status_code == 200
|
||||
a = r2.json()
|
||||
assert a["data_available"] is True
|
||||
assert a["classification"]["type_no"] == 7
|
||||
assert a["nets"]["operating"]["sign"] == "-"
|
||||
assert a["nets"]["investing"]["sign"] == "+"
|
||||
assert a["nets"]["financing"]["sign"] == "+"
|
||||
|
||||
# periods / values 可查
|
||||
r3 = client.get(f"{BASE}/periods", headers=auth_header(token))
|
||||
assert "2026-08" in r3.json()["periods"]
|
||||
|
||||
def test_upsert_overwrite_same_period(self, client: TestClient, db: Session):
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
self._post_value(client, token)
|
||||
resp2 = self._post_value(client, token, operating_net=50.0, investing_net=-20.0, financing_net=-10.0)
|
||||
assert resp2.status_code == 200
|
||||
assert resp2.json()["classification"]["type_no"] == 1 # + - - → 成熟健康型
|
||||
rows = client.get(f"{BASE}/values", headers=auth_header(token)).json()["items"]
|
||||
assert len(rows) == 1 # 同期间覆盖不重复插入
|
||||
|
||||
def test_type6_7_boundary_like_bohai(self, client: TestClient, db: Session):
|
||||
"""博海结构判定回归：经营弱/负 + 筹资依赖 → 类型6(- - +) / 类型7(- + +)"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
resp = self._post_value(client, token, operating_net=-5.0, investing_net=-2.0, financing_net=9.0)
|
||||
assert resp.json()["classification"]["type_no"] == 6
|
||||
resp = self._post_value(client, token, operating_net=-5.0, investing_net=2.0, financing_net=9.0)
|
||||
assert resp.json()["classification"]["type_no"] == 7
|
||||
|
||||
def test_invalid_period(self, client: TestClient, db: Session):
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
resp = self._post_value(client, token, period="2026/08")
|
||||
assert resp.status_code == 422
|
||||
Reference in New Issue
Block a user