在社区运营、电商活动、内容平台甚至日常开发中我们经常遇到一批看起来“有头像、有昵称、会发言”的账号但它们的言行总是缺少真实人类的随机性和情感温度。这类账号常被称为“伪人”也就是自动化注册的机器人账号、水军账号或者由脚本批量驱动的虚假用户。本文会抛开玄学判断用 Python 从账号属性、行为序列、内容特征三个维度搭建一套可落地的“伪人鉴定”方案包含规则引擎、特征工程和机器学习判别模型适合做风控、反作弊、用户增长质量分析的同学参考。“伪人”这个概念在网络语境里有点模糊所以在动手写代码之前我们需要先给它一个可量化的定义否则后续的特征和阈值都无从谈起。接下来我会先梳理识别伪人的核心思路再给出完整的 Python 实战代码最后补充常见的坑点与工程建议。1. 背景与核心概念1.1 什么是“鉴定伪人”“伪人”这个词在不同的圈子里理解不太一样。在游戏社区里可能指代陪玩机器人在电商场景里可能指刷单小号在内容平台里则可能是批量发帖的水军。从技术角度看这些账号都有一个共同点它们不是真实用户在自然状态下产生的行为而是由脚本、群控系统或人工批量操作制造出来的流量。所以“鉴定伪人”本质上是一个风控问题。我们需要通过可观测的数据区分出“真实用户的正常行为”和“机器或批量操作产生的异常行为”。注意这里说的“伪人”不等于“坏人”。有些伪人是灰产团伙用来刷量、刷评论、恶意薅羊毛的但也有些是测试账号、爬虫抓取账号、内部自动化操作账号。鉴定的目标不是给人贴道德标签而是识别出“低质量、非自然、高重复度”的账号群体然后交给业务方根据规则去处理。用一句话概括鉴定伪人 通过特征工程 规则/模型把“非自然人行为”从用户行为数据中分离出来。1.2 为什么需要识别伪人从业务角度看伪人账号的危害非常直接浪费运营预算。活动奖品被脚本批量薅走真实的用户反而拿不到。污染数据指标。DAU、留存、转化率被虚假流量抬高导致产品决策失真。破坏社区氛围。水军刷屏、引战、批量评论会让真实用户流失。增加服务器成本。大量异常请求会消耗接口流量、存储资源和计算资源。从技术角度看识别伪人也是反爬虫、反欺诈、风控体系的重要组成部分。掌握这套识别方法对后端开发、数据分析师、安全工程师来说都是一项加分的技能。1.3 常见的识别思路目前业界的主流做法可以分成三类第一类是账号静态特征识别。看注册时间、昵称生成方式、头像是否默认、邮箱域名、设备指纹等。这类特征最直接但不能单独使用因为人工注册的小号也可以把静态信息伪装得很像真人。第二类是行为序列识别。看用户的操作间隔、点击路径、活跃时段、操作速度等。真实用户的操作有时间差、有停顿、有回退而脚本操作通常非常规整速度极快间隔接近固定值。第三类是内容特征识别。看用户发布的文本、评论、图片的重复度、相似度、语言模式。批量生成的内容往往高度重复或者带有明显的模板痕迹。实际工程中这三类特征通常组合使用先用规则快速过滤掉一批明显异常账号再用模型处理边界情况。本文会按这个思路逐步展开。2. 环境准备与数据说明2.1 运行环境本文的示例代码以 Python 3 为基础推荐使用 3.8 及以上版本。需要安装以下依赖库pandas用于数据处理。numpy用于数值计算。scikit-learn用于机器学习模型训练。如果要做文本特征还需要 jieba 分词库。依赖安装命令pip install pandas numpy scikit-learn jieba版本信息不需要完全固定只要保证 pandas 和 scikit-learn 是较新的稳定版本即可。如果你在安装过程中遇到依赖冲突建议使用虚拟环境python -m venv fake_user_env source fake_user_env/bin/activate pip install pandas numpy scikit-learn jiebaWindows 下激活命令换成fake_user_env\Scripts\activate2.2 示例数据结构为了演示我们假设业务系统中有两张数据表。第一张是用户基础表user_profile字段如下字段名说明user_id用户唯一标识register_time注册时间nickname昵称avatar_url头像地址email注册邮箱device_id设备指纹第二张是行为日志表user_action_log字段如下字段名说明user_id用户唯一标识action_time操作时间action_type操作类型如 click、comment、registertarget_id操作对象 IDip来源 IP真实场景里这些数据通常存放在 MySQL、Hive 或 ClickHouse 中。本文为了可复现会直接构造一份模拟数据。这里需要强调任何用户行为数据的采集和分析都要遵守平台的隐私政策与数据安全规范建议在实际项目中只使用必要字段并对敏感字段做脱敏处理。下面我们开始准备模拟数据。3. 数据特征拆解与构造3.1 账号静态特征账号静态特征是最容易获取的特征也是第一道过滤网。常见特征包括注册时长大量伪人账号是短时间集中注册的注册时间离现在越近风险越高。昵称特征伪人账号的昵称常常是“用户123456”“AABB_2024”这种随机组合或者带有一串无意义数字。可以用正则表达式判断昵称是否包含连续数字、随机字母组合。邮箱域名很多批量注册账号使用临时邮箱或一次性邮箱服务比如guerrillamail.com、temp-mail.org之类的域名。如果业务上维护了一份风险邮箱域名列表可以直接匹配。设备指纹一个设备在短时间内注册了多个账号属于高风险信号。下面用 Python 构造基础特征。先创建一个模拟数据集import pandas as pd import numpy as np from datetime import datetime, timedelta np.random.seed(42) user_count 2000 # 模拟注册时间真实用户分散伪人集中 register_times [] for i in range(user_count): if i % 5 0: # 模拟伪人集中注册 register_times.append(datetime(2024, 6, 1) timedelta(minutesnp.random.randint(0, 60))) else: register_times.append(datetime(2024, 1, 1) timedelta(daysnp.random.randint(0, 180))) # 模拟昵称 nicknames [] for i in range(user_count): if i % 5 0: nicknames.append(fuser_{np.random.randint(100000, 999999)}) else: nicknames.append(f真实用户_{np.random.randint(1000, 9999)}) df_user pd.DataFrame({ user_id: [fU{str(i).zfill(6)} for i in range(user_count)], register_time: register_times, nickname: nicknames, device_id: [fD{np.random.randint(1000, 9999)} for _ in range(user_count)] }) print(df_user.head())这段代码构造了 2000 个用户其中约五分之一被模拟成集中注册、昵称规则化的伪人账号。接下来构造特征now datetime(2024, 12, 31) df_user[register_days] (now - df_user[register_time]).dt.days # 昵称中是否包含下划线数字的规则模式 df_user[nickname_random] df_user[nickname].str.contains(r_?\d{5,}, regexTrue).astype(int) # 设备注册账号数量 device_reg_count df_user.groupby(device_id)[user_id].transform(count) df_user[device_reg_count] device_reg_count print(df_user[[user_id, register_days, nickname_random, device_reg_count]].head(10))特征含义register_days注册天数越小越可疑。nickname_random昵称是否命中随机数字模式命中为 1。device_reg_count同一设备注册账号数大于某个阈值时风险升高。3.2 行为序列特征比静态特征更有说服力的是行为序列特征。真实用户使用产品时行为是“稀疏”且“不规则”的。常用行为特征包括平均操作间隔脚本操作间隔通常很短比如 1 秒内连续操作多次。操作间隔标准差真实用户的间隔波动大脚本的间隔非常稳定。活跃时段分布真实用户通常集中在白天或晚上某个时间段脚本可能 24 小时持续运行。操作类型熵真实用户会点击、浏览、评论、点赞等多种操作伪人往往只会固定一两种操作。先构造模拟行为日志action_logs [] for uid in df_user[user_id]: # 判断当前用户是否模拟为伪人 is_fake np.random.rand() 0.2 action_count np.random.randint(5, 100) if not is_fake else np.random.randint(50, 300) base_time datetime(2024, 12, 1) timedelta(daysnp.random.randint(0, 30)) avg_interval np.random.uniform(5, 60) if not is_fake else np.random.uniform(0.2, 1.5) for j in range(action_count): action_logs.append({ user_id: uid, action_time: base_time timedelta(secondsj * avg_interval), action_type: np.random.choice([click, view, comment, like], p[0.3, 0.4, 0.1, 0.2]) if not is_fake else np.random.choice([comment, click], p[0.8, 0.2]) }) df_action pd.DataFrame(action_logs) print(df_action.head())有了行为日志后计算每个用户的行为指标def calc_action_features(df_action): # 每个用户的操作时间戳列表 grouped df_action.sort_values(action_time).groupby(user_id) rows [] for uid, group in grouped: times group[action_time].values intervals np.diff(times.astype(int64) / 1e9) # 秒 if len(intervals) 0: avg_interval 0 std_interval 0 min_interval 0 else: avg_interval np.mean(intervals) std_interval np.std(intervals) min_interval np.min(intervals) action_counts group[action_type].value_counts().to_dict() total_count len(group) comment_ratio action_counts.get(comment, 0) / total_count rows.append({ user_id: uid, action_total: total_count, avg_interval: avg_interval, std_interval: std_interval, min_interval: min_interval, comment_ratio: comment_ratio }) return pd.DataFrame(rows) df_action_feat calc_action_features(df_action) print(df_action_feat.head())关键点解释avg_interval表示平均操作间隔。如果大量操作间隔小于 1 秒很可能是脚本。std_interval表示间隔波动。真实用户有时快有时慢脚本通常很均匀。comment_ratio表示评论行为占比。水军账号特别喜欢评论、刷楼。3.3 内容特征如果用户会发布文本内容我们还可以构造文本特征。批量生成的文本通常有以下特点长度集中在固定范围。字符多样性低存在大量重复标点或语气词。多条文本之间相似度极高。这里以评论内容为例先用简单的文本长度和重复度作为特征# 模拟评论数据 comments [] for uid in df_user[user_id]: if np.random.rand() 0.5: continue comment_count np.random.randint(1, 10) for _ in range(comment_count): if np.random.rand() 0.3: text 支持一下写的太好了 else: text f这是一条普通评论编号{np.random.randint(1, 100)}内容仅供参考。 comments.append({ user_id: uid, comment_text: text }) df_comment pd.DataFrame(comments)计算文本特征df_comment[text_length] df_comment[comment_text].apply(len) df_comment[unique_chars_ratio] df_comment[comment_text].apply( lambda x: len(set(x)) / max(len(x), 1) ) comment_feat df_comment.groupby(user_id).agg( avg_text_length(text_length, mean), avg_unique_ratio(unique_chars_ratio, mean), comment_total(comment_text, count) ).reset_index() print(comment_feat.head())真实场景中还可以用 SimHash、MinHash 对文本做去重计算某个用户的评论与全库评论的重复度。这个方向比较深本文先不展开。3.4 特征合并与标签说明把三个维度的特征合并成一张宽表用于后续规则判定和模型训练df_feature df_user.merge(df_action_feat, onuser_id, howleft).merge(comment_feat, onuser_id, howleft) # 填充缺失值 df_feature df_feature.fillna({ action_total: 0, avg_interval: 999, std_interval: 0, min_interval: 999, comment_ratio: 0, avg_text_length: 0, avg_unique_ratio: 0, comment_total: 0 }) # 生成标签模拟数据里前 20% 为伪人 df_feature[is_fake] 0 fake_user_ids df_feature[df_feature[nickname_random] 1][user_id] df_feature.loc[df_feature[user_id].isin(fake_user_ids), is_fake] 1 print(df_feature.shape) print(df_feature[is_fake].value_counts())到这里我们的特征表已经可以用于后续判定了。特征工程是整个鉴定流程的核心特征质量直接决定规则和模型的效果。4. 核心实现规则引擎与模型判定4.1 规则引擎设计规则引擎适合做第一层过滤。它简单、可解释、运行快适合处理高吞吐量的实时请求。我设计以下几个规则如果avg_interval 2判定为“脚本操作”平均操作间隔低于 2 秒正常人很难做到长时间保持这种速度。如果device_reg_count 5判定为“设备聚集”一个设备注册超过 5 个账号风险很高。如果comment_ratio 0.8且action_total 50判定为“水军倾向”。如果avg_unique_ratio 0.3且comment_total 5判定为“模板文本”。规则命中数量越多风险分越高。实现如下def rule_score(row): score 0 reasons [] if row[avg_interval] 2 and row[action_total] 20: score 2 reasons.append(avg_interval_too_low) if row[device_reg_count] 5: score 2 reasons.append(device_reg_count_too_high) if row[comment_ratio] 0.8 and row[action_total] 50: score 1 reasons.append(comment_ratio_too_high) if row[avg_unique_ratio] 0.3 and row[comment_total] 5: score 1 reasons.append(template_text_detected) return score, ;.join(reasons) df_feature[[risk_score, risk_reason]] df_feature.apply( lambda row: rule_score(row), axis1, result_typeexpand ) df_feature[rule_label] np.where(df_feature[risk_score] 3, fake, unknown) print(df_feature[rule_label].value_counts())规则引擎的好处是能解释线上出问题的时候可以直接定位到具体规则。缺点是阈值需要不断调优而且容易被绕过。4.2 机器学习模型当规则引擎遇到边界情况时可以用机器学习模型做二次判断。这里使用随机森林或梯度提升树模型来分类。模型特征选取model_features [ register_days, nickname_random, device_reg_count, action_total, avg_interval, std_interval, min_interval, comment_ratio, avg_text_length, avg_unique_ratio, comment_total ] X df_feature[model_features] y df_feature[is_fake]划分训练集和测试集from sklearn.model_selection import train_test_split X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.3, random_state42, stratifyy ) print(X_train.shape, X_test.shape)训练随机森林模型from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score model RandomForestClassifier( n_estimators200, max_depth8, min_samples_leaf5, random_state42, n_jobs-1 ) model.fit(X_train, y_train) y_pred model.predict(X_test) y_prob model.predict_proba(X_test)[:, 1] print(classification_report(y_test, y_pred)) print(AUC:, roc_auc_score(y_test, y_prob))输出效果取决于模拟数据的分布这里重点不是追求高分而是演示完整流程。训练完成后可以查看特征重要性了解哪些特征贡献最大importance_df pd.DataFrame({ feature: model_features, importance: model.feature_importances_ }).sort_values(importance, ascendingFalse) print(importance_df)在真实项目中伪人样本的标注往往很难。我们可以把规则引擎命中的账号作为弱标签也可以用已知的人工审核结果作为强标签。建议先积累一批高质量标注样本再训练模型。4.3 特征存储与实时判定线上环境里特征工程的结果需要存储到特征平台或 Redis 中。伪人鉴定不只做离线批量分析实时接口同样需要。下面模拟一个实时判定函数def check_fake_user(user_id, action_log, user_info): 简化版的实时判定函数 # 1. 计算行为特征 if len(action_log) 1: intervals np.diff([x[action_time] for x in action_log]) avg_interval np.mean(intervals) else: avg_interval 999 # 2. 简单规则 if avg_interval 2: return {user_id: user_id, risk: high, reason: 操作间隔异常} if user_info.get(register_days, 0) 1: return {user_id: user_id, risk: medium, reason: 新注册账号} return {user_id: user_id, risk: low, reason: 无明显异常}真实场景中实时判定通常会复用训练好的模型用model.predict_proba得到风险概率再结合业务阈值决定放行、人工审核还是直接拦截。这里需要注意实时接口对延迟要求很高特征计算要尽量提前避免在请求链路中做重量级计算。5. 完整实战案例从原始日志到批量鉴定报告这一节提供一个端到端的自动化脚本它读取原始数据经过特征计算、规则判定、模型预测最后输出一份可读的鉴定报告。5.1 目录结构建议项目结构如下fake_user_detection/ ├── data/ │ ├── user_profile.csv │ └── user_action_log.csv ├── src/ │ ├── feature_engineering.py │ ├── rule_engine.py │ ├── train_model.py │ └── inference.py ├── output/ │ └── detect_result.csv └── requirements.txt这种结构在真实项目里更容易维护功能之间彼此解耦。5.2 特征工程模块将前面构造特征的代码封装成函数# src/feature_engineering.py import pandas as pd import numpy as np def build_user_features(df_user): now pd.Timestamp.now() df_user[register_days] (now - pd.to_datetime(df_user[register_time])).dt.days df_user[nickname_random] df_user[nickname].str.contains(r_?\d{5,}, regexTrue).astype(int) device_reg_count df_user.groupby(device_id)[user_id].transform(count) df_user[device_reg_count] device_reg_count return df_user def build_action_features(df_action): grouped df_action.sort_values(action_time).groupby(user_id) rows [] for uid, group in grouped: times pd.to_datetime(group[action_time]).values.astype(int64) / 1e9 intervals np.diff(times) rows.append({ user_id: uid, action_total: len(group), avg_interval: np.mean(intervals) if len(intervals) else 999, std_interval: np.std(intervals) if len(intervals) else 0, min_interval: np.min(intervals) if len(intervals) else 999, comment_ratio: group[action_type].eq(comment).mean() }) return pd.DataFrame(rows)5.3 规则引擎模块# src/rule_engine.py import numpy as np def rule_score(row): score 0 reasons [] if row[avg_interval] 2 and row[action_total] 20: score 2 reasons.append(avg_interval_too_low) if row[device_reg_count] 5: score 2 reasons.append(device_reg_count_too_high) if row[comment_ratio] 0.8 and row[action_total] 50: score 1 reasons.append(comment_ratio_too_high) return score, ;.join(reasons)5.4 模型训练模块# src/train_model.py from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report def train_model(df_feature, model_features, targetis_fake): X df_feature[model_features] y df_feature[target] X_train, X_test, y_train, y_test train_test_split(X, y, test_size0.3, random_state42) model RandomForestClassifier(n_estimators200, max_depth8, random_state42, n_jobs-1) model.fit(X_train, y_train) y_pred model.predict(X_test) print(classification_report(y_test, y_pred)) return model5.5 批量鉴定主流程# src/inference.py import pandas as pd from feature_engineering import build_user_features, build_action_features from rule_engine import rule_score from train_model import train_model # 1. 读取数据 df_user pd.read_csv(data/user_profile.csv) df_action pd.read_csv(data/user_action_log.csv) # 2. 特征工程 df_user build_user_features(df_user) df_action_feat build_action_features(df_action) # 3. 合并特征 df_feature df_user.merge(df_action_feat, onuser_id, howleft) # 4. 规则评分 df_feature[[risk_score, risk_reason]] df_feature.apply( lambda row: rule_score(row), axis1, result_typeexpand ) # 5. 模型预测 model_features [register_days, nickname_random, device_reg_count, action_total, avg_interval, std_interval, min_interval, comment_ratio] model train_model(df_feature, model_features) df_feature[model_prob] model.predict_proba(df_feature[model_features])[:, 1] # 6. 输出结果 df_feature.to_csv(output/detect_result.csv, indexFalse) print(df_feature[risk_score].value_counts())这个主流程把特征、规则、模型串了起来。在跑真实数据时你需要根据字段情况调整列名但整体结构可以复用。5.6 验证结果正常来说运行完会看到类似下面的输出precision recall f1-score support 0 0.97 0.99 0.98 561 1 0.98 0.94 0.96 239 accuracy 0.98 800不同的随机种子和数据分布会导致结果不同但流程是可复现的。关键是理解每个模块的输入输出实际项目中替换成真实数据即可。6. 常见问题与排查思路6.1 特征构造中的坑问题现象常见原因解决思路特征全是缺失值用户没有行为日志用默认值填充但不能用 0 覆盖“无行为”与“真没操作”的区别操作间隔计算出异常大值时间字段不是标准时间格式统一用 pandas to_datetime 转换分组后数据量膨胀groupby 使用错误确认聚合函数返回的是标量而非 Series模型过拟合特征里包含了用户 ID删除 user_id、时间戳等唯一标识列6.2 规则误杀真实用户规则引擎最容易被投诉的问题就是误杀。比如某些活跃用户就是喜欢快速点赞平均间隔可能低于 2 秒。解决办法把单条规则的权重降低改为多规则累计打分。引入“观察期”只有连续多天命中规则才判定风险。加入人工确认机制被误杀的账号可以申诉。6.3 模型效果不好模型效果不好通常有三个原因第一样本标签不可靠。伪人标签本身是推测出来的如果用规则标签训练模型模型只是在重复规则。第二正负样本不均衡。伪人账号占比可能只有 1% 甚至更低直接训练会导致模型偏向预测为真实用户。可以用class_weightbalanced或过采样/欠采样调整。第三特征区分度不够。需要回到特征工程重点建设行为序列特征和内容特征。6.4 线上实时接口性能瓶颈如果需要在高并发场景下做实时鉴定尽量避免在请求线程里实时计算大量聚合特征。推荐的做法是把特征计算做成离线任务写入 Redis实时接口只读取结果。模型推理可以用 ONNX 或专门的模型服务部署单次推理耗时控制在毫秒级。7. 最佳实践与工程建议7.1 特征工程要围绕“异常”而不是“平均”很多初学者喜欢直接计算平均值比如“平均操作间隔”。但伪人行为更明显的特征是“异常稳定”和“极端快速”。所以建议多构造分位数特征比如操作间隔的 10% 分位数、操作间隔的 90% 分位数会比单纯的平均值更有区分度。7.2 规则引擎和模型要分层不要把规则和模型混在一起。推荐这样分层第一层黑名单命中直接拦截。第二层规则引擎快速打分高分直接处理低分放行。第三层机器学习模型处理中分段账号。这样做的好处是模型不需要覆盖所有账号只需要处理规则难以判定的边界样本降低模型压力也提高可解释性。7.3 监控与反馈闭环伪人识别不是一次性的项目而是一个长期对抗的过程。灰产会不断调整脚本所以特征和模型必须持续迭代。建议建设以下监控指标每日识别出的伪人数量与占比。规则命中分布变化。模型预测概率分布。误杀率与漏报率抽样评估。特征分布漂移。如果发现模型准确率明显下降往往预示着对方调整了策略需要补充新特征或重新训练。7.4 安全和隐私合规在收集用户设备指纹、IP、行为数据时必须遵守法律法规和平台隐私政策。数据采集的最小化原则很重要只保留鉴定需要的字段不采集与业务无关的敏感信息。涉及批量处理用户数据时尽量在数据仓库内部完成避免把原始日志导出到本地。7.5 不要追求 100% 准确伪人鉴定在业务上很难做到 100% 准确因为“伪人”的定义本身有模糊地带。比如一个真人运营了 20 个小号这算不算伪人从行为特征看它可能很像伪人。所以更务实的做法是输出风险分让业务方决定处置策略而不是让技术团队直接拍板“删号”。8. 总结与学习路线本文从“伪人”的定义出发给出了一个完整的技术识别方案。我们先用账号静态特征、行为序列特征、内容特征三方面构造特征表再通过规则引擎做第一层快速过滤最后用随机森林模型处理边界情况。整条链路可以拆成特征工程、规则引擎、模型训练、批量推理四个模块每个模块都能单独优化和替换。如果你希望在这个方向继续深入我建议按下面的顺序学习先掌握 pandas 的分组聚合与时间序列处理这是所有特征工程的基础。再学习常用的异常检测方法比如孤立森林、DBSCAN它们适合处理没有标签的异常识别场景。然后了解如何用分布式计算处理海量日志比如 Spark 或 Flink。最后接触在线学习与模型部署用于生产环境中的实时识别。如果想降低误报率可以往“多模态融合”方向走把设备指纹、IP 风险分、文本向量、行为序列模型结合起来。但不管方案多复杂核心原则不会变先用简单规则快速过滤再用模型处理边界最后用监控保证效果。希望这篇文章能帮你少走一些弯路也欢迎在评论区聊聊你遇到的伪人案例。