Use Cases

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

批量抓取美股行情或 SEC 文件时,最常卡住的不是反爬虫策略,而是 reCAPTCHA v2、Cloudflare Turnstile 这类验证码。

CaptchaAI 通过 API 自动识别并返回 token,抓取脚本照常提交表单即可,不需要人工介入。


金融平台的验证码类型与触发场景

数据来源 验证码类型 触发条件 数据价值
SEC EDGAR reCAPTCHA v2 请求频率过高 上市公司备案文件
Yahoo Finance reCAPTCHA v2 识别到抓取行为 股票行情、历史数据
Bloomberg Cloudflare Turnstile 所有自动化访问 市场数据
Finviz reCAPTCHA v2 股票筛选器访问 筛选结果
TradingView Cloudflare Challenge 触发限流规则 图表、技术指标
Morningstar reCAPTCHA v3 数据导出页面 基金分析数据

金融网站的限流与验证码应对策略

金融网站对自动化访问的容忍度更低,超出下面这些经验值容易触发验证码:

  • 请求间隔:页面间隔 2–5 秒
  • 并发连接数:每个域名最多 3–5 个
  • 网络出口:自有服务器基础设施,避免共用出口 IP
  • 会话时长:单会话 5–10 分钟
  • User-Agent:真实且同一会话内一致
  • SEC EDGAR:User-Agent 必须带联系邮箱
  • 抓取时段:避开美股开盘、收盘高峰

股票筛选器数据抓取实战

辅助函数

遇到 reCAPTCHA v2 时自动识别并重新提交表单,复用 solve_captcha()

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:[email protected]: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 文件批量提取

限流触发的验证码

超限即触发验证码,以下代码复用 solve_captcha() 并按 SEC 要求带联系邮箱:

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:[email protected]:5000",
)
filings = sec.search_filings("Apple Inc", "10-K")

Cloudflare Turnstile 保护的市场数据抓取

Turnstile 流程

拿到 sitekey,识别后把 token 放进 cf-turnstile-response 字段提交即可:

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

定时抓取:每日市场数据快照

定时批量运行

包装成定时任务,采集多只股票行情并存为 CSV:

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:[email protected]: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)

常见问题排查

常见报错:

问题 原因 处理方式
SEC EDGAR 返回 403 User-Agent 缺少联系邮箱 添加 CompanyName email@domain 请求头
每次请求都出验证码 触发了速率限制 请求之间加 3–5 秒延迟
拿到的价格数据是旧的 响应被缓存 加一个缓存穿透用的查询参数
JSON 解析报错 返回的其实是验证码页面 解析前先判断是否命中验证码
IP 被封 同一 IP 请求过多 切换到轮换自有服务器基础设施

常见问题

抓取 SEC EDGAR、Yahoo Finance 这类金融数据是否合规?

公开金融数据(股票报价、SEC 备案文件)通常允许抓取,但要遵守服务条款和速率限制。

国内网络抓取美股数据,为什么 reCAPTCHA 经常加载不出来?

reCAPTCHA 依赖 Google 托管脚本,国内网络访问不一定稳定,可能加载超时或拿不到 sitekey,需与识别失败分开排查。

批量抓取股票行情,并发和间隔怎么设置才不容易触发验证码?

每个域名最多 3–5 个并发连接,页面间隔 2–5 秒。行情数据每分钟抓一次即可,抓得越频繁越容易触发验证码。


相关阅读


免受验证码打断,稳定采集财务数据——获取你的 CaptchaAI API Key,让市场数据抓取自动运行。

该文章已禁用评论。