批量抓取美股行情或 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,让市场数据抓取自动运行。