Python量化入门：用pytdx获取800个交易日K线数据的完整指南（附代码）

Python量化入门：从零构建你的本地股票K线数据库

如果你刚开始接触量化交易，可能会被各种数据源、API接口和复杂的代码搞得晕头转向。市面上的数据要么收费昂贵，要么格式混乱，对于想快速验证策略想法的初学者来说，门槛实在不低。今天，我想分享一个我自己在早期项目中常用的方法：利用Python的pytdx库，直接从通达信行情服务器获取日线级别的K线数据。这个方法最大的好处是免费、稳定、数据质量相对可靠，特别适合个人研究和小型策略回测。

你不需要是编程高手，只要会基本的Python语法，就能跟着这篇文章一步步搭建起属于自己的股票数据获取管道。我们将从最基础的环境配置讲起，涵盖单只股票到批量获取的完整流程，并深入探讨数据清洗、存储优化以及实际应用中可能遇到的坑。最终，你将能获得一个包含近800个交易日历史数据的本地数据库，为后续的指标计算和策略回测打下坚实基础。

1. 环境搭建与核心工具初识

在开始写代码之前，我们需要一个干净、可复现的Python环境。我强烈建议使用conda或venv创建独立的虚拟环境，这能避免不同项目间的库版本冲突。假设你已经安装了Python（3.7及以上版本），那么打开终端，跟着下面的步骤操作。

首先，创建并激活一个名为quant_env的虚拟环境：

# 使用 conda
conda create -n quant_env python=3.9
conda activate quant_env

# 或者使用 venv
python -m venv quant_env
# 在Windows上激活
quant_env\\Scripts\\activate
# 在Mac/Linux上激活
source quant_env/bin/activate

环境激活后，安装我们所需的两个核心库：

pip install pytdx pandas

• pytdx: 这是我们与通达信服务器通信的桥梁。它是一个非官方的开源库，封装了通达信客户端的通信协议，让我们能用Python直接请求行情数据。
• pandas: 数据分析的瑞士军刀。获取到的原始数据是列表或元组形式，pandas的DataFrame结构能让我们像操作Excel表格一样轻松地进行筛选、计算和保存。

提示：如果pip install pytdx速度慢或失败，可以尝试使用国内的镜像源，例如 pip install pytdx -i https://pypi.tuna.tsinghua.edu.cn/simple。

安装完成后，可以在Python交互环境中简单测试一下：

import pytdx
import pandas as pd
print(pytdx.__version__, pd.__version__)

如果能看到版本号输出，说明环境配置成功。接下来，我们理解一下pytdx的工作原理。它本质上是一个“客户端模拟器”，通过TCP协议连接到通达信公开的行情服务器，发送特定的指令包来请求数据。这些服务器是面向公众的，所以我们可以免费使用，但需要注意请求频率，避免给服务器造成过大压力。

2. 建立连接与获取第一份K线数据

万事开头难，但获取第一份数据其实只需要几行代码。我们先从一个最简单的脚本开始，目标是获取上证指数（000001）最近800个交易日的日K线。

创建一个新的Python文件，比如get_data_demo.py，写入以下内容：

import os
from pytdx.hq import TdxHq_API
import pandas as pd

# 1. 创建数据保存目录
data_dir = \’./stock_data\’
os.makedirs(data_dir, exist_ok=True)

# 2. 初始化API对象
api = TdxHq_API()

# 3. 连接服务器并获取数据
try:
# 使用with语句确保连接正确关闭
with api.connect(\’124.71.163.106\’, 7709):
print(\”连接服务器成功…\”)

# 核心函数：get_security_bars
# 参数详解：
# 9 -> K线周期：9代表日线
# 1 -> 市场代码：1代表上海市场（0为深圳市场）
# \’000001\’ -> 股票代码
# 0 -> 起始位置（从最近的第0条开始取）
# 800 -> 获取的数量
data = api.get_security_bars(9, 1, \’000001\’, 0, 800)

