基于Python的B站用户行为分析系统：从数据采集到指标计算

简介：这是一套面向课程设计与Python后端学习者的B站用户行为分析系统完整项目源码，适合数据科学、机器学习方向的学生与开发者作为实战案例参考。项目围绕观看历史、弹幕、评论等行为数据展开，涵盖爬虫采集、Pandas数据处理、Matplotlib可视化，并尝试用聚类与关联规则挖掘用户偏好，最终输出用户画像与内容推荐列表，配套界面可查看分析报告。资源包共427个文件，约72.41MB，以png图表、js脚本、html页面、py源码及pyc缓存为主，另含css样式、svg图标、sql建表脚本、pkl模型文件与doc说明文档，前后端与数据模块划分清晰。已有252人学习下载，可帮助读者快速理解从数据抓取到分析展示的完整链路，并借鉴其目录组织与排错思路。

1. 从一份 B 站用户行为数据说起：这套分析系统到底在算什么

很多人第一次接触「B站用户行为分析系统」这个词，脑子里浮现的是爬虫抓弹幕、画几张词云图就完事。真做过一轮的人会告诉你，难点根本不在爬，而在「行为」两个字怎么定义。一个用户点开视频、拖进度条、发弹幕、投币、收藏、充电，这些动作在时间轴上的分布，才是行为分析真正要处理的对象。这套基于 Python 的系统，核心目标是把散落在播放页、评论区、动态流里的原始事件，整理成能回答「谁在看、看到哪、为什么走」的结构化指标。

它适合三类人：做内容运营想搞清楚完播率为什么掉的人、学 Python 数据分析想找一个真实数据集练手的人、以及需要给推荐或风控模块提供特征输入的工程师。标题里带「.zip」，说明交付形态是一份可解压运行的工程，而不是一篇论文。所以这篇笔记不聊虚的，直接按「数据从哪来 → 怎么存 → 怎么算 → 怎么避坑 → 怎么验证」的顺序，把一套能跑起来的最小闭环讲清楚。你不需要先成为爬虫高手，但得愿意动手改参数、看日志、对数据。

2. 行为数据从哪来：接口、字段与采集边界

2.1 先想清楚要采哪几类行为事件

B 站用户行为分析系统里，最容易被低估的一步是事件建模。新手常犯的错是「先把能抓的都抓下来」，结果字段几百个，真正能进分析管道的不到十个。我一般会把行为事件收敛成四类：曝光类（视频出现在推荐流/搜索页）、播放类（开始播放、暂停、拖拽、完播）、互动类（点赞、投币、收藏、弹幕、评论）、转化类（关注、充电、分享）。这四类对应分析里的漏斗层级，缺一层，后面的留存和归因就接不上。

字段设计上，每条事件至少要有：用户标识（uid 或脱敏后的 hash）、视频标识（bvid）、事件类型、事件时间戳（毫秒级）、以及一个上下文对象（来源页面、设备类型、是否登录）。时间戳精度很关键，秒级精度做不了拖拽行为分析，因为一次拖拽可能在 800 毫秒内完成。常见做法是用time.time() * 1000取毫秒，落库时统一存整数，避免浮点误差在聚合时放大。

提示：uid 属于个人信息，落库前做一次不可逆哈希，分析用哈希值，展示层不还原。这不是可选项，是底线。

2.2 用 Python 写一个最小采集与清洗脚本

采集环节我不建议一上来就上分布式，先用单机把一条链路跑通。下面这段代码演示的是「读取一批原始 JSON 事件 → 清洗 → 输出成分析用的 DataFrame」，原始数据可以来自你已有的日志文件或公开数据集，重点是清洗逻辑本身。

