简介:面向银行风控、普惠金融与信贷科技从业者及大模型工程人员,这份606页技术方案围绕小微企业信用评估展开,系统讲解以税务数据与经营信息交叉验证为核心的DeepSeek-R1落地路径。全文分61个大章节,从行业痛点与整体架构切入,依次覆盖多源税务数据采集与标准化、经营信息采集接口设计、交叉验证核心逻辑、非结构化票据结构化处理、双维度去噪、税务与经营特征工程及注意力融合机制,并延伸至信用标签体系、标注一致性校验、分布式标注框架与增量预训练、混合精度训练等环节,技术链路完整。压缩包内含1个PDF文件,约16.12MB,支持目录章节跳转与阅读器左侧书签大纲快速定位,图表文字显示正常,便于按模块查阅。目前已有97人学习。适合需要搭建普惠金融风控方案、关注大模型在信贷场景适配细节的读者对照参考。
1. 为什么小微企业信用评估必须让税务数据和经营信息互相作证
一家年开票 800 万的贸易公司来申请 200 万信用贷,客户经理手上通常只有三份材料:代账公司做的财报、近 6 个月的对公流水、年度汇总的纳税申报表。任何一份单独拿出来都不足以证明"这家企业真的有这么多生意"——财报可以调,流水可以走账,申报表只有汇总数看不出结构。能做证的,是税务侧的申报与开票数据和经营侧的账户流水、社保、用电、工商年报之间能不能互相咬合。
标题里那份 606 页方案,核心不在模型多复杂,而在于以统一社会信用代码为锚,把两条数据链路对齐到同一时间粒度,再用 DeepSeek 这类大模型承担非结构化经营信息的抽取与差异归因。它解决的是普惠金融里最实际的问题:没有抵押物、没有审计报告的小微主体,怎么把"可信"这件事量化出来。适合银行风控科技、数据中台、小微信贷模型岗位的工程师对照落地。
2. 税务数据与经营信息交叉验证的字段口径对齐
交叉验证失败,八成不是数据造假,而是口径没对齐。税务销售额是月报还是季报、含税还是不含税、总公司申报还是分支机构申报,这三点任意一个错位,算出来的偏差率就全是噪声。所以规则引擎之前,先做一次字段对齐。
2.1 税务侧能拿到什么:申报表、发票与纳税信用等级
税务侧可用的数据大致分四类。增值税申报表给出按税率分列的销售额、进项税额和应纳税额,这是判断营收规模最硬的依据;发票数据给出开票金额、作废与红冲记录、上下游客户的名称和税号、开票频次,能还原真实的交易对手结构;企业所得税年度申报表里有营业收入、利润总额、从业人数、资产总额,用来校验规模量级;纳税信用等级(A/B/M/C/D)反映的是合规性,不是规模,它的价值在于做风险分层而不是授信额度。
口径上有三个坑必须提前处理。第一,增值税申报表主表销售额是不含税口径,而发票明细通常是价税合计,两者直接相减必然对不上,需要先用适用税率还原。第二,小规模纳税人多数按季申报,一般纳税人按月申报,抽样对比时必须把季报拆到月或把月报聚到季,否则会出现"某月税务销售额为零"的假象。第三,分支机构独立申报与总机构汇总缴纳要区分,集团型企业的税务数据不能简单按主体加总。
2.2 经营信息侧的四个来源与清洗要点
经营信息侧主要看四块:对公账户流水、工商年报、社保参保、以及替代性数据(用电、用水、物流结算)。对公账户流水的价值不在余额而在贷方发生额与交易对手结构;工商年报里的从业人数和资产状况由企业自行填报,公示意愿低的企业常常字段为空;社保参保人数和缴费基数是最难伪造的指标之一;用电量适合制造业,对贸易和服务类企业参考价值有限。
清洗环节要剔除的东西很明确:关联方之间的内部转账、保证金和理财赎回产生的过路资金、集团资金池的归集与下拨、以及同名账户之间的内部划转。这些资金会让贷方发生额虚高数倍,如果不剔除,交叉验证的结论会系统性偏乐观。
| 交叉验证维度 | 税务侧字段 | 经营侧字段 | 主要偏差来源 |
|---|---|---|---|
| 营收规模 | 增值税不含税销售额 | 对公账户贷方发生额(净额) | 含税口径、资金池归集 |
| 用工规模 | 企业所得税从业人数 | 社保参保人数 | 劳务派遣、年报随意填报 |
| 交易结构 | 开票下游客户数与集中度 | 流水交易对手数 | 代开发票、关联交易 |
| 合规状态 | 纳税信用等级 | 欠税/行政处罚 | 时间滞后、评级更新周期 |
2.3 用统一社会信用代码做主键对齐
税务与经营两套系统的主体标识经常不一致:有的是 15 位纳税人识别号,有的是 18 位统一社会信用代码,名称里还带括号全半角、空格、"(有限)"与"(有限公司)"混写。对齐时必须先做标准化再做连接,不能直接按名称 join。
import re def normalize_uscc(raw: str) -> str | None: """统一社会信用代码标准化:去空格、转大写,校验 18 位长度""" if not raw: return None s = re.sub(r"\s+", "", str(raw)).upper() return s if re.fullmatch(r"[0-9A-HJ-NPQRTUWXY]{18}", s) else None def normalize_name(raw: str) -> str: """企业名称标准化:去括号差异、去空格、统一公司后缀""" if not raw: return "" s = str(raw).strip().replace("(", "(").replace(")", ")") s = re.sub(r"\s+", "", s) return s.replace("有限公司", "有限责任公司") # 归一到同一后缀 # 先按信用代码连接,再对代码缺失的记录用「标准化名称 + 法定代表人」兜底匹配normalize_uscc里的正则排除了 I、O、S、V、Z 这几个容易与数字混淆的字母,这是统一社会信用代码本身的编码规则决定的,不做这层校验会把 OCR 误识别的结果当成有效主体。normalize_name解决的是兜底匹配场景——当某一侧只有名称没有代码时,用它加上法定代表人姓名做组合键,命中率通常能到九成以上,但一定要标记为"弱匹配",后续打分时降权处理。
注意:弱匹配记录不要直接进入自动审批通道,人工复核比例建议不低于 20%,否则一个名称标准化规则写错就会批量串户。
3. 用 DeepSeek API 搭一条可跑批的交叉验证推理链
规则引擎只能处理数值比对,而实际授信材料里有大量非结构化内容:经营范围变更说明、上下游合同摘要、企业解释流水异常的情况说明。这部分交给 DeepSeek 做抽取和归因,比写几十条正则更省事。关键是把模型放在"证据整理"的位置,不要让它直接给授信结论。
3.1 最小可跑的调用:curl 与 OpenAI 兼容接口
开放平台的接口是 OpenAI 兼容格式,先用 curl 把链路跑通再写工程代码。
curl https://api.deepseek.com/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer $DEEPSEEK_API_KEY" \ -d '{ "model": "deepseek-chat", "temperature": 0, "max_tokens": 1200, "response_format": {"type": "json_object"}, "messages": [ {"role": "system", "content": "你是银行小微风控审核助手。只能基于用户提供的字段作答,缺失字段填 null,禁止推测和补充外部信息。"}, {"role": "user", "content": "税务不含税销售额(万元):812.4; 对公贷方发生额净额(万元):486.2; 开票下游客户数:23; 社保参保人数:0; 工商年报从业人数:11。输出 JSON: {\"items\":[{\"field\":\"\",\"tax_value\":\"\",\"biz_value\":\"\",\"deviation_rate\":0,\"verdict\":\"consistent|suspicious|conflict\",\"reason\":\"\"}]}"} ] }'temperature设为 0 是为了让同一批数据多次调用得到一致输出,风控场景里可复现比多样性重要得多。response_format指定json_object能强制模型输出可被json.loads解析的结构,省掉正则清洗。max_tokens要留足余量,输出被截断时会得到半截 JSON,解析直接抛异常。verdict的三档枚举是刻意设计的:把判断压缩成有限选项,比让模型自由描述更容易做后续聚合统计。
3.2 Prompt 结构与结构化输出约束
工程里不要手拼字符串,把字段白名单和输出契约固定下来,方便版本管理。
import json FIELDS = ["tax_sales_amt", "bank_credit_amt", "invoice_cust_cnt", "social_ins_cnt", "annual_emp_cnt"] SYSTEM_PROMPT = ( "你是银行小微风控审核助手。只依据用户给出的字段作答," "字段缺失时填 null,禁止推测、禁止引用外部知识。" "deviation_rate 按 |税务值-经营值|/税务值 计算,保留四位小数。" ) def build_messages(row: dict) -> list[dict]: payload = {k: row.get(k) for k in FIELDS} # 只传白名单字段,避免脏数据干扰 user = ( f"待核验字段(JSON): {json.dumps(payload, ensure_ascii=False)}\n" '输出格式: {"items":[{"field":"","tax_value":"","biz_value":"",' '"deviation_rate":0,"verdict":"consistent|suspicious|conflict","reason":""}]}' ) return [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user}]字段白名单是这套方案里最容易被忽略但最值钱的一步。直接把整行宽表丢给模型,模型会被几十个无关字段干扰,reason里开始出现"根据企业资产规模推测"这类没有依据的话。只送交叉验证需要的五个字段,输出质量会稳定很多。deviation_rate的计算公式写进 system prompt 而不是让模型自己选,是为了和规则引擎的口径保持一致,两边算出来必须能对上。
3.3 本地部署 DeepSeek 与批量跑批
数据不出行内是硬要求时,走本地部署路线。用 Ollama 拉起权重并暴露 OpenAI 兼容接口,业务代码只改base_url,其余不动。
# 拉取权重并启动本地推理服务 ollama pull deepseek-r1:7b ollama serve # 默认监听 127.0.0.1:11434,兼容 /v1/chat/completions export DEEPSEEK_BASE_URL=http://127.0.0.1:11434/v1本地部署要盯着显存。7B 量级在 FP16 下大约需要 16GB 显存,量化到 4bit 后可压到 6GB 上下,但抽取精度会下降,非结构化文本的字段召回率通常会掉几个百分点,上生产前要用同一批标注样本比一次。显存不够时的降级方案是用 CPU 推理配合更小的模型,代价是单条延迟从几百毫秒涨到数秒。
批量跑批的骨架如下,重点是限流、退避和结果落地格式。
import json, os, time from concurrent.futures import ThreadPoolExecutor from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")) def audit(row, retry=3): for i in range(retry): try: resp = client.chat.completions.create( model=os.getenv("DEEPSEEK_MODEL", "deepseek-chat"), temperature=0, response_format={"type": "json_object"}, messages=build_messages(row), timeout=30, ) return {"uscc": row["uscc"], "result": json.loads(resp.choices[0].message.content)} except Exception: time.sleep(2 ** i) # 指数退避,缓解限流与瞬时抖动 return {"uscc": row["uscc"], "result": None, "error": "max_retry"} with open("audit_result.jsonl", "w", encoding="utf-8") as f, ThreadPoolExecutor(max_workers=4) as pool: for item in pool.map(audit, rows): f.write(json.dumps(item, ensure_ascii=False) + "\n") # JSONL 便于下游直接入库max_workers不要凭感觉调大,先按 4 跑一轮看失败率和 P99 延迟,再往上加。失败记录必须落到文件里带error标记,不能静默丢弃——批量任务里最怕的就是跑完发现样本少了几百条却没报错。输出用 JSONL 而不是单个大 JSON,是为了追加写入和断点续跑,几百个主体跑批中断一次不用从头再来。
| 参数 | 建议值 | 说明 |
|---|---|---|
| temperature | 0 | 保证同输入同输出,便于审计复现 |
| response_format | json_object | 免去输出后处理,降低解析失败率 |
| max_tokens | 800~1500 | 按输出条目数估算,过小会截断 JSON |
| timeout | 30s | 本地部署可放宽到 60s |
| max_workers | 4 起步 | 配合重试观察限流阈值 |
提示:单个主体的输入控制在千字以内。一次性塞进整份授信材料会顶到上下文上限,模型对中间段落的字段召回明显变差,正确做法是按主体切片、按字段分组多次调用。
4. 交叉验证规则引擎:偏差判定、评分卡与阈值调参
模型输出的是差异说明,规则引擎输出的是可执行的授信建议。两者之间要靠一张规则表来衔接,规则表决定了哪些差异算"可疑",哪些算"冲突"。
4.1 规则表设计
规则表的核心是每条规则的判定条件可量化、分值可解释、命中后有明确处置动作。
| 规则ID | 比对项 | 判定条件 | 分值 | 命中处置 |
|---|---|---|---|---|
| R01 | 营收偏差 | 流水偏差率 > 30% | 35 | 要求补充流水说明 |
| R02 | 用工冲突 | 社保人数为 0 且年报从业人数 ≥ 5 | 25 | 转人工核实用工形式 |
| R03 | 客户集中 | 开票下游客户数 ≤ 3 | 15 | 关注单一客户依赖 |
| R04 | 合规降级 | 纳税信用等级较上期下调 | 15 | 提高利率或压缩额度 |
| R05 | 模型标记 | DeepSeek 输出 suspicious/conflict ≥ 2 条 | 10 | 触发人工复核 |
用 SQL 把偏差率一次算出来,规则判定放在下游应用层,这样口径调整时不用改数据链路。
-- 以对公账户贷方发生额净额为经营侧口径,税务侧取不含税销售额 WITH base AS ( SELECT t.uscc, t.tax_sales_amt, -- 税务不含税销售额(万元) b.bank_credit_amt, -- 流水贷方发生额净额(万元) t.invoice_cust_cnt, -- 开票下游客户数 s.social_ins_cnt, -- 社保参保人数 a.annual_emp_cnt -- 工商年报从业人数 FROM tax_declare t LEFT JOIN bank_flow b ON t.uscc = b.uscc AND t.period = b.period LEFT JOIN social_ins s ON t.uscc = s.uscc AND t.period = s.period LEFT JOIN annual_report a ON t.uscc = a.uscc AND t.period = a.period ) SELECT uscc, ROUND(ABS(tax_sales_amt - bank_credit_amt) / NULLIF(tax_sales_amt, 0), 4) AS flow_dev_rate, CASE WHEN social_ins_cnt = 0 AND annual_emp_cnt >= 5 THEN 1 ELSE 0 END AS social_conflict, CASE WHEN invoice_cust_cnt <= 3 THEN 1 ELSE 0 END AS concentration_flag FROM base;NULLIF用来防除零,税务销售额为 0 的主体(新设、停业)不能直接算偏差率,否则会得到 NULL 或无穷大,下游打分逻辑要单独处理。三张表都用LEFT JOIN并以税务侧为主表,是为了保证有税务申报但没流水的企业不被过滤掉——这类主体恰恰是最需要关注的。t.period = b.period这个条件强制期间对齐,如果一侧是月度、一侧是季度,必须先在 ETL 层做粒度归一,不要指望在 join 条件里做转换。
4.2 评分卡与权重分配
分值不是拍脑袋定的。初版可以按业务专家打分,跑半年有了一批坏样本之后,用逻辑回归的系数反推权重,把专家分替换掉。
WEIGHTS = { # 初版专家分,后续用逻辑回归系数校准 "flow_dev": 35, "social_conflict": 25, "concentration": 15, "tax_grade_drop": 15, "model_flag": 10, } THRESHOLDS = {"reject": 60, "manual": 40} # >=60 拒绝;40~60 人工;<40 通过 def score(row: dict) -> dict: total, hits = 0, [] if row.get("flow_dev_rate") is not None and row["flow_dev_rate"] > 0.30: total += WEIGHTS["flow_dev"]; hits.append("R01") if row.get("social_conflict"): total += WEIGHTS["social_conflict"]; hits.append("R02") if row.get("concentration_flag"): total += WEIGHTS["concentration"]; hits.append("R03") if row.get("tax_grade_drop"): total += WEIGHTS["tax_grade_drop"]; hits.append("R04") if row.get("model_suspicious_cnt", 0) >= 2: total += WEIGHTS["model_flag"]; hits.append("R05") decision = "reject" if total >= THRESHOLDS["reject"] else ( "manual" if total >= THRESHOLDS["manual"] else "pass") return {"score": total, "hits": hits, "decision": decision}阈值调整要先看通过率和坏账率的联动。把reject从 60 降到 50,通过率通常掉三到五个百分点,但如果坏账率没有同步改善,说明这条分界线切错了位置,问题出在权重而不是阈值。hits字段必须保留,事后追责和规则归因全靠它——只看总分无法解释为什么拒了这家企业。
4.3 命中结果排错:偏差率异常的五个常见原因
R01 命中率明显偏高时,先按下面顺序排查,不要急着改阈值。
一是含税口径没还原,用价税合计去比不含税销售额,偏差率天然在 13% 到 30% 之间浮动。二是集团资金池归集,母公司每日归集下属公司资金,贷方发生额被放大数倍。三是代开发票,企业通过税务机关代开,开票金额计入税务侧但回款走的是个人账户,流水里看不到。四是跨期确认,税务按开票时点确认、企业按收款时点记账,月末月初的几笔会造成单月偏差。五是补贴与退税,财政补贴和出口退税计入流水但不计入销售,属于合理差异,应该在规则里做白名单排除。
排查工具很朴素:把 R01 命中样本按偏差率分箱,看是集中在 30%~50% 这一档(多半是口径问题),还是散落在 200% 以上(多半是资金池或数据错位)。分布形态比平均值更能说明问题。
5. 交叉验证规则的验证与调优:回溯测试与差异归因
规则上线前必须做回溯测试:取过去 12 到 24 个月的历史申请样本,用当时的税务和经营数据重跑一遍打分,再对照这批客户后续的实际表现(逾期、展期、不良)。这一步的难点是数据要还原到申请时点,不能用现在的余额倒推,否则会引入未来信息,回溯结果虚高得没法看。
评估指标不要只用准确率。小微样本天然不均衡,通过率通常在 85% 以上,用准确率会得出"全都通过"这种毫无意义的结论。我一般看三个数:KS 衡量区分度,目标定在 0.3 以上;PSI 衡量线上线下分数分布是否漂移,超过 0.25 就要查规则是不是失效了;再就是按分数段看坏账率是否单调——如果 40~50 分段比 50~60 分段的坏账率还低,说明权重配反了。
模型输出的差异归因不要直接当分数用,把它转成特征更稳。具体做法是把每条reason文本做一次关键词映射,落到有限的归因标签上,比如"口径差异""关联交易""补贴退税""数据缺失",再统计每个标签在坏样本里的出现频次。频次显著高于均值的标签,可以单独提出来加一条规则,权重从原有的model_flag里拆出去。这样模型和规则就不是两套并行系统,而是一条链上的上下游。
灰度阶段的一个实用技巧是双跑:新规则打分后不做拦截,只记录decision字段,跟现有审批结论并排对比,观察两周。重点关注两类分歧——新规则拒而老流程通过的(看后续是否真出风险),以及新规则通过而老流程拒的(看是不是漏掉了抵押物等线下信息)。分歧样本里只要有 10% 到 20% 能在三个月内验证出风险,就说明新规则的增量信息是真实存在的,这时候再切流才有依据。
本文还有配套的精品资源,点击获取