2026年1月1日星期四

聚宽不是必须,我在本地Python实现了多指标回测框架

 

字数 1511,阅读大约需 8 分钟

在本地Python环境下构建回测框架时,很多人都会遇到一个问题:
如何像聚宽一样,生成一整套多指标的评价结果,从而更全面地衡量策略的表现?

其实,这件事情在本地是完全可以实现的。

聚宽的风险指标体系

在聚宽平台上回测时,总共能生成21个风险评估指标的结果,包括: 策略收益、策略年化收益、超额收益、基准收益、阿尔法、贝塔、夏普比率、胜率、盈亏比、最大回撤、索提诺比率、日均超额收益、超额收益最大回撤、超额收益夏普比率、日胜率、盈利次数、亏损次数、信息比率、策略波动率、基准波动率、最大回撤区间

聚宽回测结果
聚宽回测结果

同时,聚宽也在API文档里给出了各个指标的 计算方式 。因此,我们完全可以按照聚宽的公式,在本地用代码实现同样的计算逻辑。

聚宽风险评估指标介绍:https://www.joinquant.com/help/api/help#api:%E9%A3%8E%E9%99%A9%E6%8C%87%E6%A0%87

本地Backtrader回测

在本地Python环境中,常用的回测框架是Backtrader。那么,在本地使用 Backtrader 进行回测时,如何实现一套和聚宽类似的多指标回测框架呢?

首先,Backtrader 回测中最基础的输出仍然是 策略的收益率序列 ,这是后续各类风险与绩效指标计算的核心。

除此之外,还需要额外准备两个输入:

1. 第一个是每一笔交易的收益情况 ,用于计算胜率、盈亏比等基于交易层面的指标。Backtrader 在回测过程中会完整记录每一笔交易的信息。我们在策略类中添加如下代码即可:

class strategy(bt.Strategy):
    def __init__(self):
        self.trades = [] # 记录成交成功的历史订单
        
    def notify_trade(self, trade):
        if trade.isclosed:
            self.trades.append(trade.pnl)  # 记录盈亏金额

基于每一笔交易的盈亏信息,我们可以计算 胜率、盈亏比、盈利次数、亏损次数

