Tratamento CAPTCHA para coleta de dados do mercado de ações

Um coletor de dados de mercado quase nunca quebra no parser: ele quebra na sessão. O primeiro GET funciona, o segundo funciona, e na décima consulta de símbolo o portal devolve 403 com um widget do Cloudflare Turnstile no meio do HTML — o job noturno acorda com metade das cotações faltando e ninguém percebe até o relatório da manhã. A saída não é reescrever o coletor: é detectar o desafio, resolvê-lo pela API e repetir a requisição com o token em mãos.

Este guia mostra esse caminho em Python e em JavaScript, com as duas famílias que aparecem de fato em portais financeiros: Cloudflare Turnstile e reCAPTCHA. O cenário é sempre coleta autorizada, dentro dos termos de uso do portal.

Onde os portais financeiros disparam o CAPTCHA

Nem toda página do mesmo portal protege igual, e cada tipo de dado tem um gatilho previsível. Mapeá-lo antes de escrever código economiza horas de tentativa e erro:

Repare que o reCAPTCHA v3 não interrompe a navegação: ele pontua a sessão em silêncio e degrada a resposta em vez de mostrar um desafio. Por isso, monitore também o tamanho da resposta, não apenas o status HTTP.

Quanta pressão cada tipo de dado gera

O maior ganho de estabilidade vem antes do código: ajustar a frequência ao que o dado exige.

Coletar cotação a cada 30 s quando a fonte atualiza de 5 em 5 minutos multiplica os desafios sem entregar dado novo.

Coletor de cotações e histórico em Python

A classe abaixo mantém uma Session viva com os cookies do portal, reconhece a página de desafio pelo status 403 e pelos marcadores do widget, e reenvia a requisição original com o token devolvido pela CaptchaAI. A submissão vai para in.php e a consulta de resultado para res.php; o campo de retorno é cf-turnstile-response no Turnstile e g-recaptcha-response no reCAPTCHA — trocar um pelo outro é o erro mais comum aqui.

import requests
import time
import re
from datetime import datetime, timedelta

