基于Backtrader 多空量化策略程序

该策略基于 Python 的backtrader框架开发，针对 1 分钟 K 线数据设计多空双开逻辑，核心围绕价格形态、成交量异常及风险控制构建交易规则，适用于期货品种的历史回测验证。
核心逻辑
***开仓规则：***仅在每分钟 59 秒判断，无持仓时触发多 / 空开仓条件。
***多单：***连续 3 根及以上下跌 K 线，当前成交量达阈值且为近 10 根最大、是上一根 2-10 倍，价格满足特定波动范围，且不在开盘 5 分钟内，以当前 K 线低点挂限价单。
***空单：***连续 3 根及以上上涨 K 线，其余成交量、价格条件与多单对称，以当前 K 线高点挂限价单。
委托管理：多单委托 3 分钟过期，空单 6 分钟过期；价格偏离开仓价 3 元时自动取消委托。
***风控机制：***固定止盈 3 元、止损 2 元；当日连续 3 笔亏损则暂停交易，次日重置。

基础版代码如下，供大家参考和交流！

import backtrader as bt
import datetime
import pandas as pd

class MultiStrategy(bt.Strategy):
params = (
(‘volume_threshold’, 5000),
(‘day_open_hour’, 9), (‘day_open_min’, 0),
(‘night_open_hour’, 21), (‘night_open_min’, 0),
(‘long_expire_min’, 3), (‘short_expire_min’, 6),
(‘stop_win’, 3), (‘stop_loss’, 2), (‘max_loss_streak’, 3)
)

def __init__(self):
    self.loss_count = 0
    self.last_trade_date = None
    self.trade_paused = False
    self.long_order = self.short_order = None
    self.long_order_time = self.short_order_time = None
    
    self.o, self.h, self.l, self.c, self.v = self.data.open, self.data.high, self.data.low, self.data.close, self.data.volume

def next(self):
    current_time = self.data.datetime.datetime(0)
    current_date = current_time.date()

    if self.last_trade_date != current_date:
        self.loss_count = 0
        self.trade_paused = False
        self.last_trade_date = current_date

    if self.trade_paused:
        return

    self.cancel_expired_orders(current_time)

    if not self.position and current_time.minute == 59:
        if self.long_condition():
            long_price = self.l[0]
            self.long_order = self.buy(price=long_price, exectype=bt.Order.Limit,
                                     stopprice=long_price - self.p.stop_loss,
                                     limitprice=long_price + self.p.stop_win)
            self.long_order_time = current_time

        if self.short_condition():
            short_price = self.h[0]
            self.short_order = self.sell(price=short_price, exectype=bt.Order.Limit,
                                      stopprice=short_price + self.p.stop_loss,
                                      limitprice=short_price - self.p.stop_win)
            self.short_order_time = current_time

def long_condition(self):
    if len(self) < 4:
        return False
    
    if not (self.c[1] <= self.o[1] and self.c[2] <= self.o[2] and self.c[3] <= self.o[3]):
        return False

    curr_vol = self.v[0]
    if curr_vol <= self.p.volume_threshold:
        return False

    prev_vol = self.v[1]
    if not (prev_vol*2 <= curr_vol <= prev_vol*10):
        return False

    current_price = (self.data.ask + self.data.bid)/2
    if abs(current_price - self.l[0]) >=1 or abs(current_price - self.c[0]) >=1:
        return False

    if current_price >= self.l[1] -5:
        return False

    if curr_vol != max(self.v[i] for i in range(10)):
        return False

    return not self.is_in_open_5min()

def short_condition(self):
    if len(self) <4:
        return False
    
    if not (self.c[1] >= self.o[1] and self.c[2] >= self.o[2] and self.c[3] >= self.o[3]):
        return False

    curr_vol = self.v[0]
    if curr_vol <= self.p.volume_threshold:
        return False

    prev_vol = self.v[1]
    if not (prev_vol*2 <= curr_vol <= prev_vol*10):
        return False

    current_price = (self.data.ask + self.data.bid)/2
    if abs(current_price - self.h[0]) >=1 or abs(current_price - self.c[0]) >=1:
        return False

    if current_price <= self.h[1] +5:
        return False

    if curr_vol != max(self.v[i] for i in range(10)):
        return False

    return not self.is_in_open_5min()

def is_in_open_5min(self):
    current_time = self.data.datetime.datetime(0)
    day_open = datetime.datetime.combine(current_time.date(), 
                                       datetime.time(self.p.day_open_hour, self.p.day_open_min))
    night_open = datetime.datetime.combine(current_time.date(), 
                                        datetime.time(self.p.night_open_hour, self.p.night_open_min))
    return (day_open <= current_time < day_open + datetime.timedelta(minutes=5)) or \
           (night_open <= current_time < night_open + datetime.timedelta(minutes=5))

def cancel_expired_orders(self, current_time):
    if self.long_order and self.long_order.status == bt.Order.Submitted:
        if (current_time - self.long_order_time) >= datetime.timedelta(minutes=self.p.long_expire_min) or \
           (self.data.ask - self.long_order.price) >=3:
            self.cancel(self.long_order)
            self.long_order = None

    if self.short_order and self.short_order.status == bt.Order.Submitted:
        if (current_time - self.short_order_time) >= datetime.timedelta(minutes=self.p.short_expire_min) or \
           (self.short_order.price - self.data.bid) >=3:
            self.cancel(self.short_order)
            self.short_order = None

def notify_trade(self, trade):
    if trade.isclosed:
        if trade.pnl <0:
            self.loss_count +=1
            if self.loss_count >= self.p.max_loss_streak:
                self.trade_paused = True
        else:
            self.loss_count =0

if name == ‘main’:
cerebro = bt.Cerebro()
cerebro.addstrategy(MultiStrategy)

data = pd.read_csv('1min_data.csv', parse_dates=['datetime'], index_col='datetime')
bt_data = bt.feeds.PandasData(dataname=data, timeframe=bt.TimeFrame.Minutes, compression=1)
cerebro.adddata(bt_data)

cerebro.broker.setcash(100000.0)
cerebro.broker.setcommission(0.0001)

print(f"初始资金: {cerebro.broker.getvalue():.2f}")
cerebro.run()
print(f"回测结束资金: {cerebro.broker.getvalue():.2f}")
cerebro.plot()