跳到正文
本地阅读 · 用户与订单分析

用户与订单分析

计算机基础 · 综合项目

前置知识：性能、内存及 SQL 对照 · 留存、漏斗、分群及实验指标

本章目录

本章学习目标

  • 从多张表产出统一业务指标
  • 正确处理留存、漏斗及实验分母
  • 用守恒和粒度检查保证结果可解释
前置知识：性能、内存及 SQL 对照 · 留存、漏斗、分群及实验指标

任务书与口径

你是一名分析师，需要用 users、orders、events 和 assignments 形成小型业务报告，回答注册用户中有多少人付款、每日净金额多少、次日有多少用户回来、用户是否依次完成浏览加购购买，以及实验两组七日购买事件转化如何。报告交付五张表和一段结论，每项指标要注明分母、状态、时间范围和未知记录处理。观察截止从 data/dictionary.json 的 observation_end 读取，固定为二零二六年一月三十一日末尾，与 SQL 课程一致；所有事件、订单和分配都先排除截止后的记录。教学时间精确到秒，代码将完整日末截止转换成次日零点作为不包含的右边界。

用户表是每用户一行，订单是每订单一行，事件是每次行为一行，实验分配是每用户一行。净金额仅纳入 paid/refunded，按金额减退款计算；cancelled 不计入支付。空 user_id 的订单参与总账和每日金额，但无法归因到注册用户，要单独报告。退款订单不能直接删除，因为删除会同时丢失已支付和退款事实。

起始代码与分解顺序

起始时读取四表，分别检查主键与日期，再创建已支付订单副本。先把订单聚合成每用户一行，再左连接全部注册用户；这一步保留无订单用户作为转化分母。不要把事件和订单都按 user_id 直接连接，两张事实表各自多行会形成乘积，金额与事件数一起放大。

起始代码：

from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
users=load("users")
orders=load("orders")
events=load("events")
assigned=load("assignments")
assert users["user_id"].is_unique
# TODO: 每个表独立检查；按订单状态过滤；先聚合后连接。

把一个复杂报告拆成五条独立流水线，共用经过审计的输入，最后按各自粒度输出。脚本中的金额以整数分表示，网页结论如换成元须除以一百并清楚标注单位。

留存与漏斗逐步推导

次日留存分母是已经观察完整次日的注册用户：注册日期加一天不得晚于数据字典声明的完整观察截止日，故截至一月三十一日时，最晚成熟注册日为一月三十日。分子是该次日在任何事件中出现过的不同用户。先把事件按用户和日去重，再用用户键与目标日期精确连接；同一天十次访问也只计一次。最大事件时间只表明最后一次实际活动，不能证明那一天或之后的日志覆盖完整；没有事件也不能单凭数据区分真实无活动与采集故障。本课依赖数据字典的覆盖合同，真实项目应检查采集监控和分区完整性后确定截止，不能用 max(event_time) 推断。

漏斗定义为同一会话内按时间依次出现 view、cart、purchase。按事件时间和 event_id 排序，然后用状态机从浏览推进到加购再推进到购买；相同时间的排序是本课教学约定。仅分别数有浏览和有购买的用户，不保证先后顺序，也不能证明同一会话完成。最后对满足各阶段的用户去重，结果应逐级不增加。

实验指标与 SQL 对照

实验的七日转化分母为已经完整观察七天的全部分配用户，分子为分配时刻起七天内至少出现一次 purchase 事件的用户，时间范围左闭右开。分配时刻加七天不得晚于完整观察窗的不包含右边界；尚未成熟用户另列为待观察，不能当作未转化。满足成熟条件但没有任何事件的分配用户仍留在分母，分配前购买不计。本课程三十位分配用户均已成熟。购买事件是行为转化，并不自动等于支付订单，本报告分别命名，不把两种定义混用。

这里展示描述性组间差异，不据此断言因果效果或显著性。需要先检查随机化、样本分配比例、观察窗口完整性和多重比较，再进行推断。SQL 对照应在 MySQL 用相同状态和时间条件运行，COUNT(DISTINCT user_id) 与 pandas nunique 对齐，空用户归因和连接规则也必须一致；SQL 业务章的分配后至月末转化属于另一种观察窗，不能直接与本项目固定七天指标混比。Python 的结果不能代替 SQL 实际验证。

