分钟回测demos
由qxiao创建，最终由qxiao 被浏览 1 用户
期货策略一：唐奇安通道突破（螺纹钢）
CTA 趋势跟踪策略，基于 N 根分钟 K 线的最高价/最低价构建唐奇安通道。当前收盘价 >= 上轨时开多/平空，当前收盘价 <= 下轨时开空/平多。
from datetime import datetime
import pandas as pd
from bigquant import bigtrader, dai
from bigtrader.constant import Direction, OrderType
# ── 策略参数 ──────────────────────────────────────────────────────────────────
INSTRUMENT = "rb8888.SHF" # 螺纹钢主连合约
DC_PERIOD = 20 # 唐奇安通道周期（分钟 K 线根数）
LOTS = 1 # 每次开仓手数
CAPITAL = 500000 # 初始资金（元）
START_DATE = "2026-04-30"
END_DATE = "2026-05-08"
def initialize(context: bigtrader.IContext):
context.logger.info("初始化：加载分钟行情，计算唐奇安通道...")
sql = """
SELECT
cn_future_bar1m.date,
cn_future_bar1m.instrument,
cn_future_bar1m.date::datetime::date as trading_day,
dominant,
close,
high,
low,
m_max(high, {period}, pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date) AS dc_upper,
m_min(low, {period}, pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date) AS dc_lower,
CASE
WHEN close >= dc_upper THEN 1
WHEN close <= dc_lower THEN -1
ELSE 0
END AS signal
FROM cn_future_bar1m
LEFT JOIN cn_future_dominant on cn_future_bar1m.date::datetime::date=cn_future_dominant.date
""".format(period=DC_PERIOD)
df = dai.query(sql, filters={
"date": [
context.add_trading_days(context.start_date, -5),
context.end_date,
],
"instrument": [INSTRUMENT]
}).df()
df['trading_day'] = pd.to_datetime(df['trading_day'])
context.data = df
context.lots = LOTS
context.logger.info(f"数据加载完成，共 {len(df)} 条分钟记录")
def before_trading_start(context: bigtrader.IContext, data: bigtrader.IBarData):
today = pd.to_datetime(data.current_dt.strftime("%Y-%m-%d"))
df_today = context.data[context.data["trading_day"] == today]
if df_today.empty:
return
dominant_symbol = df_today["dominant"].iloc[0]
if dominant_symbol and str(dominant_symbol) != "nan":
context.subscribe_bar([dominant_symbol])
context.logger.info(f"订阅分钟行情：{dominant_symbol}")
def handle_data(context: bigtrader.IContext, data: bigtrader.IBarData):
current_dt = data.current_dt
df_bar = context.data[context.data["date"] == current_dt]
if df_bar.empty:
return
signal = int(df_bar["signal"].iloc[0])
price = float(df_bar["close"].iloc[0])
dominant_symbol = df_bar["dominant"].iloc[0]
if not dominant_symbol or str(dominant_symbol) == "nan":
return
positions = context.get_account_positions()
position_symbol = list(positions.keys())[0] if positions else dominant_symbol
long_qty = context.get_account_position(position_symbol, direction=Direction.LONG).avail_qty
short_qty = context.get_account_position(position_symbol, direction=Direction.SHORT).avail_qty
# ── 移仓换月 ──────────────────────────────────────────────────────────────
if dominant_symbol != position_symbol:
context.logger.info(f"移仓换月：{position_symbol} → {dominant_symbol}")
if long_qty > 0:
context.sell_close(position_symbol, long_qty, price, order_type=OrderType.MARKET)
context.buy_open(dominant_symbol, long_qty, price, order_type=OrderType.MARKET)
elif short_qty > 0:
context.buy_close(position_symbol, short_qty, price, order_type=OrderType.MARKET)
context.sell_open(dominant_symbol, short_qty, price, order_type=OrderType.MARKET)
return
# ── 信号交易逻辑 ──────────────────────────────────────────────────────────
if signal == 1:
if short_qty > 0:
context.buy_close(position_symbol, short_qty, price, order_type=OrderType.MARKET)
if long_qty == 0:
context.buy_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
elif signal == -1:
if long_qty > 0:
context.sell_close(position_symbol, long_qty, price, order_type=OrderType.MARKET)
if short_qty == 0:
context.sell_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
performance = bigtrader.run(
market=bigtrader.Market.CN_FUTURE,
frequency=bigtrader.Frequency.MINUTE,
start_date=START_DATE,
end_date=END_DATE,
capital_base=CAPITAL,
initialize=initialize,
before_trading_start=before_trading_start,
handle_data=handle_data,
order_price_field_buy="open",
order_price_field_sell="open",
)
期货策略二：EMA 双均线交叉 + ATR 动态止损（铜）
基于分钟 K 线的 EMA 快慢线交叉策略，结合 ATR 动态止损。EMA_fast（12周期）上穿 EMA_slow（26周期）时金叉开多/平空，下穿时死叉开空/平多。止损距离 = 2 × ATR。
from datetime import datetime
import pandas as pd
from bigquant import bigtrader, dai
from bigtrader.constant import Direction, OrderType
# ── 策略参数 ──────────────────────────────────────────────────────────────────
INSTRUMENT = "cu8888.SHF" # 铜主连合约
EMA_FAST = 12 # 快线 EMA 周期
EMA_SLOW = 26 # 慢线 EMA 周期
ATR_PERIOD = 14 # ATR 周期
ATR_MULT = 2.0 # 止损距离 = ATR_MULT × ATR
LOTS = 1
CAPITAL = 500000
START_DATE = "2026-04-30"
END_DATE = "2026-05-08"
def initialize(context: bigtrader.IContext):
context.logger.info("初始化：加载分钟行情，计算 EMA 双均线 + ATR 止损...")
sql = """
SELECT
cn_future_bar1m.date,
cn_future_bar1m.instrument,
cn_future_bar1m.date::datetime::date AS trading_day,
dominant,
close,
m_ta_ema(close, {fast}, pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date) AS ema_fast,
m_ta_ema(close, {slow}, pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date) AS ema_slow,
m_lag(m_ta_ema(close, {fast}, pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date), 1,
pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date) AS ema_fast_prev,
m_lag(m_ta_ema(close, {slow}, pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date), 1,
pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date) AS ema_slow_prev,
m_ta_atr(high, low, close, {atr}, pb:=cn_future_bar1m.instrument, ob:=cn_future_bar1m.date) AS atr,
CASE
WHEN ema_fast > ema_slow AND ema_fast_prev <= ema_slow_prev THEN 1
WHEN ema_fast < ema_slow AND ema_fast_prev >= ema_slow_prev THEN -1
ELSE 0
END AS signal
FROM cn_future_bar1m
LEFT JOIN cn_future_dominant ON
cn_future_bar1m.date::datetime::date = cn_future_dominant.date
AND cn_future_bar1m.instrument = cn_future_dominant.instrument
""".format(fast=EMA_FAST, slow=EMA_SLOW, atr=ATR_PERIOD)
df = dai.query(sql, filters={
"date": [
context.add_trading_days(context.start_date, -10),
context.end_date,
],
"instrument": [INSTRUMENT]
}).df()
df["trading_day"] = pd.to_datetime(df["trading_day"])
context.data = df
context.lots = LOTS
context.stop_prices = {}
context.logger.info(f"数据加载完成，共 {len(df)} 条分钟记录")
def before_trading_start(context: bigtrader.IContext, data: bigtrader.IBarData):
today = pd.to_datetime(data.current_dt.strftime("%Y-%m-%d"))
df_today = context.data[context.data["trading_day"] == today]
if df_today.empty:
return
dominant_symbol = df_today["dominant"].iloc[0]
if dominant_symbol and str(dominant_symbol) != "nan":
context.subscribe_bar([dominant_symbol])
def handle_data(context: bigtrader.IContext, data: bigtrader.IBarData):
current_dt = data.current_dt
df_bar = context.data[context.data["date"] == current_dt]
if df_bar.empty:
return
row = df_bar.iloc[0]
signal = int(row["signal"])
price = float(row["close"])
atr = row["atr"]
dominant_symbol = row["dominant"]
if not dominant_symbol or str(dominant_symbol) == "nan":
return
if pd.isna(atr) or atr <= 0:
return
positions = context.get_account_positions()
position_symbol = list(positions.keys())[0] if positions else dominant_symbol
long_qty = context.get_account_position(position_symbol, direction=Direction.LONG).avail_qty
short_qty = context.get_account_position(position_symbol, direction=Direction.SHORT).avail_qty
# ── 移仓换月 ──────────────────────────────────────────────────────────────
if dominant_symbol != position_symbol:
context.logger.info(f"移仓换月：{position_symbol} → {dominant_symbol}")
if long_qty > 0:
context.sell_close(position_symbol, long_qty, price, order_type=OrderType.MARKET)
context.buy_open(dominant_symbol, long_qty, price, order_type=OrderType.MARKET)
context.stop_prices[dominant_symbol] = price - ATR_MULT * atr
elif short_qty > 0:
context.buy_close(position_symbol, short_qty, price, order_type=OrderType.MARKET)
context.sell_open(dominant_symbol, short_qty, price, order_type=OrderType.MARKET)
context.stop_prices[dominant_symbol] = price + ATR_MULT * atr
context.stop_prices.pop(position_symbol, None)
return
# ── ATR 动态止损检查 ──────────────────────────────────────────────────────
stop_price = context.stop_prices.get(dominant_symbol)
if stop_price is not None:
if long_qty > 0 and price <= stop_price:
context.logger.info(f"多头止损触发：price={price:.0f} <= stop={stop_price:.0f}")
context.sell_close(dominant_symbol, long_qty, price, order_type=OrderType.MARKET)
context.stop_prices.pop(dominant_symbol, None)
return
if short_qty > 0 and price >= stop_price:
context.logger.info(f"空头止损触发：price={price:.0f} >= stop={stop_price:.0f}")
context.buy_close(dominant_symbol, short_qty, price, order_type=OrderType.MARKET)
context.stop_prices.pop(dominant_symbol, None)
return
# ── EMA 双均线交叉信号 ────────────────────────────────────────────────────
if signal == 1:
if short_qty > 0:
context.buy_close(dominant_symbol, short_qty, price, order_type=OrderType.MARKET)
context.stop_prices.pop(dominant_symbol, None)
if long_qty == 0:
context.buy_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
context.stop_prices[dominant_symbol] = price - ATR_MULT * atr
context.logger.info(f"金叉开多：price={price:.0f}，止损={context.stop_prices[dominant_symbol]:.0f}")
elif signal == -1:
if long_qty > 0:
context.sell_close(dominant_symbol, long_qty, price, order_type=OrderType.MARKET)
context.stop_prices.pop(dominant_symbol, None)
if short_qty == 0:
context.sell_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
context.stop_prices[dominant_symbol] = price + ATR_MULT * atr
context.logger.info(f"死叉开空：price={price:.0f}，止损={context.stop_prices[dominant_symbol]:.0f}")
performance = bigtrader.run(
market=bigtrader.Market.CN_FUTURE,
frequency=bigtrader.Frequency.MINUTE,
start_date=START_DATE,
end_date=END_DATE,
capital_base=CAPITAL,
initialize=initialize,
before_trading_start=before_trading_start,
handle_data=handle_data,
order_price_field_buy="open",
order_price_field_sell="open",
)
期权分钟回测一：买入并持有科创板50ETF期权
"""买入并持有华夏上证科创板50ETF期权2605认购期权"""
import bigtrader
import math
def initialize(context):
context.bought = False
def before_trading_start(context, data):
context.subscribe_bar(["10011413.SHO"], '1m')
def is_none_or_nan(x):
return x is None or math.isnan(x)
def handle_data(context, data):
dt_str = context.current_dt.strftime('%Y-%m-%d %H:%M:%S')
if not is_none_or_nan(data.current("10011413.SHO", 'close')) and not context.bought:
context.logger.info(f'{dt_str}下单500张期权标的')
ret = context.buy_open("10011413.SHO", 500)
if ret >= 0:
context.bought = True
if __name__ == "__main__":
performance = bigtrader.run(
market=bigtrader.Market.CN_STOCK_OPTION,
frequency=bigtrader.Frequency.MINUTE,
start_date="2026-05-01",
end_date="2026-05-14",
capital_base=500000,
instruments=["588000.SH"],
benchmark=None,
initialize=initialize,
before_trading_start=before_trading_start,
handle_data=handle_data,
)
期权分钟回测二：日内期权偏度策略（IV 择时 + Delta 对冲）
中证1000指数期权（MO）+ 中证1000股指期货（IM）Delta 对冲策略。等距行权价（ATM±100），9:32 开仓前实时计算双腿 IV，仅当 Put IV / Call IV > 偏度阈值时开仓（风险逆转），同时做空 IM 期货抵消 Delta，实现方向中性。盘中每 30 分钟 re-hedge，收盘 14:56 强制平仓。
# ── Black-Scholes IV 反推 ────────────────────────────────────
def bs_iv(option_price, S, K, T, r=0.02, option_type='call'):
import math, numpy as np
from scipy.stats import norm
from scipy.optimize import brentq
if any(math.isnan(x) for x in [option_price, S, K, T]):
return None
if T <= 0 or S <= 0 or K <= 0 or option_price <= 0:
return None
intrinsic = max(0.0, S - K) if option_type == 'call' else max(0.0, K - S)
if option_price <= intrinsic + 1e-6:
return None
def bs_price(sigma):
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
d2 = d1 - sigma * np.sqrt(T)
if option_type == 'call':
return S * norm.cdf(d1) - K * np.exp(-r * T) * norm.cdf(d2)
else:
return K * np.exp(-r * T) * norm.cdf(-d2) - S * norm.cdf(-d1)
try:
return brentq(lambda sigma: bs_price(sigma) - option_price, 1e-6, 10.0, maxiter=100)
except Exception:
return None
def bs_delta(S, K, T, r=0.02, sigma=0.20, option_type='call'):
import numpy as np
from scipy.stats import norm
if T <= 0 or S <= 0 or K <= 0 or sigma <= 0:
return None
try:
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
return float(norm.cdf(d1)) if option_type == 'call' else float(norm.cdf(d1) - 1)
except Exception:
return None
# 盘中 re-hedge 时间点（每 30 分钟一次）
REHEDGE_TIMES = frozenset([
(10, 0), (10, 30), (11, 0), (11, 30),
(13, 0), (13, 30), (14, 0), (14, 30),
])
def initialize(context):
from bigmodule import M
context.enter_hour = 9
context.enter_minute = 32
context.exit_hour = 14
context.exit_minute = 56
context.strike_offset = 100
context.skew_threshold = 1.10
context.portfolio_stop_loss_ratio = 0.35
context.risk_free_rate = 0.02
context.num_contracts = 2
context.option_multiplier = 100
context.im_multiplier = 200
# ... （完整代码见原始文档）
import bigtrader
Q = bigtrader.run(
market=bigtrader.Market.CN_FUTURE_OPTION,
frequency=bigtrader.Frequency.MINUTE,
instruments=["000852.SH"],
benchmark="000852.SH",
start_date="2024-01-01",
end_date="2024-06-17",
capital_base=500000,
initialize=initialize,
before_trading_start=before_trading_start,
handle_data=handle_data,
)
\