Python股票数据爬虫教程：抓取股价、校验行情与配置SOCKS5代理

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install "requests[socks]>=2.32,<3" "beautifulsoup4>=4.13,<5"

<!doctype html>
<html lang="en">
<head><meta charset="utf-8"><title>Fictional stock quote</title></head>
<body>
  <h1>DEMO Corporation — synthetic test instrument</h1>
  <p>Not a real security or a market quote.</p>
  <span class="quote-price" data-value="123.45">123.45</span>
  <span class="quote-volume" data-value="1234500">1,234,500</span>
  <time class="quote-time" datetime="2026-10-15T16:00:00-04:00">
    Synthetic timestamp
  </time>
</body>
</html>

import argparse
import csv
import ipaddress
import os
import re
import sys
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path
from urllib.parse import quote, urlsplit

import requests
from bs4 import BeautifulSoup


def build_session(use_proxy=False):
    session = requests.Session()
    session.trust_env = False
    session.headers['User-Agent'] = 'QuoteResearchTutorial/1.0'
    if use_proxy:
        host = os.environ['SOCKS5IO_HOST']
        port = int(os.environ['SOCKS5IO_PORT'])
        if not re.fullmatch(r'[A-Za-z0-9.-]+', host) or not 1 <= port <= 65535:
            raise ValueError('Invalid proxy host or port')
        user = quote(os.environ['SOCKS5IO_USER'], safe='')
        password = quote(os.environ['SOCKS5IO_PASSWORD'], safe='')
        proxy = f'socks5h://{user}:{password}@{host}:{port}'
        session.proxies.update({'http': proxy, 'https': proxy})
    return session


def is_local_url(url):
    host = (urlsplit(url).hostname or '').lower().rstrip('.')
    if host == 'localhost' or host.endswith('.localhost'):
        return True
    try:
        address = ipaddress.ip_address(host)
    except ValueError:
        return False
    mapped = getattr(address, 'ipv4_mapped', None)
    return address.is_loopback or address.is_unspecified or bool(mapped and mapped.is_loopback)


def fetch_html(session, url):
    parsed = urlsplit(url)
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ValueError('Use a public quote URL without credentials, query or fragment')
    local = is_local_url(url)
    if local and any(session.proxies.values()):
        raise ValueError('Local sources cannot use a remote proxy')
    if parsed.scheme != 'https' and not (local and parsed.scheme == 'http'):
        raise ValueError('Use HTTPS, or HTTP for the local tutorial server only')
    with session.get(url, timeout=(5, 20), allow_redirects=False) as response:
        if response.status_code == 429:
            raise ValueError('HTTP 429: stop; review Retry-After and the source quota')
        if response.status_code in {401, 403}:
            raise ValueError(f'HTTP {response.status_code}: stop; check access rights')
        if response.status_code != 200:
            raise ValueError(f'HTTP {response.status_code}: inspect before continuing')
        if 'text/html' not in response.headers.get('Content-Type', '').lower():
            raise ValueError('Expected HTML, not an API response or file download')
        return response.content


def field(soup, selector):
    matches = soup.select(selector)
    if len(matches) != 1:
        raise ValueError('Each selector must match exactly one node')
    node = matches[0]
    for attr in ('data-value', 'content', 'datetime'):
        if node.get(attr) is not None:
            return node[attr].strip()
    return node.get_text(' ', strip=True)


def number(raw, integer=False):
    pattern = r'(?:\d+|\d{1,3}(?:,\d{3})+)(?:\.\d+)?'
    if not re.fullmatch(pattern, raw):
        raise ValueError('Unsupported number format or missing value')
    value = Decimal(raw.replace(',', ''))
    if integer and value != value.to_integral_value():
        raise ValueError('Volume must be a whole number of shares')
    if not integer and value <= 0:
        raise ValueError('Price must be positive')
    return str(int(value)) if integer else format(value, 'f')


def parse_quote(html, price_selector, volume_selector, time_selector):
    soup = BeautifulSoup(html, 'html.parser')
    raw_time = field(soup, time_selector)
    source_time = datetime.fromisoformat(raw_time.replace('Z', '+00:00'))
    if source_time.tzinfo is None:
        raise ValueError('Quote timestamp needs a verified timezone offset')
    return {'price': number(field(soup, price_selector)),
            'volume_shares': number(field(soup, volume_selector), integer=True),
            'quote_time_raw': raw_time,
            'quote_time_utc': source_time.astimezone(timezone.utc).isoformat()}


