BigQuant使用文档

Tick回测demos

由qxiao创建,最终由qxiao 被浏览 5 用户

策略一:IF 双均线交叉 Tick 策略

基于双均线交叉的股指期货(IF)日内趋势跟踪策略。核心逻辑:用 5 周期和 20 周期的简单均线生成多空信号,短均线上穿长均线且偏离超过万分之二时开多,下穿时开空,并设置 10 点止损和 20 点止盈进行风险控制。

from collections import deque
from bigquant import bigtrader

# ── 策略参数 ──────────────────────────────────────────────
SHORT_WINDOW    = 5       # 短期均线 tick 数
LONG_WINDOW     = 20      # 长期均线 tick 数
SIGNAL_THRESH   = 0.0002  # 均线偏离阈值,过滤噪音(万分之二)
STOP_LOSS_PTS   = 10      # 止损点数(IF 每点 300 元,10 点 = 3000 元)
TAKE_PROFIT_PTS = 20      # 止盈点数
CONTRACT_SIZE   = 1       # 每次开仓手数

def initialize(context: bigtrader.IContext):
    print("initialize")

def before_trading_start(context: bigtrader.IContext, data: bigtrader.IBarData):
    print(f"before_trading_start instruments={context.instruments}")
    context.subscribe(context.instruments)

    # 每日开盘重置 tick 价格窗口和状态
    context.price_window = deque(maxlen=LONG_WINDOW)
    context.entry_price  = 0.0   # 入场价格
    context.position_dir = 0     # 当前方向:1=多,-1=空,0=空仓

def handle_tick(context: bigtrader.IContext, tick: bigtrader.ITickData):
    price      = tick.last_price
    instrument = tick.instrument

    if price <= 0:
        return

    # ── 1. 更新价格窗口 ───────────────────────────────────
    context.price_window.append(price)

    # 数据积累不足,不交易
    if len(context.price_window) < LONG_WINDOW:
        return

    prices   = list(context.price_window)
    short_ma = sum(prices[-SHORT_WINDOW:]) / SHORT_WINDOW
    long_ma  = sum(prices) / LONG_WINDOW

    # ── 2. 止盈 / 止损检查(优先于开仓信号)─────────────
    if context.position_dir != 0 and context.entry_price > 0:
        pnl_pts = (price - context.entry_price) * context.position_dir
        if pnl_pts <= -STOP_LOSS_PTS:
            print(f"[止损] {instrument} 价格={price:.2f}  入场={context.entry_price:.2f}  亏损={pnl_pts:.1f}点")
            context.order_target(instrument, 0)
            context.position_dir = 0
            context.entry_price  = 0.0
            return
        if pnl_pts >= TAKE_PROFIT_PTS:
            print(f"[止盈] {instrument} 价格={price:.2f}  入场={context.entry_price:.2f}  盈利={pnl_pts:.1f}点")
            context.order_target(instrument, 0)
            context.position_dir = 0
            context.entry_price  = 0.0
            return

    # ── 3. 信号生成:短均线与长均线的相对位置 ────────────
    if short_ma > long_ma * (1 + SIGNAL_THRESH):
        signal = 1
    elif short_ma < long_ma * (1 - SIGNAL_THRESH):
        signal = -1
    else:
        signal = 0

    # ── 4. 执行交易 ───────────────────────────────────────
    if signal == 1 and context.position_dir <= 0:
        context.order_target(instrument, CONTRACT_SIZE)
        context.position_dir = 1
        context.entry_price  = price
        print(f"[开多] {instrument} 价格={price:.2f}  短均={short_ma:.2f}  长均={long_ma:.2f}")
    elif signal == -1 and context.position_dir >= 0:
        context.order_target(instrument, -CONTRACT_SIZE)
        context.position_dir = -1
        context.entry_price  = price
        print(f"[开空] {instrument} 价格={price:.2f}  短均={short_ma:.2f}  长均={long_ma:.2f}")

def handle_order(context: bigtrader.IContext, order: bigtrader.IOrderData):
    print(f"收到order: {order}")

def handle_trade(context: bigtrader.IContext, trade: bigtrader.ITradeData):
    print(f"收到trade: {trade}")

bigtrader.run(
    start_date='2026-03-20',
    end_date='2026-04-01',
    market=bigtrader.Market.CN_FUTURE,
    frequency=bigtrader.Frequency.TICK,
    instruments=["IF2606.CFE"],
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_tick=handle_tick,
    handle_order=handle_order,
    handle_trade=handle_trade,
)

策略二:IF 布林带突破 Tick 策略

基于布林带(Bollinger Bands)的 tick 级别趋势突破策略,适用于沪深300股指期货(IF)。价格突破布林带上轨时做多,突破下轨时做空,并结合 ATR 动态止损控制风险。

核心逻辑:

  1. 维护滚动 tick 价格窗口(默认 60 个 tick),计算布林带上下轨
  2. 价格上穿上轨 → 做多信号;价格下穿下轨 → 做空信号
  3. 用最近 N 个 tick 的真实波幅估算 ATR,止损距离 = ATR_MULTIPLIER × ATR
  4. 固定盈亏比止盈:止盈距离 = PROFIT_RATIO × 止损距离
import math
from collections import deque
from bigquant import bigtrader
from bigtrader.constant import Direction, OrderType

