×

api 数据挖掘

Open Claw 结合淘宝高级商品 API，5 分钟搭建自动化监控选品系统（完整 Python 代码）

admin admin 发表于2026-07-02 16:31:33 浏览8 评论0

抢沙发发表评论

前言

做淘宝运营、无货源铺货、电商数据分析的同行应该都有同款痛点：每天手动翻几百个商品，对比价格、销量、库存、SKU 规格，盯竞品调价、爆款起量，纯人工不仅耗时 8 小时以上，还很容易错过红利款，竞品活动上线也无法第一时间察觉。

之前试过自建爬虫采集，频繁遇到反爬验证码、IP 封禁、页面结构改版失效，维护成本极高。后来摸索出一套轻量化方案：依托淘宝item_get_pro高级详情接口，搭配 Open Claw 做请求调度、数据清洗、定时轮询，无需服务器，本地电脑就能实现 7×24 小时商品监控、批量自动选品、价格库存异动预警，个人卖家、副业创业者零门槛落地。

本文不讲晦涩理论，全程实操，附带可直接复制运行的 Python 源码，适配 CSDN、知乎阅读，看完就能搭建属于自己的淘系数据监控工具。

一、方案核心优势：为什么选 item_get_pro 接口

市面上基础商品接口只能返回标题、售价等浅层数据，而item_get_pro是淘系选品专用高级接口，接口返回维度完全覆盖选品全需求：

  2. 全量 SKU 数据：每个规格独立售价、库存、规格名称、属性图，精准核算单品利润；

  4. 销售分层数据：总销量、实时库存、发货地、店铺动态评分（服务 / 物流 / 商品三项）；

  6. 营销信息：原价、促销价、包邮状态、套餐搭配、商品详情图文；

  8. 店铺基础：掌柜昵称、店铺 ID、是否天猫、粉丝收藏量；

  10. 商品属性：品牌、货号、附加功能、尺码颜色等筛选标签。

搭配 Open Claw 的能力互补：自动处理接口鉴权、失败重试、缓存控制、JSON 格式化输出，不用手写 MD5 签名逻辑，大幅减少代码量，新手不用钻研接口加密规则。

适用业务场景

  • 竞品实时监控：盯同行调价、上新套餐、库存清仓动作；

  • 批量选品过滤：按销量、客单价、发货地自动筛潜力爆款；

  • 无货源铺货数据采集：一键导出商品主图、详情、规格文案；

  • 库存风险预警：监控爆款库存余量，提前备货或更换货源；

  • 历史数据沉淀：定时抓取存入 Excel，长期追踪商品走势。

二、前置准备工作

  2. 申请接口调用凭证（key/secret），开通taobao.item_get_pro接口调用权限；

  4. 本地 Python3.8 及以上环境；

  6. 安装依赖库：requests（网络请求）、schedule（定时监控）、pandas（数据导出）

pip install requests schedule pandas

4.获取商品 ID（num_iid）：淘宝商品链接item.htm?id=xxxx，末尾数字即为商品 ID。

三、完整实战代码（分三大模块：单品采集、批量选品、定时监控预警）

3.1 基础接口请求封装（Open Claw 调度调用核心）

# tb_monitor.py
import requests
import json
import time
import schedule
import pandas as pd

# ===================== 配置区，自行替换凭证 =====================
API_KEY = "你的调用key"
API_SECRET = "你的调用secret"
API_GATEWAY = "https://api-gw.onebound.cn/taobao/item_get_pro"
# 监控商品池，填入需要跟踪的商品ID列表
MONITOR_ITEM_IDS = ["520813250866", "xxxxxx", "xxxxxx"]
# 选品筛选规则：可自定义
MIN_SALES = 50
PRICE_MIN = 20
PRICE_MAX = 200
# =================================================================

def get_taobao_item(num_iid):
    """
    Open Claw调度调用item_get_pro高级接口，获取商品完整数据
    :param num_iid: 淘宝商品ID
    :return: 结构化商品字典，失败返回None
    """
    params = {
        "key": API_KEY,
        "secret": API_SECRET,
        "api_name": "item_get_pro",
        "num_iid": num_iid,
        "cache": "no",  # 关闭缓存，获取实时最新数据
        "result_type": "json"
    }
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    try:
        resp = requests.get(url=API_GATEWAY, params=params, headers=headers, timeout=15)
        resp.raise_for_status()
        res_data = resp.json()
        # 接口异常判断
        if res.get("error_code", "0000") != "0000":
            print(f"商品{num_iid}查询失败：{res_data.get('reason')}")
            return None
        item_info = res_data.get("item", {})
        return parse_item_data(item_info)
    except Exception as e:
        print(f"请求异常：{str(e)}")
        return None

def parse_item_data(raw_item):
    """数据清洗，提取选品核心字段"""
    sku_list = raw_item.get("skus", {}).get("sku", [])
    sku_info = []
    for sku in sku_list:
        sku_info.append({
            "规格名称": sku.get("properties_name", "").split(":")[-1],
            "售价": float(sku.get("price", 0)),
            "原价": float(sku.get("orginal_price", 0)),
            "库存": int(sku.get("quantity", 0)),
            "SKUID": sku.get("sku_id")
        })
    clean_data = {
        "商品ID": raw_item.get("num_iid"),
        "商品标题": raw_item.get("title"),
        "现价": float(raw_item.get("price", 0)),
        "原价": float(raw_item.get("orginal_price", 0)),
        "总销量": int(raw_item.get("total_sold", 0)),
        "发货地": raw_item.get("location"),
        "店铺名称": raw_item.get("seller_info", {}).get("nick"),
        "是否天猫": raw_item.get("tmall"),
        "商品链接": raw_item.get("detail_url"),
        "主图": raw_item.get("pic_url"),
        "SKU明细": sku_info,
        "总库存": int(raw_item.get("num", 0)),
        "收藏量": int(raw_item.get("favcount", 0)),
        "采集时间": time.strftime("%Y-%m-%d %H:%M:%S")
    }
    return clean_data

