首页 编写策略 数据平台 策略社区 我的策略 宽客学院 知识库
免费注册
BigQuant使用文档

分钟回测demos

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

股票分钟策略一：RSI 超买超卖策略

标的：平安银行（000001.SZ），频率：1分钟。RSI(14) < 30 超卖买入，RSI(14) > 70 超买卖出，止损 -3%，止盈 +5%，尾盘（14:55）强制平仓。

from bigquant import bigtrader
from datetime import time

def is_trading_time(dt):
    t = dt.time()
    return time(9, 30) <= t <= time(11, 30) or time(13, 0) <= t <= time(15, 0)

def calc_rsi(prices, period=14):
    if len(prices) < period + 1:
        return None
    gains, losses = [], []
    for i in range(1, len(prices)):
        delta = prices[i] - prices[i - 1]
        gains.append(max(delta, 0))
        losses.append(max(-delta, 0))
    avg_gain = sum(gains[:period]) / period
    avg_loss = sum(losses[:period]) / period
    for i in range(period, len(gains)):
        avg_gain = (avg_gain * (period - 1) + gains[i]) / period
        avg_loss = (avg_loss * (period - 1) + losses[i]) / period
    if avg_loss == 0:
        return 100.0
    return 100 - (100 / (1 + avg_gain / avg_loss))

RSI_PERIOD    = 14
RSI_OVERSOLD  = 30
RSI_OVERBOUGHT = 70
STOP_LOSS     = -0.03
TAKE_PROFIT   = 0.05
CLOSE_TIME    = time(14, 55)
INSTRUMENT    = "000001.SZ"

def initialize(context: bigtrader.IContext):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))
    context.entry_price = {}

def before_trading_start(context: bigtrader.IContext, data):
    context.subscribe_bar(context.instruments, period='1m')

def handle_data(context: bigtrader.IContext, data):
    now = data.current_dt
    if not is_trading_time(now):
        return

    # 尾盘强制平仓
    if now.time() >= CLOSE_TIME:
        positions = context.get_account_positions()
        if INSTRUMENT in positions and positions[INSTRUMENT].current_qty > 0:
            context.order_target_percent(INSTRUMENT, 0)
            context.entry_price.pop(INSTRUMENT, None)
        return

    hist = data.history(INSTRUMENT, "close", RSI_PERIOD + 20, "1m")
    if hist is None or len(hist) < RSI_PERIOD + 1:
        return

    close_prices  = list(hist)
    rsi           = calc_rsi(close_prices, RSI_PERIOD)
    if rsi is None:
        return

    current_price = close_prices[-1]
    positions     = context.get_account_positions()
    holding       = INSTRUMENT in positions and positions[INSTRUMENT].current_qty > 0

    # 止盈止损检查
    if holding and INSTRUMENT in context.entry_price:
        entry   = context.entry_price[INSTRUMENT]
        pnl_pct = (current_price - entry) / entry
        if pnl_pct <= STOP_LOSS:
            context.order_target_percent(INSTRUMENT, 0)
            context.entry_price.pop(INSTRUMENT, None)
            return
        if pnl_pct >= TAKE_PROFIT:
            context.order_target_percent(INSTRUMENT, 0)
            context.entry_price.pop(INSTRUMENT, None)
            return

    # RSI 信号
    if not holding and rsi < RSI_OVERSOLD:
        context.order_target_percent(INSTRUMENT, 0.95)
        context.entry_price[INSTRUMENT] = current_price
    elif holding and rsi > RSI_OVERBOUGHT:
        context.order_target_percent(INSTRUMENT, 0)
        context.entry_price.pop(INSTRUMENT, None)

performance = bigtrader.run(
    market=bigtrader.Market.CN_STOCK,
    frequency=bigtrader.Frequency.MINUTE,
    instruments=[INSTRUMENT],
    start_date="2026-01-01",
    end_date="2026-05-09",
    capital_base=1_000_000,
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_data=handle_data,
    order_price_field_buy="close",
    order_price_field_sell="close",
)

股票分钟策略二：均线突破策略（沪深300成分股）

每日从沪深300中选出5日动量最强的前20只股票，盘中分钟MA5上穿MA20（金叉）买入，下穿（死叉）卖出，14:55尾盘强制平仓，每只股票等权仓位 1/20。

from bigquant import bigtrader, dai
from datetime import time

def is_trading_time(dt):
    t = dt.time()
    return time(9, 30) <= t <= time(11, 30) or time(13, 0) <= t <= time(15, 0)

