Gerçek Zamanlı Fiyat İzleme Altyapısı Oluşturma

Real-Time vs Batch Fiyat İzleme

Çoğu fiyat izleme sistemleri toplu modda çalışır: tüm ürünleri her saat kontrol edin (veya her birkaç saat), sonuçları saklar ve değişiklikler hakkında uyarılar gönderir. Bu, birçok kullanım vakası için çalışır, ancak hızlı hareket piyasalarında - flaş satışları, dinamik fiyatlandırma, pazar rekabeti - toplu izleme kontroller arasında gerçekleşen kritik fiyat değişikliklerini özlüyor.

Gerçek zamanlı fiyat izleme, saatlerden dakikaya veya hatta saniyelere kadar tespiti azaltır. Sabit bir programda her ürünü kontrol etmek yerine, gerçek zamanlı bir sistem sürekli olarak yüksek öncelikli hedefleri izler ve olduğu gibi değişikliklere tepki verir. Bu kılavuz, gerçek zamanlı bir izleme sistemi oluşturmak için gerekli olan mimarlık, proxy altyapısı ve uygulama detaylarını kapsar. Temel fiyat izleme kavramları için rehberimizi görün Rakip fiyatları otomatik olarak izleme.

Gerçek Zaman İzlemesi için Mimari

Gerçek zamanlı bir fiyat izleme sistemi sürekli bir boru hattı olarak birlikte çalışan beş temel bileşene sahiptir.

1. Öncelik Queue

Ürünler kontrol frekansı belirleyen öncelikli tiers verilir. Öncelikli bir kuyruk (Redis Sorted Setler iyi çalışır) yüksek değerli ürünler her zaman ilk kontrol edilir.

import redis
import time
import json
r = redis.Redis(host="localhost", port=6379)
def add_product(product_id, url, priority_minutes):
    """Add a product to the monitoring queue."""
    next_check = time.time()  # Check immediately on first add
    r.zadd("price_queue", {json.dumps({
        "product_id": product_id,
        "url": url,
        "interval": priority_minutes * 60,
    }): next_check})
def get_next_batch(batch_size=10):
    """Get the next batch of products due for checking."""
    now = time.time()
    items = r.zrangebyscore("price_queue", 0, now, start=0, num=batch_size)
    products = []
    for item in items:
        data = json.loads(item)
        r.zadd("price_queue", {item: now + data["interval"]})
        products.append(data)
    return products
# Example: Add products with different priorities
add_product("SKU001", "https://www.amazon.com/dp/B0CHX3QBCH", priority_minutes=2)
add_product("SKU002", "https://www.amazon.com/dp/B0D5BKRY4R", priority_minutes=5)
add_product("SKU003", "https://www.amazon.com/dp/B0CRMZHDG7", priority_minutes=15)

2. Worker Pool

Birden çok işçi süreçleri öncelikli kuyruktan çekilir, fiyatlar taşıyıcılar getirir ve sonuçları veri boru hattına taşır. İşçiler bağımsız olarak çalışır, her biri kendi proxy bağlantısı ile.

import requests
import random
from concurrent.futures import ThreadPoolExecutor
from bs4 import BeautifulSoup
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",
]
def fetch_price(product):
    """Fetch the current price for a product."""
    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(
            product["url"], headers=headers,
            proxies=proxies, timeout=30
        )
        if response.status_code == 200:
            soup = BeautifulSoup(response.text, "html.parser")
            price_el = soup.select_one("span.a-price-whole")
            if price_el:
                price = float(price_el.get_text(strip=True).replace(",", ""))
                return {
                    "product_id": product["product_id"],
                    "price": price,
                    "timestamp": time.time(),
                    "status": "success",
                }
    except Exception as e:
        pass
    return {
        "product_id": product["product_id"],
        "price": None,
        "timestamp": time.time(),
        "status": "failed",
    }
def run_workers(num_workers=10):
    """Run the monitoring worker pool."""
    with ThreadPoolExecutor(max_workers=num_workers) as executor:
        while True:
            batch = get_next_batch(batch_size=num_workers)
            if not batch:
                time.sleep(1)
                continue
            futures = [executor.submit(fetch_price, product) for product in batch]
            for future in futures:
                result = future.result()
                process_result(result)
            time.sleep(random.uniform(0.5, 2))

3. Uyarlama Motoru Değiştirin

Her fiyat kontrolünü depolamak yerine, değişim algılama motoru son bilinen değerlere karşı mevcut fiyatları karşılaştırır ve sadece gerçek değişikliklerle ilgili olayları tetikler.

class ChangeDetector:
    def __init__(self, redis_client):
        self.redis = redis_client
    def check_change(self, product_id, new_price):
        """Compare new price against last known and detect changes."""
        key = f"last_price:{product_id}"
        last_data = self.redis.get(key)
        if last_data:
            last = json.loads(last_data)
            old_price = last["price"]
            if old_price and new_price and old_price != new_price:
                change_pct = ((new_price - old_price) / old_price) * 100
                event = {
                    "product_id": product_id,
                    "old_price": old_price,
                    "new_price": new_price,
                    "change_pct": round(change_pct, 2),
                    "timestamp": time.time(),
                }
                # Publish change event
                self.redis.publish("price_changes", json.dumps(event))
                return event
        # Update last known price
        self.redis.set(key, json.dumps({
            "price": new_price,
            "timestamp": time.time(),
        }))
        return None

