stock/QMT/qmt_real_hlfx.py

# coding:utf-8
# from jqdatasdk import *
import pandas as pd
import pymysql
from sqlalchemy import create_engine
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
#原始版本

# auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
# auth('18521506014', 'Abc123!@#')
# stocks = list(get_all_securities(['stock'], date=dt.today().strftime('%Y-%m-%d')).index)
# stocks = stocks[0:200]

pd.set_option('display.max_columns', None) # 设置显示最大行
fre = '1d'

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 hlfx(data, stocks, pool_list):
    print(f'MyPid is {os.getpid()}, now is {dt.now()}')
    engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_tech?charset=utf8')
    print(stocks)
    for qmt_stock in stocks:
        # 读取qmt_stocks_whole表-前复权-信息
        try:
            df_day = pd.read_sql_query(
                'select time, open_front, close_front, high_front, low_front, volume_front, amount_front, dif, dea, macd,HL from `%s_%s`'
                % (qmt_stock, fre), engine_stock)
            df_day.columns=['time', 'open', 'close', 'high', 'low', 'volume', 'amount', 'dif', 'dea', 'macd', 'HL']
        except BaseException as e:
            print(qmt_stock, '未能读取！', e)
            pass
        else:
            # 获得最新价格信息
            get_price = data[qmt_stock]
            # 调整time时间格式
            get_price['time'] = dt.fromtimestamp(get_price['time'] / 1000.0)
            # print('成功判定', get_price['time'])

            # 先处理去包含
            # 不包含
            if (df_day.iloc[-1, 3] > get_price['high']
                and df_day.iloc[-1, 4] > get_price['low']) \
                    or (df_day.iloc[-1, 3] < get_price['high']
                        and df_day.iloc[-1, 4] < get_price['low']):
                # print('lalallala', get_price['open'], get_price['lastPrice'], get_price['high'],
                #                             get_price['low'], get_price['volume'], get_price['amount'])
                qmt_df = pd.DataFrame(data=[[get_price['time'], get_price['open'], get_price['lastPrice'], get_price['high'],
                                            get_price['low'], get_price['volume'], get_price['amount']]],
                                      columns=['time', 'open', 'close', 'high', 'low', 'volume', 'amount'])
                # print('qmt_______', qmt_df)
                df_day = pd.concat([df_day, qmt_df], ignore_index=True)
                # print('不包含，合并完成', df_day)

            # 包含
            else:
                # 左高，下降
                if df_day.iloc[-2, 3] > df_day.iloc[-1, 3]:
                    df_day.iloc[-1, 3] = min(df_day.iloc[-1, 3], get_price['high'])
                    df_day.iloc[-1, 4] = min(df_day.iloc[-1, 4], get_price['low'])
                # 右高，上升
                else:
                    df_day.iloc[-1, 3] = max(df_day.iloc[-1, 3], get_price['high'])
                    df_day.iloc[-1, 4] = max(df_day.iloc[-1, 4], get_price['low'])
                # print('包含', df_day)

            # 数合并完成，确认df_day
            # print(df_day)

            # 寻找顶底分型
            x = len(df_day.index)-1
            m = x - 1
            # 底
            if ((df_day.loc[x, 'high'] > df_day.loc[x - 1, 'high']) and (
                    df_day.loc[x - 2, 'high'] > df_day.loc[x - 1, 'high'])):
                df_day.loc[x, 'HL'] = 'L*'
                # 判断底的性质
                while m:
                    if df_day.loc[m, 'HL'] in ['H', 'HH', 'H*']:
                        if (x - m) > 3:
                            # 成笔——>L
                            df_day.loc[x, 'HL'] = 'L'

                    elif df_day.loc[m, 'HL'] == 'L':
                        if df_day.loc[m - 1, 'low'] > df_day.loc[x - 1, 'low']:
                            # pool_list.append(qmt_stock)

                            # 获得MACD，判断MACD判断背驰
                            x_macd_dif, x_macd_dea, x_macd_macd = df_day.loc[x, 'dif'], df_day.loc[x, 'dea'], \
                            df_day.loc[x, 'macd']
                            m_macd_dif, m_macd_dea, m_macd_macd = df_day.loc[m, 'dif'], df_day.loc[m, 'dea'], \
                            df_day.loc[m, 'macd']

                            # 背驰底->LL
                            if m_macd_dif < x_macd_dif:
                                df_day.loc[x, 'HL'] = 'LL'
                                # 产生信号，进入hlfx_pool
                                pool_list.append(qmt_stock)
                            # 前一个为底更高，且中间不存在更低的底
                            else:
                                df_day.loc[x, 'HL'] = 'L'
                                # 产生信号，进入hlfx_pool
                                pool_list.append(qmt_stock)
                            break
                        break
                    m = m - 1
                    if m == 0:
                        df_day.loc[x, 'HL'] = 'L'

            # 顶

            elif (df_day.loc[x, 'high'] < df_day.loc[x - 1, 'high']) and (
                    df_day.loc[x - 2, 'high'] < df_day.loc[x - 1, 'high']) and (qmt_stock in pool_list):
                df_day.loc[x, 'HL'] = 'H*'
                while m:
                    if df_day.loc[m, 'HL'] in ['L', 'LL', 'L*']:
                        if x - m > 3:
                            # 成笔->H
                            df_day.loc[x, 'HL'] = 'H'
                            # 产生卖出信号，进入hlfx_pool
                            pool_list.remove(qmt_stock)
                            break

                    elif (df_day.loc[m, 'HL'] == 'H'):
                        if df_day.loc[x - 1, 'high'] > df_day.loc[m - 1, 'high']:
                            # 获得MACD，判断MACD判断背驰
                            x_macd_dif, x_macd_dea, x_macd_macd = df_day.loc[x, 'dif'], df_day.loc[x, 'dea'], \
                                df_day.loc[x, 'macd']
                            m_macd_dif, m_macd_dea, m_macd_macd = df_day.loc[m, 'dif'], df_day.loc[m, 'dea'], \
                                df_day.loc[m, 'macd']

                            # MACD顶背驰
                            if x_macd_dif < m_macd_dif:
                                df_day.loc[x, 'HL'] = 'HH'
                                # 产生卖出信号，进入hlfx_pool
                                pool_list.remove(qmt_stock)

                            # 前一个为顶，且中间存在不包含 or 更高的顶
                            else:
                                df_day.loc[x, 'HL'] = 'H'
                                # 产生卖出信号，进入hlfx_pool
                                pool_list.remove(qmt_stock)

                            break
                        break
                    m = m - 1
                    if m == 0:
                        df_day.loc[x, 'HL'] = 'H'



