Quét dữ liệu tài chính bằng cách xử lý CAPTCHA

Các nền tảng tài chính như công cụ sàng lọc chứng khoán, SEC EDGAR và các nền tảng giao dịch bảo vệ dữ liệu bằng CAPTCHA để ngăn chặn việc trích xuất tự động. CaptchaAI xử lý những thách thức này theo chương trình để bạn có thể thu thập dữ liệu thị trường trên quy mô lớn.

Nơi CAPTCHA xuất hiện trong tài chính

Sàng lọc cổ phiếu

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

Trích xuất hồ sơ EDGAR SEC

SEC EDGAR triển khai giới hạn tốc độ và CAPTCHA để truy cập khối lượng lớn:

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

Dữ liệu thị trường được bảo vệ bằng cửa quay

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

Thu thập dữ liệu theo lịch trình

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)

Các phương pháp hay nhất về giới hạn tỷ lệ

Các trang web tài chính chặt chẽ hơn về quyền truy cập tự động:

Khắc phục sự cố

Câu hỏi thường gặp

Việc thu thập dữ liệu tài chính có hợp pháp không?

Dữ liệu tài chính công khai (hồ sơ SEC, báo giá cổ phiếu) thường được cho phép. Luôn tôn trọng các điều khoản dịch vụ và giới hạn tỷ lệ. SEC EDGAR cung cấp quyền truy cập EDGAR một cách rõ ràng cho mục đích nghiên cứu.

Tại sao các trang web tài chính sử dụng CAPTCHA?

Để ngăn chặn việc trích xuất tự động số lượng lớn có thể cho phép thao túng thị trường, thu thập thông tin cạnh tranh hoặc tải máy chủ quá mức.

Tôi nên thu thập dữ liệu thị trường bao lâu một lần?

Đối với giá cổ phiếu: tối đa một lần mỗi phút trong giờ thị trường. Đối với hồ sơ: một lần mỗi ngày là điển hình. Việc quét quá nhiều sẽ kích hoạt CAPTCHA nhanh hơn.

Hướng dẫn liên quan

• Chất lượng proxy ảnh hưởng đến tỷ lệ giải quyết
• Ủy quyền dân cư luân phiên

Thu thập dữ liệu tài chính mà không bị gián đoạn CAPTCHA —lấy khóa CaptchaAI của bạnvà tự động hóa nghiên cứu thị trường.