3.2 批量自动选品函数（过滤潜力爆款）

def batch_filter_goods(item_id_list):
    """批量采集商品并按自定义规则筛选优质品"""
    qualified = []
    all_data = []
    for item_id in item_id_list:
        item = get_taobao_item(item_id)
        if not item:
            continue
        all_data.append(item)
        # 自定义选品过滤条件
        sales = item["总销量"]
        price = item["现价"]
        if sales >= MIN_SALES and PRICE_MIN <= price <= PRICE_MAX:
            print(f"【潜力款】{item['商品标题']} | 售价：{price} | 销量：{sales}")
            qualified.append(item)
    # 导出全部商品数据到Excel，方便线下分析
    df = pd.DataFrame(all_data)
    df.to_excel("淘宝商品采集数据.xlsx", index=False)
    print(f"\n批量采集完成，共采集{len(all_data)}款，筛选潜力商品{len(qualified)}款，已导出Excel")
    return qualified

3.3 定时监控 + 价格库存异动预警（核心监控功能）

# 全局存储上一轮商品数据，用于对比变动
last_item_cache = {}

def monitor_warning():
    """定时执行监控，对比前后数据，触发异动提醒"""
    global last_item_cache
    print(f"\n===== {time.strftime('%Y-%m-%d %H:%M:%S')} 开始轮询监控 =====")
    for item_id in MONITOR_ITEM_IDS:
        curr = get_taobao_item(item_id)
        if not curr:
            continue
        # 首次采集，存入缓存
        if item_id not in last_item_cache:
            last_item_cache[item_id] = curr
            continue
        last = last_item_cache[item_id]
        # 价格变动预警
        if abs(curr["现价"] - last["现价"]) > 0.1:
            print(f"【价格异动预警】{curr['商品标题']}")
            print(f"旧价：{last['现价']} → 新价：{curr['现价']}")
        # 库存暴跌预警
        stock_drop = last["总库存"] - curr["总库存"]
        if stock_drop > 200:
            print(f"【库存暴跌预警】{curr['商品标题']}，库存减少{stock_drop}件")
        # 销量暴涨预警（短时间起量爆款）
        sales_rise = curr["总销量"] - last["总销量"]
        if sales_rise > 50:
            print(f"【爆款起量预警】{curr['商品标题']}，新增销量{sales_rise}")
        # 更新缓存
        last_item_cache[item_id] = curr

if __name__ == "__main__":
    # 1. 单次批量选品，取消注释运行
    # batch_filter_goods(MONITOR_ITEM_IDS)

    # 2. 开启定时监控，每30分钟轮询一次
    schedule.every(30).minutes.do(monitor_warning)
    print("监控程序已启动，每30分钟自动采集对比数据，等待任务执行...")
    while True:
        schedule.run_pending()
        time.sleep(10)

四、代码使用说明

  2. 修改配置区API_KEY、API_SECRET为自己的调用凭证；

  4. MONITOR_ITEM_IDS填入需要监控、采集的商品 ID；

  6. 调整MIN_SALES、PRICE_MIN、PRICE_MAX修改选品筛选标准；

  8. 两种运行模式：

    • 单次批量选品：注释定时监控代码，打开batch_filter_goods调用，运行后自动导出 Excel 数据表；

    • 7×24 小时监控：注释批量选品代码，直接运行，程序后台循环对比数据，出现价格、库存、销量变动控制台打印预警。

五、选品实操落地思路

1. 铺货无货源商家

批量抓取同类竞品 SKU 售价、发货地，筛选发货地货源集中、客单价适中、销量稳定的商品，一键导出主图、标题、规格文案，大幅节省上架整理时间。

2. 自有店铺竞品监控

添加 3-10 个头部竞品商品 ID，开启定时监控，竞品降价促销、推出新套餐、库存清仓时第一时间收到提醒，快速调整自身定价与活动策略。

3. 爆款潜力挖掘

设置销量阈值 50-1000，避开几万销量的红海头部商品，抓取刚起量、竞争小的潜力款，长期跟踪销量增长曲线，提前布局货源。

六、避坑总结（实操踩坑经验）

  2. 缓存参数 cache 设置为 no：监控场景必须关闭缓存，否则拿到几小时前的旧数据，预警失效；批量采集可开启 yes 降低调用消耗；

  4. 限流控制：接口存在调用频次限制，循环采集时可增加time.sleep(1)延迟，避免短时间大量请求被限制；

  6. SKU 数据为空排查：商品下架、链接失效会返回空 sku 数组，代码内置异常捕获，自动打印失败商品 ID；

  8. 数据存储扩展：现有代码导出 Excel，长期监控可接入 sqlite 数据库，留存每日价格销量，绘制商品走势。

七、进阶拓展方向

  2. 对接企业微信 / 钉钉 webhook，异动预警不再依赖控制台，手机实时推送消息；

  4. 新增关键词搜索接口，自动抓取分类下全部商品，实现全网自动筛品；

  6. 增加历史价格接口，抓取商品历史低价，判断当前是否入手铺货最佳时机；

  8. 部署云服务器，保持程序全天在线，本地关机也不中断监控。

结尾

这套方案对比自建爬虫，最大优势是稳定、合规、维护成本极低，接口官方标准化返回数据，不会因为页面改版导致工具报废。Open Claw 简化了接口对接的复杂逻辑，零基础运营也能看懂并修改代码，低成本搭建专属淘系数据监控工具，把人工盯品的时间释放出来做运营优化。


少长咸集

群贤毕至

访客