def ma(prices, n):
    return sum(prices[-n:]) / n

MA_SHORT   = 5
MA_LONG    = 20
STOCK_NUM  = 20
CLOSE_TIME = time(14, 55)

def initialize(context: bigtrader.IContext):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))

    sql = """
    SELECT date, instrument, close / m_lag(close, 5) - 1 AS momentum_5d
    FROM cn_stock_prefactors
    WHERE is_hs300 = 1 AND st_status = 0 AND suspended = 0 AND list_days > 365
    ORDER BY date, momentum_5d DESC
    """
    df = dai.query(sql, filters={"date": [
        context.add_trading_days(context.start_date, -10),
        context.end_date,
    ]}).df()

    context.daily_selected = (
        df.groupby("date").head(STOCK_NUM)
        .groupby("date")["instrument"].apply(list).to_dict()
    )
    context.daily_selected = {str(k)[:10]: v for k, v in context.daily_selected.items()}

def before_trading_start(context: bigtrader.IContext, data):
    today = data.current_dt.strftime("%Y-%m-%d")
    context.selected = context.daily_selected.get(today, [])
    context.subscribe_bar(context.selected, period="1m")

def handle_data(context: bigtrader.IContext, data):
    now = data.current_dt
    if not is_trading_time(now):
        return

    today = now.strftime("%Y-%m-%d")
    context.selected = context.daily_selected.get(today, [])

    if now.time() >= CLOSE_TIME:
        for ins, pos in context.get_account_positions().items():
            if pos.current_qty > 0:
                context.order_target_percent(ins, 0)
        return

    positions   = context.get_account_positions()
    holding_set = {ins for ins, pos in positions.items() if pos.current_qty > 0}

    for ins in list(holding_set):
        if ins not in context.selected:
            context.order_target_percent(ins, 0)
            holding_set.discard(ins)

    for instrument in context.selected:
        hist = data.history(instrument, "close", MA_LONG + 1, "1m")
        if hist is None or len(hist) < MA_LONG + 1:
            continue
        prices  = list(hist)
        ma_s    = ma(prices, MA_SHORT)
        ma_l    = ma(prices, MA_LONG)
        holding = instrument in holding_set

        if not holding and ma_s > ma_l:
            context.order_target_percent(instrument, 1.0 / STOCK_NUM)
            holding_set.add(instrument)
        elif holding and ma_s < ma_l:
            context.order_target_percent(instrument, 0)
            holding_set.discard(instrument)

performance = bigtrader.run(
    market=bigtrader.Market.CN_STOCK,
    frequency=bigtrader.Frequency.MINUTE,
    start_date="2026-01-01",
    end_date="2026-01-20",
    capital_base=1_000_000,
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_data=handle_data,
    order_price_field_buy="close",
    order_price_field_sell="close",
    benchmark="000300.SH",
)

ETF 分钟策略一：单 ETF 均线趋势策略

标的：沪深300ETF（510300.SH），MA5 > MA20 多头趋势持有，MA5 < MA20 空仓，止损 -2%，止盈 +3%，尾盘强制平仓。

from bigquant import bigtrader
from datetime import time

def is_trading_time(dt):
    t = dt.time()
    return time(9, 30) <= t <= time(11, 30) or time(13, 0) <= t <= time(15, 0)

def ma(prices, n):
    return sum(prices[-n:]) / n

MA_SHORT    = 5
MA_LONG     = 20
STOP_LOSS   = -0.02
TAKE_PROFIT = 0.03
CLOSE_TIME  = time(14, 55)
INSTRUMENT  = "510300.SH"

def initialize(context: bigtrader.IContext):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))
    context.entry_price = None

def before_trading_start(context: bigtrader.IContext, data):
    context.subscribe_bar(context.instruments, period="1m")
    context.entry_price = None