验收标准与参考解答

完整答案是本章运行脚本。验收一：用户汇总行数等于观察截止前注册用户数，无订单用户计数为零；用户归因净额加匿名净额等于观察窗内总已支付净额。验收二：每日净额合计与同窗订单净额完全一致。验收三：留存率在零与一之间，分母只包含完整观察用户；漏斗用户集合逐级包含。验收四：两组分配人数总和等于满足完整七天观察条件的分配人数，转化分子不超过分母。先按权威截止排除未来记录，再计算指标。

结论先报告数字，再解释限制。例如用户支付转化和行为购买转化口径不同，不能直接判定谁错。边界包括零订单、全匿名订单、用户次日尚未完整观察、购买先于浏览、跨会话完成，以及七日窗口右边界事件。不要为了通过断言删掉难解释的用户，应该定位原因并更新合同。

版本与复现约定

本课程 pandas 示例以 2.3.3 为验证基线，页面运行记录给出实际版本。pandas 3.x 默认使用 Copy-on-Write：从表中取出的对象发生修改时，不据此假定原表同步变化。跨版本都应使用一次完整的 loc 赋值表达式；需要独立工作表时显式 copy。阅读网上的旧示例时，先核对版本、输入类型和索引，再判断结果能否复现。

完整示例与实际输出

下载 project-business.py ↓
from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
import json
from decimal import Decimal
dictionary=json.loads((ROOT/"data"/"dictionary.json").read_text(encoding="utf-8"))
observation_end=pd.Timestamp(dictionary["observation_end"])
# 教学时间戳精确到秒；合同保证覆盖截至该日末尾，而非以最后一条事件推断覆盖。
cutoff=observation_end.normalize()
observation_exclusive=cutoff+pd.Timedelta(days=1)
assert observation_end == observation_exclusive-pd.Timedelta(seconds=1)
users=load("users").copy()
orders=load("orders").copy()
events=load("events").copy()
assigned=load("assignments").copy()
assert users["user_id"].is_unique and users["user_id"].notna().all()
assert orders["order_id"].is_unique and events["event_id"].is_unique
assert assigned["user_id"].is_unique
users["signup_date"]=pd.to_datetime(users["signup_date"]).dt.normalize()
orders["order_date"]=pd.to_datetime(orders["order_date"])
events["event_time"]=pd.to_datetime(events["event_time"])
assigned["assigned_at"]=pd.to_datetime(assigned["assigned_at"])
users=users.loc[users["signup_date"].le(cutoff)].copy()
orders=orders.loc[orders["order_date"].lt(observation_exclusive)].copy()
events=events.loc[events["event_time"].lt(observation_exclusive)].copy()
assigned=assigned.loc[assigned["assigned_at"].lt(observation_exclusive)].copy()
paid=orders.loc[orders["status"].isin(["paid","refunded"])].copy()
paid["net_cents"]=[int((Decimal(str(a))-Decimal(str(r)))*100) for a,r in zip(paid["amount"],paid["refund_amount"])]
known=paid.loc[paid["user_id"].notna()].copy()
summary=known.groupby("user_id").agg(paid_orders=("order_id","size"),net_cents=("net_cents","sum")).reset_index()
per_user=users.merge(summary,on="user_id",how="left",validate="one_to_one")
per_user[["paid_orders","net_cents"]]=per_user[["paid_orders","net_cents"]].fillna(0).astype("int64")
assert len(per_user)==len(users)
assert per_user["net_cents"].sum()==known["net_cents"].sum()
daily=paid.assign(day=paid["order_date"].dt.normalize()).groupby("day").agg(paid_orders=("order_id","size"),net_cents=("net_cents","sum"))
assert daily["net_cents"].sum()==paid["net_cents"].sum()
# 次日留存：注册后第一个自然日有任何事件；成熟分母使用权威完整观察截止。
active=events.assign(day=events["event_time"].dt.normalize())[["user_id","day"]].drop_duplicates()
eligible=users.loc[users["signup_date"]+pd.Timedelta(days=1)<=cutoff].copy()
eligible["target_day"]=eligible["signup_date"]+pd.Timedelta(days=1)
retained=eligible.merge(active,left_on=["user_id","target_day"],right_on=["user_id","day"],how="left",validate="one_to_one",indicator=True)
retained["retained_d1"]=retained["_merge"].eq("both")
retention=retained.groupby("signup_date").agg(eligible_users=("user_id","size"),retained_users=("retained_d1","sum"))
retention["rate"]=retention["retained_users"]/retention["eligible_users"]
assert retention["rate"].between(0,1).all()
# 会话漏斗：按事件时间与事件编号排序，用状态机保证顺序。
stages=[set(),set(),set()]
ordered=events.sort_values(["user_id","session_id","event_time","event_id"])
for (uid,sid),g in ordered.groupby(["user_id","session_id"]):
    state=0
    for kind in g["event_type"]:
        if state==0 and kind=="view":state=1
        elif state==1 and kind=="cart":state=2
        elif state==2 and kind=="purchase":state=3
    for i in range(state):stages[i].add(uid)
