Jupyter Notebook 写 PRD：可执行口径验算与参数化评审交付

简介：围绕产品需求文档（PRD）撰写与Jupyter Notebook项目实践整理的资料包，面向产品经理、数据科学初学者及需要将需求文档转化为代码实验的开发者，帮助梳理产品目标、功能范围与数据分析流程。压缩包共4个文件，以2个Python脚本和2个Jupyter Notebook为主，整体约2.04MB；脚本侧重数据加载与模型构建，Notebook用于记录交互式分析、分类实验及代码与说明混排的过程。已有337人学习下载，可作为学习热度参考。读者可从中参考PRD的产品概述、问题定义、功能列表、用户故事、性能指标、优先级规划、竞品分析与风险评估等模块的组织方式，也能观察Jupyter环境下的数据导入导出、可视化与版本控制等需求如何落到具体任务，适合用作需求文档模板和Notebook项目结构参考。

1. 用能跑起来的 Notebook 写 PRD：需求文档为什么要带执行能力

评审会上最尴尬的一幕，是产品经理翻到第三页 Word 说“这个留存口径按周算”，数仓那边抬头问“哪个留存、分母是谁、新老用户分不分”，然后全场沉默十分钟。把 PRD 放进 Jupyter Notebook 之后，这种对话会短很多：Markdown 单元写背景、用户故事和边界条件，代码单元直接跑埋点口径、拉一版真实分布、用断言把验收标准钉死。这份资源包就是一套以 Notebook 为载体的需求文档模板，解决的不是“文档写得好不好看”，而是“需求里的每个数字能不能当场算给评审看”。它适合指标定义长期和数仓对不齐、每次评审都要现场解释口径的团队；不适合只想交一份静态 PDF、后续不打算维护的人。下面按拆包、跑通、写需求、接数据、排坑、交付的顺序走一遍。

2. 拆开这个 PRD Notebook 资源包：结构、依赖与首次启动

拿到一个 .ipynb 为主的资源包，最容易犯的错是双击打开就往下跑。Notebook 的执行顺序是隐式的，单元格之间靠内核里的变量传递状态，一旦中间某个单元被跳过，后面所有结论都建立在不完整的数据上。所以先别急着看内容，把结构、依赖和入口摸清楚，这一步花二十分钟，后面能省掉一整轮返工。

2.1 目录结构：每个文件负责哪一段评审

这类资源包常见做法是按“文档层 / 数据层 / 校验层”分开，而不是把所有东西塞进一个几千行的 notebook。我一般会先确认下面这几类文件在不在，缺哪一块，后面就得自己补。

判断一个包是否值得用，看两点：params.py和metrics_dict.csv是不是独立存在。如果所有参数散落在各个单元格里、口径只写在 Markdown 文字中，那这份 PRD 的可维护性基本为零，改一个时间窗要翻二十个单元。独立出来的参数和字典，才是这份资源能长期复用的根基。

2.2 环境依赖与首次启动

Notebook 类资源的依赖通常集中在三块：内核与前端、数据处理、导出。用虚拟环境隔离，别装在系统 Python 里，这是血泪经验——某次为了省事直接全局装，结果另一个项目的 pandas 版本被顶掉，两边的 groupby 行为不一致，排查了一下午。

# 创建独立环境，Python 版本按资源包 requirements 里写的来 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 核心依赖：前端、数据处理、绘图、SQL 直连、导出 pip install jupyterlab pandas numpy matplotlib sqlalchemy ipython-sql # 可选：参数化重跑与自动化校验 pip install papermill nbconvert pytest

装完之后不要直接jupyter lab就完事，先确认内核指向的是当前虚拟环境。Notebook 最常见的翻车是“装在了 A 环境、内核用的是 B 环境”，报错信息还特别含糊，只说模块找不到。

# 在第一个单元格里跑一遍，把环境信息打印出来存档 import sys, platform import pandas as pd import numpy as np import matplotlib print("python :", sys.version.split()[0]) print("exec :", sys.executable) # 确认解释器路径落在 .venv 里 print("pandas :", pd.__version__) print("numpy :", np.__version__) print("plot :", matplotlib.__version__) print("system :", platform.platform())

