为什么股票数据采集必须用隧道代理，而不是短效代理？

股票数据的采集特性决定了代理类型的选择。三个关键差异：

• 时效性要求严格。开盘时段（9:30 到 15:00）的行情数据延迟不能超过秒级，短效代理频繁切换 IP 会导致连接重建，秒级延迟直接被拉高
• 采集频率高且稳定。分钟级或秒级轮询采集，短效代理的 IP 存活周期（几分钟到几十分钟）与轮询周期严重不匹配
• 失败重试成本高。行情数据具有时间戳属性，失败重试拿到的已是过期数据，不能补采

隧道代理的核心机制是"应用层保持稳定的连接入口，底层 IP 池由服务商托管切换"。对采集方而言，只需连接一个固定的隧道地址，IP 切换与故障转移由代理服务方后端完成。这正是股票、舆情监测、广告监测这类高频稳定采集场景的最佳匹配。

对比参考：舆情监测在突发事件时段并发翻倍，短效代理的 IP 切换开销会让延迟指数级放大；广告监测 24 小时不间断运行，短效代理频繁重新拨号会消耗大量运维精力。这两类场景的解决方案都指向同一个选型结论：隧道代理是"稳定 + 高频"业务的默认起手式。

隧道代理与其他代理类型的差异是什么？

选择隧道代理的核心理由是"稳定性 + 免运维"。应用层代码不需要处理 IP 池管理、不需要处理 IP 失效切换，把运维成本从应用侧转移到代理服务方。

完整方案的技术栈包含哪些组件？

自动化股票数据采集不是一个"跑通脚本"就完事的项目。生产环境要长期稳定运行，必须把采集、异常、存储、调度、监控做成完整闭环。一套可落地的股票数据自动化采集系统需要六个组件：

六个组件对应六个可能出错的环节，任一环节没设计好都会导致数据链路中断。

环境准备：如何配置基础运行环境？

Python 3.9 或以上版本，通过 pip 安装依赖库：

pip install requests tenacity apscheduler sqlalchemy beautifulsoup4 lxml

创建配置文件 config.py 存放隧道代理接入信息，避免硬编码：

# config.py
TUNNEL_HOST = "your-tunnel-host.example.com"
TUNNEL_PORT = 12345
TUNNEL_USERNAME = "your_username"
TUNNEL_PASSWORD = "your_password"

# 目标数据源
STOCK_DATA_URL = "https://target-finance-site.com/api/quote"

# 采集参数
REQUEST_TIMEOUT = 10
MAX_RETRIES = 3

隧道代理的接入格式是标准的 HTTP 代理协议，形如 http://username:password@host:port。这一步是所有后续代码的基础，配置错误会导致整个链路无法启动。

核心代码：如何用 Python 通过隧道代理采集股票行情？

核心采集函数封装隧道代理接入与 HTTP 请求：

# fetcher.py
import requests
from config import (
    TUNNEL_HOST, TUNNEL_PORT,
    TUNNEL_USERNAME, TUNNEL_PASSWORD,
    REQUEST_TIMEOUT
)

def build_proxies():
    """构造隧道代理接入字符串"""
    proxy_url = (
        f"http://{TUNNEL_USERNAME}:{TUNNEL_PASSWORD}"
        f"@{TUNNEL_HOST}:{TUNNEL_PORT}"
    )
    return {"http": proxy_url, "https": proxy_url}

def fetch_stock_quote(symbol: str) -> dict:
    """采集指定股票代码的实时行情"""
    url = f"https://target-finance-site.com/api/quote/{symbol}"
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; DataCollector/1.0)",
        "Accept": "application/json"
    }
    proxies = build_proxies()
    
    response = requests.get(
        url, headers=headers, proxies=proxies,
        timeout=REQUEST_TIMEOUT
    )
    response.raise_for_status()
    return response.json()

三个关键参数：proxies 参数把隧道代理接入 requests 库；timeout 参数设置合理超时，避免网络阻塞拖垮整个任务；headers 参数设置 User-Agent，减少被目标网站按"未标识客户端"限制的概率。

生产环境建议用 requests.Session 复用连接，减少每次采集的 TCP 握手与 TLS 协商开销。Session 对象在同一个 Python 进程内长期存活，隧道代理的连接池由 urllib3 自动管理，实测在秒级采集频率下性能提升可达 30% 到 50%。

异常处理：如何应对采集过程中的各种失败？

网络类失败在长期运行的采集任务里是常态，异常处理必须做到"自动重试 + 分类降级"。用 tenacity 库封装重试策略：

# fetcher_with_retry.py
from tenacity import (
    retry, stop_after_attempt,
    wait_exponential, retry_if_exception_type
)
import requests

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((
        requests.exceptions.Timeout,
        requests.exceptions.ConnectionError
    ))
)
def fetch_with_retry(symbol: str) -> dict:
    return fetch_stock_quote(symbol)

常见异常类型与处理策略：

断路器模式的作用是防止在目标站点出现全局问题时，采集任务持续发起无效请求。触发条件是短时间内连续失败超过阈值，触发后暂停采集一段时间再恢复。

一个可用的经验参数是"5 分钟内失败率超过 50% 触发断路器，暂停 5 分钟后进入半开状态尝试单次探测请求，探测成功恢复正常采集，探测失败继续暂停 5 分钟"。这套参数在多数股票数据采集场景下都是可用的起手值，具体阈值可按目标网站的响应特征做微调。

