使用验证码处理进行财务数据抓取 (实用指南)

股票筛选器、SEC EDGAR 等金融平台和交易平台使用验证码保护数据，以防止自动提取。 CaptchaAI 以编程方式处理这些挑战，以便您可以大规模收集市场数据。

验证码出现在金融领域的地方

来源	验证码类型	扳机	数据价值
SEC埃德加	reCAPTCHA v2	大量请求	公司备案
雅虎财经	reCAPTCHA v2	刮擦检测	股票行情、历史
彭博社	Cloudflare Turnstile	所有自动访问	市场数据
芬维兹	reCAPTCHA v2	股票筛选器访问	筛选结果
交易视图	Cloudflare 验证流程	速率限制	图表、指标
晨星公司	reCAPTCHA v3	数据导出页面	基金分析

股票筛选器刮擦

import requests
import time
from bs4 import BeautifulSoup
import re

CAPTCHAAI_KEY = "YOUR_API_KEY"
CAPTCHAAI_URL = "https://ocr.captchaai.com"


def solve_captcha(method, sitekey, pageurl, **kwargs):
    data = {
        "key": CAPTCHAAI_KEY,
        "method": method,
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1,
    }
    data.update(kwargs)

    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data=data)
    task_id = resp.json()["request"]

    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        r = result.json()
        if r["request"] != "CAPCHA_NOT_READY":
            return r["request"]

    raise TimeoutError("Solve timeout")


class FinancialScraper:
    def __init__(self, proxy=None):
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
            "Accept-Language": "en-US,en;q=0.9",
        })

    def scrape_screener(self, url):
        """Scrape stock screener, handling CAPTCHA if triggered."""
        resp = self.session.get(url, timeout=30)

        # Check for CAPTCHA
        sitekey_match = re.search(r'data-sitekey="([^"]+)"', resp.text)
        if sitekey_match:
            sitekey = sitekey_match.group(1)
            token = solve_captcha("userrecaptcha", sitekey, url)

            # Resubmit with token
            resp = self.session.post(url, data={
                "g-recaptcha-response": token,
            })

        return self._parse_stocks(resp.text)

    def _parse_stocks(self, html):
        soup = BeautifulSoup(html, "html.parser")
        stocks = []
        for row in soup.select("table.screener-table tr")[1:]:
            cols = row.select("td")
            if len(cols) >= 8:
                stocks.append({
                    "ticker": cols[1].get_text(strip=True),
                    "company": cols[2].get_text(strip=True),
                    "sector": cols[3].get_text(strip=True),
                    "price": cols[6].get_text(strip=True),
                    "change": cols[7].get_text(strip=True),
                })
        return stocks


# Usage
scraper = FinancialScraper(
    proxy="http://user:pass@residential.proxy.com:5000"
)
stocks = scraper.scrape_screener("https://screener.example.com/screener.ashx?v=111")
for stock in stocks[:5]:
    print(f"{stock['ticker']}: {stock['price']} ({stock['change']})")

SEC EDGAR 文件提取

SEC EDGAR 对大容量访问实施速率限制和验证码：

import json


class SECFilingScraper:
    BASE_URL = "https://efts.sec.gov/LATEST"

    def __init__(self, user_agent_email, proxy=None):
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        # SEC requires identifying User-Agent
        self.session.headers.update({
            "User-Agent": f"CompanyName admin@{user_agent_email}",
            "Accept": "application/json",
        })

    def search_filings(self, company, filing_type="10-K"):
        """Search EDGAR for specific filing types."""
        url = f"{self.BASE_URL}/search-index"
        params = {
            "q": company,
            "dateRange": "custom",
            "forms": filing_type,
        }

        resp = self.session.get(url, params=params, timeout=30)

        # Handle CAPTCHA if triggered
        if "captcha" in resp.text.lower() or resp.status_code == 403:
            sitekey = self._extract_sitekey(resp.text)
            if sitekey:
                token = solve_captcha("userrecaptcha", sitekey, url)
                resp = self.session.post(url, data={
                    **params,
                    "g-recaptcha-response": token,
                })

        return resp.json() if resp.status_code == 200 else {}

    def download_filing(self, filing_url):
        """Download individual filing document."""
        resp = self.session.get(filing_url, timeout=60)
        if resp.status_code == 200:
            return resp.text
        return None

    def _extract_sitekey(self, html):
        match = re.search(r'data-sitekey="([^"]+)"', html)
        return match.group(1) if match else None


