Backtrader 框架实现了一个简单的量化交易策略回测。首先定义了一个名为 TestStrategy 的自定义策略类，该策略在连续两天收盘价下跌时发出买入信号，买入后持有 5 个 K 线周期后卖出。通过 notify_order 方法处理订单状态的变化并记录日志。然后创建 Cerebro 对象，添加策略和股票数据，设置初始资金为 10 万。最后运行回测，输出回测开始和结束时投资组合的价值，以此评估该交易策略在指定时间段内的表现。
（1）Backtrader程序代码
# -*- coding: utf-8 -*-
from __future__ import (absolute_import, division, print_function,unicode_literals)
import datetime # 用于datetime对象操作
import os.path # 用于管理路径
import sys # 用于在argvTo[0]中找到脚本名称
import backtrader as bt # 引入backtrader框架
# 创建策略
class TestStrategy(bt.Strategy):
def log(self, txt, dt=None):
''' 策略的日志函数'''
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
# 引用data[0]数据的收盘价数据
self.dataclose = self.datas[0].close
# 用于记录订单状态
self.order = None
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('BUY EXECUTED, %.2f' % order.executed.price)
elif order.issell():
self.log('SELL EXECUTED, %.2f' % order.executed.price)
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
# 无等待处理订单
self.order = None
def next(self):
# 日志输出收盘价数据
self.log('Close, %.2f' % self.dataclose[0])
# 检查是否有订单等待处理，如果是就不再进行其他下单
if self.order:
return
# 检查是否已经进场
if not self.position:
# 还未进场，则只能进行买入
# 当日收盘价小于前一日收盘价
if self.dataclose[0] < self.dataclose[-1]:
# 前一日收盘价小于前前日收盘价
if self.dataclose[-1] < self.dataclose[-2]:
# 买买买
self.log('BUY CREATE, %.2f' % self.dataclose[0])
# 记录订单避免二次下单
self.order = self.buy()
# 如果已经在场内，则可以进行卖出操作
else:
# 卖卖卖
if len(self) >= (self.bar_executed + 5):
self.log('SELL CREATE, %.2f' % self.dataclose[0])
# 记录订单避免二次下单
self.order = self.sell()
# 创建cerebro实体
cerebro = bt.Cerebro()
# 添加策略
cerebro.addstrategy(TestStrategy)
# 先找到脚本的位置，然后根据脚本与数据的相对路径关系找到数据位置
# 这样脚本从任意地方被调用，都可以正确地访问到数据
modpath = os.path.dirname(os.path.abspath(sys.argv[0]))
datapath = os.path.join(modpath, '../../TQDat/day/stk/000001.csv')
# 创建价格数据
data = bt.feeds.GenericCSVData(
dataname = datapath,
fromdate = datetime.datetime(2024, 1, 1),
todate = datetime.datetime(2025, 1, 10),
nullvalue = 0.0,
dtformat = ('%Y-%m-%d'),
datetime = 0,
open = 1,
high = 2,
low = 3,
close = 4,
volume = 5,
openinterest = -1
)
# 在Cerebro中添加价格数据
cerebro.adddata(data)
# 设置启动资金
cerebro.broker.setcash(100000.0)
# 打印开始信息
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
# 遍历所有数据
cerebro.run()
# 打印最后结果
print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())（2）Backtrader程序代码输出结果
runfile('D:/zwPython/zwrk/2_Backtrader/B-06AddStrategyAddBuyAddSell.py', wdir='D:/zwPython/zwrk/2_Backtrader')
Starting Portfolio Value: 100000.00
2024-12-02, Close, 11.39
2024-12-02, BUY CREATE, 11.39
2024-12-03, BUY EXECUTED, 11.37
2024-12-03, Close, 11.49
2024-12-04, Close, 11.46
2024-12-05, Close, 11.44
2024-12-06, Close, 11.66
2024-12-09, Close, 11.67
2024-12-10, Close, 11.79
2024-12-10, SELL CREATE, 11.79
2024-12-11, SELL EXECUTED, 11.79
2024-12-11, Close, 11.73
2024-12-12, Close, 11.85
2024-12-13, Close, 11.56
2024-12-16, Close, 11.57
2024-12-17, Close, 11.53
2024-12-18, Close, 11.65
2024-12-19, Close, 11.59
2024-12-20, Close, 11.62
2024-12-23, Close, 11.73
2024-12-24, Close, 11.86
2024-12-25, Close, 11.92
2024-12-26, Close, 11.86
2024-12-27, Close, 11.83
2024-12-27, BUY CREATE, 11.83
2024-12-30, BUY EXECUTED, 11.78
2024-12-30, Close, 11.95
2024-12-31, Close, 11.70
2025-01-02, Close, 11.43
2025-01-03, Close, 11.38
2025-01-06, Close, 11.44
2025-01-07, Close, 11.51
2025-01-07, SELL CREATE, 11.51
2025-01-08, SELL EXECUTED, 11.50
2025-01-08, Close, 11.50
2025-01-09, Close, 11.40
2025-01-09, BUY CREATE, 11.40
Final Portfolio Value: 100000.14（3）程序代码注释
# -*- coding: utf-8 -*-
# 指定文件编码为 UTF - 8，确保支持非 ASCII 字符，如中文等
from __future__ import (absolute_import, division, print_function, unicode_literals)
# 导入 Python 未来版本的特性，保证代码在不同 Python 版本中的兼容性
import datetime # 用于 datetime 对象操作
# 导入 datetime 模块，方便进行日期和时间的处理
import os.path # 用于管理路径
# 导入 os.path 模块，用于处理文件和目录的路径
import sys # 用于在 argvTo[0] 中找到脚本名称
# 导入 sys 模块，获取命令行参数和脚本相关信息
import backtrader as bt # 引入 backtrader 框架
# 导入 backtrader 库，用于量化交易策略的回测
# 创建策略
class TestStrategy(bt.Strategy):
# 定义一个自定义策略类，继承自 backtrader 的 Strategy 类
def log(self, txt, dt=None):
''' 策略的日志函数'''
dt = dt or self.datas[0].datetime.date(0)
# 如果没有传入日期，使用当前数据的日期
print('%s, %s' % (dt.isoformat(), txt))
# 以 ISO 格式输出日期和日志信息
def __init__(self):
# 策略的初始化方法，在策略实例化时调用
# 引用 data[0] 数据的收盘价数据
self.dataclose = self.datas[0].close
# 用于记录订单状态
self.order = None
# 初始化订单状态为 None，表示没有待处理订单
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('BUY EXECUTED, %.2f' % order.executed.price)
# 记录买入订单执行价格
elif order.issell():
self.log('SELL EXECUTED, %.2f' % order.executed.price)
# 记录卖出订单执行价格
self.bar_executed = len(self)
# 记录订单执行时的 K 线数量
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
# 记录订单被取消、保证金不足或被拒绝的情况
# 无等待处理订单
self.order = None
# 订单处理完毕，将订单状态置为 None
def next(self):
# 每个 K 线数据到来时调用的方法
# 日志输出收盘价数据
self.log('Close, %.2f' % self.dataclose[0])
# 检查是否有订单等待处理，如果是就不再进行其他下单
if self.order:
return
# 检查是否已经进场
if not self.position:
# 还未进场，则只能进行买入
# 当日收盘价小于前一日收盘价
if self.dataclose[0] < self.dataclose[-1]:
# 前一日收盘价小于前前日收盘价
if self.dataclose[-1] < self.dataclose[-2]:
# 买买买
self.log('BUY CREATE, %.2f' % self.dataclose[0])
# 记录创建买入订单的信息
# 记录订单避免二次下单
self.order = self.buy()
# 发出买入订单并记录订单状态
# 如果已经在场内，则可以进行卖出操作
else:
# 卖卖卖
if len(self) >= (self.bar_executed + 5):
# 满足买入后 5 个 K 线周期的条件
self.log('SELL CREATE, %.2f' % self.dataclose[0])
# 记录创建卖出订单的信息
# 记录订单避免二次下单
self.order = self.sell()
# 发出卖出订单并记录订单状态
# 创建 cerebro 实体
cerebro = bt.Cerebro()
# 创建 Cerebro 对象，作为回测的核心管理单元
# 添加策略
cerebro.addstrategy(TestStrategy)
# 向 Cerebro 中添加自定义的交易策略
# 先找到脚本的位置，然后根据脚本与数据的相对路径关系找到数据位置
# 这样脚本从任意地方被调用，都可以正确地访问到数据
modpath = os.path.dirname(os.path.abspath(sys.argv[0]))
# 获取当前脚本的绝对路径
datapath = os.path.join(modpath, '../../TQDat/day/stk/000001.csv')
# 构建数据文件的完整路径
# 创建价格数据
data = bt.feeds.GenericCSVData(
dataname=datapath,
# 指定数据文件的路径
fromdate=datetime.datetime(2024, 1, 1),
# 数据的起始日期
todate=datetime.datetime(2025, 1, 10),
# 数据的结束日期
nullvalue=0.0,
# 处理缺失值，用 0.0 填充
dtformat=('%Y-%m-%d'),
# 日期的格式
datetime=0,
# 日期所在列的索引
open=1,
# 开盘价所在列的索引
high=2,
# 最高价所在列的索引
low=3,
# 最低价所在列的索引
close=4,
# 收盘价所在列的索引
volume=5,
# 成交量所在列的索引
openinterest=-1
# 持仓量所在列的索引，-1 表示不使用该数据
)
# 在 Cerebro 中添加价格数据
cerebro.adddata(data)
# 将创建好的数据源添加到 Cerebro 中
# 设置启动资金
cerebro.broker.setcash(100000.0)
# 设置经纪人的初始资金为 100000 元
# 打印开始信息
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
# 输出回测开始时投资组合的价值
# 遍历所有数据
cerebro.run()
# 运行回测流程，Cerebro 会根据添加的数据和策略进行回测
# 打印最后结果
print('Final Portfolio Value: %.2f' % cerebro.broker.getvalue())
# 输出回测结束后投资组合的价值发布者：股市刺客，转载请注明出处：https://www.95sca.cn/archives/913345
站内所有文章皆来自网络转载或读者投稿，请勿用于商业用途。如有侵权、不妥之处，请联系站长并出示版权证明以便删除。敬请谅解！
