如何使用代理自动监控竞争对手价格

为何自动价格监测事项

在有竞争力的电子商务市场,价格不断变化。 竞争对手可能会在凌晨2点时降低5%的价格,当你注意到时,你已经失去了一天的销售时间. 自动价格监测通过不断跟踪竞争者价格并提醒你实时变化,消除了这一盲点.

无论是零售商调整价格以保持竞争力,还是品牌监控MAP(最低广告价格)合规性,还是分析师跟踪市场趋势,精心构建的价格监控系统都迅速支付费用. 使它都可靠运作的关键要素是一个强大的代理基础设施——没有它,你的监测请求就会在数小时内被阻止。 关于电子商务数据收集的更广泛情况,见我们 电子商务数据报废指南。 。 。

价格监测系统的结构

一个生产级价格监测系统主要有四个组成部分:URL管理器,刮刮引擎,数据库,以及提示层.

时间安排战略

并非所有产品都需要同样的监测频率。 高度优先项目(你的前100SKU,直接竞争者产品)可能需要小时检查,而长尾项目则可以每天检查。 根据下列因素确定优先次序:

• 价格波动: 改变价格的产品往往需要更频繁的检查。
• 收入影响: 你们的畅销书应该有更高的优先监控
• 竞争密度: 许多竞争对手的类别需要更严格的监测。

为监测设置代理旋转

价格监测意味着在数日、数周和数月内反复点击同样的URL。 这种模式正是反机器人系统设计用来探测的. 拥有自动旋转的住宅代用品至关重要。

代理汉特配置

# Standard rotating proxy (new IP per request)
http://USERNAME:PASSWORD@gate.proxyhat.com:8080
# Geo-targeted for regional pricing (e.g., US prices)
http://USERNAME-country-US:PASSWORD@gate.proxyhat.com:8080
# Session-based for multi-page price checks
http://USERNAME-session-price001:PASSWORD@gate.proxyhat.com:8080

在价格监测方面,每次要求的轮换效果最好,因为每次价格检查都是独立的操作. 使用 地理目标代号 在监测区域价格差异时。

Python 执行

这是用Python建造的完整的价格监测系统, 代理哈特的 Python SDK。 。 。

价格计算模块

import requests
from bs4 import BeautifulSoup
import json
import time
import random
from datetime import datetime
from dataclasses import dataclass, asdict
PROXY_URL = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
]
@dataclass
class PriceResult:
    url: str
    price: float | None
    currency: str | None
    in_stock: bool
    scraped_at: str
    seller: str | None = None
def scrape_price(url: str, selectors: dict) -> PriceResult:
    """Scrape a product price from any e-commerce site."""
    headers = {
        "User-Agent": random.choice(USER_AGENTS),
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9",
    }
    proxies = {"http": PROXY_URL, "https": PROXY_URL}
    try:
        response = requests.get(url, headers=headers, proxies=proxies, timeout=30)
        response.raise_for_status()
    except requests.RequestException as e:
        return PriceResult(
            url=url, price=None, currency=None,
            in_stock=False, scraped_at=datetime.utcnow().isoformat()
        )
    soup = BeautifulSoup(response.text, "html.parser")
    price = extract_price(soup, selectors.get("price"))
    currency = selectors.get("currency", "USD")
    in_stock = check_stock(soup, selectors.get("stock"))
    return PriceResult(
        url=url,
        price=price,
        currency=currency,
        in_stock=in_stock,
        scraped_at=datetime.utcnow().isoformat(),
    )
def extract_price(soup, selector: str) -> float | None:
    """Extract and parse price from a CSS selector."""
    if not selector:
        return None
    el = soup.select_one(selector)
    if not el:
        return None
    text = el.get_text(strip=True)
    # Remove currency symbols, commas, spaces
    cleaned = "".join(c for c in text if c.isdigit() or c == ".")
    try:
        return float(cleaned)
    except ValueError:
        return None
def check_stock(soup, selector: str) -> bool:
    """Check if product is in stock."""
    if not selector:
        return True
    el = soup.select_one(selector)
    if not el:
        return False
    text = el.get_text(strip=True).lower()
    return "in stock" in text or "available" in text
# Site-specific selector configurations
SITE_SELECTORS = {
    "amazon.com": {
        "price": "span.a-price-whole",
        "stock": "#availability span",
        "currency": "USD",
    },
    "walmart.com": {
        "price": "[data-testid='price-wrap'] span.f2",
        "stock": "[data-testid='fulfillment-badge']",
        "currency": "USD",
    },
    "target.com": {
        "price": "[data-test='product-price']",
        "stock": "[data-test='fulfillmentSection']",
        "currency": "USD",
    },
}

监测调度器

