简介：这是一份面向大数据分析与电商方向毕业设计的学习资源包，围绕淘宝用户行为数据展开，时间跨度为2017年11月25日至12月3日，记录量超过1亿条。内容涵盖数据导入、清洗、异常值处理，并借助Hive完成用户流量与购物情况、行为转换率、行为习惯、RFM高价值用户识别及商品维度分析，最终以可视化方式呈现结论，适合作为毕设参考或数据分析入门实践案例。资源包共20个文件，以py脚本、json结果数据、md说明文档为主，另含csv数据集、sh安装脚本与html页面，压缩包约16.37MB，结构清晰便于按模块查阅。目前已有88人学习下载。读者可获得可运行源码、论文材料及完整分析流程，既能复用代码快速复现实验，也能借鉴RFM建模与漏斗分析思路，为电商场景下的数据挖掘与可视化项目提供参考。
1. 电商用户行为分析：从埋点到复购预测，一套能跑通的源码骨架
电商用户行为分析这件事，真正难的不是模型，而是数据从哪来、怎么串成一条链。我见过太多团队把埋点日志导出来，字段对不上、时间戳时区错乱、user_id 和 session_id 混用，最后分析出来的转化率连业务方都不信。这个标题指向的是一套可运行源码，意味着它不只是讲概念，而是把「采集→清洗→会话化→指标计算→可视化/预测」整条链路用代码串起来。适合两类人：一是刚接手用户行为数据、需要快速搭出分析框架的初中级数据开发；二是想验证自己埋点方案是否合理的后端或前端工程师。下面我按实际落地顺序拆开讲，代码能直接抄，参数能直接改。
2. 行为数据从哪来：埋点字段设计与最小采集链路
2.1 先定事件模型，再谈分析
电商用户行为分析的第一道坎是事件定义。常见做法是采用「事件 + 属性」的扁平结构，每个事件至少包含：event_id、event_type、user_id、session_id、timestamp、page_url、item_id、referrer。其中event_type建议收敛到 6～8 个核心动作：page_view、product_view、add_to_cart、remove_from_cart、checkout_start、order_submit、payment_success。不要一上来就几十个事件，后期清洗会非常痛苦。
我一般会要求前端在埋点时带上event_time的毫秒时间戳，同时后端在接收时补一个server_time。两者差值超过 5 分钟就标记为异常，后续分析时过滤掉。这个细节能解决大部分「用户行为时间倒流」的玄学问题。
2.2 用 Python 模拟一份可运行的埋点日志
没有真实数据时，先造一份结构一致的模拟数据，保证后续代码能跑通。下面这段代码生成 200 个用户、每人 5～30 条行为记录，字段与真实埋点对齐。
import random import time import pandas as pd from datetime import datetime, timedelta random.seed(42) EVENT_TYPES = [ "page_view", "product_view", "add_to_cart", "remove_from_cart", "checkout_start", "order_submit", "payment_success" ] # 权重模拟真实转化漏斗：浏览多，支付少 EVENT_WEIGHTS = [40, 30, 12, 5, 6, 4, 3] def gen_events(user_id, base_time): n = random.randint(5, 30) events = [] session_id = f"s_{user_id}_{int(base_time.timestamp())}" for i in range(n): event_type = random.choices(EVENT_TYPES, weights=EVENT_WEIGHTS, k=1)[0] # 每条行为间隔 3 秒到 10 分钟 base_time += timedelta(seconds=random.randint(3, 600)) events.append({ "event_id": f"e_{user_id}_{i}", "event_type": event_type, "user_id": user_id, "session_id": session_id, "event_time": int(base_time.timestamp() * 1000), "item_id": random.randint(1000, 9999) if "product" in event_type or "cart" in event_type else None, "page_url": f"/page/{random.randint(1, 20)}", "referrer": random.choice(["search", "recommend", "ad", "direct"]) }) return events all_events = [] start = datetime(2024, 6, 1, 8, 0, 0) for uid in range(1, 201): all_events.extend(gen_events(uid, start + timedelta(minutes=random.randint(0, 1440)))) df = pd.DataFrame(all_events) df.to_csv("mock_events.csv", index=False) print(df.head()) print(f"总记录数: {len(df)}, 用户数: {df['user_id'].nunique()}")逻辑说明：EVENT_WEIGHTS控制漏斗各环节的比例，避免生成的数据全是支付成功这种失真情况。session_id这里简单用用户 ID 加时间戳生成，真实场景中通常由前端在会话开始时生成并保持 30 分钟不变。event_time用毫秒时间戳，方便后续做会话切割。
参数说明：random.seed(42)保证每次生成的数据一致，便于调试。timedelta(seconds=random.randint(3, 600))模拟用户操作间隔，真实场景中这个间隔分布更接近长尾，可以换成指数分布。
2.3 采集链路的三个必调参数
如果走真实采集，Kafka 或类似消息队列的配置里，这三个参数直接影响数据完整性：
| 参数 | 建议值 | 作用 |
|---|---|---|
acks | all | 确保消息被所有副本确认，防止丢数据 |
retries | 5 | 网络抖动时自动重试 |
linger.ms | 20 | 攒批发送，提高吞吐，但会增加延迟 |
提示：
linger.ms设太大，实时看板会延迟明显；设 0 则吞吐下降。20 毫秒是多数电商场景的平衡点。
3. 会话切割与漏斗计算：把散点行为串成用户路径
3.1 会话切割的两种策略与选择理由
用户行为分析里，会话切割决定了后续所有指标的口径。常见两种做法：一是固定时间窗口，比如 30 分钟无操作则切新会话；二是按业务动作切，比如从page_view到payment_success算一次完整会话。我一般用第一种做基础会话，第二种做转化漏斗。
固定窗口的代码实现如下，核心是计算相邻两条记录的时间差，超过阈值就标记为新会话。
import pandas as pd df = pd.read_csv("mock_events.csv") df["event_time"] = pd.to_datetime(df["event_time"], unit="ms") df = df.sort_values(["user_id", "event_time"]).reset_index(drop=True) # 计算同一用户相邻行为的时间差（秒） df["time_diff"] = df.groupby("user_id")["event_time"].diff().dt.total_seconds() # 超过 1800 秒（30 分钟）或首条记录，标记为新会话 df["new_session"] = (df["time_diff"].isna()) | (df["time_diff"] > 1800) df["session_seq"] = df.groupby("user_id")["new_session"].cumsum() df["real_session_id"] = df["user_id"].astype(str) + "_" + df["session_seq"].astype(str) print(df[["user_id", "event_time", "event_type", "real_session_id"]].head(10)) print(f"切割后会话数: {df['real_session_id'].nunique()}")逻辑说明：groupby("user_id")["event_time"].diff()得到同一用户相邻行为的时间差，cumsum()对布尔值累加，每遇到一次True就生成一个新会话编号。这样即使用户跨天访问，只要间隔超过 30 分钟就会正确切开。
参数说明：1800 秒是行业常用值，但不同品类差异大。生鲜电商可能 15 分钟就切，长视频平台可能 60 分钟。建议先用 30 分钟跑一版，再看会话内行为数量的分布，如果大量会话只有 1 条记录，说明阈值偏小。
3.2 漏斗计算的通用函数
漏斗是电商用户行为分析的核心产出。下面这个函数接受事件序列和用户 ID，输出各环节的转化率和流失率。
def funnel_analysis(df, funnel_steps, user_col="user_id", event_col="event_type"): """ funnel_steps: 有序列表，如 ['page_view', 'product_view', 'add_to_cart', 'order_submit', 'payment_success'] """ result = [] for i, step in enumerate(funnel_steps): if i == 0: users = set(df[df[event_col] == step][user_col].unique()) else: prev_users = set(result[-1]["users"]) curr_users = set(df[df[event_col] == step][user_col].unique()) users = prev_users & curr_users # 必须经过上一步的用户才算 result.append({ "step": step, "users": users, "user_count": len(users) }) # 计算转化率 for i in range(1, len(result)): result[i]["conversion_rate"] = round(result[i]["user_count"] / result[i-1]["user_count"], 4) result[i]["drop_rate"] = round(1 - result[i]["conversion_rate"], 4) result[0]["conversion_rate"] = 1.0 result[0]["drop_rate"] = 0.0 return pd.DataFrame([{k: v for k, v in r.items() if k != "users"} for r in result]) steps = ["page_view", "product_view", "add_to_cart", "order_submit", "payment_success"] funnel_df = funnel_analysis(df, steps) print(funnel_df)逻辑说明：每一步都取上一步用户集合与当前步骤用户集合的交集，保证漏斗是「同一批用户」的连续转化，而不是各步骤独立用户数的简单相除。这是很多新手容易翻车的地方——直接用各事件的总用户数算转化率，会得到虚高的结果。
参数说明：funnel_steps必须按业务顺序排列。如果某一步骤在数据中不存在，user_count会为 0，后续转化率计算会报错，建议加一层空值保护。
3.3 用 SQL 做同样的漏斗计算
如果数据在数仓里，用 SQL 更直接。下面这段 SQL 假设事件表为user_events，按用户和会话聚合。
WITH funnel AS ( SELECT user_id, MAX(CASE WHEN event_type = 'page_view' THEN 1 ELSE 0 END) AS step1, MAX(CASE WHEN event_type = 'product_view' THEN 1 ELSE 0 END) AS step2, MAX(CASE WHEN event_type = 'add_to_cart' THEN 1 ELSE 0 END) AS step3, MAX(CASE WHEN event_type = 'order_submit' THEN 1 ELSE 0 END) AS step4, MAX(CASE WHEN event_type = 'payment_success' THEN 1 ELSE 0 END) AS step5 FROM user_events WHERE event_time >= '2024-06-01' AND event_time < '2024-06-02' GROUP BY user_id ) SELECT COUNT(*) AS total_users, SUM(step1) AS page_view_users, SUM(step2) AS product_view_users, SUM(step3) AS add_to_cart_users, SUM(step4) AS order_submit_users, SUM(step5) AS payment_users, ROUND(SUM(step2) * 1.0 / SUM(step1), 4) AS view_to_product_rate, ROUND(SUM(step3) * 1.0 / SUM(step2), 4) AS product_to_cart_rate, ROUND(SUM(step5) * 1.0 / SUM(step1), 4) AS overall_conversion_rate FROM funnel;逻辑说明：用MAX(CASE WHEN ...)把每个用户是否触发某事件打平成一行，再聚合统计。这种写法比多次 JOIN 更高效，也更容易加时间过滤。
参数说明：时间范围建议按天分区，避免全表扫描。如果数据量大，可以在WHERE里加上event_type IN (...)先过滤事件类型。
4. 复购预测与用户分层：从行为数据到可执行标签
4.1 RFM 模型的代码实现与分箱技巧
RFM（最近一次购买、购买频率、购买金额）是电商用户分层最常用的方法。但很多教程只给公式，不给分箱逻辑，导致结果没法用。下面这段代码在模拟数据上计算 RFM 并打标签。
import pandas as pd import numpy as np # 假设已有订单表 orders: user_id, order_time, amount orders = df[df["event_type"] == "payment_success"].copy() orders["order_time"] = orders["event_time"] orders["amount"] = np.random.uniform(20, 500, len(orders)).round(2) snapshot_date = orders["order_time"].max() + pd.Timedelta(days=1) rfm = orders.groupby("user_id").agg( recency=("order_time", lambda x: (snapshot_date - x.max()).days), frequency=("order_time", "count"), monetary=("amount", "sum") ).reset_index() # 分箱：按分位数切成 4 档，避免手动设阈值 rfm["r_score"] = pd.qcut(rfm["recency"], 4, labels=[4, 3, 2, 1]).astype(int) rfm["f_score"] = pd.qcut(rfm["frequency"].rank(method="first"), 4, labels=[1, 2, 3, 4]).astype(int) rfm["m_score"] = pd.qcut(rfm["monetary"], 4, labels=[1, 2, 3, 4]).astype(int) rfm["rfm_segment"] = rfm["r_score"].astype(str) + rfm["f_score"].astype(str) + rfm["m_score"].astype(str) print(rfm.head()) print(rfm["rfm_segment"].value_counts().head(10))逻辑说明：qcut按分位数分箱，保证每档用户数量接近，避免某些分段只有几个人。recency越小越好，所以标签反过来给 4 分。frequency和monetary越大越好，正常给分。
参数说明：分箱数 4 是经验值，用户量少时可以改成 3。如果数据分布极度偏斜，qcut会报边界重复错误，可以改用rank(method="first")先打散再切。
4.2 用逻辑回归做复购预测的最小流程
有了 RFM 标签，可以进一步做二分类预测：未来 30 天是否复购。下面用scikit-learn跑一个最小可用流程。
from sklearn.model_selection import train_test_split from sklearn.linear_model import LogisticRegression from sklearn.metrics import classification_report from sklearn.preprocessing import StandardScaler # 构造标签：以最后一次购买为基准，之前有购买且之后 30 天内有购买为 1 orders_sorted = orders.sort_values(["user_id", "order_time"]) last_order = orders_sorted.groupby("user_id")["order_time"].max().reset_index() last_order.columns = ["user_id", "last_order_time"] # 简化：用 frequency 和 monetary 作为特征，recency 作为标签构造依据 features = rfm[["frequency", "monetary", "r_score"]].copy() features["label"] = (rfm["recency"] < 30).astype(int) # 30 天内有过购买视为复购 X = features[["frequency", "monetary", "r_score"]] y = features["label"] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42) scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) X_test_scaled = scaler.transform(X_test) model = LogisticRegression(class_weight="balanced", max_iter=1000) model.fit(X_train_scaled, y_train) y_pred = model.predict(X_test_scaled) print(classification_report(y_test, y_pred)) print("特征系数:", dict(zip(X.columns, model.coef_[0].round(4))))逻辑说明：class_weight="balanced"处理正负样本不均衡，复购用户通常远少于非复购用户。StandardScaler对逻辑回归很重要，否则monetary的量级会压过frequency。
参数说明：max_iter=1000防止收敛警告。如果特征更多，建议换成RandomForestClassifier或XGBoost，但逻辑回归的可解释性在业务沟通中更占优势。
4.3 用户分层的业务映射表
算出 RFM 分数后，需要映射成运营能看懂的分层。下面这张表是我常用的映射规则：
| RFM 分数组合 | 分层名称 | 运营动作 |
|---|---|---|
| R≥3, F≥3, M≥3 | 高价值用户 | 专属客服、新品优先体验 |
| R≥3, F≤2, M≤2 | 潜力用户 | 首单优惠、引导复购 |
| R≤2, F≥3, M≥3 | 流失预警 | 召回券、满意度回访 |
| R≤2, F≤2, M≤2 | 低价值用户 | 低成本触达，不投过多资源 |
注意：分层规则不是固定的，不同品类差异很大。建议先用历史数据回测，看每层用户的真实复购率是否单调递减，如果不是，说明分箱或映射有问题。
5. 避坑与排查：电商行为分析里最容易翻车的 5 个点
5.1 时间戳时区不统一导致漏斗倒挂
现象：漏斗计算时，payment_success的用户数比order_submit还多，转化率超过 100%。
原因：前端埋点用本地时间，后端用 UTC，两者混在一起后排序错乱，同一用户的行为顺序被打乱。
解决：统一用 UTC 毫秒时间戳存储，展示时再转本地时区。在数据接入层加一个校验：同一session_id内，后一条记录的event_time必须大于前一条，否则标记异常并告警。
5.2 session_id 复用导致会话膨胀
现象：一个用户只有一个session_id，但行为跨度好几天，会话内行为数几百条。
原因：前端生成session_id后没有在超时后重置，或者后端直接把user_id当session_id用。
解决：会话切割不要依赖前端传来的session_id，而是用user_id+ 时间差在分析层重新切。前端session_id只作为辅助字段。
5.3 漏斗步骤顺序写错导致转化率虚高
现象：add_to_cart到order_submit的转化率有 80%，但业务方说实际只有 20%。
原因：代码里没有做用户集合交集，而是各步骤独立统计用户数。比如 100 人加购，200 人提交订单（其中只有 20 人加购过），直接相除得到 200%。
解决：严格按 3.2 节的交集逻辑计算，每一步都取上一步用户集合与当前步骤的交集。
5.4 埋点丢失导致新用户被误判为流失
现象：某天新增用户数骤降，但前端监控显示页面访问正常。
原因：埋点上报接口在高峰期超时，部分page_view事件丢失，导致这些用户没有进入分析表。
解决：在采集层加本地缓存和重试机制，前端用navigator.sendBeacon或类似方式保证页面关闭前发出。分析层对「有订单但无浏览记录」的用户做补偿标记，不直接算流失。
5.5 复购标签定义与业务口径不一致
现象：模型预测准确率很高，但运营用起来发现召回的用户大部分本来就会复购。
原因：标签定义用了「30 天内有购买」，但业务方关心的其实是「自然复购」，即没有发券情况下的复购。两者口径不同。
解决：标签定义必须和业务方对齐，最好拉上运营一起确认。可以在特征里加入「是否收到优惠券」字段，把自然复购和营销复购分开建模。
6. 把分析结果推到线上：一个可复用的调度与验证习惯
前面几章跑通了从埋点到复购预测的链路，但真正让这套东西产生价值的，是把它变成每天自动跑的任务，并且有一套验证机制。我一般会用一个轻量调度脚本，配合数据质量检查，确保每天产出的漏斗和分层结果可信。
import schedule import time import pandas as pd from datetime import datetime, timedelta def daily_job(): print(f"[{datetime.now()}] 开始执行每日用户行为分析...") # 1. 读取昨日数据 yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") # 实际项目中这里换成从数仓或数据湖读取 df = pd.read_csv("mock_events.csv") df["event_time"] = pd.to_datetime(df["event_time"], unit="ms") # 2. 数据质量检查：空值率、时间范围、事件类型覆盖 assert df["user_id"].notna().all(), "user_id 存在空值" assert df["event_time"].between("2024-06-01", "2024-06-30").all(), "时间范围异常" event_coverage = df["event_type"].nunique() assert event_coverage >= 5, f"事件类型仅覆盖 {event_coverage} 种，可能埋点丢失" # 3. 会话切割与漏斗 df = df.sort_values(["user_id", "event_time"]).reset_index(drop=True) df["time_diff"] = df.groupby("user_id")["event_time"].diff().dt.total_seconds() df["new_session"] = (df["time_diff"].isna()) | (df["time_diff"] > 1800) df["session_seq"] = df.groupby("user_id")["new_session"].cumsum() df["real_session_id"] = df["user_id"].astype(str) + "_" + df["session_seq"].astype(str) steps = ["page_view", "product_view", "add_to_cart", "order_submit", "payment_success"] funnel_df = funnel_analysis(df, steps) print(funnel_df) # 4. 结果落盘，供看板或下游模型使用 funnel_df.to_csv(f"funnel_{yesterday}.csv", index=False) print(f"[{datetime.now()}] 分析完成，结果已保存") # 每天凌晨 2 点执行 schedule.every().day.at("02:00").do(daily_job) if __name__ == "__main__": daily_job() # 先跑一次验证 # 正式环境取消下面注释 # while True: # schedule.run_pending() # time.sleep(60)这段调度脚本的关键不在schedule本身，而在第 2 步的数据质量检查。我踩过的坑是：某天上游埋点版本升级，event_type从下划线命名改成驼峰，导致漏斗里add_to_cart直接消失，但脚本没报错，看板上的转化率却异常升高。后来加了event_coverage断言，事件类型少于 5 种就中断任务并告警，才避免类似问题。
验证方法上，我习惯每天对比三个数：昨日page_view用户数与前 7 天均值的偏差是否超过 20%、漏斗各环节转化率是否在历史 ±3 个标准差内、RFM 分层中高价值用户占比是否突变。任何一个异常，先查埋点版本，再查数据接入延迟，最后才怀疑模型。
参数方面，schedule.every().day.at("02:00")适合离线数仓，如果数据源是实时流，建议改成每 15 分钟滚动窗口，用 Flink 或 Spark Structured Streaming 做增量计算。assert语句在生产环境建议换成日志告警，避免任务直接崩溃。
这套东西值不值得做？如果你手上已经有埋点数据，但每次分析都要手动导表、写 SQL、对口径，那把它脚本化、调度化、加上质量检查，投入大概两三天，之后每天省下的沟通和排查时间远超这个成本。如果还没有埋点，建议先把事件模型定清楚，再谈分析。
我自己的习惯是：任何分析结果在发给业务方之前，先自己用另一套逻辑交叉验证一遍。比如漏斗用 Python 算完，再用 SQL 算一次，两个结果对不上，一定是某一步的口径有问题。这个习惯帮我省掉了至少五次「数据打脸」的尴尬。希望帮到你。
本文还有配套的精品资源，点击获取