4. Event Stream

Fiyat değişiklikleri Redis Pub/Sub kanalına (veya Kafka daha büyük sistemler için konu). Downstream tüketiciler - uyarı hizmetleri, repricing motorları, panolar - bu olaylara abone olun ve bağımsız olarak tepki verin.

import redis
import json
def subscribe_to_changes():
    """Subscribe to price change events."""
    r = redis.Redis(host="localhost", port=6379)
    pubsub = r.pubsub()
    pubsub.subscribe("price_changes")
    for message in pubsub.listen():
        if message["type"] == "message":
            event = json.loads(message["data"])
            handle_price_change(event)
def handle_price_change(event):
    """Process a price change event."""
    change = event["change_pct"]
    product = event["product_id"]
    if change < -10:
        send_urgent_alert(event)  # Major price drop
    elif change < -5:
        send_alert(event)         # Moderate drop
    elif change > 10:
        send_alert(event)         # Significant increase
    # Always log to time-series database
    store_price_change(event)

5. Dashboard ve Uyarılar

Gerçek zamanlı veriler gerçek zamanlı görselleştirmeye ihtiyaç duyar. WebSocket bağlantılarını, fiyat güncelleştirmelerini anında paniğe zorlamak için kullanın.

Node.js Uygulama

Gerçek zamanlı izleme motorunun Node.js versiyonunu kullanarak ProxyHat's Node SDK.

const axios = require("axios");
const { HttpsProxyAgent } = require("https-proxy-agent");
const Redis = require("ioredis");
const cheerio = require("cheerio");
const PROXY_URL = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080";
const redis = new Redis();
class RealTimePriceMonitor {
  constructor(concurrency = 10) {
    this.concurrency = concurrency;
    this.running = false;
    this.agent = new HttpsProxyAgent(PROXY_URL);
  }
  async fetchPrice(product) {
    try {
      const { data } = await axios.get(product.url, {
        httpsAgent: new HttpsProxyAgent(PROXY_URL),
        headers: {
          "User-Agent":
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36",
          "Accept-Language": "en-US,en;q=0.9",
        },
        timeout: 30000,
      });
      const $ = cheerio.load(data);
      const priceText = $("span.a-price-whole").first().text().trim();
      const price = parseFloat(priceText.replace(/,/g, "")) || null;
      return { productId: product.productId, price, timestamp: Date.now(), status: "success" };
    } catch (err) {
      return { productId: product.productId, price: null, timestamp: Date.now(), status: "failed" };
    }
  }
  async checkChange(productId, newPrice) {
    const key = `last_price:${productId}`;
    const lastData = await redis.get(key);
    if (lastData) {
      const last = JSON.parse(lastData);
      if (last.price && newPrice && last.price !== newPrice) {
        const changePct = ((newPrice - last.price) / last.price) * 100;
        const event = {
          productId,
          oldPrice: last.price,
          newPrice,
          changePct: Math.round(changePct * 100) / 100,
          timestamp: Date.now(),
        };
        await redis.publish("price_changes", JSON.stringify(event));
        return event;
      }
    }
    await redis.set(key, JSON.stringify({ price: newPrice, timestamp: Date.now() }));
    return null;
  }
  async processProduct(product) {
    const result = await this.fetchPrice(product);
    if (result.price) {
      const change = await this.checkChange(result.productId, result.price);
      if (change) {
        console.log(
          `Price change: ${change.productId} $${change.oldPrice} -> $${change.newPrice} (${change.changePct}%)`
        );
      }
    }
    // Random delay
    await new Promise((r) => setTimeout(r, 500 + Math.random() * 1500));
  }
  async start() {
    this.running = true;
    console.log(`Starting monitor with ${this.concurrency} workers`);
    while (this.running) {
      const batch = await this.getNextBatch(this.concurrency);
      if (batch.length === 0) {
        await new Promise((r) => setTimeout(r, 1000));
        continue;
      }
      await Promise.all(batch.map((p) => this.processProduct(p)));
    }
  }
  async getNextBatch(size) {
    const now = Date.now() / 1000;
    const items = await redis.zrangebyscore("price_queue", 0, now, "LIMIT", 0, size);
    const products = [];
    for (const item of items) {
      const data = JSON.parse(item);
      await redis.zadd("price_queue", now + data.interval, item);
      products.push(data);
    }
    return products;
  }
}
const monitor = new RealTimePriceMonitor(10);
monitor.start();

Sürekli İzleme için Proxy Management

Gerçek zamanlı izleme, altyapınızda toplu kazılarla kıyasla benzersiz talepler tutar.

Steady-State Request Pattern

