stock/QMT/230504_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, xtconstant
import time
from sqlalchemy import create_engine, text
from jqdatasdk import *
import pymysql
import multiprocessing as mp
import math
import psutil
import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
import sys

# 指定客户端所在路径
# path = r'c:\\qmt\\userdata_mini'
path = 'C:\\方正证券FQT交易客户端\\userdata_mini'
# 创建资金账号为 800068 的证券账号对象
# acc = StockAccount('920000207040', 'SECURITY')
acc = StockAccount('2162003533')
# 生成session id 整数类型 同时运行的策略不能重复
session_id = 123456
xt_trader = None
engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_whole?charset=utf8',
                             pool_size=5000, pool_recycle=50, max_overflow=-1)
# 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, 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():
            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 0
    return


def get_fundamentals(results):
    return results
    pass


def ma(stock, num, data):
    global engine_stock
    try:
        i = (num - 1) * -1
        df = pd.read_sql_query(text(
            'select close_front from `%s_1d`' % stock), engine_stock.connect())
    except BaseException as e:
        print(e)
        return 9999999
    else:
        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
    try:
        df = pd.read_sql_query(text(
            'select close_front from `%s_1d`' % stock), engine_stock.connect())
    except BaseException as e:
        print(e)
        return 9999999
    else:
        ma_num_1 = df['close_front'][i:].mean()
        return ma_num_1


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


def ma_judge(data, list_judge, rate, results):
    # print(f'这个ma_judge的PID为:{os.getpid()}，本轮计算:{len(list_judge)}个股')
    for stock in list_judge:
        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*1.0098) & (current_price < MA5 * 1.05) \
        #         & (current_price > MA120 or current_price < MA120 * rate) & (MA5 < MA10*0.99):
        if (current_price > open_price) & (current_price > MA5) & (
                current_price < MA5 * 1.05) \
                & (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):
    # print('卖出函数:', dt.now())
    positions = xt_trader.query_stock_positions(acc)
    positions_dict = {positions[x].stock_code: positions[x].can_use_volume for x in range(0, len(positions))}
    print(
        f'目前持仓总数为：{len([positions[x].stock_code for x in range(0, len(positions)) if positions[x].volume != 0])}')

    for stock, can_use_volume in positions_dict.items():
        # if stock in data and can_use_volume != 0:
        if stock in data:
            current_price = data[stock]['lastPrice']
            open_price = data[stock]['open']
            MA5 = ma(stock, 5, data)
            MA5_1 = ma_1(stock, 5)
            print(
                f"{data[stock]['time']}, {stock}\n当前时间为：{dt.now().strftime('%Y-%m-%d %H:%M:%S')},"
                f"信号时间：{dt.fromtimestamp((data[stock]['time']) / 1000.0)}\n"
                f"持仓量为{can_use_volume}当前价:{current_price},开盘价：{open_price},"f"MA5:{MA5},昨日MA5：{MA5_1},开始判断:")
            df = pd.read_sql_query(text(
                'select close_front, high_front from `%s_1d`' % stock), engine_stock.connect())

            if current_price == xtdata.get_instrument_detail(stock).get('UpStopPrice') \
                    or (df['close_front'].iloc[-1] == df['high_front'].iloc[-1]
                        and df['close_front'].iloc[-1] / df['close_front'].iloc[-2] > 1.08):
                print(f"{stock}涨停或昨日涨幅超过8%，持股观察！{data[stock]['time']}")
                continue
            elif current_price < MA5 or MA5 < MA5_1:
                print('卖出信号！！！！！！', stock, current_price)
                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, '低于MA5趋势向下')
                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, '盈利乖离率超7%')
                print(order_id, stock, can_use_volume)
        else:
            # print(f'本轮没有持仓股票信息！')
            pass
        engine_stock.dispose()


def over_avg(stock):
    pass