funnel=pd.DataFrame({"stage":["view","ordered_cart","ordered_purchase"],"users":[len(s) for s in stages]})
assert stages[2] <= stages[1] <= stages[0]
# 固定七天转化：仅纳入七天观察窗完整的分配用户；这些用户无活动也保留在分母。
mature_7d=assigned["assigned_at"]+pd.Timedelta(days=7)<=observation_exclusive
pending_assigned=assigned.loc[~mature_7d].copy()
assigned=assigned.loc[mature_7d].copy()
candidate=events.merge(assigned,on="user_id",how="inner",validate="many_to_one")
converted=candidate.loc[candidate["event_type"].eq("purchase") & candidate["event_time"].ge(candidate["assigned_at"]) & candidate["event_time"].lt(candidate["assigned_at"]+pd.Timedelta(days=7)),"user_id"].unique()
ab=assigned.copy()
ab["converted_7d"]=ab["user_id"].isin(converted)
ab_result=ab.groupby("variant").agg(assigned_users=("user_id","size"),conversions=("converted_7d","sum"))
ab_result["rate"]=ab_result["conversions"]/ab_result["assigned_users"]
assert ab_result["assigned_users"].sum()==len(assigned)
for name,frame in [("users",per_user),("daily",daily.reset_index()),("retention",retention.reset_index()),("funnel",funnel),("experiment",ab_result.reset_index())]:
    frame.to_csv(OUT/("project_business_"+name+".csv"),index=False)
print("注册用户 / 付费用户:",len(users),int(per_user["paid_orders"].gt(0).sum()))
print("已支付净额分 / 匿名已支付净额分:",int(paid["net_cents"].sum()),int(paid.loc[paid["user_id"].isna(),"net_cents"].sum()))
print("次日留存:");print(retention.to_string())
print("有序会话漏斗:");print(funnel.to_string(index=False))
print("七日行为转化（描述性统计）:");print(ab_result.to_string())
print("权威完整观察截止:",observation_end)
print("七日观察成熟 / 待观察分配用户:",len(assigned),len(pending_assigned))

实际运行通过CPU Python execution.

注册用户 / 付费用户: 30 24
已支付净额分 / 匿名已支付净额分: 564075 1250
次日留存:
             eligible_users  retained_users  rate
signup_date                                      
2026-01-01                2               2   1.0
2026-01-02                2               1   0.5
2026-01-03                2               2   1.0
2026-01-04                2               1   0.5
2026-01-05                2               2   1.0
2026-01-06                2               1   0.5
2026-01-07                2               2   1.0
2026-01-08                2               1   0.5
2026-01-09                2               2   1.0
2026-01-10                2               1   0.5
2026-01-11                2               2   1.0
2026-01-12                2               1   0.5
2026-01-13                2               2   1.0
2026-01-14                2               1   0.5
2026-01-15                2               2   1.0
有序会话漏斗:
           stage  users
            view     30
    ordered_cart     20