class StockDataCollector:
    def __init__(self, api_key):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })

    def get_quote(self, portal_url, symbol):
        """Get current stock quote, solving CAPTCHAs if needed."""
        url = f"{portal_url}/quote/{symbol}"
        response = self.session.get(url)

        if self._is_captcha_page(response):
            response = self._solve_and_retry(response, url)

        return self._parse_quote(response.text, symbol)

    def get_historical(self, portal_url, symbol, days=365):
        """Download historical price data."""
        url = f"{portal_url}/history/{symbol}"
        params = {
            "period": f"{days}d",
            "interval": "1d"
        }
        response = self.session.get(url, params=params)

        if self._is_captcha_page(response):
            response = self._solve_and_retry(response, url)

        return self._parse_historical(response.text)

    def scan_symbols(self, portal_url, symbols, delay=2):
        """Collect quotes for multiple symbols."""
        results = {}

        for symbol in symbols:
            try:
                results[symbol] = self.get_quote(portal_url, symbol)
                time.sleep(delay)
            except Exception as e:
                results[symbol] = {"error": str(e)}

        return results

    def _is_captcha_page(self, response):
        return (
            response.status_code == 403 or
            "cf-turnstile" in response.text or
            "challenges.cloudflare.com" in response.text
        )

    def _solve_and_retry(self, response, url):
        match = re.search(r'data-sitekey="(0x[^"]+)"', response.text)
        if not match:
            # Fall back to reCAPTCHA detection
            match = re.search(r'data-sitekey="([^"]+)"', response.text)
            if match:
                return self._solve_recaptcha_and_retry(match.group(1), url)
            raise ValueError("No CAPTCHA sitekey found")

        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "turnstile",
            "sitekey": match.group(1),
            "pageurl": url,
            "json": 1
        })
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(3)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = result.json()
            if data["status"] == 1:
                return self.session.post(url, data={
                    "cf-turnstile-response": data["request"]
                })

        raise TimeoutError("CAPTCHA solve timed out")

    def _solve_recaptcha_and_retry(self, site_key, url):
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
            "json": 1
        })
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(3)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = result.json()
            if data["status"] == 1:
                return self.session.post(url, data={
                    "g-recaptcha-response": data["request"]
                })

        raise TimeoutError("reCAPTCHA solve timed out")

    def _parse_quote(self, html, symbol):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")

        def text_or_none(node):
            return node.text.strip() if node and node.text else None

        return {
            "symbol": symbol,
            "price": text_or_none(soup.select_one("[data-field='regularMarketPrice'], .price")),
            "change": text_or_none(soup.select_one("[data-field='regularMarketChange'], .change")),
            "volume": text_or_none(soup.select_one("[data-field='regularMarketVolume'], .volume")),
            "market_cap": text_or_none(soup.select_one("[data-field='marketCap'], .market-cap")),
            "timestamp": datetime.now().isoformat()
        }

    def _parse_historical(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        rows = []

        for row in soup.select("table tr")[1:]:  # Skip header
            cells = [td.text.strip() for td in row.select("td")]
            if len(cells) >= 6:
                rows.append({
                    "date": cells[0],
                    "open": cells[1],
                    "high": cells[2],
                    "low": cells[3],
                    "close": cells[4],
                    "volume": cells[5]
                })

        return rows


# Usage
collector = StockDataCollector("YOUR_API_KEY")

# Single quote
quote = collector.get_quote("https://finance.example.com", "AAPL")
print(f"AAPL: ${quote['price']} ({quote['change']})")

# Scan multiple symbols
portfolio = collector.scan_symbols(
    "https://finance.example.com",
    ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"]
)

Screener de ações em JavaScript, com o Turnstile resolvido no caminho

Quando o coletor já vive em Node.js, a lógica é a mesma com outra sintaxe: detectar cf-turnstile no HTML, extrair a sitekey, enviar a tarefa e repetir o POST do screener com o token anexado aos filtros. Mantenha o polling com intervalo fixo de alguns segundos — consultar mais rápido não acelera a resolução.

class MarketScreener {
  constructor(apiKey) {
    this.apiKey = apiKey;
  }

  async screenStocks(portalUrl, filters) {
    const params = new URLSearchParams(filters);
    const response = await fetch(`${portalUrl}/screener?${params}`);
    const html = await response.text();

    if (html.includes('cf-turnstile') || response.status === 403) {
      return this.solveAndScreen(portalUrl, filters, html);
    }

    return this.parseScreenerResults(html);
  }

  async solveAndScreen(portalUrl, filters, html) {
    const match = html.match(/data-sitekey="(0x[^"]+)"/);
    if (!match) throw new Error('Turnstile sitekey not found');

    const submitResp = await fetch('https://ocr.captchaai.com/in.php', {
      method: 'POST',
      body: new URLSearchParams({
        key: this.apiKey,
        method: 'turnstile',
        sitekey: match[1],
        pageurl: portalUrl,
        json: '1'
      })
    });
    const { request: taskId } = await submitResp.json();

    for (let i = 0; i < 60; i++) {
      await new Promise(r => setTimeout(r, 3000));
      const result = await fetch(
        `https://ocr.captchaai.com/res.php?key=${this.apiKey}&action=get&id=${taskId}&json=1`
      );
      const data = await result.json();
      if (data.status === 1) {
        const response = await fetch(`${portalUrl}/screener`, {
          method: 'POST',
          body: new URLSearchParams({
            ...filters,
            'cf-turnstile-response': data.request
          })
        });
        return this.parseScreenerResults(await response.text());
      }
    }
    throw new Error('Turnstile solve timed out');
  }

  parseScreenerResults(html) {
    const rows = [];
    const tableMatch = html.match(/<table[^>]*>[\s\S]*?<\/table>/i);
    if (!tableMatch) return rows;

    const rowMatches = tableMatch[0].matchAll(/<tr[^>]*>([\s\S]*?)<\/tr>/gi);
    for (const row of rowMatches) {
      const cells = [...row[1].matchAll(/<td[^>]*>([\s\S]*?)<\/td>/gi)]
        .map(m => m[1].replace(/<[^>]+>/g, '').trim());
      if (cells.length >= 4) {
        rows.push({
          symbol: cells[0],
          price: cells[1],
          change: cells[2],
          volume: cells[3]
        });
      }
    }
    return rows;
  }
}

