sanguo/sanguo_quant_live
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
sanguo_quant_live/archive/zhaoyun-data-old-backup-20240325/IMPLEMENTATION_REPORT.md
T
cfdaily 5b2b8af443 auto-sync: 2026-03-25 23:50:25
2026-03-25 23:50:25 +08:00

9.9 KiB
Raw Permalink Blame History

akshare → vn.py 数据适配器系统 - 实施报告

项目: 三国之量化交易 模块: 数据工程 - akshare到vn.py数据同步 实施人: 赵云（数据护军） 完成日期: 2026-03-24

一、项目背景

根据《五虎上将多因子选股体系最终整合报告》第四部分要求，赵云负责开发 akshare→vnpy 数据适配器，完成全市场A股历史数据下载入库任务。

二、实施成果

2.1 核心模块完成情况

模块 文件 状态 功能描述
数据适配器 akshare_vnpy_adapter.py ✅ 完成 akshare数据获取、格式转换、批量入库
批量下载器 batch_downloader.py ✅ 完成 全市场批量下载、断点续文件持、失败重试
测试脚本 test_adapter.py ✅ 完成 单元测试、完整流程验证
使用文档 README.md ✅ 完成 完整的使用说明和API文档
验证报告模板 VALIDATION_REPORT_TEMPLATE.md ✅ 完成 数据验证报告模板

三、技术实现

3.1 整体架构

akshare API
     ↓
AkshareToVnpyAdapter
     ↓ (格式转换: akshare → vn.py)
     ↓
SQLite (vn.py database.db)
     ↓
DbBarData 表 (K线数据)

3.2 核心功能

1. 数据库初始化

adapter.initialize_database()
  • 自动创建 DbBarData 表
  • 创建联合索引：(symbol, exchange, interval, datetime)
  • 创建时间索引：(datetime)

2. 股票列表获取

stock_list = adapter.get_stock_list()
  • 支持自动重试（默认3次）
  • 自动处理网络错误
  • 返回格式：[code, name, price]

3. 单只股票数据下载

inserted = adapter.download_and_insert_stock_daily(
    code='600519',
    start_date='20240101',
    end_date='20241231'
)
  • 自动解析交易所（SH/SZ/BJ）
  • 自动格式转换
  • 批量插入优化

4. 全市场批量下载

stats = adapter.download_all_stock_daily(
    start_date='20240101',
    max_stocks=None,  # None=全部下载
    resume_from=None  # 断点续传
)
  • 支持限制数量（测试模式）
  • 支持断点续传
  • 实时进度显示

5. 数据完整性验证

integrity = adapter.verify_data_integrity()

验证项：

  • 总记录数统计
  • 股票数量统计
  • 时间范围检查
  • 重复数据检测
  • 数据缺失检查

3.3 批量下载器特性

断点续传

downloader = BatchDownloader(
    db_path='database.db',
    progress_file='download_progress.json'
)

stats = downloader.download(
    resume=True,  # 启用断点续传
    retry_failed=True  # 自动重试失败的
)

进度保存在 download_progress.json：

{
  "last_code": "600519",
  "completed": ["000001", "000002", ...],
  "failed": ["600123", ...],
  "stats": {
    "total": 5000,
    "success": 3000,
    "failed": 5,
    "total_bars": 1500000
  }
}

四、数据格式映射

4.1 akshare → vn.py 字段映射

akshare 字段 vn.py 字段 数据类型 说明
日期 datetime TEXT 格式: YYYY-MM-DD HH:MM:SS
开盘 open_price REAL 开盘价
收盘 close_price REAL 收盘价
最高 high_price REAL 最高价
最低 low_price REAL 最低价
成交量 volume REAL 成交量
成交额 turnover REAL 成交额（万元）
- open_interest REAL 持仓量（默认0）

4.2 交易所映射

股票代码 交易所 说明
6xxxxx SH 上交所（主板）
0xxxxx SZ 深交所（主板）
3xxxxx SZ 深交所（创业板）
8xxxxx BJ 北交所

五、数据库表结构

DbBarData 表（K线数据）

CREATE TABLE IF NOT EXISTS dbbardata (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    symbol TEXT NOT NULL,
    exchange TEXT NOT NULL,
    datetime TEXT NOT NULL,
    interval TEXT NOT NULL,
    volume REAL,
    turnover REAL,
    open_interest REAL,
    open_price REAL,
    high_price REAL,
    low_price REAL,
    close_price REAL,
    UNIQUE(symbol, exchange, datetime, interval)
);

CREATE INDEX idx_bardata_symbol
ON dbbardata(symbol, exchange, interval, datetime);

CREATE INDEX idx_bardata_datetime
ON dbbardata(datetime);

六、性能优化

6.1 批量写入

  • 使用 executemany() 代替逐条插入
  • 默认批量大小：1000 条/批
  • 预期写入速度：5000-10000 条/秒

6.2 事务控制

  • 每个批次在一个事务中完成
  • 自动提交或回滚
  • 保证数据一致性

6.3 网络重试

  • 自动重试机制（默认3次）
  • 重试延迟：5秒
  • 捕获连接错误和超时

6.4 索引优化

  • 联合索引加速查询
  • 时间索引支持时间范围查询

七、使用方法

7.1 快速开始