sys.executable是最关键的一行。它打印出来的路径如果不含.venv，说明内核选错了，后面所有 import 都可能来自另一套环境。把这段输出留在 notebook 顶部，等于给整份文档留了一个环境快照，半年后别人接手时能立刻知道当初跑在什么条件下。

2.3 参数配置：把路径和口径抽到一处

参数集中管理是这份资源能不能被别人接手的分水岭。常见做法是在 notebook 第一个代码单元做一次全局注入，或者单独建params.py然后from params import *。我倾向后者，因为params.py能被脚本、测试、导出流程共用，而 notebook 里的单元只能被自己用。

# params.py —— 所有可变项集中在这里，notebook 只读取不硬编码 from pathlib import Path BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" / "raw" EXPORT_DIR = BASE_DIR / "export" # 分析时间窗：评审前统一改成最近一个完整自然周 START_DATE = "2024-03-04" END_DATE = "2024-03-10" # 口径阈值：留存判定窗口、最小样本量、异常波动容忍度 RETENTION_DAYS = 7 # 次日/7日留存的判定天数 MIN_SAMPLE_SIZE = 500 # 低于这个量级的分组不参与结论 ANOMALY_THRESHOLD = 0.15 # 同比波动超过 15% 标记为待确认 # 数据源连接串走环境变量，别写死在文件里 import os DB_URL = os.getenv("PRD_DB_URL", "sqlite:///" + str(BASE_DIR / "data" / "demo.db")) EXPORT_DIR.mkdir(parents=True, exist_ok=True)

几个参数的取舍值得说清楚。START_DATE和END_DATE用字符串而不是 datetime，是因为 notebook 里做展示和拼 SQL 都更方便，转换放在读取数据那一步统一做。MIN_SAMPLE_SIZE是为了防止小分组得出误导性结论——某个渠道只有三十个用户，留存率波动能从 20% 跳到 60%，这种数字放进评审稿就是给自己埋雷。ANOMALY_THRESHOLD不是业务硬指标，是我自己的检查线，超过就人工看一眼再决定要不要写进结论。

数据库连接串务必走环境变量。资源包里如果带了示例连接串，那是给本地 demo 用的，真接数仓时改成os.getenv读取，别把账号密码提交进版本库。

3. 需求章节怎么写：Markdown 骨架、pandas 口径验算、断言验收

Notebook 写 PRD 的核心价值不在排版，在于“需求描述”和“数据验证”能放在同一页。用户故事写完，紧跟着一个代码单元把这条故事涉及的分母、分子、边界条件算一遍，评审时谁有异议，改参数重跑就行，不用会后拉群对口径。

3.1 用 Markdown 单元搭需求骨架

每个需求点建议占一个 Markdown 单元，结构固定成四段：背景一句话、用户故事一句、边界条件若干条、验收标准一条可量化描述。固定结构的好处是评审时能快速扫，也方便后面用脚本检查有没有漏填。

资源包里的模板一般长这样，直接照着改：

### 需求 R-03：新用户首周留存看板 **背景**：运营侧目前只能看到整体留存，无法区分渠道来源，投放决策缺少依据。 **用户故事**：作为投放负责人，我希望按渠道查看新用户 7 日留存， 以便判断哪个渠道带来的用户质量更高，从而调整预算分配。 **边界条件** - 只统计注册后 7 个自然日内有任意活跃行为的用户 - 渠道以首次注册时的归因为准，后续变更不回溯 - 单渠道样本量低于 500 时不出结论，仅展示原始值 **验收标准**：同一时间窗内，看板数值与数仓宽表按渠道聚合的结果偏差不超过 0.5%。

这四段里，边界条件是含金量最高的部分。绝大多数口径争议都出在“算不算”“归到哪”，把这些写死，后面代码单元的断言才有依据。验收标准要落到可测的数字上，“误差不超过 0.5%”能被写成断言，“结果准确”不能。

3.2 把用户故事翻成 pandas 口径验算

