量化策略——准备3 数据、Backtrader回测框架与quantstats评价指标

news2025/1/11 5:02:42

我们一般使用AKShare这个库来获取股票数据或策略中用得到的数据：
AKShare github主页：https://github.com/akfamily/akshare

使用Backtrader框架作为回测的框架：
Backtrader github主页：https://github.com/mementum/backtrader

使用quantstats库作为回测结果评价的库：
quantstats github主页：https://github.com/ranaroussi/quantstats

这一部分准备好之后，后续我们将关注点主要放在【策略】上，对于数据、评价指标这些如无特殊处理，将不再赘述。整个量化的框架构造起来不太容易，如果以前有所了解，可以用自己习惯的方式；如果觉得困难较大，也可以先跳过，等后面能力够了之后，再上手构建。

ps：大家要慎重的使用网上的量化平台，因为偷策略这种事太正常了，大家还是最好自己本地搭一个测试的平台~

文章目录

1. 数据准备
2. Backtrader回测框架准备
3. 评价指标

下面的数据准备与Backtrader回测准备，只是博主提供的一个参考，在开始正式介绍量化策略的时候是不会涉及到每个数据的采集，手把手的代码，这些都是不会提及的，只会提供一个backtrader的策略类，作为对策略的编程实现。

1. 数据准备

比如股票数据：

首先创建一个data文件夹，然后在文件夹里创建一个stock_data的文件夹
创建一个code文件夹用来存放程序文件

然后新建一个python文件，使用如下代码：

import time
import akshare as ak
from tqdm import tqdm
from loguru import logger


def extract_data():
    start_date = "20150101"
    end_date = "20221101"
    stock_list = ak.stock_zh_a_spot_em()  # 东方财富网-沪深京 A 股-实时行情数据

    for stock_code in tqdm(stock_list['代码']):
        time.sleep(1)
        stock_df = ak.stock_zh_a_hist(
            symbol=stock_code, period="daily", start_date=start_date, end_date=end_date, adjust="hfq")  # 后复权
        stock_df.to_pickle("../data/stock_data/{}.pkl".format(stock_code))
        logger.debug("ADD DATA {}", stock_code)


if __name__ == '__main__':
    extract_data()

这里博主把股票数据保存到data/stock_data/文件夹下，以股票代码.pkl的格式保存：

在这里插入图片描述

2. Backtrader回测框架准备

Backtrader足够简单，同时也非常接近实盘（国外可以一键切换实盘，国内没有接口）

Backtrader的使用请参考：Backtrader量化&回测1——基本的交易策略与挂单买卖

从策略到最终影响金额，都会经历四个步骤：

策略信号
委托
订单
金额与标的的置换

因此盯紧订单的变化就可以了解策略对金额变动的影响，为了将更多精力用于策略本身的编写上，我们写一个策略模版，然后以后的策略都可以通过继承这个模版，把与策略无关的变量、操作都写在模版里：

from loguru import logger
import backtrader as bt


