Construindo Infraestrutura de Monitoramento de Precos em Tempo Real

Monitoramento de preços em tempo real vs em lote

A maioria dos sistemas de monitoramento de preços operam em modo batelada: verifique todos os produtos a cada hora (ou a cada poucas horas), guarde os resultados e envie alertas sobre alterações. Isso funciona para muitos casos de uso, mas em mercados em movimento rápido — vendas em flash, preços dinâmicos, concorrência de mercado — monitoramento de lotes perde mudanças críticas de preços que acontecem entre os cheques.

O monitoramento de preços em tempo real reduz a defasagem de detecção de horas para minutos ou até mesmo segundos. Em vez de verificar cada produto em um cronograma fixo, um sistema em tempo real monitora continuamente alvos de alta prioridade e reage às mudanças à medida que acontecem. Este guia cobre a arquitetura, infraestrutura proxy e detalhes de implementação necessários para construir um sistema de monitoramento em tempo real. Para os conceitos de monitoramento de preços fundamentais, consulte nosso guia sobre monitorização automática dos preços dos concorrentes.

Arquitetura para monitoramento em tempo real

Um sistema de monitoramento de preços em tempo real tem cinco componentes centrais que trabalham juntos como um pipeline contínuo.

1. Fila de Prioridade

Os produtos são atribuídos níveis prioritários que determinam a frequência de verificação. Uma fila de prioridades (Redessorted Sets funcionam bem) garante que produtos de alto valor sejam sempre verificados primeiro.

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. Grupo de Trabalhadores

Vários processos de trabalhadores tiram da fila de prioridades, obtêm preços através de proxies e empurram resultados para o pipeline de dados. Os trabalhadores operam de forma independente, cada um com sua própria conexão proxy.

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. Motor de detecção de mudança

Em vez de armazenar cada verificação de preços, o motor de detecção de mudanças compara os preços atuais com os últimos valores conhecidos e apenas desencadeia eventos em mudanças reais.

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. Fluxo de eventos

Mudanças de preços são publicadas em um canal Redis Pub/Sub (ou tópico Kafka para sistemas maiores). Os consumidores a jusante — serviços de alerta, motores de recarga, painéis — subscrevem estes eventos e reagem de forma independente.

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. Painel e Alertas

Dados em tempo real precisam de visualização em tempo real. Use conexões WebSocket para empurrar atualizações de preço para painéis instantaneamente.

Implementação Node.js

Uma versão Node.js do motor de monitoramento em tempo real usando O Nó do ProxyHat 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();

Gerenciamento de Proxy para monitoramento contínuo

O monitoramento em tempo real coloca exigências únicas na sua infraestrutura de proxy em comparação com a raspagem em lote.

Padrão de Pedido de Estado Firme

Ao contrário da raspagem em lote que envia rajadas de pedidos, o monitoramento em tempo real cria um fluxo constante. Isso é realmente melhor para a saúde do proxy — um fluxo constante de 5-10 solicitações por segundo parece mais natural do que 1.000 solicitações em uma explosão de 2 minutos.

Configuração do ProxyHat para Tempo Real

# 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

Monitoramento de Saúde IP

Acompanhe as taxas de sucesso por site alvo e ajuste sua abordagem dinamicamente. Se as taxas de sucesso caírem em um mercado específico, aumente os atrasos ou mude para um pool geo-alvo diferente. A grande piscina residencial da ProxyHat garante que você sempre tenha IPs frescos disponíveis. Verifique o nosso locais do proxy para cobertura total.

Key takeaway: O monitoramento em tempo real requer uma estratégia de proxy estável e sustentável. O objetivo é consistente pedidos de baixo volume em muitos IPs, não estouros de alto volume de poucos IPs.

Armazenamento de dados para dados em tempo real

Dados de preço em tempo real precisam de uma solução de armazenamento otimizada para inserções de alta frequência e consultas de intervalo de tempo.

Esquema de tempoDB

-- 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');

Considerações de Escala

• Escala horizontal do trabalhador: Adicione trabalhadores em várias máquinas, cada uma puxando da mesma fila do Redis. Não é necessária coordenação — a distribuição da fila lida com ela.
• Troteamento prioritário: Quando o orçamento proxy é limitado, reduza automaticamente a frequência de verificação de produtos de baixa prioridade, mantendo a cobertura em tempo real para itens críticos.
• Intervalos adaptativos: Se o preço de um produto estiver estável por 24 horas, aumente o intervalo de verificação. Se mudou duas vezes numa hora, diminua-a.
• Convênio específico do site: Locais de destino diferentes têm tolerâncias diferentes. Execute mais trabalhadores concorrentes para Shopify (mais permissive) e menos para Amazon (mais agressiva detecção).

Para mais informações sobre estratégias proxy que suportam monitoramento de alta frequência, explore nosso guia sobre melhores proxies para raspagem web e Planos de preços da ProxyHat para uso de alto volume.

Tiras de Chaves

• O monitoramento em tempo real reduz a detecção de mudança de preço de horas a minutos, crítica para repricing e resposta competitiva.
• Use uma fila de prioridades para focar recursos em produtos de alto valor enquanto ainda cobre a cauda longa.
• Um pool de trabalhadores com conexões proxy simultâneas fornece rendimento sem padrões de ruptura.
• Alterar o ruído do filtro dos motores de detecção — apenas processar e alertar para as alterações reais dos preços.
• Armazene dados brutos em um banco de dados de séries temporais (TimescaleDB) com políticas de retenção para gerenciamento de custos.
• Proxies residenciais com rotação em estado estacionário são essenciais para o monitoramento contínuo. Iniciar com ProxyHat para um acesso fiável.

Construir infra-estruturas em tempo real? Leia o nosso Guia de raspagem do comércio electrónico para a estratégia completa e verificar nossos guias sobre usando proxies em Python e Node.js Para mais pormenores sobre a aplicação.