def main():
    parser = argparse.ArgumentParser()
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument('--fixture', type=Path)
    source.add_argument('--url')
    parser.add_argument('--approved', action='store_true')
    parser.add_argument('--proxy', action='store_true')
    parser.add_argument('--demo-http', action='store_true')
    for name in ('symbol', 'exchange', 'currency', 'market-session', 'delay-status'):
        parser.add_argument('--' + name, required=True)
    parser.add_argument('--price-selector', default='.quote-price')
    parser.add_argument('--volume-selector', default='.quote-volume')
    parser.add_argument('--time-selector', default='.quote-time')
    parser.add_argument('--out', type=Path, required=True)
    args = parser.parse_args()
    if args.fixture:
        if args.proxy or args.demo_http:
            parser.error('Fixture mode does not use a proxy or HTTP')
        html, source_url, kind = args.fixture.read_bytes(), 'local:quote-demo.html', 'synthetic_fixture'
    else:
        if not args.approved:
            parser.error('Review source permissions, terms and robots rules before --approved')
        local_source = is_local_url(args.url)
        if args.demo_http and not local_source:
            parser.error('--demo-http is reserved for the local tutorial server')
        if local_source and args.proxy:
            parser.error('Local sources cannot use a remote proxy')
        with build_session(args.proxy) as session:
            html = fetch_html(session, args.url)
        source_url = args.url
        kind = 'synthetic_http_fixture' if local_source else 'user_approved_html'
    row = {'symbol': args.symbol, 'exchange': args.exchange, 'currency': args.currency,
           **parse_quote(html, args.price_selector, args.volume_selector, args.time_selector),
           'market_session': args.market_session, 'delay_status': args.delay_status,
           'retrieved_at_utc': datetime.now(timezone.utc).isoformat(),
           'source_url': source_url, 'source_kind': kind}
    with args.out.open('x', encoding='utf-8', newline='') as handle:
        writer = csv.DictWriter(handle, fieldnames=list(row))
        writer.writeheader()
        writer.writerow(row)
    print('Saved one validated record. Source kind:', kind)


if __name__ == '__main__':
    try:
        main()
    except ValueError as error:
        print('Stopped:', error, file=sys.stderr)
        sys.exit(1)
    except (KeyError, OSError, requests.RequestException):
        print('Stopped: check HTTP/access status, selectors, fields, environment or output filename.', file=sys.stderr)
        sys.exit(1)

python quote_scraper.py --fixture quote-demo.html --symbol DEMO --exchange TEST --currency USD --market-session regular --delay-status synthetic --out demo-quotes.csv

symbol: DEMO
exchange: TEST
currency: USD
price: 123.45
volume_shares: 1234500
quote_time_utc: 2026-10-15T20:00:00+00:00
delay_status: synthetic
source_kind: synthetic_fixture

python quote_scraper.py --fixture quote-demo.html --symbol DEMO --exchange TEST --currency USD --market-session regular --delay-status synthetic --price-selector ".missing-price" --out must-not-exist.csv

python -m http.server 8765 --bind 127.0.0.1

python quote_scraper.py --url http://127.0.0.1:8765/quote-demo.html --approved --demo-http --symbol DEMO --exchange TEST --currency USD --market-session regular --delay-status synthetic --out demo-http-quotes.csv

python quote_scraper.py --url "https://quotes.example.invalid/quote/REPLACE_SYMBOL" --approved --symbol "REPLACE_SYMBOL" --exchange "REPLACE_EXCHANGE" --currency "REPLACE_CURRENCY" --market-session "REPLACE_SESSION" --delay-status "REPLACE_DELAY" --price-selector "REPLACE_PRICE_CSS" --volume-selector "REPLACE_VOLUME_CSS" --time-selector "REPLACE_TIME_CSS" --out authorized-direct.csv

$env:SOCKS5IO_HOST = Read-Host 'Issued proxy host'
$env:SOCKS5IO_PORT = Read-Host 'Issued proxy port'
$env:SOCKS5IO_USER = Read-Host 'Issued proxy username'
$proxySecret = Read-Host 'Issued proxy password' -AsSecureString
$env:SOCKS5IO_PASSWORD = [System.Net.NetworkCredential]::new('', $proxySecret).Password

python quote_scraper.py --url "https://quotes.example.invalid/quote/REPLACE_SYMBOL" --approved --proxy --symbol "REPLACE_SYMBOL" --exchange "REPLACE_EXCHANGE" --currency "REPLACE_CURRENCY" --market-session "REPLACE_SESSION" --delay-status "REPLACE_DELAY" --price-selector "REPLACE_PRICE_CSS" --volume-selector "REPLACE_VOLUME_CSS" --time-selector "REPLACE_TIME_CSS" --out authorized-proxy.csv