批量拉取历史数据是量化回测的第一道工序。这道工序的完成质量，直接决定了后续所有策略验证的可信度——一个因限频、超时或数据缺失而产生偏差的历史数据集，会让回测结果与实盘表现产生系统性背离。

本文拆解历史数据批量拉取的完整工程方案：从单次请求的限频自适应处理，到分片并发拉取与断点续传，再到本地存储与完整性校验。你可以直接将本文代码用于美股、港股、A股及数字货币的分钟级历史数据获取。

在动手写代码之前，先用一张表把核心问题梳理清楚。

一句话总结：批量拉取历史数据不是简单的API循环调用，而是一个需要处理限频、重试、分页、断点续传和流式存储的数据迁移工程。

在写代码之前，先把整体设计画清楚。

┌─────────────────────────────────────────────────────────────────┐ │ 控制层（主程序） │ │ - 读取待拉取symbol列表 │ │ - 根据本地进度文件跳过已完成symbol │ │ - 为每个symbol生成时间切片任务 │ └─────────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ 任务队列 & 并发控制层 │ │ - asyncio 任务队列，控制并发数（默认3-5） │ │ - 每个任务：拉取一个(symbol, 时间切片)的数据块，内含分页循环 │ │ - 带指数退避的重试机制（限频/超时/5xx） │ └─────────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ API 调用层 │ │ - TickDB /v1/market/kline 接口 │ │ - 自动处理3001限频，读取Retry-After │ │ - 超时设置：(3.05, 30) 连接超时3秒，读取超时30秒 │ │ - 分页：每次最多1000条，用最后一条时间戳+1继续拉取 │ └─────────────────────────────┬───────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ 数据存储层 │ │ - 每个symbol独立CSV文件 │ │ - 流式追加写入，不在内存累积 │ │ - 记录进度文件（JSON），支持断点续传 │ └─────────────────────────────────────────────────────────────────┘ 

运行时的典型日志输出：

[INFO] 加载进度文件，已记录 2 个symbol [DEBUG] AAPL.US 切片 00 已完成，跳过 [WARNING] 触发限频，等待 5 秒 (HTTP 3001) [INFO] TSLA.US [00-00] 拉取成功，流式落盘 11700 条 [INFO] TSLA.US 切片 00 已标记完成 

import os import json import time import asyncio import aiohttp import aiofiles from datetime import datetime, timedelta from typing import List, Optional, Dict, Any import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ⚠️ 工程预警：API Key 必须从环境变量读取，严禁硬编码 API_KEY = os.environ.get("TICKDB_API_KEY") if not API_KEY: raise ValueError("请设置环境变量 TICKDB_API_KEY") BASE_URL = "//api.tickdb.ai" HEADERS = {"X-API-Key": API_KEY} MAX_RETRIES = 5 BASE_DELAY = 1 MAX_DELAY = 60 CONCURRENT_LIMIT = 3 # 并发数，免费层建议保守设置 async def fetch_kline_slice( session: aiohttp.ClientSession, symbol: str, start_time: int, end_time: int, interval: str = "1m", limit: int = 1000 # TickDB 单次最大 1000 条 ) -> Optional[List[Dict[str, Any]]]: """ 拉取一个时间切片的历史K线，带分页循环、指数退避重试和限频自适应。 ⚠️ 工程预警： - 免费层限频严格，建议并发数不超过3 - 该函数已包含指数退避重连、限频自适应和分页逻辑，可直接用于生产 """ all_klines = [] current_start = start_time url = f"{BASE_URL}/v1/market/kline" while current_start < end_time: params = { "symbol": symbol, "interval": interval, "start_time": current_start, "end_time": end_time, "limit": limit } retry_count = 0 page_success = False while retry_count <= MAX_RETRIES: try: async with session.get( url, headers=HEADERS, params=params, timeout=aiohttp.ClientTimeout(connect=3.05, sock_read=30) ) as resp: data = await resp.json() code = data.get("code", 0) if code == 0: klines = data.get("data", {}).get("klines", []) if not klines: return all_klines if all_klines else None all_klines.extend(klines) logger.debug(f"{symbol} 分页拉取 {len(klines)} 条，累计 {len(all_klines)} 条") if len(klines) < limit: # 已拉完该时间切片 logger.info(f"{symbol} [{start_time}-{end_time}] 拉取完成，共 {len(all_klines)} 条") return all_klines else: # 可能还有更多数据，用最后一条的时间戳+1作为下一页起点 current_start = klines[-1]['time'] + 1 page_success = True retry_count = 0 # 重置重试计数 break elif code == 3001: # 限频 retry_after = int(resp.headers.get("Retry-After", 5)) logger.warning(f"触发限频，等待 {retry_after} 秒") await asyncio.sleep(retry_after) retry_count += 1 continue elif code in (1001, 1002): raise ValueError("API Key 无效，请检查环境变量") elif code == 2002: logger.error(f"{symbol} 不存在，跳过") return None else: logger.error(f"未知错误 {code}: ") retry_count += 1 except asyncio.TimeoutError: logger.warning(f"{symbol} 请求超时，重试 {retry_count+1}/{MAX_RETRIES}") retry_count += 1 except Exception as e: logger.error(f"{symbol} 请求异常: {e}") retry_count += 1 # 指数退避 + 抖动 if retry_count <= MAX_RETRIES and not page_success: delay = min(BASE_DELAY * (2 (retry_count - 1)), MAX_DELAY) jitter = delay * 0.1 * (hash(symbol) % 100) / 100 await asyncio.sleep(delay + jitter) if not page_success: logger.error(f"{symbol} 分页拉取失败，已获取 {len(all_klines)} 条") return all_klines if all_klines else None return all_klines 

