干金融风控的朋友应该都清楚在信贷审批、额度测算这些业务场景里运营商账单数据一直是个很有分量的辅助判断依据。一个人每个月的通信消费金额、缴费是否准时、号码在网时长这些东西看似细碎却能从侧面反映出收入水平、居住稳定性、消费习惯甚至潜在的欺诈信号。我之前处理过一批来自海宇运营商的账单数据目标非常具体利用近3个月的平均账单金额把金融风控合规环节从用户上传纸质账单、人工一张张核验的老流程改造成系统自动拉取、自动计算、自动生成核验结果的新流程。这篇文章就把整个项目的来龙去脉、技术实现、踩坑记录完整复盘一遍适合正在做风控数据、数据仓库或者Python数据处理的朋友参考。先说明一下文中的海宇运营商是我们项目里的脱敏代称实际对接的可以是运营商本身也可以是持有相关资质的持牌数据服务商。账单数据结构我做了泛化处理但核心逻辑和工程思路完全一致你可以直接套用到自己项目里。1. 项目背景风控合规为什么盯上运营商账单1.1 运营商账单在金融风控里的独特价值在信贷场景里传统风控主要依赖央行征信报告、社保公积金缴纳记录、银行流水这些强数据。但问题在于这些数据不是每个人都有也不是每个用户都愿意第一时间提供。很多年轻用户、自由职业者、蓝领群体征信记录比较薄银行流水也不规整这时候运营商账单就成了很好的补充维度。运营商账单能看出哪些东西最直接的就是消费金额。一个人每月的手机话费套餐费、流量费、增值业务费加起来大概处于什么水平能粗略反映他的消费能力和支付意愿。比如一个用户每月通信账单稳定在200元以上且连续数月没有欠费停机通常说明他有相对稳定的收入来源和较好的信用习惯。反过来如果账单金额忽高忽低、频繁出现停机缴费记录甚至号码使用时间极短那就要警惕是不是新办号码用于短期骗贷。除了单月金额账单里还包含缴费渠道、缴费时间、套餐变更记录、通话详单规模等信息。这些字段在反欺诈环节很有价值。举个例子一个用户申请贷款时填的居住地址和工作地址都在A城市但他的手机号码归属地在B城市且近3个月主要在B城市产生基站数据那这个信息就要进入风控规则里做交叉验证。回到我们项目本身近3个月平均账单这个指标之所以被风控合规部门重点提出是因为它比单月账单更稳定。单月数据可能受偶然因素影响比如某个月用户办理了新手机合约导致账单激增或者某个月出差到外地产生了大量漫游费用。取3个月平均值可以平滑掉这些短期波动让特征更接近用户的真实消费水平。1.2 海宇运营商数据源解读与接入方式海宇运营商这个数据源在我们项目里扮演的是账单数据提供方的角色。具体来说用户在我们的金融App上发起贷款申请后经过用户本人的授权同意我们通过合规接口向海宇运营商发起账单查询请求拿到用户在近3个月内的账单明细数据。这里有个关键点一定要有用户的明确授权。我们当时的做法是在贷款申请流程中增加单独的一步数据查询授权书用户需要勾选同意并完成人脸核验系统才会向运营商发起数据请求。这个授权记录会作为合规留痕保存后续稽核时可以随时调取。接口返回的数据结构大概是这样的用户基础信息脱敏后的手机号码中间4位打码、实名认证姓名、证件号加密传输、号码归属地账单明细列表每个自然月的账单总额、套餐月费、语音通话费、流量费、增值业务费、缴费金额、缴费时间、当月是否停机、停机次数在网信息号码入网时间、在网时长、当前套餐档位字段不算特别复杂但坑在于不同运营商的字段命名和取值格式不统一。哪怕我们只对接海宇运营商这一个数据源不同批次返回的数据里也可能出现金额单位不一致、日期格式混乱、空值标记方式不同等问题。这些细节后面我会在数据清洗部分展开讲。接入方式上海宇运营商提供的是HTTP接口基于HTTPS协议请求需要带签名参数。我们当时的方案是业务后端在用户授权后生成一个一次性token然后数据管道服务拿着这个token去调用海宇的查询接口。每笔请求都有独立的消息ID方便追踪全链路日志。2. 数据工程整体链路设计从接口到风控特征2.1 一条完整的账单数据管道包含什么很多刚开始做数据处理的朋友容易犯一个错误拿到数据就开始写计算逻辑忽略了管道本身的工程化设计。实际上一个稳定的数据管道应该包含采集、清洗、计算、存储、输出五个环节每个环节都要有独立的日志、异常处理和监控告警。我们这条管道的流程是触发机制用户完成授权后消息队列MQ里产生一条消息包含用户ID和授权token采集层Python脚本订阅MQ消息调用海宇运营商接口拉取账单数据原始JSON直接落盘到对象存储作为备份清洗层对原始JSON做格式校验、字段映射、缺失值处理、数据标准化计算层基于清洗后的数据计算近3个月平均账单、缴费稳定性、在网时长等特征存储层特征结果写入MySQL原始数据与清洗结果写入Hive/OSS用于离线分析输出层将计算结果同步给风控决策引擎同时生成一份合规核验报告供人工审核岗查看为什么坚持把原始JSON落盘这是很重要的经验。接口方返回的数据结构有可能会变一旦线上计算结果出现异常我们可以回溯原始数据排查问题不需要重新找接口方要日志。而且合规审计时监管如果问这个特征是怎么算出来的原始数据就是最有力的证据。2.2 技术选型解析为什么用Python搭这条管道整个项目的数据处理部分我用的是Python这应该算是最主流也最稳妥的选择。原因很实在第一生态成熟。pandas、numpy这两个库在数据处理上几乎是事实标准写起来比Java、Go要快很多。我们的清洗逻辑里有大量的条件判断、字段映射、分组聚合用pandas的DataFrame操作非常顺手一个简单的groupby加agg就能完成复杂的聚合计算。第二接口调用方便。Python的requests库、httpx库封装得很好处理JSON、签名、超时重试都很灵活。尤其面对运营商接口这种需要自定义签名逻辑的场景Python可以很快迭代调试。第三团队协作成本低。风控团队和数据分析团队日常就在用Python写策略和分析脚本我们做数据管道的代码放在同一个代码仓库里后续维护和功能迭代很顺畅不需要跨语言沟通。当然Python也不是没有缺点比如性能上限不如C/Go但在我们这种每天几万笔请求、每笔请求只处理几个月账单数据的场景下性能完全不是瓶颈。真正需要关注的是代码的工程规范性而不是语言本身的性能。项目里用到的库主要有requests调用海宇运营商接口pandas / numpy数据清洗与特征计算pydantic定义返回数据模型做数据校验SQLAlchemy写MySQL特征结果APScheduler / cron定时跑批与重试任务redis做分布式锁和幂等控制2.3 近3个月平均账单的口径定义与计算窗口任何一个数据指标如果没有明确的计算口径上线后一定会出问题。近3个月平均账单这个指标看起来简单但是细究起来有很多口径问题第一个问题是近3个月怎么定义。是以当前日期往前推3个自然月比如今天是3月5日就取12月、1月、2月的账单还是以账单出账周期为准实际操作中我们采用了截至当前已出账的最近3个完整自然月这个口径并且在实际计算时如果用户有某个月没有账单记录比如新入网不满3个月则按实际有账月份数折算平均。第二个问题是平均金额的计算方式。是三个月的账单总额除以3还是除以有账单的月份数如果用户某个月账单为0算不算这个月我们的口径是用户近3个月有账单记录的月份中每月账单金额的算术平均值。也就是说只有在用户这3个月的账单都正常出账的情况下才能用总额除以3如果有一个月缺失就按两个月的总额除以2来算。这样处理更公平也避免因为数据缺失导致平均账单被低估。第三个问题是是否剔除异常月份。比如用户某个月买了一部合约机账单金额从平时的100元暴涨到5000元这种极端值如果不处理会直接拉高平均账单导致风控模型误判。我们当时加了一个规则单月账单金额超过该用户近3个月中位数的3倍时将该月标记为异常值不参与平均值计算。这个规则有点像统计学里的MAD中位数绝对偏差方法但更简单直观。这里贴一段核心计算逻辑的伪代码后面我在第3章会给出完整实现计算步骤 1. 获取用户在近3个自然月内的账单记录 2. 过滤掉状态为欠费停机且金额为0的无效记录 3. 计算账单金额的中位数标记超过3倍中位数的异常月 4. 对剩余有效月份的账单金额求算术平均 5. 如果有效月份数小于2则该特征置为NULL不参与模型评分这个口径确认好之后我们和风控策略团队、合规团队一起做了评审确保三方对指标的理解完全一致才进入开发。3. 核心实现从原始账单到风控特征3.1 账单数据的标准化清洗海宇运营商接口返回的原始JSON基本上可以直接用但字段值是原始状态必须清洗标准化后才能参与计算。我们踩过的坑主要集中在几类问题上。金额字段格式不统一。有时候返回的是字符串128.00有时候是整数128有时候是科学计数法。最稳妥的做法是全部转成Decimal类型避免浮点数精度问题。比如计算平均账单金额如果直接用float累加再除可能出现0.10.2不等于0.3这种尴尬情况虽然金额只有两位小数误差不大但在金融场景里风控特征值喂给模型后误差会被放大还是应该用Decimal。日期字段的时区问题。运营商返回的账单周期字段是2025-01这种格式但缴费时间可能是完整的日期时间字符串而且接口方使用的时区跟我们业务库不一致。处理方式统一是所有时间字段解析后转为东八区再格式化为标准字符串。空值和缺失值处理。不同字段的空值标记不一样金额字段可能返回null也可能返回0还有可能是空字符串。我们在pydantic模型里定义了校验规则金额字段必须大于等于0字符串字段长度不能超过指定值。如果不合法就直接抛弃这条记录并记录异常原因。文本字段的脏数据也值得注意。号码归属地返回的值可能是浙江省杭州市也可能是杭州我们需要维护一套映射表做归一化。套餐名称字段就比较乱了同一档位在不同时期的名字可能不一样不过这个字段我们暂时没参与特征计算只做了存储后续如果需要再做解析。清洗层实现的核心代码我用pycharm写了一个数据清洗模块核心思路是这样的使用pydantic定义账单数据的数据模型在模型字段中增加validator方法做校验和标准化。模型的每个字段都会经过类型检查和值域校验这样原始数据一旦有问题就能在最早的时间点暴露出来而不是等到算特征时才报错。清洗逻辑里比较重要的是幂等性设计。因为我们有重试机制同一笔用户的账单数据可能被拉取多次所以清洗模块要保证无论来几次数据清洗结果是一致的。我们的做法是给每一笔计算任务生成一个唯一的task_id清洗结果表里以task_id和用户ID作为联合主键重复写入时执行upsert更新不会产生重复数据。3.2 平均账单金额的计算实现计算层的核心代码我直接贴出来这段代码经过线上验证逻辑清晰各位可以直接参考。import pandas as pd import numpy as np from decimal import Decimal from typing import List, Dict, Optional def calculate_avg_bill( bill_records: List[Dict[str, object]], current_date: str None, ) - Dict[str, Optional[Decimal]]: 计算用户的近3个月平均账单金额 口径说明 1. 取当前日期往前推的最近3个完整自然月已出账月份 2. 过滤欠费停机且金额为0的记录 3. 剔除金额超过中位数3倍的异常月份 4. 对剩余有效月份求算术平均 5. 有效月份小于2个月时特征置为None :param bill_records: 海宇运营商返回的账单明细列表 :param current_date: 当前日期字符串格式为 YYYY-MM-DD :return: 平均账单金额的Decimal值以及辅助诊断信息 if not bill_records: return { avg_bill: None, valid_month_count: 0, anomaly_month_count: 0, raw_month_count: 0, } # 获取近3个自然月的月份列表 if current_date is None: current_date pd.Timestamp.now().strftime(%Y-%m-%d) date_ts pd.Timestamp(current_date) # 获取当月的起始日期往前推3个月 three_months_ago_start (date_ts - pd.DateOffset(months3)).replace(day1) recent_months pd.date_range( startthree_months_ago_start, enddate_ts, freqMS ) recent_month_str set(month.strftime(%Y-%m) for month in recent_months) # 构建DataFrame df pd.DataFrame(bill_records) # 确保账单月份字段存在且格式正确 if bill_month not in df.columns: raise ValueError(bill_records中缺少bill_month字段) df[bill_month] df[bill_month].astype(str) # 1. 筛选近3个月的账单 df df[df[bill_month].isin(recent_month_str)] # 2. 过滤欠费停机且金额为0的无效记录 if bill_status in df.columns: df df[~((df[bill_status] 欠费停机) (df[bill_amount] 0))] if df.empty: return { avg_bill: None, valid_month_count: 0, anomaly_month_count: 0, raw_month_count: len(bill_records), } # 3. 剔除异常月份金额超过中位数3倍 df[bill_amount] pd.to_numeric(df[bill_amount], errorscoerce) df df.dropna(subset[bill_amount]) if df.empty: return { avg_bill: None, valid_month_count: 0, anomaly_month_count: 0, raw_month_count: len(bill_records), } median_amount df[bill_amount].median() threshold median_amount * 3 anomaly_mask df[bill_amount] threshold anomaly_count int(anomaly_mask.sum()) valid_months_df df[~anomaly_mask] valid_month_count valid_months_df.shape[0] # 4. 有效月份少于2不参与计算 if valid_month_count 2: return { avg_bill: None, valid_month_count: valid_month_count, anomaly_month_count: anomaly_count, raw_month_count: len(df), } # 5. 计算算术平均值使用Decimal避免精度问题 avg_bill Decimal(str(valid_months_df[bill_amount].mean())).quantize(Decimal(0.01)) return { avg_bill: avg_bill, valid_month_count: valid_month_count, anomaly_month_count: anomaly_count, raw_month_count: len(df), }这段代码里有几个值得细说的点。pd.date_range配合replace(day1)可以准确地生成最近3个自然月的月份列表。这里有个坑如果直接用date_ts - pd.DateOffset(months3)得到的是3个月前的这个日期比如3月5日往前推90天左右而不是2月1日。我们想要的是从2月1日到4月1日之间出账的账单所以必须把起始日期修正到月份的第一天。中位数和3倍阈值的方案相对简单但很实用。实际操作中我还加了一个下限保护当阈值小于30元时不剔除任何月份因为如果用户的基础套餐非常低比如月费才19元中位数的3倍可能只有57元正常一个月的流量超出费就超了不能因为阈值设置过死误杀正常月份。valid_month_count 2这个条件可以通过写清楚逻辑来避免平均账单偏高/偏低的错误判断。如果用户只出了1个月的账单说明他入网时间很短本身就是一个风险信号这时候不但不能给他算平均账单反而应该在风控规则里加分。3.3 合规报告与决策辅助输出计算完平均账单金额之后这个特征值并不会直接决定用户能不能通过贷款审批而是作为众多风控特征之一输入到决策引擎里。同时我们要生成一份面向合规审核岗的核验报告让审核人员能够快速理解这笔账单数据是否可信、是否存在异常。合规报告我们当时输出成JSON格式同时渲染成PDF便于留档。里面包含几个核心板块用户脱敏信息手机号打码、姓名脱敏、证件号脱敏账单数据摘要近3个月各月的账单金额、缴费状态、异常标记计算过程说明用了哪些月份、剔除了哪些异常月、计算依据特征结果平均账单金额、有效月份数、数据质量评分授权信息授权时间、授权方式、授权编号这个报告最大的价值在于可解释性。金融风控合规要求决策需要有据可查不能因为系统说拒绝就拒绝用户。有了合规报告审核人员可以在3秒内判断出系统计算逻辑是否合理用户申诉时也能快速核查。我们用Python的jinja2模板加weasyprint库生成PDF报告模板里定义好表格样式后端只要填充数据就能自动渲染出格式统一的文件。这个方案在Linux服务器上部署很方便不需要额外的Office软件。4. 合规体验优化的落地效果4.1 从用户交材料到系统拉数据的体验转变用户侧体验的提升是这个项目最直观的成果。原来的流程是用户申请贷款如果风控模型判断需要补充运营商账单核验用户就要自己登录运营商App查找和下载近几个月的电子账单然后上传到我们的平台。听起来不难但实际操作中经常会遇到几个问题用户找不到账单下载入口。运营商App的功能层级很复杂不是每个人都能熟练找到电子发票或账单查询在哪客服电话咨询高峰期又打不进去。我们当时做过统计用户从收到补件通知到成功上传账单平均耗时超过1天有接近15%的用户在这个过程中放弃申请。上传的账单格式五花八门。有截图的、有PDF的、有网页转图片的审核人员需要逐一打开查看效率低且容易出错。还有用户会把敏感信息截掉或者上传了经过PS修改的账单这给合规审核埋了很大的雷。新流程改成了用户授权即拉取。用户在申请页面上点击授权我们系统自动向海宇运营商发起查询数据直接进入数据管道全程不需要用户手动操作。从用户授权到风控特征计算完成平均耗时不到10秒基本上用户还没从申请页面离开风控结果就出来了。这个体验上的提升用户层面感受最明显审批时长从原来的平均2天缩短到最快10分钟。而且因为运营商数据是直接接口拉取的从源头上杜绝了修改账单的可能合规性也更强。4.2 指标上线后的效果评估指标上线不是终点关键要看它对业务的实际影响。我们跟踪了几个核心指标审批耗时大幅降低。原来需要人工核验账单的用户群体现在大部分实现了自动化。人工审核工单量下降了60%以上审核人员可以集中精力处理真正有争议的案例。用户放弃率明显下降。因为流程变短了用户在贷款申请过程中的流失率从原来的12%降到了4%左右。这一点业务团队感受最深直接反映在进件量和最终促成率上。当然了还要关注风控的质量。我们对比了用运营商账单特征前后整体逾期率基本没有上升甚至在部分客群上略有下降。这说明新增的特征并没有放松风控标准而是更准确地识别了好用户和差用户。在监控方面我们搭建了一个简单的看板每天追踪几个关键指标平均账单金额的分布、有效月份数不足2个月的占比、异常月剔除率、接口调用成功率、数据质量异常率。任何一个指标出现异常波动都会触发告警。特别是接口调用成功率这个指标直接反映了数据管道是否稳定。有一次海宇运营商的接口在晚间升级导致半小时内调用失败率超过30%我们的告警系统第一时间发现了问题及时切到了备用数据源避免了线上业务受影响。5. 实操中踩过的坑与排查方法5.1 账单缺失与跨月数据延迟问题运营商的账单出账时间不是统一的有的用户是每月1日出上月账单有的可能要到5号甚至10号。这意味着我们在月初查近3个月数据时上个月的账单可能还没出账。如果按照有账单就算的逻辑月初计算的用户平均账单会用到旧月份数据口径就偏了。我们的处理方式是在计算前先拉取账单的出账状态字段只有状态为已出账的月份才纳入计算。如果用户已经有2个月以上的已出账账单即使当月账单还未出账也能正常计算平均账单。如果不足2个月就需要等待账单出账后再触发一次重试计算。还有一个问题账单缺失不一定是运营商的问题也可能是接口查询失败、用户号码状态异常、授权过期。这就需要排查链路里每一层的日志。我们给每个任务都分配了唯一的trace_id从MQ消息产生到最终特征入库全程记录日志。排查问题时只要拿着trace_id在日志平台搜索就能定位到具体是哪个环节出了问题。5.2 接口限流与批量重试机制海宇运营商的接口有很严格的QPS限制单账号并发请求数不能超过5否则会直接返回限流错误码。我们在批量处理存量用户时比如上线初期要预计算所有存量的授权用户单线程跑又太慢必须设计一套合理的并发控制方案。最终采用的是信号量加指数退避重试的方案。用一个线程池控制最大并发数为4留一个余量每个请求如果遇到限流错误就退避一段时间后重试退避时间按1秒、2秒、5秒、10秒递增最大不超过30秒。实测下来批量跑100万笔存量数据加上限流控制大约需要7个小时跑完完全可以接受。重试机制里有一个容易踩的坑接口超时时间不能设置得太短。运营商接口在高峰期响应可能会超过10秒如果我们的超时时间设置成5秒会导致大量请求被主动中断反而加重了服务端的压力。我们把连接超时设成3秒、读取超时设成30秒实际运行中几乎很少出现超时问题。5.3 隐私合规与数据安全细节最后说说合规和数据安全这块在金融场景里是红线碰都不能碰。用户授权记录必须完整保存。谁在什么时间、通过哪种方式授权我们查询了哪些数据这些信息都要有完整的台账而且台账不能随便改最好用独立的只读存储。数据字段的展示要脱敏。运营商的原始账单数据里有手机号、姓名、证件号这些在系统里必须加密存储。我们用的是AES-256加密密钥由独立的KMS管理日常服务运行时只能动态解密不能落盘明文。数据的使用范围要最小化。从海宇运营商拉下来的原始账单JSON只有数据管道服务和风控决策引擎有权限访问其他任何系统都拿不到。即使是给审核人员看的合规报告敏感字段也做了打码处理比如手机号中间4位显示为星号。日志里避免出现敏感信息。调用接口时我们只记录脱敏后的用户标识比如user_id后端分配的匿名ID而不是直接把手机号打进日志。因为日志系统通常是明文存储、长期保留万一日志泄露手机号和身份证号就会一起泄露。还有一点务必要提醒和运营商的合同里通常会约定数据用途限制。我们的合同明确写了数据只能用于贷款申请的风控评估不能用于营销、不能转让给第三方。在使用数据做模型训练时也要遵守合同约定训练用的特征数据同样需要脱敏和加密。结尾几点经验性的体会这次项目做完我自己最深的一个体会是数据工程这件事真正难的地方往往不在算法和模型而在工程细节。一个近3个月平均账单看起来就是求个平均值实际上牵扯到口径定义、异常处理、缺失值应对、幂等重试、合规留痕、监控告警这么多环节。任何一个环节考虑不到位线上跑一段时间就会冒出新问题。另外一个感受是数据特征上线前的口径评审非常重要。我当时和风控策略、合规团队开了三次评审会把近3个月怎么定义金额异常怎么处理有效月份不足怎么返回这些问题反复确认清楚开发阶段几乎没有返工。相比之下我们之前做过的一个项目因为口径没对齐上线两周后才发现不同团队的理解不一样不得不推翻重算教训很深刻。如果你们也在做类似的运营商数据接入项目我建议先把计算口径文档写清楚拿到业务方和合规方签字确认再动手写代码。数据管道搭建上优先保证幂等性和可追溯性其他的性能优化可以后面再慢慢做。祝各位早日跑通自己的数据链路。