×

api 数据挖掘

电商运营自动化：低成本搭建商品实时监控选品体系（附完整可运行代码）

admin admin 发表于2026-07-06 16:21:02 浏览17 评论0

抢沙发发表评论

前言

做淘宝、跨境电商的朋友应该都有同感，人工选品、盯竞品是最耗时间的重复性工作。每天手动翻竞品链接、记录价格、销量、评价、活动变动，一天下来大半时间浪费在复制粘贴上，数据滞后还容易错过蓝海机会。

之前踩过不少坑：手动整理 Excel 效率极低，浏览器批量采集容易触发平台限制，第三方付费工具年费动辄几千，数据维度不全还无法自定义监控规则。经过一段时间摸索，我搭建了一套轻量化自动化监控方案，依托 OpenClaw 数据能力，无需复杂环境部署，几行代码就能批量拉取商品全维度信息，自动保存、定时巡检，新手也能快速落地，下面完整分享实操流程与源码。

一、方案核心优势（解决运营选品痛点）

  2. 全维度商品信息一键获取 覆盖商品基础标题、售价、sku 规格、历史价格、月销、总销量、好评差评、类目层级、活动优惠、主图详情、库存状态、店铺层级等全部运营核心数据，不用多页面切换搜集。

  4. 定时自动监控，异常实时捕捉 设置定时循环执行，自动对比前后两次数据，价格跳水、销量暴涨、上新活动、差评激增都会标记，第一时间筛选潜力爆款、规避内卷红海品。

  6. 轻量化部署，无高额订阅成本 不用本地搭建复杂爬虫环境，规避平台访问限制，支持国内、跨境淘宝商品同步采集，单次任务批量处理上百个商品链接。

  8. 数据本地留存，自由分析 采集结果自动存入 Excel / 本地 JSON 文件，可搭配 pandas 做筛选、排序、数据可视化，快速筛选高销量低竞争、高转化蓝海款。

二、前期准备工作

  2. 注册登录获取专属密钥凭证（个人后台直接复制两组字符备用）；

  4. 本地安装 Python3.8 及以上版本，安装依赖库：requests、pandas、openpyxl；

pip install requests pandas openpyxl

3.整理需要监控的商品 ID / 商品链接清单，存入 txt 文档，一行一个链接方便批量读取。

三、完整实操代码（可直接复制运行）

3.1 基础单商品数据获取脚本

功能：输入商品 ID，返回完整结构化商品数据，包含价格、销量、评价、活动信息

import requests
import json
import time

# 替换为你自己后台获取的密钥
CLIENT_ID = "你的ClientID"
CLIENT_SECRET = "你的ClientSecret"

def get_goods_detail(item_id):
    """获取单款商品完整详情数据"""
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    params = {
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "item_id": item_id
    }
    # 数据请求地址
    url = "https://open.onebound.cn/server/goods/query"
    try:
        resp = requests.post(url, data=params, headers=headers, timeout=30)
        res_data = resp.json()
        if res_data.get("code") == 200:
            return res_data.get("data")
        else:
            print(f"商品{item_id}查询失败：{res_data.get('msg')}")
            return None
    except Exception as e:
        print(f"网络异常：{str(e)}")
        return None

# 测试调用
if __name__ == "__main__":
    target_item = "728956123456"  # 替换目标商品ID
    info = get_goods_detail(target_item)
    if info:
        print("商品基础信息：")
        print(f"商品标题：{info.get('title')}")
        print(f"当前售价：{info.get('price')}")
        print(f"月销量：{info.get('month_sales')}")
        print(f"总评价数：{info.get('review_count')}")
        print(f"店铺名称：{info.get('shop_name')}")

3.2 批量监控 + 数据导出 Excel 脚本

功能：批量读取多款商品，循环采集，自动写入表格，记录采集时间，用于长期对比选品

import requests
import json
import pandas as pd
import time

CLIENT_ID = "你的ClientID"
CLIENT_SECRET = "你的ClientSecret"

def get_goods_detail(item_id):
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    params = {
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "item_id": item_id
    }
    url = "https://open.onebound.cn/server/goods/query"
    try:
        resp = requests.post(url, data=params, headers=headers, timeout=30)
        res_data = resp.json()
        if res_data.get("code") == 200:
            return res_data.get("data")
        else:
            print(f"{item_id} 查询异常：{res_data.get('msg')}")
            return None
    except Exception as e:
        print(f"请求出错：{e}")
        return None

