# coding=utf-8 from __future__ import print_function, absolute_import from gm.api import * from MyTT import * """ 示例策略仅供参考，不建议直接实盘使用。 本策略采用布林线进行均值回归交易。当价格触及布林线上轨的时候进行卖出，当触及下轨的时候，进行买入。 """ # 策略中必须有init方法 def init(context): # 设置布林线的三个参数 context.maPeriod = 26 # 计算BOLL布林线中轨的参数 context.stdPeriod = 150 # 计算BOLL 标准差的参数 context.stdRange = 1 # 计算BOLL 上下轨和中轨距离的参数 # 设置要进行回测的合约 context.symbol2 = 'SZSE.399673' # 订阅&交易标的, 此处订阅的是600004 ['SHSE.600519', 'SHSE.600419'], context.symbol = 'SZSE.159949' context.symbolTwo = ['SZSE.399673','SZSE.159949'] context.period = max(context.maPeriod, context.stdPeriod, context.stdRange) + 1 # 订阅数据滑窗长度 # 订阅行情 subscribe(symbols= context.symbolTwo, frequency='1d', count=context.period) def on_bar(context, bars): # 获取数据滑窗，只要在init里面有订阅，在这里就可以取的到，返回值是pandas.DataFrame data2 = context.data(symbol=context.symbol, frequency='1d', count=context.period, fields='close,open,high,low,bob,eob') data = context.data(symbol=context.symbol2, frequency='1d', count=context.period, fields='close,open,high,low,bob,eob') # print(data2) #print(data.bob.values[-1]) close = data.close.values low = data.low.values high = data.high.values dopen = data.open.values H1_5 = EMA(close, 8) H2_5 = EMA(H1_5, 20) # print('h1-5:', H1_5[-4:]) # print('h2-5:', H2_5[-4:]) H1H2_CROSS = CROSS(H1_5, H2_5) H2H1_CROSS = CROSS(H2_5, H1_5) # print('cross H1 H2：', H1H2_CROSS) # print('cross H2 H1', H2H1_CROSS) # qrld rsv = (close - LLV(low, 7)) / (HHV(high, 7) - LLV(low, 7)) * 100 rsv = np.nan_to_num(rsv) # print('rsv :', rsv) Y0 = SMA(rsv, 3, 1) Y0 = np.nan_to_num(Y0) # print('Y0 :', Y0) Y1 = SMA(Y0, 3, 1) Y1 = np.nan_to_num(Y1) # print('Y1 :', Y1) CROSS_Y0_Y1 = CROSS(Y0, Y1) CROSS_Y1_Y0 = CROSS(Y1, Y0) RSV1 = (close - LLV(low, 38)) / (HHV(high, 38) - LLV(low, 38)) * 100 RSV1 = np.nan_to_num(RSV1) # print('RSV1 :', RSV1) Y2 = SMA(RSV1, 5, 1) Y2 = np.nan_to_num(Y2) # print('Y2 :', Y2) Y3 = SMA(Y2, 10, 1) Y3 = np.nan_to_num(Y3) # print('Y3 :', Y3) # lq N = 1 xopen = (REF(dopen, N) + REF(close, N)) / 2 xclose = close xhigh = MAX(high, xopen) xlow = MIN(low, xopen) volality = MA(xhigh - xlow, 8) h_line = MA(xclose, 5) + volality / 2 f_line = MA(xclose, 5) - volality / 2 bu = CROSS(xclose, h_line) sel = CROSS(f_line, xclose) var1 = BARSLAST(bu) var2 = BARSLAST(sel) # 布林带上轨 bollUpper = H1H2_CROSS # 布林带下轨 bollBottom = H2H1_CROSS # cross_kdj_up = cross_y0_y1[-1] and (Y0[-1] > Y1[-1]); a1 = (H1_5[-1]-H2_5[-1])/((H1_5[-1]+H2_5[-1])/2) b1 = (Y2[-1] - Y3[-1]) / 100 c1 = (Y2[-1] + Y3[-1]) / 2 if(CROSS_Y0_Y1[-1] == 0 or CROSS_Y1_Y0[-1] == 0): return else: #print('有信号：CROSS_Y0_Y1[-1]:'+str(CROSS_Y0_Y1[-1])) #print('有信号：CROSS_Y1_Y0[-1]:'+str(CROSS_Y1_Y0[-1])) pos = list(filter(lambda x:x['symbol']==context.symbol,get_position())) if(Y0[-1]>Y1[-1] and (b1>0) and ( a1>-0.02 or a1<0.02)): #BUY if( (a1<-0.04 or b1<-0.17)): # print('a#1<-0.04 or b1<-0.17 cant buy') return # 交易逻辑与下单 # 当有持仓，且股价穿过BOLL上界的时候卖出股票。 # # 当没有持仓，且股价穿过BOLL下界的时候买入股票。 if not pos: cash_info = get_cash() # 返回 DictLikeObject balance = cash_info['balance'] # 仍然可以通过键访问 # print("balance b:"+str(balance)) volumeTmp = int(balance/data2.close.values[-1]) volumeTmp = (volumeTmp // 100) * 100 - 500 order_volume(symbol=context.symbol, volume=volumeTmp, side=OrderSide_Buy, order_type=OrderType_Limit, position_effect=PositionEffect_Open, price=data2.close.values[-1]) elif(Y0[-1]<=Y1[-1] and ( a1>-0.02 or a1<0.02)): #sell # print("a1:"+str(a1)) if(a1>0.05): print("a1>0.05") return if pos : cash_info = get_cash() # 返回 DictLikeObject balance = cash_info['balance'] # 仍然可以通过键访问 #print("balance s:"+str(balance)) #print("pos:"+str(pos)) volume_value = next((p['volume'] for p in pos if p['symbol'] == 'SZSE.159949'), None) volumeTmp = int(volume_value) order_volume(symbol=context.symbol, volume=volumeTmp, side=OrderSide_Sell, order_type=OrderType_Limit, position_effect=PositionEffect_Close, price=data2.close.values[-1]) # 交易逻辑与下单 # 当有持仓，且股价穿过BOLL上界的时候卖出股票。 # if pos and (bollBottom[-1] or bollUpper[-1]) and (H1_5[-1]=H2_5[-1])): # cash_info = get_cash() # 返回 DictLikeObject # balance = cash_info['balance'] # 仍然可以通过键访问 # print("balance b:"+str(balance)) # volumeTmp = int(balance/data2.close.values[-1]) # volumeTmp = (volumeTmp // 100) * 100 - 500 # order_volume(symbol=context.symbol, volume=volumeTmp, side=OrderSide_Buy, # order_type=OrderType_Limit, position_effect=PositionEffect_Open, price=data2.close.values[-1]) def on_order_status(context, order): # 标的代码 symbol = order['symbol'] # 委托价格 price = order['price'] # 委托数量 volume = order['volume'] # 目标仓位 target_percent = order['target_percent'] # 查看下单后的委托状态，等于3代表委托全部成交 status = order['status'] # 买卖方向，1为买入，2为卖出 side = order['side'] # 开平仓类型，1为开仓，2为平仓 effect = order['position_effect'] # 委托类型，1为限价委托，2为市价委托 order_type = order['order_type'] if status == 3: if effect == 1: if side == 1: side_effect = '开多仓' else: side_effect = '开空仓' else: if side == 1: side_effect = '平空仓' else: side_effect = '平多仓' order_type_word = '限价' if order_type==1 else '市价' print('{}:标的：{}，操作：以{}{}，委托价格：{}，委托数量：{}'.format(context.now,symbol,order_type_word,side_effect,price,volume)) def on_backtest_finished(context, indicator): print('*'*50) print('回测已完成，请通过右上角“回测历史”功能查询详情。') if __name__ == '__main__': ''' strategy_id策略ID,由系统生成 filename文件名,请与本文件名保持一致 mode实时模式:MODE_LIVE回测模式:MODE_BACKTEST token绑定计算机的ID,可在系统设置-密钥管理中生成 backtest_start_time回测开始时间 backtest_end_time回测结束时间 backtest_adjust股票复权方式不复权:ADJUST_NONE前复权:ADJUST_PREV后复权:ADJUST_POST backtest_initial_cash回测初始资金 backtest_commission_ratio回测佣金比例 backtest_slippage_ratio回测滑点比例 backtest_match_mode市价撮合模式，以下一tick/bar开盘价撮合:0，以当前tick/bar收盘价撮合：1 ''' run(strategy_id='3b7f1e3f-8e41-11f0-aae9-00155d8e2a12', filename='main.py', mode=MODE_BACKTEST, token='a130ee5e899369902d1e95d0a116b297ad1edbfb', backtest_start_time='2016-08-01 08:00:00', backtest_end_time='2025-09-19 15:30:00', backtest_adjust=ADJUST_PREV, backtest_initial_cash=10000, backtest_commission_ratio=0.0001, backtest_slippage_ratio=0.0001, backtest_match_mode=1)