tqsdk-python 数据下载工具 DataDownloader 详解：批量下载期货、期权、股票历史行情并导出 CSV-平芜编程栈

金融科技

【免费下载链接】tqsdk-python

天勤量化开发包, 期货量化, 实时行情/历史数据/实盘交易

项目地址：https://gitcode.com/gh_mirrors/tq/tqsdk-python

本文围绕天勤量化开发包（tqsdk-python）中的tqsdk.tools.DataDownloader历史数据下载工具展开，结合其 API 参考文档 tqsdk.tools.download 与源码实现 tqsdk/tools/downloader.py，完整讲解其构造参数、多合约时间对齐、Tick/K线两种数据形态、股票复权处理以及底层"图表窗口"式数据拉取原理。读完后你将能够编写可复制运行的批量历史行情下载脚本，理解输出 CSV 的字段结构，并能针对已下市合约、流式输出（StreamWriter）等场景做工程化配置。

一、DataDownloader 的定位与使用前提

DataDownloader定义在 tqsdk/tools/downloader.py，并通过 tqsdk/tools/init.py 导出，导入方式为：

from tqsdk.tools import DataDownloader

它的设计目标是：让用户以几行代码完成期货、期权和股票类历史数据的批量下载，数据支持Tick 级别精度和任意 K 线周期，最终输出为 CSV 文件（或直接流式写入asyncio.StreamWriter），多合约数据按时间横向对齐到同一行中。

需要特别注意的使用前提（源码中有硬性检查）：

仅限专业版权限。构造时会校验账户是否具备tq_dl功能点，否则直接抛出异常：

# tqsdk/tools/downloader.py 第 105-106 行 if not self._api._auth._has_feature("tq_dl"): raise Exception("您的账户不支持下载历史数据功能，需要购买后才能使用。...")

这一点与专业版说明文档 doc/profession.rst 中"数据下载工具tqsdk.tools.DataDownloader是 TqSdk 专业版中的功能"的表述一致。 2.Tick 数据仅支持单合约。dur_sec=0且symbol_list为多个合约时抛出"Tick序列不支持多合约"（tqsdk/tools/downloader.py）。 3. 下载是一个异步任务：构造函数通过self._api.create_task(self._run())把下载挂到TqApi的事件循环中，调用方需要通过api.wait_update()驱动它前进，而不是同步等待返回。

二、构造函数参数完整说明

构造函数签名为（tqsdk/tools/downloader.py）：

DataDownloader(api, symbol_list, dur_sec, start_dt, end_dt, csv_file_name, write_mode="w", adj_type=None)

参数	类型	说明
`api`	`TqApi`	TqApi 实例，下载器复用该实例的连接与事件循环
`symbol_list`	`str`/`List[str]`	需要下载数据的合约代码；指定多个合约时，其他合约按第一个合约的交易时间对齐
`dur_sec`	`int`	数据周期（秒）。`60`为 1 分钟线、`3600`为 1 小时线、`86400`为日线、`0`为 Tick 数据
`start_dt`	`date`/`datetime`	起始时间。`date`表示交易日，`datetime`表示具体时间点，无时区信息时默认为北京时间
`end_dt`	`date`/`datetime`	结束时间，含义同上
`csv_file_name`	`str`/`asyncio.StreamWriter`	[必填] 输出方式：文件名字符串，或直接写入 StreamWriter
`write_mode`	`str`	写入模式，默认`"w"`（覆盖写入，写标题行）；`"a"`为追加写入，不写标题行，直接写数据
`adj_type`	`str`/`None`	复权方式：`"F"`前复权、`"B"`后复权、`None`不复权（默认）。只对股票、基金合约有效；源码中`"FORWARD"`/`"BACK"`写法也兼容，实际取首字母判断（tqsdk/tools/downloader.py）

2.1 时间参数的底层处理

