A 股分钟级量化数据集成系统 -- 专为中国 A 股设计的分钟级数据获取与预处理工具。
通过适配量化 SaaS 平台（当前支持 JoinQuant），解决动态指数成分股追踪、分钟级高频数据下载以及 Qlib 机器学习框架格式转换的痛点，为研究员提供开箱即用的高质量实验数据。
┌──────────────────────────────────────────────────────────┐
│ CLI / Config │
├───────────┬──────────────────┬────────────────────────────┤
│ Provider │ Universe │ Pipeline │
│ 数据适配 │ 动态股票池 │ 处理与转换 │
│ │ │ │
│ • JQ SDK │ • 指数成分股追踪 │ • 逐日下载 (配额管理) │
│ • 代码归一 │ • 时序还原 │ • 数据清洗 (FFill) │
│ • 限流器 │ • 并集/快照模式 │ • 前复权计算 │
│ • 账号管理 │ │ • 增量合并 │
│ • 配额跟踪 │ │ • Qlib .bin 转换 │
│ │ │ • 完整性校验 (240根/日) │
├───────────┴──────────────────┴────────────────────────────┤
│ Storage / Monitoring / Utils │
│ Parquet 存储 · JSON 索引 · Loguru 中文日志 · Rich 报告 │
│ QuotaTracker 配额跟踪 · 进度估算 · 自动等待次日配额重置 │
└──────────────────────────────────────────────────────────┘
qt_data_loader/
├── src/qt_data_loader/
│ ├── config/ # Pydantic 配置模型 + YAML 加载器
│ ├── provider/ # 数据源抽象基类 + JoinQuant 适配器
│ │ ├── base.py # BaseProvider ABC (6 个抽象方法)
│ │ ├── joinquant.py # JoinQuantProvider (jqdatasdk 封装)
│ │ ├── code_mapper.py # 股票代码归一化 (XSHE/SZ/SZ000001)
│ │ └── rate_limiter.py # 令牌桶限流器
│ ├── universe/ # 动态指数成分股管理
│ ├── pipeline/ # 下载 → 清洗 → 复权 → 合并 → 转换 → 校验
│ ├── storage/ # 本地索引 + 路径管理
│ ├── monitoring/ # 日志 + 报告 + QuotaTracker 配额跟踪
│ └── utils/ # 日期工具
├── tests/ # 237 个单元测试 + 集成测试 (覆盖率 95%)
├── config.example.yaml # 配置模板
├── Dockerfile # 容器化部署
└── docker-compose.yml # 编排配置
方式一：Conda（推荐）
conda create -n qt_data_loader python=3.11 -y
conda activate qt_data_loader
pip install -e ".[dev]"方式二：pip
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
pip install -e .
jqdatasdk已包含在项目依赖中，安装时会自动拉取。
cp config.example.yaml config.yaml编辑 config.yaml，填入 JoinQuant 凭证和参数：
provider:
name: "joinquant"
joinquant:
username: "your_phone" # 或使用 ${JQ_USERNAME} 环境变量
password: "your_password" # 或使用 ${JQ_PASSWORD} 环境变量
account_type: "trial" # trial=试用账号 | pro=正式账号
universe:
index_codes:
- "000300.XSHG" # 沪深300
mode: "union" # union=历史并集 | snapshot=当日快照
date_range:
start_date: "2025-01-15" # 试用账号注意数据范围限制
end_date: "2026-01-15"
download:
frequency: "1m" # 1m | 5m | 60m
fields: # JQData 全部行情字段 (默认 12 个)
- open # 开盘价
- close # 收盘价
- high # 最高价
- low # 最低价
- volume # 成交量
- money # 成交额
- factor # 复权因子
- high_limit # 涨停价
- low_limit # 跌停价
- avg # 平均价
- paused # 是否停牌
- pre_close # 前收盘价
adjust_type: "none" # none=不复权 | pre=前复权 | post=后复权支持通过环境变量注入敏感信息：
export JQ_USERNAME="13141200430"
export JQ_PASSWORD="your_password"# 查看账号信息和可下载时间范围
qt-data-loader info -c config.yaml
# 全量下载（自动逐日下载 + 配额管理 + 断点续传）
qt-data-loader download -c config.yaml
# 后处理：对已下载数据执行复权计算 + Qlib 格式转换（不重新下载）
qt-data-loader postprocess -c config.yaml
# 增量更新（仅下载缺失数据，自动检测除权事件）
qt-data-loader update -c config.yaml
# 校验本地数据完整性
qt-data-loader validate -c config.yaml典型工作流：先 download 下载全部分钟数据（试用账号需多天），完成后执行 postprocess 计算复权并生成 Qlib 格式。两个阶段均支持断点续传和配额自动等待。
系统根据 account_type 自动适配数据范围、限流策略和配额管理：
| 特性 | 试用账号 (trial) |
正式账号 (pro) |
|---|---|---|
| 历史数据范围 | 前15个月 ~ 前3个月 | 不限制 |
| 每日流量 | 100 万条 | 20000 万条 |
| 默认限流 | 3 requests/s | 10 requests/s |
| 沪深300全量下载 | ~18 个自然日 | < 1 天 |
- 连接时自动查询并显示剩余配额
- 试用账号请求超出范围的日期时，自动裁剪并输出警告
- 到达每日配额上限时，自动等待至次日 00:05 后继续下载
JQData 试用申请：https://www.joinquant.com/help/api/doc?name=logon&id=9830
系统下载 JQData 返回的全部 12 个行情字段：
| 字段 | 说明 | 字段 | 说明 |
|---|---|---|---|
open |
开盘价 | money |
成交额 |
close |
收盘价 | factor |
复权因子 |
high |
最高价 | high_limit |
涨停价 |
low |
最低价 | low_limit |
跌停价 |
volume |
成交量 | avg |
平均价 |
paused |
是否停牌 | pre_close |
前收盘价 |
系统在两个阶段均支持断点续传，随时可以 Ctrl+C 中断，重新执行会从断点继续：
- download 阶段：已完成的交易日记录在
data/download_progress.json，重启后自动跳过，不浪费配额 - postprocess 阶段：复权计算和 Qlib 转换可独立重复执行，幂等操作
系统内置 QuotaTracker，在下载过程中实时跟踪数据量消耗：
2026-04-15 10:30:00 | INFO | 已连接聚宽 (trial账号) | 剩余配额: 1,000,000 条
2026-04-15 10:30:01 | INFO | 可下载数据范围: 2025-01-15 ~ 2026-01-15
2026-04-15 10:30:02 | INFO | 开始下载 | 交易日: 250天 | 预估总数据量: 18,000,000 条 | 预计需要: 19 个自然日
2026-04-15 10:30:10 | INFO | [1/250天] 2025-02-03 | 成分股: 298只 | 成功: 298 失败: 0 |
累计: 71,520/18,000,000 (0.4%) | 今日剩余配额: 928,480
...
2026-04-15 18:20:00 | WARNING | 今日配额已用完 (1,000,000/1,000,000 条) | 等待至明日 00:05 (约5小时45分钟)...
2026-04-16 00:05:00 | INFO | 新的一天开始，配额已重置，继续下载...
- 支持沪深 300、中证 500、中证 800 等指数追踪
- 并集模式 (union): 获取区间内曾经入选过的所有股票，规避幸存者偏差
- 快照模式 (snapshot): 获取指定日期当天的成分股
- 逐日获取成分股列表，精确追踪每日的真实构成
- 基于本地 JSON 索引记录每只股票的最后有效时间戳
- 启动时自动对比本地与服务器时间，仅下载缺失片段
- 检测到除权除息事件时，自动触发受影响股票的全量重下载
输出完全兼容 Qlib 框架，全部 12 个字段均转为 .bin 文件：
data/qlib_1min/
├── calendars/
│ └── 1min.txt # 分钟级交易日历
├── instruments/
│ └── all.txt # 股票列表 + 起止日期
└── features/
├── SZ000001/
│ ├── open.1min.bin # little-endian float32
│ ├── close.1min.bin
│ ├── high.1min.bin
│ ├── low.1min.bin
│ ├── volume.1min.bin
│ ├── money.1min.bin
│ ├── factor.1min.bin
│ ├── high_limit.1min.bin
│ ├── low_limit.1min.bin
│ ├── avg.1min.bin
│ ├── paused.1min.bin
│ └── pre_close.1min.bin
└── SH600000/
└── ...
转换后即可在 Qlib 中直接使用：
from qlib.data import D
data = D.features(instruments='csi300', fields=['$close', '$high_limit', '$avg'], freq='1min')- 完整性校验: 每个交易日必须有 240 根分钟线
- OHLC 一致性:
high >= max(open, close),low <= min(open, close) - 缺失填充: FFill 补全非交易时段缺失值
- 停牌对齐: 价格前值填充，成交量归零
- 异常值处理: 自动修正零价格和负价格
# 构建镜像
docker build -t qt-data-loader .
# 运行（挂载数据目录 + 注入凭证）
docker run \
-e JQ_USERNAME=13141200430 \
-e JQ_PASSWORD=your_password \
-v $(pwd)/data:/app/data \
-v $(pwd)/config.yaml:/app/config.yaml:ro \
qt-data-loader download
# 或使用 docker-compose
docker-compose up# 运行全部测试
pytest tests/ -v
# 运行测试并查看覆盖率
pytest tests/ --cov=qt_data_loader --cov-report=term-missing
# 仅运行单元测试
pytest tests/unit/ -v
# 仅运行某个模块的测试
pytest tests/unit/pipeline/ -v当前状态：237 个测试全部通过，代码覆盖率 95%
完整配置项见 config.example.yaml，关键参数说明：
| 配置项 | 说明 | 默认值 |
|---|---|---|
provider.joinquant.account_type |
trial 试用 / pro 正式 |
trial |
universe.mode |
union 历史并集 / snapshot 当日快照 |
union |
download.frequency |
分钟线频率：1m / 5m / 60m |
1m |
download.fields |
JQData 行情字段列表 | 全部 12 个字段 |
download.adjust_type |
复权方式：none / pre / post |
none |
resilience.rate_limit.calls_per_second |
API 每秒最大请求数 | 3 (trial) / 10 (pro) |
resilience.retry.max_attempts |
单股下载最大重试次数 | 3 |
logging.level |
日志级别 | INFO |
dashboard.output |
报告输出：terminal / json / both |
both |
实现 BaseProvider 的 6 个抽象方法即可接入新数据源：
from qt_data_loader.provider.base import BaseProvider
class MyProvider(BaseProvider):
def connect(self) -> None: ...
def disconnect(self) -> None: ...
def get_minute_bars(self, order_book_id, start_date, end_date, ...) -> pd.DataFrame: ...
def get_index_components(self, index_code, date) -> list[str]: ...
def get_trading_dates(self, start_date, end_date) -> list[date]: ...
def get_ex_factor(self, order_book_id, start_date, end_date) -> pd.DataFrame: ...| 组件 | 技术 |
|---|---|
| 数据源 | JoinQuant (jqdatasdk>=1.9) |
| 配置管理 | Pydantic v2 + PyYAML |
| 数据处理 | Pandas + NumPy + PyArrow |
| CLI | Click |
| 日志 | Loguru (全中文输出) |
| 容错 | 令牌桶限流 + 指数退避重试 + 配额自动等待 |
| 终端展示 | Rich |
| 测试 | pytest + pytest-cov + pytest-mock |
| 容器化 | Docker + docker-compose |