Markdown 写完就该上代码了。这一步的目的不是出最终报表，而是用一小份样本数据把口径跑通，证明“按这个定义算出来的数长这样”。样本可以从数仓抽一天，也可以直接用资源包自带的 demo 数据。

import pandas as pd from params import START_DATE, END_DATE, MIN_SAMPLE_SIZE # 读注册与活跃两张事件表，字段名按资源包 demo 数据结构 users = pd.read_csv(DATA_DIR / "users.csv", parse_dates=["register_time"]) active = pd.read_csv(DATA_DIR / "active.csv", parse_dates=["active_date"]) # 第一步：框定注册时间窗内的新用户，渠道取首次归因 new_users = users[ (users["register_time"] >= START_DATE) & (users["register_time"] < pd.Timestamp(END_DATE) + pd.Timedelta(days=1)) ].copy() new_users["register_date"] = new_users["register_time"].dt.normalize() # 第二步：算每个用户注册后第 N 天内是否活跃（留存判定） window_end = new_users["register_date"] + pd.Timedelta(days=RETENTION_DAYS) act_flag = active.merge( new_users[["user_id", "register_date", "channel"]], on="user_id", how="inner" ) act_flag = act_flag[ (act_flag["active_date"] >= act_flag["register_date"]) & (act_flag["active_date"] <= window_end) ] # 第三步：按渠道聚合，分子是窗口内有活跃的去重用户 retained = act_flag.groupby("channel")["user_id"].nunique().rename("retained") total = new_users.groupby("channel")["user_id"].nunique().rename("total") result = pd.concat([total, retained], axis=1).fillna(0) result["retention_rate"] = (result["retained"] / result["total"]).round(4) # 第四步：样本量不足的渠道单独标记，不进结论 result["sample_ok"] = result["total"] >= MIN_SAMPLE_SIZE print(result.sort_values("retention_rate", ascending=False))

逻辑上有三个点容易写错。第一，register_date必须做归一化，注册时间是带时分秒的，直接和日期比较会把当天注册的用户排除掉。第二，活跃表 merge 之后要重新按register_date过滤，不能直接用原始时间窗，否则会把注册前就活跃的老行为算进来。第三，分子用nunique而不是count，一个用户 7 天内活跃 5 次只算一次，这是留存和活跃的区别。

MIN_SAMPLE_SIZE在这一步就生效，输出里直接带一列sample_ok。评审时看到某个渠道标着 False，就知道它的高留存率不能当结论用，省得后面被追问。

3.3 验收标准写成断言，别写成形容词

需求文档里写“数据要准确”等于没写。真正有用的做法是把验收标准翻译成 assert，跑不过就报错，跑过了就是一份自证清白的文档。资源包里的checks/目录一般就是干这个的。

# checks/acceptance.py —— 验收断言，供 notebook 与 CI 共用 import pandas as pd def check_retention_bounds(df, lo=0.0, hi=1.0): """留存率必须落在合理区间，超出说明分子分母写反了""" bad = df[(df["retention_rate"] < lo) | (df["retention_rate"] > hi)] assert bad.empty, f"留存率越界：\n{bad}" def check_total_consistency(df, expected_total): """各渠道用户数之和应等于去重后的新用户总数""" got = int(df["total"].sum()) assert got == expected_total, f"用户数不一致：聚合 {got} vs 全量 {expected_total}" def check_sample_flag(df, min_size): """标记为可用的渠道，样本量必须真的够""" bad = df[(df["sample_ok"]) & (df["total"] < min_size)] assert bad.empty, f"样本量标记错误：\n{bad}" # notebook 中调用 total_all = new_users["user_id"].nunique() check_retention_bounds(result) check_total_consistency(result, total_all) check_sample_flag(result, MIN_SAMPLE_SIZE) print("全部验收断言通过")

这几个函数的价值在于把“人眼检查”换成“机器检查”。check_total_consistency尤其重要，按渠道聚合最常见的错误是用户跨渠道重复计数，导致各渠道之和大于全量，这种偏差在小数据上看不出来，上了全量能差出几个百分点。断言写完，notebook 从头跑一遍，全绿才允许导出评审稿，这比会后被人挑出数字对不上体面得多。