# ── 策略参数 ──────────────────────────────────────────────────────────────────
BOLL_WINDOW    = 60    # 布林带计算窗口(tick 数)
BOLL_K         = 2.0   # 布林带宽度系数(标准差倍数)
ATR_WINDOW     = 20    # ATR 估算窗口(tick 数)
ATR_MULTIPLIER = 1.5   # 止损距离 = ATR_MULTIPLIER * ATR
PROFIT_RATIO   = 2.0   # 盈亏比:止盈距离 = PROFIT_RATIO * 止损距离
CONTRACT_SIZE  = 1     # 每次开仓手数

def initialize(context: bigtrader.IContext):
    print("=" * 60)
    print("策略初始化:IF 布林带突破 Tick 策略")
    print("=" * 60)

def before_trading_start(context: bigtrader.IContext, data: bigtrader.IBarData):
    print(f"[开盘前] {data.current_dt.strftime('%Y-%m-%d')} instruments={context.instruments}")
    context.subscribe(context.instruments)

    context.price_window = deque(maxlen=BOLL_WINDOW)
    context.range_window = deque(maxlen=ATR_WINDOW)
    context.prev_price   = None

    context.position_dir = 0
    context.entry_price  = 0.0
    context.stop_loss    = 0.0
    context.take_profit  = 0.0

def _calc_boll(price_window):
    if len(price_window) < BOLL_WINDOW:
        return None, None, None
    prices = list(price_window)
    n      = len(prices)
    mean   = sum(prices) / n
    variance = sum((p - mean) ** 2 for p in prices) / n
    std    = math.sqrt(variance)
    return mean + BOLL_K * std, mean, mean - BOLL_K * std

def _calc_atr(range_window):
    if len(range_window) < ATR_WINDOW:
        return None
    return sum(range_window) / len(range_window)

def handle_tick(context: bigtrader.IContext, tick: bigtrader.ITickData):
    price      = tick.last_price
    instrument = tick.instrument

    if price <= 0:
        return

    # ── 1. 更新数据窗口 ───────────────────────────────────────────────────────
    if context.prev_price is not None:
        context.range_window.append(abs(price - context.prev_price))
    context.price_window.append(price)
    context.prev_price = price

    # ── 2. 止盈 / 止损检查 ─────────────────────────────────────────────────────
    if context.position_dir != 0:
        hit_stop   = (context.position_dir == 1  and price <= context.stop_loss) or \
                     (context.position_dir == -1 and price >= context.stop_loss)
        hit_profit = (context.position_dir == 1  and price >= context.take_profit) or \
                     (context.position_dir == -1 and price <= context.take_profit)

        if hit_stop:
            pnl = (price - context.entry_price) * context.position_dir
            print(f"[止损] {instrument} 价格={price:.2f}  入场={context.entry_price:.2f}  盈亏={pnl:.2f}点")
            if context.position_dir == 1:
                context.sell_close(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
            else:
                context.buy_close(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
            context.position_dir = 0
            context.entry_price  = 0.0
            return

        if hit_profit:
            pnl = (price - context.entry_price) * context.position_dir
            print(f"[止盈] {instrument} 价格={price:.2f}  入场={context.entry_price:.2f}  盈亏={pnl:.2f}点")
            if context.position_dir == 1:
                context.sell_close(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
            else:
                context.buy_close(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
            context.position_dir = 0
            context.entry_price  = 0.0
            return

    # ── 3. 计算布林带和 ATR ───────────────────────────────────────────────────
    upper, middle, lower = _calc_boll(context.price_window)
    atr = _calc_atr(context.range_window)

    if upper is None or atr is None or atr <= 0:
        return

    # ── 4. 信号生成与下单 ─────────────────────────────────────────────────────
    prices_list = list(context.price_window)
    if len(prices_list) < 2:
        return
    prev_price_for_signal = prices_list[-2]

    stop_dist   = ATR_MULTIPLIER * atr
    profit_dist = PROFIT_RATIO * stop_dist

    # 做多信号:价格从布林带上轨下方突破到上方
    if prev_price_for_signal <= upper < price and context.position_dir <= 0:
        if context.position_dir == -1:
            context.buy_close(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
        context.buy_open(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
        context.position_dir = 1
        context.entry_price  = price
        context.stop_loss    = price - stop_dist
        context.take_profit  = price + profit_dist
        print(f"[开多] {instrument} 价格={price:.2f}  上轨={upper:.2f}  止损={context.stop_loss:.2f}  止盈={context.take_profit:.2f}")

    # 做空信号:价格从布林带下轨上方突破到下方
    elif prev_price_for_signal >= lower > price and context.position_dir >= 0:
        if context.position_dir == 1:
            context.sell_close(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
        context.sell_open(instrument, CONTRACT_SIZE, price, order_type=OrderType.MARKET)
        context.position_dir = -1
        context.entry_price  = price
        context.stop_loss    = price + stop_dist
        context.take_profit  = price - profit_dist
        print(f"[开空] {instrument} 价格={price:.2f}  下轨={lower:.2f}  止损={context.stop_loss:.2f}  止盈={context.take_profit:.2f}")

def handle_order(context: bigtrader.IContext, order: bigtrader.IOrderData):
    print(f"[委托] {order}")

def handle_trade(context: bigtrader.IContext, trade: bigtrader.ITradeData):
    print(f"[成交] {trade}")

performance = bigtrader.run(
    start_date='2026-03-20',
    end_date='2026-04-01',
    market=bigtrader.Market.CN_FUTURE,
    frequency=bigtrader.Frequency.TICK,
    instruments=["IF2606.CFE"],
    capital_base=500000,
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_tick=handle_tick,
    handle_order=handle_order,
    handle_trade=handle_trade,
)

\

{link}