import json import hashlib import pandas as pd from datetime import datetime def hash_uid(uid: str) -> str: # 对 uid 做加盐哈希，盐值从环境变量读取，不写死在代码里 salt = "bili_behavior_salt" return hashlib.sha256((salt + str(uid)).encode()).hexdigest()[:16] def clean_event(raw: dict) -> dict | None: # 丢弃缺少关键字段的脏数据，而不是用默认值填充 required = ["uid", "bvid", "event_type", "ts"] if any(k not in raw or raw[k] in (None, "") for k in required): return None ts = int(raw["ts"]) # 时间戳合理性校验：早于 2015 年或晚于当前时间 1 天的直接丢 if ts < 1420041600000 or ts > int(datetime.now().timestamp() * 1000) + 86400000: return None return { "uid_hash": hash_uid(raw["uid"]), "bvid": raw["bvid"], "event_type": raw["event_type"], "event_ts": ts, "source": raw.get("source", "unknown"), "device": raw.get("device", "unknown"), } def load_events(path: str) -> pd.DataFrame: rows = [] with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: raw = json.loads(line) except json.JSONDecodeError: continue # 单行解析失败不影响整批 cleaned = clean_event(raw) if cleaned: rows.append(cleaned) df = pd.DataFrame(rows) df["event_ts"] = pd.to_datetime(df["event_ts"], unit="ms") return df if __name__ == "__main__": df = load_events("raw_events.jsonl") print(df.shape) print(df["event_type"].value_counts())

逻辑说明：hash_uid用加盐 SHA256 截前 16 位，既保证同一用户可关联，又不可逆推。clean_event采用「缺字段即丢弃」策略，而不是填默认值，因为行为分析里一条时间戳错误的记录比少一条记录危害大得多。load_events逐行解析，单行 JSON 出错就跳过，保证批量任务不会因为一行脏数据整体失败。

参数说明：盐值bili_behavior_salt实际部署时应从环境变量注入；时间戳下限1420041600000对应 2015-01-01，可按你的数据范围调整；event_type的取值建议提前定义枚举，避免同一行为出现「click」「Click」「CLICK」三种写法。

2.3 采集频率与限速的现实取舍

做 B 站用户行为分析，绕不开请求频率问题。我的血泪经验是：不要试图在短时间内拉全量，而是按「热点视频优先、长尾视频抽样」的策略分层采集。热点视频（比如榜单前 500）可以高频更新，长尾视频按天抽样即可。单机采集时，请求间隔设 1 到 2 秒，配合指数退避重试，比硬扛限速要稳得多。如果你只是做分析练手，直接用官方公开的脱敏数据集或自己构造的模拟日志，能省掉大量和接口斗智斗勇的时间。

3. 把行为存成能算的结构：表设计与聚合管道

3.1 事件表、会话表、指标表的三层拆分

数据存不好，后面算什么都别扭。我一般拆三张表：事件表（明细，一行一个动作）、会话表（把同一用户 30 分钟内的连续行为切成一个 session）、指标表（按视频/按天聚合的宽表）。事件表用列式存储（Parquet）比 CSV 快一个量级，尤其是做时间范围扫描的时候。会话表的关键是「30 分钟」这个阈值，它来自通用的会话切分惯例，但 B 站场景下可以调到 20 分钟，因为短视频消费的间隔更短。

指标表要提前想好维度：视频维度（bvid、播放量、完播率、互动率）、用户维度（uid_hash、活跃天数、行为分布）、时间维度（小时、天、周）。三张表之间用 bvid 和 uid_hash 关联，不要用自增 ID，否则跨表 join 时容易错位。

3.2 用 Pandas 做会话切分与完播率计算

会话切分是行为分析里最容易写错的一段。核心逻辑是：按用户分组，按时间排序，当前事件与上一条事件间隔超过阈值就开新会话。下面这段代码可以直接抄。