Taleplerin patlamalarını gönderen toplu kazı aksine, gerçek zamanlı izleme sürekli bir akış yaratır. Bu aslında proxy sağlığı için daha iyidir - ikinci başına kalıcı bir akış 2 dakikalık bir patlamada 1.000'den fazla istek görünüyor.

ProxyHat Build for Real-Time

# Per-request rotation (default, recommended for most checks)
http://USERNAME:PASSWORD@gate.proxyhat.com:8080
# Geo-targeted for marketplace-specific monitoring
http://USERNAME-country-US:PASSWORD@gate.proxyhat.com:8080
http://USERNAME-country-DE:PASSWORD@gate.proxyhat.com:8080
# SOCKS5 for lower-level protocol control
socks5://USERNAME:PASSWORD@gate.proxyhat.com:1080

IP Health Watch

Hedef site başına başarı oranları takip edin ve yaklaşımınızı dinamik olarak ayarlar. Başarı oranları belirli bir pazarda düşerse, gecikmeleri artırmak veya farklı bir geo-targeted havuza geçiş yapın. ProxyHat'ın büyük konut havuzu, her zaman mevcut taze IP'lere sahip olmanızı sağlar. Kontrolümüzü kontrol edin lokasyon yerleri lokasyon yerleri lokasyon yerleri lokasyon yerleri lokasyon yerleri lokasyon yerleri Tam kapsama için.

Key takeaway: Gerçek zamanlı izleme istikrarlı, sürdürülebilir bir proxy stratejisi gerektirir. Hedef birçok IP'de tutarlı düşük hacimli istekler, birkaç IP'den yüksek hacimli patlamalar değildir.

Gerçek Zamanlı Veri Deposu

Gerçek zamanlı fiyat verileri yüksek frekanslı eklemeler ve zaman aralığı sorgular için optimize edilmiş bir depolama çözümüne ihtiyaç duyar.

TimescaleDB Schema

-- TimescaleDB hypertable for price data
CREATE TABLE price_ticks (
    time        TIMESTAMPTZ NOT NULL,
    product_id  TEXT NOT NULL,
    price       DECIMAL(10,2),
    currency    VARCHAR(3) DEFAULT 'USD',
    source_url  TEXT,
    status      VARCHAR(20)
);
SELECT create_hypertable('price_ticks', 'time');
-- Continuous aggregate for hourly summaries
CREATE MATERIALIZED VIEW price_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS hour,
    product_id,
    AVG(price) AS avg_price,
    MIN(price) AS min_price,
    MAX(price) AS max_price,
    COUNT(*) AS check_count
FROM price_ticks
WHERE status = 'success'
GROUP BY hour, product_id;
-- Retention policy: keep raw ticks for 30 days
SELECT add_retention_policy('price_ticks', INTERVAL '30 days');
-- Keep hourly aggregates for 1 year
SELECT add_retention_policy('price_hourly', INTERVAL '365 days');

Scaling Thinkations

• Yatay işçi ölçeklendirme: Birden fazla makinede işçiler ekleyin, her biri aynı Redis kuyruğundan çekiyor. Hiçbir koordinasyon gerekli - kuyruk dağıtımını ele alır.
• Öncekilik temelli throttling: Bütçe sınırlı olduğunda, eleştirel maddeler için gerçek zamanlı kapsama devam ederken düşük öncelikli ürünler için otomatik olarak kontrol frekansı azaltır.
• Adaptif aralıklar: Bir ürünün fiyatı 24 saat boyunca stabilse, kontrol aralığı artırmak. Bir saat içinde iki kez değiştiyse, azaltın.
• Siteye özgü koncurrency: Farklı hedef siteleri farklı toleranslara sahiptir. Azure için daha fazla eş zamanlı çalışan çalıştırın (daha fazla ikna edici) ve Amazon için daha az (daha agresif algılama).

Yüksek frekanslı izlemeyi destekleyen proxy stratejileri hakkında daha fazla bilgi için rehberimizi keşfedin Web için en iyi referanslar ve ProxyHat'ın fiyatlandırma planları Yüksek hacimli kullanım için.

Key Takeaways

• Gerçek zamanlı izleme, fiyat değişikliği algılamasını saatlerce dakikalar, repricing ve rekabetçi yanıt için kritik azaltır.
• Yüksek değerli ürünler üzerinde kaynaklara odaklanmak için öncelikli bir kuyruk kullanın, hala uzun kuyrukları kapsarken.
• Eş zamanlı proxy bağlantıları olan bir işçi havuzu patlama kalıpları olmadan bağlantı sağlar.
• Doğrulama motorları filtre gürültüyü değiştirin - sadece gerçek fiyat değişikliklerinde süreç ve uyarı.
• Maliyet yönetimi için saklama politikaları ile zaman serisi veritabanında ham veriler.
• Sürekli izleme için kalıcı devlet rotasyonu olan ev sahipleri önemlidir. Başlangıç ProxyHat Güvenilir erişim için.

Gerçek zamanlı altyapı mı? Devamını Oku e-ticaret kılavuzluk Tam strateji için ve rehberlerimizi kontrol edin Python'daki referansları ve Hayır Uygulama detayları için.