4. 数据与指标：SQL 直连、指标字典与图表结论

需求章节跑通之后，接下来是把口径接到真实数据上。这一步的难点不在写 SQL，而在“同一指标在不同人嘴里是不是同一个东西”。指标字典存在的意义就是把这个歧义消掉，让 PRD 里的每个数字都能追溯到唯一定义。

4.1 指标字典：先对齐口径再动手

字典建议做成 CSV 而不是写在 Markdown 表格里，因为 CSV 能被代码读进来做校验——检查 PRD 正文里出现的指标名是不是都在字典里，有没有漏登记的野指标。表结构一般是这几列：

实际用的时候，我会加一段代码做一致性检查：把 PRD 正文里所有反引号包起来的指标名抽出来，和字典的metric_name做差集，差集非空就说明有指标没登记。这个检查很土，但拦住过好几次“会上临时加了个指标、忘了写口径”的情况。

import re, pandas as pd from pathlib import Path metrics = pd.read_csv(BASE_DIR / "metrics_dict.csv") # 从 notebook 的 Markdown 单元里抽 `指标名` 形式的引用 nb_text = Path(BASE_DIR / "prd.ipynb").read_text(encoding="utf-8") referenced = set(re.findall(r"`([\u4e00-\u9fa5A-Za-z0-9_]{2,20})`", nb_text)) known = set(metrics["metric_name"]) | set(metrics["metric_key"]) missing = {m for m in referenced if m not in known and not m.endswith((".py", ".csv", ".sql"))} print("未登记指标：", missing or "无")

re.findall里的字符集限制了长度，是为了避免把代码片段里的长变量名也当成指标抓出来；endswith那行是排除文件名引用。这类小检查不优雅，但实用，尤其适合在评审前一天跑一遍。

4.2 用 SQL 单元格直连取数

Notebook 接数仓，常见做法是ipython-sql的魔术命令，或者用 SQLAlchemy 建 engine 后pd.read_sql。前者写起来爽，后者可控性强。资源包里两种都支持，我一般对固定报表用read_sql，对临时探索用魔术命令。

from sqlalchemy import create_engine import pandas as pd from params import DB_URL, START_DATE, END_DATE engine = create_engine(DB_URL, pool_pre_ping=True) sql = """ SELECT channel, register_date, COUNT(DISTINCT user_id) AS new_users, SUM(CASE WHEN d7_active = 1 THEN 1 ELSE 0 END) AS retained_users FROM dws_user_retention_di WHERE register_date BETWEEN :start_date AND :end_date GROUP BY channel, register_date ORDER BY register_date, channel """ df = pd.read_sql( sql, engine, params={"start_date": START_DATE, "end_date": END_DATE} ) print(df.head()) print("行数：", len(df), "渠道数：", df["channel"].nunique())

参数化传参（:start_date）这一步别省。字符串拼接 SQL 在本地 demo 里跑得通，接上真实数仓后一旦时间窗写错，扫全表能把调度队列堵住。pool_pre_ping=True是为了应对连接空闲超时，Notebook 经常挂着几小时不动，回来一跑就报连接断开，加这个参数能自动重连。

取完数先看行数和渠道数，这两个数对不上预期，后面所有图都白搭。我习惯在取数单元后面立刻加一个分布检查，看每个渠道每天的用户量级是不是在同一档，出现量级差十倍的情况，八成是渠道字段有空值或者枚举值变了。

4.3 图表与结论单元：让评审看到趋势而不是表格

图表的作用是替代“请各位看第三行第七列”这种对话。留存类指标最适合画分组折线：横轴日期，纵轴留存率，每个渠道一条线，样本量不足的渠道用虚线并标注。这样评审一眼能看出哪个渠道在掉，而不是在一堆数字里找。