import schedule
import threading
from collections import defaultdict
class PriceMonitor:
    def __init__(self, db_connection):
        self.db = db_connection
        self.price_history = defaultdict(list)
    def add_product(self, url: str, site: str, check_interval_minutes: int = 60):
        """Register a product for monitoring."""
        selectors = SITE_SELECTORS.get(site, {})
        def check():
            result = scrape_price(url, selectors)
            self.price_history[url].append(result)
            self.store_result(result)
            self.check_alerts(url, result)
            time.sleep(random.uniform(1, 3))
        schedule.every(check_interval_minutes).minutes.do(check)
    def store_result(self, result: PriceResult):
        """Store price result in database."""
        # Insert into price_history table
        self.db.execute(
            "INSERT INTO price_history (url, price, currency, in_stock, scraped_at) "
            "VALUES (%s, %s, %s, %s, %s)",
            (result.url, result.price, result.currency,
             result.in_stock, result.scraped_at)
        )
    def check_alerts(self, url: str, result: PriceResult):
        """Check if price change triggers an alert."""
        history = self.price_history[url]
        if len(history) < 2:
            return
        prev = history[-2]
        curr = history[-1]
        if prev.price and curr.price:
            change_pct = ((curr.price - prev.price) / prev.price) * 100
            if abs(change_pct) >= 5:  # 5% threshold
                self.send_alert(url, prev.price, curr.price, change_pct)
        # Stock status change
        if prev.in_stock and not curr.in_stock:
            self.send_alert(url, msg="Product went out of stock")
        elif not prev.in_stock and curr.in_stock:
            self.send_alert(url, msg="Product back in stock")
    def send_alert(self, url, old_price=None, new_price=None,
                   change_pct=None, msg=None):
        """Send price change notification."""
        if msg:
            print(f"ALERT [{url}]: {msg}")
        else:
            direction = "dropped" if change_pct < 0 else "increased"
            print(f"ALERT [{url}]: Price {direction} {abs(change_pct):.1f}% "
                  f"(${old_price} -> ${new_price})")
    def run(self):
        """Start the monitoring loop."""
        while True:
            schedule.run_pending()
            time.sleep(1)
# Usage
if __name__ == "__main__":
    monitor = PriceMonitor(db_connection=None)  # Replace with actual DB
    # Monitor competitor products
    monitor.add_product(
        "https://www.amazon.com/dp/B0CHX3QBCH",
        site="amazon.com",
        check_interval_minutes=60,
    )
    monitor.add_product(
        "https://www.amazon.com/dp/B0D5BKRY4R",
        site="amazon.com",
        check_interval_minutes=30,  # Higher priority
    )
    monitor.run()

节点.js 执行

对于使用Node.js的团队来说,这里有一个相应的监控设置,使用 代理哈特节点 SDK。 。 。

const axios = require("axios");
const cheerio = require("cheerio");
const { HttpsProxyAgent } = require("https-proxy-agent");
const cron = require("node-cron");
const PROXY_URL = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080";
const agent = new HttpsProxyAgent(PROXY_URL);
const USER_AGENTS = [
  "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
  "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
];
async function scrapePrice(url, selectors) {
  try {
    const { data } = await axios.get(url, {
      httpsAgent: agent,
      headers: {
        "User-Agent": USER_AGENTS[Math.floor(Math.random() * USER_AGENTS.length)],
        "Accept-Language": "en-US,en;q=0.9",
      },
      timeout: 30000,
    });
    const $ = cheerio.load(data);
    const priceText = $(selectors.price).first().text().trim();
    const price = parseFloat(priceText.replace(/[^0-9.]/g, "")) || null;
    return {
      url,
      price,
      currency: selectors.currency || "USD",
      inStock: $(selectors.stock).text().toLowerCase().includes("in stock"),
      scrapedAt: new Date().toISOString(),
    };
  } catch (err) {
    return { url, price: null, currency: null, inStock: false, scrapedAt: new Date().toISOString() };
  }
}
class PriceMonitor {
  constructor() {
    this.products = [];
    this.history = new Map();
  }
  addProduct(url, selectors, cronExpression = "0 * * * *") {
    this.products.push({ url, selectors, cronExpression });
    this.history.set(url, []);
    cron.schedule(cronExpression, async () => {
      const result = await scrapePrice(url, selectors);
      const prev = this.history.get(url);
      prev.push(result);
      if (prev.length >= 2) {
        const last = prev[prev.length - 2];
        if (last.price && result.price) {
          const changePct = ((result.price - last.price) / last.price) * 100;
          if (Math.abs(changePct) >= 5) {
            console.log(`ALERT [${url}]: Price changed ${changePct.toFixed(1)}%`);
          }
        }
      }
      console.log(`Checked ${url}: $${result.price} (${result.inStock ? "in stock" : "out of stock"})`);
    });
  }
}
// Usage
const monitor = new PriceMonitor();
monitor.addProduct(
  "https://www.amazon.com/dp/B0CHX3QBCH",
  { price: "span.a-price-whole", stock: "#availability span", currency: "USD" },
  "0 * * * *"  // Every hour
);
monitor.addProduct(
  "https://www.amazon.com/dp/B0D5BKRY4R",
  { price: "span.a-price-whole", stock: "#availability span", currency: "USD" },
  "*/30 * * * *"  // Every 30 minutes
);

数据储存和分析

在分析长期趋势时,原始价格数据变得非常宝贵。

数据库Schema

