简介一套基于 Python 的完整用户画像生成系统源码适合具备一定数据分析基础、希望上手用户画像落地的开发者。项目围绕数据收集预处理、特征工程、用户行为分析、聚类与特征权重计算展开覆盖从 CSV 清洗到画像存储、可视化和 Web 接口发布的全流程便于二次开发或直接运用于精细化运营与精准推荐场景。压缩包共 149 个文件大小 2.45MB核心逻辑集中在 69 个 py 脚本中另有 53 个 pyc 编译文件、5 个 html 页面、css/js 静态资源、csv 示例数据及字体图标等前后端结构较为清晰能快速定位数据、算法与展示模块。当前已有 352 人学习下载。源码中不仅包含 KMeans 聚类、TF-IDF 权重计算等关键实现还提供了 Flask/Django 集成思路与可视化图表对想系统掌握用户画像构建流程的开发者具有不错的参考价值。1. 用户画像生成系统解决什么问题标签工厂比算法模型先跑起来拿到“基于python实现用户画像生成系统源码”这个标题很多人的第一反应是去调大模型、训练兴趣模型结果项目卡在环境配置上一个月都出不了活。我用Python做用户画像系统真正能落地的那套东西不是某个炫技的机器学习模型而是一条条可靠可解释的标签近30天消费金额、最近一次活跃时间、品类偏好、流失风险等级。运营要的答案永远是一张可以直接筛选人的表而不是一个黑匣子般的概率分数。一个用户画像系统要回答三件事用户是谁、做过什么、接下来该给他推什么。做这件事适合两类人。一类是电商或内容平台的后端工程师手里的埋点数据和订单数据一直躺着想变成运营能直接用的标签另一类是SaaS工具的产品负责人要给客户交付一套可解释的用户洞察能力而不是又一个数据大屏。它不依赖海量数据和GPU集群一台普通开发机加上MySQL和Scikit-learn就能跑起来这正是用Python而不是Spark、Flink来做第一版的原因——迭代快、好排查、代码难维护时也容易重构。2. 画像系统的数据层设计标签体系、ID打通与宽表建模2.1 标签体系三种分类统计标签、规则标签、算法标签怎么选我见过很多团队一上来就买算法标签的账问“能不能预测一下用户性别”结果连基础标签都没建。落地时第一件事是给标签分类分清轻重缓急。通常分成三类统计标签、规则标签和算法标签它们的计算复杂度、更新频率和维护成本完全不一样不能混为一谈。统计标签是画像系统的基本盘。它直接用SQL或pandas对行为数据做聚合例如“近7天登录天数”“累计消费金额”“最近一次购买距今天数”这类标签逻辑简单、解释成本低出了偏差一眼就能看到源头适合高频更新。规则标签是在统计结果上套业务规则最典型的是RFM用户分层、沉睡用户定义、高价值客户识别规则一旦写清楚运行结果完全可复现运营也认这套逻辑。算法标签才轮到机器学习上场常见的有用TF-IDF或Word2Vec给用户打内容兴趣标签、用聚类做用户分群、用逻辑回归预测流失概率。我的排序建议是先把统计标签全部跑通再用规则标签覆盖业务核心场景最后才上算法标签。原因很现实运营日常使用的标签中统计和规则两类能覆盖掉八成需求算法标签更多是加分项而不是救命稻草。而且统计和规则标签的产出物是人和标签键值算法标签产出的是一堆概率分或向量后者需要额外的解释和验证成本。另一个重要原因是更新频率不同统计标签可以每小时刷新规则标签每天刷新一次就够了算法标签一周跑一次都算高频把这三类塞进同一张表、同一个调度频率运维上迟早翻车。2.2 ID打通没有统一用户标识画像就是一堆碎片做画像系统第一个绕不过去的坎是ID打通。一个用户在未登录状态下浏览商品埋点系统记的是设备ID等到下单时他登录了订单系统记的是user_id他可能还在小程序里用openid访问过。如果不做打通“浏览过某商品”和“买了某商品”会被算到两个不同的人头上画像标签自相矛盾是必然的。常见的做法是建一张id_mapping映射表把同一个人的多类ID归并成一个唯一的用户主键。我把这张表的建表语句放在下面它是整个画像系统的地基。CREATE TABLE id_mapping ( user_id BIGINT COMMENT 用户主键统一标识, device_id VARCHAR(64) COMMENT 设备ID未登录时用于关联, openid VARCHAR(64) COMMENT 小程序openid可选, mobile VARCHAR(20) COMMENT 手机号可选, merge_time DATETIME COMMENT 该ID关系首次确认时间, is_primary TINYINT DEFAULT 1 COMMENT 是否主ID1是0否 ) COMMENT 用户ID映射表做身份打通;这张表的逻辑是以user_id为主键它代表一个真实用户device_id和openid算作从属ID。实际处理流程是用户登录后把当前设备上积累的浏览行为全部归属到user_id名下并在id_mapping里写入一条device_id到user_id的映射。后续跑标签时凡是遇到只有device_id的记录先查id_mapping如果映射存在就替换成user_id不存在就单独标记为“未登录访客”不参与画像计算。这里有个关键参数值得注意merge_time。它记录的是第一次确认设备与用户归属关系的时间不是最近一次活跃时间。这个时间戳是审计和回溯的依据一旦运营质疑某个标签归属错误可以用它追查是何时、基于什么行为把设备分配给这个用户的。很多团队图省事只在代码里写replace替换丢了这条审计信息后期排查会非常痛苦。2.3 用户宽表与标签表两张表撑起整个画像系统ID打通之后下一步是设计存储结构。画像系统不需要复杂的图数据库或列式存储两张表就能撑起大部分业务一张用户宽表profile_wide一张标签表user_tags。用户宽表的结构是“一用户一行”每一列是一个统计特征。它的消费者是模型和报表查询时按user_id取一整行特征或者一次性批量取几万行做统计分析。比如“取近30天有加购但没下单的用户”在宽表里就是一条SQLtotal_add_cart 0 AND order_count 0。宽表建议用Parquet列式存储加分区每天一个分区查询时只扫当天分区速度和成本都可控。标签表的结构是“一用户多行”每一行是一个标签键值对。它的消费者是运营系统他们要的是“给我所有‘高价值’用户”“把所有‘90天未活跃’用户的标签推给短信系统”。标签表用MySQL或StarRocks这类支持索引的数据库存储即可核心字段包括user_id、tag_key、tag_value、tag_type和updated_at。之所以不放宽表是因为标签的键是动态的今天加一个“大促敏感度”明天删一个“会员等级”用宽表存会产生大量空列运维起来是灾难。宽表和标签表之间靠user_id关联更新频率错开。我一般把宽表设计成每小时更新一次标签表每天凌晨全量重建一次。这样既保证了运营早上打开后台看到的是当天最新标签又不会因为频繁重建导致数据库压力过大。有一点要提前想到标签表不要做成更新插入而是每天先删除当天过期标签再插入新值保留tag_type字段做分类方便后续清理无用标签。这就是标签系统的后悔药——留了type和updated_at真出了问题能快速回滚到前一天版本。3. 用Python计算标签生成画像表从清洗到落库的完整代码3.1 从订单表生成第一批统计标签筛选、聚合与合并一次到位这一节我给你一套可以直接复制运行的最小实现假设数据在MySQL的orders表里字段包括user_id、order_amount、order_time、status。目标是生成profile_wide表里的几个核心字段累计消费金额、累计订单数、最近一次购买时间距离今天的天数、消费品类数。import pandas as pd from datetime import datetime, timedelta from sqlalchemy import create_engine # 连接MySQL生产环境建议用配置文件管理连接串 engine create_engine(mysqlpymysql://user:passlocalhost:3306/app_db) # 读取订单表只取近365天有效订单减少数据量 sql SELECT user_id, order_amount, order_time, category_id FROM orders WHERE status paid AND order_time DATE_SUB(NOW(), INTERVAL 365 DAY) df pd.read_sql(sql, engine) print(原始订单行数:, len(df)) # 按用户聚合生成宽表统计特征 wide df.groupby(user_id).agg( total_amount(order_amount, sum), # 累计消费金额 total_orders(order_id, count), # 累计订单数 last_order_time(order_time, max), # 最近一次订单时间 category_cnt(category_id, nunique), # 购买过的品类数 ).reset_index() # 计算最近一次购买距离今天的天数 wide[recency_days] (datetime.now() - pd.to_datetime(wide[last_order_time])).dt.days wide wide[[user_id, total_amount, total_orders, recency_days, category_cnt]] # 写出到宽表使用replace策略每天全量覆盖 wide.to_sql(profile_wide, engine, if_existsreplace, indexFalse)这段代码里有三个参数值得解释。第一是筛选近365天这是画像宽表最常见的统计窗口既可以控制数据量也覆盖了大多数业务场景如果你的业务周期短改成180天或90天都行但窗口一旦确定就不要频繁改否则前后对比会失真。第二是status paid这一步是很多人忽视的坑订单表里会混入取消、退货、未付款的订单不做状态过滤的话金额和订单数标签全是脏的。第三是if_existsreplace宽表每天全量重建简单直接到数据量上亿后再改成按日分区增量写入第一版不必为性能过度设计。运行完这段代码之后你应该得到一张最小可用的画像宽表。拿着这张表给运营就能回答“累计消费前10%的用户是谁”“最近90天没买过东西的人有多少”这些是运营每天都在问的高频问题。宽表是画像系统里最不值钱又最值钱的东西——不值钱是因为它只是个聚合查询值钱是因为它让后续所有标签和模型都站在同一套数据口径上。3.2 规则标签落地RFM模型用户分层的一段完整实现统计标签打底之后业务最想要的是分层标签RFM模型是规则标签里最经典的一条。RFM用三个维度给用户打分Recency最近购买时间、Frequency购买频次、Monetary消费金额。它的价值不在算法在于让运营一眼看清哪些人是高价值客户、哪些人正在流失边缘。下面是一段完整实现包含打分和分层两个环节。import pandas as pd import numpy as np # 读取宽表数据这里直接用上一节生成的结果 wide pd.read_sql(SELECT * FROM profile_wide, engine) # 计算RFM分数分母加1防止除零 wide[R_score] pd.qcut(wide[recency_days], 5, labels[5, 4, 3, 2, 1]) wide[F_score] pd.qcut(wide[total_orders].rank(methodfirst), 5, labels[1, 2, 3, 4, 5]) wide[M_score] pd.qcut(wide[total_amount].rank(methodfirst), 5, labels[1, 2, 3, 4, 5]) # 把分数字段转为数值类型便于比较 for col in [R_score, F_score, M_score]: wide[col] wide[col].astype(int) # 规则分层金字塔型的客户价值判断 def rfm_level(row): if row[R_score] 4 and row[F_score] 4 and row[M_score] 4: return 高价值用户 elif row[R_score] 3 and row[M_score] 3: return 潜力用户 elif row[R_score] 2 and row[F_score] 3: return 流失风险用户 elif row[R_score] 1: return 沉睡用户 else: return 普通用户 wide[rfm_level] wide.apply(rfm_level, axis1)这里有两个关键选择。第一是打分用pd.qcut做分位数切分而不是固定阈值原因是业务量级在增长用户金额和频次的绝对值一直在变用分位数能保证每层人数比例相对稳定运营不需要每季度调一次阈值。第二是分层规则用if-elif而不是复杂模型因为规则要能解释给业务听运营只要知道“高价值用户是R和F和M都排前20%的人”就够了不需要理解评分卡或密度聚类。这段代码在数据量低于一两万行时跑得很快但有个边界注意如果某个维度数据分布极度偏斜比如大量用户recency_days相同pd.qcut会报Bin edges must be unique错误。解决办法是像代码里F_score那样先rank(methodfirst)再加扰动或者对重复值做去重处理。这是RFM落地最常翻车的地方不是模型问题是数据分布问题。3.3 算法标签上手用TF-IDF给用户打内容兴趣标签统计标签和规则标签跑通之后可以上一个算法标签练手TF-IDF打内容兴趣是性价比最高的选择。适用场景是用户浏览过一批文章或商品标题你要给每个用户提炼出3到5个关键词作为兴趣标签比如“健身”“数码”“考研”。核心思路是把用户看过的所有标题拼成一篇文章再用TF-IDF算关键词权重取权重最高的几个词作为标签。import jieba import pandas as pd from sklearn.feature_extraction.text import TfidfVectorizer # 读取用户浏览记录user_id, content_title, click_time logs pd.read_sql(SELECT * FROM user_content_logs, engine) # 按用户拼接浏览过的标题文本 user_texts ( logs.groupby(user_id)[content_title] .apply(lambda x: .join(x)) .reset_index() ) # 使用固定的停用词表和jieba分词构造TF-IDF向量 vectorizer TfidfVectorizer( tokenizerjieba.lcut, stop_words[的, 了, 是, 我, 你, 他, 在, 有, 和], max_features5000, # 控制词表大小防止维度爆炸 min_df3, # 词至少在3个用户文本中出现过 max_df0.8 # 超过80%用户都有的词无区分度去掉 ) tfidf_matrix vectorizer.fit_transform(user_texts[title]) feature_names vectorizer.get_feature_names_out() # 取每个用户权重最高的3个词作为兴趣标签 interest_tags {} for idx, row in user_texts.iterrows(): scores tfidf_matrix[idx].toarray().flatten() top3 [feature_names[i] for i in scores.argsort()[-3:][::-1]] interest_tags[row[user_id]] top3 print(示例结果:, list(interest_tags.items())[:3])这里三个参数是必须调的。max_features5000是为了控制词表规模不设的话中文的分词结果可能有几万个词矩阵稀疏而且占用内存大。min_df3让小众词保留但过滤掉只出现过一次的词避免某个用户因为看了篇冷门文章就被打上一个奇怪标签。max_df0.8处理的是“视频”“文章”这类高频泛词超过80%用户都有的词没有区分度必须去掉。运行完这段代码拿到的兴趣标签本质上是“用户在某个内容子集上的集中度体现”不是真正的意图理解。但它足够支撑一个初版推荐理由展示“因为你常看健身内容所以给你推荐这篇”。这个效果对用户的感知已经非常强了。如果后续数据量上来可以平滑升级到Word2Vec甚至预训练模型但第一版用TF-IDF把链路跑通比一步到位上深度模型稳妥得多。4. 用户画像系统落地避坑五连ID纷争、时区、口径、内存与调度4.1 坑一未登录浏览数据和登录后订单数据永远对不上现象画像里出现两类用户一类只有浏览记录没有订单另一类只有订单记录没有浏览运营按“浏览了但没下单”的人群做营销结果触达名单里一半是只浏览过但根本没注册的访客。原因埋点系统和订单系统使用两套ID浏览行为挂在device_id上订单挂在user_id上没有做ID打通就直接聚合。解决按2.2节里的id_mapping表做归并。处理订单和浏览数据时统一先通过id_mapping把device_id转成user_id再计算。对没匹配到user_id的设备ID单独标记为匿名访客不进入画像主表。这个坑最可怕的不是技术复杂而是发现得晚——第一版画像如果没打通ID等于整个标签体系都建立在流沙上后期修数据成本极高。4.2 坑二昨天还是高价值用户今天一觉醒来变成了流失用户现象运营反馈同一个用户昨天的标签还是“高价值用户”今天整个标签变成了“沉睡用户”两天的标签结果对不上运营对数据的信任一夜崩塌。原因标签口径在代码里被悄悄改过。比如RFM的评分从“全量历史订单”改成了“近90天订单”或者上游订单表新增了退款状态过滤导致同一批用户的M_score大幅下降。这种口径漂移在代码版本迭代时极易发生没有版本管理的标签系统等于没有刹车。解决建一张标签口径字典表把每个标签的计算口径、SQL片段、版本号、生效时间登记在案。每次修改标签口径必须新建版本号而不是原地覆盖。跑批任务完成后对关键标签做波动检测比如“高价值用户”占比如果相比昨天波动超过20%任务告警并冻结结果发布让开发确认是业务波动还是口径变化。把口径当成代码一样对待画像系统才谈得上可维护。4.3 坑三凌晨三点画像任务跑挂了早上十点运营才知道现象任务夜里运行失败没有重试也没有告警邮件报表照样发出去了里面全部是前一天甚至更早的旧数据运营拿着过期画像做了一上午外呼。原因画像任务的调度系统没有设置失败重试和告警或者任务之间有依赖但调度器不知道上游任务挂了下游任务还傻傻地跑成功。解决给每个画像调度任务加三个配置依赖检查、失败重试、完成告警。依赖检查指任务启动前先确认上游表和前一天数据分区已经就绪失败重试设3次间隔5分钟应对数据库连接抖动完成告警分成功和失败两类失败通知责任人成功但数据量异常也要通知。我见过太多团队只做失败告警不做成功校验结果失败重试成功后没人知道数据其实已经晚了两个小时。4.4 坑四pandas全量聚合导致内存溢出现象画像任务跑着跑着进程被系统杀掉报MemoryError一看日志是df.groupby().agg()处理全量订单表时内存爆了。原因订单表几千万行按user_id做groupby聚合时pandas要把全表加载进内存加上中间生成的临时对象内存翻两三倍很正常。很多Python工程师习惯性read_sql(SELECT * FROM orders)完全不考虑数据量。解决分三步优化按时间分区读取数据用增量聚合而不是全量聚合数据量超过千万行时改用polars或DuckDB。第一版常用的是时间分区按天读取订单数据先reduce成(user_id, 当天金额, 当天订单数)再和昨天结果累加。这样内存里任何时候只保留几天的聚合结果而不是全表。这个优化不改变标签口径只改变计算路径收益立竿见影。4.5 坑五标签空值率一夜之间飙升到60%现象第二天数据质量检查发现“消费金额”标签的空值率从2%涨到了60%排查了一圈发现不是程序问题而是上游订单表新增了order_type字段一部分新类型订单没有被识别为有效订单。原因上游表结构变更下游消费逻辑没有同步更新。数据系统的表结构变更永远是画像系统的最大风险源数据库不会通知你没读到新数据。解决给关键标签配置空值率和波动率监控。画像任务跑完必须执行一段质量检查代码统计每个核心标签的空值率、非空行数、与昨天的差异幅度超过阈值立即阻断发布并通知数据负责人。这段检查代码不是可选项它和计算代码一样重要我在后面第6章会给出一份可以直接用的巡检脚本。5. 把画像变成服务用FastAPI提供标签查询接口与可视化基础5.1 一个最小的标签查询接口输入user_id返回用户全量标签画像表建好之后业务方要能查得到人。最直接的方式是提供一个HTTP接口输入user_id返回这个用户的所有标签。用FastAPI写这个接口很顺手天然支持参数校验和JSON输出。下面是一个可以直接运行的查询服务。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import pymysql app FastAPI() DB_CONFIG { host: localhost, user: app_user, password: your_password, database: user_profile, charset: utf8mb4 } app.get(/user/tags/{user_id}) def get_user_tags(user_id: int): conn pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: sql SELECT tag_key, tag_value, tag_type, updated_at FROM user_tags WHERE user_id %s cursor.execute(sql, (user_id,)) rows cursor.fetchall() if not rows: raise HTTPException(status_code404, detailuser not found) return {user_id: user_id, tags: rows} finally: conn.close()这段代码里最需要注意的是SQL参数化直接用%s占位符而不是字符串拼接防止SQL注入画像接口一旦被注入等于把所有用户标签数据暴露给攻击者。另一个细节是查不到用户时返回404而不是空列表让调用方清楚地区分“用户不存在”和“用户没有标签”两种场景。生产环境还要加一个缓存像FastAPI里用lru_cache装饰器处理高频标签或者用Redis缓存热点用户接口QPS能提升一个量级。5.2 群体圈人接口给运营一个按标签筛人的查询能力单个用户查询只解决了“这个用户是谁”的问题运营更多时候要的是“帮我拉出所有高价值用户”这类圈人场景。这时候需要一个按标签查询用户列表的接口参数是tag_key和tag_value返回符合条件的人群明细。app.get(/segment/users) def get_users_by_tag(tag_key: str, tag_value: str, limit: int 100): conn pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: sql SELECT user_id, updated_at FROM user_tags WHERE tag_key %s AND tag_value %s ORDER BY updated_at DESC LIMIT %s cursor.execute(sql, (tag_key, tag_value, limit)) rows cursor.fetchall() return {total: len(rows), users: rows} finally: conn.close()这个接口有三个参数要调。第一是limit必须限制返回行数画像人群有可能几十万甚至上百万不限制会把应用拖垮需要全量导出时另做异步导出任务不走同步接口。第二是tag_key的枚举控制数据库里tag_key是受限集合接口层要把可查询的key白名单暴露出去防止运营随意传一个不存在的标签让数据库全表扫描。第三是tag_value支持模糊匹配还是精确匹配建议默认精确匹配性能更好模糊查询另开接口用ES。5.3 画像服务的权限与请求日志别让经营数据裸奔画像数据属于核心经营数据接口一旦上线就面向多个内部系统权限控制和审计日志必须在第一天就加好否则后面很难补。常见做法是调用方注册API Key请求头携带Key网关校验通过后才转发到FastAPI服务。from fastapi import Header, Depends, HTTPException VALID_API_KEYS {ops_web: key_ops_2024, recommendation: key_rec_2024} def verify_api_key(x_api_key: str Header(...)): if x_api_key not in VALID_API_KEYS: raise HTTPException(status_code401, detailinvalid api key) return x_api_key app.get(/segment/users) def get_users_by_tag(tag_key: str, tag_value: str, limit: int 100, api_key: str Depends(verify_api_key)): # 实际查询逻辑与上一节相同 pass这里要强调一个容易被忽略的点API Key不要按人员分配要按系统分配。一个人调用了接口离职了Key要能单独吊销而不影响其他系统按人员分配的话人走了只能全部换Key牵连面太大。日志方面每一次接口调用要记录调用方、查询参数、返回数据量、耗时写入独立的审计日志表。运营用画像圈人做营销活动一旦出现用户投诉“你们怎么知道我浏览过什么”没有审计日志就是哑巴吃黄连。6. 画像系统的日常体检标签监控与数据质量验证画像系统上线只是开始长期稳定运行靠的是每天几分钟的自动体检而不是出了问题再排查。我要求我的画像任务跑完后必须执行一段质量检查脚本把核心指标写入监控表异常直接推送告警。下面这段代码是体检的核心逻辑完全可以复用。from datetime import datetime, timedelta import pymysql import pandas as pd # 读取昨天画像标签 conn pymysql.connect(**DB_CONFIG) yesterday (datetime.now() - timedelta(days1)).strftime(%Y-%m-%d) sql fSELECT tag_key, COUNT(*) as cnt, SUM(tag_value IS NULL) as null_cnt FROM user_tags WHERE dt {yesterday} GROUP BY tag_key df pd.read_sql(sql, conn) # 计算核心监控指标 df[null_rate] df[null_cnt] / df[cnt] # 空值率 df[cnt_diff] df[cnt] - df[cnt].shift() # 波动量 abnormal df[(df[null_rate] 0.05) | (df[cnt_diff].abs() 50000)] if not abnormal.empty: # 触发告警实际接入飞书/钉钉/邮件机器人 print(异常标签:, abnormal[[tag_key, null_rate, cnt_diff]])体检只看三个指标空值率有没有超过5%、标签覆盖人数波动有没有超过某个阈值、与昨天相比数据分布有没有异常。这三个指标能拦截掉我在第4章里讲的空值飙升和口径漂移问题。阈值怎么定我习惯先跑两周基线取正常波动范围的两倍作为告警阈值不拍脑袋设数字。这段脚本放进调度器每天早上9点自动执行比任何人工巡检都可靠。我个人的习惯是把这套质量检查沉淀成一个独立脚本和画像计算任务完全分开不让计算任务自己检查自己——自己检查自己等于让运动员当裁判日志会被任务的异常中断波及。计算任务只负责出数检查任务单独跑两个链路互不干扰。这样做还有一个额外收益新同事接手画像系统时不用翻阅一堆计算代码去理解数据预期直接看检查脚本就能明白哪些标签是核心资产、正常波动范围是多少。画像系统做到这一步本质上已经从“生成一批标签”进化成了“一个可信的标签服务平台”。回头来看整个落地路径里最花时间的不是写代码而是把ID打通、口径管理和监控体系做扎实。希望帮到你。本文还有配套的精品资源点击获取