// Usage
const screener = new MarketScreener('YOUR_API_KEY');
const results = await screener.screenStocks('https://finance.example.com', {
  sector: 'technology',
  marketCap: 'large',
  peRatio: '<25'
});

Janela de mercado, latência e LGPD: o recorte brasileiro

Uma equipe em São Paulo que acompanha carteiras internacionais convive com duas janelas: o pregão local, que abre às 10h, e a abertura de Nova York, que cai no fim da manhã no horário de Brasília. A partir daí os dois mercados ficam abertos ao mesmo tempo, e é aí que os desafios se concentram. O padrão que funciona: resolva uma vez antes da abertura, mantenha a sessão viva com requisições espaçadas e deixe o trabalho pesado — histórico, demonstrações, reprocessamento — para a madrugada, quando o portal está ocioso.

Latência também pesa: um worker em sa-east-1 responde rápido a portais brasileiros, mas paga RTT alto contra portais hospedados nos Estados Unidos. Se o pipeline mistura as duas fontes, separe os workers por região em vez de aceitar o pior caso para todos.

Na conformidade, cotação é dado público, mas o pipeline em volta dela raramente é só isso. Se você guarda logs com identificadores de usuário ou cruza os dados com carteiras de clientes, considere as obrigações da LGPD sobre finalidade e retenção (em Portugal, o RGPD).

Quando o coletor falha: causas e correções

Perguntas frequentes

Quantas threads eu preciso para acompanhar uma carteira grande?

Depende de quantos desafios acontecem ao mesmo tempo, não de quantos símbolos você monitora. Como a cobrança da CaptchaAI é por thread simultânea, com resoluções ilimitadas dentro do plano, um coletor diário de algumas centenas de símbolos costuma caber no BASIC (US$ 15/mês, 5 threads). Pipelines com vários portais em paralelo e reprocessamento noturno pedem algo como o ADVANCE (US$ 90/mês, 50 threads).

E se o portal exibir hCaptcha ou FunCaptcha?

Esses dois tipos não são suportados hoje, assim como o GeeTest v4 (anunciado apenas como "em breve"). O que está coberto aqui é justamente o que aparece nesses portais: Cloudflare Turnstile, reCAPTCHA v2 e v3, GeeTest v3 e CAPTCHAs de imagem/OCR.

Preciso de um navegador headless para essa coleta?

Na maioria dos casos, não. Os exemplos deste artigo usam apenas cliente HTTP: você lê a sitekey do HTML, envia a tarefa e reenvia a requisição com o token. Um navegador headless só se justifica quando a página monta os dados via JavaScript e não há resposta útil no HTML inicial.

Por que o token funcionou no teste e falhou em produção?

Quase sempre porque a pageurl enviada não corresponde à URL onde o widget foi renderizado, ou porque o token demorou demais para ser usado. Envie a URL exata da página protegida e consuma o token logo após recebê-lo. Se o 403 persistir, o diagnóstico de 403 no Turnstile indicado no fim deste artigo cobre as causas restantes.

Artigos relacionados

• Como coletar dados para pesquisa de mercado
• GeeTest v3 e Cloudflare Turnstile lado a lado
• Turnstile devolvendo 403 mesmo com token válido

Próximas etapas

Comece pelo trecho mais frágil: a cotação que falha em lote. Obtenha sua chave de API da CaptchaAI, plugue a detecção de desafio no coletor que já existe e rode a primeira coleta completa hoje.