不用生意参谋，自建淘宝竞品监控选品工具｜OpenClaw 淘宝商品 API 完整实操教程（附 Python 源码）

pip install requests pandas

import requests
import json
import time
import pandas as pd
from datetime import datetime

# 替换成你自己OpenClaw后台的密钥
API_KEY = "你的key"
API_SECRET = "你的secret"
# 目标接口地址：淘宝高级商品详情接口 item_get_pro
API_URL = "https://open.onebound.cn/taobao/item_get_pro"

class TaobaoItemMonitor:
    def __init__(self):
        self.key = API_KEY
        self.secret = API_SECRET
        self.url = API_URL
        # 存储历史数据文件
        self.save_file = "taobao_goods_data.csv"
        # 初始化文件不存在则创建表头
        self.init_data_file()

    def init_data_file(self):
        try:
            pd.read_csv(self.save_file)
        except FileNotFoundError:
            cols = [
                "crawl_time", "item_id", "title", "price", "sale_price",
                "month_sold", "total_comment", "good_rate", "shop_name",
                "shop_level", "is_tmall", "stock", "category_id"
            ]
            df = pd.DataFrame(columns=cols)
            df.to_csv(self.save_file, index=False, encoding="utf-8-sig")

    # 调用item_get_pro接口获取单商品完整数据
    def get_item_detail(self, item_id):
        params = {
            "key": self.key,
            "secret": self.secret,
            "api_name": "item_get_pro",
            "item_id": item_id
        }
        try:
            resp = requests.get(self.url, params=params, timeout=15)
            res_json = resp.json()
            if res_json.get("code") != 0:
                print(f"商品{item_id}接口请求失败：{res_json.get('msg')}")
                return None
            return res_json.get("result", {})
        except Exception as e:
            print(f"请求异常 item_id:{item_id}, 错误：{str(e)}")
            return None

    # 解析接口返回数据，提取选品核心字段
    def parse_goods_data(self, raw_data):
        if not raw_data:
            return {}
        item_info = raw_data.get("item", {})
        price_info = raw_data.get("price", {})
        sales_info = raw_data.get("sales", {})
        shop_info = raw_data.get("shop", {})
        category_info = raw_data.get("category", {})

        data = {
            "crawl_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "item_id": item_info.get("item_id", ""),
            "title": item_info.get("title", ""),
            "price": price_info.get("original_price", ""),
            "sale_price": price_info.get("current_price", ""),
            "month_sold": sales_info.get("month_sold", 0),
            "total_comment": sales_info.get("comment_count", 0),
            "good_rate": sales_info.get("good_comment_rate", "0%"),
            "shop_name": shop_info.get("shop_name", ""),
            "shop_level": shop_info.get("shop_rank", ""),
            "is_tmall": shop_info.get("is_tmall", False),
            "stock": item_info.get("stock_num", 0),
            "category_id": category_info.get("cid", "")
        }
        return data

    # 写入本地CSV存储
    def save_to_csv(self, data_row):
        df = pd.DataFrame([data_row])
        df.to_csv(self.save_file, mode="a", header=False, index=False, encoding="utf-8-sig")

    # 爆款筛选规则：自定义蓝海潜力款标准，可自行调整阈值
    def filter_hot_goods(self, data_row):
        # 筛选条件：月销>200，好评率>95%，现货库存大于50，非天猫低价竞争款
        month_sold = int(data_row["month_sold"])
        good_rate = float(data_row["good_rate"].replace("%", ""))
        stock = int(data_row["stock"])
        is_tmall = data_row["is_tmall"]

        if month_sold > 200 and good_rate >= 95 and stock >= 50 and not is_tmall:
            return True
        return False

    # 批量循环监控主逻辑
    def monitor_batch_items(self, item_id_list, interval=300):
        """
        :param item_id_list: 待监控商品ID列表
        :param interval: 轮询间隔，单位秒，示例300=5分钟采集一次
        """
        print("=====淘宝商品监控程序启动=====")
        while True:
            for item_id in item_id_list:
                raw = self.get_item_detail(item_id)
                if not raw:
                    continue
                row_data = self.parse_goods_data(raw)
                self.save_to_csv(row_data)
                # 判断是否为潜力爆款
                if self.filter_hot_goods(row_data):
                    print(f"【发现潜力爆款】{row_data['title']} 月销:{row_data['month_sold']} 售价:{row_data['sale_price']}")
                # 单商品请求间隔，防止接口调用频率超限
                time.sleep(2)
            print(f"本轮商品采集完成，等待{interval}秒后下一轮扫描...")
            time.sleep(interval)

if __name__ == "__main__":
    # 需要监控的商品ID集合，自行替换
    watch_items = [
        "723456123456",
        "723456789012",
        "723456987654"
    ]
    monitor = TaobaoItemMonitor()
    # 每5分钟轮询一次全部商品
    monitor.monitor_batch_items(watch_items, interval=300)