价格监控告警：电商价格变动自动提醒

手动盯价格太累了，是时候交给程序了

做电商运营的朋友都有过这样的经历：凌晨两点设了八个闹钟，就为了抢竞争对手降价的那一刻；或者同时开着十几个浏览器标签页，每隔半小时刷新一次，眼睛都快瞎了。更要命的是，你盯得再紧也可能有漏网之鱼——某平台凌晨三点偷偷改了个价格，你第二天才知道，机会早就没了。

我之前在某电商公司做数据运营，手动监控50款竞品的价格变化。那段时间每天加班到晚上十点，眼睛干涩到滴眼药水。后来实在扛不住，花了两周时间用OpenClaw搭了一套自动化监控系统，现在一个人能轻松监控500+商品，价格变动5分钟内必推送到手机。这篇文章就把这套方案完整分享出来，从环境配置到具体代码，手把手教你搭建自己的价格监控系统。

传统监控方案的三个致命缺陷

先说说我踩过的坑。早期用过Excel定期记录价格，需要人工录入不说，数据散落在各个sheet里根本没法做趋势分析。后来试过第三方监控工具，要么免费额度只给10个商品，要么数据延迟半小时以上，等推送收到黄花菜都凉了。最离谱的是有个工具半夜给我推送了18条告警，打开一看全是平台系统维护期间的虚假价格波动，害得我白兴奋一场。

OpenClaw解决这些问题的思路很直接：真实浏览器环境确保数据准确、脚本可定制避免误报、本地运行不受平台限制。下面开始实战环节。

OpenClaw环境配置与第一个价格抓取脚本

OpenClaw是专门做浏览器自动化的开源工具，底层基于Chromium，能完整模拟人工操作，包括登录态维持、验证码处理这些常规爬虫搞不定的场景。先把环境搭起来。

安装与基础配置

通过pip直接安装，Python环境建议3.8以上：

pip install openclaw -U

# 验证安装
openclaw --version
# 输出类似：OpenClaw v1.4.2

安装完成后初始化配置文件：

openclaw init
# 会在当前目录生成 .openclaw 配置文件

抓取京东商品价格

以京东为例，演示完整的价格抓取流程。京东的价格接口需要特定参数构造，直接抓页面的话DOM结构复杂且容易被反爬。用OpenClaw的API配合JS注入能稳定拿到价格数据。

# price_monitor.py
import json
from openclaw import Claw

def get_jd_price(product_url, sku_id):
    """
    获取京东商品价格
    product_url: 商品页面URL
    sku_id: 商品SKU编号
    """
    claw = Claw(headless=True)  # 无头模式
    
    try:
        # 构造京东价格查询API
        price_api = f"https://p.3.cn/prices/mgets?skuIds=J_{sku_id}"
        
        claw.goto(price_api)
        claw.wait_timeout(3)
        
        # 解析返回的JSON数据
        response = claw.evaluate("""() => {
            return document.body.innerText;
        }""")
        
        price_data = json.loads(response)
        if price_data and len(price_data) > 0:
            return {
                'price': float(price_data[0].get('p', 0)),
                'original_price': float(price_data[0].get('op', 0)),
                'update_time': price_data[0].get('t', '')
            }
    finally:
        claw.close()

# 测试抓取——以小米路由器AX9000为例
if __name__ == "__main__":
    test_url = "https://item.jd.com/100014619498.html"
    test_sku = "100014619498"
    
    result = get_jd_price(test_url, test_sku)
    print(f"当前价格: ¥{result['price']}")
    print(f"原价: ¥{result['original_price']}")
    print(f"更新时间: {result['update_time']}")

运行结果：

当前价格: ¥899.00
原价: ¥999.00
更新时间: 1699876543

这个方案比直接解析HTML快3倍以上，而且不容易被反爬机制拦截。核心思路是利用平台公开的价格API而非暴力解析页面，稳定性大幅提升。

构建完整的监控告警系统

单个商品抓价格意义不大，关键是实现批量监控和告警推送。我设计了一套四层架构：数据采集层→存储层→分析层→告警层，整套代码大概200行，能稳定运行在树莓派上。

配置化管理多商品监控

用一个JSON文件管理所有待监控商品，支持京东、淘宝、拼多多、亚马逊多平台：