def bridge(data):
    # 连接数据库
    '''
    db = pymysql.connect(host='localhost',
                         user='root',
                         port=3307,
                         password='r6kEwqWU9!v3',
                         database='hlfx')
    cursor = db.cursor()
    cursor.execute("show tables like '%%%s%%' " % fre)
    pool_list = [tuple[0] for tuple in cursor.fetchall()]
    print('取得 table_list %s' % fre)
    '''
    '''
    
    1.获取hlfx_pool中隔夜的标的
    2.将本此的data均分，给到进程池
    3.将data总数据、分配的任务stocklist、hlfx_pool 送入realtime_hlfx中进行计算
    4.将实时刷新的hlfx存入hlfx_pool 以过滤出现顶分型的标的
    
    '''

    # 获得hlfx_pool池子
    engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')
    results = mp.Manager().list()
    results.extend(pd.read_sql_query(
        'select value from `%s`' % fre, engine_hlfx_pool).iloc[-1, 0].split(","))
    print(results)

    to_hlfx_list = []
    keys = list(data.keys())
    print(len(keys))
    cpu_count = 4
    step = math.ceil(len(keys) / cpu_count)
    for i in range(0, len(keys), step):
        to_hlfx_list.append([x for x in keys[i:i+step]])

    pool = mp.Pool(processes=cpu_count)
    for m in range(cpu_count):
        pool.apply_async(func=hlfx,
                         args=(data, to_hlfx_list[m], results,), error_callback=err_call_back)
    pool.close()
    pool.join()



    db_pool = pymysql.connect(host='localhost',
                              user='root',
                              port=3307,
                              password='r6kEwqWU9!v3',
                              database='hlfx_pool')
    cursor_pool = db_pool.cursor()

    print(set(results))
    results_list = ','.join(set(results))
    sql = "INSERT INTO %s (date,value) VALUES('%s','%s')" % (fre, dt.now().strftime('%Y-%m-%d %H:%M:%S'), results_list)
    cursor_pool.execute(sql)
    db_pool.commit()
    print(f'{dt.now()}写入新的results，hlfx_pool更新')
    # hlfx(data, engine_stock, engine_hlfx)
    pass


def prepare():
    engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')

    results = pd.read_sql_query(
        'select value from `%s`' % fre, engine_hlfx_pool).iloc[-1, 0].split(",")
    results = [x.replace('XSHG', 'SH').replace('XSHE', 'SZ') for x in results]
    print('数据库读取,并转化后缀格式', len(results))
    # print(results[0:10])
    return results


if __name__ == '__main__':
    mp.freeze_support()
    pus = psutil.Process()
    pus.cpu_affinity([0, 1, 2, 3])

    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)

    stocks = xtdata.get_stock_list_in_sector('沪深A股')
    xtdata.subscribe_whole_quote(stocks, callback=bridge)

    xtdata.run()