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

pip install requests schedule pandas

# 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

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

# 全局存储上一轮商品数据，用于对比变动
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)