1. 项目背景业务场景食光集市每周一的运营周会上运营总监要一份上周各商圈超时订单分布 热销品类排名的报表。过去三个月这份报表的产出流程是运维小李登录数据库跑一条 200 行的 SQL每次都要微调日期范围导出 CSV用 Excel 打开——单文件 80 万行打开要 45 秒在 Excel 里做数据透视表 VLOOKUP 条件格式——操作 20 分钟截图粘贴到周报 PPT一个 2 小时的活每周一早上雷打不动。直到上周一——小李请假了没人会跑那条 SQLPPT 上只有一行数据暂缺。运营总监当着 CTO 的面说你们连个自动报表都做不出来CTO 把任务派给开发团队用 Python 脚本自动化。开发也能搞定’够用的’数据分析——不需要多炫但要满足三个要求每周一自动跑出 CSV 一张折线图代码能跑在 CI 上无 GUI 环境数据口径和那条 SQL 完全一致本章用 pandas matplotlib 构建一个够用的报表自动化流水线——不求 BI 工具的复杂度但求可复现、可 CI、可交接。痛点手工数据分析流程的问题不可复现SQL 在运维的 Navicat 里日期靠手动改换人就断档。Excel 性能瓶颈百万行数据 Excel 直接崩VLOOKUP 跨表查询卡十分钟。口径不一致运营用 Excel 算的超时率和开发用代码算的超时率可能差 2%——因为空值的处理方式不同。无版本管理每次改日期生成的新报表覆盖旧报表——想回溯上周的报表“被覆盖了没留备份。”2. 项目设计三人剧本交锋场景小李的 200 行 SQL 投在大屏上——5 层嵌套子查询 3 个CASE WHENGROUP BY ROLLUP。小胖看了愣住。小胖“大师这 SQL 比我写的任何 Python 代码都复杂……我有一个质朴的问题——为什么不在数据库里做完全部聚合然后 Python 只负责展示毕竟 SQL 是专门干这个的。”小白“小胖问得合理——这是一个经典的’计算该在哪层做’的问题。数据库做聚合效率最高数据在磁盘上不用传输但 200 行 SQL 的维护成本太高——每次需求微调‘超时阈值从 30 分钟改到 45 分钟’就要改 SQL而 SQL 越改越难读。折中方案是什么”大师“计算分层的原则——能下推的就下推需要灵活性的上提到应用层。”分层策略数据库层做过滤WHERE、简单聚合GROUP BY、排序——把 80 万行数据精简到 5000 行以内的中间结果集。Python 层做复杂计算同环比、移动平均、自定义分组、格式美化、可视化。输出层生成 CSV给运营二次加工 一张 PNG 图给周报 PPT 直接用。“判断标准——如果一段计算逻辑改动频率 半年一次如’超时判断标准 配送时间 预计时间 1.2 倍’可以放 SQL 里。如果改动频率 每次周会如运营想按商圈、按品类、按时段做不同维度的分析放 Python 里更灵活。”技术映射数据库 中央厨房——负责洗菜、切菜、焯水过滤、聚合降低运输量。Python 精加工间——把半成品拼成摆盘多维分析、可视化每道菜的摆盘方式各不相同需求多变。小胖“那 pandas 的核心 DataFrame 操作有哪些我每次查 pandas 文档都有一种’函数太多记不住’的绝望。”大师pandas 日常高频五件套pd.read_csv(file, parse_dates[order_date])——读取 自动解析日期列df[df[status] completed]——行过滤df.groupby(store_id)[amount].sum()——分组聚合pd.merge(orders, stores, onstore_id)——表关联df.to_csv(output.csv, indexFalse)——输出80% 的数据处理操作都可以用这五个函数拼出来。剩下的 20% 在需要的时候查文档——不要尝试记全。小白“我关心的是——matplotlib 在无 GUI 的 CI 环境比如 GitHub Actions 的 Linux runner里能正常出图吗之前部署的机器上不装 X11 就报错。”大师“用matplotlib.use(Agg)——在 import pyplot 之前设置后端为 Agg非交互式渲染能在无 GUI 环境下正常生成 PNG/SVG 图片。这是 CI/CD 跑数据脚本的必修课。”importmatplotlib matplotlib.use(Agg)# 必须在 import pyplot 之前importmatplotlib.pyplotasplt3. 项目实战超时订单分析报表环境准备依赖版本说明Python3.13.14基准版本pandas2.2数据处理matplotlib3.9可视化mkdirfoodmarket-ch22cdfoodmarket-ch22 python-mvenv .venv .venv\Scripts\activate pipinstallpandas matplotlib pytest分步实现步骤1生成模拟订单数据src/generate_data.py生成模拟订单数据importrandomimportpandasaspdfromdatetimeimportdatetime,timedelta STORES[(ST001,朝阳店,朝阳商圈),(ST002,浦东店,陆家嘴商圈),(ST003,南山店,科技园商圈),(ST004,武侯店,春熙路商圈),(ST005,西湖店,武林商圈),]DISHES[宫保鸡丁,麻婆豆腐,水煮鱼,蛋炒饭,烤串,冰粉,酸梅汤]defgenerate_orders(rows:int10000)-pd.DataFrame:生成模拟订单 CSVnowdatetime.now()data[]foriinrange(rows):storerandom.choice(STORES)order_timenow-timedelta(daysrandom.randint(0,29),hoursrandom.randint(0,23),minutesrandom.randint(0,59),)etarandom.randint(20,60)actual_minutesrandom.randint(15,90)data.append({order_id:fSG-{i:08d},store_id:store[0],store_name:store[1],district:store[2],dish_name:random.choice(DISHES),amount:random.randint(10,200),eta_minutes:eta,actual_minutes:actual_minutes,is_overtime:actual_minuteseta*1.2,order_time:order_time,status:random.choice([completed,cancelled,completed]),})returnpd.DataFrame(data)步骤2实现报表分析引擎src/report_engine.py运营报表引擎——pandas 聚合 matplotlib 可视化importmatplotlib matplotlib.use(Agg)# 无 GUI 后端importmatplotlib.pyplotaspltimportpandasaspdfrompathlibimportPathclassWeeklyReport:def__init__(self,output_dir:str./reports):self.output_dirPath(output_dir)self.output_dir.mkdir(parentsTrue,exist_okTrue)defanalyze(self,df:pd.DataFrame)-dict:执行全部分析并生成报表文件 Returns: dict: 各指标汇总 # 确保日期列为 datetimedf[order_time]pd.to_datetime(df[order_time])results{}# 1. 超时率 by 商圈results[overtime_by_district]self._overtime_by_district(df)self._plot_overtime_by_district(results[overtime_by_district])# 2. 每日订单趋势results[daily_trend]self._daily_trend(df)self._plot_daily_trend(results[daily_trend])# 3. 热销品类 Top5按菜品名替代品类results[top_dishes]self._top_dishes(df)# 4. 总结报告 CSVself._export_summary(df,results)# 5. 异常订单明细超时 2 倍abnormaldf[df[actual_minutes]df[eta_minutes]*2]abnormal.to_csv(self.output_dir/abnormal_orders.csv,indexFalse,encodingutf-8-sig)results[abnormal_count]len(abnormal)returnresultsdef_overtime_by_district(self,df:pd.DataFrame)-pd.DataFrame:各商圈超时率district_statsdf.groupby(district).agg(total_orders(order_id,count),overtime_orders(is_overtime,sum),avg_actual(actual_minutes,mean),avg_eta(eta_minutes,mean),).reset_index()district_stats[overtime_rate](district_stats[overtime_orders]/district_stats[total_orders]*100).round(2)returndistrict_stats.sort_values(overtime_rate,ascendingFalse)def_daily_trend(self,df:pd.DataFrame)-pd.DataFrame:每日订单量趋势df[date]df[order_time].dt.date dailydf.groupby(date).agg(orders(order_id,count),revenue(amount,sum),overtime_rate(is_overtime,lambdax:x.sum()/x.count()*100),).reset_index()returndailydef_top_dishes(self,df:pd.DataFrame)-pd.DataFrame:热销菜品 Top5returndf.groupby(dish_name).size().sort_values(ascendingFalse).head(5).reset_index(namecount)def_plot_overtime_by_district(self,data:pd.DataFrame):商圈超时率柱状图fig,axplt.subplots(figsize(8,5))barsax.bar(data[district],data[overtime_rate],colorsteelblue)ax.set_title(各商圈超时率 (%),fontsize14)ax.set_ylabel(超时率 %)forbar,rateinzip(bars,data[overtime_rate]):ax.text(bar.get_x()bar.get_width()/2,bar.get_height()0.5,f{rate}%,hacenter,fontsize10)plt.tight_layout()plt.savefig(self.output_dir/overtime_by_district.png,dpi100)plt.close()def_plot_daily_trend(self,data:pd.DataFrame):每日订单趋势折线图fig,ax1plt.subplots(figsize(10,5))ax1.plot(data[date],data[orders],b-,label订单量)ax1.set_ylabel(订单量,colorb)ax2ax1.twinx()ax2.plot(data[date],data[overtime_rate],r--,label超时率 %)ax2.set_ylabel(超时率 %,colorr)ax1.set_title(每日订单量 超时率趋势)fig.autofmt_xdate()plt.tight_layout()plt.savefig(self.output_dir/daily_trend.png,dpi100)plt.close()def_export_summary(self,df:pd.DataFrame,results:dict):导出汇总 CSVrows[{指标:总订单数,值:len(df)},{指标:总营收,值:f¥{df[amount].sum():,.0f}},{指标:整体超时率,值:f{df[is_overtime].mean()*100:.1f}%},{指标:严重异常订单,值:results[abnormal_count]},{指标:活跃商圈数,值:df[district].nunique()},]summary_dfpd.DataFrame(rows)summary_df.to_csv(self.output_dir/summary.csv,indexFalse,encodingutf-8-sig)if__name____main__:fromsrc.generate_dataimportgenerate_orders dfgenerate_orders(20000)reportWeeklyReport(./reports)resultsreport.analyze(df)print(f报表生成完成:{report.output_dir})forkey,valinresults.items():ifisinstance(val,pd.DataFrame):print(f{key}:{len(val)}行)else:print(f{key}:{val})步骤3编写测试tests/test_report.pyimportpandasaspdfrompathlibimportPathfromsrc.report_engineimportWeeklyReport SAMPLE_DATApd.DataFrame({order_id:[SG-001,SG-002,SG-003,SG-004],store_id:[ST001,ST001,ST002,ST002],store_name:[朝阳店,朝阳店,浦东店,浦东店],district:[朝阳商圈,朝阳商圈,陆家嘴商圈,陆家嘴商圈],dish_name:[宫保鸡丁,水煮鱼,麻婆豆腐,冰粉],amount:[28,68,18,8],eta_minutes:[30,30,25,20],actual_minutes:[28,55,30,18],is_overtime:[False,True,True,False],order_time:pd.to_datetime([2025-03-15 10:00,2025-03-15 11:00,2025-03-16 09:00,2025-03-16 10:00]),status:[completed,completed,cancelled,completed],})classTestWeeklyReport:deftest_overtime_by_district(self,tmp_path):reportWeeklyReport(str(tmp_path))resultreport._overtime_by_district(SAMPLE_DATA)assertlen(result)2# 两个商圈chaoyangresult[result[district]朝阳商圈].iloc[0]assertchaoyang[overtime_rate]50.0deftest_daily_trend(self,tmp_path):reportWeeklyReport(str(tmp_path))resultreport._daily_trend(SAMPLE_DATA)assertlen(result)2# 两天deftest_top_dishes(self,tmp_path):reportWeeklyReport(str(tmp_path))resultreport._top_dishes(SAMPLE_DATA)assertresult.iloc[0][count]1# 每道菜 1 次deftest_full_report_generates_files(self,tmp_path):reportWeeklyReport(str(tmp_path))resultsreport.analyze(SAMPLE_DATA)assert(tmp_path/overtime_by_district.png).exists()assert(tmp_path/daily_trend.png).exists()assert(tmp_path/summary.csv).exists()assertresults[abnormal_count]0运行python-mpytest tests/-v完整代码清单foodmarket-ch22/ ├── src/ │ ├── __init__.py │ ├── generate_data.py # 模拟数据 │ └── report_engine.py # 报表引擎 ├── tests/ │ └── test_report.py ├── reports/ # 生成报表目录 └── requirements.txt4. 项目总结优点 缺点维度pandas matplotlib纯 SQL ExcelPower BI / TableauApache Spark成本★★★★★ 免费★★★★★★ 商业授权★★★ 集群成本灵活性★★★★★★★★★★★★★★百万行数据★★★★★★★★★★★★★★学习曲线★★★★★★★★★★★★自动化/CI★★★★★★★★★★★★适用场景周期性运营报表周报/月报——定时任务自动跑图表直接贴 PPT。异常数据巡检找出与正常模式偏差最大的数据点——如超时 3 倍的订单。AB 测试结果分析pandas 做透视表 置信区间计算。数据迁移前的质量检查新老系统数据比对——pandas merge 找出差异行。探索性分析Jupyter 里df.describe()df.plot()快速了解数据分布。不适用场景实时流处理pandas 是批处理模式不适合逐条实时聚合——用 Flink/Spark Streaming。TB 级数据单机 pandas 受内存限制超过内存 50% 的数据量就跑不动——用 Spark/Dask。注意事项pd.to_datetime默认推断格式可能出错如果日期列混合2025-03-15和03/15/2025pandas 可能推断错误。用format%Y-%m-%d显式指定。groupby后索引被重置agg后索引层级可能不符合预期习惯性加.reset_index()。中文乱码to_csv(encodingutf-8-sig)——-sig后缀自动加 BOM确保 Excel 正确识别。matplotlib 中文字体Linux CI 环境通常没有中文字体展示中文报表需先安装fonts-noto-cjk或关闭中文显示。常见踩坑经验故障案例1inplaceTrue的链式操作陷阱现象df.dropna(inplaceTrue).groupby(...)报AttributeError: NoneType object has no attribute groupby。根因dropna(inplaceTrue)返回None不能链式调用。修复永远不用inplaceTrue——df df.dropna()更清晰可链式。故障案例2SettingWithCopyWarning但代码看起来没问题现象df_filtered df[df[a] 0]; df_filtered[new_col] 1触发 Warning。根因df_filtered可能是原始 DataFrame 的视图view而非副本copy赋值操作有歧义。修复df_filtered df[df[a] 0].copy()——显式创建副本。故障案例3matplotlib 内存泄漏——图片没关闭现象循环中生成 100 张报表图内存持续增长。根因每次plt.savefig()后没有plt.close()Figure 对象堆积在内存中。修复plt.savefig(...); plt.close(all)——或使用with plt.ioff():上下文。思考题df.groupby(store_id).agg({amount: sum, status: count})和df.groupby(store_id).amount.sum()的结果有什么不同什么情况下用agg比直接调用.sum()更好一个 500MB 的 CSV 文件无法全部读入内存。除了用第 16 章的生成器管道pandas 自身有什么分块读取的机制如何实现分块聚合答案见中级篇综合实战章附录。延伸阅读与资源Python 3实战精进从脚本到高并发订单引擎MongoDB 实战进阶与内核修炼python入门Rquests从菜鸟脚本到企业级SDK的网络实战圣经Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析