ECサイトと競合分析 9 分で読めます

プロキシで競合の価格を自動監視する方法

住宅用プロキシを使用して自動競合価格監視システムを構築します。 PythonとNode.jsコード、スケジューリング戦略、およびアラート設定で完全なアーキテクチャガイド.

ProxyHat Team
プロキシで競合の価格を自動監視する方法

なぜ自動価格監視マットレス

競争力のある電子商取引市場では、価格は常に変化します。 競合他社は、午前2時00分に価格を落とす可能性があり、通知した時点で、すでに販売日を失います。 自動価格監視は、競争相手の価格を継続的に追跡し、リアルタイムで変化に警告することにより、この盲点を排除します。

競争力を維持するために価格を調整小売業者であるかどうか, ブランド監視マップ (最小広告価格) コンプライアンス, またはアナリスト追跡市場の傾向, 十分に構築された価格監視システムは、すぐにそれを支払う. すべての作業を確実にする重要な成分は、堅牢なプロキシインフラストラクチャです。それなしで、監視リクエストは数時間以内にブロックされます。 より広範なECデータ収集を見るには、 eコマースデータスクレイピングガイド. .

価格監視システムのアーキテクチャ

製造グレードの価格監視システムは、URL マネージャー、スクレーピング エンジン、データ ストア、および警告層の 4 つの主要なコンポーネントを持っています。

価格監視システムのアーキテクチャ
コンポーネントミッションテクノロジー
URL マネージャーターゲット URL を保存し、メタデータをスケジューリングし、周波数をスクレイピングPostgreSQL、Redis
スクレイピングエンジンプロキシ、抽出物価格によるフェッチページPython/Node.js、ProxyHat、BeautifulSoup/Cheerio
データストアタイムスタンプで価格履歴を保存PostgreSQL、タイムスケールDB、ClickHouse
アラートシステム変更を検出し、通知を送信Webhooks、Slack、メール、SMS

スケジューリング戦略

すべての製品は同じ監視周波数を必要としません。 高優先アイテム(トップ100のSKU、ダイレクト競合製品)は、毎日ロングテールアイテムがチェックできますが、毎時チェックが必要な場合があります。 優先順位:

  • 価格のボラティリティ: 価格を頻繁に変えるプロダクトはより頻繁な点検を必要とします。
  • 収益の影響: あなたのベストセラーはより高い監視優先順位に値します。
  • 競争力のある密度: 多くの競合他社とのカテゴリには、より厳しい監視が必要です。

監視用のプロキシ回転の設定

価格監視は、日数、週数、月数の繰り返し同じURLを打つことを意味します。 このパターンは、アンチボットシステムが検出するように設計されています。 自動回転の住宅用プロキシは不可欠です。

ProxyHat の設定

# 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

各価格の点検が独立した操作であるので価格の監視のために、/requestの回転は最もよく働きます。 使用条件 ジオターゲティングプロキシ 地域の価格設定の違いを監視するとき。

Pythonの実装

ここでは、Pythonで構築された完全な価格監視システムです。 ProxyHatの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()

Node.js 実装

Node.js を使用するチームでは、 Node.js を使用するチームでは、 Node.js を使用して同等の監視設定があります。 ProxyHat ノード 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
);

データストレージと分析

トレンドを時間をかけて分析できると、原価データが貴重になります。

データベーススキーマ

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;

アラートと通知

自動化されたアラートは、価格変化に迅速に反応します。 共通の通知チャネルは下記のものを含んでいます:

  • Slack の webhooks: チーム全体の可視性に最適です。 構造化されたメッセージを価格変更の細部と送ります。
  • メールダイジェスト: あなたのしきい値の上にすべての価格の変更の毎日または1時間合計。
  • Webhookコールバック: 価格が変更されるときあなたのreplricingエンジンか他のオートメーションをトリガーして下さい。
  • ダッシュボード: すべての監視されたプロダクトを渡る価格の傾向の実時間視覚化。

アラートの閾値

異なるシナリオの異なるアラート境界を設定:

アラートの閾値
スケナリオパスワードアクション
競争価格の低下> 5%税率Slack通知
競争価格の低下> 15%15%の価格設定チーム+自動価格へのメール
在庫切れ株式変更機会アラート
サイトマップMAP 値の下コンプライアンス警告

監視のためのプロキシベストプラクティス

継続的な監視により、1回のスクレイピングと比較して、プロキシ管理のためのユニークな課題が生まれます。

  • 時間を割く要求: 深夜に10,000製品をすべてチェックする代わりに、間隔全体でチェックをスプレッドします。 これは、着実に低プロファイルの要求パターンを作成します。
  • 住宅のプロキシを使用して下さい: 住宅用プロキシ 同じデータセンターのIPが毎日同じサイトに当たるので、長期にわたる監視には不可欠です。
  • 一致の地理位置: 地域価格設定を監視する場合、対象地域からプロキシを使用する。 ドイツの価格をチェックする米国のIPは、誤ったデータを参照するか、リダイレクトされます。
  • ハンドルの失敗は優雅に: リクエストが失敗した場合は、すぐに再要求するのではなく、指数関数的なバックオフで待機して再試行してください。 成功率を監視し、それが落ちるならば、通貨を削減します。
  • キャッシュと重複排除: 価格が変更されていない場合は、重複したレコードを保存しないでください。 データベースのリーンを維持し、分析を高速化します。
主要なテイクアウト:価格の監視は、スプリントではなくマラソンです。 破裂ではなく、安定した持続可能なパターンを要求するためのシステムの設計。

モニタリングシステムをスケーリング

製品カタログが成長するにつれて、スケーリングが重要になります。 ここが働くパターンです。

  • ワーカープール: ジョブキュー(Redis, RabbitMQ)からプルする複数のワーカーを使用します。 各ワーカーは、独自のプロキシ接続を持ち、独立して運営しています。
  • 優先キュー: 高価な製品は、まず、より頻繁にチェックされます。 低い優先項目は残りの容量を満たします。
  • 適応スケジューリング: 商品の価格が7日で変更されていない場合は、自動的にチェック頻度を削減します。 今日に2回変更した場合、頻度を増加させます。
  • サイトあたりのレート制限: 各ターゲットサイトのレート制限値を尊重します。 アマゾン、ウォルマート、ニッチの店はすべて異なる公差を持っています。

スクレイピング操作のスクレイピングの詳細については、ガイドを参照してください 2026年にWebスクレイピングに最適なプロキシ 探検する ProxyHatの料金プラン 大量の監視のため。

キーテイクアウト

  • 自動価格監視では、堅牢なアーキテクチャが必要です。URLマネージャー、スクレイピングエンジン、データストア、アラートシステム。
  • レジデンシャルプロキシは、ブロックなしで持続的な監視のために不可欠です。
  • スケジュールは優先的にチェックします。すべての製品が毎時監視を必要としません。
  • トレンド分析のためのタイムシリーズフレンドリーなスキーマで価格履歴を保存します。
  • tiered アラートのしきい値を設定し、ノイズリダクションによる応答性のバランスを整えます。
  • 持続可能で控えめなスクレイピングパターンを時間をかけても要求を分散させます。

あなたの価格監視システムを構築する準備はできましたか? まずは ProxyHatの住宅用プロキシ お問い合わせ eコマーススクレイピングガイド 完全な戦略のため。 技術的な実装の詳細については、ガイドを参照してください。 Pythonでプロキシを使う そして、 Node.js でプロキシを使用する. .

始める準備はできましたか？

AIフィルタリングで148か国以上、5,000万以上のレジデンシャルIPにアクセス。

料金を見るレジデンシャルプロキシ
← ブログに戻る