"""
本土红旗五因子计算器（R1-R5）。
作为 Beneish M-Score 的本土业务场景补充，本模块提供以下可复算规则：
R1（净现比异常）：
net_profit_cur > 0 且 cfo_cur < 0 → 严重红旗
net_profit_cur > 0 且 cfo_cur > 0 且 cfo_cur/net_profit_cur < 0.5 → 普通红旗
R2（存货+资金异常增幅）：
任一分支满足即触发：
(A) monetary_funds YoY > 1.70 AND inventory YoY > 1.45
(B) (monetary_funds_cur + inventory_cur) / current_assets_cur > 0.82
R3（存贷双高）：
monetary_funds_cur/total_assets_cur > 0.20
AND (short_term_borrowings_cur+long_term_borrowings_cur)/total_assets_cur > 0.15
R4（关联方/应收款占比异常）：
accounts_receivable_cur / total_assets_cur > 0.18（通用阈值，可由行业参数覆盖）
R5（存货异常加强版）：
inventory YoY > 0.30 AND revenue YoY < 0.10
所有 None 情况均返回 triggered=False，不抛异常。
"""
from typing import Dict, Optional, Tuple
import logging
from config.settings import settings
logger = logging.getLogger(__name__)
class RedFlagCalculator:
"""本土红旗五因子（R1-R5）计算器。"""
R1_CFO_NEGATIVE = 0.0
R1_NETPROFIT_TO_CFO_RATIO = 0.5
R2_YOY_THRESHOLD_FUNDS = 1.7
R2_YOY_THRESHOLD_INVENTORY = 1.45
R2_FUNDS_INVENTORY_RATIO = 0.82
R3_FUNDS_TO_ASSETS = 0.20
R3_BORROWINGS_TO_ASSETS = 0.15
R4_AR_TO_ASSETS = 0.18
R5_INVENTORY_GROWTH = 0.30
R5_REVENUE_GROWTH = 0.10
INDUSTRY_ADJUSTMENTS: Dict[str, Dict[str, float]] = {
"白酒": {"R2_B": 0.92, "R3_funds_to_assets": 0.40},
"零售": {"R2_B": 0.90, "R3_funds_to_assets": 0.40},
"食品饮料": {"R2_B": 0.88, "R3_funds_to_assets": 0.32},
"半导体": {"R2_B": 0.84, "R3_funds_to_assets": 0.28},
"农牧": {"R2_B": 0.84},
"化工": {"R2_B": 0.84},
"房地产": {"R2_B": 0.84},
"汽车": {"R3_funds_to_assets": 0.28},
"通信": {"R4": 0.24},
"科技": {"R4": 0.22},
"军工": {"R4": 0.22},
"医药": {"R4": 0.22},
}
@staticmethod
def _safe_get(data: Dict, key: str, idx: int) -> Optional[float]:
"""安全获取 (current, previous) 元组/列表的某一位。
兼容 tuple 和 list（JSON 反序列化产生 list，代码内用 tuple），
与 f_score.py._safe_get 行为对齐。
"""
v = data.get(key, (None, None))
if not isinstance(v, (tuple, list)) or len(v) <= idx:
return None
return v[idx]
@staticmethod
def _yoy(current: Optional[float], previous: Optional[float]) -> Optional[float]:
"""计算同比 ratio = current/previous，None 或零分母返回 None"""
if current is None or previous is None:
return None
if previous == 0:
return None
return current / previous
@classmethod
def _get_industry_threshold(
cls, industry: Optional[str], threshold_key: str, default: float
) -> float:
"""
获取行业修正后的阈值。
如果行业在 INDUSTRY_ADJUSTMENTS 中且该 key 存在，返回行业专用值；
否则返回通用默认值。
"""
if industry and industry in cls.INDUSTRY_ADJUSTMENTS:
adjusted = cls.INDUSTRY_ADJUSTMENTS[industry].get(threshold_key)
if adjusted is not None:
return adjusted
return default
def calculate_r1(self, financial_data: Dict[str, Tuple[float, float]]) -> Dict:
"""
R1（净现比异常）：
net_profit_cur > 0 且 cfo_cur < 0 → 严重红旗
net_profit_cur > 0 且 cfo_cur > 0 且 cfo_cur/net_profit_cur < 0.5 → 普通红旗
"""
net_profit_cur = self._safe_get(financial_data, 'net_profit', 0)
cfo_cur = self._safe_get(financial_data, 'cash_flow_operations', 0)
triggered = False
severity = "未触发"
reason = ""
detail = {
'net_profit_cur': net_profit_cur,
'cfo_cur': cfo_cur,
'cfo_to_net_profit_ratio': None,
}
if net_profit_cur is None or cfo_cur is None:
return self._build_r1_result(False, "未触发", "数据缺失", detail)
if net_profit_cur > 0 and cfo_cur < 0:
triggered = True
severity = "严重红旗"
reason = (f"净利润为正({net_profit_cur:.2f})但经营现金流为负"
f"({cfo_cur:.2f})，盈余质量严重可疑")
elif net_profit_cur > 0 and cfo_cur > 0:
if net_profit_cur == 0:
ratio = None
else:
ratio = cfo_cur / net_profit_cur
detail['cfo_to_net_profit_ratio'] = ratio
if ratio is not None and ratio < self.R1_NETPROFIT_TO_CFO_RATIO:
triggered = True
severity = "普通红旗"
reason = (f"经营现金流/净利润={ratio:.3f} < {self.R1_NETPROFIT_TO_CFO_RATIO}，"
f"净利润缺少现金流支撑")
return self._build_r1_result(triggered, severity, reason, detail)
@staticmethod
def _build_r1_result(triggered: bool, severity: str, reason: str, detail: Dict) -> Dict:
return {
'flag_id': 'R1',
'name': '净现比异常',
'description': '净利润与经营活动现金流的匹配度',
'triggered': triggered,
'severity': severity,
'reason': reason,
'detail': detail,
}
def calculate_r2(self, financial_data: Dict[str, Tuple[float, float]],
industry: Optional[str] = None) -> Dict:
"""
R2（存货+资金异常增幅）：
(A) monetary_funds YoY > 1.7 AND inventory YoY > 1.45
(B) (monetary_funds_cur + inventory_cur) / current_assets_cur > 阈值（含行业修正）
任一分支满足即触发。
"""
funds_cur = self._safe_get(financial_data, 'monetary_funds', 0)
funds_prev = self._safe_get(financial_data, 'monetary_funds', 1)
inv_cur = self._safe_get(financial_data, 'inventory', 0)
inv_prev = self._safe_get(financial_data, 'inventory', 1)
ca_cur = self._safe_get(financial_data, 'current_assets', 0)
funds_yoy = self._yoy(funds_cur, funds_prev)
inv_yoy = self._yoy(inv_cur, inv_prev)
branch_a_triggered = (
funds_yoy is not None
and inv_yoy is not None
and funds_yoy > self.R2_YOY_THRESHOLD_FUNDS
and inv_yoy > self.R2_YOY_THRESHOLD_INVENTORY
)
r2b_threshold = self._get_industry_threshold(industry, "R2_B", self.R2_FUNDS_INVENTORY_RATIO)
branch_b_triggered = False
funds_inv_ratio = None
if (funds_cur is not None and funds_cur != 0
and inv_cur is not None and inv_cur != 0
and ca_cur is not None and ca_cur != 0):
funds_inv_ratio = (funds_cur + inv_cur) / ca_cur
if funds_inv_ratio > r2b_threshold:
branch_b_triggered = True
triggered = branch_a_triggered or branch_b_triggered
reasons = []
if branch_a_triggered:
reasons.append(
f"分支A：货币资金YoY={funds_yoy:.2f}>{self.R2_YOY_THRESHOLD_FUNDS} "
f"且 存货YoY={inv_yoy:.2f}>{self.R2_YOY_THRESHOLD_INVENTORY}"
)
if branch_b_triggered:
threshold_type = "行业" if r2b_threshold != self.R2_FUNDS_INVENTORY_RATIO else "通用"
reasons.append(
f"分支B：(资金+存货)/流动资产={funds_inv_ratio:.3f}>{r2b_threshold}（{threshold_type}阈值）"
)
if not triggered:
reasons.append("未达到任一分支阈值")
severity = "普通红旗" if triggered else "未触发"
detail = {
'funds_cur': funds_cur,
'funds_prev': funds_prev,
'funds_yoy': funds_yoy,
'inventory_cur': inv_cur,
'inventory_prev': inv_prev,
'inventory_yoy': inv_yoy,
'current_assets_cur': ca_cur,
'funds_inventory_to_current_assets': funds_inv_ratio,
'branch_a_triggered': branch_a_triggered,
'branch_b_triggered': branch_b_triggered,
}
return {
'flag_id': 'R2',
'name': '存货+资金异常增幅',
'description': '货币资金/存货同比异常增长 或 占流动资产比例过高',
'triggered': triggered,
'severity': severity,
'reason': "；".join(reasons),
'detail': detail,
}
def calculate_r3(self, financial_data: Dict[str, Tuple[float, float]],
industry: Optional[str] = None) -> Dict:
"""
R3（存贷双高）：
monetary_funds_cur/total_assets_cur > 阈值（含行业修正）
AND (short_term_borrowings_cur+long_term_borrowings_cur)/total_assets_cur > 0.15
同时满足 → 触发
"""
funds_cur = self._safe_get(financial_data, 'monetary_funds', 0)
ta_cur = self._safe_get(financial_data, 'total_assets', 0)
stb_cur = self._safe_get(financial_data, 'short_term_borrowings', 0)
ltb_cur = self._safe_get(financial_data, 'long_term_borrowings', 0)
r3_funds_threshold = self._get_industry_threshold(industry, "R3_funds_to_assets", self.R3_FUNDS_TO_ASSETS)
funds_to_assets = None
if funds_cur is not None and ta_cur is not None and ta_cur != 0:
funds_to_assets = funds_cur / ta_cur
borrowings_to_assets = None
if (stb_cur is not None or ltb_cur is not None) and ta_cur is not None and ta_cur != 0:
stb = stb_cur if stb_cur is not None else 0.0
ltb = ltb_cur if ltb_cur is not None else 0.0
borrowings_to_assets = (stb + ltb) / ta_cur
cond_a = (funds_to_assets is not None
and funds_to_assets > r3_funds_threshold)
cond_b = (borrowings_to_assets is not None
and borrowings_to_assets > self.R3_BORROWINGS_TO_ASSETS)
triggered = cond_a and cond_b
severity = "严重红旗" if triggered else "未触发"
threshold_type = "行业" if r3_funds_threshold != self.R3_FUNDS_TO_ASSETS else "通用"
reasons = []
if cond_a:
reasons.append(
f"资金/总资产={funds_to_assets:.3f}>{r3_funds_threshold}（{threshold_type}阈值）"
)
else:
reasons.append(
f"资金/总资产={'N/A' if funds_to_assets is None else f'{funds_to_assets:.3f}'}"
f"未超{r3_funds_threshold}"
)
if cond_b:
reasons.append(
f"借款/总资产={borrowings_to_assets:.3f}>{self.R3_BORROWINGS_TO_ASSETS}"
)
else:
reasons.append(
f"借款/总资产={'N/A' if borrowings_to_assets is None else f'{borrowings_to_assets:.3f}'}"
f"未超{self.R3_BORROWINGS_TO_ASSETS}"
)
if not triggered:
reasons.append("两条件未同时满足")
detail = {
'funds_cur': funds_cur,
'total_assets_cur': ta_cur,
'short_term_borrowings_cur': stb_cur,
'long_term_borrowings_cur': ltb_cur,
'funds_to_assets': funds_to_assets,
'funds_to_assets_threshold': r3_funds_threshold,
'borrowings_to_assets': borrowings_to_assets,
'cond_a_satisfied': cond_a,
'cond_b_satisfied': cond_b,
}
return {
'flag_id': 'R3',
'name': '存贷双高',
'description': f'货币资金占总资产>{r3_funds_threshold*100:.0f}% 且 短借+长借占总资产>{self.R3_BORROWINGS_TO_ASSETS*100:.0f}%',
'triggered': triggered,
'severity': severity,
'reason': "；".join(reasons),
'detail': detail,
}
def calculate_r4(self, financial_data: Dict[str, Tuple[float, float]],
industry: Optional[str] = None) -> Dict:
"""
R4（关联方/应收款占比异常）：
accounts_receivable_cur / total_assets_cur > 阈值（含行业修正）→ 严重红旗
（应收款占总资产比例过高，可能存在虚增收入或关联方占款）
"""
ar_cur = self._safe_get(financial_data, 'accounts_receivable', 0)
ta_cur = self._safe_get(financial_data, 'total_assets', 0)
r4_threshold = self._get_industry_threshold(industry, "R4", self.R4_AR_TO_ASSETS)
ar_to_assets = None
if ar_cur is not None and ta_cur is not None and ta_cur != 0:
ar_to_assets = ar_cur / ta_cur
triggered = (ar_to_assets is not None
and ar_to_assets > r4_threshold)
severity = "严重红旗" if triggered else "未触发"
threshold_type = "行业" if r4_threshold != self.R4_AR_TO_ASSETS else "通用"
if triggered:
reason = (f"应收账款/总资产={ar_to_assets:.3f} > {r4_threshold}（{threshold_type}阈值）"
f"，应收款占比过高，可能存在虚增收入或关联方占款")
elif ar_to_assets is not None:
reason = (f"应收账款/总资产={ar_to_assets:.3f} 未超 {r4_threshold}")
else:
reason = "数据缺失"
detail = {
'accounts_receivable_cur': ar_cur,
'total_assets_cur': ta_cur,
'ar_to_assets': ar_to_assets,
'threshold': r4_threshold,
'threshold_type': threshold_type,
}
return {
'flag_id': 'R4',
'name': '关联方/应收款占比异常',
'description': f'应收账款/总资产 > {r4_threshold*100:.0f}%（虚增收入或关联方占款风险）',
'triggered': triggered,
'severity': severity,
'reason': reason,
'detail': detail,
}
def calculate_r5(self, financial_data: Dict[str, Tuple[float, float]]) -> Dict:
"""
R5（存货异常加强版）：
inventory 同比增长 > 30% AND revenue 同比增长 < 10% → 普通红旗
（存货大增但营收停滞，存货真实性存疑）
与 R2 区分：R2 关注"资金+存货双增"，R5 关注"存货增但营收不增"
"""
inv_cur = self._safe_get(financial_data, 'inventory', 0)
inv_prev = self._safe_get(financial_data, 'inventory', 1)
rev_cur = self._safe_get(financial_data, 'revenue', 0)
rev_prev = self._safe_get(financial_data, 'revenue', 1)
inv_yoy = self._yoy(inv_cur, inv_prev)
rev_yoy = self._yoy(rev_cur, rev_prev)
inv_growth = None
if inv_yoy is not None:
inv_growth = inv_yoy - 1.0
rev_growth = None
if rev_yoy is not None:
rev_growth = rev_yoy - 1.0
triggered = (
inv_growth is not None and inv_growth > self.R5_INVENTORY_GROWTH
and rev_growth is not None and rev_growth < self.R5_REVENUE_GROWTH
)
severity = "普通红旗" if triggered else "未触发"
reasons = []
if inv_growth is not None:
if inv_growth > self.R5_INVENTORY_GROWTH:
reasons.append(f"存货同比+{inv_growth*100:.1f}% > {self.R5_INVENTORY_GROWTH*100:.0f}%")
else:
reasons.append(f"存货同比+{inv_growth*100:.1f}% 未超 {self.R5_INVENTORY_GROWTH*100:.0f}%")
else:
reasons.append("存货同比数据缺失")
if rev_growth is not None:
if rev_growth < self.R5_REVENUE_GROWTH:
reasons.append(f"营收同比+{rev_growth*100:.1f}% < {self.R5_REVENUE_GROWTH*100:.0f}%")
else:
reasons.append(f"营收同比+{rev_growth*100:.1f}% 未低于 {self.R5_REVENUE_GROWTH*100:.0f}%")
else:
reasons.append("营收同比数据缺失")
if not triggered:
reasons.append("两条件未同时满足")
detail = {
'inventory_cur': inv_cur,
'inventory_prev': inv_prev,
'inventory_growth': inv_growth,
'revenue_cur': rev_cur,
'revenue_prev': rev_prev,
'revenue_growth': rev_growth,
}
return {
'flag_id': 'R5',
'name': '存货异常加强版',
'description': '存货同比增长>30%且营收增长<10%（存货真实性存疑）',
'triggered': triggered,
'severity': severity,
'reason': "；".join(reasons),
'detail': detail,
}
def evaluate_all(self, financial_data: Dict[str, Tuple[float, float]],
industry: Optional[str] = None) -> Dict:
"""
综合计算 R1/R2/R3/R4/R5 并返回汇总结果。
输出口径：
total_red_flags / triggered_count = 实际触发项数
severity_score = 严重红旗 2 分 + 普通红旗 1 分
升级规则（risk_upgrade）：
triggered_count ≥ 3 → "高风险（M-Score盲区警告）"
triggered_count = 2 → "中高风险（M-Score盲区警告）"
triggered_count = 1 → "关注"
0 → None
Args:
financial_data: 财务数据字典
industry: 行业分类（可选），用于行业专用阈值修正
"""
r1 = self.calculate_r1(financial_data)
r2 = self.calculate_r2(financial_data, industry=industry)
r3 = self.calculate_r3(financial_data, industry=industry)
r4 = self.calculate_r4(financial_data, industry=industry)
r5 = self.calculate_r5(financial_data)
flags = [r1, r2, r3, r4, r5]
severity_score = 0
for f in flags:
if not f['triggered']:
continue
sev = f['severity']
if sev == "严重红旗":
severity_score += 2
elif sev == "普通红旗":
severity_score += 1
triggered_flags = [f for f in flags if f['triggered']]
triggered_count = len(triggered_flags)
if triggered_count >= 3:
risk_upgrade = "高风险（M-Score盲区警告）"
elif triggered_count == 2:
risk_upgrade = "中高风险（M-Score盲区警告）"
elif triggered_count == 1:
risk_upgrade = "关注"
else:
risk_upgrade = None
result = {
'R1': r1,
'R2': r2,
'R3': r3,
'R4': r4,
'R5': r5,
'total_red_flags': triggered_count,
'triggered_count': triggered_count,
'severity_score': severity_score,
'risk_upgrade': risk_upgrade,
'triggered_flag_ids': [f['flag_id'] for f in triggered_flags],
'triggered_flag_names': [f['name'] for f in triggered_flags],
}
result = self.attach_llm_interpretations(result)
return result
@staticmethod
def _build_interpretation_system_prompt() -> str:
"""构建红旗解读系统 prompt"""
return (
"你是一名资深审计专家。请用50-80字中文解读以下财务红旗。"
"要求：1. 说明该红旗的财务含义与数据可疑点；"
"2. 对应常见舞弊手法（必须使用虚构公司名案例，如"
"'某制药公司2017''某白酒企业2025''某新材料公司2017'"
"'某水产公司2017''某中药公司2019'，不允许出现任何真实A股上市公司名称）；"
"3. 给出针对该红旗的1条具体审计建议。"
"不要输出免责套话，严格控制在80字以内。"
)
@staticmethod
def _build_interpretation_user_prompt(flag_id: str, detail: dict) -> str:
"""构建红旗解读用户 prompt"""
flag_desc = {
'R1': '净现比异常红旗（盈利企业经营现金流/净利润 < 0.5 或现金流为负）',
'R2': '存货+资金异常增幅红旗（货币资金和存货同比大幅增长 或 两者合计占流动资产比例过高）',
'R3': '存贷双高红旗（账上大量货币资金同时持有大量有息负债）',
'R4': '关联方/应收款占比异常红旗（通用阈值18%，部分行业按行业阈值覆盖；以触发数据中的threshold为准，可能存在虚增收入或关联方占款）',
'R5': '存货异常加强版红旗（存货同比增长>30%但营收增长<10%，存货真实性存疑）',
}
return (
f"红旗类型：{flag_desc.get(flag_id, flag_id)}\n"
f"触发数据：{detail}\n\n"
f"请按系统指令生成50-80字中文深度解读。"
)
def get_llm_interpretation(self, flag_id: str, flag_detail: dict) -> str:
"""
调用昇腾 API 生成红旗因子的 AI 深度解读。
永不抛异常——所有失败场景返回兜底字符串。
"""
if not settings.LLM_ASCEND.get('enabled', True):
return "AI 解读未启用（离线演示模式）"
try:
from core.ascend_adapter import AscendLLMClient
client = AscendLLMClient()
system_prompt = self._build_interpretation_system_prompt()
user_prompt = self._build_interpretation_user_prompt(flag_id, flag_detail)
return client.chat_completion(
user_prompt=user_prompt,
system_prompt=system_prompt,
)
except RuntimeError as e:
msg = str(e)
if "超时" in msg or "不可用" in msg or "失败" in msg or "未配置" in msg or "空" in msg:
logger.warning(f"Ascend LLM 调用失败（{flag_id}）: {msg}")
return f"AI 解读生成失败（API不可用或超时）"
logger.warning(f"Ascend LLM 配置异常（{flag_id}）: {msg}")
return f"AI 解读生成失败（配置异常）"
except Exception as e:
logger.warning(f"Ascend LLM 未知错误（{flag_id}）: {str(e)[:200]}")
return f"AI 解读生成失败（未知错误）"
def attach_llm_interpretations(self, red_flags_result: dict) -> dict:
"""
为已触发的红旗追加 llm_interpretation 字段。
未触发的红旗塞空字符串。
不修改原数据其他字段。
"""
for flag_id in ['R1', 'R2', 'R3', 'R4', 'R5']:
flag = red_flags_result.get(flag_id, {})
if flag.get('triggered', False):
flag['llm_interpretation'] = self.get_llm_interpretation(
flag_id, flag.get('detail', {})
)
else:
flag['llm_interpretation'] = ''
return red_flags_result
if __name__ == "__main__":
calc = RedFlagCalculator()
test_case_2017 = {
'revenue': (26476970977.57, 21642324070.28),
'net_profit': (4100926077.16, 3340000000.00),
'cash_flow_operations': (1842794237.84, 1603189351.32),
'monetary_funds': (34151000000.00, 27325000000.00),
'inventory': (15700000000.00, 12619000000.00),
'short_term_borrowings': (11370000000.00, 8252000000.00),
'long_term_borrowings': (None, None),
'total_assets': (68722020630.61, 54823896576.81),
'current_assets': (56479077718.23, 44461544324.71),
}
result = calc.evaluate_all(test_case_2017)
import json
print(json.dumps(result, ensure_ascii=False, indent=2, default=str))