数据入库：采集到的数据如何持久化存储？

用 SQLAlchemy 定义数据结构，SQLite 作为轻量级存储（单机部署可用，多节点部署建议 PostgreSQL）：

# models.py
from sqlalchemy import (
    create_engine, Column, Integer,
    String, Float, DateTime
)
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class StockQuote(Base):
    __tablename__ = "stock_quotes"
    id = Column(Integer, primary_key=True)
    symbol = Column(String(16), index=True)
    price = Column(Float)
    volume = Column(Integer)
    timestamp = Column(DateTime, default=datetime.utcnow, index=True)

engine = create_engine("sqlite:///stock_data.db")
Base.metadata.create_all(engine)

入库时采用"批量写入 + 增量去重"策略：

# storage.py
from sqlalchemy.orm import sessionmaker
from models import engine, StockQuote

Session = sessionmaker(bind=engine)

def save_quotes(quotes: list) -> int:
    """批量入库,返回成功入库条数"""
    session = Session()
    try:
        objects = [StockQuote(**q) for q in quotes]
        session.bulk_save_objects(objects)
        session.commit()
        return len(objects)
    except Exception as e:
        session.rollback()
        raise
    finally:
        session.close()

批量写入的性能显著优于逐条写入。同批次入库 100 条数据，批量写入比逐条写入快 5 到 10 倍。索引字段选择 symbol 和 timestamp，覆盖常见查询模式。

行情数据的天然属性是时序数据，长期存储建议考虑 TimescaleDB 或 InfluxDB 这类专用时序数据库。相比通用关系数据库，时序库在按时间范围查询、聚合统计、数据自动降采样上都有原生优化，单机可支撑数亿级数据点的查询延迟在毫秒级。

定时调度：如何让采集脚本自动化运行？

三种调度方案的对比：

股票数据采集建议用 APScheduler，支持秒级触发和运行时动态调整：

# scheduler.py
from apscheduler.schedulers.blocking import BlockingScheduler
from fetcher_with_retry import fetch_with_retry
from storage import save_quotes

scheduler = BlockingScheduler()

STOCK_SYMBOLS = ["000001", "600036", "000858"]

@scheduler.scheduled_job("cron",
    day_of_week="mon-fri",
    hour="9-15", minute="*/1"
)
def collect_quotes_job():
    """交易时段每分钟采集一次"""
    quotes = []
    for symbol in STOCK_SYMBOLS:
        try:
            data = fetch_with_retry(symbol)
            quotes.append(data)
        except Exception as e:
            log_error(symbol, e)
    save_quotes(quotes)

if __name__ == "__main__":
    scheduler.start()

配合 logging 和监控告警 webhook，构成完整的自动化闭环：

# monitor.py
import logging
import requests as req

logging.basicConfig(
    filename="collector.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

WEBHOOK_URL = "https://your-alert-webhook.example.com"

def log_error(symbol: str, error: Exception):
    logging.error(f"采集失败 {symbol}: {error}")
    if is_critical(error):
        req.post(WEBHOOK_URL, json={
            "level": "critical",
            "symbol": symbol,
            "error": str(error)
        })

def is_critical(error) -> bool:
    """判断是否为需要立即告警的严重错误"""
    return isinstance(error, (
        req.exceptions.ConnectionError,
        req.exceptions.SSLError
    ))

监控告警的粒度需要分级设计。单次采集失败属于常态，只写日志不触发告警；连续 5 次同一股票代码失败触发普通告警；整个采集任务停摆超过 3 分钟触发严重告警。粒度过细会导致告警疲劳，粒度过粗又会让真正的故障被淹没在噪声里。日常运行时可以每周复盘一次告警日志，动态调整阈值。

FAQ

Q：隧道代理和普通 HTTP 代理接入代码有什么区别？

接入代码几乎没有区别，都是通过 requests 的 proxies 参数传入。区别在于隧道代理的地址是固定的入口，底层 IP 由服务方后端切换；普通 HTTP 代理需要应用层自己维护 IP 池、处理 IP 失效切换，代码复杂度显著更高。

Q：SQLite 能撑住多大规模的股票数据？

单机 SQLite 撑住 1000 万条记录量级没问题，读写并发不高的场景下性能也够用。超过这个量级或需要多节点访问，建议迁移到 PostgreSQL 或 TimescaleDB（时序数据库更适合行情数据的场景）。

Q：APScheduler 和 crontab 应该怎么选？

简单定时任务、频率在分钟级以上、单机部署，用 crontab 就够。需要秒级调度、动态调整任务、或需要与 Python 代码深度集成，用 APScheduler。多任务依赖复杂、需要可视化任务编排，才考虑 Airflow。

Q：怎么监控采集任务是否正常运行？

基础监控三件套：每次采集的成功率日志、每日采集条数汇总报表、异常时的实时告警 webhook。进阶方案可以接入 Prometheus + Grafana 做时序指标监控，观察采集延迟、成功率、代理响应时间等指标的趋势变化。

Q：这套方案能扩展到其他数据采集场景吗？

可以。核心结构（隧道代理接入 + 异常重试 + 批量入库 + 定时调度）适用于所有高频稳定采集场景，例如舆情监测、广告监测、APP大数据分析。差异主要在数据源接入方式和解析逻辑上，其他组件可复用。