# 胜率、盈亏比、盈利次数、亏损次数
wins = [x for x in strategy.trades if x > 0]  # 盈利交易
losses = [x for x in strategy.trades if x <= 0# 亏损交易
total_trades = len(strategy.trades)
win_rate = len(wins) / total_trades if total_trades > 0 else 0 # 胜率
total_profit = sum(wins)
total_loss = abs(sum(losses))
profit_factor = total_profit / total_loss if total_loss > 0 else float('inf')   # 盈亏比 

2. 第二个是基准指数的收益率序列 。这一部分可以直接通过 Tushare 获取对应指数数据,再转换为收益率序列,用于计算超额收益、Alpha、Beta 等指标。

  • 用Tushare获取基准指数收益率序列
# 基准指数收益
Ashare = pro.index_daily(**{'ts_code':'000300.SH'
                        'trade_date'""
                        'start_date': start_date, 
                        'end_date': end_date, 
                        'limit':""
                        'offset':""}, 
                    field=['ts_code''trade_date''close'
                            'pre_close''pct_chg'])  
Ashare.index = pd.to_datetime(Ashare.trade_date)
Ashare = Ashare.sort_index(axis=0)    
  • 计算超额收益、Alpha、Beta等指标

# beta
X = sm.add_constant(market_pnl)  # 自变量:基准收益
y = port_pnl                     # 因变量:策略收益
model = sm.OLS(y, X).fit()
beta = model.params[1]  # 斜率即beta  

# alpha
annualized_strategy_return = perf_stats.loc['Annual return'][0]
annualized_market_return = perf_stats_market.loc['Annual return'][0]
alpha = annualized_strategy_return - \
    (risk_free_rate + beta * (annualized_market_return - risk_free_rate))    

小结

基于以上思路和数据,我们就可以在本地 Python 环境中搭建出一套与聚宽类似的多指标回测框架,对策略表现进行全面、系统的评估。

本地回测结果
本地回测结果

同时,本地实现的回测框架能带来更高的 自由度 ,所有数据与计算过程都可随时调试和核查,相比平台上的“黑箱式”回测要 直观、可控得多

实现以上过程的核心代码如下:

# 分析函数
def analyzer(port_pnl, market_pnl, strategy, risk_free_rate=0.03):
    stats_index = ['策略收益''策略年化收益''超额收益','基准收益',
                   '阿尔法','贝塔','夏普比率','胜率','盈亏比','最大回撤','索提诺比率',
                   '日均超额收益','超额收益最大回撤','超额收益夏普比率','日胜率',
                   '盈利次数','亏损次数','信息比率','策略波动率','基准波动率']
    stats_all = pd.Series(index = stats_index, dtype='float64')
    
    # 用pyfolio包统计时间段的收益指标,包括累计收益、年化收益、夏普比率、最大回撤、索提诺比率
    perf_stats = pf.timeseries.perf_stats((port_pnl)).to_frame(name='all')
    perf_stats_market = pf.timeseries.perf_stats((market_pnl)).to_frame(name='all')

    # 回撤序列
    cumulative = (port_pnl + 1).cumprod()/(port_pnl[0]+1)
    max_return = cumulative.cummax()
    drawdown = (cumulative - max_return) / max_return
    
    cumulative_market = (market_pnl + 1).cumprod()/(market_pnl[0]+1)
    
    # beta
    X = sm.add_constant(market_pnl)  # 自变量:基准收益
    y = port_pnl                     # 因变量:策略收益
    model = sm.OLS(y, X).fit()
    beta = model.params[1]  # 斜率即beta  
            
    # alpha
    annualized_strategy_return = perf_stats.loc['Annual return'][0]
    annualized_market_return = perf_stats_market.loc['Annual return'][0]
    alpha = annualized_strategy_return - \
        (risk_free_rate + beta * (annualized_market_return - risk_free_rate))

    # 策略波动率
    port_volatility = np.sqrt(250*port_pnl.var(ddof=1))
    # 基准波动率
    market_volatility = np.sqrt(250*market_pnl.var(ddof=1))
    
    # 超额收益  
    daily_ratio_alpha = (1 + port_pnl) / (1 + market_pnl) - 1   
    # 日均超额收益
    AEI = daily_ratio_alpha.mean()
    # 超额收益最大回撤
    cumulative_alpha = (daily_ratio_alpha + 1).cumprod()/(daily_ratio_alpha[0] + 1)
    max_return_alpha = cumulative_alpha.cummax()
    maxdrawdown_alpha = np.min((cumulative_alpha - max_return_alpha) / max_return_alpha)   
    # 超额收益夏普比率
    perf_stats_alpha = pf.timeseries.perf_stats((daily_ratio_alpha)).to_frame(name='all')
    sharp_ratio_alpha = perf_stats_alpha.loc['Sharpe ratio'][0]
    
    # 日胜率
    daily_win = port_pnl > market_pnl  # 返回布尔序列(True表示跑赢)
    total_days = len(daily_win)
    win_days = daily_win.sum()  # True的个数即为赢的天数
    daily_win_rate = (win_days / total_days) 
    
    # 信息比率
    daily_stdv = (port_pnl - market_pnl).std()
    annualized_stdv = daily_stdv * np.sqrt(250)    
    information_ratio = \
        (perf_stats.loc['Annual return'][0] - perf_stats_market.loc['Annual return'][0])/annualized_stdv
    
    # 胜率、盈亏比、盈利次数、亏损次数
    wins = [x for x in strategy.trades if x > 0]  # 盈利交易
    losses = [x for x in strategy.trades if x <= 0# 亏损交易
    total_trades = len(strategy.trades)
    win_rate = len(wins) / total_trades if total_trades > 0 else 0
    total_profit = sum(wins)
    total_loss = abs(sum(losses))
    profit_factor = total_profit / total_loss if total_loss > 0 else float('inf')    
    
    stats_all.loc['策略收益'] = perf_stats.loc['Cumulative returns'][0]
    stats_all.loc['策略年化收益'] = perf_stats.loc['Annual return'][0]
    stats_all.loc['超额收益'] = cumulative_alpha[-1] - 1
    stats_all.loc['基准收益'] = perf_stats_market.loc['Cumulative returns'][0]
    stats_all.loc['阿尔法'] = alpha
    stats_all.loc['贝塔'] = beta
    stats_all.loc['夏普比率'] = perf_stats.loc['Sharpe ratio'][0]
    stats_all.loc['胜率'] = win_rate
    stats_all.loc['盈亏比'] = profit_factor
    stats_all.loc['最大回撤'] = abs(perf_stats.loc['Max drawdown'][0])
    stats_all.loc['索提诺比率'] = perf_stats.loc['Sortino ratio'][0]
    stats_all.loc['日均超额收益'] = AEI
    stats_all.loc['超额收益最大回撤'] = abs(maxdrawdown_alpha)
    stats_all.loc['超额收益夏普比率'] = sharp_ratio_alpha
    stats_all.loc['日胜率'] = daily_win_rate
    stats_all.loc['盈利次数'] = len(wins)
    stats_all.loc['亏损次数'] = len(losses)
    stats_all.loc['信息比率'] = information_ratio
    stats_all.loc['策略波动率'] = port_volatility
    stats_all.loc['基准波动率'] = market_volatility

 

没有评论:

发表评论

博客目录

    🔧 本地量化项目入口

    📦 本地量化接口库
    前往仓库 →
    💹 实盘交易系统(Live Trading System)
    前往仓库 →
    📊 策略模拟实盘(Strategy Live Trading)
    前往仓库 →