做电商数据分析的完整链路很多人只做了半截。有人写爬虫把商品数据抓下来存成Excel就收工有人做可视化直接拿别人的表格画几个图交差。但真实业务里运营更想知道的是“这款商品凭什么卖爆”“定价调高十块钱销量会掉多少”“哪几个商品需要盯紧价格”这需要采集、建模、展示三件事真正连起来。去年我动手把这条链路完整实现了一遍用Selenium采集公开商品数据用多元线性回归做销量影响分析用Flask把数据库和模型封装成接口最后投到可视化大屏上做实时监控。整套系统技术栈很亲民——Python Flask Selenium ECharts代码量不夸张难的是流程怎么串。这篇就把整个项目的设计思路、关键代码、踩坑记录一次性写透想练手完整项目的Python学习者、要交课程设计或毕设的同学、用数据辅助选品定价的运营都能从这里拿走能直接改的东西。1. 项目整体拆解这套系统到底解决了什么问题1.1 做之前先搞清楚数据要回答什么问题很多爬虫项目死在第一步不是因为不会写采集代码而是不知道采集完要干什么。这个项目开始前我把需求收敛成三个具体问题第一一组关键词下哪些商品的销量表现最好这个月环比有什么变化第二价格、评价数、店铺评分这些公开字段里哪些因素对销量影响最大影响方向是什么第三每天定时采集一次当有商品价格出现明显异动或销量异常飙升时能不能在大屏上直接把预警顶出来。这三个问题直接决定了系统结构。第一个问题需要采集层和数据库第二个问题需要建模层第三个问题需要定时任务、接口层和展示层。千万别一上来就写爬虫先把要回答的问题列出来后面每一步都有据可依。1.2 系统架构五层拆解整个系统按数据流方向分成五层每层只干一件事采集层Selenium驱动浏览器打开平台公开搜索页滚动加载商品列表提取商品标题、价格、销量、店铺、所在地等字段写成结构化数据。存储层用SQLite做本地存储一张表存商品基础信息一张表存每次采集的价格和销量快照方便后面画趋势线。建模层用statsmodels做多元线性回归把销量作为因变量价格、评价数、店铺评分等作为自变量输出系数和显著性并保存模型供接口调用。服务层Flask提供数据接口前端页面从这里取数同时挂一个预测接口输入商品特征返回预测销量。展示层ECharts做可视化大屏布局采用三栏式中间放核心指标两侧放趋势、排行、占比和地域分布。这套分层的好处是每一层都可以单独替换。比如今天用Selenium明天想换成官方API只要保持输出表格结构不变上层完全不用动。建模层也是线性回归跑明白了后续换XGBoost只是替换一个函数的事。1.3 技术选型为什么不赶时髦选的都是成熟方案有人问为什么不用Django、不用Scrapy、不用深度学习这里把选型逻辑摊开说。先说Flask和Django。这个项目本质是“几个数据接口 一个展示页面”Flask的轻量完全够用路由写起来直观部署也简单。Django适合需要用户体系、后台管理、ORM大量模型的复杂应用对这个小项目来说偏重。FastAPI性能好但生态和资料不如Flask老牌课程设计或入门项目选FastAPI反而给自己添堵。再说Selenium和Scrapy。平台商品页大量依赖动态渲染很多字段是JavaScript异步加载出来的Scrapy默认拿不到渲染后的DOM需要额外接Splash或Playwright链路复杂。Selenium直接驱动浏览器所见即所得还能模拟滚动、点击“加载更多”这类交互。缺点也明显——慢一个关键词几十个商品要几分钟但采集频率控制在每天一次这个开销完全可接受。最后说多元线性回归。为什么不直接上LightGBM或深度学习因为这个小项目的数据量就几千条树模型容易过拟合深度模型解释性差而业务方要的不是准确率99%的黑盒他们要的是一个能说清楚的结论“评价数每增加100条销量大概涨多少”。线性回归的系数天然具备可解释性这也是statsmodels输出p值和置信区间的原因。技术本项目的选型不选它的理由Flask轻量路由和开发效率高Django太重FastAPI生态不够老牌Selenium动态页面渲染后采集模拟真实用户操作Scrapy抓不到异步加载内容需二次对接SQLite单机运行、不需要独立数据库服务MySQL部署复杂数据量没必要上statsmodels可直接输出回归表、p值、置信区间sklearn的LinearRegression解释性输出太弱ECharts大屏图表适配成熟社区案例多其他JS图表库在大屏布局上资料少2. Selenium采集层从抓取到清洗的完整链路2.1 为什么绕不开Selenium平台商品页只要搜索一个关键词页面会先加载骨架屏然后通过异步接口把商品卡片渲染出来。requests直接拿到的HTML里面连商品标题都找不到因为数据是脚本执行后填充的。Selenium的思路就是打开一个真实浏览器等页面渲染完再像人一样滚动、读取、翻页。虽然笨重但在“必须拿到渲染后数据”的场景下是最直接的办法。我用的是Chrome的headless模式不弹浏览器窗口在服务器上也能跑。Selenium版本升到4.x以后不需要再单独引webdriver_manager也能跑但建议还是用webdriver_manager自动匹配浏览器版本省去手动下载驱动的麻烦。2.2 一个可复用的商品列表采集脚本下面这个脚本以某大型电商平台公开搜索页为例只采集无需登录就能看到的列表信息。我第一次跑的时候是直接打开浏览器手动看页面结构的F12找到商品卡片的公共CSS路径再写定选择器。from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.chrome.options import Options from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import time import random options Options() options.add_argument(--headless) options.add_argument(--no-sandbox) options.add_argument(--disable-gpu) options.add_argument(--window-size1920,1080) options.add_argument(--disable-blink-featuresAutomationControlled) driver webdriver.Chrome(optionsoptions) wait WebDriverWait(driver, 10) def search_and_collect(keyword, max_scroll5): url https://搜索页地址/search?q keyword driver.get(url) # 等待商品列表出现 wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, .商品列表容器选择器))) # 模拟滚动触发懒加载 for _ in range(max_scroll): driver.execute_script(window.scrollBy(0, document.body.scrollHeight);) time.sleep(random.uniform(1, 2)) items driver.find_elements(By.CSS_SELECTOR, .商品卡片选择器) result [] for item in items: try: name item.find_element(By.CSS_SELECTOR, .标题选择器).text.strip() price item.find_element(By.CSS_SELECTOR, .价格选择器).text.strip() sales item.find_element(By.CSS_SELECTOR, .销量选择器).text.strip() shop item.find_element(By.CSS_SELECTOR, .店铺选择器).text.strip() location item.find_element(By.CSS_SELECTOR, .所在地选择器).text.strip() result.append({ keyword: keyword, name: name, price: price, sales: sales, shop: shop, location: location, collected_at: time.strftime(%Y-%m-%d %H:%M:%S) }) except Exception: continue return result几个让我踩过坑的细节滚动间隔不能写死加random.uniform(1, 2)模拟真人节奏每次循环重新查找元素不要缓存旧的元素列表因为滚动后DOM会更新旧引用会失效headless模式下窗口大小必须设置不然某些版本Chrome渲染结果和头戴模式不一致导致元素定位不到。2.3 数据清洗字段规范化比爬数据更费时间页面爬到的price字段看起来是“¥1,299.00”sales字段是“已售3万”这些字符串直接进数据库没法分析。清洗逻辑必须写在这里否则后面建模全是坑。def clean_price(raw): if not raw: return None raw raw.replace(¥, ).replace(, ).replace(,, ).strip() try: return float(raw) except ValueError: return None def clean_sales(raw): if not raw: return 0 raw raw.replace(已售, ).replace(, ).strip() if 万 in raw: return int(float(raw.replace(万, )) * 10000) if 亿 in raw: return int(float(raw.replace(亿, )) * 100000000) digits .join(filter(str.isdigit, raw)) return int(digits) if digits else 0清洗完再补一列“是否包邮”和“是否品牌旗舰店”这两个字段模型里有用。我实际采集过程中发现有些店铺名带“旗舰店”字样的商品普遍比普通店铺销量高一截做成布尔特征后回归系数会很显著。2.4 合规与反爬不是教你绕过而是教你体面地采这个项目对外说话必须讲清楚边界。我只请求不需要登录的公开页面控制请求频率随机间隔至少1秒以上不给对方服务器增加压力。如果遇到验证码或滑块验证脚本立即停止并记录日志绝不尝试绕过。生产环境建议优先使用平台官方开放接口或找有授权的数据服务商个人学习项目以少量低频公开数据为限即可。反爬策略上要用巧劲而非蛮力。加随机延迟比加代理池有效且稳妥登录态能不用就不用登录接口本身就涉及更多合规风险。采集频率设成一天一次比一下子抓几万条再存起来更符合“监控”这个场景的定位。3. 多元线性回归建模把采集数据变成可解释的决策依据3.1 横截面数据怎么做回归这个模型的目的是回答“已经采集到的这些商品哪些特征能解释销量差异”。这是典型的横截面回归问题每一行是一个商品因变量是销量自变量是商品属性。多元线性回归假设销量和特征之间存在线性关系虽然现实世界不完全线性但在局部分析中足够给出方向性结论而且样本量不大时比复杂模型更稳。这里有个认知要端正回归系数不能直接解读为因果关系。比如“价格系数为负”不能简单说“降价就一定能提升销量”因为可能存在价格和品质、品牌等因素的混杂。它能做的是告诉你统计关联的方向和强度辅助决策。3.2 特征工程实操原始采集字段不能直接用我整理成下面的特征表特征类型处理方式price连续数值清洗后直接入模total_comment连续数值评价数取对数压缩量纲shop_score连续数值缺失值填所在类目平均值free_shipping0/1二值“包邮”映射为1is_brand0/1二值店铺名含“旗舰店/官方”映射为1location分类pandas get_dummies生成哑变量评价数取对数这条很多人会忽略。评价数和销量都带着长尾分布直接放进线性回归个别爆款会被当成异常值拉扯系数。取log之后分布更接近正态模型效果立刻明显好转。3.3 statsmodels建模与评估用statsmodels而不是sklearn是因为它能直接输出回归报告表格。我在Jupyter里跑完summary()会把系数、标准误、t值、p值、R方、F统计量全部列出来省去手动计算验证。import pandas as pd import numpy as np import statsmodels.api as sm from sklearn.model_selection import train_test_split df pd.read_csv(goods_data.csv) df[price] pd.to_numeric(df[price], errorscoerce) df[sales] pd.to_numeric(df[sales], errorscoerce) df[total_comment] pd.to_numeric(df[total_comment], errorscoerce) df[log_comment] np.log1p(df[total_comment]) df[free_shipping] (df[free_shipping] 是).astype(int) df[is_brand] df[shop].str.contains(旗舰店|官方, naFalse).astype(int) df pd.get_dummies(df, columns[location], drop_firstTrue) feature_cols [price, log_comment, shop_score, free_shipping, is_brand] \ [c for c in df.columns if c.startswith(location_)] df df.dropna(subset[price, sales]) X df[feature_cols].fillna(0) y df[sales].fillna(0) X_train, X_test, y_train, y_test train_test_split(X, y, test_size0.2, random_state42) X_train_const sm.add_constant(X_train) model sm.OLS(y_train, X_train_const).fit() print(model.summary())模型跑出来的结果大概长这样基于我实际采集的数据变量系数p值说明price-86.30.002价格每提高1元销量平均少约86件log_comment1240.50.000评价数对数每增加1个单位销量涨得明显free_shipping452.10.031包邮商品销量平均高约450件is_brand983.40.000品牌店铺销量平均高约983件R²0.68-特征解释了68%的销量差异负价格系数对运营有直接参考价值log_comment系数说明“口碑积累”比单纯降价更能带动销量。R方0.68在横截面回归里属于可接受水平毕竟真实的销量还受促销、季节、流量投放影响这些字段平台不会公开。3.4 模型落地保存、加载、复用模型不能只留在Jupyter里要保存成文件供Flask调用。statsmodels的模型可以用joblib直接序列化。import joblib joblib.dump(model, models/sales_model.pkl)以后每次增量采集完数据重新训练一次并覆盖保存即可。模型的更新周期我设置为每周一次这样预测接口用的参数不会长期停留在“上周的认知”。4. Flask后端接口让数据库和模型变成可调用的服务4.1 项目目录结构与数据库设计采集和建模都在脚本里跑但Web端必须要有一个常驻服务把数据带出来。Flask在这个项目里的角色很单纯读SQLite转JSON加载模型做预测托管静态大屏页面。project/ ├── app.py # Flask 入口 ├── crawler.py # Selenium 采集脚本 ├── train_model.py # 训练并保存模型 ├── models/ │ └── sales_model.pkl # 序列化后的模型 ├── data/ │ └── goods.db # SQLite 数据库 ├── static/ │ ├── css/ │ └── js/ └── templates/ └── dashboard.html # 大屏页面数据库两张表。goods表存商品基础信息price_snapshots存每次采集的价格和销量快照两张表通过商品去重ID关联。这样既能看当前榜单也能画价格趋势。CREATE TABLE goods ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT, name TEXT, price REAL, sales INTEGER, shop TEXT, location TEXT, free_shipping INTEGER, is_brand INTEGER, first_seen TEXT ); CREATE TABLE price_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, goods_name TEXT, price REAL, sales INTEGER, captured_at TEXT );4.2 核心接口设计接口设计围绕大屏的四个模块来。汇总指标给大屏顶部KPI卡趋势数据给价格折线图排行数据给销量柱状图地域数据给分布地图。多一个预测接口给前端做“如果调价预估销量变化”的测算。接口方法返回内容/api/summaryGET商品总数、平均价、总销量、今日采集数/api/trend?keywordxxxGET某关键词商品近30天平均价格和销量走势/api/rank?limit10GET销量TOP10商品列表/api/locationGET商品所在地分布统计/api/predictPOST传入价格、评价数等特征返回预测销量summary接口的实现非常直接from flask import Flask, jsonify, request from flask_cors import CORS import sqlite3 import pandas as pd import joblib app Flask(__name__) CORS(app) model joblib.load(models/sales_model.pkl) def query_df(sql): conn sqlite3.connect(data/goods.db) df pd.read_sql(sql, conn) conn.close() return df app.route(/api/summary) def summary(): df query_df( SELECT COUNT(*) AS total, AVG(price) AS avg_price, SUM(sales) AS total_sales FROM goods ) return jsonify(df.to_dict(orientrecords))predict接口需要注意的一点前端传来的字段必须和训练时保持一致连特征顺序都不能乱。我的处理是前端传来一个JSON对象后端用固定顺序的feature_cols列表重新构造二维数组再调用model.predict。另外statsmodels模型用predict需要带常数项我封装了一层app.route(/api/predict, methods[POST]) def predict(): data request.get_json() # 按训练时的特征顺序构造向量 row [ float(data[price]), float(data[log_comment]), float(data[shop_score]), int(data[free_shipping]), int(data[is_brand]) ] # 这里假设只取这5个特征location哑变量在简版接口中省略 import numpy as np X_new np.array([row]) X_new_const sm.add_constant(X_new, has_constantadd) pred model.predict(X_new_const)[0] return jsonify({pred_sales: round(float(pred), 2)})4.3 前后端对接的小细节跨域问题直接用Flask-CORS解决把CORS(app)挂在入口大屏页面和数据接口不在同一域名时不会报错。SQLite被Flask多线程访问时可能会报database is locked我在连接时加了check_same_threadFalse并设置PRAGMA journal_modeWAL读写并发基本不冲突。还有一个很隐蔽的坑pandas的to_dict会把NaN转成NaNJSON序列化会失败接口里需要fillna(None)再返回。5. 大屏可视化把接口数据变成实时监控看板5.1 大屏布局思路大屏不是把所有图表都铺在一张页面上需要有信息层级。我采用常见的16:9三栏布局中间顶部是核心KPI大数字中间腰部是价格销量走势大图左右两栏分别是销量排行、类目占比、地域分布底部留一条滚动的预警信息栏。页面用CSS Grid分区宽度按百分比和rem适配保证在1920宽的显示器上不被拉伸变形。背景用深色渐变图表用亮色高亮这是监控类大屏的通用风格。ECharts本身不提供整体主题我通过自带的color数组统一设置一套品牌色避免每个图表各用各的颜色。5.2 ECharts核心图表配置大屏页面里引用CDN的ECharts然后按接口数据渲染图表。核心是折线图和柱状图配置不算复杂关键是init时要把图表实例存起来后面轮询更新数据时复用同一个实例调用setOption更新而不是反复创建销毁。const trendChart echarts.init(document.getElementById(trendChart)); async function fetchTrend() { const res await fetch(/api/trend?keyword运动鞋); const data await res.json(); trendChart.setOption({ tooltip: { trigger: axis }, grid: { left: 60, right: 30, top: 40, bottom: 40 }, xAxis: { type: category, data: data.map(d d.captured_at) }, yAxis: [ { type: value, name: 价格 }, { type: value, name: 销量 } ], series: [ { name: 平均价格, type: line, data: data.map(d d.price), smooth: true, lineStyle: { width: 3 } }, { name: 总销量, type: bar, yAxisIndex: 1, data: data.map(d d.sales), itemStyle: { color: #2f6bff, opacity: 0.6 } } ] }); }大屏还有个关键点图表容器必须有明确高度ECharts初始化时容器隐藏或高度为0渲染出来就是空白。我在初始化前先确保页面布局完成或者直接给容器写死高度。5.3 实时刷新与异常预警大屏的“实时”用轮询实现。setInterval每30秒请求一次所有接口把最新数据setOption进去。WebSocket在小项目里没有必要30秒级别的刷新用HTTP轮询够用还省掉了Socket连接管理的复杂度。预警逻辑放在后端更合适前端只负责展示。我在Flask里加了一个/api/warnings接口用SQL查最近两次快照的销量和价格变化率变化超过阈值就返回预警列表app.route(/api/warnings) def warnings(): df query_df( SELECT name, price, sales, LAG(price) OVER (ORDER BY captured_at) AS prev_price, LAG(sales) OVER (ORDER BY captured_at) AS prev_sales FROM price_snapshots ) df[price_change] (df[price] - df[prev_price]) / df[prev_price] df[sales_change] (df[sales] - df[prev_sales]) / df[prev_sales] warn df[(df[price_change].abs() 0.15) | (df[sales_change] 0.3)] return jsonify(warn.to_dict(orientrecords))前端轮询这个接口后如果有数据就用列表滚动组件把预警信息投到底部。这里用CSS动画实现无缝滚动不用引入额外插件。6. 定时采集与部署让系统在生产环境真正跑起来6.1 用APScheduler给系统加上自动采集任务人工每天跑一次爬虫不现实得让系统自己干活。APScheduler是Python生态里最常用的定时任务库和Flask集成方式简单。from apscheduler.schedulers.background import BackgroundScheduler def scheduled_job(): keywords [运动鞋, 手机壳, 保温杯] for keyword in keywords: data search_and_collect(keyword) save_to_db(data) retrain_model_if_needed() scheduler BackgroundScheduler() scheduler.add_job(scheduled_job, cron, hour2, minute30) scheduler.start()这里有个非常容易踩的坑Flask在debug模式下会启动两个进程一个监听一个reloaderAPScheduler会跟着跑两遍导致定时任务重复执行。解决方法是判断当前进程是否为主进程或者直接用app.run(debugFalse)启动开发时可以接受生产环境用gunicorn就不会有这个问题。6.2 打包部署与服务器运行部署环境我建议用Linux服务器加gunicorn。Selenium需要Chrome和ChromeDriver服务器上也要装好。先把依赖导出pip freeze requirements.txt然后在服务器上用gunicorn起服务gunicorn -w 1 -b 0.0.0.0:5000 app:app注意worker数量写1。这个系统本身就是单个Flask进程如果开多个workerAPScheduler定时任务会被启动多份而且SQLite并发写也会被锁折磨。数据规模不大时单worker完全够用。前端大屏页面Flask直接托管不需要单独配Nginx如果以后要上HTTPS或并发增加再把静态文件交给NginxFlask只留API。6.3 运行期常见坑位与排查建议把我在实际运行里遇到的一批问题整理成表给后来人省点排查时间现象原因处理方案SessionNotCreatedExceptionChromeDriver版本和Chrome版本不匹配用webdriver_manager自动匹配headless模式下定位不到元素窗口尺寸太小导致懒加载逻辑异常设置--window-size1920,1080sqlite3.OperationalError: database is locked多线程同时写SQLite连接参数check_same_threadFalse开启WAL接口返回NaN导致前端渲染空pandas的NaN序列化失败返回前df.fillna(None)APScheduler任务重复执行Flask reloader加载两次进程生产用gunicorn单worker中文乱码采集时编码不一致文件头声明utf-8数据库连接加charset参数这些坑单个看都不大但会在你第一次完整部署时连环爆。我的建议是先本地跑通全链路再上服务器每上一次就记一次坑比什么都管用。7. 复盘这个项目带给我最值钱的经验如果让我重做一遍我会把更多时间花在特征工程上而不是纠结爬虫代码优化。第一次做的时候我把大量时间花在让Selenium更快、更稳后来发现采集频率一天一次快十分钟慢十分钟根本没区别反倒是把“评价数取对数”“是否品牌店”这种特征处理好之后模型R方从0.45涨到0.68那才是真正影响项目价值的部分。另一个体会是接口返回的数据结构一定要在设计大屏之前定好。我一开始是前端需要什么字段后端就临时加接口改来改去浪费了不少时间。后来学乖了先画出大屏原型再把每个图表要的数据列成清单后端一次性把接口出好效率高很多。最后是拆分的价值。整个项目分成采集、建模、服务、展示四块每一块都能单独测试。出问题的时候先看日志在哪个模块几秒钟就能定位。不要学网上有些教程把所有代码堆在一个文件里看起来“一键运行”实际上排查问题的时候会非常痛苦。这套系统的完整源码我已经整理好了目录结构就是第4章展示的那个样子代码片段也可以直接复制到自己的项目里改。想动手的话先从搜索一个关键词跑通采集开始然后慢慢加上建模和可视化整个流程走一遍你对“数据产品”这个概念的理解会比看十篇教程都深刻。