ordered_purchase     10
七日行为转化（描述性统计）:
         assigned_users  conversions      rate
variant                                       
A                    15            5  0.333333
B                    15            5  0.333333
权威完整观察截止: 2026-01-31 23:59:59
七日观察成熟 / 待观察分配用户: 30 0

五道自测：先回答，再核对

01 · 为何先汇总订单再连用户？

查看提示

两边最终粒度。

查看参考答案

避免事实表乘积，得到每用户一行并保留无订单用户。

02 · 匿名支付订单参与什么？

查看提示

总账与归因不同。

查看参考答案

参与总额和每日净额，不能归因注册用户，单列报告。

03 · 次日尚未完整观察的用户怎么办？

查看提示

不能把未知视为未留存。

查看参考答案

从本次次日留存分母排除并说明观察截止。

04 · 漏斗为什么用状态机？

查看提示

事件顺序。

查看参考答案

保证同会话浏览后加购再购买，简单独立计数无法验证顺序。

05 · 实验转化分母可只用活跃用户吗？

查看提示

随机分配对象。

查看参考答案

不可以，应保留全部分配用户，避免按结果选择样本。

两项动手任务

任务 1 · 完成五张业务表并写出各指标分母和限制。

提示

按任务书实现独立流水线。

完整参考解答与检查方式
from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
import json
from decimal import Decimal
dictionary=json.loads((ROOT/"data"/"dictionary.json").read_text(encoding="utf-8"))
observation_end=pd.Timestamp(dictionary["observation_end"])
# 教学时间戳精确到秒；合同保证覆盖截至该日末尾，而非以最后一条事件推断覆盖。
cutoff=observation_end.normalize()
observation_exclusive=cutoff+pd.Timedelta(days=1)
assert observation_end == observation_exclusive-pd.Timedelta(seconds=1)
users=load("users").copy()
orders=load("orders").copy()
events=load("events").copy()
assigned=load("assignments").copy()
assert users["user_id"].is_unique and users["user_id"].notna().all()
assert orders["order_id"].is_unique and events["event_id"].is_unique
assert assigned["user_id"].is_unique
users["signup_date"]=pd.to_datetime(users["signup_date"]).dt.normalize()
orders["order_date"]=pd.to_datetime(orders["order_date"])
events["event_time"]=pd.to_datetime(events["event_time"])
assigned["assigned_at"]=pd.to_datetime(assigned["assigned_at"])
users=users.loc[users["signup_date"].le(cutoff)].copy()
orders=orders.loc[orders["order_date"].lt(observation_exclusive)].copy()
events=events.loc[events["event_time"].lt(observation_exclusive)].copy()
assigned=assigned.loc[assigned["assigned_at"].lt(observation_exclusive)].copy()
paid=orders.loc[orders["status"].isin(["paid","refunded"])].copy()
paid["net_cents"]=[int((Decimal(str(a))-Decimal(str(r)))*100) for a,r in zip(paid["amount"],paid["refund_amount"])]
known=paid.loc[paid["user_id"].notna()].copy()
summary=known.groupby("user_id").agg(paid_orders=("order_id","size"),net_cents=("net_cents","sum")).reset_index()
per_user=users.merge(summary,on="user_id",how="left",validate="one_to_one")
per_user[["paid_orders","net_cents"]]=per_user[["paid_orders","net_cents"]].fillna(0).astype("int64")
assert len(per_user)==len(users)
assert per_user["net_cents"].sum()==known["net_cents"].sum()
daily=paid.assign(day=paid["order_date"].dt.normalize()).groupby("day").agg(paid_orders=("order_id","size"),net_cents=("net_cents","sum"))
assert daily["net_cents"].sum()==paid["net_cents"].sum()
# 次日留存：注册后第一个自然日有任何事件；成熟分母使用权威完整观察截止。
active=events.assign(day=events["event_time"].dt.normalize())[["user_id","day"]].drop_duplicates()
eligible=users.loc[users["signup_date"]+pd.Timedelta(days=1)<=cutoff].copy()
eligible["target_day"]=eligible["signup_date"]+pd.Timedelta(days=1)
retained=eligible.merge(active,left_on=["user_id","target_day"],right_on=["user_id","day"],how="left",validate="one_to_one",indicator=True)
retained["retained_d1"]=retained["_merge"].eq("both")
retention=retained.groupby("signup_date").agg(eligible_users=("user_id","size"),retained_users=("retained_d1","sum"))
retention["rate"]=retention["retained_users"]/retention["eligible_users"]
assert retention["rate"].between(0,1).all()
# 会话漏斗：按事件时间与事件编号排序，用状态机保证顺序。
stages=[set(),set(),set()]
ordered=events.sort_values(["user_id","session_id","event_time","event_id"])
for (uid,sid),g in ordered.groupby(["user_id","session_id"]):
    state=0
    for kind in g["event_type"]:
        if state==0 and kind=="view":state=1
        elif state==1 and kind=="cart":state=2
        elif state==2 and kind=="purchase":state=3
    for i in range(state):stages[i].add(uid)
