把蹲点抢购这事儿交给Python是我最近半年做得最值的一个决定。事情是这样的我一直在关注几款数码产品和家电的价格想等一个合适的入手时机。但手动刷新页面太累了而且很多商品是半夜改价等你早上看到便宜的时候早就过去了。后来我干脆写了一个Python脚本通过淘宝关键词API定期搜索这些商品的实时价格把价格数据存下来再设置一个心理阈值价格一旦低于预期就自动推送提醒到我的钉钉。这套逻辑跑通之后我基本不用再盯页面了该干嘛干嘛等到手机弹通知再去看一眼下单就行。这篇文章就把这套方案的完整思路和代码实现拆开讲一遍。核心就三件事怎么通过淘宝关键词API去搜商品、怎么把价格数据存成历史记录、怎么根据价格变化触发预警通知。不管是想学Python接口调用的新手还是想做个竞品价格监控的电商运营都可以直接照着抄。1. 为什么我决定用Python做价格监控1.1 手动盯价的痛点想必你也遇到过先说一个很常见的场景。你购物车里躺着一件商品比如一台两千多的显示器。你知道它历史低价在1800左右但你没时间天天看。于是你每天中午打开App看一眼结果连续一周都是原价。某天晚上你加班到十一点回家刷手机发现这件显示器搞限时秒杀只要1799但活动只剩半小时——你一边付款一边骂自己为什么没早看。这种经历我反复遇到而且我发现电商平台上先涨价再降价的玩法太普遍了。页面显示限时特惠四个字你以为捡了大便宜实际上比上个月的基本盘还贵。这时候最需要的就是历史价格数据这个商品过去三个月到底卖多少钱当前价格处于什么位置。没有数据你所谓的便宜和贵都是凭感觉。所以我的需求变得很明确第一自动搜索指定关键词下的商品第二定期记录每个商品的价格、销量、店铺信息第三当价格满足某个条件低于心理价、低于历史最低价、降价幅度超过一定比例时给我发一个通知。注意这个方案不是为了薅羊毛抢限时补贴而是建立一个长期观察商品价格的自动化机制弥补人工盯价的各种盲区。1.2 爬虫和API我为什么选了后者市面上的技术方案其实就两条路用爬虫直接抓淘宝页面或者调用淘宝开放API。爬虫这条路我最早也试过。直接用requests请求商品搜索页能拿到HTML源码但问题一堆。淘宝的反爬机制升级很频繁初次访问可能正常翻几页之后就要求滑块验证再往后IP会被临时限制。你就算用上Selenium模拟浏览器也扛不住登录态和验证码的连环折腾。就算你费老大劲把数据抓下来了页面结构一改版解析代码当场作废这种方案太脆了。API方案就稳得多。淘宝开放平台以及第三方数据服务商提供了标准化的接口比如通过关键词搜索商品、获取商品详情、解析优惠信息。你只要带上合法的app_key和app_secret按照文档拼接参数、生成签名发起HTTP请求就能拿到结构化的JSON数据。接口返回的字段是固定的不会动不动就变也没有验证码的烦恼。稳定性和开发效率都远超爬虫。这两种方案的对比如下表对比维度自建爬虫淘宝API开发成本高需处理登录、验证码、页面解析低按文档请求即可稳定性差页面改版或风控升级就挂好接口字段固定数据质量需要自己清洗截取返回结构化字段合规风险较高可能违反平台规则相对可控走官方/第三方授权费用主要是IP和人工成本按调用量计费初期成本低我最终选择的是第三方数据服务商提供的淘宝关键词搜索API。这里需要说明一下淘宝开放平台对个人开发者申请商品搜索类接口的门槛比较高而市面上一些正规的数据服务商把这类能力做成了标准化的付费API个人开发者可以比较方便地申请使用。这类服务本质上还是围绕淘宝商品数据做整合前提是你自己要有合法的应用场景。下面所有代码我以这类通用API为例来写。2. 动手前的准备环境、依赖与API申请2.1 Python环境与依赖这个项目对Python版本没有太苛刻的要求3.8以上都行。我自己用的是3.10Windows上开发的后来部署到一台Linux小主机上跑代码完全没改。如果你是刚接触Python建议先装好Python解释器再把pip的国内镜像配置一下不然装包速度很痛苦。Windows用户在命令行执行python -m pip install -i https://pypi.tuna.tsinghua.edu.cn/simple requests apscheduler pandasLinux下如果系统自带的Python版本偏低可以先升级或者用apt安装新版再执行同样的命令。依赖方面核心就三个库requests发HTTP请求、apscheduler定时调度、pandas可选做价格数据分析时方便。数据库我直接用Python内置的sqlite3连第三方库都不用装。另外强烈建议用虚拟环境尤其是你机器上还有别的Python项目时。我用的是venvpython -m venv price_env # Windows激活 price_env\Scripts\activate # Linux激活 source price_env/bin/activate跑在独立环境里避免依赖冲突这是最基本的工程习惯。别省这一步以后有你省心的时候。2.2 淘宝关键词API的申请与鉴权市面上提供淘宝关键词API的服务商有好几家申请流程大同小异。一般是注册账号、完成企业或者个人实名认证、在控制台创建一个应用然后系统会给你一对密钥——app_key和app_secret。app_key相当于你的身份标识app_secret是签名密钥绝对不能泄露到前端或者公开仓库里。如果你打算把代码发布到GitHub请一定把密钥放到config文件里然后加入.gitignore。有了这对密钥之后调用API时还需要传入一个重点参数方法名。以关键词搜索商品为例方法名一般是item_search。接口的通用请求参数包括app_key应用标识method方法名这里就是item_searchkeyword搜索关键词比如无线机械键盘page页码page_size每页返回条数有些服务商限制单页最多40sign签名串用于身份校验签名算法是所有第三方API的通用逻辑把除sign以外的所有请求参数按参数名字母升序排列拼成key1value1key2value2的形式然后拼接上app_secret最后做MD5得到的字符串就是签名。以q机械键盘和page1两个参数为例拼接逻辑就是先把参数按字母排好再和密钥一起做MD5摘要。给你一段通用的签名函数代码几乎可以直接移植到任何一家API上import hashlib import requests APP_KEY 你的app_key APP_SECRET 你的app_secret API_URL https://api.xxx.com/api # 以实际服务商为准 def generate_sign(params: dict) - str: # 1. 过滤掉空值 filtered {k: v for k, v in params.items() if v ! and v is not None} # 2. 按key字母升序 sorted_keys sorted(filtered.keys()) # 3. 拼接成key1value1key2value2 plain_text .join(f{k}{filtered[k]} for k in sorted_keys) # 4. 拼接app_secret后再做MD5 plain_text plain_text APP_SECRET return hashlib.md5(plain_text.encode(utf-8)).hexdigest().upper()注意签名里最后做了个大写处理这是因为有些服务商要求签名全部大写。如果实际对接过程中被告知sign error大概率是大小写问题或者参数漏传了。另外app_secret串到待签名的字符串里时不需要URL编码这一点比较容易踩坑。2.3 请求封装与通用返回结构所有接口的调用方式都是一样的无非是换方法名和参数。所以我习惯把请求逻辑封装成一个统一函数这样后面不管是搜索商品还是获取商品详情都直接复用同一套签名和请求逻辑。def call_taobao_api(method: str, params: dict) - dict: request_params { app_key: APP_KEY, method: method, **params } request_params[sign] generate_sign(request_params) resp requests.get(API_URL, paramsrequest_params, timeout10) resp.raise_for_status() result resp.json() if result.get(code) ! 200: raise Exception(fAPI error: {result.get(code)} {result.get(msg)}) return result[data] if isinstance(result, dict) else result注意这里有一个比较实用的细节有些数据服务商的返回字段名有差异有的叫price有的叫current_price有的带了原始促销价。你在初期联调时最好先把返回的JSON完整打印一遍看清楚字段再写解析逻辑不要想当然。3. 核心代码搜索商品、解析价格与入库3.1 数据模型设计既要存当前价也要存历史价有了API能力之后下一步就是设计数据怎么存。我的方案是两张表商品表和价格历史表。商品表存的是商品的相对静态信息比如标题、商品ID、店铺名称、商品链接这些字段不会频繁变化重复搜索同一商品时直接做去重。价格历史表存的是每次抓取到的价格、原价、销量和时间点一个商品在历史表中会有多行记录代表它在不同时间点的价格快照。这两张表拆开是很有必要的。如果只建一张表每插入一条新价格记录就要重复存一遍商品标题和链接浪费空间不说以后想分析某个价格区间的商品中销量最高的有哪些这种问题SQL写起来也会很繁琐。对应的建表语句如下CREATE TABLE IF NOT EXISTS product ( item_id TEXT PRIMARY KEY, title TEXT, shop_name TEXT, url TEXT, first_seen_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, item_id TEXT, price REAL, orginal_price REAL, sales INTEGER, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );price字段之所以用REAL严格来说应该用DECIMAL避免浮点误差但SQLite对DECIMAL的支持比较弱所以我在Python层统一用Decimal做计算入库时转成float查询做比较时也走Decimal这样双保险。3.2 调用关键词API拉取商品列表接下来是核心的拉取逻辑。我定义一个函数入参是关键词循环拉取前几页数据把每件商品的详情写入数据库。import time import sqlite3 def scrape_keyword(keyword: str, max_pages: int 3): conn sqlite3.connect(price_monitor.db) for page in range(1, max_pages 1): data call_taobao_api(item_search, { keyword: keyword, page: page, page_size: 40 }) items data.get(items, []) if not items: break for item in items: item_id item.get(num_iid) or item.get(item_id) if not item_id: continue title item.get(title, ) price item.get(price) or item.get(current_price) orginal_price item.get(orginal_price) or item.get(original_price) or price sales int(item.get(sales) or 0) shop_name item.get(shop_name, ) url item.get(detail_url) or fhttps://item.taobao.com/item.htm?id{item_id} conn.execute( INSERT OR IGNORE INTO product (item_id, title, shop_name, url) VALUES (?, ?, ?, ?), (item_id, title, shop_name, url) ) conn.execute( INSERT INTO price_history (item_id, price, orginal_price, sales) VALUES (?, ?, ?, ?), (item_id, float(price), float(orginal_price), sales) ) conn.commit() print(f[{keyword}] 第{page}页完成累计{len(items)}条) time.sleep(1) # 控制请求频率防止限流 conn.close()有几个很关键的处理细节我必须说明。第一price和orginal_price字段名在每个服务商那里不一样我用or逐个尝试。这是长期踩坑踩出来的经验不要假设一个字段一定存在也别假设它是字符串还是数字。第二sales直接int转换前先判断是否为None有些非热销商品可能没有销量字段。同理如果接口返回的是已售100这种带文字的量级需要自己写正则把数字提取出来。第三INSERT OR IGNORE是SQLite的去重写法第一次见到这个商品就写入product表price_history表则每轮都插入一条。这样同一件商品在历史表里有多条价格快照画趋势图、算最低价都靠它。3.3 解析返回数据与价格清洗你以为拿到price字段就完事了天真。实际工作中我发现几个很坑的问题。第一个坑是价格类型不统一。有的API返回的是字符串299.00有的直接是数字299有的更过分返回一个列表比如299.00-399.00表示这个商品有多 SKU价格是个区间。遇到区间价格如果直接float转换会直接报错。我的处理方式是如果返回的是字符串且包含-或者~取最低价因为做价格监控关心的是入手门槛取最低区间更符合场景。第二个坑是到手价和标价的区别。接口返回的price通常是页面标价但实际成交时可能有店铺优惠券、跨店满减、淘金币抵扣最终到手价可能比标价低很多。有些API服务商会提供coupon_price或者promotion_price字段如果有优先级要高于price。我在代码里这样处理def parse_price(item: dict) - tuple: raw_price item.get(coupon_price) or item.get(promotion_price) or item.get(price) raw_orginal item.get(orginal_price) or item.get(original_price) or raw_price def to_float(raw): if raw is None: return 0.0 if isinstance(raw, (int, float)): return float(raw) if isinstance(raw, str): raw raw.strip().replace(元, ).replace(¥, ) if - in raw or ~ in raw: raw re.split(r[-~至], raw)[0] try: return float(raw) except ValueError: return 0.0 return 0.0 return to_float(raw_price), to_float(raw_orginal)配合上3.2节里面的字段兜底逻辑这套解析函数到目前为止还没在哪个商品上报过错。第三个坑是商品ID的类型。淘宝的商品IDnum_iid是纯数字但位数很多用int类型存没问题但如果用JSON传输后再解析可能会被某些JSON库弄成科学计数法字符串。所以存product表时item_id字段我用的是TEXT类型而不是INTEGER。写SQL建表时一定要写成TEXT不然你会看到很多商品ID变成了1.23457e18这种鬼东西后悔都来不及。4. 价格预警机制阈值、定时任务与通知推送4.1 基于历史价格的动态阈值策略数据存下来了接下来最关键的就是怎么判断该出手了。我这里实现了两套预警策略你可以根据自己的习惯选择或者叠加。第一套是绝对阈值最简单暴力。比如我给自己设定无线键盘降到399以下就提醒。这个阈值写死在配置里价格一旦低于阈值就直接触发通知。适合对目标商品的心理价位比较明确的人。第二套是动态阈值基于历史价格判断。核心逻辑如果当前价格低于该商品历史最低价的3%以上或者低于近30天平均价格的8%以上就说明现在的价格已经处于一个相对低位可以考虑入手。这个策略的好处是不用你自己定具体数值系统帮你算。动态阈值的本质是识别价格的相对位置。我举个例子一件商品旺季卖699淡季卖529历史最低是499。你如果设绝对阈值500可能永远等不到因为499只出现过一次且是秒杀价。但动态阈值会告诉你当前529相对历史均值已经下降了20%这时候就值得关注了。用一段SQL就能算出来SELECT item_id, MIN(price) AS min_price, AVG(price) AS avg_price_30d FROM price_history WHERE fetched_at datetime(now, -30 days) GROUP BY item_id;拿到这个结果后在Python里和当前价做比较低于阈值就进入预警流程。4.2 APScheduler定时调度每30分钟跑一轮做完拉取逻辑和判断逻辑之后需要让它定时自动化跑起来。这里我用了APScheduler而不是自己写while True sleep。原因很简单APScheduler支持cron表达式可以精确控制每天的运行时间而且自带异常处理单个任务崩了不会拖垮整个调度器。我设置了两个任务一个是整点拉取任务每30分钟执行一次另一个是每天早上9点整合前一天的监控数据生成一份汇总报告。from apscheduler.schedulers.blocking import BlockingScheduler def job_fetch_and_check(): try: for keyword in WATCH_KEYWORDS: scrape_keyword(keyword, max_pages2) check_and_notify() except Exception as e: print(f[任务失败] {e}) # 这里应该接入日志别只print scheduler BlockingScheduler() scheduler.add_job(job_fetch_and_check, cron, minute*/30, idfetch_job) scheduler.start()注意cron表达式里minute*/30表示每30分钟执行一次但这里的实现是从凌晨0点开始每30分钟一次也就是0分和30分。如果你希望是每小时的第15分和第45分跑就写成minute15,45这样更符合实际业务节奏。还有一点要强调BlockingScheduler会阻塞主线程所以你需要把它放在单独的进程里运行或者索性像我一样放到Linux的systemd服务里。我是用nohup python main.py 直接挂着跑的简单粗暴日志重定向到文件就行。如果你用的是Windows可以考虑创建计划任务。4.3 预警通知接入钉钉/邮件预警信息要能及时触达到你手上否则系统跑得再勤也没用。我这里提供了两种通知方式钉钉机器人和SMTP邮件。钉钉机器人的成本最低配置也最简单。你在钉钉群里添加一个自定义机器人拿到webhook地址后用requests直接POST一段JSON就能推送消息def send_dingtalk(message: str): webhook 你的钉钉机器人webhook地址 headers {Content-Type: application/json} payload { msgtype: text, text: {content: message} } requests.post(webhook, jsonpayload, headersheaders, timeout5)我实际运行下来钉钉机器人的延迟基本在1秒以内消息到达率很高。唯一要注意的是钉钉机器人对每分钟消息条数有限制大概20条价格监控这个场景完全够用了。邮件的话用smtplib发即可。我选择的是QQ邮箱的SMTP服务开启授权码后填到配置里。邮件的好处是排版可以做得更丰富可以把当天所有降价商品列成表格一起发。但如果只为了即时性钉钉或者企业微信的webhook体验更好。import smtplib from email.mime.text import MIMEText def send_email(subject: str, html_content: str): sender 你的邮箱 password 你的SMTP授权码 receiver 接收邮箱 msg MIMEText(html_content, html, utf-8) msg[Subject] subject msg[From] sender msg[To] receiver with smtplib.SMTP_SSL(smtp.qq.com, 465, timeout10) as server: server.login(sender, password) server.sendmail(sender, [receiver], msg.as_string())我个人现在的做法是实时预警走钉钉每天早上9点再汇总一封邮件把昨日价格波动明显的商品列表发出来。流量不大成本也低一个月的API调用费用控制在几块钱到几十块钱之内信息密度却比堆在消息群里高很多。5. 常见问题与排查技巧实录5.1 高频报错排查速查表这套方案跑起来之后你大概率会遇到下面几个问题。我做了个速查表按我踩坑的频率从高到低排的。报错现象可能的根因排查与解决办法sign错误签名串生成逻辑与服务商要求不一致重点检查参数是否过滤了空值、签名是否需要大写、app_secret是否拼错返回code 401或403app_key权限不足或IP白名单去服务商控制台确认应用是否已经审核通过检查是否需要绑定服务器IP返回code 403但key正常账号余额不足续费这类接口是预付费扣点模式欠费就直接拒绝部分商品字段缺失接口对非热销商品返回的字段不全解析时用or做字段兜底不要强制访问某个key被限流rate limit短时请求太密集严格控制page间的sleep最好加上随机抖动比如sleep(0.8random.uniform(0,0.5))中文乱码响应编码解析错误requests通过resp.encoding判断有时候需要手动指定resp.encoding utf-8数据重复插入商品ID类型不一致导致去重失效统一在Python层把item_id先str()再入库确保product表主键稳定5.2 那些API文档里不会写的事第一件事别在整点和大促节点跑定时任务。我一开始设置的是每天0点、6点、12点、18点拉取价格结果发现整点的时候接口延迟从平时的300毫秒飙到两三秒偶发超时。后来我把拉取时间调整到每小时的第23分和第53分避开整点高峰稳定了很多。促销节点比如双11当天也一样十点开始十点整点绝对是最卡的。第二件事一个大词下面会有很多不相关的商品。比如你搜键盘出来的结果除了电脑键盘还有门锁的键盘、钢琴的键盘。所以关键词要尽量具体到型号比如MX Keys机械版肯定比键盘准确。另外同一个型号可能有官方店、第三方店、个人闲置转卖等多个商品最好在判断逻辑里加一层白名单筛选只监控你信任的店铺。我是把这些白名单店铺名放在config里爬完数据后再过滤。第三件事别只盯价格这一个维度。有些商品降价是清仓或者过期品比如食品的临期价格会暴跌数码产品的官翻机价格也会低很多。所以我在商品表里加了一个condition字段用来标记是全新还是二手还是官翻不然你看着价格触底很兴奋点进去发现是二手白高兴一场。第四件事API的价格字段和页面成交价可能差出一截。最初我把接口返回的price直接当成真实成交价来做预警结果发现预警之后点进链接价格比预警价格高了不少。后来仔细对比才发现接口返回的price是商品的基础标价实际成交价要结合SKU比如小容量版便宜大容量版贵、优惠券、活动折扣来计算。所以预警的时候我建议宁可漏报不要错报把阈值压低一点留出缓冲空间。6. 我的实操体会与后续扩展思路这套监控系统从写第一行代码到现在已经稳定跑了几个月我个人最大的体会就是掌握了历史数据之后再回来看商品价格你会有一种完全不同的洞察力。以前看到限时6折会冲动下单现在会先查一下这商品过去30天的价格分布很多所谓的大促其实就是把价格先抬高再砍下来一点根本没到我的动态阈值线。整个项目真正写代码的时间其实也就是一个周末大部分时间都耗在接口联调和字段梳理上。如果你也想做一套类似的工具我的建议是先锁定一个商品、一个平台把端到端的流程跑通再逐步增加关键词和预警策略。别一开始就想做一个功能完整的子系统那样会很累而且容易失去动手的乐趣。下一步我打算在现有代码基础上做两件小事一是把价格历史数据导出成图表用matplotlib画每个商品的价格趋势线直观看到价格波动二是把触发预警后的历史记录单独存一张表等一个购买周期结束后回看看哪些预警真正促成了低价下单哪些优化了购物决策。说白了这个工具不复杂但真正跑起来之后节省的是时间改变的是决策方式。