date与datetime两种输入在 tqsdk/datetime.py 的_convert_user_input_to_nano中被统一转换为内部纳秒时间戳，且内部时间强制按东八区（北京时间）解释。两者的关键差异在于边界计算：

传入date时，起始时间由_get_trading_day_start_time（tqsdk/datetime.py）换算：以交易日当天 0 点再向前回退 6 小时作为锚点，并把周六、周日回推到周四/周五之前——这样"一个交易日"实际覆盖前一晚的夜盘；结束时间由_get_trading_day_end_time（tqsdk/datetime.py）加 18 小时，覆盖到当日晚夜盘结束。
传入datetime时则按字面时间点计算，适合精确截取某时段的夜盘或日盘。

因此，若希望某交易日的数据完整（含夜盘），推荐使用date；若只要白天时段或跨交易日的任意区间，使用datetime更直观。

2.2 已下市合约的超时优化

源码维护了一份已下市（不再产生交易）合约清单，来自打包资源 tqsdk/tools/dead_ins.lzma（LZMA 压缩的文本列表，由MANIFEST.in声明随包分发，当前包含上百个已摘牌期权合约代码）：

# tqsdk/tools/downloader.py 第 110-111 行 # 下载合约超时时间（默认 30s），已下市的没有交易的合约，超时时间可以设置短一点（2s） self._timeout_seconds = 2 if any([symbol in DEAD_INS for symbol in self._symbol_list]) else 30

从源码结构看，当symbol_list中任一合约命中该清单时，"等待数据序列建立"的超时窗口从 30 秒缩短到 2 秒——因为已下市合约必然没有新成交，不必长时间空等。配合版本记录中"DataDownloader在下载没有任何成交数据的合约时能够及时退出，避免等待超时或报错"的改进说明，下载空数据合约时任务会直接结束而不是卡住。

三、完整使用示例

官方文档字符串（tqsdk/tools/downloader.py）给出的示例覆盖了日线、5 分钟线、多合约分钟线和盘口 Tick 四类场景，仓库中另有可直接运行的演示脚本 tqsdk/demo/tutorial/downloader.py。核心用法如下：

from datetime import datetime, date from contextlib import closing from tqsdk import TqApi, TqAuth from tqsdk.tools import DataDownloader api = TqApi(auth=TqAuth("快期账户", "账户密码")) download_tasks = {} # 1) 下载 2018-01-01 ~ 2018-09-01 的 SR901 日线 download_tasks["SR_daily"] = DataDownloader(api, symbol_list="CZCE.SR901", dur_sec=24*60*60, start_dt=date(2018, 1, 1), end_dt=date(2018, 9, 1), csv_file_name="SR901_daily.csv") # 2) 下载 2017-01-01 ~ 2018-09-01 的 rb 主连 5 分钟线（KQ.m@ 为主连合约代码） download_tasks["rb_5min"] = DataDownloader(api, symbol_list="KQ.m@SHFE.rb", dur_sec=5*60, start_dt=date(2017, 1, 1), end_dt=date(2018, 9, 1), csv_file_name="rb_5min.csv") # 3) 多合约分钟线：cu1805/cu1807/IC1803 按第一个合约 cu1805 的时间对齐 # cu1805 夜盘时段 IC1803 无交易 → 相应字段为 #N/A # cu1805 在 13:00-13:30 不交易 → 该时段 K 线直接跳过 download_tasks["cu_min"] = DataDownloader(api, symbol_list=["SHFE.cu1805", "SHFE.cu1807", "CFFEX.IC1803"], dur_sec=60, start_dt=datetime(2018, 1, 1, 6, 0, 0), end_dt=datetime(2018, 6, 1, 16, 0, 0), csv_file_name="cu_min.csv") # 4) 下载 T1809 盘口 Tick 数据（dur_sec=0，仅限单合约） download_tasks["T_tick"] = DataDownloader(api, symbol_list=["CFFEX.T1809"], dur_sec=0, start_dt=datetime(2018, 5, 1), end_dt=datetime(2018, 6, 1), csv_file_name="T1809_tick.csv") # 使用 with closing 机制确保下载完成后释放对应资源 with closing(api): while not all([v.is_finished() for v in download_tasks.values()]): api.wait_update() print("progress: ", {k: ("%.2f%%" % v.get_progress()) for k, v in download_tasks.items()})

