日频回测demos
由qxiao创建，最终由qxiao 被浏览 1 用户
期货日频策略一：唐奇安通道突破（沪深300股指期货 IF）
使用 20 日唐奇安通道，收盘价突破 20 日最高价时做多，跌破 20 日最低价时做空，持仓期间自动移仓换月。
from datetime import datetime
from bigquant import bigtrader, dai
from bigtrader.constant import Direction, OrderType
def initialize(context: bigtrader.IContext):
context.logger.info("初始化：计算唐奇安通道信号...")
channel_period = 20
sql = f"""
SELECT
date,
instrument,
dominant,
close,
high,
low,
m_lag(m_max(high, {channel_period}), 1) AS upper_band,
m_lag(m_min(low, {channel_period}), 1) AS lower_band,
CASE
WHEN close >= upper_band THEN 1
WHEN close <= lower_band THEN -1
ELSE 0
END AS signal
FROM cn_future_bar1d
LEFT JOIN cn_future_dominant USING (date, instrument)
WHERE instrument = 'IF8888.CFE'
"""
df = dai.query(sql, filters={
"date": [
context.add_trading_days(context.start_date, -(channel_period + 10)),
context.end_date
]
}).df()
context.data = df
context.lots = 1
context.logger.info(f"数据加载完成，共 {len(df)} 条记录")
def handle_data(context: bigtrader.IContext, data: bigtrader.IBarData):
today = data.current_dt.strftime('%Y-%m-%d')
df_today = context.data[context.data['date'] == today]
if df_today.empty:
return
signal = int(df_today['signal'].values[0])
price = float(df_today['close'].values[0])
dominant_symbol = df_today['dominant'].values[0]
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
total_qty = long_qty + short_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)
position_symbol = dominant_symbol
# ── 信号交易 ──────────────────────────────────────────────
if long_qty > 0:
if signal == -1:
context.logger.info("平多开空：价格跌破 20 日最低价")
context.sell_close(position_symbol, long_qty, price, order_type=OrderType.MARKET)
context.sell_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
elif short_qty > 0:
if signal == 1:
context.logger.info("平空开多：价格突破 20 日最高价")
context.buy_close(position_symbol, short_qty, price, order_type=OrderType.MARKET)
context.buy_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
elif total_qty == 0:
if signal == 1:
context.logger.info("空仓开多：价格突破 20 日最高价")
context.buy_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
elif signal == -1:
context.logger.info("空仓开空：价格跌破 20 日最低价")
context.sell_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
performance = bigtrader.run(
market=bigtrader.Market.CN_FUTURE,
frequency=bigtrader.Frequency.DAILY,
start_date="2020-01-01",
end_date=datetime.now().strftime("%Y-%m-%d"),
capital_base=5_000_000,
initialize=initialize,
handle_data=handle_data,
order_price_field_buy='open',
order_price_field_sell='open',
)
期货日频策略二：双均线交叉趋势跟踪（螺纹钢 rb）
短期均线（SMA5）上穿长期均线（SMA20）时金叉做多，下穿时死叉做空，持仓期间自动移仓换月。
from datetime import datetime
from bigquant import bigtrader, dai
from bigtrader.constant import Direction, OrderType
def initialize(context: bigtrader.IContext):
context.logger.info("初始化：计算双均线交叉信号...")
short_period = 5
long_period = 20
sql = f"""
SELECT
date,
instrument,
dominant,
close,
m_avg(close, {short_period}) AS sma_short,
m_avg(close, {long_period}) AS sma_long,
m_lag(m_avg(close, {short_period}), 1) AS sma_short_prev,
m_lag(m_avg(close, {long_period}), 1) AS sma_long_prev,
CASE
WHEN sma_short > sma_long AND sma_short_prev <= sma_long_prev THEN 1
WHEN sma_short < sma_long AND sma_short_prev >= sma_long_prev THEN -1
ELSE 0
END AS signal
FROM cn_future_bar1d
LEFT JOIN cn_future_dominant USING (date, instrument)
WHERE instrument = 'rb8888.SHF'
"""
df = dai.query(sql, filters={
"date": [
context.add_trading_days(context.start_date, -(long_period + 10)),
context.end_date
]
}).df()
context.data = df
context.lots = 1
context.logger.info(f"数据加载完成，共 {len(df)} 条记录")
def handle_data(context: bigtrader.IContext, data: bigtrader.IBarData):
today = data.current_dt.strftime('%Y-%m-%d')
df_today = context.data[context.data['date'] == today]
if df_today.empty:
return
signal = int(df_today['signal'].values[0])
price = float(df_today['close'].values[0])
dominant_symbol = df_today['dominant'].values[0]
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
total_qty = long_qty + short_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)
position_symbol = dominant_symbol
# ── 信号交易 ──────────────────────────────────────────────
if long_qty > 0:
if signal == -1:
context.logger.info("死叉：平多开空")
context.sell_close(position_symbol, long_qty, price, order_type=OrderType.MARKET)
context.sell_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
elif short_qty > 0:
if signal == 1:
context.logger.info("金叉：平空开多")
context.buy_close(position_symbol, short_qty, price, order_type=OrderType.MARKET)
context.buy_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
elif total_qty == 0:
if signal == 1:
context.logger.info("空仓金叉：开多")
context.buy_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
elif signal == -1:
context.logger.info("空仓死叉：开空")
context.sell_open(dominant_symbol, context.lots, price, order_type=OrderType.MARKET)
performance = bigtrader.run(
market=bigtrader.Market.CN_FUTURE,
frequency=bigtrader.Frequency.DAILY,
start_date="2020-01-01",
end_date=datetime.now().strftime("%Y-%m-%d"),
capital_base=1_000_000,
initialize=initialize,
handle_data=handle_data,
order_price_field_buy='open',
order_price_field_sell='open',
)
期权日频策略一：保护性看跌策略（Protective Put）
持有沪深300ETF（510300.SH）现货多头仓位，同时买入下月到期的平值或虚值1档看跌期权作为下行保护。每5个交易日检查并调整期权保护仓位，期权仓位与ETF现货仓位保持1:1对冲比例。
import math
from datetime import datetime
from bigquant import bigtrader, dai
from dateutil.relativedelta import relativedelta
def get_next_month(cur_date):
date = datetime.strptime(cur_date, '%Y-%m-%d')
next_month_date = date + relativedelta(months=1)
return f"{next_month_date.year % 100:02d}{next_month_date.month:02d}"
def get_option_contracts(my_month, option_type, underlying, current_date):
sql = f"""
SELECT strike_price, english_name, instrument
FROM cn_option_basic_info
WHERE english_name LIKE '{underlying}{option_type}{my_month}%'
AND list_date <= '{current_date}'
ORDER BY strike_price
"""
try:
return dai.query(sql).df()
except Exception:
return None
def find_nearest_option(target_price, my_month, option_type, underlying, current_date):
contracts_df = get_option_contracts(my_month, option_type, underlying, current_date)
if contracts_df is None or contracts_df.empty:
return None, None, None
contracts_df['price_diff'] = abs(contracts_df['strike_price'] - target_price)
nearest = contracts_df.loc[contracts_df['price_diff'].idxmin()]
return nearest['strike_price'], nearest['english_name'], nearest['instrument']
def initialize(context: bigtrader.IContext):
context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))
context.etf = "510300.SH"
context.option_prefix = "510300"
context.rebalance_days = 5
context.otm_level = 1
context.strike_step = 0.05
context.hedge_ratio = 1.0
context.etf_position_ratio = 0.80
sql = """
SELECT date, instrument, close
FROM cn_fund_bar1d
WHERE instrument = '510300.SH'
ORDER BY date
"""
df = dai.query(sql, filters={"date": [context.start_date, context.end_date]}).df()
context.etf_data = df
def handle_data(context: bigtrader.IContext, data: bigtrader.IBarData):
current_date = context.current_dt.strftime('%Y-%m-%d')
etf_price = data.current(context.etf, "close")
if etf_price is None or math.isnan(etf_price) or etf_price <= 0:
return
# 维持 ETF 现货多头仓位
etf_position = context.portfolio.positions.get(context.etf)
etf_long_qty = etf_position.long_avail_qty() if etf_position else 0
target_etf_qty = int(context.portfolio.total_value * context.etf_position_ratio / etf_price / 100) * 100
if etf_long_qty == 0 and target_etf_qty > 0:
context.order_target(context.etf, target_etf_qty)
# 确定下月看跌期权合约
my_month = get_next_month(current_date)
put_target_price = etf_price - context.strike_step * context.otm_level
put_strike, put_contract, put_instrument = find_nearest_option(
put_target_price, my_month, 'P', context.option_prefix, current_date
)
if put_strike is None:
return
context.subscribe_bar(put_instrument)
if context.trading_day_index % context.rebalance_days != 0:
return
put_price = data.current(put_instrument, "close")
if put_price is None or math.isnan(put_price) or put_price <= 0:
return
current_etf_qty = etf_long_qty if etf_long_qty > 0 else target_etf_qty
target_put_qty = max(1, int(current_etf_qty * context.hedge_ratio / 10000))
# 调整期权仓位（省略持仓扫描逻辑，完整代码见原始文档）
context.buy_open(put_instrument, target_put_qty)
performance = bigtrader.run(
market=bigtrader.Market.CN_STOCK_OPTION,
frequency=bigtrader.Frequency.DAILY,
start_date="2026-01-01",
end_date=datetime.now().strftime("%Y-%m-%d"),
capital_base=1000000,
instruments=["510300.SH"],
benchmark="510300.SH",
initialize=initialize,
handle_data=handle_data,
)
期权日频策略二：牛市看涨期权价差策略（Bull Call Spread）
买入平值（ATM）看涨期权 + 卖出虚值（OTM）看涨期权，两腿合约数量相同，净权利金支出有限。每 5 个交易日检查并调仓，选择下月到期合约。
- 最大收益 = (卖出行权价 - 买入行权价) × 合约单位 × 张数 - 净权利金
- 最大亏损 = 净权利金支出（有限风险）
- 适用场景：温和看涨，希望以较低成本参与上涨行情
import math
from datetime import datetime
from bigquant import bigtrader, dai
from dateutil.relativedelta import relativedelta
def initialize(context: bigtrader.IContext):
context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))
context.sh50_etf = "510050.SH"
context.rebalance_days = 5
context.capital_ratio = 0.05
context.buy_otm_level = 0
context.sell_otm_level = 2
context.strike_step = 0.05
# 预缓存 ETF 日线数据
etf_df = dai.query(
"SELECT date, instrument, close FROM cn_fund_bar1d WHERE instrument = '510050.SH' ORDER BY date",
filters={"date": [context.start_date, context.end_date]}
).df()
context.etf_data = etf_df
# 预缓存所有 510050 看涨期权基本信息
option_df = dai.query(
"SELECT instrument, english_name, strike_price, list_date FROM cn_option_basic_info WHERE english_name LIKE '510050C%' ORDER BY english_name"
).df()
context.option_info_by_instrument = {}
context.option_contracts_by_month = {}
for _, row in option_df.iterrows():
context.option_info_by_instrument[row['instrument']] = {
'english_name': row['english_name'],
'strike_price': row['strike_price'],
}
month_key = row['english_name'][7:11]
if month_key not in context.option_contracts_by_month:
context.option_contracts_by_month[month_key] = []
context.option_contracts_by_month[month_key].append({
'instrument': row['instrument'],
'english_name': row['english_name'],
'strike_price': row['strike_price'],
'list_date': str(row['list_date']),
})
def get_next_month(cur_date):
date = datetime.strptime(cur_date, '%Y-%m-%d')
next_month_date = date + relativedelta(months=1)
return f"{next_month_date.year % 100:02d}{next_month_date.month:02d}"
def find_nearest_option(context, target_price, my_month, current_date):
contracts = context.option_contracts_by_month.get(my_month, [])
valid = [c for c in contracts if c['list_date'] <= current_date]
if not valid:
return None, None, None
nearest = min(valid, key=lambda c: abs(c['strike_price'] - target_price))
return nearest['strike_price'], nearest['english_name'], nearest['instrument']
def handle_data(context: bigtrader.IContext, data: bigtrader.IBarData):
current_date = context.current_dt.strftime('%Y-%m-%d')
current_etf_price = data.current(context.sh50_etf, "close")
if current_etf_price is None or math.isnan(current_etf_price):
return
my_month = get_next_month(current_date)
buy_strike, buy_contract, buy_instrument = find_nearest_option(
context, current_etf_price + context.strike_step * context.buy_otm_level, my_month, current_date)
sell_strike, sell_contract, sell_instrument = find_nearest_option(
context, current_etf_price + context.strike_step * context.sell_otm_level, my_month, current_date)
if buy_strike is None or sell_strike is None or buy_strike >= sell_strike:
return
if context.trading_day_index % context.rebalance_days != 0:
return
buy_price = data.current(buy_instrument, "close")
sell_price = data.current(sell_instrument, "close")
if buy_price is None or sell_price is None or math.isnan(buy_price) or math.isnan(sell_price):
return
if buy_price <= 0 or sell_price <= 0:
return
net_cost_per_contract = (buy_price - sell_price) * 10000
if net_cost_per_contract <= 0:
return
quantity = int(context.portfolio.total_value * context.capital_ratio / net_cost_per_contract)
if quantity <= 0:
return
# 买入腿 + 卖出腿（省略持仓扫描换仓逻辑，完整代码见原始文档）
context.buy_open(buy_instrument, quantity)
context.sell_open(sell_instrument, quantity)
performance = bigtrader.run(
market=bigtrader.Market.CN_STOCK_OPTION,
frequency=bigtrader.Frequency.DAILY,
start_date="2025-01-01",
end_date=datetime.now().strftime("%Y-%m-%d"),
capital_base=500000,
instruments=["510050.SH"],
benchmark='510050.SH',
initialize=initialize,
handle_data=handle_data,
)
\