先还原一个让不少玩家血压升高的场景排位对局打到关键团战手机屏幕顶端突然弹出来电直接盖住操作界面刚挂掉另一个陌生号码又打进来。局还没打完语音频道里已经有人开始报“船新阵容”随后飘过来一句吐槽打着打着打过来诈骗电话笑死船新阵容电羊未登场罗隐。这句话放在游戏社区里是个段子意思是“对局正激烈骚扰电话突然打断节奏而队友讨论的新阵容里还有个关键角色没上场”。但在做在线业务的技术人眼里这个场景暴露的是一个非常典型的实时风控问题骚扰电话和诈骗电话对线上业务的伤害远比表面看起来严重。玩家层面的损失是操作被打断、语音卡顿、心态爆炸业务层面的损失则是对局掉线率上升、用户投诉增加、在线时长下降。更危险的是如果接下来的电话话术是“您的账号存在异常”“装备交易需要验证”那它就不再只是骚扰而是可能演变成盗号、代充诈骗等安全事故的入口。本文不打算停留在聊天吐槽层面。我会把“游戏对局中突遭诈骗电话”这件事拆成一个可落地技术问题如何构建一套能实时识别和拦截高风险来电的反诈识别系统。内容包括核心概念、整体架构、最小可运行的 Python 工程、规则引擎与机器学习模型的组合方式、线上评估方法以及生产落地时最容易踩的坑。如果你想理解风控系统是怎么搭建的或者正在为游戏、直播、社交平台接入反欺诈能力这篇文章会给你一条清晰的技术路线。1. 游戏对局遇到诈骗电话问题到底出在哪很多技术人会下意识觉得诈骗电话是运营商和手机安全软件的事游戏平台为什么要关心但把视角拉回业务现场问题就清楚了。第一个问题是实时性。玩家在游戏内时业务方通常拿不到“此刻这个号码正在呼叫玩家手机”的实时事件。等到玩家打完电话、意识被骗再发起投诉盗号或资金损失可能已经发生。反诈识别的价值窗口往往只有几分钟甚至几十秒过了这个窗口拦截动作再准确也没有意义。第二个问题是干扰。即使没有诈骗成功高频呼叫也会直接影响游戏体验。假设某次大版本更新后诈骗团伙批量使用语音群呼骚扰玩家玩家在关键时刻被迫接听第一反应不是骂诈骗分子而是觉得“这游戏怎么这么多骚扰电话”“体验变差了”。用户不会把账算到黑产头上只会算到产品头上。第三个问题是关联风险。诈骗电话从来不是孤立事件。黑产通常先确认活跃玩家再通过话术诱导开启屏幕共享、扫码登录、语音操作验证码最终完成账号盗取或装备虚拟财产转移。如果游戏平台能提前识别出这批号码的风险等级在客服申诉、交易校验、设备风控等环节联动整个账号安全水位都会提升。所以更稳妥的判断是诈骗电话识别不是一个纯电话业务问题而是在线业务风控体系的前置触角。它要解决的不只是“这个电话是不是骗子”还包括“风险事件发生时业务系统能不能快速响应”。类似的系统并不是只有大厂才能做。市面上已有不少运营商级和互联网公司级的号码风险接口自己也可以基于通话记录、号码画像、行为频次搭建一套精简版反诈识别流程。这篇文章后面的示例就是要用最小的成本把这条链路跑通。2. 诈骗电话识别为什么是一个技术活如果只是做个黑名单拦截那叫名单管理不叫风控系统。诈骗电话识别的难点在于对手是动态对抗的。黑产手里有成本极低的号码池。过去是固定电话、手机号现在还有虚拟运营商号段、物联网号卡、网络电话、国际来显。一个号码被标记拉黑后面还有成千上万个新号。单纯的黑名单库永远追不上号码更换速度。这就是为什么需要“特征识别”。同样是陌生号码正常外卖电话、快递电话和诈骗电话在行为模式上有明显差异。诈骗电话通常有这些特征短时间大量呼出、平均通话时长很短、被叫号码覆盖范围分散、呼叫时段集中在夜间或午休、频繁更换主叫号码、同一批次号码存在关联关系。靠人工总结这些规律就是规则引擎让模型从历史标注数据中自动学习这些规律就是机器学习方案。两者不是替代关系而是组合关系。先看一张传统识别和智能识别的对比对比维度传统黑名单方式规则引擎机器学习模型更新速度滞后依赖人工新增可快速调整阈值依赖定期重训可解释性强直接看名单强命中哪条规则清晰较弱概率输出需要分析对抗能力差改号即失效中需持续维护规则较强能捕捉复杂关联实施成本低低到中中到高典型定位全局第一道闸兜底与可解释策略深度风险识别在实际工程里三者是叠加的。号码先过黑名单再进规则引擎最后让模型打分。任一环节给出高风险结论都可以触发拦截、标记或二次确认。这里还要强调一个容易被误解的地方识别诈骗电话不等于读取通话内容。在绝大多数合规场景下系统使用的是通话信令、话单记录、号码归属、频次统计等元数据而不是通话录音和聊天内容。文本后面讲到的所有特征都只使用脱敏后的统计信息与行为标记。3. 整体架构从呼叫数据到拦截动作的闭环一个完整的反诈识别系统可以从下往上拆成五层。第一层是接入层负责接收呼叫相关数据。常见来源包括运营商信令网关、CDR 话单、软交换平台日志、App 通话上报接口。接入层要做的事情就是统一数据格式把不同来源的字段映射成标准字段。第二层是计算层分实时和离线两条链路。实时链路处理在线呼叫事件比如收到“号码A呼叫号码B”的实时信令离线链路负责周期性加工号码画像、更新名单、重训模型。对中小团队来说实时链路可以先用一个 HTTP 接口承接离线链路用定时任务跑。第三层是特征层负责把原始数据变成风险特征。比如这个号码近 24 小时呼出多少次被叫号码是否分散是否命中高风险号段是否在夜间高频呼叫。特征层强调规范化训练和推断必须使用同一套特征逻辑。第四层是决策层按照“黑名单 - 规则引擎 - 机器学习模型”的顺序做综合判断。决策层最后输出一个状态放行pass、人工复核review、拦截block以及风险分数和命中原因。第五层是执行层负责把决策结果落地。常见动作包括运营商侧挂断、来电彩印标记、App 弹出风险提示、客服工单生成、设备风控联动。执行动作要在“保障用户体验”和“控制风险”之间取平衡不能只要拦截量。完整的数据闭环是关键。每次识别后系统需要记录结果并允许运营人员对拦截和放行样本进行标注。标注数据回流到训练集定期更新模型。这个闭环做不转前几个月的规则和模型就会慢慢失效。为了快速让读者理解链路下面先用一个最小工程把“接入层、特征层、决策层、执行层”跑通。计算层的分布式部分先不展开本地单机服务已经足够演示核心思路。4. 环境准备一个可运行的最小工程长什么样本项目使用 Python 3.9依赖较少便于在本地或测试服务器部署。需要安装的库如下pip install fastapi uvicorn pandas scikit-learn joblib项目目录建议这样组织phone-risk-demo/ ├── src/ │ ├── api.py # FastAPI 接口 │ ├── features.py # 特征提取 │ ├── rule_engine.py # 规则引擎 │ └── train_model.py # 训练并保存模型 ├── models/ # 模型存放目录 └── data/ └── call_records_labeled.csv # 带标注的历史数据首先约定一条呼叫记录的数据格式。下面是一个标准 JSON 示例{ phone: 17012345678, callee: 13900001111, start_time: 2025-05-01 23:30:00, duration: 3, call_type: normal, call_count_24h: 35, blacklist_score: 1 }各字段含义如下字段类型说明phonestring主叫号码calleestring被叫号码start_timestring呼叫开始时间durationint通话时长单位秒call_typestringnormal 表示普通呼叫international 表示国际来电voip 表示网络电话call_count_24hint该号码近 24 小时呼出次数blacklist_scoreint黑名单命中程度0 表示未命中数值越高风险越大这些字段只是演示用途。在生产环境中任何用户数据采集都必须有合法授权并且要做脱敏处理。实际项目中可以接入运营商合规提供的信令数据或者使用企业已授权的呼叫日志。5. 特征提取与规则引擎先把兜底阵容搭起来我建议任何风控系统都先把规则引擎做好再上机器学习。原因是规则引擎可解释性极强业务和风控同学都可以快速理解某个号码为什么被拦截因为命中了几条规则、累计了多少分。规则引擎也可以作为模型上线前的兜底阵容即使后面模型失效它还能挡住一部分明显风险。先编写特征提取模块把原始数据转成模型和规则都能用的标准化特征。# 文件路径src/features.py from datetime import datetime def extract_call_features(call_record: dict) - dict: 根据一条呼叫记录构造风险特征。 call_record 字段包括 - phone: 主叫号码 - callee: 被叫号码 - start_time: 呼叫开始时间字符串格式如 2025-05-01 23:30:00 - duration: 通话时长秒 - call_type: normal / international / voip - call_count_24h: 近24小时呼出次数 - blacklist_score: 黑名单命中程度 start datetime.fromisoformat(call_record[start_time]) features { hour: start.hour, is_night: int(start.hour 22 or start.hour 6), duration: call_record.get(duration, 0), is_international: int(call_record.get(call_type) international), call_count_24h: call_record.get(call_count_24h, 1), blacklist_hit: int(call_record.get(blacklist_score, 0) 0), prefix_risk: int(call_record.get(phone, )[:3] in {400, 952, 170}), } return features然后编写规则引擎。这里采用打分制每个风险特征累加一个分数最后根据总分判断状态。# 文件路径src/rule_engine.py def rule_scan(features: dict) - dict: risk_score 0 reasons [] if features[is_night]: risk_score 20 reasons.append(夜间高频呼叫) if features[duration] 5 and features[call_count_24h] 30: risk_score 40 reasons.append(短通话高频呼叫) if features[is_international]: risk_score 15 reasons.append(国际来电) if features[blacklist_hit]: risk_score 50 reasons.append(命中黑名单) if features[prefix_risk]: risk_score 10 reasons.append(高风险号段) if risk_score 60: state block elif risk_score 30: state review else: state pass return { risk_score: risk_score, state: state, reasons: reasons, }规则引擎设计有三个要点。第一分数阈值要来自数据分布不要拍脑袋。可以拉一批历史标注数据把所有样本跑一遍规则观察高风险样本和正常样本的分数分布再选择能拉开差距的阈值。第二每条规则的权重不是一成不变。比如黑名单命中往往可信度最高可以给较高权重号段风险只是弱提示权重不宜过大。如果某个权重在线上导致误拦激增就要及时调低。第三规则引擎必须保留命中日志。每一单被拦截都要能查到是哪条规则命中的。否则后续优化无从下手。6. 机器学习升级用历史标注数据做风险概率预测规则引擎适合快速兜底但面对黑产不断变化的策略会显得僵硬。此时可以用历史标注数据训练一个二分类模型输入特征输出该号码是诈骗电话的概率。这里采用逻辑回归主要考虑两点一是可解释性较好权重能对应到具体特征二是工程成本低训练快线上推断延迟低。数据集在真实项目中一般来自历史人工标注、用户投诉结果、运营商协作标记等渠道。# 文件路径src/train_model.py import pandas as pd from sklearn.model_selection import train_test_split from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler from sklearn.metrics import classification_report import joblib # 特征列必须与 features.py 中输出的特征保持一致 feature_cols [ hour, is_night, duration, is_international, call_count_24h, blacklist_hit, prefix_risk, ] def train_model(df: pd.DataFrame, model_path: str models/phone_risk_model.joblib): X df[feature_cols] y df[target] # 1 表示诈骗电话0 表示正常电话 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, random_state42, stratifyy ) model Pipeline([ (scaler, StandardScaler()), (lr, LogisticRegression(class_weightbalanced, max_iter1000)), ]) model.fit(X_train, y_train) y_pred model.predict(X_test) print(classification_report(y_test, y_pred)) joblib.dump(model, model_path) print(model saved to, model_path) return model if __name__ __main__: # 请准备一份带 target 标注的历史话单数据 # df pd.read_csv(data/call_records_labeled.csv) # train_model(df) pass这个训练脚本虽然简单但有几处需要注意。第一特征一致性。训练时用哪些字段线上推断就必须用同样顺序、同样口径的特征。举例来说训练时is_night用的是“22 点到次日 6 点”线上特征抽取函数必须保持一致否则模型效果会急剧退化。第二样本不平衡问题。正常电话数量通常远大于诈骗电话如果直接用原始比例训练模型会倾向于把所有样本都判为正常。这里使用了class_weightbalanced但更完整的做法是过采样、欠采样或使用更复杂模型并在评估时关注召回率与精确率而不是只看准确率。第三模型校准。逻辑回归输出的概率并不一定等于真实风险概率线上做阈值判断时不能机械地认为0.5就是最佳切分点。建议在验证集上绘制精确率/召回率曲线选择一个符合业务容忍度的阈值比如希望高风险拦截更激进可以把阈值降到0.4希望减少误拦可以把阈值提升到0.8。生产环境更推荐给模型做定期重训频率可以是每天或每周。黑产策略会变训练数据的时间窗口越近模型对当前风险的捕捉能力越强。7. 接口集成与结果验证业务侧如何接入风控服务模型训练完成后用一个 FastAPI 服务把特征提取、规则引擎、模型推断串起来。这个服务对外只暴露一个风险查询接口业务侧传入呼叫记录服务返回风险状态。# 文件路径src/api.py from fastapi import FastAPI from pydantic import BaseModel import pandas as pd import joblib from features import extract_call_features from rule_engine import rule_scan app FastAPI() # 生产环境建议在启动时从模型仓库加载并做好新旧版本切换 model joblib.load(models/phone_risk_model.joblib) feature_cols [ hour, is_night, duration, is_international, call_count_24h, blacklist_hit, prefix_risk, ] class CallRecord(BaseModel): phone: str callee: str start_time: str duration: int call_type: str normal call_count_24h: int 1 blacklist_score: int 0 app.post(/risk/check) def check_call(record: CallRecord): features extract_call_features(record.dict()) rule_result rule_scan(features) X pd.DataFrame([features])[feature_cols] ml_score model.predict_proba(X)[0][1] if ml_score 0.8: rule_result[state] block rule_result[reasons].append(高分模型判定) return { phone: record.phone, state: rule_result[state], risk_score: rule_result[risk_score], ml_score: round(ml_score, 4), reasons: rule_result[reasons], }启动服务uvicorn src.api:app --host 0.0.0.0 --port 8000用 curl 发一条测试请求curl -X POST http://127.0.0.1:8000/risk/check \ -H Content-Type: application/json \ -d { phone: 17012345678, callee: 13900001111, start_time: 2025-05-01 23:30:00, duration: 3, call_type: normal, call_count_24h: 35, blacklist_score: 1 }预期返回结果{ phone: 17012345678, state: block, risk_score: 70, ml_score: 0.9364, reasons: [ 夜间高频呼叫, 短通话高频呼叫, 命中黑名单, 高分模型判定 ] }判断服务是否正常工作可以从三个层面看第一接口能返回结构化 JSON说明链路已经接通。第二观察精确率和误拦情况这个请求被判定为 block是因为同时命中夜间呼叫、短通话高频、黑名单多条规则模型概率也较高结果符合预期。第三通过日志观察响应耗时。如果后续并发上来单个请求处理时间超过业务容忍上限就需要加缓存、前置名单层或者把特征计算下沉到流计算框架。在高并发场景下更推荐的接入方式是先走本机缓存的黑白名单再走规则引擎最后才调用模型。这样可以把大部分正常号码挡在第一层避免所有请求都进入模型推断。8. 线上效果好坏的评估方法与常见问题排查风控系统上线后不能只盯拦截量。拦截量高可能是模型激进导致的误拦反而伤害正常用户。建议建立一套围绕“风险识别准确度”和“用户体验影响”的评估体系。先看核心指标指标计算方式说明精确率被拦截号码中真正为风险号码的比例越高代表误拦越少召回率真实风险号码中被正确拦截的比例越高代表漏过越少误拦率正常号码被错误拦截的比例直接关系用户体验人工复核率状态为 review 的号码占全部请求的比例过高会消耗运营人力用户投诉率用户反馈“号码被错误标记”的比例线上效果的重要信号风控评估最稳妥的做法是“先标记后拦截”。新策略上线初期不要直接拦截而是先在用户端展示“该号码存在风险”的提示观察用户的后续行为确认误拦率可控后再切换为拦截策略。同时在线上设置 AB 实验一部分用户走旧策略一部分用户走新策略对比两组的投诉率、挂断率、账号安全事件发生率。没有对比很难证明新策略真的带来收益。下面是常见问题排查表问题现象可能原因排查方式解决方案正常号码被大量拦截规则阈值过低或模型阈值过低分析被拦截号码的命中原因分布调高阈值降低对应规则权重诈骗号码漏过率上升黑产换了新号段或新呼叫模式对比近期风险样本特征变化重训模型更新规则库上线一天后拦截量骤降特征口径不一致检查线上日志特征是否与训练一致统一特征抽取逻辑增加校验接口响应耗时过高每个请求都走完整模型推断查看每个环节耗时增加名单缓存和规则前置分层模型效果越训越差漂移样本污染训练集检查回流标注质量人工审核后再进入训练集同一号码反复出现名单未覆盖新号码池分析号码之间的关联关系基于关联图谱扩充名单线上排查的第一原则是先看日志再看特征最后看策略。不要在数据链路还没确认的情况下贸然调阈值。9. 生产环境最佳实践与合规红线文章开头说“船新阵容”竞赛阵容不能只靠一个核心选手风控系统也一样。规则、黑名单、模型、人工复核这些能力应该像一套有替补的阵容那样互相补位。某个模型未生效时规则引擎要能顶上规则失效时名单和人工审核要能兜底。在生产环境落地时下面几条经验值得记录。第一从最小可用闭环开始。先用黑名单和规则引擎上线每天能处理几万条呼叫记录再做模型升级。不要一上来就搭建复杂的大数据平台业务没验证清楚时系统再豪华也是浪费。第二策略灰度发布。任何新阈值、新模型权重都有风险。可以把流量复制一份离线回放历史数据观察新策略是否比旧策略更好。回放通过后再小流量上线逐步扩大到全量。第三定期红蓝对抗。风控系统也需要模拟攻击测试。安全团队可以扮演黑产使用新号段、改变呼叫时长、调整呼叫时间检验系统能否识别出来。这种对抗能提前暴露规则死角。第四所有操作遵循最小权限和审计原则。风控系统权限相对较大能拦截用户通信入口因此必须做严格权限控制。查看隐私数据的操作要有留痕决策变更要可回滚。第五数据合规是红线。任何系统都不能在未授权的情况下采集用户通话内容和个人隐私信息。建议使用运营商合规接口、脱敏话单或用户主动授权上报的数据。系统建设前先让法务或合规团队介入评审。第六保持模型的可解释性。即使用上深度学习模型也要为每个拦截决策保留证据链条因为什么特征、命中了什么规则、模型分数是多少。这样运营团队才能向用户解释也才能在误拦发生时快速纠正。对于刚接触风控的新手建议先拿一个离线数据集做实验使用 pandas 做特征分析用 sklearn 训练逻辑回归再用本地 FastAPI 服务暴露接口。等这条链路跑通再逐步考虑流计算、分布式存储和更复杂的模型。技术难度不是风控落地的最大障碍对数据闭环的耐心、对误拦成本的敬畏、对合规边界的重视才是决定系统能做多远的关键。