def generate_time_slices( start_date: datetime, end_date: datetime, slice_days: int = 30 ) -> List[tuple]: """ 将长时间范围切分为多个小片，便于断点续传。 切分粒度可调整。每个切片内部会通过分页循环完整拉取。 """ slices = [] current = start_date while current < end_date: slice_end = min(current + timedelta(days=slice_days), end_date) slices.append(( int(current.timestamp() * 1000), int(slice_end.timestamp() * 1000) )) current = slice_end return slices 

import json from pathlib import Path class ProgressManager: """管理拉取进度，支持断点续传""" def __init__(self, progress_file: str): self.progress_file = Path(progress_file) self.progress: Dict[str, List[int]] = {} self._load() def _load(self): if self.progress_file.exists(): with open(self.progress_file, 'r') as f: self.progress = json.load(f) logger.info(f"加载进度文件，已记录 {len(self.progress)} 个symbol") def save(self): with open(self.progress_file, 'w') as f: json.dump(self.progress, f) def is_slice_done(self, symbol: str, start_ts: int) -> bool: return symbol in self.progress and start_ts in self.progress[symbol] def mark_slice_done(self, symbol: str, start_ts: int): if symbol not in self.progress: self.progress[symbol] = [] if start_ts not in self.progress[symbol]: self.progress[symbol].append(start_ts) self.save() 

import aiofiles import csv async def append_klines_to_csv(symbol: str, klines: List[Dict], output_dir: str): """流式追加写入CSV，不在内存累积""" output_path = Path(output_dir) / f"{symbol}.csv" file_exists = output_path.exists() async with aiofiles.open(output_path, 'a', newline='') as f: writer = csv.writer(f) if not file_exists: await writer.writerow(['timestamp', 'open', 'high', 'low', 'close', 'volume']) for k in klines: await writer.writerow([ k['time'], k['open'], k['high'], k['low'], k['close'], k['volume'] ]) 

