|  | @@ -12,31 +12,130 @@ import pymysql
 | 
	
		
			
				|  |  |  import multiprocessing as mp
 | 
	
		
			
				|  |  |  import math
 | 
	
		
			
				|  |  |  import psutil
 | 
	
		
			
				|  |  | +import datetime
 | 
	
		
			
				|  |  |  from apscheduler.schedulers.blocking import BlockingScheduler
 | 
	
		
			
				|  |  | +import sys
 | 
	
		
			
				|  |  | +import gc
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 | 
	
		
			
				|  |  | -db_pool = pymysql.connect(host='localhost',
 | 
	
		
			
				|  |  | -                          user='root',
 | 
	
		
			
				|  |  | -                          port=3307,
 | 
	
		
			
				|  |  | -                          password='r6kEwqWU9!v3',
 | 
	
		
			
				|  |  | -                          database='hlfx_pool')
 | 
	
		
			
				|  |  | -cursor_pool = db_pool.cursor()
 | 
	
		
			
				|  |  | +# auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 | 
	
		
			
				|  |  |  engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_whole?charset=utf8')
 | 
	
		
			
				|  |  | +auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +class MyXtQuantTraderCallback(XtQuantTraderCallback):
 | 
	
		
			
				|  |  | +    def on_disconnected(self):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        连接断开
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        print(datetime.datetime.now(), '连接断开回调')
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    def on_stock_order(self, order):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        委托回报推送
 | 
	
		
			
				|  |  | +        :param order: XtOrder对象
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        print(datetime.datetime.now(), '委托回调', order.order_remark)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    def on_stock_trade(self, trade):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        成交变动推送
 | 
	
		
			
				|  |  | +        :param trade: XtTrade对象
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        print(datetime.datetime.now(), '成交回调', trade.order_remark)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    def on_order_error(self, order_error):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        委托失败推送
 | 
	
		
			
				|  |  | +        :param order_error:XtOrderError 对象
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        # print("on order_error callback")
 | 
	
		
			
				|  |  | +        # print(order_error.order_id, order_error.error_id, order_error.error_msg)
 | 
	
		
			
				|  |  | +        print(f"委托报错回调 {order_error.order_remark} {order_error.error_msg}")
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    def on_cancel_error(self, cancel_error):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        撤单失败推送
 | 
	
		
			
				|  |  | +        :param cancel_error: XtCancelError 对象
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        print(datetime.datetime.now(), sys._getframe().f_code.co_name)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    def on_order_stock_async_response(self, response):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        异步下单回报推送
 | 
	
		
			
				|  |  | +        :param response: XtOrderResponse 对象
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        print(f"异步委托回调 {response.order_remark}")
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    def on_cancel_order_stock_async_response(self, response):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        :param response: XtCancelOrderResponse 对象
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        print(datetime.datetime.now(), sys._getframe().f_code.co_name)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    def on_account_status(self, status):
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        :param response: XtAccountStatus 对象
 | 
	
		
			
				|  |  | +        :return:
 | 
	
		
			
				|  |  | +        """
 | 
	
		
			
				|  |  | +        print(datetime.datetime.now(), sys._getframe().f_code.co_name)
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -def run(seq):
 | 
	
		
			
				|  |  | -    print(seq)
 | 
	
		
			
				|  |  | -    '''阻塞线程接收行情回调'''
 | 
	
		
			
				|  |  | +# 指定客户端所在路径
 | 
	
		
			
				|  |  | +path = r'c:\\qmt\\userdata_mini'
 | 
	
		
			
				|  |  | +# 生成session id 整数类型 同时运行的策略不能重复
 | 
	
		
			
				|  |  | +session_id = int(time.time())
 | 
	
		
			
				|  |  | +xt_trader = XtQuantTrader(path, session_id)
 | 
	
		
			
				|  |  | +# 创建资金账号为 800068 的证券账号对象
 | 
	
		
			
				|  |  | +acc = StockAccount('920000207040', 'SECURITY')
 | 
	
		
			
				|  |  | +# 创建交易回调类对象,并声明接收回调
 | 
	
		
			
				|  |  | +callback = MyXtQuantTraderCallback()
 | 
	
		
			
				|  |  | +xt_trader.register_callback(callback)
 | 
	
		
			
				|  |  | +# 启动交易线程
 | 
	
		
			
				|  |  | +xt_trader.start()
 | 
	
		
			
				|  |  | +# 建立交易连接,返回0表示连接成功
 | 
	
		
			
				|  |  | +connect_result = xt_trader.connect()
 | 
	
		
			
				|  |  | +print('建立交易连接,返回0表示连接成功', connect_result)
 | 
	
		
			
				|  |  | +# 对交易回调进行订阅,订阅后可以收到交易主推,返回0表示订阅成功
 | 
	
		
			
				|  |  | +subscribe_result = xt_trader.subscribe(acc)
 | 
	
		
			
				|  |  | +print('对交易回调进行订阅,订阅后可以收到交易主推,返回0表示订阅成功', subscribe_result)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +def run(seq, pid):
 | 
	
		
			
				|  |  | +    mor = datetime.datetime.strptime(
 | 
	
		
			
				|  |  | +        str(dt.now().date()) + '11:30', '%Y-%m-%d%H:%M')
 | 
	
		
			
				|  |  | +    afternoon = datetime.datetime.strptime(
 | 
	
		
			
				|  |  | +        str(dt.now().date()) + '15:00', '%Y-%m-%d%H:%M')
 | 
	
		
			
				|  |  | +    mor_1 = datetime.datetime.strptime(
 | 
	
		
			
				|  |  | +        str(dt.now().date()) + '12:59', '%Y-%m-%d%H:%M')
 | 
	
		
			
				|  |  | +    """阻塞线程接收行情回调"""
 | 
	
		
			
				|  |  |      import time
 | 
	
		
			
				|  |  |      client = xtdata.get_client()
 | 
	
		
			
				|  |  |      while True:
 | 
	
		
			
				|  |  |          time.sleep(3)
 | 
	
		
			
				|  |  |          now_date = dt.now()
 | 
	
		
			
				|  |  | -        if not client.is_connected() or dt.now() > now_date.replace(hour=13, minute=26, second=0):
 | 
	
		
			
				|  |  | +        if not client.is_connected():
 | 
	
		
			
				|  |  |              xtdata.unsubscribe_quote(seq)
 | 
	
		
			
				|  |  |              raise Exception('行情服务连接断开')
 | 
	
		
			
				|  |  | +        # if mor < dt.now() < mor_1:
 | 
	
		
			
				|  |  | +        #     xtdata.unsubscribe_quote(seq)
 | 
	
		
			
				|  |  | +        #     print(f'现在时间:{dt.now()},已休市')
 | 
	
		
			
				|  |  | +        #     sys.exit()
 | 
	
		
			
				|  |  | +        #     break
 | 
	
		
			
				|  |  | +        #     return 0
 | 
	
		
			
				|  |  | +        elif dt.now() > afternoon:
 | 
	
		
			
				|  |  | +            xtdata.unsubscribe_quote(seq)
 | 
	
		
			
				|  |  | +            print(f'现在时间:{dt.now()},已收盘')
 | 
	
		
			
				|  |  | +            sys.exit()
 | 
	
		
			
				|  |  |              break
 | 
	
		
			
				|  |  | -    return
 | 
	
		
			
				|  |  | +            # return 0
 | 
	
		
			
				|  |  | +    # return
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  def real_price(datas):
 | 
	
	
		
			
				|  | @@ -57,13 +156,13 @@ def ma(stock, num, data):
 | 
	
		
			
				|  |  |      except:
 | 
	
		
			
				|  |  |          return 9999999
 | 
	
		
			
				|  |  |      else:
 | 
	
		
			
				|  |  | -        ma_num = (sum(df['close_front'][i:]) + data[stock]['lastPrice'])/num
 | 
	
		
			
				|  |  | +        ma_num = (sum(df['close_front'][i:]) + data[stock]['lastPrice']) / num
 | 
	
		
			
				|  |  |          return ma_num
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  def ma_1(stock, num):
 | 
	
		
			
				|  |  |      global engine_stock
 | 
	
		
			
				|  |  | -    i = (num) * -1
 | 
	
		
			
				|  |  | +    i = num * -1
 | 
	
		
			
				|  |  |      try:
 | 
	
		
			
				|  |  |          df = pd.read_sql_query(
 | 
	
		
			
				|  |  |              'select close_front from `%s_1d`' % stock, engine_stock)
 | 
	
	
		
			
				|  | @@ -86,53 +185,61 @@ def his_vol(stock, num):
 | 
	
		
			
				|  |  |          return df['volume_front'].iloc[num]
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -def ma_judge(data, stock_list, rate, results):
 | 
	
		
			
				|  |  | +def ma_judge(data, list_judge, rate, results):
 | 
	
		
			
				|  |  |      # print(f',收到的data数据为:{len(data.keys())},stock_pool长度为{len(stock_list)},now is {dt.now()}')
 | 
	
		
			
				|  |  | -    list_judge = list(set(data.keys()) & set(stock_list))
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  |      print(f'这个ma_judge的PID为:{os.getpid()},本轮计算:{len(list_judge)}个股')
 | 
	
		
			
				|  |  |      for stock in list_judge:
 | 
	
		
			
				|  |  | -        i = stock.replace('XSHG', 'SH').replace('XSHE', 'SZ')
 | 
	
		
			
				|  |  | -        current_price, open_price = data[i]['lastPrice'], data[i]['open']
 | 
	
		
			
				|  |  | -        MA5, MA10, MA20, MA30, MA60, MA120 = ma(i, 5, data), ma(i, 10, data), ma(i, 20, data), ma(i, 30, data),\
 | 
	
		
			
				|  |  | -            ma(i, 60, data), ma(i, 120, data)
 | 
	
		
			
				|  |  | -        MA5_1 = ma_1(i, 5)
 | 
	
		
			
				|  |  | +        current_price, open_price = data[stock]['lastPrice'], data[stock]['open']
 | 
	
		
			
				|  |  | +        MA5, MA10, MA20, MA30, MA60, MA120 = ma(stock, 5, data), ma(stock, 10, data), ma(stock, 20, data), ma(stock, 30,
 | 
	
		
			
				|  |  | +                                                                                                              data), \
 | 
	
		
			
				|  |  | +            ma(stock, 60, data), ma(stock, 120, data)
 | 
	
		
			
				|  |  | +        MA5_1 = ma_1(stock, 5)
 | 
	
		
			
				|  |  |          # print(i, current_price, open_price, MA5, MA10, MA20, MA5_1)
 | 
	
		
			
				|  |  |          # 入交易池标准:阳线\大于MA5\MA5向上\MA20<MA10\离120线有距离
 | 
	
		
			
				|  |  |          if (current_price > open_price) & (current_price > MA5) & (MA5 > MA5_1) & (current_price < MA5 * 1.03) & (
 | 
	
		
			
				|  |  | -                MA20 < MA10) & (current_price > MA120 or current_price < MA120*rate):
 | 
	
		
			
				|  |  | -            if his_vol(i, -1) > his_vol(i, -2):
 | 
	
		
			
				|  |  | -                results.append(i.replace('SH', 'XSHG').replace('SZ', 'XSHE'))
 | 
	
		
			
				|  |  | +                MA20 < MA10) & (current_price > MA120 or current_price < MA120 * rate):
 | 
	
		
			
				|  |  | +            if his_vol(stock, -1) > his_vol(stock, -2):
 | 
	
		
			
				|  |  | +                results.append(stock.replace('SH', 'XSHG').replace('SZ', 'XSHE'))
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -def sell_trader(data, positions_dict):
 | 
	
		
			
				|  |  | -    # for m in data:
 | 
	
		
			
				|  |  | -    #     print(m, data[m]['lastPrice'])
 | 
	
		
			
				|  |  | +def get_fundamentals(results):
 | 
	
		
			
				|  |  | +    return results
 | 
	
		
			
				|  |  | +    pass
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +def sell_trader(data):
 | 
	
		
			
				|  |  |      print('卖出函数:', dt.now())
 | 
	
		
			
				|  |  |      positions = xt_trader.query_stock_positions(acc)
 | 
	
		
			
				|  |  | -    print('持仓总数:', len(positions))
 | 
	
		
			
				|  |  | +    positions_dict = {positions[x].stock_code: positions[x].can_use_volume for x in range(0, len(positions))}
 | 
	
		
			
				|  |  | +    print(f'今日可卖出个股总数:{len([value for value in positions_dict.values() if value != 0])}')
 | 
	
		
			
				|  |  | +    print(
 | 
	
		
			
				|  |  | +        f'目前持仓总数为:{len([positions[x].stock_code for x in range(0, len(positions)) if positions[x].volume != 0])}')
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -    for stock, volume in positions_dict.items():
 | 
	
		
			
				|  |  | -        if stock in data:
 | 
	
		
			
				|  |  | +    for stock, can_use_volume in positions_dict.items():
 | 
	
		
			
				|  |  | +        if stock in data and can_use_volume != 0:
 | 
	
		
			
				|  |  |              current_price = data[stock]['lastPrice']
 | 
	
		
			
				|  |  |              open_price = data[stock]['open']
 | 
	
		
			
				|  |  |              MA5 = ma(stock, 5, data)
 | 
	
		
			
				|  |  |              MA5_1 = ma_1(stock, 5)
 | 
	
		
			
				|  |  | -            print(f'{stock},持仓量为{volume}当前价:{current_price},MA5:{MA5},昨日MA5:{MA5_1},开始判断:')
 | 
	
		
			
				|  |  | -            if current_price < MA5 or MA5 < MA5_1 or current_price > MA5 * 1.07:
 | 
	
		
			
				|  |  | +            print(f'{stock},持仓量为{can_use_volume}当前价:{current_price},MA5:{MA5},昨日MA5:{MA5_1},开始判断:')
 | 
	
		
			
				|  |  | +            if current_price < MA5 or MA5 < MA5_1:
 | 
	
		
			
				|  |  |                  print('卖出信号!!!!!!', stock, current_price)
 | 
	
		
			
				|  |  | -                order_id = xt_trader.order_stock(acc, stock, xtconstant.STOCK_SELL, volume,
 | 
	
		
			
				|  |  | -                                                 xtconstant.LATEST_PRICE, 0, 'strategy1', 'order_test')
 | 
	
		
			
				|  |  | +                order_id = xt_trader.order_stock(acc, stock, xtconstant.STOCK_SELL, can_use_volume,
 | 
	
		
			
				|  |  | +                                                 xtconstant.LATEST_PRICE, 0, 'MA5策略', '低于MA5趋势向下')
 | 
	
		
			
				|  |  |                  print('价格:', current_price, open_price, MA5, MA5_1)
 | 
	
		
			
				|  |  | -                print(order_id, stock, volume)
 | 
	
		
			
				|  |  | +                print(order_id, stock, can_use_volume)
 | 
	
		
			
				|  |  | +            elif current_price > MA5 * 1.07:
 | 
	
		
			
				|  |  | +                print('盈利乖离率超7%!!!!!!', stock, current_price)
 | 
	
		
			
				|  |  | +                order_id = xt_trader.order_stock(acc, stock, xtconstant.STOCK_SELL, can_use_volume,
 | 
	
		
			
				|  |  | +                                                 xtconstant.LATEST_PRICE, 0, 'MA5策略', '盈利乖离率超7%')
 | 
	
		
			
				|  |  | +                print('价格:', current_price, open_price, MA5, MA5_1)
 | 
	
		
			
				|  |  | +                print(order_id, stock, can_use_volume)
 | 
	
		
			
				|  |  |          else:
 | 
	
		
			
				|  |  |              print(f'本轮没有持仓股票信息!')
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -def get_fundamentals(results):
 | 
	
		
			
				|  |  | -    return results
 | 
	
		
			
				|  |  | -    pass
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -def buy_trader(data, positions):
 | 
	
		
			
				|  |  | +def buy_trader(data):
 | 
	
		
			
				|  |  |      print('买入函数:', dt.now(), f'接受到{len(data.keys())}个个股')
 | 
	
		
			
				|  |  |      results = mp.Manager().list()
 | 
	
		
			
				|  |  |      mp_list = []
 | 
	
	
		
			
				|  | @@ -140,10 +247,10 @@ def buy_trader(data, positions):
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |      try:
 | 
	
		
			
				|  |  |          stock_pool = pd.read_sql_query(
 | 
	
		
			
				|  |  | -            'select value from `%s`' % '1d', engine_hlfx_pool)
 | 
	
		
			
				|  |  | -        stock_pool = stock_pool.iloc[-1, 0].split(",")
 | 
	
		
			
				|  |  | +            'select value from `%s` order by `index` desc limit 10' % '1d', engine_hlfx_pool)
 | 
	
		
			
				|  |  | +        stock_pool = stock_pool.iloc[0, 0].split(",")
 | 
	
		
			
				|  |  |          stock_pool.sort()
 | 
	
		
			
				|  |  | -        print('stock_pool',len(stock_pool))
 | 
	
		
			
				|  |  | +        print('stock_pool', len(stock_pool))
 | 
	
		
			
				|  |  |      except BaseException:
 | 
	
		
			
				|  |  |          pass
 | 
	
		
			
				|  |  |      '''
 | 
	
	
		
			
				|  | @@ -159,12 +266,13 @@ def buy_trader(data, positions):
 | 
	
		
			
				|  |  |                      results.append(stock.replace('SH', 'XSHG').replace('SZ', 'XSHE'))
 | 
	
		
			
				|  |  |                      print('append')
 | 
	
		
			
				|  |  |      '''
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    step = math.ceil(len(stock_pool) / (mp.cpu_count()/2))
 | 
	
		
			
				|  |  | +    list_judge = list(set(data.keys()) & set(stock_pool))
 | 
	
		
			
				|  |  | +    print(f'本轮有{len(data.keys())}条个股信息,而list_judge有:{len(list_judge)}')
 | 
	
		
			
				|  |  | +    step = math.ceil(len(list_judge) / 2)
 | 
	
		
			
				|  |  |      print('step:', step)
 | 
	
		
			
				|  |  |      rate = 0.8
 | 
	
		
			
				|  |  | -    for i in range(0, len(stock_pool), step):
 | 
	
		
			
				|  |  | -        p = mp.Process(target=ma_judge, args=(data, stock_pool[i:i + step], rate, results))
 | 
	
		
			
				|  |  | +    for i in range(0, len(list_judge), step):
 | 
	
		
			
				|  |  | +        p = mp.Process(target=ma_judge, args=(data, list_judge[i:i + step], rate, results))
 | 
	
		
			
				|  |  |          mp_list.append(p)
 | 
	
		
			
				|  |  |          p.start()
 | 
	
		
			
				|  |  |      for j in mp_list:
 | 
	
	
		
			
				|  | @@ -192,6 +300,12 @@ def buy_trader(data, positions):
 | 
	
		
			
				|  |  |          results_industry = ','.join(set(results_industry))
 | 
	
		
			
				|  |  |          print('1d', '\n', results_industry)
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | +        db_pool = pymysql.connect(host='localhost',
 | 
	
		
			
				|  |  | +                                  user='root',
 | 
	
		
			
				|  |  | +                                  port=3307,
 | 
	
		
			
				|  |  | +                                  password='r6kEwqWU9!v3',
 | 
	
		
			
				|  |  | +                                  database='hlfx_pool')
 | 
	
		
			
				|  |  | +        cursor_pool = db_pool.cursor()
 | 
	
		
			
				|  |  |          sql = "INSERT INTO MA5_%s (date,value) VALUES('%s','%s')" % ('1d', dt.now().strftime('%Y-%m-%d %H:%M:%S'),
 | 
	
		
			
				|  |  |                                                                       results_industry)
 | 
	
		
			
				|  |  |          cursor_pool.execute(sql)
 | 
	
	
		
			
				|  | @@ -199,18 +313,21 @@ def buy_trader(data, positions):
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |          # print(len(results_industry), results_industry)
 | 
	
		
			
				|  |  |          print(dt.now(), '数据库数据已赋值!')
 | 
	
		
			
				|  |  | +        cursor_pool.close()
 | 
	
		
			
				|  |  | +        db_pool.close()
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |          # 取值交易
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  |          keep_stocks = results_industry.split(",")
 | 
	
		
			
				|  |  |          new_keep_stock = [stock.replace('XSHG', 'SH').replace('XSHE', 'SZ') for stock in keep_stocks]
 | 
	
		
			
				|  |  |          print(f'new_keep_stock is:{len(new_keep_stock)}')
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -        #进入购买程序
 | 
	
		
			
				|  |  | +        # 进入购买程序
 | 
	
		
			
				|  |  |          max_pos = 7
 | 
	
		
			
				|  |  |          for stock in new_keep_stock:
 | 
	
		
			
				|  |  | +            positions = xt_trader.query_stock_positions(acc)
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  |              asset = xt_trader.query_stock_asset(acc)
 | 
	
		
			
				|  |  | +            print('bbbb', positions, asset)
 | 
	
		
			
				|  |  |              cash = asset.cash
 | 
	
		
			
				|  |  |              positions_dict = {positions[x].stock_code: positions[x].volume for x in range(0, len(positions)) if
 | 
	
		
			
				|  |  |                                positions[x].volume > 0}
 | 
	
	
		
			
				|  | @@ -218,138 +335,59 @@ def buy_trader(data, positions):
 | 
	
		
			
				|  |  |              current_price = data[stock]['lastPrice']
 | 
	
		
			
				|  |  |              current_high = data[stock]['high']
 | 
	
		
			
				|  |  |              if cash > 5000 and len(positions_dict) < max_pos and current_price > 9 \
 | 
	
		
			
				|  |  | -                    and current_price > (current_high*0.98):
 | 
	
		
			
				|  |  | +                    and current_price > (current_high * 0.98):
 | 
	
		
			
				|  |  |                  volume = int((cash / 3 / current_price) // 100 * 100)
 | 
	
		
			
				|  |  |                  print('买入信号!!!!!!', stock, volume, current_price)
 | 
	
		
			
				|  |  |                  order_id = xt_trader.order_stock(acc, stock, xtconstant.STOCK_BUY, volume, xtconstant.LATEST_PRICE,
 | 
	
		
			
				|  |  | -                                                 current_price, 'strategy1', 'order_test')
 | 
	
		
			
				|  |  | +                                                 current_price, 'MA5策略', 'MA5趋势向上')
 | 
	
		
			
				|  |  |                  print(order_id)
 | 
	
		
			
				|  |  |              else:
 | 
	
		
			
				|  |  | -                print(f'Cash只有:{cash} 或者 现有持仓{len(positions)} 超过了{max_pos}')
 | 
	
		
			
				|  |  | +                print(f'Cash只有:{cash} 或者 现有持仓{len(positions_dict)} 超过了{max_pos}')
 | 
	
		
			
				|  |  |      engine_hlfx_pool.dispose()
 | 
	
		
			
				|  |  |      print('一轮结束了,现在时间是:', dt.now())
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  def trader(data):
 | 
	
		
			
				|  |  | -    print(len(data.keys()))
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    # 先判断卖出条件
 | 
	
		
			
				|  |  | -    positions = xt_trader.query_stock_positions(acc)
 | 
	
		
			
				|  |  | -    print('持仓数量', len(positions))
 | 
	
		
			
				|  |  | -    if len(positions) != 0:
 | 
	
		
			
				|  |  | -        positions_dict = {positions[x].stock_code: positions[x].volume for x in range(0, len(positions))}
 | 
	
		
			
				|  |  | -        sell_trader(data, positions_dict)
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | +    sell_trader(data)
 | 
	
		
			
				|  |  |      # 买入条件
 | 
	
		
			
				|  |  | -    buy_trader(data, positions)
 | 
	
		
			
				|  |  | +    buy_trader(data)
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  def bridge():
 | 
	
		
			
				|  |  | -    print("start")
 | 
	
		
			
				|  |  | +    pid = os.getpid()
 | 
	
		
			
				|  |  | +    print(f'MyPid is {os.getpid()}, now is {dt.now()},开盘了')
 | 
	
		
			
				|  |  |      stocks = xtdata.get_stock_list_in_sector('沪深A股')
 | 
	
		
			
				|  |  |      seq = xtdata.subscribe_whole_quote(stocks, callback=trader)
 | 
	
		
			
				|  |  | -    run(seq)
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -class MyXtQuantTraderCallback(XtQuantTraderCallback):
 | 
	
		
			
				|  |  | -    def on_disconnected(self):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        连接断开
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        print(datetime.datetime.now(), '连接断开回调')
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    def on_stock_order(self, order):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        委托回报推送
 | 
	
		
			
				|  |  | -        :param order: XtOrder对象
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        print(datetime.datetime.now(), '委托回调', order.order_remark)
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    def on_stock_trade(self, trade):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        成交变动推送
 | 
	
		
			
				|  |  | -        :param trade: XtTrade对象
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        print(datetime.datetime.now(), '成交回调', trade.order_remark)
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    def on_order_error(self, order_error):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        委托失败推送
 | 
	
		
			
				|  |  | -        :param order_error:XtOrderError 对象
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        # print("on order_error callback")
 | 
	
		
			
				|  |  | -        # print(order_error.order_id, order_error.error_id, order_error.error_msg)
 | 
	
		
			
				|  |  | -        print(f"委托报错回调 {order_error.order_remark} {order_error.error_msg}")
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    def on_cancel_error(self, cancel_error):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        撤单失败推送
 | 
	
		
			
				|  |  | -        :param cancel_error: XtCancelError 对象
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        print(datetime.datetime.now(), sys._getframe().f_code.co_name)
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    def on_order_stock_async_response(self, response):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        异步下单回报推送
 | 
	
		
			
				|  |  | -        :param response: XtOrderResponse 对象
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        print(f"异步委托回调 {response.order_remark}")
 | 
	
		
			
				|  |  | +    run(seq, pid)
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -    def on_cancel_order_stock_async_response(self, response):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        :param response: XtCancelOrderResponse 对象
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        print(datetime.datetime.now(), sys._getframe().f_code.co_name)
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -    def on_account_status(self, status):
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        :param response: XtAccountStatus 对象
 | 
	
		
			
				|  |  | -        :return:
 | 
	
		
			
				|  |  | -        """
 | 
	
		
			
				|  |  | -        print(datetime.datetime.now(), sys._getframe().f_code.co_name)
 | 
	
		
			
				|  |  | +def job_func():
 | 
	
		
			
				|  |  | +    print(f"Job started at {dt.now()}")
 | 
	
		
			
				|  |  | +    # 创建子进程
 | 
	
		
			
				|  |  | +    p = mp.Process(target=bridge)
 | 
	
		
			
				|  |  | +    # 启动子进程
 | 
	
		
			
				|  |  | +    p.start()
 | 
	
		
			
				|  |  | +    # 等待子进程结束
 | 
	
		
			
				|  |  | +    p.join()
 | 
	
		
			
				|  |  | +    print(f"Job finished at {dt.now()}")
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |  if __name__ == '__main__':
 | 
	
		
			
				|  |  | -    auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  |      mp.freeze_support()
 | 
	
		
			
				|  |  |      print('cpu_count =', mp.cpu_count())
 | 
	
		
			
				|  |  |      pus = psutil.Process()
 | 
	
		
			
				|  |  | -    pus.cpu_affinity([4, 5, 6, 7])
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  | -    # 指定客户端所在路径
 | 
	
		
			
				|  |  | -    path = r'c:\\qmt\\userdata_mini'
 | 
	
		
			
				|  |  | -    # 生成session id 整数类型 同时运行的策略不能重复
 | 
	
		
			
				|  |  | -    session_id = int(time.time())
 | 
	
		
			
				|  |  | -    xt_trader = XtQuantTrader(path, session_id)
 | 
	
		
			
				|  |  | -    # 创建资金账号为 800068 的证券账号对象
 | 
	
		
			
				|  |  | -    acc = StockAccount('920000207040', 'SECURITY')
 | 
	
		
			
				|  |  | -    # 创建交易回调类对象,并声明接收回调
 | 
	
		
			
				|  |  | -    callback = MyXtQuantTraderCallback()
 | 
	
		
			
				|  |  | -    xt_trader.register_callback(callback)
 | 
	
		
			
				|  |  | -    # 启动交易线程
 | 
	
		
			
				|  |  | -    xt_trader.start()
 | 
	
		
			
				|  |  | -    # 建立交易连接,返回0表示连接成功
 | 
	
		
			
				|  |  | -    connect_result = xt_trader.connect()
 | 
	
		
			
				|  |  | -    print('建立交易连接,返回0表示连接成功', connect_result)
 | 
	
		
			
				|  |  | -    # 对交易回调进行订阅,订阅后可以收到交易主推,返回0表示订阅成功
 | 
	
		
			
				|  |  | -    subscribe_result = xt_trader.subscribe(acc)
 | 
	
		
			
				|  |  | -    print('对交易回调进行订阅,订阅后可以收到交易主推,返回0表示订阅成功', subscribe_result)
 | 
	
		
			
				|  |  | +    pus.cpu_affinity([12, 13, 14, 15, 16, 17])
 | 
	
		
			
				|  |  | +
 | 
	
		
			
				|  |  | +    # job_func()
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  |      scheduler = BlockingScheduler()
 | 
	
		
			
				|  |  | -    scheduler.add_job(func=bridge, trigger='cron', day_of_week='0-4', hour='9', minute='40',
 | 
	
		
			
				|  |  | +    scheduler.add_job(func=job_func, trigger='cron', day_of_week='0-4', hour='09', minute='40',
 | 
	
		
			
				|  |  |                        timezone="Asia/Shanghai")
 | 
	
		
			
				|  |  | +    # scheduler.add_job(func=job_func, trigger='cron', day_of_week='0-4', hour='13', minute='05',
 | 
	
		
			
				|  |  | +    #                   timezone="Asia/Shanghai")
 | 
	
		
			
				|  |  |      try:
 | 
	
		
			
				|  |  |          scheduler.start()
 | 
	
		
			
				|  |  |      except (KeyboardInterrupt, SystemExit):
 | 
	
		
			
				|  |  |          pass
 | 
	
		
			
				|  |  |  
 | 
	
		
			
				|  |  | -
 | 
	
		
			
				|  |  |      # xtdata.subscribe_quote('000001.SZ', '1d', '', '', count=1, callback=MA)
 |