Обработка CAPTCHA для сбора данных фондового рынка

Проверка появляется не там, где её ждут: не на первой странице портала, а на двадцатом тикере подряд, на выгрузке истории за год и на сложном запросе к скринеру. Финансовые порталы закрывают котировки, отчётность и рейтинги аналитиков через Cloudflare Turnstile и reCAPTCHA, поэтому сборщик данных должен уметь три вещи: распознать страницу проверки, отправить её в решатель и повторить исходный запрос с готовым токеном. Ниже — рабочая схема: где именно срабатывает проверка, код на Python и JavaScript, разумные интервалы опроса и разбор ошибок, которые чаще всего ломают ночной сбор.

Где именно срабатывает проверка CAPTCHA на финансовых порталах

Полезно заранее знать, какой тип проверки ждёт на каждом маршруте: от этого зависит и значение параметра method, и то, в какое поле возвращать токен.

CaptchaAI решает все перечисленные типы: reCAPTCHA v2 и v3 (включая Enterprise-варианты), Cloudflare Turnstile и Cloudflare Challenge, GeeTest v3, image/OCR и grid-капчи. Типы hCaptcha и FunCaptcha сервис не поддерживает, GeeTest v4 — в разработке; если портал закрыт одним из них, сразу планируйте другой источник данных.

Сценарий: ночная выгрузка портфеля из 300 тикеров

Типичная задача аналитической команды в Алматы или Минске выглядит так: раз в сутки после закрытия рынка нужно снять цены закрытия по 300 бумагам и обновить витрину в ClickHouse. Проверка появляется примерно на каждом десятом запросе, и последовательный сборщик с паузой в 2 секунды укладывается в окно. При трёх параллельных потоках проверка срабатывает чаще, и решения начинают выстраиваться в очередь.

Здесь важна тарификация по потокам: CaptchaAI считает не решения, а одновременные задачи, и внутри тарифа число решений на поток не ограничено. Тариф BASIC ($15/мес, 5 потоков) закрывает такую ночную выгрузку с запасом, потому что решения расходуются короткими всплесками. Если же вы снимаете внутридневные котировки по нескольким сотням бумаг в течение всей сессии, имеет смысл смотреть на STANDARD ($30/мес, 15 потоков) или ADVANCE ($90/мес, 50 потоков) — предсказуемая месячная сумма в долларах удобнее для планирования бюджета, чем оплата за каждое решение.

Сборщик котировок на Python

Класс ниже делает три вещи: ходит за котировкой или историей в рамках одной сессии, распознаёт страницу проверки по коду 403 и маркерам Cloudflare, а затем отправляет sitekey в CaptchaAI и повторяет запрос с полученным токеном. Обратите внимание на два разных поля ответа: cf-turnstile-response для Turnstile и g-recaptcha-response для reCAPTCHA — перепутать их нельзя, форма просто не примет отправку.

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"]
)

Ключевая деталь — общий requests.Session(). Куки, выданные после успешного прохождения проверки, живут в сессии и позволяют сделать следующие запросы уже без решения. Если создавать новую сессию на каждый тикер, проверка будет срабатывать буквально на каждом шаге, а число задач вырастет на порядок.

Скринер рынка на JavaScript

Тот же цикл в Node-варианте: запрос к скринеру, проверка HTML на маркер cf-turnstile, отправка задачи, опрос res.php и повторная отправка фильтров вместе с токеном.

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

Опрос здесь ограничен 60 итерациями по 3 с — это осознанный верхний предел, а не ожидаемое время решения. Turnstile обычно решается заметно быстрее, а длинный лимит нужен для редких пиков нагрузки; при исчерпании лимита корректнее отдать ошибку наверх и повторить попытку позже, чем висеть в ожидании бесконечно.

Интервалы сбора: сколько данных нужно на самом деле

Половина проблем со сбором биржевых данных решается не кодом, а расписанием. Чем реже вы обращаетесь к порталу, тем реже видите проверку и тем меньше потоков вам нужно.

Отдельно про сами данные: собирайте только ту информацию, которую вы вправе обрабатывать, и сверяйтесь с условиями использования портала. Если в выгрузке появляются персональные данные — например, имена аналитиков или клиентские идентификаторы, — к ним применяются требования 152-ФЗ «О персональных данных» и аналогичных норм в вашей юрисдикции. Это зона ответственности вашей команды, а не сервиса решения CAPTCHA.

Разбор типичных сбоев

Часто задаваемые вопросы

Сколько потоков нужно для сбора по 300 тикерам?

Считайте не по числу тикеров, а по числу одновременных решений. При последовательном обходе с паузой достаточно одного-двух потоков, и тариф BASIC ($15/мес, 5 потоков) даёт запас. Параллельный сбор сразу по нескольким порталам — повод смотреть на STANDARD ($30/мес, 15 потоков).

Почему проверка возвращается сразу после успешного решения?

Чаще всего токен приняли, но куки не сохранились: следующий запрос уходит из новой сессии и выглядит для портала как первый визит. Держите одну сессию на портал и проверьте, что заголовки и User-Agent не меняются между запросами.

Что делать, если портал закрыт hCaptcha?

CaptchaAI не решает hCaptcha и FunCaptcha. В этом случае ищите альтернативный источник тех же данных — официальный API портала, агрегатора или регуляторную выгрузку — либо согласуйте доступ напрямую с площадкой.

Можно ли собирать данные с нескольких порталов одновременно?

Да. Заведите отдельную сессию на каждый портал и отправляйте задачи в CaptchaAI независимо — параллельные задачи ограничены только числом потоков вашего тарифа.

Чем решение CAPTCHA отличается от официального API портала?

Официальный API стабильнее и предсказуемее, поэтому начинать стоит с него. Но бесплатные тарифы обычно ограничены по частоте и покрывают только базовые поля: фильтры скринера, комментарии аналитиков и нишевые показатели остаются доступны только на самом сайте — там и нужна обработка проверки.

Следующие шаги

• Быстрый старт CaptchaAI: первый решённый токен за пять минут
• Решение reCAPTCHA v2 через API: пошаговый разбор
• Решение Cloudflare Turnstile через API
• Решение GeeTest v3 через API