def handle_data(context: bigtrader.IContext, data):
    now = data.current_dt
    if not is_trading_time(now):
        return

    if now.time() >= CLOSE_TIME:
        positions = context.get_account_positions()
        if INSTRUMENT in positions and positions[INSTRUMENT].current_qty > 0:
            context.order_target_percent(INSTRUMENT, 0)
            context.entry_price = None
        return

    hist = data.history(INSTRUMENT, "close", MA_LONG + 1, "1m")
    if hist is None or len(hist) < MA_LONG + 1:
        return

    prices        = list(hist)
    ma_s          = ma(prices, MA_SHORT)
    ma_l          = ma(prices, MA_LONG)
    current_price = prices[-1]

    positions = context.get_account_positions()
    holding   = INSTRUMENT in positions and positions[INSTRUMENT].current_qty > 0

    if holding and context.entry_price is not None:
        pnl_pct = (current_price - context.entry_price) / context.entry_price
        if pnl_pct <= STOP_LOSS:
            context.order_target_percent(INSTRUMENT, 0)
            context.entry_price = None
            return
        if pnl_pct >= TAKE_PROFIT:
            context.order_target_percent(INSTRUMENT, 0)
            context.entry_price = None
            return

    if not holding and ma_s > ma_l:
        context.order_target_percent(INSTRUMENT, 0.95)
        context.entry_price = current_price
    elif holding and ma_s < ma_l:
        context.order_target_percent(INSTRUMENT, 0)
        context.entry_price = None

performance = bigtrader.run(
    market=bigtrader.Market.CN_FUND,
    frequency=bigtrader.Frequency.MINUTE,
    instruments=[INSTRUMENT],
    start_date="2026-04-25",
    end_date="2026-05-09",
    capital_base=1_000_000,
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_data=handle_data,
    order_price_field_buy="close",
    order_price_field_sell="close",
    benchmark="000300.SH",
)

ETF 分钟策略二：多 ETF 动量轮动

标的池：沪深300ETF / 纳指ETF / 黄金ETF / 国债ETF / 科创50ETF。每日开盘前用前一日日线数据计算各ETF的5日动量，选出动量最强的前3只；盘中分钟MA5 > MA20持有，MA5 < MA20空仓，每只ETF等权仓位 1/3，尾盘强制平仓。

from bigquant import bigtrader, dai
from datetime import time

def is_trading_time(dt):
    t = dt.time()
    return time(9, 30) <= t <= time(11, 30) or time(13, 0) <= t <= time(15, 0)

def ma(prices, n):
    return sum(prices[-n:]) / n

MA_SHORT   = 5
MA_LONG    = 20
ETF_NUM    = 3
CLOSE_TIME = time(14, 55)
ETF_POOL   = ["510300.SH", "513100.SH", "518880.SH", "511010.SH", "588000.SH", "159941.SZ", "513520.SH"]

def initialize(context: bigtrader.IContext):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))

    sql = """
    SELECT date, instrument, close / m_lag(close, 5) - 1 AS momentum_5d
    FROM cn_fund_bar1d
    WHERE instrument IN $etf_pool
    ORDER BY date, momentum_5d DESC
    """
    df = dai.query(sql, filters={"date": [
        context.add_trading_days(context.start_date, -10),
        context.end_date,
    ]}, params={"etf_pool": ETF_POOL}).df()

    context.daily_selected = (
        df.dropna(subset=["momentum_5d"])
        .groupby("date").head(ETF_NUM)
        .groupby("date")["instrument"].apply(list).to_dict()
    )
    context.daily_selected = {str(k)[:10]: v for k, v in context.daily_selected.items()}

def before_trading_start(context: bigtrader.IContext, data):
    today = data.current_dt.strftime("%Y-%m-%d")
    context.selected = context.daily_selected.get(today, [])
    if context.selected:
        context.subscribe_bar(context.selected, period="1m")

def handle_data(context: bigtrader.IContext, data):
    now      = data.current_dt
    selected = context.selected
    if not is_trading_time(now):
        return

    if now.time() >= CLOSE_TIME:
        for ins, pos in context.get_account_positions().items():
            if pos.current_qty > 0:
                context.order_target_percent(ins, 0)
        return

    positions   = context.get_account_positions()
    holding_set = {ins for ins, pos in positions.items() if pos.current_qty > 0}

    for ins in list(holding_set):
        if ins not in selected:
            context.order_target_percent(ins, 0)
            holding_set.discard(ins)

    for instrument in selected:
        hist = data.history(instrument, "close", MA_LONG + 1, "1m")
        if hist is None or len(hist) < MA_LONG + 1:
            continue
        prices  = list(hist)
        ma_s    = ma(prices, MA_SHORT)
        ma_l    = ma(prices, MA_LONG)
        holding = instrument in holding_set

        if not holding and ma_s > ma_l:
            context.order_target_percent(instrument, 1.0 / ETF_NUM)
            holding_set.add(instrument)
        elif holding and ma_s < ma_l:
            context.order_target_percent(instrument, 0)
            holding_set.discard(instrument)

