绝对型收益的策略，backtrader从零实现风险平价，年化8.69%，夏普1.61，附python代码

import backtrader as btimport ffnimport pandas as pdimport numpy as np
class RiskParityStrategy(bt.Strategy):    params = (        ('lookback', 60),       # 计算协方差矩阵的历史数据天数        ('rebalance_freq', 22),  # 再平衡频率（天）        ('covar_method', 'ledoit-wolf'),  # 协方差矩阵计算方法        ('max_iter', 100),       # ERC最大迭代次数        ('tol', 1e-08)           # 收敛容忍度    )
    def notify_order(self, order):        """订单状态通知"""        if order.status in [order.Submitted, order.Accepted]:            return
        if order.status in [order.Completed]:            if order.isbuy():                #self.log(f"买入执行, 价格: {order.executed.price:.2f}, 数量: {order.executed.size}, 成本: {order.executed.value:.2f}, 佣金: {order.executed.comm:.2f}")                self.buy_price = order.executed.price                self.buy_date = self.datas[0].datetime.date(0)            elif order.issell():                profit = order.executed.value - order.created.value                #self.log(f"卖出执行, 价格: {order.executed.price:.2f}, 数量: {order.executed.size}, 收益: {profit:.2f}, 佣金: {order.executed.comm:.2f}")


        elif order.status in [order.Canceled, order.Margin, order.Rejected]:            self.log(f"订单取消/保证金不足/拒绝")
        self.order = None
    def log(self, txt, dt=None, doprint=True):        """日志记录函数"""        if doprint:            dt = dt or self.datas[0].datetime.date(0)            print(f"{dt.isoformat()}, {txt}")
    def __init__(self):        # 跟踪再平衡日期        self.day_counter = 0        self.assets = self.datas[0:]  # 第0个数据是基准/现金，资产从1开始        self.initial_weights = np.ones(len(self.assets)) / len(self.assets)  # 初始等权重        self.last_month = -1  # 初始化为无效值
    def next(self):        self.day_counter += 1
        # 到达再平衡日        #if self.day_counter % self.params.rebalance_freq == 0:        current_date = self.datetime.date(0)        current_month = current_date.month
        # 检查月份是否变化（包括跨年情况）        if current_month != self.last_month:            # 更新记录月份            self.last_month = current_month            # 准备收益率数据


    
            returns = self._prepare_returns()            if returns is None:                return  # 数据不足时跳过
            # 计算ERC权重            weights = self._calculate_erc_weights(returns)            #print(self.datetime.date(),weights)
            # 调整仓位            self._rebalance_portfolio(weights)
    def _prepare_returns(self):        """收集历史数据并计算收益率"""        prices = []        valid_length = True
        # 检查所有资产是否有足够数据        for i, data in enumerate(self.assets):            if len(data) < self.params.lookback + 1:                valid_length = False                break            # 获取收盘价（从当前时刻回溯）            prices.append(data.close.get(size=self.params.lookback + 1))
        if not valid_length:            return None
        # 转换为DataFrame并计算收益率        price_df = pd.DataFrame(np.array(prices).T)        returns = price_df.pct_change().dropna()        return returns
    def _calculate_erc_weights(self, returns_df):        """调用ffn计算ERC权重"""        try:            # 确保输入是pd.DataFrame            if not isinstance(returns_df, pd.DataFrame):                returns_df = pd.DataFrame(returns_df)
            weights = ffn.core.calc_erc_weights(                returns=returns_df,                initial_weights=None,#self.initial_weights,                covar_method=self.params.covar_method,                risk_parity_method='ccd',                maximum_iterations=self.params.max_iter,                tolerance=self.params.tol            )            return weights.values*0.995        except Exception as e:            print(f"计算ERC权重失败: {e}")            return self.initial_weights  # 失败时使用初始权重
    def _rebalance_portfolio(self, weights):        """根据新权重调整仓位"""        total_value = self.broker.getvalue()
        to_buy = {}        for i, data in enumerate(self.assets):            # 计算目标市值            target_value = total_value * weights[i]
            # 计算当前持仓市值            current_value = self.getposition(data).size * data.close[0]
            # 计算需要交易的数量            size_diff = (target_value - current_value) / data.close[0]
            # 执行订单            if size_diff > 0:                to_buy[data]=size_diff
            elif size_diff 0:                self.sell(data=data, size=abs(size_diff))
        #self.buy(data=data, size=size_diff)        for data,size in to_buy.items():            self.buy(data=data, size=size)