要点说明：

驱动方式：DataDownloader本身不阻塞，必须靠api.wait_update()轮询推进，with closing(api)保证全部任务完成后连接被正常关闭。仓库演示脚本 tqsdk/demo/tutorial/downloader.py 中同时驱动了 K 线与 Tick 两个任务并分别打印进度，可直接照抄该模式。
进度与完成检查：get_progress()返回百分比，其实现（tqsdk/tools/downloader.py）是按时间推进估算的：(当前时间戳 - 起始) / (结束 - 起始) × 100，任务完成（_task.done()）时返回 100.0；is_finished()判断底层 task 是否结束。
多任务共享一个 api：如上例，多个下载器可以共用同一个TqApi实例，各自独立输出文件，只需在所有is_finished()为真后再退出循环。

四、输出 CSV 的字段结构与多合约对齐规则

4.1 表头与字段

表头由_get_headers生成（tqsdk/tools/downloader.py）：前两列固定为datetime（可读时间字符串，保留纳秒精度）和datetime_nano（纳秒时间戳），之后每个合约按{symbol}.{col}展开一组列。具体数据列由_get_data_cols决定（tqsdk/tools/downloader.py）：

K 线模式（dur_sec != 0）：open, high, low, close, volume, open_oi, close_oi（开高低收、成交量、开盘/收盘持仓量）。
Tick 模式（dur_sec == 0）：last_price, highest, lowest, average, volume, amount, open_interest，再追加盘口列bid_price1..N / bid_volume1..N / ask_price1..N / ask_volume1..N；盘口档位数取决于交易所——SHFE、INE、SSE、SZSE取 5 档，其余交易所取 1 档（tqsdk/tools/downloader.py）。

datetime列由_nano_to_str格式化为YYYY-MM-DD HH:MM:SS.fffffffff（纳秒 9 位，tqsdk/tools/downloader.py），注释明确说明是为了保留 nano 精度而未复用通用的时间戳转字符串函数。

4.2 价格精度与缺失值

_get_value（tqsdk/tools/downloader.py）负责取字段并做两件事：

价格类字段（PRICE_KEYS，即open/high/low/close/last_price/highest/lowest及五档买卖价，定义于 tqsdk/tools/downloader.py）会按该合约price_decs（最小变动价位对应的小数位数）round一次，避免 CSV 中出现浮点噪声；
字段不存在（KeyError）时写入字符串#N/A，数值类型异常（TypeError）时写入nan。

4.3 多合约横向对齐

文档对多合约行为的关键承诺是："多合约按时间横向对齐"。具体规则（来自 tqsdk/tools/downloader.py 的说明与_download_data的实现）：

以symbol_list中第一个合约的行情序列为主轴，逐行推进；
第一个合约某时刻不交易（例如 cu1805 在 13:00-13:30 休市）时，整行 K 线直接跳过，后续合约该时段数据不会单独落盘；
其他合约（副合约）在该时刻无数据时，通过主合约序列中的binding绑定关系查找副合约对应的行 ID，找不到则该合约各字段写#N/A（例如 cu1805 夜盘时段，IC1803 的各项数据为#N/A）；
行序严格按时间升序推进，因此 CSV 可直接作为宽表用于跨合约价差、套利策略的研究。

4.4 流式输出：StreamWriter

csv_file_name传入asyncio.StreamWriter时，下载结果不落本地文件，而是逐行写入网络流，并在每行写入后await self._csv_file_name.drain()（tqsdk/tools/downloader.py），结束时write_eof()关闭流。这适合把数据直接推送到服务端或管道。注意：_get_data_series()只支持csv_file_name为str的情形（见下节）。