def batch_monitor(item_id_list):
    """批量采集商品并整理数据"""
    result_list = []
    current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    for item in item_id_list:
        data = get_goods_detail(item)
        if not data:
            continue
        row = {
            "采集时间": current_time,
            "商品ID": item,
            "商品标题": data.get("title", ""),
            "售价": data.get("price", ""),
            "划线价": data.get("original_price", ""),
            "月销量": data.get("month_sales", 0),
            "总销量": data.get("total_sales", 0),
            "好评数": data.get("good_review", 0),
            "差评数": data.get("bad_review", 0),
            "店铺名": data.get("shop_name", ""),
            "一级类目": data.get("cat_name_1", ""),
            "活动标签": data.get("promotion_tag", ""),
            "商品链接": f"https://item.taobao.com/item.htm?id={item}"
        }
        result_list.append(row)
        time.sleep(1.2)  # 间隔防限流
    return result_list

def save_to_excel(data, file_name="商品监控数据表.xlsx"):
    df = pd.DataFrame(data)
    df.to_excel(file_name, index=False)
    print(f"数据已保存至 {file_name}")

if __name__ == "__main__":
    # 填入需要监控的全部商品ID
    monitor_ids = [
        "728956123456",
        "654321987654",
        "123456789012"
    ]
    goods_data = batch_monitor(monitor_ids)
    if goods_data:
        save_to_excel(goods_data)

3.3 定时自动巡检 + 异常预警简易脚本

新增逻辑：对比历史表格数据，识别价格下跌、销量暴涨商品，打印预警信息，实现自动化监控

import requests
import json
import pandas as pd
import time
import os

CLIENT_ID = "你的ClientID"
CLIENT_SECRET = "你的ClientSecret"
EXCEL_FILE = "商品监控数据表.xlsx"

def get_goods_detail(item_id):
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    params = {
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "item_id": item_id
    }
    url = "https://open.onebound.cn/server/goods/query"
    try:
        resp = requests.post(url, data=params, headers=headers, timeout=30)
        res_data = resp.json()
        if res_data.get("code") == 200:
            return res_data.get("data")
        return None
    except Exception as e:
        return None

def check_data_change(new_data):
    """对比历史数据，输出异常商品预警"""
    if not os.path.exists(EXCEL_FILE):
        print("暂无历史监控记录，本次仅保存数据")
        return
    old_df = pd.read_excel(EXCEL_FILE)
    old_dict = dict(zip(old_df["商品ID"], old_df.to_dict("records")))
    alert_list = []
    for row in new_data:
        item_id = row["商品ID"]
        if item_id not in old_dict:
            alert_list.append(f"【新品上线】{row['商品标题']}")
            continue
        old = old_dict[item_id]
        # 价格下跌预警
        if float(row["售价"]) < float(old["售价"]) * 0.9:
            alert_list.append(f"【价格暴跌】{row['商品标题']} 原价{old['售价']} 现价{row['售价']}")
        # 销量暴涨预警（月销增长超500）
        if int(row["月销量"]) - int(old["月销量"]) > 500:
            alert_list.append(f"【爆款飙升】{row['商品标题']} 月销新增{int(row['月销量']) - int(old['月销量'])}")
    if alert_list:
        print("===== 监控异常预警 =====")
        for alert in alert_list:
            print(alert)

def cycle_monitor(monitor_ids, cycle_minute=60):
    """循环定时监控，默认60分钟执行一次"""
    while True:
        print(f"\n===== {time.strftime('%Y-%m-%d %H:%M:%S')} 开始一轮监控 =====")
        result = []
        now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        for iid in monitor_ids:
            info = get_goods_detail(iid)
            if not info:
                continue
            line = {
                "采集时间": now_time,
                "商品ID": iid,
                "商品标题": info.get("title", ""),
                "售价": info.get("price", 0),
                "月销量": info.get("month_sales", 0),
                "店铺名": info.get("shop_name", "")
            }
            result.append(line)
            time.sleep(1)
        # 异常检测
        check_data_change(result)
        # 追加写入表格
        df_new = pd.DataFrame(result)
        if os.path.exists(EXCEL_FILE):
            df_old = pd.read_excel(EXCEL_FILE)
            df_all = pd.concat([df_old, df_new], ignore_index=True)
        else:
            df_all = df_new
        df_all.to_excel(EXCEL_FILE, index=False)
        print(f"本轮监控完成，等待{cycle_minute}分钟后再次巡检")
        time.sleep(cycle_minute * 60)

if __name__ == "__main__":
    watch_list = ["728956123456", "654321987654"]
    cycle_monitor(watch_list, cycle_minute=30)  # 30分钟自动轮询一次

3.3 定时自动巡检 + 异常预警简易脚本

新增逻辑：对比历史表格数据，识别价格下跌、销量暴涨商品，打印预警信息，实现自动化监控

import requests
import json
import pandas as pd
import time
import os

CLIENT_ID = "你的ClientID"
CLIENT_SECRET = "你的ClientSecret"
EXCEL_FILE = "商品监控数据表.xlsx"

def get_goods_detail(item_id):
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    params = {
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "item_id": item_id
    }
    url = "https://open.onebound.cn/server/goods/query"
    try:
        resp = requests.post(url, data=params, headers=headers, timeout=30)
        res_data = resp.json()
        if res_data.get("code") == 200:
            return res_data.get("data")
        return None
    except Exception as e:
        return None

