数据并行抓取方案

问题背景

量化系统中存在两个独立的数据拉取链路，均使用纯串行的 HTTP API 调用，成为系统运行时的主要瓶颈：

每次 API 调用耗时 0.5~3s（取决于数据源响应），25 个代码串行约 20~40 秒纯网络等待。

方案设计

使用 concurrent.futures.ThreadPoolExecutor 将独立的 HTTP API 请求并行化。Python 的 GIL 在 I/O 等待时释放，多线程对 HTTP 请求效果极佳。

并发数选择

实测探测 tushare 代理 + adata 库：

收益在 6 workers 附近趋于饱和，平台 API 无明显限流。最终默认 max_workers=6。

---

实现一：截面选基 DataFetcher 并行化

文件：engine.py

fetch_batch 改为当 max_workers > 1 时使用 ThreadPoolExecutor，保留 max_workers=1 的串行回退路径。

class DataFetcher:
    def __init__(self, max_workers: int = 6):
        ...

    def fetch_batch(self, codes, start_date, end_date, progress_callback=None):
        total = len(codes)
        if self.max_workers <= 1:
            return self._fetch_batch_serial(...)

        results = {}
        completed = [0]
        lock = threading.Lock()

        def _fetch_one(code):
            df = self.fetch_single(code, start_date, end_date)
            with lock:
                completed[0] += 1
                if progress_callback:
                    progress_callback(completed[0], total, code, ...)
            return code, df

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {executor.submit(_fetch_one, code): code for code in codes}
            for future in as_completed(futures):
                code, df = future.result()
                results[code] = df
        return results

实测：45 只 ETF 全量拉取，从 ~18s 降至 3.5s（5x 加速）。

---

实现二：择时策略 fetch_data_batch

文件：df_loader.py

新增 fetch_data_batch() 函数，与 fetch_data() 参数完全兼容，内部并行分发：

def fetch_data_batch(codes, start_date='2010-01-01', end_date='2030-01-01',
                     k_type=1, future_days=1, us_api="tushare",
                     max_workers=6, progress_callback=None):
    ...
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(_fetch_one, code): code for code in codes}

调用者改为：show_marker.py

时性检查从：

# 旧：串行
for code in codes_to_process:
    df_check = fetch_data(code=clean_code, ...)

变为：

# 新：并行批量
check_results = fetch_data_batch(clean_codes, start_date=check_start, ...)
for clean_code, df_check in check_results.items():
    ...

实测：

3 个代码:  串行 2.3s  →  并行 1.0s   (2.2x)
10 个代码: 串行 ~8s   →  并行 ~1.5s  (5x)

---

使用方法

截面选基引擎

from adata.aohan.quant.selection.engine import ETFSelectionEngine, DataFetcher

# 默认 6 线程并行
engine = ETFSelectionEngine()
engine.fetch_all_data(years_back=2)

# 自定义线程数
engine = ETFSelectionEngine(data_fetcher=DataFetcher(max_workers=4))

# 回退串行
engine = ETFSelectionEngine(data_fetcher=DataFetcher(max_workers=1))

择时策略 / 通用场景

from adata.aohan.quant.utils.df_loader import fetch_data_batch

results = fetch_data_batch(
    ["399102.SZ", "000015.SH", "510880", "513050"],
    start_date="2026-04-01",
    end_date="2026-05-24",
    max_workers=6
)
# results: {"399102.SZ": DataFrame, "000015.SH": DataFrame, ...}

自行测试加速比

cd /Users/aohan/PycharmProjects/adata
PYTHONPATH=$(pwd) .venv/bin/python

import time
from adata.aohan.quant.utils.df_loader import fetch_data, fetch_data_batch

CODES = ["399102.SZ", "000015.SH", "510880", "513050", "513180", "000933.SH"]

# 串行
t0 = time.time()
for c in CODES:
    fetch_data(code=c, start_date="2026-04-01", end_date="2026-05-24")
print(f"Serial: {time.time()-t0:.1f}s")

# 并行
t0 = time.time()
fetch_data_batch(CODES, start_date="2026-04-01", end_date="2026-05-24", max_workers=6)
print(f"Parallel: {time.time()-t0:.1f}s")

---

测试覆盖

test_data_fetcher.py 5 个测试：

---

关联优化：推送耗时打点

文件：show_marker.py

每轮执行后终端和推送都会输出各阶段耗时：

⏱ 耗时 | 截面选基 拉取3.5s+评分0.7s | 时效检查 1.8s | 择时 拉取4.1s+计算8.4s(25品种) | 总计 18.5s

六段分界：

耗时写入 TXT 推送文件和终端 stdout。通过对比「拉取」和「计算」各自的占比，可以精确定位瓶颈在 I/O 还是 CPU。

---

关联优化：Tushare 数据更新时机探测

文件：tushare_update_probe.py

不确定 tushare.citydata.club 数据每天几点更新时，用此脚本探测。

PYTHONPATH=$(pwd) .venv/bin/python adata/aohan/quant/utils/tushare_update_probe.py

脚本会在后台每 15 分钟拉一次 510300.SH 的当天数据，检测到 trade_date 更新到当日时打印首次检测时间并停止。日志写入 adata/aohan/quant/data/tushare_update_log.txt。

运行一天后即可从日志中确认更新时间窗口，后续把每日推送的 crontab 设在该时间点之后即可最早拿到当日数据。