2385 BBI 强势市场补涨 ETF 轮动策略 BBI指标-大盘涨跌-宽基轮动 模型 44521

 

1. 市场强度判断体系：

• 实时监控 5 大核心指数（上证、深成指、创业板、中证 1000、红利指数）
• 计算前一日涨幅最大的指数作为市场风向标
• 设置涨幅阈值（默认 0%），仅当领涨指数突破阈值时触发交易

2. 多空信号生成机制：

• 采用 BBI 多空指标（3/6/12/24 周期均线综合值）
• 30 分钟线级别计算 BBI 比值（BBI 值 / 当前价）
• 筛选 BBI 比值最低的指数，判定为短期超跌品种
• 双重确认：领涨指数强势 + 弱势指数 BBI<1 时开仓

3. ETF 动态轮动规则：

• 构建 13 组指数 - ETF 对应组合（涵盖主流宽基与策略指数）
• 自动排除上市未满 20 个交易日的 ETF
• 全仓集中持有单一弱势指数对应的 ETF
• 无符合条件的品种时自动清仓

4. 风险控制体系：

• 逆向选择：专门捕捉 "强势市场中补涨品种"
• 强制冷却：新上市 ETF 隔离期机制
• 高频监控：每 30 分钟刷新交易信号
• 滑点控制：0.1% 的固定滑点缓冲

策略代码

# 导入函数模块
import jqdata
from   jqlib.technical_analysis  import *
from   jqdata import *


# 初始化函数，设定基准等等
def initialize(context):

    # 过滤掉order系列API产生的比error级别低的log
    # 默认是 'debug' 参数。最低的级别，日志信息最多
    # 系统推荐尽量使用'debug'参数或不显式设置，方便找出所有错误    
    log.set_level('order', 'error')
    
    # 开启动态复权模式(使用真实价格)
    set_option('use_real_price', True)
    
    # 避免使用未来数据
    set_option('avoid_future_data', True)
    
    # 交易量不超过实际成交量的 0.1    
    set_option('order_volume_ratio',0.1)
    
    # 设定基准
    # '000300.XSHG' 沪深300指数    
    set_benchmark('000300.XSHG')
    
    # 设置滑点
    set_slippage(FixedSlippage(0.001))
    
    # 基金和股票的手续费设置有些不同
    set_order_cost(OrderCost(close_tax=0.000, 
                   open_commission=0.00006, 
                   close_commission=0.00006, 
                   min_commission=0), type='fund')



    # 初始化全局变量
    
    # g. 开头的是全局变量
    # 一经声明整个程序都可以使用

    # 最强指数涨了多少，可以开仓
    g.dapan_threshold =0
    g.signal= 'BUY'
    

    # 基金上市了多久可以买卖
    g.lag = 20
    g.decrease_days = 0 
    g.increase_days = 0 
    
    # bbi动量的单位
    g.unit = '30m'
    # g.unit = '1d'
    
    
    g.zs_list = [
        '000001.XSHG', # 上证
        '399001.XSHE', # 深成指
        '399006.XSHE', # 创业板指数
        '000852.XSHG', # 中证1000指数
        '000015.XSHG'  # 红利指数
        ]
        
        
    
    # 指数、基金对, 所有想交易的etf都可以，会自动过滤掉交易时没有上市的
    g.ETF_list =  {
      
        '399905.XSHE':'159902.XSHE',  # 中小板指
        '399632.XSHE':'159901.XSHE',  # 深100etf
        '000016.XSHG':'510050.XSHG',  # 上证50
        '000010.XSHG':'510180.XSHG',  # 上证180
        
        '000852.XSHG':'512100.XSHG',  # 中证1000etf
        '399295.XSHE':'159966.XSHE',  # 创蓝筹
        '399958.XSHE':'159967.XSHE',  # 创成长
        '000015.XSHG':'510880.XSHG',  # 红利ETF
        '399324.XSHE':'159905.XSHE',  # 深红利
        '399006.XSHE':'159915.XSHE',  # 创业板
        '000300.XSHG':'510300.XSHG',  # 沪深300
        '000905.XSHG':'510500.XSHG',  # 中证500
        '399673.XSHE':'159949.XSHE',  # 创业板50
        '000688.XSHG':'588000.XSHG'   # 科创50

    }
    
    
    g.not_ipo_list = g.ETF_list.copy()
    
    g.available_indexs = []

    # 每日9:15调用函数
    # 分析 ETF 上市情况
    run_daily(check_etf, time='9:15')

    
    # 每日 11:15 实际进行交易
    run_daily(check_trade, time='11:15')
    # run_daily(check_trade, time='14:45')
    # run_weekly(check_trade, weekday=1,time='11:15')
  


# 获取指定日期之前或之后 N 个交易日的日期
def get_N_trade_date(date, N, is_before=True):
    
    # get_all_trade_days() 获取所有交易日
    # 获取所有交易日, 不需要传入参数
    # 返回一个包含所有交易日的 numpy.ndarray, 每个元素为一个datetime.date类型
    # 注： 需导入 jqdata 模块，即在策略或研究起始位置加入 import jqdata
    all_date = pd.Series(get_all_trade_days())
    
    # 如果输入数据是字符串就对数据进行转换
    # 最后用 date() 得到日期数据
    if isinstance(date, str):
        date = datetime.datetime.strptime(date, '%Y-%m-%d').date()
    
    # 如果date参数的数据类型为 datetime.datetime
    # 则直接使用date()函数获取日期
    if isinstance(date, datetime.datetime):
        date = date.date()

    # 如果是准备获取更早期的数据
    if is_before:
        
        # 这里的写法非常巧妙
        return all_date[all_date <= date].tail(N).values[0]
    
    # 否则是获取更晚的数据
    else:
        return all_date[all_date >= date].head(N).values[-1]    

    

# 确认ETF已经上市交易
# 对待交易基金列表中的每只基金都做分析判断是否可以交易
# 有可能在列表中的基金在回测的时间点还不能交易
def check_etf(context):
    
    # 在初始化函数中已经赋值待交易的 ETF 列表
    # 至少刚开始回测时不会为空
    if len(g.not_ipo_list) == 0:
        return 
    
    idxs = []
    
    # 确保交易标的已经上市g.lag个交易日以上
    yesterday = context.previous_date
    
    # 今天的前g.lag个交易日的日期
    list_date = get_N_trade_date(yesterday, g.lag)  
  
    # 上个交易日之前上市的所有基金
    all_funds = get_all_securities(types='fund', date=yesterday) 
    
    # 上个交易日之前就已经存在的指数
    all_idxes = get_all_securities(types='index', date=yesterday) 
    
    # 遍历待交易基金列表中的每只ETF
    for idx in g.not_ipo_list:
        
        if idx in all_idxes.index:
            
            # 指数已经在要求的日期前上市
            if all_idxes.loc[idx].start_date <= list_date:  
                
                symbol = g.not_ipo_list[idx]
                
                if symbol in all_funds.index:
                    
                    # 对应的基金也已经在要求的日期前上市
                    if all_funds.loc[symbol].start_date <= list_date:
                        
                        # 将这只基金加入可交易列表中
                        g.available_indexs.append(idx)  
                        
                        idxs.append(idx)
    
    # 删除不可交易列表中的对应基金
    for idx in idxs:
        
        del g.not_ipo_list[idx]
    
    log.info('输出不可交易基金列表：')    
    log.info(g.not_ipo_list)
    
    return    

关键函数解锁后查看：

2025-02-24