# Usage
sec = SECFilingScraper(
    user_agent_email="example.com",
    proxy="http://user:pass@proxy.example.com:5000",
)
filings = sec.search_filings("Apple Inc", "10-K")

旋转门保护的市场数据

def scrape_turnstile_market_data(url, sitekey):
    """Handle Cloudflare Turnstile on financial data sites."""
    token = solve_captcha("turnstile", sitekey, url)

    session = requests.Session()
    session.headers.update({
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
    })

    resp = session.post(url, data={
        "cf-turnstile-response": token,
    }, timeout=30)

    return resp.json() if resp.status_code == 200 else None

预定的数据收集

import csv
from datetime import datetime


def daily_market_snapshot(tickers, output_dir="data"):
    """Collect daily stock data, handling CAPTCHAs automatically."""
    scraper = FinancialScraper(
        proxy="http://user:pass@residential.proxy.com:5000"
    )

    date_str = datetime.now().strftime("%Y-%m-%d")
    results = []

    for ticker in tickers:
        url = f"https://screener.example.com/quote.ashx?t={ticker}"
        try:
            data = scraper.scrape_screener(url)
            if data:
                results.extend(data)
            time.sleep(2)  # Rate limit
        except Exception as e:
            print(f"Error on {ticker}: {e}")

    # Save to CSV
    filepath = f"{output_dir}/market_{date_str}.csv"
    with open(filepath, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["ticker", "company", "sector", "price", "change"])
        writer.writeheader()
        writer.writerows(results)

    print(f"Saved {len(results)} records to {filepath}")
    return results


# Run daily
tickers = ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"]
daily_market_snapshot(tickers)

速率限制最佳实践

金融网站对自动访问更加严格：

实践	推荐
请求延迟	页面之间间隔 2-5 秒
并发连接数	每个域最多 3-5 个
代理类型	住宅或 ISP
会话时长	5-10 分钟的粘性会话
用户代理	现实、每次会话一致
SEC埃德加	在 UA 中包含联系电子邮件（必填）
市场营业时间	尽可能在非高峰时段进行刮擦

故障排除

问题	原因	处理方式
SEC 埃德加 403	缺少带有电子邮件的用户代理	添加 `CompanyName email@domain` 标头
针对每个请求的验证码	超出速率限制	在请求之间添加 3-5 秒的延迟
陈旧的价格数据	缓存响应	添加缓存清除查询参数
JSON 解析错误	返回验证码页面	解析前检查验证码
IP 被封锁	来自同一 IP 的请求过多	切换到旋转自有服务器基础设施

常问问题

抓取财务数据合法吗？

公共财务数据（SEC 文件、股票报价）通常是允许的。始终遵守服务条款和费率限制。 SEC EDGAR 明确提供 EDGAR 访问权限用于研究目的。

为什么金融网站使用验证码？

防止可能导致市场操纵、竞争情报收集或服务器负载过高的大量自动提取。

我应该多久抓取一次市场数据？

对于股票价格：在市场交易时间内最多每分钟一次。对于归档：通常每天一次。过度抓取会更快地触发验证码。

使用验证码处理进行财务数据抓取

验证码出现在金融领域的地方

股票筛选器刮擦

SEC EDGAR 文件提取

旋转门保护的市场数据

预定的数据收集

速率限制最佳实践

故障排除

常问问题

抓取财务数据合法吗？

为什么金融网站使用验证码？

我应该多久抓取一次市场数据？

相关指南

使用验证码解决进行学术研究网络抓取

Bright Data + CaptchaAI：完整的代理集成指南

注册流程测试中的验证码处理

使用验证码处理自动提交表单

持续集成测试中的验证码处理

弹出模式中的验证码：检测和token 提交

验证码出现在金融领域的地方

股票筛选器刮擦

SEC EDGAR 文件提取

旋转门保护的市场数据

预定的数据收集

速率限制最佳实践

故障排除

常问问题

抓取财务数据合法吗？

为什么金融网站使用验证码？

我应该多久抓取一次市场数据？

相关指南

相关文章

使用验证码解决进行学术研究网络抓取

Bright Data + CaptchaAI：完整的代理集成指南

注册流程测试中的验证码处理

使用验证码处理自动提交表单

持续集成测试中的验证码处理

弹出模式中的验证码：检测和token 提交