4.5 write_mode 与追加写

write_mode只接受"w"/"a"，否则构造时即抛异常（tqsdk/tools/downloader.py）：

"w"：覆盖写入，先写标题行再写数据；
"a"：追加写入，不写标题行，直接追加数据行。

该参数在版本记录 doc/version.rst 中有明确说明："DataDownloader 增加 write_mode 参数，作为写入模式参数，并且在模式为 'a' 时，不写入标题行"。典型用法是把一个大时间区间切成多个任务（每段一个新实例、同一文件名、write_mode="a"），首段用"w"，后续段用"a"，从而分段下载超长区间。

五、股票/基金复权（adj_type）的实现原理

adj_type只对ins_class为STOCK/FUND的合约生效。下载循环中（tqsdk/tools/downloader.py）：

通过_ensure_dividend_factor（tqsdk/tools/downloader.py）为每个合约预先构建复权因子矩阵并缓存在self._dividend_cache中，每个合约只计算一次。因子来源是 tqsdk/utils.py 的_get_dividend_factor：它读取合约的stock_dividend_ratio（送转股）与cash_dividend_ratio（现金分红）事件序列，对每个除权除息日拉取前一日日线的收盘价pre_close，按公式
```
factor = (pre_close - cash_dividend) / pre_close / (1 + stock_dividend)
```
计算该日的除权因子（见 tqsdk/utils.py）。
前复权（F）把因子序列逆序累乘得到各时点的累计前复权因子；后复权（B）按因子累乘的倒数并 shift 一位，语义统一为"截止 datetime 之前（不包含）应使用 factor 复权"（tqsdk/tools/downloader.py）。
由于复权事件表与行情序列都是时间升序，主循环用归并方式滚动推进last_dt指针，取当前时点适用因子；因子不为 1 时，对价格列（K 线为open/high/low/close，Tick 为last_price/highest/lowest及五档盘口价）逐列相乘后写盘（tqsdk/tools/downloader.py）。

版本记录中还有一条"修复: DataDownloader 计算复权因子错误"的条目（doc/version.rst），说明复权路径经历过实际修正，使用前以当前仓库实现为准。

六、底层工作原理：基于"图表窗口"的分页拉取

从源码结构看，DataDownloader并非逐条调用数据接口，而是复用了 TqApi 与天勤服务器之间的图表（chart）数据通道，这是理解其性能与限制的关键。核心流程在_download_data（tqsdk/tools/downloader.py）：

建立图表窗口：向self._api._send_chan发送set_chart消息，携带唯一的chart_id（_generate_uuid("PYSDK_downloader")）、ins_list（逗号拼接的全部合约）、duration（dur_sec换算成的纳秒周期）、view_width: 2000（一次视口 2000 根 K 线/Tick）以及定位参数focus_datetime（起始时间纳秒戳）、focus_position: 0（定位到窗口左端点）。
等待数据就绪：通过self._api.register_update_notify()注册更新通知，循环中先检查服务器回执（chart.state必须包含全部请求参数）与chart.ready（主合约和所有副合约的数据序列都已到达），再开始取数。
窗口内逐行推进：读取chart的left_id/right_id，从max(left_id, 0)开始逐 ID 取出主合约行（K 线存于charts.*.klines.{symbol}.{dur_nano}，Tick 存于ticks.{symbol}），并按 4.3 节的绑定规则拼出副合约列，yield给上层写 CSV。若serials[0]的last_id == -1（完全没有数据）或行时间超出end_dt_nano，立即return。
窗口滚动：当current_id推进超过right_id，移除focus_datetime/focus_position，改用left_kline_id = current_id重新发送set_chart请求后续数据，实现分页续传；每推进一行都会更新_current_dt_nano，即get_progress()的数据来源。
资源释放：finally中再次发送set_chart并把ins_list置空，释放服务端图表资源（tqsdk/tools/downloader.py）。

