量化交易实战：基于vnpy与Tushare的股票多周期数据获取与处理

1. 为什么需要多周期股票数据做量化交易的朋友都知道数据就是策略的粮食。就像厨师做菜需要新鲜的食材一样我们需要高质量、多周期的股票数据来开发和测试交易策略。很多新手刚开始接触量化时往往只关注日线数据这就像只用一种调料做菜很难做出丰富的味道。我在实际开发中遇到过这样的情况一个基于日线的突破策略回测表现很好但实盘却频频失效。后来发现是因为没有考虑周线和月线的整体趋势。这就好比开车只看眼前几米的路况很容易错过重要的转弯提示。vnpy作为国内知名的量化交易框架提供了完善的数据处理模块。而Tushare则是获取A股数据的利器两者结合可以构建稳定可靠的数据获取管道。我实测下来这套组合特别适合个人和小型机构使用既避免了自建数据采集系统的复杂性又能获得足够丰富的数据维度。2. 环境准备与基础配置2.1 安装必备工具包首先需要准备好Python环境建议使用3.7以上版本。我习惯用conda创建独立的虚拟环境避免包冲突conda create -n vnpy python3.8 conda activate vnpy然后安装核心依赖pip install vnpy tushare pandas这里有个小坑要注意Tushare的pro接口需要token认证记得先去官网注册账号获取API密钥。我建议把token保存在环境变量中不要硬编码在代码里import os import tushare as ts os.environ[TUSHARE_TOKEN] 你的token ts.set_token(os.getenv(TUSHARE_TOKEN))2.2 vnpy基础配置vnpy的数据模块需要做一些初始化设置。我通常会在项目根目录创建config.py文件存放这些配置from vnpy.trader.constant import Exchange, Interval # 交易所映射配置 EXCHANGE_MAPPING { SSE: Exchange.SSE, # 上交所 SZSE: Exchange.SZSE # 深交所 } # 周期映射配置 INTERVAL_MAPPING { daily: Interval.DAILY, weekly: Interval.WEEKLY, monthly: Interval.MONTHLY }3. 多周期数据获取实战3.1 封装Tushare数据接口原始代码已经给出了很好的基础实现我在实际使用中做了些优化。比如增加异常处理和缓存机制from datetime import datetime import pandas as pd from vnpy.trader.object import BarData class TushareDataFetcher: def __init__(self, token): self.pro ts.pro_api(token) self.cache {} # 简单内存缓存 def get_bars(self, symbol, exchange, interval, start_dateNone, end_dateNone, days20): cache_key f{symbol}_{exchange}_{interval} if cache_key in self.cache: return self.cache[cache_key] ts_code self._convert_symbol(symbol, exchange) if interval Interval.WEEKLY: df self.pro.weekly(ts_codets_code) elif interval Interval.MONTHLY: df self.pro.monthly(ts_codets_code) else: df self.pro.daily(ts_codets_code) bars self._convert_to_bardata(df, symbol, exchange, interval) self.cache[cache_key] bars return bars def _convert_symbol(self, symbol, exchange): # 转换股票代码格式 return f{symbol}.{SH if exchange Exchange.SSE else SZ} def _convert_to_bardata(self, df, symbol, exchange, interval): bars [] for _, row in df.iterrows(): bar BarData( symbolsymbol, exchangeexchange, datetimedatetime.strptime(row[trade_date], %Y%m%d), intervalinterval, open_pricerow[open], high_pricerow[high], low_pricerow[low], close_pricerow[close], volumerow[vol], gateway_nameTushare ) bars.append(bar) return bars3.2 多周期数据同步技巧获取不同周期数据时时间对齐是个常见问题。我的经验是统一以交易日历为基准def sync_multiple_periods(fetcher, symbol, exchange): # 获取交易日历 trade_dates fetcher.pro.trade_cal(exchangeSSE, start_date20230101) trade_dates trade_dates[trade_dates[is_open] 1] # 初始化数据容器 data { daily: [], weekly: [], monthly: [] } # 获取日线数据 daily_bars fetcher.get_bars(symbol, exchange, Interval.DAILY) data[daily] daily_bars # 获取周线数据每周五收盘 weekly_dates trade_dates[trade_dates[cal_date].apply( lambda x: datetime.strptime(x, %Y%m%d).weekday() 4)] weekly_bars [] for date in weekly_dates[cal_date]: dt datetime.strptime(date, %Y%m%d) weekly_bars.extend(fetcher.get_bars( symbol, exchange, Interval.WEEKLY, end_datedt.strftime(%Y%m%d), days1)) data[weekly] weekly_bars # 获取月线数据每月最后交易日 monthly_dates trade_dates.groupby( trade_dates[cal_date].str[:6]).last()[cal_date] monthly_bars [] for date in monthly_dates: dt datetime.strptime(date, %Y%m%d) monthly_bars.extend(fetcher.get_bars( symbol, exchange, Interval.MONTHLY, end_datedt.strftime(%Y%m%d), days1)) data[monthly] monthly_bars return data4. 数据处理与质量控制4.1 数据清洗实战从Tushare获取的原始数据需要经过清洗才能使用。我总结了几个常见问题及处理方法缺失值处理遇到停牌等情况会有缺失数据def handle_missing_data(bars): # 前向填充 df pd.DataFrame([bar.__dict__ for bar in bars]) df.fillna(methodffill, inplaceTrue) # 重新生成BarData对象 clean_bars [] for _, row in df.iterrows(): bar BarData(**row) clean_bars.append(bar) return clean_bars异常值检测价格突然跳变可能是数据错误def detect_outliers(bars, threshold0.2): prices [bar.close_price for bar in bars] returns np.diff(prices) / prices[:-1] outliers np.where(np.abs(returns) threshold)[0] return outliers4.2 数据存储优化对于大量历史数据我推荐使用vnpy的MongoDB存储from vnpy.trader.database import database_manager def save_to_database(bars): # 批量插入 database_manager.save_bar_data(bars) # 创建索引加速查询 database_manager.db.client[vnpy][bar_data].create_index([ (symbol, 1), (interval, 1), (datetime, -1) ])5. 实战案例双均线策略数据准备以经典的双均线策略为例展示如何使用多周期数据def prepare_ma_strategy_data(symbol, exchange): fetcher TushareDataFetcher(os.getenv(TUSHARE_TOKEN)) # 获取1年数据 end_date datetime.now().strftime(%Y%m%d) start_date (datetime.now() - timedelta(days365)).strftime(%Y%m%d) # 日线数据短期均线 daily_bars fetcher.get_bars( symbol, exchange, Interval.DAILY, start_datestart_date, end_dateend_date) # 周线数据长期均线 weekly_bars fetcher.get_bars( symbol, exchange, Interval.WEEKLY, start_datestart_date, end_dateend_date) # 转换为DataFrame daily_df pd.DataFrame([bar.__dict__ for bar in daily_bars]) weekly_df pd.DataFrame([bar.__dict__ for bar in weekly_bars]) # 计算均线 daily_df[ma5] daily_df[close_price].rolling(5).mean() weekly_df[ma20] weekly_df[close_price].rolling(20).mean() return daily_df, weekly_df这个案例中我们同时使用日线和周线数据日线的5日均线作为短期信号周线的20日均线判断长期趋势。实测下来这种多周期配合的策略比单周期策略稳定很多。