import pandas as pd SESSION_GAP_MS = 20 * 60 * 1000 # 20 分钟无行为则切分会话 def build_sessions(df: pd.DataFrame) -> pd.DataFrame: df = df.sort_values(["uid_hash", "event_ts"]).copy() # 计算同一用户相邻事件的时间差 df["gap"] = df.groupby("uid_hash")["event_ts"].diff().dt.total_seconds() * 1000 # gap 为空（每个用户第一条）或超过阈值，标记为新会话 df["is_new_session"] = (df["gap"].isna()) | (df["gap"] > SESSION_GAP_MS) df["session_seq"] = df.groupby("uid_hash")["is_new_session"].cumsum() df["session_id"] = df["uid_hash"] + "_" + df["session_seq"].astype(str) sessions = df.groupby("session_id").agg( uid_hash=("uid_hash", "first"), start_ts=("event_ts", "min"), end_ts=("event_ts", "max"), event_count=("event_type", "count"), ).reset_index() return sessions def calc_finish_rate(df: pd.DataFrame) -> pd.DataFrame: # 完播定义：同一用户同一视频出现 finish 事件，或播放时长 >= 视频时长的 90% play = df[df["event_type"].isin(["play", "finish"])] agg = play.groupby("bvid")["event_type"].apply( lambda s: (s == "finish").sum() / max((s == "play").sum(), 1) ).reset_index(name="finish_rate") return agg

逻辑说明：build_sessions先用diff算相邻事件毫秒差，再用cumsum给每个新会话打递增序号，拼出唯一 session_id。calc_finish_rate用 finish 事件数除以 play 事件数，分母用max(..., 1)防止除零。

参数说明：SESSION_GAP_MS是唯一需要按业务调的参数，短视频场景可降到 15 分钟，长视频可升到 30 分钟；完播率的分母如果换成「去重用户数」而不是「播放次数」，得到的是另一套指标，两者不要混用。

3.3 聚合管道的调度与增量更新

全量重算在数据量上来之后会变得不可接受。常见做法是按天分区，每天只重算最近 7 天的会话和指标，因为会话切分只依赖相邻事件，历史数据不会因为新数据而改变。用 Airflow 或简单的 cron + Python 脚本都能做，关键是给每张表加dt分区字段，重跑时先删对应分区再写入，避免重复累加。这一步不做，后面所有指标都会偏大，而且很难排查。

4. 可视化与特征输出：让分析结果能被用起来

4.1 用 Matplotlib 画留存曲线和漏斗

分析结果如果只停在 DataFrame 里，运营和产品是用不起来的。最实用的两张图是留存曲线和行为漏斗。留存曲线按「首次行为日期」分组，看后续每天的活跃比例；漏斗按「曝光 → 播放 → 互动 → 转化」四层算转化率。下面这段代码画漏斗，留存曲线同理，只是分组维度换成日期差。

import matplotlib.pyplot as plt def plot_funnel(df: pd.DataFrame, bvid: str): stages = ["exposure", "play", "interact", "convert"] sub = df[df["bvid"] == bvid] counts = [sub[sub["event_type"] == s]["uid_hash"].nunique() for s in stages] fig, ax = plt.subplots(figsize=(8, 4)) ax.barh(stages, counts, color="#4C72B0") for i, c in enumerate(counts): ax.text(c, i, f" {c}", va="center") ax.set_xlabel("unique users") ax.set_title(f"funnel for {bvid}") plt.tight_layout() plt.savefig(f"funnel_{bvid}.png", dpi=150)

逻辑说明：用nunique而不是count，因为漏斗每一层要的是去重用户数，同一用户多次播放只算一次。barh横向条形图比纵向更适合展示阶段名称较长的漏斗。

参数说明：dpi=150是发布到文档的清晰度下限，屏幕展示 100 即可；如果阶段名称要中文化，记得先设置plt.rcParams["font.sans-serif"]，否则中文会显示成方块，这是新手最常见的翻车点之一。

4.2 把指标导出成推荐/风控可用的特征