if data:
print(f\”成功获取到 {len(data)} 条K线数据\”)
# 4. 转换为DataFrame
df = pd.DataFrame(
data,
columns=[\’datetime\’, \’open\’, \’high\’, \’low\’, \’close\’, \’amount\’, \’volume\’]
)
# 转换时间戳
df[\’datetime\’] = pd.to_datetime(df[\’datetime\’])

# 5. 保存为CSV文件
file_path = os.path.join(data_dir, \’sh000001.csv\’)
df.to_csv(file_path, index=False)
print(f\”数据已保存至: {file_path}\”)
print(df.head()) # 预览前5行数据
else:
print(\”未获取到数据，请检查股票代码或网络连接。\”)

except Exception as e:
print(f\”程序执行出错: {e}\”)

运行这个脚本，如果一切顺利，你会在当前目录下看到一个stock_data文件夹，里面有一个sh000001.csv文件。用Excel或文本编辑器打开，你会看到类似这样的内容：

这里有几个关键点需要解释：

• 市场代码判断：上海市场的股票代码通常以6、9开头（如600000，688111），而深圳市场以0、3开头（如000001，300750）。上证指数000001是个特例，它属于上海市场，所以市场代码用了1。
• K线周期代码：get_security_bars的第一个参数决定了数据粒度。除了9（日线），其他常用值包括：
  • 1: 1分钟线
  • 2: 5分钟线
  • 3: 15分钟线
  • 4: 30分钟线
  • 5: 60分钟线
  • 7: 周线
  • 8: 月线
• 数据字段：返回的7个字段分别对应时间、开盘价、最高价、最低价、收盘价、成交额（元）、成交量（股）。其中成交额和成交量单位较大，在分析时需要注意。

注意：服务器地址 124.71.163.106:7709 是一个常用的公共服务器，但并非永久稳定。如果连接超时或失败，可以尝试替换为其他已知的服务器IP，例如 119.147.212.81:7709 或 106.14.95.149:7709。pytdx库的源码或相关社区中常会维护一个可用的服务器列表。

3. 构建健壮的多股票数据批量下载器

只下载一只股票的数据显然不够。一个实用的量化数据模块，需要能够稳定、高效地批量获取数百甚至数千只股票的历史数据。我们需要解决几个问题：如何自动识别市场？如何优雅地处理网络异常？如何避免因请求过快被封IP？

下面，我们来构建一个更健壮的批量下载类。我们将它封装在一个类中，提高代码的可重用性和可配置性。

import os
import time
import pandas as pd
from pytdx.hq import TdxHq_API
from typing import List, Optional

class StockDataDownloader:
\”\”\”
股票历史K线数据批量下载器
\”\”\”

def __init__(self, data_dir: str = \’./stock_data\’, server: str = \’124.71.163.106\’, port: int = 7709):
\”\”\”
初始化下载器

参数:
data_dir: 数据保存目录
server: 通达信行情服务器地址
port: 服务器端口
\”\”\”
self.data_dir = data_dir
self.server = server
self.port = port
self.api = TdxHq_API()

# 创建数据目录
os.makedirs(self.data_dir, exist_ok=True)

# 市场代码映射规则
self.market_rules = {
\’SH\’: [\’6\’, \’9\’, \’5\’], # 上海主板、科创板、基金
\’SZ\’: [\’0\’, \’3\’, \’2\’] # 深圳主板、创业板、B股（仅作示例，B股代码特殊）
}

def _get_market_code(self, stock_code: str) -> int:
\”\”\”
根据股票代码判断市场

返回:
0: 深圳市场
1: 上海市场
\”\”\”
prefix = stock_code[0]
for market, prefixes in self.market_rules.items():
if prefix in prefixes:
return 1 if market == \’SH\’ else 0
# 默认处理：000001（上证指数）等特殊代码
if stock_code in [\’000001\’]:
return 1 # 上海市场
return 0 # 默认深圳市场，但实际应记录未知代码

def download_single_stock(self, stock_code: str, kline_type: int = 9, limit: int = 80