1. 多平台比价系统到底在解决什么问题多平台商品比价系统本质是把淘宝、京东、1688、拼多多等平台的同款商品价格通过 API 聚合拉到一个统一的数据结构里再用定时任务持续轮询价格变化当价格跌破你设定的阈值时自动触发降价提醒。它适合三类人个人网购想蹲历史低价的、跨境或反向海淘卖家需要批量监控货源底价的、企业采购要长期追踪办公物资行情的。手动比价的痛点很具体。同一款蓝牙耳机京东标价 399 券后 349拼多多百亿补贴 3291688 批发价 285 但起订量 10 件你挨个 App 切换、截图、记价格半小时就没了。更麻烦的是价格波动——今天 329 明天可能 299你不可能 24 小时盯着。所以这套系统的核心价值就三件事聚合归一让比价有统一基准定时监控让价格变化被记录成曲线阈值提醒让你只在真正值得出手时才被打扰。我试过用纯手工表格维护 20 个商品的价格坚持不到一周就放弃了因为平台改价太频繁人工根本追不上。后来把抓取和比对交给定时任务只在降价时收通知效率完全不一样。下面按 API 聚合层、价格监控层、降价提醒层三层拆开讲每层都给可复制的配置骨架。2. TaoToken 在比价链路里的前置准备比价系统里有一环容易被忽略商品标题的同款匹配和价格异动的原因归纳。各平台标题写法差异极大比如「Apple AirPods Pro 2 代 USB-C 版」和「苹果 AirPods Pro 第二代 Type-C 接口」纯字符串匹配会把同款判成不同款。这时候可以用大模型做标题语义归一和规格抽取把品牌、型号、版本、接口类型结构化出来再入库比对。TaoToken 在这里的角色是提供统一的模型调用入口。你不需要分别去对接多家模型厂商的 SDK 和计费体系通过一个 API 地址和一把 Key 就能调用对话模型完成标题归一、规格抽取、降价原因摘要生成。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 接入地址是 https://taotoken.net/api 注意 API 地址不带 UTM 参数。前置准备分三步。第一步注册后在控制台创建 API Key地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。第二步确认你要用的模型名称可以在模型对话页面试跑地址是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。第三步把 Key 写进环境变量不要硬编码在代码里。注意比价系统里模型调用只用于标题归一和摘要不要用它替代价格抓取本身。价格数据必须来自平台 API 或合规采集模型只做文本理解层的事。如果你后续要做长期的编码和 Agent 任务比如自动生成比价报表、维护抓取脚本可以了解 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。3. API 聚合层与价格监控的可复制配置这一层是整个系统的地基。API 聚合层要做的事统一字段、去重同款、限流缓存。价格监控层要做的事定时轮询、历史入库、波动比对。下面给一份配置文件骨架用 YAML 写你可以直接改成自己的参数。# config/price_monitor.yaml aggregator: platforms: - name: jd enabled: true api_base: https://api.jd.com/routerjson rate_limit: 30 # 每分钟最大请求数 timeout: 8 - name: pdd enabled: true api_base: https://gw-api.pinduoduo.com/api/router rate_limit: 20 timeout: 8 - name: alibaba_1688 enabled: true api_base: https://gw.open.1688.com/openapi rate_limit: 15 timeout: 10 normalize: fields: [item_id, title, brand, model, spec, price, coupon_price, final_price, stock, shop, platform] dedup_keys: [brand, model, spec] similarity_threshold: 0.86 # 标题相似度阈值超过视为同款 monitor: schedule: default_interval_minutes: 240 # 普通商品 4 小时 hot_interval_minutes: 45 # 爆款 45 分钟 idle_interval_minutes: 720 # 非活动商品 12 小时 storage: mysql_dsn: mysql://user:pass127.0.0.1:3306/price_db redis_url: redis://127.0.0.1:6379/2 history_table: price_history retry: max_attempts: 3 backoff_seconds: [5, 30, 120] alert: rules: - type: fixed_price item_id: SKU10086 target_price: 299.00 - type: drop_ratio item_id: SKU10087 base: historical_max ratio: 0.15 # 较历史最高价降 15% 触发 channels: - type: webhook url: https://your-domain.com/hook/price-alert timeout: 5 - type: email smtp_host: smtp.example.com smtp_port: 465 from: alertexample.com to: [youexample.com] cooldown_minutes: 120 # 同一商品 2 小时内不重复提醒字段归一化是聚合层的核心。各平台返回的 JSON 结构完全不同京东可能叫price拼多多叫min_group_price1688 叫priceRange。你需要写一层适配器把每个平台的响应映射到统一字段。下面是一个 Python 适配器示例用 requests 调用并归一化。# aggregator/normalize.py import requests from decimal import Decimal def fetch_jd(item_id, cfg): resp requests.get( cfg[api_base], params{method: jingdong.ware.price.get, skuId: item_id}, timeoutcfg[timeout], ) resp.raise_for_status() raw resp.json() return { item_id: item_id, title: raw.get(wareTitle, ), price: Decimal(str(raw.get(jdPrice, 0))), coupon_price: Decimal(str(raw.get(couponPrice, 0))), final_price: Decimal(str(raw.get(finalPrice, raw.get(jdPrice, 0)))), stock: raw.get(stock, 0), platform: jd, } def normalize_title(title, model_client): prompt f抽取商品标题中的品牌、型号、规格用JSON返回{title} result model_client.chat(prompt) return result # {brand: ..., model: ..., spec: ...}定时任务用 Celery 或 Crontab 都行。Crontab 适合轻量部署Celery 适合需要任务队列和重试的场景。下面是一个 Celery beat 配置按商品热度动态调整轮询频率。# tasks/scheduler.py from celery import Celery from celery.schedules import crontab app Celery(price_monitor, brokerredis://127.0.0.1:6379/0) app.conf.beat_schedule { poll-hot-items: { task: tasks.poll_items, schedule: crontab(minute*/45), args: (hot,), }, poll-default-items: { task: tasks.poll_items, schedule: crontab(minute0, hour*/4), args: (default,), }, poll-idle-items: { task: tasks.poll_items, schedule: crontab(minute30, hour*/12), args: (idle,), }, }价格入库时每次抓取都往price_history表插一条记录字段包括 item_id、record_time、price、coupon_price、final_price、platform、activity_status。Redis 里缓存当前最新价key 用price:latest:{item_id}设置 10 分钟过期避免频繁查库。4. 端到端验证从抓取到提醒的完整动作配置写完后必须做一次端到端验证确认抓取、归一、入库、比对、提醒五个环节都通。下面给一套验证动作按顺序执行。第一步单独跑一次抓取任务确认能拿到数据。用 Python 直接调用适配器python -c from aggregator.normalize import fetch_jd cfg {api_base: https://api.jd.com/routerjson, timeout: 8} print(fetch_jd(100012043978, cfg)) 如果返回的字典里final_price是有效数字说明抓取和归一通了。如果报超时或 403先检查 API Key 和请求频率。第二步手动触发一次 Celery 任务确认入库celery -A tasks.scheduler call tasks.poll_items --args[default]然后查 MySQLSELECT item_id, record_time, final_price, platform FROM price_history WHERE item_id SKU10086 ORDER BY record_time DESC LIMIT 5;第三步模拟一次降价验证提醒触发。把某商品的历史最高价手动改高让当前价低于阈值然后跑比对逻辑# verify/alert_test.py from decimal import Decimal from alert.engine import check_alert history_max Decimal(399.00) current Decimal(299.00) rule {type: drop_ratio, base: historical_max, ratio: 0.15} triggered check_alert(history_max, current, rule) print(alert triggered:, triggered) # 期望 True第四步确认提醒通道收到消息。Webhook 通道可以用本地起一个临时服务接收python -m http.server 9000把 alert 配置里的 webhook url 改成http://127.0.0.1:9000/hook/price-alert触发后看终端是否打印 POST 请求。邮件通道则检查收件箱。成功结果应该是抓取返回归一化字典、MySQL 有新记录、比对返回 True、Webhook 收到 JSON、邮件收到通知。五个都通过闭环就算搭好了。5. 本篇常见错误排查报错一requests.exceptions.HTTPError: 429 Too Many Requests这是接口限流。原因通常是轮询频率超过平台限制或者多个任务并发打同一个接口。排查方法看 config 里的rate_limit是否和平台文档一致检查 Celery 是否同时跑了多个 poll 任务。解决在适配器里加令牌桶限流或者把default_interval_minutes调大。Redis 可以做分布式限流key 用ratelimit:{platform}:{minute}。报错二同款商品被拆成多条记录标题相似度阈值设太低或者归一化字段没对齐。比如「AirPods Pro 2」和「AirPods Pro 第二代」相似度可能只有 0.7。解决把similarity_threshold调到 0.85 以上同时用模型做标题归一把「第二代」和「2」映射到同一规格。入库前用dedup_keys做一次去重。报错三降价提醒重复推送同一商品在短时间内多次触发。原因是cooldown_minutes没生效或者比对逻辑每次都拿历史最高价做基准。解决在 Redis 里记录alert:cooldown:{item_id}触发后设置过期时间比对前先检查这个 key 是否存在。另外base字段要明确是historical_max还是last_price避免逻辑混淆。报错四MySQL 写入慢历史表越来越大price_history表没有索引或者单表数据量过千万。解决给item_id和record_time建联合索引按月份分表或者把超过 90 天的历史数据归档到冷存储。查询价格曲线时只查最近 30 天用 Redis 缓存聚合结果。报错五模型调用超时导致归一化失败标题归一化是同步调用如果模型响应慢会拖垮整个抓取任务。解决把归一化改成异步抓取先入库原始标题归一化任务单独跑或者设置超时和降级策略模型不可用时用规则匹配兜底。TaoToken 的 API 地址是 https://taotoken.net/api 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 可以先在模型对话页面确认模型可用性。6. 接入与排障的下一步如果你在接入 API 聚合层时遇到 Key 配置或限流问题先去 API Keys 页面确认 Key 状态和额度地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 再对照接入文档检查请求格式地址是 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。验证模型是否可用直接在模型对话页面发一条测试消息地址是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。如果你要把这套比价系统做成长期运行的 Agent自动生成报表、自动维护抓取脚本可以看 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。最后给一个实用技巧价格监控的轮询频率不要一刀切。把商品按「是否在活动期」「历史降价频率」「用户关注度」三个维度打分高分商品高频轮询低分商品拉长间隔。这样既能抓住限时降价又不会把接口额度浪费在长期不动的商品上。我实测下来动态频率比固定频率省了大约 40% 的请求量而降价捕获率几乎没降。