外层_run（tqsdk/tools/downloader.py）则负责：打开 CSV 文件（或包装 StreamWriter 为同步csv.writer）、"w"模式下写表头、消费上述异步生成器逐行写盘，并在finally中关闭文件、await gen.aclose()收尾。

七、任务控制与结果获取

除构造参数外，DataDownloader实例公开的方法有：

is_finished() -> bool：底层 task 是否完成（tqsdk/tools/downloader.py）；
get_progress() -> float：下载进度百分比，完成返回 100（tqsdk/tools/downloader.py）；
_get_data_series() -> Optional[pandas.DataFrame]：下载完成后把结果 CSV 读成 pandas DataFrame 返回；下载完成前返回None。注意它是下划线开头的半私有方法（tqsdk/tools/downloader.py），且仅在csv_file_name为字符串时可用，StreamWriter模式会抛出异常。版本记录（doc/version.rst）中显示该接口曾有"导致内存泄漏暂时下线修复"的历史，如需在内存中拿 DataFrame，更稳妥的做法是下载完成后再自行pandas.read_csv。

八、与 TqApi 数据接口的选型对比

同为专业版数据能力（_has_feature("tq_dl")校验同时出现在 tqsdk/api.py 与 tqsdk/tools/downloader.py），TqApi.get_kline_data_series/TqApi.get_tick_data_series（tqsdk/api.py）与DataDownloader可以互为补充：

维度	`DataDownloader`	`TqApi.get_kline_data_series`/`get_tick_data_series`
输出形态	CSV 文件 / StreamWriter 流	pandas DataFrame
合约数量	支持多合约（K 线模式横向对齐）	单合约（多合约请求会抛"暂不支持多合约请求"，见 tqsdk/api.py）
调用方式	异步任务，需`wait_update()`驱动	同步阻塞式获取，且不支持在协程中调用（tqsdk/api.py）
复权	支持`adj_type`F/B	同样支持`adj_type`参数
回测环境	可用于实盘 api 实例	明确"不支持在回测中调用"（tqsdk/api.py）
适用场景	大批量、长区间、多合约导出落盘	小规模、直接进 DataFrame 的分析

从 doc/profession.rst 的"其他相关函数"一节看，官方也把get_kline_data_series定位成"以起始日期获取 Dataframe 格式的 kline 数据"的轻量入口；而需要 CSV 导出或跨合约对齐宽表时，DataDownloader是更合适的工具。

九、实践注意事项小结

权限：非专业版账户构造即抛异常，先确认账户具备tq_dl功能点；
时间类型选择：date会覆盖完整交易日（含前一晚夜盘），datetime是精确到点的区间，无时区输入一律按北京时间解释；
多合约：只有第一个合约的行情节奏决定主行序，副合约缺数据表现为#N/A，第一个合约休市时段整行跳过；
Tick 限制：dur_sec=0时只能传单合约；SHFE/INE/SSE/SZSE 输出 5 档盘口，其余交易所 1 档；
分段下载：用"w"起始 +"a"追加可在不重写下表头的前提下分段续传；
进度监控：get_progress()是按时间戳推算的近似值，数据稀疏的区间接近完成前可能出现长时间"卡进度"，属正常现象；
资源管理：始终用with closing(api):包裹驱动循环，下载器内部也会释放图表资源，避免服务端会话泄漏。

以上内容的源码依据集中于 tqsdk/tools/downloader.py（核心实现）、tqsdk/datetime.py（交易日边界换算）、tqsdk/utils.py（复权因子计算）、演示脚本 tqsdk/demo/tutorial/downloader.py，以及 API 参考页 doc/reference/tqsdk.tools.download.rst 与专业版说明 doc/profession.rst，可按路径在当前仓库中逐一核验。