import matplotlib.pyplot as plt import matplotlib as mpl mpl.rcParams["font.sans-serif"] = ["DejaVu Sans"] # 中文字体按本机可用字体替换 mpl.rcParams["axes.unicode_minus"] = False pivot = df.pivot_table( index="register_date", columns="channel", values=["new_users", "retained_users"], aggfunc="sum" ) rate = (pivot["retained_users"] / pivot["new_users"]).round(4) fig, ax = plt.subplots(figsize=(10, 4.5)) for ch in rate.columns: small = df.loc[df["channel"] == ch, "new_users"].mean() < MIN_SAMPLE_SIZE ax.plot(rate.index, rate[ch], marker="o", ms=3, linestyle="--" if small else "-", label=ch) ax.set_title("分渠道新用户 7 日留存率") ax.set_ylabel("retention_rate") ax.grid(alpha=0.3) ax.legend(fontsize=8, ncol=3) plt.tight_layout() plt.savefig(EXPORT_DIR / "retention_by_channel.png", dpi=150) plt.show()

虚线区分小样本渠道这个细节，比在图下面写一行“注：某渠道样本量不足”有效得多，因为看图的人不一定读注。dpi=150是导出到 HTML 评审稿的合适值，再高文件就大了，嵌进页面加载会卡。

图后面紧跟一个 Markdown 单元写结论，格式固定成“观察到什么、可能原因、建议动作”三句。结论必须能被上面的代码支撑，不能出现图里没有的数字。这条纪律看起来多余，但实际评审里最常见的质疑就是“这个 12% 是从哪来的”，而它往往来自上一版没更新的手写文字。

5. 避坑与排查：Notebook 写 PRD 的五类翻车

Notebook 的便利和它的坑是同一件事的两面：状态存在内核里，所以快；状态存在内核里，所以不可信。下面这几类问题，几乎每个用 Notebook 写需求文档的团队都会撞上至少两三个。

5.1 环境与执行类：内核、路径、乱序

现象：本地跑得好好的 notebook，换台机器或者重启内核后FileNotFoundError，或者 import 到的包版本不对，报一些莫名其妙的属性错误。

原因：路径用了相对路径且依赖当前工作目录；内核指向了另一个 Python 环境；或者 notebook 里存在被手动跳过、但后面单元依赖其变量的单元格。

解决：路径一律基于Path(__file__).resolve().parent或 notebook 所在目录推导，不用裸相对路径；启动后第一个单元就打印sys.executable并肉眼确认；交付前必须 Restart & Run All 跑一遍，任何需要“先跑这个再跑那个”的顺序依赖都说明代码该重构了。这一条我吃过亏，某次评审现场演示，第二个单元一直报变量未定义，最后发现是自己调试时跳过了初始化单元，全场等着我重启。

5.2 数据与口径类：时区、空值、分母

现象：同一个指标，PRD 里算出来和数仓看板差几个百分点，查了半天发现是某一天的数对不上。

原因：时间字段有时区差异，注册时间存的是 UTC，统计按本地日期切分，跨零点注册的用户被归到了前一天；或者渠道字段存在空值，groupby默认丢弃导致各渠道之和小于全量；再或者分母用了count而不是nunique，把重复注册算了两遍。

解决：所有时间字段在读入时统一tz_localize到同一时区再转日期；groupby加dropna=False，空值单独成一类并显式处理；分子分母的聚合函数在指标字典里写死，代码里不许临时改。前面 3.3 的check_total_consistency断言就是专门拦这一类问题的。

5.3 交付与协作类：导出、diff、大文件

现象：notebook 提交到代码库后，同事改了一行文字，diff 显示整个文件全变了，代码评审没法做；或者导出 HTML 有几十兆，发出去打不开。

原因：.ipynb 本质是 JSON，包含输出、执行计数、图片 base64，任何一次运行都会让文件内容大变；图表以高分辨率内嵌，体积迅速膨胀。

解决：提交前清空输出（jupyter nbconvert --ClearOutputPreprocessor.enabled=True --inplace prd.ipynb），或者配nbstripout钩子自动处理；图表统一dpi=150并保存为独立 png 引用，而不是内嵌；代码评审看导出的 HTML 或 py 转换版，不看原始 ipynb。团队里如果有人坚持把带输出的 notebook 直接推上去，代码库很快就没法看了。

