简介这是一套面向直播电商开发者与二次创业者的技术源码资源聚焦‘直播本地生活菜市场/生鲜’融合场景提供可快速部署的双端直播带货系统基础框架。资源包含完整前后端代码、配套视频教程及作者原始分享说明适用于具备Java/PHP或小程序开发基础的工程师进行功能定制与业务拓展。压缩包大小为55.1MB文件总数虽未明确统计但根据典型直播类项目结构应涵盖服务端核心逻辑如直播推拉流对接、商品秒杀模块、管理后台含主播审核、菜品库存管理、小程序前端页面及部署配置脚本等关键类型。已有1120人下载学习内容延续作者早期版本优化新增实操录屏讲解覆盖环境搭建、核心接口调试与常见联调问题处理路径便于开发者高效理解架构设计与业务流程闭环。 做直播带货的兄弟应该都有过这种体验一场直播下来销售额是知道了但流量从哪来的、观众停留多久、哪个商品真正起了量、哪个款只是看着热闹全凭感觉。我前阵子帮一个做生鲜的朋友整理了整套直播运营的数据体系顺手把整套东西做成了开源格式的源码也就是这个“cainao_直播菜源码”的项目。简单说它是一套用 Python 写的直播带货辅助数据分析源码能帮你把直播间的基础数据、商品表现、实时转化情况汇总到一块看板上全程不需要手动整理表格跑起来之后你看一眼屏幕就知道下一步该推哪个品、该不该补流量。这套源码不是那种特别重型的商业级系统它的定位很明确给个人主播、小团队、或者刚接触直播电商的技术爱好者提供一个能看懂、能改、能用的轻量方案。技术栈用的是 Python SQLite Flask ECharts全部都是常见技术哪怕你只会点 Python 基础也能把这套东西跑起来。这篇文章我就把这套“直播带菜源码”从设计思路到核心实现完整拆一遍代码我会直接贴出来你照着操作就行。1. 项目整体设计与思路拆解1.1 为什么直播带货需要一套自己的数据看板先说个很现实的问题。现在主流直播平台的后台其实都有自己的数据面板能看成交金额、看流量来源、看转化率但大部分人用起来有个痛点后台数据是平台定义的指标体系它不一定会按你自己的运营节奏来组织。举个例子你这场直播有A、B、C三款商品平台只会告诉你每款卖了多少钱但不会告诉你“前20分钟推A款的时候观众停留时长是明显偏高的说明这个品的内容吸引力强下次应该把A放在开场”。再一个痛点就是数据分散。你在直播间讲品的时候实时互动数据、商品点击数据、成交数据往往分散在不同页面要同时盯几个屏幕才能拼出完整画面。这种场景下自己搭建一个聚合看板就很有必要。我这套源码的核心思路就是把平台提供的基础数据手动导出或通过开放接口拉取汇总到本地 SQLite 数据库然后统一计算指标、统一展示。它不追求实时到秒级而是给运营一个整合视角。1.2 技术选型的几个关键抉择这套系统最初选型的时候其实纠结过要不要上重型方案比如引入消息队列、实时计算框架什么的。后来我放弃了原因很简单直播间的数据量级和数据时效要求用不到那么大的架构。一场直播从开播到下播核心数据也就几万条SQLite 完全撑得住。Flask 做后端也就够了不需要上 Spring Cloud 那种全家桶维护成本对于个人主播和团队来说太高了。所以最终选型是下面这套模块选型理由数据存储SQLite零配置、单文件、足够支撑直播数据量后端服务Flask轻量、灵活、和数据分析生态兼容数据处理Pandas清洗、分组、聚合计算非常方便前端图表ECharts交互能力好、图表类型丰富、适合实时刷新部署方式本地单机运行不用买服务器笔记本就能跑这个组合的另外一个好处是它特别适合用来学习源码。你不需要理解分布式系统、不需要配置复杂的中间件整个项目的代码量控制在千行级别每一段都能看懂、能改。如果你想把它改造成团队共用的系统加一层登录鉴权和 MySQL 存储就行改造空间也很大。1.3 系统的功能模块划分整套源码按功能划分成四个模块分别是数据采集模块负责从平台后台导出文件或通过开放接口拉取数据统一解析成结构化数据入库。数据处理模块对原始数据进行清洗剔除异常值计算核心指标。指标计算与存储模块把计算好的指标按直播间场次、商品维度存储方便历史对比。可视化看板模块提供 Web 页面以图表形式展示实时与历史数据。后面我会逐模块讲解实现这里先铺垫一个大框架你脑子里先有个地图后面看代码的时候不会迷路。2. 核心模块选型与实现准备2.1 开发环境与依赖安装我假设你用的是 Windows 或 macOSPython 版本建议 3.9 以上。我是在 Python 3.10 环境下开发的3.9 以下有些语法特性可能不兼容这点要注意。项目依赖不多核心就这几个库pip install flask pandas numpy openpyxl requestsFlaskWeb 服务用来提供接口给前端页面调数据。Pandas处理表格数据做聚合运算。Numpy数值计算有些指标需要用到。Openpyxl读取 Excel 格式的导出文件。Requests从开放接口拉数据用的。装完依赖之后建议你先把项目目录建好。我的目录结构是live_dashboard/ ├── app.py # Flask 主程序 ├── config.py # 配置文件 ├── database.py # 数据库连接与建表 ├── data_loader.py # 数据采集与解析 ├── data_processor.py # 数据清洗与指标计算 ├── templates/ │ └── index.html # 看板页面 ├── static/ │ ├── echarts.min.js # ECharts 库 │ └── dashboard.js # 前端逻辑 └── data/ └── live_records.db # SQLite 数据库文件自动生成你不需要完全照搬这个结构但我建议你至少把后端逻辑和数据解析分开不然写到后面代码会越来越乱自己都不想维护。2.2 数据库表结构设计数据库设计这块我踩过坑。最初我把所有数据都塞进一张大表结果指标加的越来越多查询速度越来越慢代码也越来越难读。后来重新拆成三张核心表才顺起来。第一张是直播间基础信息表live_sessions记录每场直播的元数据CREATE TABLE IF NOT EXISTS live_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT UNIQUE, -- 直播场次ID live_title TEXT, -- 直播标题 start_time DATETIME, -- 开播时间 end_time DATETIME, -- 下播时间 total_viewers INTEGER, -- 累计观看人数 peak_viewers INTEGER, -- 最高在线人数 avg_view_duration INTEGER, -- 平均观看时长秒 total_orders INTEGER, -- 总成交订单数 total_sales_amount REAL, -- 总成交金额 goods_count INTEGER -- 商品数量 );第二张是商品表现表product_performance记录每个商品在单场直播中的数据CREATE TABLE IF NOT EXISTS product_performance ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, product_id TEXT, product_name TEXT, category TEXT, -- 商品品类 price REAL, -- 售价 impressions INTEGER, -- 商品曝光次数 clicks INTEGER, -- 商品点击次数 orders INTEGER, -- 商品成交订单数 sales_amount REAL, -- 商品成交金额 stock INTEGER, -- 库存数量 FOREIGN KEY (session_id) REFERENCES live_sessions(session_id) );第三张是实时趋势表live_trends记录直播过程中每分钟的快照数据用于画折线图CREATE TABLE IF NOT EXISTS live_trends ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, record_time DATETIME, online_count INTEGER, -- 当前在线人数 cumulative_viewers INTEGER, -- 累计观看人数 total_orders INTEGER, -- 累计订单数 sales_amount REAL, -- 累计成交金额 impression_count INTEGER, -- 累计曝光量 click_count INTEGER -- 累计点击量 );为什么要建三张表而不是一张因为查询维度不一样。看板首页要看场次总览用的是第一张表点进某个商品要看单品转化漏斗用的是第二张表看直播过程中的实时曲线用的是第三张表。三张表用session_id关联逻辑很清晰。3. 数据采集与清洗模块详解3.1 数据采集的合规姿势与常见方式数据采集是整个系统的输入源头也是最容易踩坑的地方。我的原则是只对接官方开放平台的接口或者后台导出的数据文件不做任何绕过平台限制的抓取动作。这不只是合规问题也是稳定性问题——绕过限制的采集方式随时可能失效你辛辛苦苦跑通的流程第二天就断掉得不偿失。目前我实际使用中比较稳的方式有三种后台导出Excel平台后台一般支持导出直播记录、商品明细手动下载后放到data/exports/目录源码会自动扫描解析。开放平台API部分平台提供开放接口用requests定时拉取适合对实时性要求较高的场景。手动录入实在没有接口和导出文件时就手动整理成 CSV 格式放入指定目录。考虑到大部分个人主播拿不到开放平台的接口权限我在源码里默认采用第一种方式同时预留了第二种方式的扩展接口。你在实际使用中可以按自己的数据权限灵活切换。3.2 Excel 导入与数据解析实现我写了一个data_loader.py负责扫描指定目录下的 Excel 文件解析成 DataFrame 然后入库。核心代码大概是这样import os import pandas as pd from datetime import datetime from database import get_db_connection EXPORT_DIR data/exports def load_excel_files(): 扫描导出目录解析所有Excel文件并入库 files [f for f in os.listdir(EXPORT_DIR) if f.endswith((.xlsx, .xls))] for file_name in files: file_path os.path.join(EXPORT_DIR, file_name) df pd.read_excel(file_path) if session_id in df.columns: save_live_sessions(df) if product_id in df.columns: save_product_performance(df) if record_time in df.columns: save_live_trends(df) # 处理完的文件可以重命名加 .done 后缀避免重复导入 os.rename(file_path, file_path .done)这段代码有个细节值得说处理完的文件要加.done后缀。不然你每次启动程序都会重复导入同一批文件数据库里就会出现重复数据。这个坑我第一次跑的时候就踩了后来才补上这个逻辑。3.3 数据清洗的关键步骤原始数据基本都需要清洗。我总结了四个最常见的脏数据场景缺失值比如某商品没有曝光数据需要填充 0 或者剔除。重复记录同一场直播的数据被导出了两次需要去重。异常值在线人数突然变成 999999这种明显是异常数据要过滤掉。时间格式不统一有的时间是字符串有的是 datetime需要统一格式。清洗代码如下def clean_data(df): 数据清洗的核心方法 # 去重 df df.drop_duplicates(subset[session_id, product_id], keeplast) # 填充缺失值 numeric_cols [price, impressions, clicks, orders, sales_amount] for col in numeric_cols: if col in df.columns: df[col] df[col].fillna(0) # 过滤异常值比如订单数不可能为负 if orders in df.columns: df df[df[orders] 0] # 统一时间格式 if record_time in df.columns: df[record_time] pd.to_datetime(df[record_time]) return df很多新手做数据分析时喜欢直接拿原始数据开干结果算出来的指标总是“怪怪的”其实就是少了清洗这一步。我个人的习惯是清洗逻辑要单独写函数不要把清洗代码散落在各个分析模块里这样后面维护起来非常痛苦。4. 数据指标计算与可视化看板实现4.1 直播核心指标的计算逻辑数据入库之后重头戏就是指标计算。看板不能只是把原始数据罗列出来那就没有意义了。我重点做了几个指标曝光点击率CTR商品点击次数 / 商品曝光次数反映商品在直播间的吸引力。浏览转化率CVR商品成交订单数 / 商品点击次数反映话术和逼单效果。客单价AOV成交金额 / 成交订单数反映价格带定位。UV价值成交金额 / 累计观看人数衡量流量的变现效率。峰值在线率最高在线人数 / 累计观看人数衡量直播间留人能力。指标计算用 Pandas 的groupby函数非常方便def calculate_metrics(df): 计算商品维度的核心指标 grouped df.groupby(product_id).agg( product_name(product_name, first), price(price, first), impressions(impressions, sum), clicks(clicks, sum), orders(orders, sum), sales_amount(sales_amount, sum) ).reset_index() # 计算衍生指标 grouped[ctr] grouped[clicks] / grouped[impressions].replace(0, float(inf)) grouped[cvr] grouped[orders] / grouped[clicks].replace(0, float(inf)) grouped[aov] grouped[sales_amount] / grouped[orders].replace(0, float(inf)) return grouped这里要注意replace(0, float(inf))这个操作。如果曝光数为 0直接除会出现除零警告和 NaN替换成无穷大的话我们需要在后续显示时做一些特殊处理比如显示为“—”。4.2 Flask 接口设计与 JSON 返回后端我用 Flask 提供两个接口一个返回场次总览数据一个返回商品维度明细。前端页面加载时通过 fetch 调用这两个接口拿数据画图。from flask import Flask, jsonify, render_template from database import query_data app Flask(__name__) app.route(/) def index(): return render_template(index.html) app.route(/api/summary) def api_summary(): 返回直播场次汇总数据 sql SELECT * FROM live_sessions ORDER BY start_time DESC LIMIT 10 rows query_data(sql) return jsonify(rows) app.route(/api/products/session_id) def api_products(session_id): 返回某场直播的商品表现数据 sql SELECT * FROM product_performance WHERE session_id ? ORDER BY sales_amount DESC rows query_data(sql, (session_id,)) return jsonify(rows) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)接口设计有一个原则要分享不要在前端直接查数据库一定要通过接口做一层隔离。这样后端可以随时调整查询逻辑而不影响前端也方便以后加权限控制、加缓存。4.3 ECharts 可视化看板的实现前端页面我用了 ECharts这个库最强大的一点是——你只需要给它一个数据数组它就能画出各种漂亮的图表。看板页面设计了三块核心图表第一块是“本场直播实时趋势折线图”展示在线人数和累计成交金额的变化曲线。直播运营看这个图就知道了什么时候是转化高峰、什么时候是流量低谷。第二块是“商品销售排行柱状图”按成交金额从高到低排列。这个图直接决定下一场的排品策略——记住了你卖的好的品不一定是转化率最高的品可能是流量倾斜导致的要结合转化率一起看才有意义。第三块是“单品转化漏斗图”从曝光、点击、成交三个环节展示漏斗。这个图帮你看清楚问题出在哪个环节是曝光不够导致点击少还是点了不买导致转化差。前端核心代码fetch(/api/summary) .then(response response.json()) .then(data { const chart echarts.init(document.getElementById(trendChart)); const option { tooltip: { trigger: axis }, legend: { data: [在线人数, 成交金额] }, xAxis: { type: category, data: data.map(d d.record_time) }, yAxis: [ { type: value, name: 在线人数 }, { type: value, name: 成交金额 } ], series: [ { name: 在线人数, type: line, data: data.map(d d.online_count), smooth: true }, { name: 成交金额, type: line, yAxisIndex: 1, data: data.map(d d.sales_amount), smooth: true } ] }; chart.setOption(option); });这里有个小的实操建议ECharts 不建议直接用 CDN 链接因为你直播的时候网络可能会出现波动加载一个本地版本的echarts.min.js文件更稳妥。这是我在真实直播现场吃过亏才补上的——有一次页面加载不出图表全场气氛一度非常尴尬。5. 直播辅助小工具话术提示与选品建议5.1 实时话术提示模块聊完可视化我再分享一个实用性极强的辅助模块——话术提示。很多主播下播之后复盘发现自己在转化率最高的时间点没有及时逼单在观众流失严重的时间点又在讲无关话题。这个模块解决的就是这个问题。我在系统里加了一个简单的话术提示逻辑根据实时数据变化自动生成运营提示。比如当在线人数连续5分钟下降超过10%系统会提示“当前在线人数下滑建议启动互动游戏或发放福袋留人”当某个商品的点击率高但转化率低系统会提示“该商品点击良好建议加强信任背书和限时优惠话术”。这个逻辑实现起来不复杂就是基于规则引擎的判断def generate_tips(current_metrics, history_metrics): 根据当前指标生成话术提示 tips [] # 检测在线人数下滑 online_drop (current_metrics[online_count] - history_metrics[online_count]) / history_metrics[online_count] if online_drop -0.1: tips.append(在线人数下滑超过10%建议互动留人) # 检测商品点击好但转化差 if current_metrics[ctr] 0.08 and current_metrics[cvr] 0.02: tips.append(当前商品点击良好但转化偏低建议加强逼单话术) # 检测成交高峰期 if current_metrics[orders] history_metrics[orders] * 1.5: tips.append(成交速度加快建议追加一波福利款) return tips当然这套规则不一定完全贴合你的直播风格你可以根据自己的经验调整阈值。我强烈建议你把提示文本改成自己说话的方式这样系统提示出来直接就能当提词器用。5.2 基于历史数据的选品建议另外一个很实用的功能是选品建议。系统会根据历史直播中每个品类的平均转化率、客单价、退款率如果有数据的话算出一个“推荐指数”帮助你在直播前决定主推哪些品。推荐指数 转化率 × 0.4 客单价 × 0.2 库存深度 × 0.2 历史销量 × 0.2这只是一个经验公式你可以按自己的业务逻辑调整权重。核心思路是——不要让“直觉”单独决定选品让数据参与决策过程。我见过太多主播凭感觉选品结果翻了车不是说感觉完全不靠谱而是数据能帮你增加确定性降低风险。6. 常见问题与排查技巧实录6.1 数据库不写入数据这算是我被问到最多的问题其实大部分情况是目录路径写错了。建议你把data/exports/这个目录放在项目根目录下并且先在代码里打印一下绝对路径确认程序能不能找到文件。还有一个容易被忽视的点就是 Excel 文件的列名必须和代码里的字段名一致不能是“商品名称”或“product name”这种变体否则 Pandas 解析出来就是空列。6.2 前端图表不显示图表不显示的问题90% 是接口返回的数据格式不对。ECharts 对数据的格式非常挑剔如果data字段是字符串而不是数字图表就会白屏。建议你在浏览器开发者工具里打开 Network 面板看看接口返回的 JSON 里online_count是不是数字类型。如果发现是字符串就在后端做一次类型转换rows [dict(row, online_countint(row[online_count])) for row in rows]6.3 Flask 端口被占用这个也遇到过5000端口被其他程序占用了。解决办法很简单在app.run()里换一个端口比如 5001app.run(host0.0.0.0, port5001, debugTrue)注意要同步更新前端代码里请求接口的 URL 端口号不然页面还是打不开。6.4 中文乱码问题Excel 导入的时候中文乱码一般是编码问题。Pandas 读取 Excel 文件时需要用openpyxl引擎同时确保系统里安装了中文字体。如果是读取 CSV 文件需要指定编码为utf-8或者gbk具体看文件保存时的编码df pd.read_csv(file_path, encodingutf-8) # 如果报错就试 gbk df pd.read_csv(file_path, encodinggbk)6.5 指标计算结果异常如果你发现转化率超过 100% 这种明显不合理的数据先怀疑是不是重复导入导致的数据重复。我刚才在数据库设计部分提到过给session_id和product_id加唯一索引是很有必要的能在数据库层面拦住重复数据CREATE UNIQUE INDEX idx_unique_product ON product_performance(session_id, product_id);这个索引加上之后重复插入直接报错你就能第一时间发现问题而不是等指标算出来才发现不对。6.6 历史数据对比时的时间口径问题最后一个常见问题对比两场直播的数据时发现趋势对不上。原因通常是把不同时长的直播直接按分钟对比了。一场 2 小时直播的 60 分钟节点和一场 4 小时直播的 60 分钟节点流量结构完全不同。解决方法是把时间轴归一化成百分比也就是把直播总时长当作 100%看每个时间百分点的数据变化。这样不同时长的直播才有对比价值。这个功能我在源码里做了核心思路是计算每个记录点的总时长占比然后按占比重新采样。你如果自己改动代码记住这个口径要一致不然所有对比分析都会失真。7. 这套源码还能怎么扩展最后聊点我个人实际操作的体会和扩展方向。如果你已经把这套源码跑通了我建议下一步往这几个方向去升级。第一接入消息通知。把 Flask 后端加一个定时任务每 5 分钟检查一次数据如果转化率明显下降或者出现库存告警就通过 Webhook 推送到企业微信群或者钉钉群这样团队成员都能及时看到预警。第二增加多场次对比分析。现在的看板重点在单场直播其实把近 30 天所有场次的数据放在一起对比你会看到很多规律比如周日晚上 8 点到 10 点的转化率普遍高于周四下午这直接决定了你的排播策略。第三引入商品生命周期管理。同一个商品在不同场次的表现会变化第一场可能是福利款拉流量第二场可能是利润款收转化。通过追踪单品的历史表现曲线你就能更好地规划每个商品的角色定位。最后再分享一个小技巧我实际用下来这套系统最大的价值不是数据准确率高其实手工导出的 Excel 数据难免有延迟而是让你养成了“每场直播结束后 10 分钟看数据复盘”的习惯。数据复盘这种事坚持下去比工具本身重要得多。希望这套源码能帮你在直播带货的路上少踩几个坑多看透一些数据背后的门道。本文还有配套的精品资源点击获取