2417 协方差收缩多资产动量优化策略 多品种ETF动量轮动+EPO优化 47208

1. 多维度资产配置：

• 覆盖 15 类 ETF：包含商品（黄金 / 豆粕）、海外（纳指 / 恒生科技）、宽基（沪深 300 / 创业板）及 7 大行业 ETF
• 风险分散设计：通过商品与权益资产负相关性对冲系统性风险

2. 改进型动量模型：

• 双重趋势验证：34 日对数收益率线性回归计算年化收益（捕捉趋势强度）与 R 平方值（衡量趋势稳定性）
• 正向筛选机制：仅保留动量得分 > 0 的标的，规避下行趋势品种

3. 组合优化引擎：

• 协方差矩阵收缩：引入 1200 日历史波动率，通过权重参数 w 平衡长短期波动
• 风险平价优化：EPO 模型动态调整组合权重，使组合方差最小化
• 锚定波动控制：设置 lambda=10 的风险厌恶系数约束极端风险

4. 动态再平衡机制：

• 月度调仓频率：每月首个交易日执行组合再平衡
• 三层风控体系：0 滑点设置 + 最大回撤控制 + 正权重约束
• 流动性保障：限定 3 只 ETF 集中持仓，降低冲击成本

策略代码

from jqdata import *
from jqfactor import *
import numpy as np
import pandas as pd
import talib
from scipy.optimize import minimize
import statsmodels.api as sm
from scipy.linalg import solve
#初始化函数 
def initialize(context):
    # 设定基准
    set_benchmark('000300.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option('avoid_future_data', True)
    # 设置滑点为0 https://www.joinquant.com/view/community/detail/a31a822d1cfa7e83b1dda228d4562a70
    set_slippage(FixedSlippage(0))
    # 设置交易成本
    set_order_cost(OrderCost(open_tax=0, close_tax=0, open_commission=0.0002, close_commission=0.0002, close_today_commission=0, min_commission=5), type='fund')
    # 过滤一定级别的日志
    log.set_level('system', 'error')
    
    g.stock_num = 3
    g._lambda = 10
    g.w = 0.2


    # 参数
    g.etf_pool = [
        # 商品
        '518880.XSHG',#黄金ETF
        '159985.XSHE',#豆粕ETF
        # 海外
        '513100.XSHG',#纳指ETF
        # 宽基
        '510300.XSHG',#沪深300ETF
        '159915.XSHE',#创业板
        # 窄基
        '159992.XSHE',#创新药ETF
        '515700.XSHG',#新能车ETF
        '510150.XSHG',#消费ETF
        '515790.XSHG',#光伏ETF
        '515880.XSHG',#通信ETF
        '512720.XSHG',#计算机ETF
        '512660.XSHG',#军工ETF
        '159740.XSHE',#恒生科技ETF
        ]	
    run_monthly(trade, 1, '9:30')
    # run_daily(trade, '9:30') #每天运行确保即时捕捉动量变化
    g.m_days = 34 #动量参考天数

#============基于年化收益和判定系数打分的动量因子轮动=============#
def get_rank(etf_pool):
    score_list = []
    for etf in etf_pool:
        df = attribute_history(etf, g.m_days, '1d', ['close'])
        y = df['log'] = np.log(df.close)
        x = df['num'] = np.arange(df.log.size)
        slope, intercept = np.polyfit(x, y, 1)
        annualized_returns = math.pow(math.exp(slope), 250) - 1
        r_squared = 1 - (sum((y - (slope * x + intercept))**2) / ((len(y) - 1) * np.var(y, ddof=1)))
        score = annualized_returns * r_squared
        score_list.append(score)
    df = pd.DataFrame(index=etf_pool, data={'score':score_list})
    df = df.sort_values(by='score', ascending=False)
    df = df.dropna()
    rank_list = list(df.index)
    print (df)
    filtered_rank_list = [etf for etf in rank_list if df.loc[etf, 'score'] > 0]
    return filtered_rank_list
    #return rank_list   

关键函数解锁后查看：

# 定义获取数据并调用优化函数的函数
def run_optimization(stocks, end_date):
    prices = get_price(stocks, count=1200, end_date=end_date, frequency='daily', fields=['close'])['close']
    returns = prices.pct_change().dropna() # 计算收益率
    d = np.diag(returns.cov())
    a = (1/d) / (1/d).sum()
    # a= np.array([0.25,0.25,0.25,0.25])
    weights = epo(x = returns, signal = returns.mean(), lambda_ = g._lambda, method = 'anchored', w = g.w, anchor=a)
    return weights
    
# 交易
def trade(context):
    end_date = context.previous_date 
    target_list = get_rank(g.etf_pool)[:g.stock_num]
    
    # 卖出
    hold_list = list(context.portfolio.positions)
    for etf in hold_list:
        if etf not in target_list:
            order_target_value(etf, 0)
            print( '卖出' + str(etf))
        else:
            print( '继续持有' + str(etf))
            
    # 买入
    weights = run_optimization(target_list, end_date)

    if weights is None:
        return
    total_value = context.portfolio.total_value 
    index = 0
    for w in weights:
        value = total_value * w 
        order_target_value(target_list[index], value) 
        index+=1   

2025-02-25