Tick回测demos
由qxiao创建，最终由qxiao 被浏览 2 用户
策略一：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 动态止损控制风险。
核心逻辑：
- 维护滚动 tick 价格窗口（默认 60 个 tick），计算布林带上下轨
- 价格上穿上轨 → 做多信号；价格下穿下轨 → 做空信号
- 用最近 N 个 tick 的真实波幅估算 ATR，止损距离 = ATR_MULTIPLIER × ATR
- 固定盈亏比止盈：止盈距离 = 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,
)
\