电商选品自动化监控方案:用 Python 定时抓取商品数据并筛选潜力爆款
做电商,选品是绕不开的基本功。很多中小卖家每天要花大量时间手动刷榜单、翻商品列表、比对价格和销量,效率低不说,还容易因为盯盘不及时错过新品红利。人工整理的数据很难持续积累,过几天再看,价格变了、销量涨了,之前的笔记对不上,很难形成有效的长期监控。
下面分享一套自己实际在用的轻量级自动化方案,无需复杂的页面逆向解析,也不用操心反爬虫限制。通过简单的 Python 脚本定时拉取商品数据,自动完成去重、筛选、存储,每天生成一份选品报表。代码可在本地跑起来,稍作修改就能适配不同的数据源和监控需求。
环境准备
本地需要 Python 3.8 及以上版本。安装以下依赖:
# 系统演示、API测试控制台:http://console.open.onebound.cn/console/?i=NewRookie
pip install requests pandas openpyxl schedule
一、人工选品的常见问题
-
效率瓶颈:手动翻页记录标题、价格、销量、评价数,一个类目整理下来动辄两三个小时,多个类目并行监控几乎不现实。
-
数据断层:今天看到的价格和明天不一样,无法自动留存历史数据,判断商品是短期冲量还是长期稳定款全靠印象。
-
标准难统一:面对海量商品,凭肉眼过滤掉转化低、竞争大、售后多的产品,主观性太强,容易看走眼。
-
新品滞后:平台上新、同行调价、活动降价等信息无法实时感知,等人工发现时往往已过了最佳切入时机。
用程序替代重复劳动,核心要解决的就是这几个点:批量获取结构化数据、定时执行、按规则筛选、历史可比对。
二、整体实现思路
整个流程拆成三层,各司其职:
-
数据获取层:通过调用第三方数据接口(或者官方开放 API)获取商品信息。传入关键词或类目 ID,返回结构化的商品列表,包含标题、价格、销量、评价数、店铺信息、发货地等字段。这样就不用自己写页面解析和反爬对抗,稳定性和可维护性都高得多。
-
数据处理层:Python 脚本对返回的数据进行清洗、去重,计算一些派生指标(比如预估月销量、价格波动幅度、竞争度打分),并按预设阈值筛选出潜力商品。
-
存储与对比层:将结果存入本地 Excel 或 SQLite 数据库,并保留每次采集的时间戳。定时任务每天执行一次,自动对比历史记录,发现价格暴跌、销量暴涨、上新等异动时打上标记,输出爆款候选清单。
这套方案把大部分精力放在数据处理和业务逻辑上,避开了爬虫维护的泥潭,适合长期稳定的类目监控。
三、代码实现
下面是可直接运行的脚本,做了适当的错误处理和日志输出。数据获取函数 fetch_products 是一个抽象接口,实际使用时需要替换成你自己的数据源(例如官方 API 或者合规的第三方数据服务)。
import requests
import pandas as pd
import schedule
import time
import logging
from datetime import datetime
# ---------- 日志配置 ----------
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# ---------- 配置区 ----------
# 监控的关键词列表,可自行增删
MONITOR_KEYWORDS = ["家用收纳盒", "厨房沥水篮", "桌面手机支架"]
# 潜力商品的最低月销量阈值
SALES_THRESHOLD = 500
# 输出报表文件名
OUTPUT_FILE = "选品监控报表.xlsx"
# ---------- 数据获取函数(需自行实现) ----------
def fetch_products(keyword, page=1, page_size=30):
"""
根据关键词获取商品列表。
返回一个 list,每个元素为 dict,包含商品的标准字段。
实际使用时,替换为具体的数据接口调用逻辑。
"""
# 示例:假设有一个商品搜索接口
# 注意:这只是示例,不可直接使用,需要替换为你的实际数据源
api_url = "https://api.example.com/search"
headers = {"Content-Type": "application/json"}
params = {
"keyword": keyword,
"page": page,
"page_size": page_size,
# "key": "your_api_key" # 如果需要鉴权
}
try:
resp = requests.get(api_url, params=params, headers=headers, timeout=15)
resp.raise_for_status()
data = resp.json()
# 根据实际返回结构提取商品列表
items = data.get("data", {}).get("items", [])
except Exception as e:
logging.error(f"获取关键词 '{keyword}' 数据失败: {e}")
return []
# 统一转换为标准格式,字段不足的给默认值
products = []
for item in items:
product = {
"采集时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"搜索词": keyword,
"商品ID": item.get("item_id", ""),
"标题": item.get("title", ""),
"售价": float(item.get("price", 0)),
"促销价": float(item.get("promo_price", 0)) if item.get("promo_price") else None,
"月销量": int(item.get("monthly_sales", 0)),
"累计评价": int(item.get("comment_count", 0)),
"店铺名": item.get("shop_name", ""),
"店铺类型": item.get("shop_type", ""), # 如:天猫、个人店
"发货地": item.get("ship_from", ""),
"主图链接": item.get("main_image", ""),
"类目路径": item.get("category_path", "")
}
products.append(product)
return products
# ---------- 数据筛选 ----------
def filter_hot_products(all_products):
"""按销量阈值过滤爆款,并输出统计"""
hot = [p for p in all_products if p["月销量"] >= SALES_THRESHOLD]
normal = [p for p in all_products if p["月销量"] < SALES_THRESHOLD]
logging.info(f"筛选完成,潜力爆款 {len(hot)} 个,普通商品 {len(normal)} 个")
return hot, normal
# ---------- 数据存储 ----------
def save_to_excel(hot_products, all_products):
"""写入 Excel,分为两个 Sheet"""
try:
with pd.ExcelWriter(OUTPUT_FILE, engine="openpyxl", mode="w") as writer:
df_all = pd.DataFrame(all_products)
df_hot = pd.DataFrame(hot_products)
df_all.to_excel(writer, sheet_name="全部商品", index=False)
if not df_hot.empty:
df_hot.to_excel(writer, sheet_name="潜力爆款", index=False)
logging.info(f"报表已保存至 {OUTPUT_FILE}")
except Exception as e:
logging.error(f"保存 Excel 失败: {e}")
# ---------- 单次监控任务 ----------
def run_once():
all_products = []
for kw in MONITOR_KEYWORDS:
logging.info(f"正在采集关键词: {kw}")
data = fetch_products(kw)
all_products.extend(data)
time.sleep(2) # 控制请求频率,避免触发限流
if not all_products:
logging.warning("未获取到任何商品数据,请检查数据源配置。")
return
hot, _ = filter_hot_products(all_products)
save_to_excel(hot, all_products)
# ---------- 定时任务 ----------
def schedule_task():
schedule.every().day.at("09:00").do(run_once)
logging.info("定时任务已启动,每天 09:00 自动执行选品监控...")
while True:
schedule.run_pending()
time.sleep(60)
if __name__ == "__main__":
# 选择运行模式:
# 1) 手动执行一次
run_once()
# 2) 启动定时任务(需注释掉上面那行)
# schedule_task()
使用说明
-
实现数据获取函数:上面的
fetch_products是占位的,你需要根据实际对接的 API 来完善它。常见的来源包括:-
电商平台的官方开放 API(需申请开发者权限)
-
经过授权的第三方数据服务(注意合规性)
-
如果是自建爬虫,则需在该函数内处理请求和解析,但仍建议将获取和解析分离。
-
-
调整监控关键词:修改
MONITOR_KEYWORDS列表,放入你主营的类目词。 -
设定筛选阈值:
SALES_THRESHOLD根据类目特点调整,小众类目可以设低些(如 200),标品大类目可以设高些(如 1000)。 -
运行方式:直接运行脚本会执行一次采集并生成报表;如需每日自动运行,注释掉
run_once(),启用schedule_task(),让程序在后台常驻即可。也可以配合系统的计划任务(如 cron)来每日执行单次脚本,这样更轻量。
四、进阶玩法:让数据更有价值
1. 价格趋势与异动监控
每次采集都会带上时间戳,持续运行几天后,你可以用 Excel 的数据透视表或直接在 pandas 里按商品 ID 分组,观察同一商品的价格走势。举个例子,某款收纳盒连续三天售价为 29.9、29.9、19.9,说明正在进行降价促销,可能值得关注。
更进一步,可以在脚本里增加异动告警:
def price_change_alert(new_data, old_file):
# 读取昨日报表,对比同商品ID的售价和月销量
# 价格跌幅超过20% 或 销量增长超过50% 时输出告警信息
pass
2. 竞争度分析
利用返回的店铺类型、发货地、类目路径等字段,可以快速估算一个品类的竞争态势。比如统计同一类目下天猫店铺的占比,如果超过 70%,说明品牌集中度高,小卖家切入难度大。或者按发货地分组,看货源是否集中在某几个城市,侧面反映供应链的集中程度。
3. 新品识别
定时任务每天采集,通过比对今日和昨日的商品 ID 集合,新增的 ID 即为当天上架的新品。对这些新品可以单独记录下来,持续跟踪 3-7 天的销量增速,很多蓝海爆款往往就出现在这类刚冒头的商品里。
4. 数据持久化
Excel 适合日常查看,但当数据量大、要做复杂查询时,建议将每次采集的结果存入 SQLite 数据库。表结构可以很简单:
CREATE TABLE products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
crawl_time TEXT,
keyword TEXT,
item_id TEXT,
title TEXT,
price REAL,
monthly_sales INTEGER,
...
);
然后用 SQL 就能轻松计算每个商品的历史最高/最低价、销量均值等,比 Excel 灵活得多。
五、注意点
-
请求频率:对接任何数据源都要注意调用频率限制,在
fetch_products里已经加了 2 秒的间隔,根据实际 API 的 QPS 调整。 -
数据源稳定性:接口可能不定时调整字段或返回结构,代码中要做好兼容,对缺失字段给默认值,避免程序崩溃。
-
合规使用:获取数据时遵守平台协议,不要抓取未公开的信息或用于违规用途。
-
阈值灵活调整:不同季节、不同类目的销量基准差异很大,建议每周根据实际情况微调
SALES_THRESHOLD。
整套方案把重复的人工浏览和记录工作转变为程序自动化执行,每天花几分钟看报表就能掌握多个类目的动态。代码是开箱即用的基础版,可以根据自己的业务需求进一步扩展,比如接入更多数据维度、加入利润率计算、对接企业微信/钉钉告警等。选品拼的终究是信息获取和分析的效率,用对工具能省下大量时间,让精力集中在选品决策本身。
更多推荐




所有评论(0)