diff --git a/backend/tests/test_seven_flow_rules.py b/backend/tests/test_seven_flow_rules.py new file mode 100644 index 00000000..acbacd0f --- /dev/null +++ b/backend/tests/test_seven_flow_rules.py @@ -0,0 +1,243 @@ +"""七流一致自检规则测试 — 金税四期七流框架 + +覆盖 /root/cma_seven_flow_design_20260825.md 的7条自检规则: +1. 进销项倒挂 +2. 应交税费异常 +3. 账面毛利异常 +4. 收入vs现金 +5. 收入vs存货 +6. 人员vs收入 +7. 成本vs收入 + +规则是纯函数(不依赖DB),测试直接验证判定逻辑。 +""" + +import pytest + + +# ── 七流自检规则引擎(与设计方案一致) ── + +def rule_invoice_inverted(input_tax, output_tax): + """规则1:进项税额 > 销项税额×1.2 → 倒挂风险""" + if output_tax <= 0: + return None + ratio = input_tax / output_tax + return {"passed": ratio <= 1.2, "value": round(ratio, 2), + "threshold": 1.2, "level": "red" if ratio > 1.2 else "green", + "message": f"进项/销项比 {ratio:.2f}(阈值1.2)"} + + +def rule_tax_payable_ratio(tax_payable, income): + """规则2:应交税费/收入 > 20% → 异常""" + if income <= 0: + return None + ratio = tax_payable / income + return {"passed": ratio <= 0.2, "value": round(ratio * 100, 1), + "threshold": 20.0, "level": "red" if ratio > 0.2 else "green", + "message": f"应交税费/收入 {ratio*100:.1f}%(阈值20%)"} + + +def rule_margin_abnormal(margin_rate, industry_base=10.0): + """规则3:毛利率 < 行业基准×0.5 → 异常(需解释链)""" + threshold = industry_base * 0.5 + return {"passed": margin_rate >= threshold, "value": round(margin_rate, 2), + "threshold": threshold, "level": "red" if margin_rate < threshold else "green", + "message": f"毛利率 {margin_rate}%(行业基准{industry_base}%×0.5={threshold}%)"} + + +def rule_income_vs_cash(cashflow, revenue_growth): + """规则4:经营现金流为负 且 收入负增长 → 预警""" + risky = (cashflow < 0) and (revenue_growth < 0) + return {"passed": not risky, "value": f"现金流{cashflow}, 增长{revenue_growth}%", + "level": "red" if risky else "green", + "message": "现金流负+收入负增长" if risky else "现金流/收入正常"} + + +def rule_income_vs_inventory(revenue_growth, inventory_change): + """规则5:收入降>20% 但 存货不降 → 预警""" + risky = (revenue_growth < -20) and (inventory_change >= 0) + return {"passed": not risky, "value": f"收入{revenue_growth}%, 存货{inventory_change}%", + "level": "yellow" if risky else "green", + "message": "收入下滑但存货未降" if risky else "收入/存货匹配"} + + +def rule_personnel_vs_income(revenue_growth, employee_change): + """规则6:收入降>20% 但 人员不变 → 预警""" + risky = (revenue_growth < -20) and (employee_change == 0) + return {"passed": not risky, "value": f"收入{revenue_growth}%, 人员{employee_change}%", + "level": "yellow" if risky else "green", + "message": "收入下滑但人员未变" if risky else "收入/人员匹配"} + + +def rule_cost_vs_income(cost_rate): + """规则7:成本率 > 100% → 异常(需厂补/渠补解释)""" + return {"passed": cost_rate <= 100, "value": round(cost_rate, 1), + "threshold": 100.0, "level": "red" if cost_rate > 100 else "green", + "message": f"成本率 {cost_rate:.1f}%(阈值100%)"} + + +def seven_flow_score(checks): + """七流健康评分:green=100, yellow=60, red=0,取平均""" + scores = {"green": 100, "yellow": 60, "red": 0} + if not checks: + return 0 + return round(sum(scores.get(c["level"], 50) for c in checks) / len(checks)) + + +# ── 测试类 ── + +class TestRuleInvoiceInverted: + def test_normal(self): + r = rule_invoice_inverted(100, 120) + assert r["passed"] is True + assert r["level"] == "green" + + def test_inverted(self): + r = rule_invoice_inverted(200, 120) # 1.67 > 1.2 + assert r["passed"] is False + assert r["level"] == "red" + + def test_boundary(self): + r = rule_invoice_inverted(120, 100) # 1.2 = 阈值 + assert r["passed"] is True + + def test_zero_output(self): + assert rule_invoice_inverted(100, 0) is None + + +class TestRuleTaxPayableRatio: + def test_normal(self): + r = rule_tax_payable_ratio(10, 100) # 10% + assert r["passed"] is True + + def test_abnormal(self): + r = rule_tax_payable_ratio(429, 100) # 429% (酣客场景) + assert r["passed"] is False + assert r["level"] == "red" + + def test_zero_income(self): + assert rule_tax_payable_ratio(10, 0) is None + + +class TestRuleMarginAbnormal: + def test_normal_margin(self): + r = rule_margin_abnormal(13.0, 10.0) # 博海IT经销13% + assert r["passed"] is True + + def test_abnormal_margin(self): + r = rule_margin_abnormal(-0.43, 10.0) # 酣客账面-0.43% + assert r["passed"] is False + assert r["level"] == "red" + + def test_threshold_boundary(self): + r = rule_margin_abnormal(5.0, 10.0) # = 5 = 10×0.5 + assert r["passed"] is True + + +class TestRuleIncomeVsCash: + def test_both_negative(self): + r = rule_income_vs_cash(-13.89, -37.0) # 酣客:现金流负+收入负增长 + assert r["passed"] is False + assert r["level"] == "red" + + def test_positive_cash(self): + r = rule_income_vs_cash(33.86, -37.0) + assert r["passed"] is True + + def test_negative_cash_positive_growth(self): + r = rule_income_vs_cash(-5, 10) + assert r["passed"] is True + + +class TestRuleIncomeVsInventory: + def test_income_down_inventory_up(self): + r = rule_income_vs_inventory(-37, 5) # 收入降37%存货增5% + assert r["passed"] is False + assert r["level"] == "yellow" + + def test_income_down_inventory_down(self): + r = rule_income_vs_inventory(-37, -2) # 都降 + assert r["passed"] is True + + def test_slight_income_drop(self): + r = rule_income_vs_inventory(-10, 5) # 降幅<20% + assert r["passed"] is True + + +class TestRulePersonnelVsIncome: + def test_income_down_staff_same(self): + r = rule_personnel_vs_income(-37, 0) # 收入降37%人员不变 + assert r["passed"] is False + assert r["level"] == "yellow" + + def test_staff_reduced(self): + r = rule_personnel_vs_income(-37, -10) # 人员也降 + assert r["passed"] is True + + def test_income_stable(self): + r = rule_personnel_vs_income(-10, 0) + assert r["passed"] is True + + +class TestRuleCostVsIncome: + def test_cost_over_100(self): + r = rule_cost_vs_income(100.4) # 酣客成本率100.4% + assert r["passed"] is False + assert r["level"] == "red" + + def test_cost_normal(self): + r = rule_cost_vs_income(87.0) # 博海成本率87% + assert r["passed"] is True + + +class TestSevenFlowScore: + def test_all_green(self): + checks = [{"level": "green"}, {"level": "green"}, {"level": "green"}] + assert seven_flow_score(checks) == 100 + + def test_mixed(self): + checks = [{"level": "green"}, {"level": "yellow"}, {"level": "red"}] + assert seven_flow_score(checks) == 53 # (100+60+0)/3 + + def test_all_red(self): + checks = [{"level": "red"}, {"level": "red"}] + assert seven_flow_score(checks) == 0 + + def test_empty(self): + assert seven_flow_score([]) == 0 + + +class TestHankeScenario: + """酣客2026H1真实场景(应触发多项红色预警)""" + + def test_full_hanke_checks(self): + checks = [ + rule_invoice_inverted(input_tax=27157, output_tax=31445), # 进项2.7亿vs销项3.1亿 + rule_tax_payable_ratio(tax_payable=429, income=713), # 应交429万vs收入713万 + rule_margin_abnormal(margin_rate=-0.43, industry_base=10), # 账面毛利-0.43% + rule_income_vs_cash(cashflow=-13.89, revenue_growth=-37), # 现金流负+收入负增长 + rule_income_vs_inventory(revenue_growth=-37, inventory_change=0), # 收入降库存平 + rule_personnel_vs_income(revenue_growth=-37, employee_change=0), # 收入降人员平 + rule_cost_vs_income(cost_rate=100.4), # 成本率100.4% + ] + reds = [c for c in checks if c and c["level"] == "red"] + yellows = [c for c in checks if c and c["level"] == "yellow"] + score = seven_flow_score([c for c in checks if c]) + assert len(reds) >= 3, f"酣客应至少3项红色, 实际{len(reds)}: {[c['message'] for c in reds]}" + assert len(yellows) >= 1 + assert score < 60, f"酣客健康评分应<60, 实际{score}" + + +class TestBohaiScenario: + """博海2026H1真实场景(应风险较低)""" + + def test_full_bohai_checks(self): + checks = [ + rule_margin_abnormal(margin_rate=13.0, industry_base=10), # 毛利率13%合理 + rule_cost_vs_income(cost_rate=87.0), # 成本率87% + rule_tax_payable_ratio(tax_payable=5, income=100), # 正常税负 + ] + reds = [c for c in checks if c and c["level"] == "red"] + score = seven_flow_score([c for c in checks if c]) + assert len(reds) == 0, f"博海不应有红色预警: {[c['message'] for c in reds]}" + assert score >= 60, f"博海健康评分应>=60, 实际{score}"