class TemplateStrategy(bt.Strategy):

    def __init__(self):
        # 记录用
        self.buy_bond_record = {}  # 记录购买的订单
        self.sell_bond_record = {} # 记录卖出的订单

    def next(self):
        """最核心的触发策略"""
        raise

    def notify_order(self, order):
        """通知订单状态,当订单状态变化时触发"""
        today_time_string = self.datetime.datetime().strftime('%Y-%m-%d')
        if order.status in [order.Submitted, order.Accepted]:  # 接受订单交易，正常情况
            return
        if order.status in [order.Completed]:
            if order.isbuy():
                self.buy_bond_record.setdefault(today_time_string, {})
                self.buy_bond_record[today_time_string].setdefault(order.data._name.replace(".", "_"), [])
                self.buy_bond_record[today_time_string][order.data._name.replace(".", "_")].append({
                    "order_ref": order.ref,
                    "bond_name": order.data._name,
                    "size": order.size,
                    "price": order.executed.price,
                    "value": order.executed.value,
                    "trade_date": self.datetime.datetime(0),
                })
                logger.debug('{} 订单{} 已购入 {} , 购入单价 {:.2f}, 数量 {}, 费用 {:.2f}, 手续费 {:.2f}'.
                             format(self.datetime.date(), order.ref, order.data._name, order.executed.price, order.size,
                                    order.executed.value, order.executed.comm))
            elif order.issell():
                self.sell_bond_record.setdefault(today_time_string, {})
                self.sell_bond_record[today_time_string].setdefault(order.data._name.replace(".", "_"), [])
                self.sell_bond_record[today_time_string][order.data._name.replace(".", "_")].append({
                    "order_ref": order.ref,
                    "bond_name": order.data._name,
                    "size": order.size,
                    "price": order.executed.price,
                    "value": - order.executed.price * order.size,
                    "sell_type": order.info.sell_type,
                    "trade_date": self.datetime.datetime(0),
                })
                logger.debug('{} 订单{} 已卖出 {}, 卖出金额 {:.2f}, 数量 {}, 费用 {:.2f}, 手续费 {:.2f}'.
                             format(self.datetime.date(), order.ref, order.data._name, order.executed.price, order.size,
                                    -order.executed.price * order.size, order.executed.comm))
        elif order.status in [order.Margin, order.Rejected]:
            logger.warning('{} 订单{} 现金不足、金额不足拒绝交易', self.datetime.date(), order.ref)
        elif order.status in [order.Canceled]:
            logger.debug("{} 订单{} 已取消", self.datetime.date(), order.ref)
        elif order.status in [order.Expired]:
            logger.warning('{} 订单{} 超过有效期已取消, 订单开价 {}, 当天最高价{}, 最低价{}', self.datetime.date(), order.ref, order.price,
                           order.data.high[0], order.data.low[0])

之后的策略均继承此TemplateStrategy策略类，并覆写def next(self)函数即可，这一部分会将所有的订单在日志中打印下来

在程序中，购买的订单可以使用如下代码：

self.buy(
	data=self.getdatabyname(stock_name), # 针对哪一个股票代码
	size=100, # 数量
	price=self.getdatabyname(stock_name).close[0], # 以当天的收盘价购买
	exectype=bt.Order.Limit, # 限价单
	valid=self.getdatabyname(stock_name).datetime.date(1),  # 有效期1天
	)

3. 评价指标

我们使用quantstats这个库来对回测结果进行评价，这个库里的计算方法简单粗暴，通过对已有的计算方法的封装，我们得到可以方便的进行评价的方法：

import pandas as pd
import quantstats as qs


def cal_daily_return(fund_values: pd.Series):
    """根据资金变动，计算日资产的变化率
    :param fund_values: 每日的总资产
    """
    fund_values = fund_values.sort_index()
    daily_re: pd.Series = (fund_values / fund_values.shift(1)) - 1
    daily_re.iloc[0] = 0
    return daily_re


def cal_rolling_feature(daily_return_series: pd.Series, rf=0.02, record_dict: dict = None):
    """计算各种指标
    :param daily_return_series: 日收益的变化率
    :param rf: 无风险收益，这里定为0.02
    :param record_dict: 指标的结果会追加到这个字典中
    """
    if record_dict is None:
        record_dict = {}
    daily_return_series.index = pd.to_datetime(daily_return_series.index.values)
    feature_df = pd.DataFrame(index=daily_return_series.index)
    feature_df['累积收益率'] = qs.stats.compsum(daily_return_series).values
    feature_df['回撤'] = qs.stats.to_drawdown_series(daily_return_series)
    record_dict.update({"累积收益率": feature_df['累积收益率'].iloc[-1]})
    feature_dict = {
        "复合年增长": qs.stats.cagr(daily_return_series, rf=rf),
        "夏普比率": qs.stats.sharpe(daily_return_series, rf=rf),
        "索蒂诺": qs.stats.sortino(daily_return_series, rf=rf),
        "omega": qs.stats.omega(pd.DataFrame(daily_return_series), rf=rf),
        "最大回撤": qs.stats.max_drawdown(daily_return_series),
        "最大回撤期(天)": int(qs.stats.drawdown_details(feature_df['回撤'])['days'].max()),
        "年波动率": qs.stats.volatility(daily_return_series),
    }
    record_dict.update(feature_dict)
    # 决定保留的小数
    for key, value in record_dict.items():
        if isinstance(value, float):
            record_dict[key] = value.round(3)
    return feature_df, record_dict