# 计算乖离因子:价格相对BIAS_N日均线的偏离程度
def calculate_bias_ratio(close_prices, bias_n=90):
# 计算移动平均线
moving_avg = close_prices.rolling(window=bias_n, min_periods=1).mean()
# 计算乖离率
bias_ratio = close_prices / moving_avg
return bias_ratio
# 使用示例
close_prices = df['close'] # 获取收盘价数据
bias_ratio = calculate_bias_ratio(close_prices, BIAS_N)
bias_recent = bias_ratio.iloc[-MOMENTUM_DAY:] # 取最近25天的乖离率
乖离率 = (当前价格 - 90日均线) / 90日均线 × 100%
动量指标是通过对最近25天的乖离率进行线性回归,计算出趋势的斜率。
这个斜率告诉我们:斜率为正且增大:趋势强劲斜率为负且减小:趋势疲软斜率接近零:趋势不明朗策略最终得分就是动量斜率得分,选择动量得分最高的ETF进行配置。
from sklearn.linear_model import LinearRegression
import numpy as np
def calculate_momentum_score(bias_recent, momentum_day=25):
# 创建时间序列数组
x_array = np.arange(momentum_day)
# 标准化处理,将第一个值设为1
y_normalized = bias_recent / bias_recent.iloc[0]
# 使用线性回归计算斜率
lr_model = LinearRegression()
lr_model.fit(x_array.reshape(-1, 1), y_normalized)
momentum_score = lr_model.coef_[0].real * 10000# 放大10000倍
return momentum_score
# 使用示例
momentum_score = calculate_momentum_score(bias_recent, MOMENTUM_DAY)
print(f'动量得分: {momentum_score:.2f}')
猫哥把这个策略简化为三个步骤:
第一步:计算各类ETF的乖离率和动量
获取90日均线数据
计算每日乖离率
对25天乖离率进行线性回归,得到动量值
第二步:动量评分排序
对25天乖离率进行线性回归,计算趋势斜率
斜率越大得分越高,表示趋势越强劲
选择动量得分最高的1个ETF进行配置
第三步:每日调仓执行
每日开盘时执行调仓操作
卖出非目标ETF,买入目标ETF
轮动策略是什么?说到轮动策略,很多朋友可能不太熟悉。简单来说,就是在一篮子资产中,定期挑选表现最好的进行持仓,表现差的就踢出去。
为什么轮动策略有效呢?这背后有几个市场逻辑:
动量效应:涨得好的资产短期内往往会继续涨(趋势延续)
行业轮动:经济周期不同阶段,不同行业表现差异巨大
相对强弱:同一时期总有强势板块和弱势板块今天我们要实现的策略,就是在9个行业ETF中,每天计算它们的"动量得分",然后选出得分最高的持仓。
# 计算斜率动量得分
x = np.arange(1, N_DAYS + 1) # 时间序列
y = df['close'] / df['close'].iloc[0] # 价格归一化
lr = LinearRegression()
lr.fit(x.reshape(-1, 1), y)
slope = lr.coef_[0] # 斜率
r_squared = lr.score(x.reshape(-1, 1), y) # R²
# 最终得分 = 斜率 × R² × 10000
score = 10000 * slope * r_squared
RSRS全称是"阻力支撑相对强度",由光大证券在2017年发布的研报《基于阻力支撑相对强度的市场择时》中提出。
对过去18天的最高价和最低价做线性回归得到的斜率代表"上涨空间相对下跌空间的大小"计算这个斜率在过去600天中的Z-score(标准分)再乘以R²修正,得到最终的RSRS得分
RSRS得分的含义:
大于0.7:市场处于强势,支撑远大于阻力,可以买入
小于-0.7:市场处于弱势,阻力远大于支撑,应该清仓
-0.7到0.7之间:震荡行情,保持现状
# 计算RSRS指标
high_data = data['high'].values
low_data = data['low'].values
_, slope, r2 = get_ols_regression(low_data, high_data)
# 计算Z-score
rsrs_score = get_zscore(slope_series) * r2
# 均线信号
today_MA = close_data.iloc[MEAN_DIFF_DAY:].mean() # 当前均线
before_MA = close_data.iloc[:-MEAN_DIFF_DAY].mean() # 5天前的均线
# 综合判断
if rsrs_score > 0.7and today_MA > before_MA:
signal = "BUY"# RSRS强势 + 均线上升 = 买入
elif rsrs_score < -0.7and today_MA < before_MA:
signal = "SELL"# RSRS弱势 + 均线下降 = 清仓
else:
signal = "KEEP"# 维持现状