5.4 参数与复用类：硬编码和隐式全局

现象：换一个时间窗，要改十几个单元格；某次只改了一半，图上是新数据、结论文字还是旧的。

原因：参数散落在各个单元，没有单一入口；结论文字手写在 Markdown 里，不随数据更新。

解决：参数全部收敛到params.py，notebook 只读不写；结论部分能自动生成的（比如“环比变化 X%”）就用 f-string 拼出来，减少手写；确实需要人工判断的结论，加一个# TODO 复核标记并在导出前搜索一遍。这条的收益在第二次评审时最明显，第一次大家都能记住自己改了哪。

5.5 版本与回溯类：这版数字怎么来的

现象：两周后有人问“当时那个 8.3% 是怎么算的”，翻遍聊天记录和邮件都找不到对应的代码版本。

原因：notebook 改了又改，没有打标签，数据快照也被下一次运行覆盖了。

解决：每次导出评审稿时，把params.py的值、数据快照日期、notebook 的 git commit 短哈希写进导出文件头部；原始数据落地到data/raw/日期/而不是原地覆盖。这是最不起眼但最省心的一条，等于给每个结论留了后悔药。

6. 把 Notebook PRD 交付出去：参数化重跑与导出评审稿

文档在自己机器上跑通只是及格线，能交给别人重跑才算完成。Notebook 交付最大的问题是“我这儿能跑”，所以导出前建议走一遍参数化重跑，用脚本模拟一个陌生人拿到文件后的执行路径。

# 用 papermill 注入参数并完整重跑，输出到临时文件 papermill prd.ipynb export/prd_run.ipynb \ -p START_DATE "2024-03-11" \ -p END_DATE "2024-03-17" \ --kernel python3 # 重跑无报错后，再导出成 HTML 评审稿 jupyter nbconvert export/prd_run.ipynb \ --to html --output prd_review.html \ --HTMLExporter.theme=light

papermill的-p参数会覆盖 notebook 里标记为parameters的单元的变量值，所以顶部参数单元记得加标签，否则注入不生效。这一步的实际意义是强制一次从头到尾的执行——所有依赖隐式状态的单元都会在这一步暴露。我一般把这条命令写进Makefile，交付前跑一次，跑不过就不发。

导出成 HTML 而不是直接发 ipynb，是考虑到评审对象不一定装了 Jupyter。HTML 里图表、代码、结论都在，但不可执行，正好符合“给人看”的定位；需要复现的人拿 ipynb 和params.py自己跑。两套产物分工明确，别指望一个文件同时满足两种需求。

最后一个技巧，也是我认为最值钱的：给导出的 HTML 加一个固定的头部区块，写清数据截止日期、参数取值、代码版本。评审时所有“这个数是什么时候的”问题都能在这一块找到答案，不用现场翻记录。

from datetime import datetime import subprocess commit = subprocess.check_output( ["git", "rev-parse", "--short", "HEAD"], text=True ).strip() header = f""" > 数据截止：{END_DATE} ｜ 统计窗口：{START_DATE} ~ {END_DATE} > 参数：RETENTION_DAYS={RETENTION_DAYS}, MIN_SAMPLE_SIZE={MIN_SAMPLE_SIZE} > 代码版本：{commit} ｜ 生成时间：{datetime.now():%Y-%m-%d %H:%M} """ print(header)

这段输出放在 notebook 最顶部，导出时会渲染成 HTML 的第一个区块。看起来只是几行字，但它把“结论”和“产生结论的条件”绑在了一起，半年后回看也不会变成黑匣子。

我以前交付 PRD 的习惯是写完就发，觉得文档嘛，内容对就行。直到有一次评审通过了三周，数仓那边问某个渠道留存为什么和看板对不上，我翻出当时的 notebook，发现参数早就被自己改过两轮，原始那份数据也被覆盖了，最后只能重新跑一遍，还得跟人解释“数字可能和当时略有出入”。从那以后，我每次导出评审稿都强制走一遍参数化重跑和头部区块生成，哪怕只改了一个字。希望帮到你。

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