OpenClaw 对接淘宝商品详情 API，低成本搭建自动化选品监控系统（附完整 Python 代码）

clawhub install DataStorageSkill
clawhub install AlertPushSkill

pip install requests json time pandas

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

# ========== 配置区，自行修改 ==========
API_URL = "https://open.onebound.cn/api/taobao/item_get_pro"
APP_KEY = "你的平台AppKey"
APP_SECRET = "你的平台AppSecret"
# 需要监控的淘宝商品ID列表
MONITOR_ITEM_IDS = ["723456123456", "723456123789", "723456123901"]
# 监控轮询间隔，单位秒，建议600（10分钟）防止调用超限
LOOP_INTERVAL = 600
# 选品筛选阈值
MIN_SALE = 500        # 最低月销
MAX_PRICE = 199.0     # 最高售价
GOOD_RATE = 0.95      # 最低好评率
PRICE_DROP_RATE = 0.1 # 价格下跌10%触发告警
# ======================================

# 存储历史数据，用于对比波动
history_data = {}

def call_taobao_api(item_id):
    """调用item_get_pro高级商品接口"""
    params = {
        "appkey": APP_KEY,
        "secret": APP_SECRET,
        "api_name": "taobao.item_get_pro",
        "num_iid": item_id,
        "cache": 0
    }
    try:
        resp = requests.get(API_URL, params=params, timeout=15)
        result = resp.json()
        if result.get("code") != 0:
            print(f"商品{item_id}接口调用失败：{result.get('msg')}")
            return None
        return result.get("data", {})
    except Exception as e:
        print(f"请求异常：{str(e)}")
        return None

def filter_good_product(item_info):
    """根据阈值筛选潜力爆款"""
    try:
        sale = int(item_info.get("sales", 0))
        price = float(item_info.get("price", 9999))
        good_comment_rate = float(item_info.get("good_rate", 0))
        if sale >= MIN_SALE and price <= MAX_PRICE and good_comment_rate >= GOOD_RATE:
            return True
        return False
    except:
        return False

def check_price_fluctuate(item_id, new_price):
    """对比历史价格，判断是否大幅降价"""
    if item_id not in history_data:
        history_data[item_id] = {"price": new_price, "time": datetime.now()}
        return False
    old_price = history_data[item_id]["price"]
    if old_price == 0:
        return False
    drop_ratio = (old_price - new_price) / old_price
    if drop_ratio >= PRICE_DROP_RATE:
        print(f"【价格告警】商品{item_id}降价{round(drop_ratio*100,2)}%，原价{old_price}，现价{new_price}")
        # 此处可对接OpenClaw推送插件，发送企业微信/短信提醒
    # 更新历史价格
    history_data[item_id]["price"] = new_price
    history_data[item_id]["time"] = datetime.now()
    return drop_ratio >= PRICE_DROP_RATE

def save_to_excel(valid_items):
    """将筛选出的优质商品写入Excel存档"""
    if not valid_items:
        return
    df = pd.DataFrame(valid_items)
    save_name = f"淘宝选品记录_{datetime.now().strftime('%Y%m%d_%H%M')}.xlsx"
    df.to_excel(save_name, index=False)
    print(f"优质商品数据已保存至：{save_name}")

def monitor_loop():
    print("===== OpenClaw淘宝商品监控程序启动 =====")
    while True:
        qualified_list = []
        for item_id in MONITOR_ITEM_IDS:
            item_data = call_taobao_api(item_id)
            if not item_data:
                time.sleep(2)
                continue
            # 提取核心字段
            info = {
                "item_id": item_id,
                "title": item_data.get("title", ""),
                "price": float(item_data.get("price", 0)),
                "original_price": item_data.get("original_price", ""),
                "sales": item_data.get("sales", 0),
                "good_rate": item_data.get("good_rate", 0),
                "shop_name": item_data.get("shop_name", ""),
                "coupon_info": item_data.get("coupon", "无优惠券"),
                "monitor_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            }
            # 价格波动检测
            check_price_fluctuate(item_id, info["price"])
            # 筛选爆款
            if filter_good_product(info):
                qualified_list.append(info)
                print(f"【潜力品】{info['title']} 月销{info['sales']} 售价{info['price']}")
            time.sleep(1)
        # 保存筛选结果
        save_to_excel(qualified_list)
        print(f"本轮监控完成，等待{LOOP_INTERVAL/60}分钟后下一轮扫描\n")
        time.sleep(LOOP_INTERVAL)

if __name__ == "__main__":
    monitor_loop()

task_name: 淘宝选品监控
task_type: python_script
script_path: ./taobao_monitor.py
run_mode: loop
max_retry: 3
alert_rule:
  trigger_keyword: 【价格告警】
  push_channel: wecom
storage:
  save_data: true
  file_type: excel