1. 为什么你需要一个量化交易数据仓库做量化交易的朋友都知道数据就是我们的弹药库。我刚开始做量化的时候最头疼的就是每天到处找数据今天从这个网站爬一点明天从那个API调一些最后发现数据格式不统一时间跨度不完整简直是一场噩梦。直到发现了akshare这个神器才算是真正解决了我的数据焦虑。akshare就像是一个金融数据的百宝箱股票、基金、期货、外汇你能想到的金融数据它基本都有。而且最良心的是它完全免费不像某些商业数据源动不动就收你几万块的年费。不过光会调用akshare的接口还不够要想长期稳定地做量化策略你需要建立一个属于自己的数据仓库。我自己的数据仓库是从2020年开始搭建的现在已经有将近4年的完整数据了。有了这个仓库回测新策略时再也不用临时抱佛脚去到处找数据所有历史数据都在本地随时调用。更重要的是你可以对数据进行统一的清洗和校验确保数据质量。接下来我就详细分享一下如何从零开始构建这样一个数据仓库。2. 搭建基础环境2.1 安装与配置akshare首先当然是安装akshare。我强烈建议使用Python虚拟环境来管理你的量化交易环境这样可以避免各种依赖冲突。下面是我的标准配置流程# 创建虚拟环境 python -m venv quant_env source quant_env/bin/activate # Linux/Mac # quant_env\Scripts\activate # Windows # 安装基础包 pip install akshare --upgrade pip install pandas numpy sqlalchemy这里有个小技巧akshare的更新比较频繁我建议每周都升级一次确保能用上最新的数据接口。我在2021年就遇到过因为akshare版本太旧导致某些接口无法使用的情况。2.2 数据库选择数据仓库需要一个存储后端我对比过几种方案CSV文件最简单但查询效率低不适合大数据量SQLite轻量级适合个人使用MySQL/PostgreSQL功能强大但需要额外维护MongoDB适合非结构化数据对于个人量化交易来说SQLite是个不错的选择。它不需要单独安装服务一个文件就是一个数据库备份也方便。下面是如何用SQLAlchemy连接SQLitefrom sqlalchemy import create_engine # 创建数据库引擎 engine create_engine(sqlite:///quant_data.db, echoFalse)3. 构建核心数据采集系统3.1 股票数据采集股票数据是量化交易的基础。akshare提供了非常全面的A股数据接口我常用的有这几个import akshare as ak import pandas as pd # 获取个股历史数据 def get_stock_daily(symbol, start_date, end_date): df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjusthfq # 后复权 ) # 统一列名 df.columns [date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover] return df # 示例获取贵州茅台最近一年的数据 df get_stock_daily(600519, 20230101, 20231231)这里有几个经验分享一定要用后复权数据(hfq)这样才能准确反映股价的真实变动字段名称最好统一成英文方便后续处理日期格式要标准化我习惯用YYYY-MM-DD3.2 基金数据采集除了股票基金数据也是很多量化策略的重要原料。特别是ETF数据可以用来做行业轮动等策略。# 获取ETF日线数据 def get_etf_daily(code, start_date, end_date): df ak.fund_etf_fund_info_em(code, start_date, end_date) df.columns [date, open, close, high, low, volume, amount, pct_change] return df # 示例获取沪深300ETF数据 df get_etf_daily(510300, 20230101, 20231231)4. 数据存储与管理4.1 设计数据库表结构有了数据采集函数下一步就是设计存储方案。我的经验是不同类型的金融数据最好分开存储。这是我的数据库表结构设计stock_daily存储股票日线数据symbol (文本)股票代码date (日期)交易日期open (浮点)开盘价close (浮点)收盘价...其他字段etf_daily存储ETF日线数据类似股票表结构index_daily存储指数数据# 将数据存入SQLite def save_to_db(df, table_name): df.to_sql(table_name, engine, if_existsappend, indexFalse)4.2 数据更新策略数据仓库最难的不是第一次采集而是持续更新。我设计了一个简单的更新策略每天收盘后运行更新脚本对于每只股票先查询数据库中最新日期只获取最新日期之后的数据将新数据追加到数据库def update_stock_data(symbol): # 查询该股票最新日期 latest_date pd.read_sql( fSELECT MAX(date) FROM stock_daily WHERE symbol{symbol}, engine ).iloc[0,0] if latest_date is None: # 全新股票 start_date 20000101 else: start_date (pd.to_datetime(latest_date) pd.Timedelta(days1)).strftime(%Y%m%d) # 获取新数据 new_data get_stock_daily(symbol, start_date, datetime.now().strftime(%Y%m%d)) if not new_data.empty: save_to_db(new_data, stock_daily)5. 数据质量保障5.1 数据校验金融数据经常会有各种问题比如缺失数据特别是停牌日的处理异常值价格突然暴涨暴跌节假日数据错误我通常会做这些校验def validate_data(df): # 检查是否有重复日期 if df[date].duplicated().any(): print(发现重复日期) # 检查价格是否合理 if (df[close] 0).any(): print(发现异常价格) # 检查成交量是否为0 if (df[volume] 0).any(): print(发现零成交量)5.2 数据备份策略数据仓库最怕的就是数据丢失。我的备份方案是每天自动备份数据库到本地另一个硬盘每周备份到云存储每月做一次完整备份可以使用简单的shell脚本实现自动备份#!/bin/bash DATE$(date %Y%m%d) cp quant_data.db /backup/quant_data_$DATE.db6. 进阶应用构建数据API当你的数据仓库越来越完善后可以把它封装成一个内部API方便其他程序调用。我用Flask实现了一个简单的数据服务from flask import Flask, jsonify import pandas as pd app Flask(__name__) app.route(/stock/symbol/start_date/end_date) def get_stock_data(symbol, start_date, end_date): query f SELECT * FROM stock_daily WHERE symbol{symbol} AND date BETWEEN {start_date} AND {end_date} df pd.read_sql(query, engine) return jsonify(df.to_dict(records)) if __name__ __main__: app.run(port5000)这样你就可以通过HTTP请求获取数据了http://localhost:5000/stock/600519/20230101/202312317. 实战建议与避坑指南在构建数据仓库的过程中我踩过不少坑这里分享几个关键经验数据频率选择日线数据对大多数策略已经足够高频数据不仅存储压力大而且容易遇到质量问题。除非做高频交易否则不必追求tick级数据。代码健壮性网络请求一定要有重试机制。我遇到过因为网络波动导致数据采集中断的情况后来加了自动重试后稳定多了。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def safe_get_stock_data(symbol, start_date, end_date): return get_stock_daily(symbol, start_date, end_date)存储优化当数据量很大时可以考虑按年份分表存储使用Parquet等列式存储格式对常用查询字段建立索引监控报警设置简单的监控当数据更新失败时能及时通知你。我用的是邮件提醒import smtplib from email.mime.text import MIMEText def send_alert(subject, content): msg MIMEText(content) msg[Subject] subject msg[From] your_emailexample.com msg[To] your_emailexample.com with smtplib.SMTP(smtp.example.com) as server: server.send_message(msg)构建一个完善的数据仓库可能需要几个月的时间但一旦建成你会发现量化研究和策略开发的效率会大幅提升。我现在回测一个新策略从数据准备到结果分析可能只需要几个小时而在以前光收集数据就要花上好几天。