performance = bigtrader.run(
    market=bigtrader.Market.CN_FUND,
    frequency=bigtrader.Frequency.MINUTE,
    start_date="2026-04-25",
    end_date="2026-05-09",
    capital_base=1_000_000,
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_data=handle_data,
    order_price_field_buy="close",
    order_price_field_sell="close",
    benchmark="000300.SH",
)

可转债分钟策略一：均线趋势策略

每日开盘前，从可转债池中按双低因子（低价格+低溢价率）选出前5只；盘中分钟MA5 > MA20持有，MA5 < MA20空仓，每只等权仓位 1/5，尾盘强制平仓。

from bigquant import bigtrader, dai
from datetime import time

def is_trading_time(dt):
    t = dt.time()
    return time(9, 30) <= t <= time(11, 30) or time(13, 0) <= t <= time(15, 0)

def ma(prices, n):
    return sum(prices[-n:]) / n

MA_SHORT   = 5
MA_LONG    = 20
CB_NUM     = 5
CLOSE_TIME = time(14, 55)

def initialize(context: bigtrader.IContext):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))

    sql = """
    SELECT date, instrument, close + conversion_premium_rate AS double_low
    FROM cn_cbond_bar1d
    INNER JOIN cn_cbond_analyze_metric USING (date, instrument)
    WHERE bond_balance > 1 AND remaining_days > 120
      AND conversion_premium_rate > -10 AND conversion_premium_rate < 50
      AND close > 80 AND close < 150
    ORDER BY date, double_low ASC
    """
    df = dai.query(sql, filters={"date": [
        context.add_trading_days(context.start_date, -5),
        context.end_date,
    ]}).df()

    context.daily_selected = (
        df.dropna(subset=["double_low"])
        .groupby("date").head(CB_NUM)
        .groupby("date")["instrument"].apply(list).to_dict()
    )
    context.daily_selected = {str(k)[:10]: v for k, v in context.daily_selected.items()}

def before_trading_start(context: bigtrader.IContext, data):
    today = data.current_dt.strftime("%Y-%m-%d")
    context.selected = context.daily_selected.get(today, [])
    if context.selected:
        context.subscribe_bar(context.selected, period="1m")

def handle_data(context: bigtrader.IContext, data):
    now      = data.current_dt
    selected = context.selected
    if not is_trading_time(now):
        return

    if now.time() >= CLOSE_TIME:
        for ins, pos in context.get_account_positions().items():
            if pos.current_qty > 0:
                context.order_target_percent(ins, 0)
        return

    positions   = context.get_account_positions()
    holding_set = {ins for ins, pos in positions.items() if pos.current_qty > 0}

    for ins in list(holding_set):
        if ins not in selected:
            context.order_target_percent(ins, 0)
            holding_set.discard(ins)

    for instrument in selected:
        hist = data.history(instrument, "close", MA_LONG + 1, "1m")
        if hist is None or len(hist) < MA_LONG + 1:
            continue
        prices  = list(hist)
        ma_s    = ma(prices, MA_SHORT)
        ma_l    = ma(prices, MA_LONG)
        holding = instrument in holding_set

        if not holding and ma_s > ma_l:
            context.order_target_percent(instrument, 1.0 / CB_NUM)
            holding_set.add(instrument)
        elif holding and ma_s < ma_l:
            context.order_target_percent(instrument, 0)
            holding_set.discard(instrument)

performance = bigtrader.run(
    market=bigtrader.Market.CN_CBOND,
    frequency=bigtrader.Frequency.MINUTE,
    start_date="2026-04-25",
    end_date="2026-05-09",
    capital_base=500_000,
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_data=handle_data,
    order_price_field_buy="close",
    order_price_field_sell="close",
    benchmark="000300.SH",
)

可转债分钟策略二：RSI 超买超卖策略

每日开盘前按双低因子选出前5只转债，盘中 RSI(14) < 30 超卖买入，RSI > 70 超买卖出，止损 -3%，止盈 +5%，尾盘强制平仓。

from bigquant import bigtrader, dai
from datetime import time

def is_trading_time(dt):
    t = dt.time()
    return time(9, 30) <= t <= time(11, 30) or time(13, 0) <= t <= time(15, 0)

