基于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 python=3.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的历史数据很简单，但批量获取时需要注意效率问题。我总结了几点经验：

1. 控制请求频率，避免被封IP
2. 合理设置时间范围，不要一次性拉取过多数据
3. 对获取的数据做好本地缓存

这里分享我的数据获取模板代码：

def get_etf_history(etf_code, start_date="20100101", end_date="20231231"): try: df = ak.fund_etf_fund_info_em( fund=etf_code, start_date=start_date, end_date=end_date ) df.rename(columns={'净值日期':'date','日增长率':'chg_pct'}, inplace=True) 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', inplace=True) # 计算周收益率(复利) 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(inplace=True) return weekly_df

这个版本改用周五作为周线截止日，更符合交易习惯。收益率计算采用复利方式，比简单取最后一天的值更准确。我还添加了详细的文档字符串，方便后续维护。

3.2 动量指标的计算技巧

动量策略的核心是准确计算各ETF的相对强弱。常见的计算方法有：

1. 过去N周收益率
2. 过去N周收益率的排名
3. 过去N周收益率的Z-score标准化

我比较推荐使用排名法，因为它对极端值不敏感，更能反映相对强弱。计算排名的代码很简单：

def calculate_rank(df, lookback=4): """ 计算各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(ascending=False) return df

实际应用中，我发现4-8周的回看周期效果较好。太短容易受噪声影响，太长则反应迟钝。这个参数需要根据市场环境适当调整，牛市时可以缩短周期，熊市则应该延长。

4. 策略构建与回测

4.1 策略逻辑实现

有了处理好的数据，策略实现就水到渠成了。我的基本思路是：

1. 每周五收盘后计算各ETF的动量排名
2. 买入排名前20%的ETF
3. 持有固定周期后调仓

具体代码实现如下：

def momentum_strategy(data, top_pct=0.2, hold_weeks=4): """ 动量滚动策略实现 参数: 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 回测要点与陷阱规避

回测是检验策略的关键环节，但也是最容易产生误导的地方。根据我的经验，有以下几个常见陷阱需要注意：

1. 前视偏差：确保只用历史数据做决策
2. 幸存者偏差：考虑已经退市的ETF
3. 交易成本：合理估算买卖价差和手续费
4. 流动性限制：避免买入成交量过小的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='动量策略', linewidth=2) 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), textcoords='offset points') # 美化图表 plt.title('ETF动量策略表现', pad=20) plt.xlabel('日期') plt.ylabel('累计收益') plt.legend() plt.tight_layout() return plt

除了收益曲线，还可以绘制月度收益热力图、滚动夏普比率等辅助图表。这些可视化能帮助我们更直观地发现策略在不同市场环境下的表现特点。

5.2 参数优化与过拟合防范

参数优化是策略开发的重要环节，但也是最容易导致过拟合的陷阱。我的经验法则是：

1. 使用不同时间段的数据进行参数测试
2. 保持参数数量尽可能少
3. 避免在优化时使用未来数据

这里展示一个简单的网格搜索实现：

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. 实盘注意事项

经过充分回测后，如果策略表现令人满意，就可以考虑小资金实盘测试了。根据我的实盘经验，有几点特别需要注意：

1. 交易时机：严格按策略设定的时间点执行交易，避免临时起意
2. 滑点控制：对于流动性较差的ETF，使用限价单而非市价单
3. 资金管理：初期建议用不超过10%的资金测试新策略
4. 记录分析：详细记录每笔交易的执行价格、数量、时间等信息

实盘中最常见的问题是策略在回测中表现很好，但实盘却不尽如人意。这通常是由于回测中忽略了一些现实约束。我的做法是先在模拟盘运行1-2个月，确认各项流程顺畅后再投入真金白银。

最后提醒一点，任何策略都有其适应的市场环境。ETF动量策略在趋势行情中表现优异，但在震荡市中可能会频繁止损。因此最好不要把所有资金押注在单一策略上，构建多策略组合是更稳健的选择。