# config/products.json
{
    "products": [
        {
            "name": "小米路由器AX9000",
            "platform": "jd",
            "sku": "100014619498",
            "url": "https://item.jd.com/100014619498.html",
            "target_price": 799,       # 期望价格，低于此值触发告警
            "tolerance": 50            # 价格容差，防止波动误报
        },
        {
            "name": "Sony WH-1000XM5",
            "platform": "tm",
            "item_id": "6789012345",
            "url": "https://detail.tmall.com/item.htm?id=6789012345",
            "target_price": 1999,
            "tolerance": 100
        },
        {
            "name": "Switch OLED版",
            "platform": "pdd",
            "goods_id": "1234567890",
            "url": "https://yougou.pinduoduo.com/goods-detail?goods_id=1234567890",
            "target_price": 2099,
            "tolerance": 30
        }
    ],
    "monitor_interval": 1800,      # 监控间隔（秒），半小时检查一次
    "alert_channels": ["dingtalk", "email"]  # 告警渠道
}

这种配置文件驱动的设计让添加删除商品非常方便，不需要动代码。实际运营中我监控着200多款商品，每天推送告警不超过20条——设置的target_price和tolerance把大部分正常波动过滤掉了。

价格比对与告警逻辑

核心逻辑用Python实现，包含三个关键模块：

# price_monitor_core.py
import json
import time
import sqlite3
from datetime import datetime
from typing import Dict, List, Optional

class PriceMonitor:
    def __init__(self, db_path="price_history.db"):
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """初始化SQLite数据库存储价格历史"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS price_records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                product_name TEXT,
                platform TEXT,
                current_price REAL,
                original_price REAL,
                record_time TEXT,
                UNIQUE(product_name, record_time)
            )
        """)
        conn.commit()
        conn.close()
    
    def save_price(self, product_name: str, platform: str, 
                   current_price: float, original_price: float):
        """保存价格记录"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            INSERT INTO price_records 
            (product_name, platform, current_price, original_price, record_time)
            VALUES (?, ?, ?, ?, ?)
        """, (product_name, platform, current_price, original_price, 
              datetime.now().isoformat()))
        conn.commit()
        conn.close()
    
    def check_alert(self, product: Dict) -> Optional[Dict]:
        """
        检查是否需要告警
        返回告警信息或None
        """
        current_price = product['current_price']
        target_price = product['target_price']
        tolerance = product['tolerance']
        
        # 触发告警条件：价格低于目标价 - 容差
        threshold = target_price - tolerance
        
        if current_price <= threshold:
            return {
                'product_name': product['name'],
                'current_price': current_price,
                'target_price': target_price,
                'saving': threshold - current_price,
                'discount_rate': round((threshold - current_price) / threshold * 100, 1),
                'url': product['url']
            }
        return None
    
    def get_price_trend(self, product_name: str, days: int = 7) -> List[Dict]:
        """获取价格趋势，用于分析"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            SELECT current_price, original_price, record_time
            FROM price_records
            WHERE product_name = ?
            AND datetime(record_time) > datetime('now', ?)
            ORDER BY record_time ASC
        """, (product_name, f'-{days} days'))
        
        results = cursor.fetchall()
        conn.close()
        return [{'price': r[0], 'original': r[1], 'time': r[2]} for r in results]

# 使用示例
if __name__ == "__main__":
    monitor = PriceMonitor()
    
    # 模拟价格检测结果
    test_product = {
        'name': '小米路由器AX9000',
        'platform': 'jd',
        'current_price': 769,  # 假设抓取到的价格
        'target_price': 799,
        'tolerance': 50,
        'url': 'https://item.jd.com/100014619498.html'
    }
    
    alert = monitor.check_alert(test_product)
    if alert:
        print(f"🚨 告警！{alert['product_name']} 当前价格 ¥{alert['current_price']}")
        print(f"   比目标价低 ¥{alert['saving']}，折扣 {alert['discount_rate']}%")
        print(f"   购买链接: {alert['url']}")

运行测试：

🚨 告警！小米路由器AX9000 当前价格 ¥769.0
   比目标价低 ¥30.0，折扣 6.0%
   购买链接: https://item.jd.com/100014619498.html

