1. 初识akshare与ETF动量策略第一次接触量化投资时我被各种复杂的数据获取和处理流程搞得晕头转向。直到发现了akshare这个宝藏库才真正体会到用Python玩转金融数据的乐趣。akshare就像是一个免费的数据管家能帮我们轻松获取股票、基金、ETF等各种金融数据省去了自己爬取数据的麻烦。ETF动量滚动策略听起来高大上其实原理很简单。就像我们逛超市会挑选最近销量最好的商品一样这个策略的核心思想就是定期买入近期表现最好的ETF卖出表现较差的ETF。实际操作中我习惯用周线数据来计算动量因为日线波动太大容易产生噪音而月线又显得过于迟钝。记得刚开始尝试时我犯了个典型错误——直接用日收益率排序选ETF。结果发现策略频繁调仓交易成本高得吓人。后来改用周线动量排序配合适当的持仓周期效果明显稳定多了。这也让我明白好的策略不仅要有正确的逻辑还需要考虑实际交易中的各种细节。2. 搭建你的数据获取管道2.1 安装与配置akshare工欲善其事必先利其器。使用akshare前建议创建一个干净的Python环境。我习惯用conda管理环境这样可以避免各种依赖冲突conda create -n etf_strategy python3.8 conda activate etf_strategy pip install akshare pandas numpy matplotlib安装完成后先测试下基础功能是否正常。获取ETF列表是最常用的功能之一akshare提供了非常简洁的接口import aksare as ak # 获取所有ETF基本信息 etf_list ak.fund_etf_fund_daily_em() print(etf_list.head())这个接口返回的DataFrame包含ETF代码、名称、最新净值等关键信息。我通常会先筛选流动性较好的ETF比如剔除日均成交额低于1000万的品种避免买卖时的冲击成本。2.2 高效获取历史数据获取单只ETF的历史数据很简单但批量获取时需要注意效率问题。我总结了几点经验控制请求频率避免被封IP合理设置时间范围不要一次性拉取过多数据对获取的数据做好本地缓存这里分享我的数据获取模板代码def get_etf_history(etf_code, start_date20100101, end_date20231231): try: df ak.fund_etf_fund_info_em( fundetf_code, start_datestart_date, end_dateend_date ) df.rename(columns{净值日期:date,日增长率:chg_pct}, inplaceTrue) return df[[date, chg_pct]] except Exception as e: print(f获取{etf_code}数据失败: {str(e)}) return None对于需要批量获取的场景建议使用线程池加速但要注意控制并发数量。我一般设置5个左右的线程既能提高效率又不会给服务器造成太大压力。3. 数据处理与特征工程3.1 日线转周线的正确姿势原始文章提供了日线转周线的函数但在实际使用中我发现几个可以优化的地方。首先是时区问题akshare返回的时间戳有时会带有时区信息直接处理可能会报错。其次是缺失值处理原函数只是简单丢弃可能会影响后续计算。这是我改进后的版本def daily_to_weekly(daily_df): 将日线数据转换为周线数据 参数: daily_df: 包含date和chg_pct列的DataFrame 返回: 转换后的周线DataFrame # 确保日期格式正确 df daily_df.copy() df[date] pd.to_datetime(df[date]).dt.tz_localize(None) # 设置日期为索引 df.set_index(date, inplaceTrue) # 计算周收益率(复利) weekly_df df[chg_pct].add(1).resample(W-FRI).prod().sub(1).to_frame() # 过滤无效数据 weekly_df weekly_df[weekly_df[chg_pct].notna()] weekly_df.reset_index(inplaceTrue) return weekly_df这个版本改用周五作为周线截止日更符合交易习惯。收益率计算采用复利方式比简单取最后一天的值更准确。我还添加了详细的文档字符串方便后续维护。3.2 动量指标的计算技巧动量策略的核心是准确计算各ETF的相对强弱。常见的计算方法有过去N周收益率过去N周收益率的排名过去N周收益率的Z-score标准化我比较推荐使用排名法因为它对极端值不敏感更能反映相对强弱。计算排名的代码很简单def calculate_rank(df, lookback4): 计算各ETF的动量排名 参数: df: 包含date, code, chg_pct的DataFrame lookback: 回溯周期(周) 返回: 包含排名结果的DataFrame # 计算过去N周累计收益 df[cum_ret] df.groupby(code)[chg_pct].transform( lambda x: x.add(1).rolling(lookback).prod().sub(1) ) # 按周计算排名 df[rank] df.groupby(date)[cum_ret].rank(ascendingFalse) return df实际应用中我发现4-8周的回看周期效果较好。太短容易受噪声影响太长则反应迟钝。这个参数需要根据市场环境适当调整牛市时可以缩短周期熊市则应该延长。4. 策略构建与回测4.1 策略逻辑实现有了处理好的数据策略实现就水到渠成了。我的基本思路是每周五收盘后计算各ETF的动量排名买入排名前20%的ETF持有固定周期后调仓具体代码实现如下def momentum_strategy(data, top_pct0.2, hold_weeks4): 动量滚动策略实现 参数: data: 处理好的周线数据 top_pct: 买入前百分之多少的ETF hold_weeks: 持有周期(周) 返回: 策略收益序列 # 计算排名 ranked_data calculate_rank(data) # 确定买入数量 etf_count len(ranked_data[code].unique()) buy_num max(1, int(etf_count * top_pct)) # 生成交易信号 ranked_data[signal] ranked_data[rank] buy_num # 计算策略收益 strategy_ret [] for week in ranked_data[date].unique(): week_data ranked_data[ranked_data[date] week] selected week_data[week_data[signal]] if len(selected) 0: avg_ret selected[chg_pct].mean() strategy_ret.append(avg_ret) # 转换为累计收益 cum_ret pd.Series(strategy_ret).add(1).cumprod().sub(1) return cum_ret这个基础版本还有很多优化空间比如可以加入止损机制或者对不同排名的ETF分配不同仓位。我在实际使用中会记录每次调仓的ETF列表方便后续分析策略的选股能力。4.2 回测要点与陷阱规避回测是检验策略的关键环节但也是最容易产生误导的地方。根据我的经验有以下几个常见陷阱需要注意前视偏差确保只用历史数据做决策幸存者偏差考虑已经退市的ETF交易成本合理估算买卖价差和手续费流动性限制避免买入成交量过小的ETF这是我常用的回测框架def backtest_strategy(etf_list, start_date, end_date): 完整回测流程 参数: etf_list: ETF代码列表 start_date: 回测开始日期 end_date: 回测结束日期 返回: 回测结果DataFrame # 获取历史数据 all_data [] for code in etf_list: data get_etf_history(code, start_date, end_date) if data is not None: data[code] code all_data.append(data) # 合并数据 full_data pd.concat(all_data) # 转换为周线 weekly_data [] for code in full_data[code].unique(): code_data full_data[full_data[code] code] weekly daily_to_weekly(code_data) weekly[code] code weekly_data.append(weekly) weekly_data pd.concat(weekly_data) # 运行策略 strategy_ret momentum_strategy(weekly_data) # 计算基准收益 benchmark ... # 对比分析 result pd.DataFrame({ strategy: strategy_ret, benchmark: benchmark }) return result回测完成后除了看最终收益率更应该关注最大回撤、夏普比率等风险指标。我习惯用pyfolio库进行专业的绩效分析它能生成各种详细的统计图表。5. 可视化与策略优化5.1 收益曲线的正确绘制原始文章中的绘图代码比较简略实际分析时需要更专业的可视化。matplotlib虽然功能强大但默认样式比较简陋。我通常会用seaborn进行美化并添加关键标注import seaborn as sns import matplotlib.pyplot as plt def plot_performance(result): 绘制策略与基准的对比曲线 参数: result: 包含strategy和benchmark列的DataFrame plt.figure(figsize(12, 6)) sns.set_style(whitegrid) # 绘制曲线 plt.plot(result.index, result[strategy], label动量策略, linewidth2) plt.plot(result.index, result[benchmark], label基准, linestyle--) # 添加标注 max_drawdown result[strategy].min() max_date result[strategy].idxmin() plt.annotate(f最大回撤: {max_drawdown:.1%}, xy(max_date, max_drawdown), xytext(10, 10), textcoordsoffset points) # 美化图表 plt.title(ETF动量策略表现, pad20) plt.xlabel(日期) plt.ylabel(累计收益) plt.legend() plt.tight_layout() return plt除了收益曲线还可以绘制月度收益热力图、滚动夏普比率等辅助图表。这些可视化能帮助我们更直观地发现策略在不同市场环境下的表现特点。5.2 参数优化与过拟合防范参数优化是策略开发的重要环节但也是最容易导致过拟合的陷阱。我的经验法则是使用不同时间段的数据进行参数测试保持参数数量尽可能少避免在优化时使用未来数据这里展示一个简单的网格搜索实现def parameter_search(data, lookback_range, top_pct_range): 参数网格搜索 参数: data: 处理好的周线数据 lookback_range: 回溯周期范围 top_pct_range: 买入比例范围 返回: 参数组合结果DataFrame results [] for lookback in lookback_range: for top_pct in top_pct_range: # 计算策略收益 ret momentum_strategy(data, lookback, top_pct) # 计算绩效指标 sharpe calculate_sharpe(ret) mdd calculate_max_drawdown(ret) results.append({ lookback: lookback, top_pct: top_pct, return: ret.iloc[-1], sharpe: sharpe, mdd: mdd }) return pd.DataFrame(results)执行优化后不要简单地选择历史表现最好的参数组合。我通常会选择参数高原区(performance plateau)的参数这些参数在邻近值表现也较好说明策略对这些参数不敏感未来更可能保持稳定。6. 实盘注意事项经过充分回测后如果策略表现令人满意就可以考虑小资金实盘测试了。根据我的实盘经验有几点特别需要注意交易时机严格按策略设定的时间点执行交易避免临时起意滑点控制对于流动性较差的ETF使用限价单而非市价单资金管理初期建议用不超过10%的资金测试新策略记录分析详细记录每笔交易的执行价格、数量、时间等信息实盘中最常见的问题是策略在回测中表现很好但实盘却不尽如人意。这通常是由于回测中忽略了一些现实约束。我的做法是先在模拟盘运行1-2个月确认各项流程顺畅后再投入真金白银。最后提醒一点任何策略都有其适应的市场环境。ETF动量策略在趋势行情中表现优异但在震荡市中可能会频繁止损。因此最好不要把所有资金押注在单一策略上构建多策略组合是更稳健的选择。
基于akshare的ETF动量滚动策略实战解析
1. 初识akshare与ETF动量策略第一次接触量化投资时我被各种复杂的数据获取和处理流程搞得晕头转向。直到发现了akshare这个宝藏库才真正体会到用Python玩转金融数据的乐趣。akshare就像是一个免费的数据管家能帮我们轻松获取股票、基金、ETF等各种金融数据省去了自己爬取数据的麻烦。ETF动量滚动策略听起来高大上其实原理很简单。就像我们逛超市会挑选最近销量最好的商品一样这个策略的核心思想就是定期买入近期表现最好的ETF卖出表现较差的ETF。实际操作中我习惯用周线数据来计算动量因为日线波动太大容易产生噪音而月线又显得过于迟钝。记得刚开始尝试时我犯了个典型错误——直接用日收益率排序选ETF。结果发现策略频繁调仓交易成本高得吓人。后来改用周线动量排序配合适当的持仓周期效果明显稳定多了。这也让我明白好的策略不仅要有正确的逻辑还需要考虑实际交易中的各种细节。2. 搭建你的数据获取管道2.1 安装与配置akshare工欲善其事必先利其器。使用akshare前建议创建一个干净的Python环境。我习惯用conda管理环境这样可以避免各种依赖冲突conda create -n etf_strategy python3.8 conda activate etf_strategy pip install akshare pandas numpy matplotlib安装完成后先测试下基础功能是否正常。获取ETF列表是最常用的功能之一akshare提供了非常简洁的接口import aksare as ak # 获取所有ETF基本信息 etf_list ak.fund_etf_fund_daily_em() print(etf_list.head())这个接口返回的DataFrame包含ETF代码、名称、最新净值等关键信息。我通常会先筛选流动性较好的ETF比如剔除日均成交额低于1000万的品种避免买卖时的冲击成本。2.2 高效获取历史数据获取单只ETF的历史数据很简单但批量获取时需要注意效率问题。我总结了几点经验控制请求频率避免被封IP合理设置时间范围不要一次性拉取过多数据对获取的数据做好本地缓存这里分享我的数据获取模板代码def get_etf_history(etf_code, start_date20100101, end_date20231231): try: df ak.fund_etf_fund_info_em( fundetf_code, start_datestart_date, end_dateend_date ) df.rename(columns{净值日期:date,日增长率:chg_pct}, inplaceTrue) return df[[date, chg_pct]] except Exception as e: print(f获取{etf_code}数据失败: {str(e)}) return None对于需要批量获取的场景建议使用线程池加速但要注意控制并发数量。我一般设置5个左右的线程既能提高效率又不会给服务器造成太大压力。3. 数据处理与特征工程3.1 日线转周线的正确姿势原始文章提供了日线转周线的函数但在实际使用中我发现几个可以优化的地方。首先是时区问题akshare返回的时间戳有时会带有时区信息直接处理可能会报错。其次是缺失值处理原函数只是简单丢弃可能会影响后续计算。这是我改进后的版本def daily_to_weekly(daily_df): 将日线数据转换为周线数据 参数: daily_df: 包含date和chg_pct列的DataFrame 返回: 转换后的周线DataFrame # 确保日期格式正确 df daily_df.copy() df[date] pd.to_datetime(df[date]).dt.tz_localize(None) # 设置日期为索引 df.set_index(date, inplaceTrue) # 计算周收益率(复利) weekly_df df[chg_pct].add(1).resample(W-FRI).prod().sub(1).to_frame() # 过滤无效数据 weekly_df weekly_df[weekly_df[chg_pct].notna()] weekly_df.reset_index(inplaceTrue) return weekly_df这个版本改用周五作为周线截止日更符合交易习惯。收益率计算采用复利方式比简单取最后一天的值更准确。我还添加了详细的文档字符串方便后续维护。3.2 动量指标的计算技巧动量策略的核心是准确计算各ETF的相对强弱。常见的计算方法有过去N周收益率过去N周收益率的排名过去N周收益率的Z-score标准化我比较推荐使用排名法因为它对极端值不敏感更能反映相对强弱。计算排名的代码很简单def calculate_rank(df, lookback4): 计算各ETF的动量排名 参数: df: 包含date, code, chg_pct的DataFrame lookback: 回溯周期(周) 返回: 包含排名结果的DataFrame # 计算过去N周累计收益 df[cum_ret] df.groupby(code)[chg_pct].transform( lambda x: x.add(1).rolling(lookback).prod().sub(1) ) # 按周计算排名 df[rank] df.groupby(date)[cum_ret].rank(ascendingFalse) return df实际应用中我发现4-8周的回看周期效果较好。太短容易受噪声影响太长则反应迟钝。这个参数需要根据市场环境适当调整牛市时可以缩短周期熊市则应该延长。4. 策略构建与回测4.1 策略逻辑实现有了处理好的数据策略实现就水到渠成了。我的基本思路是每周五收盘后计算各ETF的动量排名买入排名前20%的ETF持有固定周期后调仓具体代码实现如下def momentum_strategy(data, top_pct0.2, hold_weeks4): 动量滚动策略实现 参数: data: 处理好的周线数据 top_pct: 买入前百分之多少的ETF hold_weeks: 持有周期(周) 返回: 策略收益序列 # 计算排名 ranked_data calculate_rank(data) # 确定买入数量 etf_count len(ranked_data[code].unique()) buy_num max(1, int(etf_count * top_pct)) # 生成交易信号 ranked_data[signal] ranked_data[rank] buy_num # 计算策略收益 strategy_ret [] for week in ranked_data[date].unique(): week_data ranked_data[ranked_data[date] week] selected week_data[week_data[signal]] if len(selected) 0: avg_ret selected[chg_pct].mean() strategy_ret.append(avg_ret) # 转换为累计收益 cum_ret pd.Series(strategy_ret).add(1).cumprod().sub(1) return cum_ret这个基础版本还有很多优化空间比如可以加入止损机制或者对不同排名的ETF分配不同仓位。我在实际使用中会记录每次调仓的ETF列表方便后续分析策略的选股能力。4.2 回测要点与陷阱规避回测是检验策略的关键环节但也是最容易产生误导的地方。根据我的经验有以下几个常见陷阱需要注意前视偏差确保只用历史数据做决策幸存者偏差考虑已经退市的ETF交易成本合理估算买卖价差和手续费流动性限制避免买入成交量过小的ETF这是我常用的回测框架def backtest_strategy(etf_list, start_date, end_date): 完整回测流程 参数: etf_list: ETF代码列表 start_date: 回测开始日期 end_date: 回测结束日期 返回: 回测结果DataFrame # 获取历史数据 all_data [] for code in etf_list: data get_etf_history(code, start_date, end_date) if data is not None: data[code] code all_data.append(data) # 合并数据 full_data pd.concat(all_data) # 转换为周线 weekly_data [] for code in full_data[code].unique(): code_data full_data[full_data[code] code] weekly daily_to_weekly(code_data) weekly[code] code weekly_data.append(weekly) weekly_data pd.concat(weekly_data) # 运行策略 strategy_ret momentum_strategy(weekly_data) # 计算基准收益 benchmark ... # 对比分析 result pd.DataFrame({ strategy: strategy_ret, benchmark: benchmark }) return result回测完成后除了看最终收益率更应该关注最大回撤、夏普比率等风险指标。我习惯用pyfolio库进行专业的绩效分析它能生成各种详细的统计图表。5. 可视化与策略优化5.1 收益曲线的正确绘制原始文章中的绘图代码比较简略实际分析时需要更专业的可视化。matplotlib虽然功能强大但默认样式比较简陋。我通常会用seaborn进行美化并添加关键标注import seaborn as sns import matplotlib.pyplot as plt def plot_performance(result): 绘制策略与基准的对比曲线 参数: result: 包含strategy和benchmark列的DataFrame plt.figure(figsize(12, 6)) sns.set_style(whitegrid) # 绘制曲线 plt.plot(result.index, result[strategy], label动量策略, linewidth2) plt.plot(result.index, result[benchmark], label基准, linestyle--) # 添加标注 max_drawdown result[strategy].min() max_date result[strategy].idxmin() plt.annotate(f最大回撤: {max_drawdown:.1%}, xy(max_date, max_drawdown), xytext(10, 10), textcoordsoffset points) # 美化图表 plt.title(ETF动量策略表现, pad20) plt.xlabel(日期) plt.ylabel(累计收益) plt.legend() plt.tight_layout() return plt除了收益曲线还可以绘制月度收益热力图、滚动夏普比率等辅助图表。这些可视化能帮助我们更直观地发现策略在不同市场环境下的表现特点。5.2 参数优化与过拟合防范参数优化是策略开发的重要环节但也是最容易导致过拟合的陷阱。我的经验法则是使用不同时间段的数据进行参数测试保持参数数量尽可能少避免在优化时使用未来数据这里展示一个简单的网格搜索实现def parameter_search(data, lookback_range, top_pct_range): 参数网格搜索 参数: data: 处理好的周线数据 lookback_range: 回溯周期范围 top_pct_range: 买入比例范围 返回: 参数组合结果DataFrame results [] for lookback in lookback_range: for top_pct in top_pct_range: # 计算策略收益 ret momentum_strategy(data, lookback, top_pct) # 计算绩效指标 sharpe calculate_sharpe(ret) mdd calculate_max_drawdown(ret) results.append({ lookback: lookback, top_pct: top_pct, return: ret.iloc[-1], sharpe: sharpe, mdd: mdd }) return pd.DataFrame(results)执行优化后不要简单地选择历史表现最好的参数组合。我通常会选择参数高原区(performance plateau)的参数这些参数在邻近值表现也较好说明策略对这些参数不敏感未来更可能保持稳定。6. 实盘注意事项经过充分回测后如果策略表现令人满意就可以考虑小资金实盘测试了。根据我的实盘经验有几点特别需要注意交易时机严格按策略设定的时间点执行交易避免临时起意滑点控制对于流动性较差的ETF使用限价单而非市价单资金管理初期建议用不超过10%的资金测试新策略记录分析详细记录每笔交易的执行价格、数量、时间等信息实盘中最常见的问题是策略在回测中表现很好但实盘却不尽如人意。这通常是由于回测中忽略了一些现实约束。我的做法是先在模拟盘运行1-2个月确认各项流程顺畅后再投入真金白银。最后提醒一点任何策略都有其适应的市场环境。ETF动量策略在趋势行情中表现优异但在震荡市中可能会频繁止损。因此最好不要把所有资金押注在单一策略上构建多策略组合是更稳健的选择。