funnel=pd.DataFrame({"stage":["view","ordered_cart","ordered_purchase"],"users":[len(s) for s in stages]})
assert stages[2] <= stages[1] <= stages[0]
# 固定七天转化：仅纳入七天观察窗完整的分配用户；这些用户无活动也保留在分母。
mature_7d=assigned["assigned_at"]+pd.Timedelta(days=7)<=observation_exclusive
pending_assigned=assigned.loc[~mature_7d].copy()
assigned=assigned.loc[mature_7d].copy()
candidate=events.merge(assigned,on="user_id",how="inner",validate="many_to_one")
converted=candidate.loc[candidate["event_type"].eq("purchase") & candidate["event_time"].ge(candidate["assigned_at"]) & candidate["event_time"].lt(candidate["assigned_at"]+pd.Timedelta(days=7)),"user_id"].unique()
ab=assigned.copy()
ab["converted_7d"]=ab["user_id"].isin(converted)
ab_result=ab.groupby("variant").agg(assigned_users=("user_id","size"),conversions=("converted_7d","sum"))
ab_result["rate"]=ab_result["conversions"]/ab_result["assigned_users"]
assert ab_result["assigned_users"].sum()==len(assigned)
for name,frame in [("users",per_user),("daily",daily.reset_index()),("retention",retention.reset_index()),("funnel",funnel),("experiment",ab_result.reset_index())]:
    frame.to_csv(OUT/("project_business_"+name+".csv"),index=False)
print("注册用户 / 付费用户:",len(users),int(per_user["paid_orders"].gt(0).sum()))
print("已支付净额分 / 匿名已支付净额分:",int(paid["net_cents"].sum()),int(paid.loc[paid["user_id"].isna(),"net_cents"].sum()))
print("次日留存:");print(retention.to_string())
print("有序会话漏斗:");print(funnel.to_string(index=False))
print("七日行为转化（描述性统计）:");print(ab_result.to_string())
print("权威完整观察截止:",observation_end)
print("七日观察成熟 / 待观察分配用户:",len(assigned),len(pending_assigned))

任务 2 · 检验购买先于浏览不能通过有序漏斗。

提示

同会话时间顺序推进状态。

完整参考解答与检查方式
from pathlib import Path
import pandas as pd
import numpy as np
from _shared import load
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
kinds=["purchase","view","cart"]
state=0
for k in kinds:
    if state==0 and k=="view":state=1
    elif state==1 and k=="cart":state=2
    elif state==2 and k=="purchase":state=3
assert state==2
print("最终状态:",state,"尚未完成有序购买")

官方资料与掌握状态

完成两项任务后，按独立完成程度记录。状态会尽量保存在当前浏览器，可导出备份。

动手任务记录

掌握状态与复习

关掉资料独立完成后记 A；提示后完成记 B；需要重学记 C。

记录尽量保存在当前浏览器，建议定期导出备份。