async def download_symbol( session: aiohttp.ClientSession, symbol: str, start_date: datetime, end_date: datetime, progress: ProgressManager, output_dir: str, semaphore: asyncio.Semaphore ): """下载单个symbol的全量历史数据""" async with semaphore: slices = generate_time_slices(start_date, end_date, slice_days=30) for start_ts, end_ts in slices: if progress.is_slice_done(symbol, start_ts): logger.debug(f"{symbol} 切片 {start_ts} 已完成，跳过") continue klines = await fetch_kline_slice(session, symbol, start_ts, end_ts) if klines: await append_klines_to_csv(symbol, klines, output_dir) progress.mark_slice_done(symbol, start_ts) else: logger.warning(f"{symbol} 切片 {start_ts} 拉取失败，将在下次运行时重试") # 礼貌性等待，避免连续请求触发限频 await asyncio.sleep(0.2) async def main(symbols: List[str], start_date: datetime, end_date: datetime, output_dir: str): """主入口""" Path(output_dir).mkdir(parents=True, exist_ok=True) progress = ProgressManager(f"{output_dir}/progress.json") semaphore = asyncio.Semaphore(CONCURRENT_LIMIT) async with aiohttp.ClientSession() as session: tasks = [ download_symbol(session, sym, start_date, end_date, progress, output_dir, semaphore) for sym in symbols ] await asyncio.gather(*tasks, return_exceptions=True) if __name__ == "__main__": # 示例：拉取苹果和特斯拉过去5年的1分钟线 symbols = ["AAPL.US", "TSLA.US"] end = datetime.now() start = end - timedelta(days=365 * 5) asyncio.run(main(symbols, start, end, "./data")) 

代码能跑通只是第一步。要让回测结果可靠，还需要校验数据完整性。

TickDB 返回的时间戳是 UTC 毫秒，这是正确的工程实践。但在美股回测中，直接使用 UTC 时间会产生两个问题：

1. 美股交易时段是美东时间 09:30-16:00，每年 3 月和 11 月夏令时切换，与 UTC 的偏移量不同
2. 如果用 UTC 时间直接过滤“开盘前 30 分钟”，在夏令时切换前后会错位

正确做法：

import pandas as pd import pytz # 读取数据后转换时区 df['timestamp_utc'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True) df['timestamp_et'] = df['timestamp_utc'].dt.tz_convert('America/New_York') # 过滤正常交易时段（自动处理夏令时） mask = (df['timestamp_et'].dt.time >= pd.Timestamp('09:30').time()) & (df['timestamp_et'].dt.time <= pd.Timestamp('16:00').time()) df_trading = df[mask] 

这个转换是跨市场回测的基础——TickDB 统一返回 UTC 的优势在于，你只需要维护一套时区转换逻辑，而不是每个市场分别适配。

TickDB 返回的K线在停牌期间不返回空K线。这意味着不同股票的时间索引维度不一致。回测前必须做两件事：

1. 生成标准交易日历的完整分钟时间轴
2. 用 pandas.merge_asof 或 reindex 对齐，停牌期间用前向填充

如果只拉取了当前活跃股票的历史数据，回测收益会被幸存者偏差高估。TickDB 支持获取历史成分股列表和已退市股票数据。在构建回测标的池时，务必包含退市股票，并在退市日之后将其剔除。

import pandas as pd def validate_completeness(df: pd.DataFrame, symbol: str, start_date: datetime, end_date: datetime): """简单校验：检查起止时间和记录数是否在合理范围""" expected_days = (end_date - start_date).days actual_days = df['timestamp'].dt.date.nunique() coverage = actual_days / expected_days print(f"{symbol}: 预期 {expected_days} 天，实际 {actual_days} 天，覆盖率 {coverage:.1%}") if coverage < 0.95: print(f"⚠️ {symbol} 数据覆盖率偏低，请检查停牌或拉取失败的时间段") 

批量拉取历史数据是量化工程的“第一公里”。这公里走不稳，后面回测跑出来的收益都是空中楼阁。

本文给出的代码可以直接用于生产——它处理了限频、超时、分页、断点续传、并发控制、时区转换，并在本地落盘时采用了流式写入。你只需要替换 symbols 列表和日期范围，就能拉取美股、港股、A股、加密货币的历史分钟数据。

如果你是个人开发者：可以到官网注册申请 API KEY。免费层足够拉取中等规模的标的。

如果你是量化团队：需要更高并发或企业级SLA，可到官网联系官方获取团队方案。

如果你习惯用AI辅助开发：到 Clawhub 搜索“tickdb-market-data SKILL”，用自然语言查询历史行情。

本文不构成任何投资建议。市场有风险，投资需谨慎。