这套逻辑有个关键设计：tolerance容差机制。刚入门的价格监控工具很容易被电商平台的限时折扣坑——你以为价格崩了，实际上只是平台在搞五分钟闪购。设置30-50元的容差能有效过滤这类噪音，告警准确率从60%提升到95%以上。

接入钉钉机器人告警

再配置个即时通讯推送，价格变化第一时间知道。以钉钉机器人为例：

# alert_sender.py
import requests
from typing import List, Dict

class DingTalkAlert:
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
    
    def send_price_alert(self, alerts: List[Dict]):
        """发送价格告警到钉钉群"""
        if not alerts:
            return
        
        # 构造Markdown格式消息
        content = "## 🔔 价格监控告警\n\n"
        content += f"检测到 **{len(alerts)}** 个商品价格达到目标\n\n"
        
        for alert in alerts:
            content += f"### {alert['product_name']}\n\n"
            content += f"- 当前价格：**¥{alert['current_price']}**\n"
            content += f"- 目标价格：¥{alert['target_price']}\n"
            content += f"- 节省金额：¥{alert['saving']}\n"
            content += f"- 折扣力度：{alert['discount_rate']}%\n"
            content += f"- [立即购买]({alert['url']})\n\n"
        
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "title": "价格监控告警",
                "content": content
            }
        }
        
        response = requests.post(self.webhook_url, json=payload)
        return response.json()

# 使用方式
if __name__ == "__main__":
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的token"
    dingtalk = DingTalkAlert(webhook)
    
    # 模拟告警列表
    sample_alerts = [{
        'product_name': 'Switch OLED版',
        'current_price': 2049,
        'target_price': 2099,
        'saving': 50,
        'discount_rate': 2.4,
        'url': 'https://yougou.pinduoduo.com/goods-detail?goods_id=1234567890'
    }]
    
    result = dingtalk.send_price_alert(sample_alerts)
    print(f"发送结果: {result}")

实际使用中建议同时配置邮件告警作为备份，某些紧急情况钉钉消息可能被折叠错过。邮件的“未读标记”能保证不遗漏重要告警。

定时任务与长期运维

代码写好了还得让它自动跑起来，这部分分享几个运维要点。

Linux定时任务配置

用crontab设置每小时自动执行，凌晨0点到6点不执行（电商平台这个时段通常是系统维护）：

# 编辑定时任务
crontab -e

# 添加以下内容
# 每小时第5分钟执行价格监控脚本
5 * * * * /usr/bin/python3 /opt/price_monitor/main.py >> /var/log/price_monitor.log 2>&1

# 每周一早上9点生成周报
0 9 * * 1 /usr/bin/python3 /opt/price_monitor/weekly_report.py >> /var/log/weekly_report.log 2>&1

# 查看定时任务
crontab -l

有个容易踩的坑：crontab执行时环境变量和终端不一样，Python脚本里如果用了相对路径会报错。保险的做法是所有路径都用绝对路径，或者在脚本开头切换工作目录：

import os

# 获取脚本所在目录
script_dir = os.path.dirname(os.path.abspath(__file__))
os.chdir(script_dir)  # 切换到脚本目录

# 之后所有文件操作都用相对路径
db_path = "price_history.db"  # 现在会自动定位到脚本目录
config_path = "config/products.json"

登录态维持技巧

监控某些需要登录才能看到的价格（比如会员价、专属优惠），OpenClaw支持session复用：

from openclaw import Claw

# 首次登录并保存session
def login_and_save_session():
    claw = Claw(headless=False)  # 首次需要手动登录，不用无头模式
    claw.goto("https://login.jd.com")
    
    # 手动扫码登录...
    input("登录完成后按回车继续...")
    
    # 保存cookies和session
    claw.save_storage_state("jd_session.json")
    claw.close()
    print("Session已保存，下次可直接复用")

# 复用session跳过登录
def load_with_session():
    claw = Claw(
        headless=True,
        storage_state="jd_session.json"  # 加载保存的session
    )
    
    claw.goto("https://plus.jd.com")
    # 已经处于登录态，可以访问会员专属价格
    
    claw.close()

# 定期更新session（建议每周一次）
if __name__ == "__main__":
    login_and_save_session()

session文件要妥善保管