CREATE TABLE monitored_products (
    id SERIAL PRIMARY KEY,
    url TEXT NOT NULL,
    site VARCHAR(100) NOT NULL,
    product_name VARCHAR(500),
    our_sku VARCHAR(100),
    check_interval_minutes INT DEFAULT 60,
    is_active BOOLEAN DEFAULT true,
    created_at TIMESTAMPTZ DEFAULT now()
);
CREATE TABLE price_history (
    id SERIAL PRIMARY KEY,
    product_id INT REFERENCES monitored_products(id),
    price DECIMAL(10, 2),
    currency VARCHAR(3) DEFAULT 'USD',
    in_stock BOOLEAN,
    scraped_at TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX idx_price_history_product_time
    ON price_history (product_id, scraped_at DESC);

价格趋势查询

-- Average daily price for the last 30 days
SELECT
    date_trunc('day', scraped_at) AS day,
    AVG(price) AS avg_price,
    MIN(price) AS min_price,
    MAX(price) AS max_price
FROM price_history
WHERE product_id = 1
  AND scraped_at >= now() - INTERVAL '30 days'
GROUP BY day
ORDER BY day;
-- Products with price drops > 10% in the last 24 hours
SELECT
    mp.product_name,
    mp.url,
    old_prices.avg_price AS price_yesterday,
    new_prices.avg_price AS price_today,
    ((new_prices.avg_price - old_prices.avg_price) / old_prices.avg_price * 100) AS change_pct
FROM monitored_products mp
JOIN LATERAL (
    SELECT AVG(price) AS avg_price
    FROM price_history
    WHERE product_id = mp.id
      AND scraped_at BETWEEN now() - INTERVAL '48 hours' AND now() - INTERVAL '24 hours'
) old_prices ON true
JOIN LATERAL (
    SELECT AVG(price) AS avg_price
    FROM price_history
    WHERE product_id = mp.id
      AND scraped_at >= now() - INTERVAL '24 hours'
) new_prices ON true
WHERE ((new_prices.avg_price - old_prices.avg_price) / old_prices.avg_price * 100) < -10;

警报和通知

自动提示可以确保您对价格变化做出快速反应. 常见的通知渠道包括:

• 黑网呼声 : 理想的全团队知名度。 发送带有价格变化细节的结构化消息 。
• 电子邮件摘要 : 每日或小时汇总超过门槛的所有价格变动。
• Webhook 回电: 当价格变化时触发您重置的引擎或其他自动化.
• 盘片 : 实时可视化所有受监测产品的价格趋势。

警戒阈值

为不同的情景配置不同的警报阈值 :

监测的代理最佳做法

与一次性报废相比,持续监测给代理管理带来了独特的挑战。

• 随时间分配请求 : 而不是在午夜检查所有10 000种产品,而是在整个间隔期间进行分散检查。 这形成了一种稳定,低调的请求模式.
• 使用住宅代理 : 住宅代理机构 对长期监测至关重要,因为每天袭击同一网站的同一数据中心IP将被禁止。
• 匹配地理位置 : 在监测区域定价时,使用目标区域的代理。 一个美国IP检查德国价格会看到错误的数据或被重新定向.
• 优雅地处理失败: 如果请求失败, 请等待并重试 。 监视您的成功率, 如果下降的话, 则降低货币 。
• 缓存和调试 : 价格不变的,不要存储重复记录. 这样可以保持数据库的精度,使分析更快.

钥匙外卖:价格监测是马拉松,不是冲刺. 设计您的系统 稳定,可持续的要求模式 而不是破碎的刮。

扩大你的监测系统

随着产品目录的增长,缩放变得至关重要. 以下是行之有效的模式:

• 员工池: 使用多个工人从工作队列中拉出(Redis, RabbitMQ). 每个工人都有自己的代理关系,独立运作。
• 优先级队列 : 高价值产品首先要检查,而且更经常地检查。 低优先项目填补剩余能力。
• 适应性时间安排 : 如果产品的价格在7天内没有改变,则自动减少检查频率. 如果今天改变两次,增加频率.
• 限制每个地点的费率 : 遵守每个目标地点的速率限制. 亚马逊,沃尔玛,和利基商店都有不同的耐受性.

更多关于拆卸操作,请检查我们的指南 2026年网络刮刮的最佳代理 并探索 代理哈特的定价计划 用于高容量监测。

关键外卖

• 自动价格监测需要一个强有力的架构:URL管理器、刮刮引擎、数据储存和警报系统。
• 按要求轮换的住所代理人对于无阻的持续监测至关重要。
• 根据优先次序进行时间表检查——并非所有产品都需要小时监测。
• 将价格历史存储在便于时间序列分析的计划中。
• 配置分级警戒阈值,以平衡反应和降低噪音.
• 随着时间的推移,平均分配关于可持续、低调刮刮方式的请求。

准备好建立价格监控系统了吗? 开始 代理哈特的住宅代理 读我们 电子商务刮刮指南 为整个战略。 关于技术实施的细节,见我们的指南: 使用 Python 中的代理 和 使用节点中的代理。 。 。