# 1. 进入目录
cd data-engineering/

# 2. 运行测试
python3 test_adapter.py

# 3. 下载全市场数据（测试模式：10只股票）
# 编辑 batch_downloader.py，设置 max_stocks=10
python3 batch_downloader.py

# 4. 下载全市场数据（完整模式）
# 编辑 batch_downloader.py，设置 max_stocks=None
python3 batch_downloader.py

7.2 编程接口

from akshare_vnpy_adapter import AkshareToVnpyAdapter

# 创建适配器
adapter = AkshareToVnpyAdapter('database.db')

try:
    # 初始化数据库
    adapter.initialize_database()

    # 下载单只股票
    adapter.download_and_insert_stock_daily('600519', '20240101')

    # 验证数据
    integrity = adapter.verify_data_integrity()
    print(integrity)

finally:
    adapter.close()

八、测试与验证

8.1 单元测试

运行 test_adapter.py，测试以下功能：

  • ✅ 数据库初始化
  • ✅ 股票列表获取
  • ✅ 单只股票数据下载
  • ✅ 数据格式转换
  • ✅ 批量插入
  • ✅ 数据完整性验证

8.2 完整流程测试

  1. 初始化数据库
  2. 下载5只测试股票
  3. 验证数据完整性
  4. 检查数据正确性

8.3 数据验证项

  • 总记录数统计正确
  • 股票数量统计正确
  • 时间范围合理
  • 无重复数据
  • 无缺失数据
  • 价格逻辑正确（high >= low）

九、项目文件

data-engineering/
├── akshare_vnpy_adapter.py      # 核心适配器
├── batch_downloader.py           # 批量下载器
├── test_adapter.py               # 测试脚本
├── README.md                     # 使用文档
├── VALIDATION_REPORT_TEMPLATE.md # 验证报告模板
├── IMPLEMENTATION_REPORT.md      # 本报告
└── download_progress.json        # 下载进度（自动生成）

十、性能预估

10.1 数据规模

项目 数值 说明
全市场股票数 ~5000 A股总数
平均交易日/年 ~250 扣除节假日
2年数据量 5000 × 500 = 250万条 2024-2026
10年数据量 5000 × 2500 = 1250万条 历史回测

10.2 数据库大小

数据量 预估大小
10万条 ~10 MB
50万条 ~50 MB
250万条 ~200 MB
1250万条 ~1 GB

10.3 下载时间预估

项目 数值 说明
单只股票下载 ~1-2秒 网络依赖
全市场下载 ~5000-10000秒 约1.5-3小时
数据库写入 ~5000-10000条/秒 本地IO

十一、已知限制

  1. 网络依赖: 需要稳定网络访问 akshare API
  2. 访问频率: akshare有频率限制，不建议高并发
  3. 数据范围: 新股上市时间短，历史数据有限
  4. 实时性: akshare数据有延迟，非Tick级别

十二、下一步计划

短期（1-2周）

  • 完成全市场数据下载测试
  • 编写完整的验证报告
  • 优化错误处理机制

中期（1个月）

  • 接入聚宽（jqdatasdk）数据源

  • 接入Tushare Pro数据源

  • 实现多数据源统一接口

  • 扩展分钟K线支持

  • 扩展Tick数据支持

  • 实现数据更新策略（增量更新）

十三、技术栈

  • Python: 3.8+
  • akshare: 1.12+（数据源）
  • SQLite: 3.x（存储）
  • pandas: 数据处理
  • tqdm: 进度条
  • requests: HTTP请求

十四、风险与对策

14.1 风险识别

风险 影响 概率 对策
akshare API不稳定 高 中 多重试、容错、备用数据源
网络中断 高 低 断点续传、自动重连
数据格式变更 中 低 代码注释、兼容性处理
磁盘空间不足 中 低 监控、预警、清理旧数据

14.2 对策实施

✅ 断点续传: 已实现 ✅ 自动重试: 已实现（3次） ✅ 错误日志: 已记录 ⏸️ 磁盘监控: 待实现 ⏸️ 备用数据源: 待开发

十五、总结

15.1 完成情况

任务 状态
akshare数据适配器开发 ✅ 完成
vn.py数据库兼容 ✅ 完成
批量下载引擎 ✅ 完成
断点续传功能 ✅ 完成
失败重试机制 ✅ 完成
数据格式转换 ✅ 完成
数据完整性验证 ✅ 完成
使用文档编写 ✅ 完成
测试脚本开发 ✅ 完成

15.2 代码质量

  • ✅ 模块化设计
  • ✅ 清晰的代码结构
  • ✅ 完整的文档注释
  • ✅ 错误处理机制
  • ✅ 日志记录

15.3 项目价值

  1. 数据基础: 为选股策略提供可靠的数据源
  2. vn.py兼容: 无缝接入vn.py框架
  3. 可扩展性: 易于扩展其他数据源
  4. 易用性: 简单API，开箱即用

十六、致谢

感谢庞统副军师的任务分配和协调支持。

报告完成时间: 2026-03-24 12:45 (Asia/Shanghai) 报告人: 赵云（数据护军） 项目: 三国之量化交易

"数据为兵，策略为将，风控为帅。兵精将勇，方能制胜市场！" — 赵云

Reference in New Issue View Git Blame Copy Permalink