做量化回测第一道坎永远不是策略而是数据。这一点我自己吃过亏。刚开始学 Python 量化那会儿我一度觉得写策略才是最难的结果在数据源上反复折腾了一个多星期这个要注册那个要积分还有一个爬到一半网站改版直接断流。后来一个做交易的老哥提醒我试试 pytdx直接从通达信的行情服务器拉K线。我半信半疑装了一下五分钟跑通了第一段代码才算真的进了门。这篇博客把我从零开始用 pytdx 获取 800 个交易日K线数据的完整过程整理了出来环境搭建、服务器连接、接口参数、完整代码、数据落盘还有我踩过的几个坑。整篇的代码都是可以直接复制运行的适合理顺过量化基本概念、想自己搞定A股日线数据的同学。目标很简单跑完之后你手里就有一份干净的本地 CSV代码、日期、开高低收、成交量、成交额齐活。1. 为什么我选 pytdx而不是 tushare 或 akshare先说结论对个人研究、尤其是 A 股日线级别的数据获取来说pytdx 是性价比最高的方案之一。开源社区里常见的免费数据源大概是这几类tushare pro 数据质量高但很多常用接口需要积分新用户注册之后有不同程度的门槛baostock 免费且稳定可接口风格偏老用起来不算顺手akshare 覆盖面广、文档全但不少数据本质上是爬虫抓取目标网站一改版接口就断一两天。pytdx 走的是另一条路它不去爬网页而是直接和通达信的行情服务器建立 TCP 连接用二进制协议交换数据。数据源注册门槛费用获取日K便利性稳定性tushare pro需要 token部分免费积分限制较多很稳baostock需要登录免费可用但接口老较稳akshare不需要免费爬虫依赖网页一般pytdx不需要免费直接协议连接依赖行情服务器pytdx 最大的优点是快和简单pip 装完几行代码就能拉到你要的行情不需要注册任何账号也没有积分概念。缺点也很明显它只覆盖通达信客户端的常见品种比如沪深 A 股、指数、场内基金、债券之类期货和美股完全没有。所以我现在的工作流里A 股研究用 pytdx 拉日线其他品种再单独维护别的数据源。还得多说一句pytdx 不是官方库属于社区逆向工程的成果。因为依赖的是公开行情服务器不能保证永远长期有效有时候某个服务器地址会挂掉换个 IP 又能继续用。但对量化入门来说拿它学习、跑策略原型完全够用。2. 先把连接跑通环境、服务器地址和最小验证代码在碰K线接口之前先确认 pytdx 装好、服务器能连上。这一步不跑通后面全白搭。2.1 安装 pytdx直接装就行pip install pytdx我测试过的环境包括 Python 3.8 到 3.11都能正常工作。pytdx 对第三方库的依赖很少基本就是 pandas、numpy 这种常见库装完不会给项目带来额外负担。如果你用的是 Anaconda 或者 Miniconda直接在 base 环境里装也没问题不影响其他项目。2.2 行情服务器地址怎么填pytdx 连的是通达信行情服务器这类地址网上能搜到不少但时效性参差不齐。我实测下来下面这几个当时是可以连通的不过并不保证永久有效写着供大家参考IP 地址端口119.147.212.817709115.238.90.1657709123.125.108.147709218.108.98.2447709端口一般用 7709有些历史攻略会写 7708也可以试。connect 方法支持time_out参数我建议改成 10 秒默认值在部分网络环境下容易被误判成超时。2.3 最小验证代码连接测试不一定一次成功所以我习惯写一个循环挨个试服务器地址from pytdx.hq import TdxHq_API api TdxHq_API() server_list [ (119.147.212.81, 7709), (115.238.90.165, 7709), (123.125.108.14, 7709), ] connected False for ip, port in server_list: try: ok api.connect(ip, port, time_out10) if ok: print(f连接成功: {ip}:{port}) connected True break except Exception as e: print(f{ip}:{port} 连接失败: {e}) if connected: count api.get_security_count(0) print(深圳市场证券总数:, count) api.disconnect() else: print(所有服务器连接失败请更换服务器地址)get_security_count(0)返回的是深圳市场证券总数如果返回一个正常正整数说明连接没问题。不同服务器的数据略有差异数字不用完全一致能通就行。3. get_security_bars 参数拆解800根K线是从哪来的连接通了之后最核心的接口就是get_security_bars。标题里说的 800 个交易日就是通过它的count参数控制的。3.1 接口签名和参数含义这个函数的调用方式是api.get_security_bars(category, market, code, start, count)参数含义常见取值categoryK线周期41分钟, 55分钟, 615分钟, 730分钟, 860分钟, 9日Kmarket市场代码0深圳, 1上海code证券代码六位数字字符串start从最新K线往前数的偏移量0 表示从最新一根开始取count取多少根单次最大 800注意category9 才是日线。如果看到别人的代码里 category 传的是 7 或 8那一般是分钟线或小时线不要照抄先确认自己要什么周期。3.2 market 和 code 的配对最容易踩的隐蔽坑这一点必须单独拿出来说。market 不是按股票代码开头区分的而是按交易所区分的。0 是深交所1 是上交所。同一个代码在不同市场可能对应完全不同的标的。举两个最常见的例子000001在 market1 时是上证指数在 market0 时是平安银行。600519这种 6 开头的是沪市market 要填 1000858这种 0 开头的是深市market 填 0。如果你把 market 和 code 配对搞反了接口并不会报错拉回来的数据是另一个标的这种错最坑因为很难一眼看出来。核对代码是不是自己想要的最简单的办法是看返回数据里的最新日期和最近几根K线价格再和你平时用的行情软件对一下几秒钟就能确认。3.3 start、count 与 800 根K线的换算关系行情服务器单次最多返回 800 根K线。start0 代表最新的一根start1 代表往前推一根以此类推。所以如果你只要最近 800 个交易日一次调用就够了start0, count800。A 股一年大概 250 个交易日800 个交易日大约是三年多的数据。做动量、趋势这类日线级策略这个长度完全够用。如果后续你想拉 2000 个交易日就得循环调用了具体怎么写放到下一章详细展开。这里先记住一个核心结论单次调用最多 800 根要更多就多调几次靠start偏移量往前翻。每次调用返回的是一个列表列表里每个元素是一根K线。字段包括open、close、high、low、vol、amount、datetime等还有year、month、day这类拆分字段日线里基本用不到。4. 拉取800个交易日K线的完整代码循环、拼接与容错单次调用已经能覆盖 800 天需求但考虑到通用性和稳定性我会把它封装成一个函数顺便支持更长周期。4.1 一个通用的按天数拉取函数def fetch_kline(api, market, code, days800): records [] rounds (days 799) // 800 for i in range(rounds): start i * 800 count min(800, days - i * 800) data api.get_security_bars(9, market, code, start, count) if not data: print(f{code} 第 {i} 轮返回空数据) break records.extend(data) return recordsrounds是调用次数。比如 days800 时 rounds1days2000 时 rounds3。starti*800是核心逻辑第一轮从最新开始取 800 根第二轮从第 801 根开始再取 800 根第三轮再继续往前最后用min处理不足 800 的余数。这个函数只依赖get_security_bars没有引入复杂依赖理解起来也容易。4.2 加上连接失败重试防止“跑一半崩掉”直接连续调用接口偶尔会遇到返回 None 或者抛异常。比如长时间连接被服务器静默断掉再不然就是网络抖动。我的做法是加一层重试失败后断开重连再试一次。SERVER_IP 119.147.212.81 SERVER_PORT 7709 def safe_fetch(api, market, code, days800, retry3): for attempt in range(retry): try: records fetch_kline(api, market, code, days) if records: return records except Exception as e: print(f第 {attempt1} 次尝试失败: {e}) try: api.disconnect() api.connect(SERVER_IP, SERVER_PORT, time_out10) except Exception: pass return []安全调用逻辑很简单失败就断开重连再跑一次最多重试三次。第三次还不行就直接返回空列表让主程序跳过这只股票不把整个流程卡死。如果你想要更稳可以维护一个服务器列表每次重连换一个 IP能进一步提高成功率。4.3 批量拉多只股票时的节奏控制很多新手会把一批股票丢进 for 循环里猛拉结果拉到一半被服务器限流。我自己实测下来拉几十只没事拉几百只就不能太蛮干。推荐在每次请求后 sleep 0.2 到 0.5 秒给服务器一点喘息时间。import time code_list [600519, 000001, 000858] result {} for code in code_list: market 1 if code.startswith(6) else 0 records safe_fetch(api, market, code, days800) if records: result[code] records time.sleep(0.3)注意这里market的判断只是入门写法6 开头一般是沪市0 和 3 开头一般是深市。严谨一点应该维护一份每个代码的市场归属表。对于批量数据更新这种长期任务我还有个习惯每处理 50 只左右主动断开连接再重新连一次。因为行情服务器对长时间占用同一条连接并不友好主动重连反而能降低异常概率。另外8 开头的北交所代码在 pytdx 的 market 参数里通常没有对应市场所以这套代码主要覆盖沪深 A 股。北交所的数据比较特殊要另行处理。5. 数据处理与落盘把接口返回的 dict 转成干净的 DataFrame接口返回的是 list of dict直接拿去做回测不够方便。我的习惯是一步到位转成 pandas DataFrame清洗排序后存 CSV。5.1 转 DataFrame 与字段说明import pandas as pd def records_to_df(records): df pd.DataFrame(records) df[datetime] pd.to_datetime(df[datetime]) df df.set_index(datetime).sort_index() return dfDataFrame 里的字段主要有这些open开盘价、close收盘价、high最高价、low最低价、vol成交量、amount成交额。返回结果里还有year、month、day、hour、minute这些拆分字段转成 datetime 之后基本用不上了可以按需删掉。vol的单位通常是“手”成交额单位是“元”。不同服务器版本可能存在细微差异写策略前先打印几行确认一下。别在单位问题上想当然否则最后收益率算出来差了好几位数找 bug 找到崩溃。5.2 日期排序、去重与缺失处理分页拉取的数据顺序是从新到旧所以必须sort_index()。如果拉长周期时分页边界出现重叠直接用去重解决df df[~df.index.duplicated(keeplast)]日线数据本身不包含停牌日期所以 df 里的日期不连续是正常的。回测时如果需要补齐交易日建议用行情日历去 reindex而不是手动生成工作日序列。简单 demo 里也可以先不做但心里要知道这个差距。5.3 存 CSV 和快速读取import os os.makedirs(data, exist_okTrue) df.to_csv(fdata/{code}.csv) df pd.read_csv(fdata/{code}.csv, index_coldatetime, parse_datesTrue)把数据落盘之后后续回测就不需要每次联网拉取。这也是我把 pytdx 当“数据下载器”用的主要原因下载一次本地复用速度比每次现拉快得多。6. 实战中的坑连接断开、缺K线、未复权数据的处理代码能跑通只是第一步真正写策略回测时你会碰到下面这些在文档里看不到的问题。6.1 程序跑着跑着返回全是 None最常见的情况是连接被服务器断开。pytdx 的连接不是永久连接长时间闲置或者大量请求之后服务器可能直接清理掉这条 TCP 连接。表现就是前几百次调用没问题某一刻开始接口返回 None。排查方法在循环里打印每次调用的 code 和是否成功定位第一个失败点。如果失败点怪异地出现在同一批股票的中间位置基本可以确定是连接被重置。解决办法就是我上面写的重连重试机制再配合每处理几十只主动断开重连。实测下来这个组合能让几千只股票的批量拉取稳定跑到最后。6.2 新股数据不足800根停牌日直接“消失”新股上市没到 800 个交易日调用返回的K线数量肯定少于期望值甚至可能为 0。这属于正常现象。处理方式很简单判断返回记录数是否大于策略需要的最少K线数比如做双均线至少要 20 根以上达不到就跳过。再说停牌。股票停牌期间不会产生K线所以 DataFrame 里的日期会缺失很多天。如果直接算pct_change()会把停牌前最后一天到复牌后第一天的价格变化当作“一天收益”这在理论上是对的但如果策略在停牌期间还有持仓就需要特别处理。入门阶段至少要知道缺失的日期不代表价格不变而是可能没有交易。6.3 未复权数据带来的信号突变pytdx 返回的日K线是未复权价格。分红送股之后价格会突然向下跳空K线图上像崩盘一样。如果不做复权均线类策略会在除权日附近出现假信号收益回测也不准确。我入门时为了解决这个问题走了不少弯路。正规做法是用 pytdx 的get_xdxr_info拉取复权因子自己算前复权但这会引入额外复杂度。如果你只是想先把流程跑通我的建议是暂时选择研究区间内没有大额分红的股票来验证策略等把回测框架搭好之后再回来补复权处理。做任何实盘相关决策之前复权这一步不能省。7. 数据到手之后我的第一步量化尝试数据落盘之后我建议马上做一个小实验把整个流程串起来这样你才会知道数据到底有什么用。7.1 先用最简单的双均线信号练手双均线是量化教程里最常见的例子。思路很简单5 日均线上穿 20 日均线时看多下穿时看空。用 pandas 几行就能算出来。先说清楚这个实验只是为了验证数据链路不是投资建议也不是什么能赚钱的信号。df[ma5] df[close].rolling(5).mean() df[ma20] df[close].rolling(20).mean() df[position] (df[ma5] df[ma20]).astype(int)position是 1 时表示持有0 时表示空仓。这里已经出现了第一个要命的陷阱直接用当天的收盘价算出的均线信号当天收盘就买入在真实场景中是做不到的因为信号要等收盘后才能确认。严谨的做法是把position结果shift(1)也就是第二天才执行昨天的信号避免用到“未来数据”。很多人回测收益非常漂亮一上实盘就崩多半就是这种细节没处理好。shift(1)是回测代码里最值得养成的好习惯。7.2 做一个粗糙的手工回测import matplotlib.pyplot as plt df[ret] df[close].pct_change().fillna(0) df[strategy_ret] df[position].shift(1).fillna(0) * df[ret] net df[strategy_ret].add(1).cumprod() net.plot() plt.show()这段代码会把策略净值曲线画出来。它虽然粗糙手续费、滑点、涨跌停限制都没考虑但至少能帮你验证数据链路是否正常。如果不想画图把累计收益打印出来做个对比也行。7.3 给入门者的一句话数据获取、清洗、存储这条链路是量化的地基。我见过太多人一上来就研究复杂的选股策略结果数据本身有问题结论全是空中楼阁。用 pytdx 把这套流程跑通之后你会发现后面的策略实验顺畅很多。先把地基打牢再往上盖房子。
用pytdx高效获取A股日线K线：从800根数据到量化回测实战
做量化回测第一道坎永远不是策略而是数据。这一点我自己吃过亏。刚开始学 Python 量化那会儿我一度觉得写策略才是最难的结果在数据源上反复折腾了一个多星期这个要注册那个要积分还有一个爬到一半网站改版直接断流。后来一个做交易的老哥提醒我试试 pytdx直接从通达信的行情服务器拉K线。我半信半疑装了一下五分钟跑通了第一段代码才算真的进了门。这篇博客把我从零开始用 pytdx 获取 800 个交易日K线数据的完整过程整理了出来环境搭建、服务器连接、接口参数、完整代码、数据落盘还有我踩过的几个坑。整篇的代码都是可以直接复制运行的适合理顺过量化基本概念、想自己搞定A股日线数据的同学。目标很简单跑完之后你手里就有一份干净的本地 CSV代码、日期、开高低收、成交量、成交额齐活。1. 为什么我选 pytdx而不是 tushare 或 akshare先说结论对个人研究、尤其是 A 股日线级别的数据获取来说pytdx 是性价比最高的方案之一。开源社区里常见的免费数据源大概是这几类tushare pro 数据质量高但很多常用接口需要积分新用户注册之后有不同程度的门槛baostock 免费且稳定可接口风格偏老用起来不算顺手akshare 覆盖面广、文档全但不少数据本质上是爬虫抓取目标网站一改版接口就断一两天。pytdx 走的是另一条路它不去爬网页而是直接和通达信的行情服务器建立 TCP 连接用二进制协议交换数据。数据源注册门槛费用获取日K便利性稳定性tushare pro需要 token部分免费积分限制较多很稳baostock需要登录免费可用但接口老较稳akshare不需要免费爬虫依赖网页一般pytdx不需要免费直接协议连接依赖行情服务器pytdx 最大的优点是快和简单pip 装完几行代码就能拉到你要的行情不需要注册任何账号也没有积分概念。缺点也很明显它只覆盖通达信客户端的常见品种比如沪深 A 股、指数、场内基金、债券之类期货和美股完全没有。所以我现在的工作流里A 股研究用 pytdx 拉日线其他品种再单独维护别的数据源。还得多说一句pytdx 不是官方库属于社区逆向工程的成果。因为依赖的是公开行情服务器不能保证永远长期有效有时候某个服务器地址会挂掉换个 IP 又能继续用。但对量化入门来说拿它学习、跑策略原型完全够用。2. 先把连接跑通环境、服务器地址和最小验证代码在碰K线接口之前先确认 pytdx 装好、服务器能连上。这一步不跑通后面全白搭。2.1 安装 pytdx直接装就行pip install pytdx我测试过的环境包括 Python 3.8 到 3.11都能正常工作。pytdx 对第三方库的依赖很少基本就是 pandas、numpy 这种常见库装完不会给项目带来额外负担。如果你用的是 Anaconda 或者 Miniconda直接在 base 环境里装也没问题不影响其他项目。2.2 行情服务器地址怎么填pytdx 连的是通达信行情服务器这类地址网上能搜到不少但时效性参差不齐。我实测下来下面这几个当时是可以连通的不过并不保证永久有效写着供大家参考IP 地址端口119.147.212.817709115.238.90.1657709123.125.108.147709218.108.98.2447709端口一般用 7709有些历史攻略会写 7708也可以试。connect 方法支持time_out参数我建议改成 10 秒默认值在部分网络环境下容易被误判成超时。2.3 最小验证代码连接测试不一定一次成功所以我习惯写一个循环挨个试服务器地址from pytdx.hq import TdxHq_API api TdxHq_API() server_list [ (119.147.212.81, 7709), (115.238.90.165, 7709), (123.125.108.14, 7709), ] connected False for ip, port in server_list: try: ok api.connect(ip, port, time_out10) if ok: print(f连接成功: {ip}:{port}) connected True break except Exception as e: print(f{ip}:{port} 连接失败: {e}) if connected: count api.get_security_count(0) print(深圳市场证券总数:, count) api.disconnect() else: print(所有服务器连接失败请更换服务器地址)get_security_count(0)返回的是深圳市场证券总数如果返回一个正常正整数说明连接没问题。不同服务器的数据略有差异数字不用完全一致能通就行。3. get_security_bars 参数拆解800根K线是从哪来的连接通了之后最核心的接口就是get_security_bars。标题里说的 800 个交易日就是通过它的count参数控制的。3.1 接口签名和参数含义这个函数的调用方式是api.get_security_bars(category, market, code, start, count)参数含义常见取值categoryK线周期41分钟, 55分钟, 615分钟, 730分钟, 860分钟, 9日Kmarket市场代码0深圳, 1上海code证券代码六位数字字符串start从最新K线往前数的偏移量0 表示从最新一根开始取count取多少根单次最大 800注意category9 才是日线。如果看到别人的代码里 category 传的是 7 或 8那一般是分钟线或小时线不要照抄先确认自己要什么周期。3.2 market 和 code 的配对最容易踩的隐蔽坑这一点必须单独拿出来说。market 不是按股票代码开头区分的而是按交易所区分的。0 是深交所1 是上交所。同一个代码在不同市场可能对应完全不同的标的。举两个最常见的例子000001在 market1 时是上证指数在 market0 时是平安银行。600519这种 6 开头的是沪市market 要填 1000858这种 0 开头的是深市market 填 0。如果你把 market 和 code 配对搞反了接口并不会报错拉回来的数据是另一个标的这种错最坑因为很难一眼看出来。核对代码是不是自己想要的最简单的办法是看返回数据里的最新日期和最近几根K线价格再和你平时用的行情软件对一下几秒钟就能确认。3.3 start、count 与 800 根K线的换算关系行情服务器单次最多返回 800 根K线。start0 代表最新的一根start1 代表往前推一根以此类推。所以如果你只要最近 800 个交易日一次调用就够了start0, count800。A 股一年大概 250 个交易日800 个交易日大约是三年多的数据。做动量、趋势这类日线级策略这个长度完全够用。如果后续你想拉 2000 个交易日就得循环调用了具体怎么写放到下一章详细展开。这里先记住一个核心结论单次调用最多 800 根要更多就多调几次靠start偏移量往前翻。每次调用返回的是一个列表列表里每个元素是一根K线。字段包括open、close、high、low、vol、amount、datetime等还有year、month、day这类拆分字段日线里基本用不到。4. 拉取800个交易日K线的完整代码循环、拼接与容错单次调用已经能覆盖 800 天需求但考虑到通用性和稳定性我会把它封装成一个函数顺便支持更长周期。4.1 一个通用的按天数拉取函数def fetch_kline(api, market, code, days800): records [] rounds (days 799) // 800 for i in range(rounds): start i * 800 count min(800, days - i * 800) data api.get_security_bars(9, market, code, start, count) if not data: print(f{code} 第 {i} 轮返回空数据) break records.extend(data) return recordsrounds是调用次数。比如 days800 时 rounds1days2000 时 rounds3。starti*800是核心逻辑第一轮从最新开始取 800 根第二轮从第 801 根开始再取 800 根第三轮再继续往前最后用min处理不足 800 的余数。这个函数只依赖get_security_bars没有引入复杂依赖理解起来也容易。4.2 加上连接失败重试防止“跑一半崩掉”直接连续调用接口偶尔会遇到返回 None 或者抛异常。比如长时间连接被服务器静默断掉再不然就是网络抖动。我的做法是加一层重试失败后断开重连再试一次。SERVER_IP 119.147.212.81 SERVER_PORT 7709 def safe_fetch(api, market, code, days800, retry3): for attempt in range(retry): try: records fetch_kline(api, market, code, days) if records: return records except Exception as e: print(f第 {attempt1} 次尝试失败: {e}) try: api.disconnect() api.connect(SERVER_IP, SERVER_PORT, time_out10) except Exception: pass return []安全调用逻辑很简单失败就断开重连再跑一次最多重试三次。第三次还不行就直接返回空列表让主程序跳过这只股票不把整个流程卡死。如果你想要更稳可以维护一个服务器列表每次重连换一个 IP能进一步提高成功率。4.3 批量拉多只股票时的节奏控制很多新手会把一批股票丢进 for 循环里猛拉结果拉到一半被服务器限流。我自己实测下来拉几十只没事拉几百只就不能太蛮干。推荐在每次请求后 sleep 0.2 到 0.5 秒给服务器一点喘息时间。import time code_list [600519, 000001, 000858] result {} for code in code_list: market 1 if code.startswith(6) else 0 records safe_fetch(api, market, code, days800) if records: result[code] records time.sleep(0.3)注意这里market的判断只是入门写法6 开头一般是沪市0 和 3 开头一般是深市。严谨一点应该维护一份每个代码的市场归属表。对于批量数据更新这种长期任务我还有个习惯每处理 50 只左右主动断开连接再重新连一次。因为行情服务器对长时间占用同一条连接并不友好主动重连反而能降低异常概率。另外8 开头的北交所代码在 pytdx 的 market 参数里通常没有对应市场所以这套代码主要覆盖沪深 A 股。北交所的数据比较特殊要另行处理。5. 数据处理与落盘把接口返回的 dict 转成干净的 DataFrame接口返回的是 list of dict直接拿去做回测不够方便。我的习惯是一步到位转成 pandas DataFrame清洗排序后存 CSV。5.1 转 DataFrame 与字段说明import pandas as pd def records_to_df(records): df pd.DataFrame(records) df[datetime] pd.to_datetime(df[datetime]) df df.set_index(datetime).sort_index() return dfDataFrame 里的字段主要有这些open开盘价、close收盘价、high最高价、low最低价、vol成交量、amount成交额。返回结果里还有year、month、day、hour、minute这些拆分字段转成 datetime 之后基本用不上了可以按需删掉。vol的单位通常是“手”成交额单位是“元”。不同服务器版本可能存在细微差异写策略前先打印几行确认一下。别在单位问题上想当然否则最后收益率算出来差了好几位数找 bug 找到崩溃。5.2 日期排序、去重与缺失处理分页拉取的数据顺序是从新到旧所以必须sort_index()。如果拉长周期时分页边界出现重叠直接用去重解决df df[~df.index.duplicated(keeplast)]日线数据本身不包含停牌日期所以 df 里的日期不连续是正常的。回测时如果需要补齐交易日建议用行情日历去 reindex而不是手动生成工作日序列。简单 demo 里也可以先不做但心里要知道这个差距。5.3 存 CSV 和快速读取import os os.makedirs(data, exist_okTrue) df.to_csv(fdata/{code}.csv) df pd.read_csv(fdata/{code}.csv, index_coldatetime, parse_datesTrue)把数据落盘之后后续回测就不需要每次联网拉取。这也是我把 pytdx 当“数据下载器”用的主要原因下载一次本地复用速度比每次现拉快得多。6. 实战中的坑连接断开、缺K线、未复权数据的处理代码能跑通只是第一步真正写策略回测时你会碰到下面这些在文档里看不到的问题。6.1 程序跑着跑着返回全是 None最常见的情况是连接被服务器断开。pytdx 的连接不是永久连接长时间闲置或者大量请求之后服务器可能直接清理掉这条 TCP 连接。表现就是前几百次调用没问题某一刻开始接口返回 None。排查方法在循环里打印每次调用的 code 和是否成功定位第一个失败点。如果失败点怪异地出现在同一批股票的中间位置基本可以确定是连接被重置。解决办法就是我上面写的重连重试机制再配合每处理几十只主动断开重连。实测下来这个组合能让几千只股票的批量拉取稳定跑到最后。6.2 新股数据不足800根停牌日直接“消失”新股上市没到 800 个交易日调用返回的K线数量肯定少于期望值甚至可能为 0。这属于正常现象。处理方式很简单判断返回记录数是否大于策略需要的最少K线数比如做双均线至少要 20 根以上达不到就跳过。再说停牌。股票停牌期间不会产生K线所以 DataFrame 里的日期会缺失很多天。如果直接算pct_change()会把停牌前最后一天到复牌后第一天的价格变化当作“一天收益”这在理论上是对的但如果策略在停牌期间还有持仓就需要特别处理。入门阶段至少要知道缺失的日期不代表价格不变而是可能没有交易。6.3 未复权数据带来的信号突变pytdx 返回的日K线是未复权价格。分红送股之后价格会突然向下跳空K线图上像崩盘一样。如果不做复权均线类策略会在除权日附近出现假信号收益回测也不准确。我入门时为了解决这个问题走了不少弯路。正规做法是用 pytdx 的get_xdxr_info拉取复权因子自己算前复权但这会引入额外复杂度。如果你只是想先把流程跑通我的建议是暂时选择研究区间内没有大额分红的股票来验证策略等把回测框架搭好之后再回来补复权处理。做任何实盘相关决策之前复权这一步不能省。7. 数据到手之后我的第一步量化尝试数据落盘之后我建议马上做一个小实验把整个流程串起来这样你才会知道数据到底有什么用。7.1 先用最简单的双均线信号练手双均线是量化教程里最常见的例子。思路很简单5 日均线上穿 20 日均线时看多下穿时看空。用 pandas 几行就能算出来。先说清楚这个实验只是为了验证数据链路不是投资建议也不是什么能赚钱的信号。df[ma5] df[close].rolling(5).mean() df[ma20] df[close].rolling(20).mean() df[position] (df[ma5] df[ma20]).astype(int)position是 1 时表示持有0 时表示空仓。这里已经出现了第一个要命的陷阱直接用当天的收盘价算出的均线信号当天收盘就买入在真实场景中是做不到的因为信号要等收盘后才能确认。严谨的做法是把position结果shift(1)也就是第二天才执行昨天的信号避免用到“未来数据”。很多人回测收益非常漂亮一上实盘就崩多半就是这种细节没处理好。shift(1)是回测代码里最值得养成的好习惯。7.2 做一个粗糙的手工回测import matplotlib.pyplot as plt df[ret] df[close].pct_change().fillna(0) df[strategy_ret] df[position].shift(1).fillna(0) * df[ret] net df[strategy_ret].add(1).cumprod() net.plot() plt.show()这段代码会把策略净值曲线画出来。它虽然粗糙手续费、滑点、涨跌停限制都没考虑但至少能帮你验证数据链路是否正常。如果不想画图把累计收益打印出来做个对比也行。7.3 给入门者的一句话数据获取、清洗、存储这条链路是量化的地基。我见过太多人一上来就研究复杂的选股策略结果数据本身有问题结论全是空中楼阁。用 pytdx 把这套流程跑通之后你会发现后面的策略实验顺畅很多。先把地基打牢再往上盖房子。