如果这套系统的下游是推荐或风控，指标表要转成「用户×视频」的特征向量。常见特征包括：该用户对该视频的播放次数、平均播放进度、互动次数、距上次互动天数。导出时用 Parquet 按天分区，特征名统一小写下划线风格，缺失值用 -1 而不是 0，因为 0 在树模型里是有意义的取值，用 0 填缺失会引入偏差。这一步的细节决定了特征能不能直接用，而不是还要下游再洗一遍。

5. 避坑与排查：这套系统最容易翻车的五个地方

5.1 时间戳单位混用导致会话全乱

现象：会话数量比预期多出几倍，每个会话只有一两条事件。原因：一部分数据用秒级时间戳，一部分用毫秒级，diff算出来的间隔忽大忽小。解决：在清洗阶段统一转成毫秒整数，并加一条断言assert df["event_ts"].max() > 1e12，秒级时间戳不会超过这个量级，能在入口就拦住。

5.2 完播率分母用了播放次数而非去重用户

现象：完播率超过 100%。原因：同一用户重复播放被多次计入分母，而 finish 事件只记一次。解决：分母改用nunique去重用户数，或者明确把指标命名为「完播次数/播放次数」并在文档里写清口径，两者不能混叫「完播率」。

5.3 中文图表乱码让整张图报废

现象：图表里所有中文变成方框。原因：Matplotlib 默认字体不含中文字形。解决：在绘图前设置plt.rcParams["font.sans-serif"] = ["SimHei"]和plt.rcParams["axes.unicode_minus"] = False，Linux 环境下如果没有 SimHei，换成Noto Sans CJK SC并确认字体已安装。

5.4 增量重算时忘记删分区导致指标翻倍

现象：某天指标突然变成前一天的近两倍。原因：重跑任务时直接 append，没有先清空对应dt分区。解决：写入前执行DELETE FROM table WHERE dt = '目标日期'，或者用mode="overwrite"配合分区列写入，让存储层自己处理覆盖。

5.5 uid 未脱敏直接落库

现象：代码 review 时被安全同学拦下。原因：原始 uid 属于个人信息，直接存明文违反最小必要原则。解决：入口处统一哈希，盐值走环境变量，分析全程只用哈希值。这件事没有后悔药，必须在第一版就做对，后期补做要重刷全量历史数据。

6. 验证分析结果是否可信：三个可复现的校验技巧

系统跑通不等于结果可信。我一般用三个校验动作来确认数据没骗我。第一个是「总量对齐」：把事件表的记录数按天汇总，和采集端日志的行数对比，差异超过 1% 就要查是哪段清洗逻辑丢多了。第二个是「抽样回放」：随机抽 20 个 session，把原始事件按时间打印出来，人工看会话切分是否符合直觉，这一步能抓出阈值设错的问题。第三个是「指标交叉验证」：用两套独立实现算同一个完播率，比如一套用 Pandas，一套用 SQL，结果对不上就说明有一边的口径理解错了。

def validate_totals(raw_path: str, df: pd.DataFrame): raw_lines = sum(1 for _ in open(raw_path, encoding="utf-8")) cleaned = len(df) drop_rate = 1 - cleaned / raw_lines print(f"raw={raw_lines}, cleaned={cleaned}, drop_rate={drop_rate:.2%}") # 丢弃率超过 5% 需要人工确认清洗规则是否过严 assert drop_rate < 0.05, "drop rate too high, check clean_event rules"

这段校验代码放在每次批量任务末尾，丢弃率超过 5% 直接让任务失败，而不是静默通过。参数上，5% 这个阈值对小数据集可以放宽到 10%，但生产环境建议收紧到 2%。

最后一个习惯：任何指标上线前，我都会先在一个已知答案的小数据集上跑一遍，比如手工构造 100 条事件、预期完播率 40%，跑出来对不上就先修代码再谈业务。这个习惯帮我省掉了无数次「指标看着合理但其实是错的」的尴尬。希望帮到你。

本文还有配套的精品资源，点击获取