def buy_trader(data):
    print('买入函数:', dt.now(), f'接受到{len(data.keys())}个个股')
    results = mp.Manager().list()
    mp_list = []
    engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8',
                                     pool_pre_ping=True, pool_size=20, pool_timeout=60)

    try:
        stock_pool = pd.read_sql_query(
            text('select value from `%s` order by `index` desc limit 10' % '1d'), engine_hlfx_pool.connect())
        stock_pool = stock_pool.iloc[0, 0].split(",")
        print(stock_pool)
        print('---------')
        stock_pool.sort()
        # print('stock_pool', len(stock_pool))

    except BaseException as e:
        print(e)

    if len(stock_pool) != 0:
        print(set(data.keys()))
        list_judge = list(set(data.keys()) & set(stock_pool))
        print(f'本轮有{len(data.keys())}条个股信息，而list_judge有:{len(list_judge)}')
    else:
        print(f'stock_pool为{len(stock_pool)}个')
    step = math.ceil(len(list_judge) / 4)
    rate = 0.9
    if len(list_judge) != 0:
        print(f'list_judge:{list_judge}')
        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:
            j.join()
    results = list(set(results))
    engine_hlfx_pool.dispose()

    # 选择板块
    if len(results) != 0:
        print(f'进入板块选择{results}')
        # 基本面过滤
        results = get_fundamentals(results)
        num_industry = get_industry(results)
        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:2].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)
        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)
        db_pool.commit()

        # 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'{dt.now()},new_keep_stock is:{len(new_keep_stock)}')

        if len(new_keep_stock) != 0:
            # 进入购买程序
            max_pos = 7
            for stock in new_keep_stock:
                positions = xt_trader.query_stock_positions(acc)

                asset = xt_trader.query_stock_asset(acc)
                cash = asset.cash
                positions_dict = {positions[x].stock_code: positions[x].volume for x in range(0, len(positions)) if
                                  positions[x].volume > 0}
                print(f'判断{stock}:cash={cash},持仓数量为{len(positions_dict)}')
                current_price = data[stock]['lastPrice']
                current_high = data[stock]['high']
                if stock not in positions_dict:
                    overavg = over_avg(stock)

                    if len(positions_dict) < max_pos and current_price > 9 \
                            and current_price > (current_high * 0.98):
                        if 5000 > cash:
                            i = 1
                        else:
                            i = 2
                        volume = int((cash / i / 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, 'MA5策略', 'MA5趋势向上')
                        print(order_id)
                    else:
                        print(f'Cash只有：{cash} 或者 现有持仓{len(positions_dict)} 超过了{max_pos}')
                else:
                    print(f'{stock}已持仓！')
    engine_hlfx_pool.dispose()
    # print('一轮结束了，现在时间是：', dt.now())


def trader(data):
    print(f'本轮订阅{len(data)}', dt.now())
    # print(f'xt_trader = {xt_trader},{session_id}')
    # print(len(xt_trader.query_stock_positions(acc)))
    # 卖出判断
    # sell_trader(data)
    # 买入条件
    buy_trader(data)


def bridge():
    global session_id, xt_trader
    pid = os.getpid()
    connect_result = -1
    subscribe_result = -1

    while True:
        if connect_result != 0 or subscribe_result != 0:
            session_id = int(time.time())
            xt_trader = XtQuantTrader(path, session_id)
            # 创建交易回调类对象，并声明接收回调
            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)
            # 建立交易连接，返回0表示连接成功
            connect_result = xt_trader.connect()
            print('建立交易连接，返回0表示连接成功', connect_result)
            # 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
            subscribe_result = xt_trader.subscribe(acc)
            print('对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功', subscribe_result)
            time.sleep(3)
        else:
            break
    print(f'MyPid is {os.getpid()}, now is {dt.now()},开盘了,session_id = {session_id}, \n')
    positions = xt_trader.query_stock_positions(acc)
    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])}')
    stocks = xtdata.get_stock_list_in_sector('沪深A股')
    # seq_s = xtdata.subscribe_whole_quote(stocks, callback=sell_trader)
    seq_b = xtdata.subscribe_whole_quote(stocks, callback=buy_trader)
    # run(seq_s, pid)
    run(seq_b, pid)


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__':
    mp.freeze_support()
    # print('cpu_count =', mp.cpu_count())
    pus = psutil.Process()
    # pus.cpu_affinity([10, 11, 12, 13, 14, 15])

    job_func()

    scheduler = BlockingScheduler()
    scheduler.add_job(func=job_func, trigger='cron', day_of_week='0-4', hour='09', minute='40',
                      timezone="Asia/Shanghai", max_instances=5)
    # scheduler.add_job(func=job_func, trigger='cron', day_of_week='0-4', hour='12', minute='35',
    #                   timezone="Asia/Shanghai")
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass