发布于vn.py社区公众号【vnpy-community】
原文作者： 丛子龙 | 发布时间：2023-08-25
2023年VeighNa小班特训营【机器学习CTA】、【套利价差交易】报名进行中，目前半数名额已经被报名锁定，感兴趣的同学请抓紧！内容大纲戳这里。
聊聊PortfolioStrategy
对于不少VeighNa社区的同学来说，CtaStrategy和PortfolioStrategy这两个策略模块到底有什么区别，可能一直傻傻分不清楚。这里做了个对比表格，希望能够给大家一个比较清晰的认识：
过去几篇的【Elite量化策略实验室】系列文章围绕CTA策略模块（CtaStrategy）相关的内容展开，主要是针对期货市场的中高频策略，实盘中需要程序化自动交易执行。
本篇文章中将要分享的则是一套基于投组策略模块（PortfolioStrategy）的大类资产ETF轮动策略，该策略采用中低频日线级别数据生成交易信号，实盘中即使手动交易执行基本也能满足需求。
同样先来看一下策略的历史回测绩效：
策略基本信息
策略核心原理
本文中策略的思路来源于微信公众号【量化君也】的文章：
对于该策略的一些背景情况和详细原理，推荐直接看这篇文章（同时也强烈推荐【量化君也】公众号，其中有不少精彩的量化策略分享），所以这里只梳理下策略的核心逻辑。
对众多普通投资者来说，ETF基金是一种非常适合用来追踪大类资产价格波动，并且拥有较好流动性的交易品种，这里选择了四只比较典型的ETF来构建轮动组合：
- 黄金ETF（518880），代表大宗商品；
- 红利ETF（510880），代表价值股；
- 创业板ETF（159915），代表成长股；
- 纳指ETF（513100），代表美国科技股。
在具体决定每日要持仓的ETF时，使用的则是在金融领域中已经被广泛应用的时序动量（Time-Series Momentum）作为策略信号。
对每只ETF的收盘价时间序列，计算其线性回归后的的斜率，斜率越大代表走势越强。同时计算线性回归结果中的决定系数R平方（又名R方），其数值越大（越接近1）则说明拟合效果越好，反之（越接近0）则说明效果越差。
利用斜率和R平方的乘积得出一个趋势强弱评分score，score越高就表示动量越强，每日都选择持有当前score排名靠前的ETF进行轮动。
策略代码实现
策略模板构造
EtfRotationStrategy基于PortfolioStrategy模块下的策略模板类StrategyTemplate开发，可以直接在VeighNa开源版中使用（不依赖Elite版）。
class EtfRotationStrategy(StrategyTemplate):
"""ETF轮动策略"""
author: str = "CZL"
regression_window: int = 25 # 线性回归窗口
fixed_capital: int = 1_000_000 # 固定持仓市值
holding_symbol: str = "" # 持仓合约代码
parameters = [
"regression_window",
"fixed_capital"
]
variables = [
"holding_symbol"
]
def on_init(self) -> None:
"""策略初始化"""
# 确保缓存数据足够回归计算
size: int = self.regression_window + 1
# 创建每个合约的时序数据容器
self.ams: dict[str, ArrayManager] = {}
for vt_symbol in self.vt_symbols:
self.ams[vt_symbol] = ArrayManager(size)
self.write_log("策略初始化")
因为要使用前N天的收盘价历史来计算score，所以这里在创建ArrayManager实例时传入了size参数（self.regression_window + 1），在保证有足够缓存数据满足计算需求的同时，减少花费在回测初始化上的数据长度（日线的总数据长度相对分钟线要少几个数量级）。
信号指标计算
首先需要使用sklearn库中的线性回归功能，来实现对ETF当前趋势强弱得分score的计算函数：
from sklearn.linear_model import LinearRegression
def calculate_score(data: np.ndarray) -> float:
"""计算强弱得分"""
# 执行回归
x: np.ndarray = np.arange(1, len(data) + 1).reshape(-1, 1)
y: np.ndarray = data / data[0]
reg: LinearRegression = LinearRegression().fit(x, y)
# 返回得分
slope: float = reg.coef_[0]
r2: float = reg.score(x, y)
return slope * r2
然后在on_bars回调函数中即可计算各只ETF的得分score，并统一缓存到数据字典score_data中：
def on_bars(self, bars: dict[str, BarData]) -> None:
"""K线切片推送"""
# 更新K线到时序容器
for vt_symbol, bar in bars.items():
am: ArrayManager = self.ams[vt_symbol]
am.update_bar(bar)
# 计算每只ETF的分数
score_data: dict[str, float] = {}
for vt_symbol, bar in bars.items():
am: ArrayManager = self.ams[vt_symbol]
if not am.inited:
return
data: np.array = am.close[-self.regression_window:]
score_data[vt_symbol] = calculate_score(data)
目标交易执行
多标的截面类的PortfolioStrategy经常需要对多个合约同时交易，由用户在策略中直接实现具体的买卖委托操作可能较为麻烦，因此这里使用StrategyTemplate提供的目标仓位交易执行功能：
# 重置所有合约目标
for vt_symbol in self.vt_symbols:
self.set_target(vt_symbol, 0)
# 选出得分领先的ETF
self.holding_symbol: str = max(score_data, key=score_data.get)
price: float = bars[self.holding_symbol].close_price
# 交易数量必须是100整数倍
volume: int = 100 * int((self.fixed_capital / (price * 100)))
self.set_target(self.holding_symbol, volume)
# 根据设置好的目标仓位进行交易
self.rebalance_portfolio(bars)
# 推送UI更新
self.put_event()
在上文代码中，每日的交易执行步骤可以分解为：
- 将所有ETF上的持仓目标重置为0；
- 选出当前强弱分数排序第一的ETF，并基于其当前价格计算固定市值所需的持仓数量，使用set_target函数设置该ETF的持仓目标；
- 调用rebalance_portfolio，由策略引擎自动计算每只ETF上目标持仓target和实际持仓pos的差值执行委托交易；
回测结果
该策略的历史回测需要用到前文提及的四只ETF日线数据，可以下载zip数据文件后解压，找到其中csv格式的数据文件，然后使用DataManager模块导入数据库即可。
具体的回测参数配置如下：
回测结果的资金曲线和绩效统计可以参考前文中的内容。
需要注意的是，本文中EtfRotationStrategy的策略代码采用了固定市值持仓进行交易，在回测机制上属于单利模式（忽略了由于盈利带来的复利增长），更多为了体现策略本身逻辑的稳健性，而在实际交易中仍然应该根据账户的盈亏变化进行动态市值的持仓轮动。
完整策略代码和回测数据文件，可以通过【VeighNa进阶用户交流群】获取：
免责声明
文章中的信息或观点仅供参考，作者不对其准确性或完整性做出任何保证。读者应以其独立判断做出投资决策，作者不对因使用本报告的内容而引致的损失承担任何责任。