def calc_rsi(prices, period=14):
    if len(prices) < period + 1:
        return None
    gains, losses = [], []
    for i in range(1, len(prices)):
        delta = prices[i] - prices[i - 1]
        gains.append(max(delta, 0))
        losses.append(max(-delta, 0))
    avg_gain = sum(gains[:period]) / period
    avg_loss = sum(losses[:period]) / period
    for i in range(period, len(gains)):
        avg_gain = (avg_gain * (period - 1) + gains[i]) / period
        avg_loss = (avg_loss * (period - 1) + losses[i]) / period
    if avg_loss == 0:
        return 100.0
    return 100 - (100 / (1 + avg_gain / avg_loss))

RSI_PERIOD    = 14
RSI_OVERSOLD  = 30
RSI_OVERBOUGHT = 70
STOP_LOSS     = -0.03
TAKE_PROFIT   = 0.05
CB_NUM        = 5
CLOSE_TIME    = time(14, 55)

def initialize(context: bigtrader.IContext):
    context.set_commission(bigtrader.PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))

    sql = """
    SELECT date, instrument, close + conversion_premium_rate AS double_low
    FROM cn_cbond_bar1d
    INNER JOIN cn_cbond_analyze_metric USING (date, instrument)
    WHERE bond_balance > 1 AND remaining_days > 120
      AND conversion_premium_rate > -10 AND conversion_premium_rate < 50
      AND close > 80 AND close < 150
    ORDER BY date, double_low ASC
    """
    df = dai.query(sql, filters={"date": [
        context.add_trading_days(context.start_date, -5),
        context.end_date,
    ]}).df()

    context.daily_selected = (
        df.dropna(subset=["double_low"])
        .groupby("date").head(CB_NUM)
        .groupby("date")["instrument"].apply(list).to_dict()
    )
    context.daily_selected = {str(k)[:10]: v for k, v in context.daily_selected.items()}

def before_trading_start(context: bigtrader.IContext, data):
    today = data.current_dt.strftime("%Y-%m-%d")
    context.selected = context.daily_selected.get(today, [])
    if context.selected:
        context.subscribe_bar(context.selected, period="1m")
    context.entry_price = {}

def handle_data(context: bigtrader.IContext, data):
    now      = data.current_dt
    selected = context.selected
    if not is_trading_time(now):
        return

    if now.time() >= CLOSE_TIME:
        for ins, pos in context.get_account_positions().items():
            if pos.current_qty > 0:
                context.order_target_percent(ins, 0)
                context.entry_price.pop(ins, None)
        return

    positions   = context.get_account_positions()
    holding_set = {ins for ins, pos in positions.items() if pos.current_qty > 0}

    for ins in list(holding_set):
        if ins not in selected:
            context.order_target_percent(ins, 0)
            holding_set.discard(ins)
            context.entry_price.pop(ins, None)

    for instrument in selected:
        hist = data.history(instrument, "close", RSI_PERIOD + 20, "1m")
        if hist is None or len(hist) < RSI_PERIOD + 1:
            continue
        prices  = list(hist)
        rsi     = calc_rsi(prices, RSI_PERIOD)
        if rsi is None:
            continue
        current_price = prices[-1]
        holding       = instrument in holding_set

        if holding and instrument in context.entry_price:
            entry   = context.entry_price[instrument]
            pnl_pct = (current_price - entry) / entry
            if pnl_pct <= STOP_LOSS:
                context.order_target_percent(instrument, 0)
                holding_set.discard(instrument)
                context.entry_price.pop(instrument, None)
                continue
            if pnl_pct >= TAKE_PROFIT:
                context.order_target_percent(instrument, 0)
                holding_set.discard(instrument)
                context.entry_price.pop(instrument, None)
                continue

        if not holding and rsi < RSI_OVERSOLD:
            context.order_target_percent(instrument, 1.0 / CB_NUM)
            holding_set.add(instrument)
            context.entry_price[instrument] = current_price
        elif holding and rsi > RSI_OVERBOUGHT:
            context.order_target_percent(instrument, 0)
            holding_set.discard(instrument)
            context.entry_price.pop(instrument, None)

performance = bigtrader.run(
    market=bigtrader.Market.CN_CBOND,
    frequency=bigtrader.Frequency.MINUTE,
    start_date="2026-04-25",
    end_date="2026-05-09",
    capital_base=500_000,
    initialize=initialize,
    before_trading_start=before_trading_start,
    handle_data=handle_data,
    order_price_field_buy="close",
    order_price_field_sell="close",
    benchmark="000300.SH",
)

\

{link}