stock/QMT/real_time.py

# coding:utf-8
from datetime import datetime as dt
import os
import pandas as pd
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xttype import StockAccount
from xtquant import xtdata
import time
from sqlalchemy import create_engine
from jqdatasdk import *
import pymysql
import multiprocessing as mp
import math

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()
engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_front?charset=utf8')


def real_price(datas):
    # i = '000001.SZ'
    for i in datas:
        if i == '000001.SZ':
            print(i, datas[i])
    # trader(datas)
    # return datas


def ma(stock, num, data):
    global engine_stock
    try:
        i = (num - 1) * -1
        df = pd.read_sql_query(
            'select close from `%s_1d`' % stock, engine_stock)
    except:
        return 9999999
    else:
        ma_num = sum(df['close'][i:-1] + data[stock]['lastPrice'])/num
        return ma_num


def ma_1(stock, num):
    global engine_stock
    i = num  * -1
    try:
        df = pd.read_sql_query(
            'select close from `%s_1d`' % stock, engine_stock)
    except BaseException:
        return 9999999
    else:
        ma_num_1 = df['close'][i:-1].mean()
        return ma_num_1


def his_vol(stock, num):
    global engine_stock
    num = num * -1
    try:
        df = pd.read_sql_query(
            'select volume from `%s_1d`' % stock, engine_stock)
    except BaseException:
        return 9999999
    else:
        return df['volume'].iloc[num]


def ma_judge(data, stock_list, results):
    print('这个ma_judge的PID为：', os.getpid())
    for stock in data:
        i = stock.replace('XSHG', 'SH').replace('XSHE', 'SZ')
        current_price, open_price = data[i]['lastPrice'], data[i]['open']
        MA5, MA10, MA20 = ma(i, 5, data), ma(i, 10, data), ma(i, 20, data)
        MA5_1 = ma_1(i, 5)
        print(i, current_price, open_price, MA5, MA10, MA20, MA5_1)
        if (current_price > open_price) & (current_price > MA5) & (MA5 > MA5_1) & (current_price < MA5 * 1.03) & (
                MA20 < MA10):
            if his_vol(i, -1) > his_vol(i, -2):
                results.append(i.replace('SH', 'XSHG').replace('SZ', 'XSHE'))
    print('RRRRRRR,', results)


def sell_trader(data, positions):
    # for m in data:
    #     print(m, data[m]['lastPrice'])
    print('卖出函数:', dt.now())
    # positions = xt_trader.query_stock_positions(acc)
    # print('持仓总数：', len(positions))
    for stock in data:
        if stock in positions:
            print('持仓', stock, data[stock])
            current_price = data[stock]['lastPrice']
            open_price = data[stock]['open']
            print('价格：', current_price, open_price)
            MA5 = ma(stock, 5, data)
            MA5_1 = ma_1(stock, 5)
            if current_price < MA5 or MA5 < MA5_1 or current_price > MA5 * 1.07:
                print('卖出信号！！！！！！', stock, current_price)

            #     order_id = xt_trader.order_stock(acc, stock, xtconstant.STOCK_SELL,
            #                                     i.volume, xtconstant.LATEST_PRICE, 0, 'strategy1', 'order_test')
            # print(order_id, i)


def buy_trader(data):
    print('买入函数:', dt.now())
    results = mp.Manager().list()
    mp_list = []
    engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')

    try:
        stock_pool = pd.read_sql_query(
            'select value from `%s`' % '1d', engine_hlfx_pool)
        stock_pool = stock_pool.iloc[-1, 0].split(",")
        stock_pool.sort()
        print('stock_pool', stock_pool)
    except BaseException:
        pass
    '''
    for stock in data:
        if stock.replace('SH', 'XSHG').replace('SZ', 'XSHE') in stock_pool:
            # 真实买入策略
            current_price, open_price = data[stock]['lastPrice'], data[stock]['open']
            MA5, MA10, MA20 = ma(stock, 5), ma(stock, 10), ma(stock, 20)
            MA5_1 = ma_1(stock, 5)
            print(stock, current_price, open_price, MA5, MA10, MA20, MA5_1)
            if (current_price > open_price) & (current_price > MA5) & (MA5 > MA5_1) & (current_price < MA5 * 1.03) & (MA20 < MA10):
                if his_vol(stock, -1) > his_vol(stock, -2):
                    results.append(stock.replace('SH', 'XSHG').replace('SZ', 'XSHE'))
                    print('append')
    '''

    step = math.ceil(len(stock_pool) / mp.cpu_count())
    print('step:', step)
    print('cpu_count =', mp.cpu_count())
    for i in range(0, len(stock_pool), math.ceil(len(stock_pool) / mp.cpu_count())):
        p = mp.Process(target=ma_judge, args=(data, stock_pool[i:i + step], results))
        mp_list.append(p)
        p.start()
    for j in mp_list:
        j.join()
    results = list(set(results))
    print('results!!!!', len(results))

    # 选择板块
    if len(results) != 0:
        num_industry = get_industry(results)
        print(num_industry)
        industry_list = []
        for key in num_industry.values():
            for key2 in key.values():
                industry_list.append(key2['industry_name'])
        industry_list = pd.value_counts(industry_list)
        # 最热集中的n个板块
        max_industry_list = list(industry_list[0:3].index)
        results_industry = []
        for key, value in num_industry.items():
            for key2 in value.values():
                if key2['industry_name'] in max_industry_list:
                    results_industry.append(key)
        print('所有:', set(results_industry))
        results_industry = ','.join(set(results_industry))
        print('1d', '\n', results_industry)

        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)
        db_pool.commit()

        # print(len(results_industry), results_industry)
        print(dt.now(), '数据库数据已赋值！')

        # 取值交易

        keep_stocks = results_industry.split(",")
        new_keep_stock = [stock.replace('XSHG', 'SH').replace('XSHE', 'SZ') for stock in keep_stocks]
        print(new_keep_stock)

        for stock in data:
            if stock in new_keep_stock:
                current_price = data[stock]['lastPrice']
                if acc.cash > 2000:
                    volume = int((acc.cash / 2 / current_price) // 100 * 100)
                    print('volume：', volume)
                    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')
                    # print(order_id)
    print('一轮结束了，现在时间是：', dt.now())


def trader(data):
    print(len(data.keys()))

    # 先判断卖出条件
    positions = xt_trader.query_stock_positions(acc)
    print('持仓数量', len(positions))
    if len(positions) != 0:
        sell_trader(data, positions)

    # 买入条件
    buy_trader(data)


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)


if __name__ == '__main__':
    auth('18616891214', 'Ea?*7f68nD.dafcW34d!')

    print("start")
    # 指定客户端所在路径
    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=trader)
    xtdata.run()
    # xtdata.subscribe_quote('000001.SZ', '1d', '', '', count=1, callback=MA)