# coding:utf-8 import pandas as pd import pymysql from sqlalchemy import create_engine, text import threading from datetime import datetime as dt import datetime from jqdatasdk.technical_analysis import * from xtquant import xtdata, xtconstant from xtquant.xttype import StockAccount from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback import time import math import multiprocessing as mp import os import psutil import traceback from apscheduler.schedulers.blocking import BlockingScheduler import sys pd.set_option('display.max_columns', None) # 设置显示最大行 # 全局变量 stocks_dict stocks_dict = {} stocks = xtdata.get_stock_list_in_sector('沪深A股') for stock in stocks: stocks_dict[stock] = [] print(stocks_dict) 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 err_call_back(err): print(f'问题在这里~ error：{str(err)}') traceback.print_exc() def run(seq): 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()) + '11:10', '%Y-%m-%d%H:%M') """阻塞线程接收行情回调""" import time client = xtdata.get_client() while True: now_date = dt.now() 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 def speed(stock_list, datas): global stocks_dict print('..............................',stocks_dict) for i in range(len(stock_list)): stock = stock_list[i] try: stocks_dict[stock].append(datas[stock]['lastPrice']) except BaseException as e: print(e) print(stock, stocks_dict[stock], datas[stock]['lastPrice']) # print(datas[i]['lastPrice'], stocks_dict[i]) ''' for stock in stock_list: print(stock, datas[stock]['lastPrice']) print(stocks_dict[stock]) try: stocks_dict[stock].append(datas[stock]['lastPrice']) except BaseException as e: print(e) print(stock, stocks_dict[stock], datas[stock]['lastPrice']) ''' def prepare(datas): for k, v in datas.items(): # print(k) stocks_dict[k].append(v['lastPrice']) print(k, stocks_dict[k], v['lastPrice']) # stocks_dict[stock] = datas[stock]['lastPrice'] # 将数据添加至全局变量 stocks_dict[stock] key为股票代码，值为最新价 # print(stocks_dict[stock]) ''' stock_list = list(datas.keys()) if len(datas.keys()) >= 12: cpu_count = 12 else: cpu_count = len(datas.keys()) step = math.ceil(len(stock_list) / cpu_count) to_list = [] for i in range(0, len(stock_list), step): to_list.append([x for x in stock_list[i:i + step]]) pool = mp.Pool(processes=int(cpu_count/2)) for m in range(len(to_list)): pool.apply_async(func=speed, args=(to_list[m], datas), error_callback=err_call_back) pool.close() pool.join() ''' def bridge(): # 全局变量 stocks_dict global stocks print(f'bridge is {os.getpid()}, now is {dt.now()},开盘了') # stocks_dict = dict.fromkeys(stocks, []) # print(stocks_dict) seq = xtdata.subscribe_whole_quote(stocks, callback=prepare) # 建立一个以stocks中值为键，值为空列表的字典 # print(f'stocks_dict is {stocks_dict}') run(seq) if __name__ == '__main__': print(f'总进程pid：{os.getpid()}') mp.freeze_support() pus = psutil.Process() # pus.cpu_affinity([0, 1, 2, 3, 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) bridge() scheduler = BlockingScheduler() scheduler.add_job(func=bridge, trigger='cron', day_of_week='0-4', hour='09', minute='25', timezone="Asia/Shanghai", max_instances=5) # # scheduler.add_job(func=job_func, trigger='cron', day_of_week='0-4', hour='13', minute='00', # # timezone="Asia/Shanghai") try: scheduler.start() except (KeyboardInterrupt, SystemExit): pass