def check_data_change(new_data):
    """对比历史数据，输出异常商品预警"""
    if not os.path.exists(EXCEL_FILE):
        print("暂无历史监控记录，本次仅保存数据")
        return
    old_df = pd.read_excel(EXCEL_FILE)
    old_dict = dict(zip(old_df["商品ID"], old_df.to_dict("records")))
    alert_list = []
    for row in new_data:
        item_id = row["商品ID"]
        if item_id not in old_dict:
            alert_list.append(f"【新品上线】{row['商品标题']}")
            continue
        old = old_dict[item_id]
        # 价格下跌预警
        if float(row["售价"]) < float(old["售价"]) * 0.9:
            alert_list.append(f"【价格暴跌】{row['商品标题']} 原价{old['售价']} 现价{row['售价']}")
        # 销量暴涨预警（月销增长超500）
        if int(row["月销量"]) - int(old["月销量"]) > 500:
            alert_list.append(f"【爆款飙升】{row['商品标题']} 月销新增{int(row['月销量']) - int(old['月销量'])}")
    if alert_list:
        print("===== 监控异常预警 =====")
        for alert in alert_list:
            print(alert)

def cycle_monitor(monitor_ids, cycle_minute=60):
    """循环定时监控，默认60分钟执行一次"""
    while True:
        print(f"\n===== {time.strftime('%Y-%m-%d %H:%M:%S')} 开始一轮监控 =====")
        result = []
        now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        for iid in monitor_ids:
            info = get_goods_detail(iid)
            if not info:
                continue
            line = {
                "采集时间": now_time,
                "商品ID": iid,
                "商品标题": info.get("title", ""),
                "售价": info.get("price", 0),
                "月销量": info.get("month_sales", 0),
                "店铺名": info.get("shop_name", "")
            }
            result.append(line)
            time.sleep(1)
        # 异常检测
        check_data_change(result)
        # 追加写入表格
        df_new = pd.DataFrame(result)
        if os.path.exists(EXCEL_FILE):
            df_old = pd.read_excel(EXCEL_FILE)
            df_all = pd.concat([df_old, df_new], ignore_index=True)
        else:
            df_all = df_new
        df_all.to_excel(EXCEL_FILE, index=False)
        print(f"本轮监控完成，等待{cycle_minute}分钟后再次巡检")
        time.sleep(cycle_minute * 60)

if __name__ == "__main__":
    watch_list = ["728956123456", "654321987654"]
    cycle_monitor(watch_list, cycle_minute=30)  # 30分钟自动轮询一次

四、代码落地使用说明

  2. 密钥替换 登录 OpenClaw 个人后台，在个人中心复制两组身份凭证，替换代码中CLIENT_ID、CLIENT_SECRET，凭证长期有效，无需频繁更换。

  4. 商品 ID 提取 淘宝商品链接item.htm?id=xxxx，等号后数字即为商品 ID，批量整理到列表即可一次性监控几十上百款竞品。

  6. 定时任务调整 循环监控脚本中cycle_minute参数控制轮询间隔，新店 / 竞品多建议 30~60 分钟一次，避免短时间高频请求。

  8. 数据筛选思路（选品实操） 表格导出后，用 pandas 简单筛选即可快速锁定潜力款：

  • 筛选月销 500~5000、差评占比低于 5%：竞争适中转化稳定；

  • 筛选近一轮销量上涨 300+：近期流量爆发蓝海品；

  • 筛选有大额满减、优惠券活动：短期冲量机会款。

五、实操落地避坑指南

  2. 请求间隔控制 代码内自带 1 秒左右延迟，不要删除 sleep 语句，短时间大量查询会触发访问限制，导致数据返回为空。

  4. 数据缺失处理 部分预售、下架商品会返回空数据，脚本内置异常捕获，会打印提示并跳过，不中断整体批量任务。

  6. 本地文件备份 监控表格会持续追加数据，建议每周复制一份备份，避免文件损坏丢失长期对比数据。

  8. 跨境商品适配 后台支持切换海外站点数据源，只需在后台配置区域，代码无需修改即可同步采集海外淘宝商品信息，适合跨境卖家。

六、总结

这套自动化监控方案把原本每天 2-3 小时的人工选品工作压缩到零操作，代码一次部署后自动运行，实时捕捉竞品价格、销量、活动变动，依托 OpenClaw 稳定的数据获取能力，不用搭建复杂采集环境，中小卖家、个人运营都能低成本上手。

获取完整商品结构化数据后，结合表格筛选、简单数据分析，能精准避开内卷红海，快速捕捉短期爆发的潜力新品，大幅提升选品效率与店铺运营效率。


少长咸集

群贤毕至

访客