提交

QMT/5m_data_whole.py

@@ -0,0 +1,410 @@
+from xtquant import xtdata
+from datetime import datetime as dt
+import pandas as pd
+import math
+from sqlalchemy import create_engine, text
+import multiprocessing as mp
+import os
+from apscheduler.schedulers.blocking import BlockingScheduler
+import traceback
+import psutil
+import pymysql
+import talib as ta
+import numpy as np
+
+
+pd.set_option('display.max_columns', None) # 设置显示最大行
+
+
+path = 'C:\\qmt\\userdata_mini'
+
+field = ['time', 'open', 'close', 'high', 'low', 'volume', 'amount']
+cpu_count = mp.cpu_count()
+
+
+def err_call_back(err):
+    print(f'问题在这里~ error：{str(err)}')
+    traceback.print_exc()
+
+
+def err_call_back(err):
+    print(f'出错啦~ error：{str(err)}')
+    traceback.print_exc()
+
+
+def myself_kdj(df):
+    low_list = df['low_back'].rolling(9, min_periods=9).min()
+    low_list.fillna(value=df['low_back'].expanding().min(), inplace=True)
+    high_list = df['high_back'].rolling(9, min_periods=9).max()
+    high_list.fillna(value=df['high_back'].expanding().max(), inplace=True)
+    rsv = (df['close_back'] - low_list) / (high_list - low_list) * 100
+    df['k'] = pd.DataFrame(rsv).ewm(com=2).mean()
+    df['d'] = df['k'].ewm(com=2).mean()
+    df['j'] = 3 * df['k'] - 2 * df['d']
+    return df
+
+
+# macd指标
+def get_macd_data(data, short=0, long1=0, mid=0):
+    if short == 0:
+        short = 12
+    if long1 == 0:
+        long1 = 26
+    if mid == 0:
+        mid = 9
+    data['sema'] = pd.Series(data['close_back']).ewm(span=short).mean()
+    data['lema'] = pd.Series(data['close_back']).ewm(span=long1).mean()
+    data.fillna(0, inplace=True)
+    data['dif'] = data['sema'] - data['lema']
+    data['dea'] = pd.Series(data['dif']).ewm(span=mid).mean()
+    data['macd'] = 2 * (data['dif'] - data['dea'])
+    data.fillna(0, inplace=True)
+    # return data[['dif', 'dea', 'macd']]
+
+
+# rsi指标
+def get_ris(data):
+    data["rsi_6"] = ta.RSI(data['close_back'], timeperiod=6)
+    data["rsi_12"] = ta.RSI(data['close_back'], timeperiod=12)
+    data["rsi_24"] = ta.RSI(data['close_back'], timeperiod=24)
+
+
+def get_bias(data):
+    # 计算方法：
+    # bias指标
+    # N期BIAS=(当日收盘价-N期平均收盘价)/N期平均收盘价*100%
+    data['bias_6'] = (data['close_back'] - data['close_back'].rolling(6, min_periods=1).mean()) / \
+                     data['close_back'].rolling(6, min_periods=1).mean() * 100
+    data['bias_12'] = (data['close_back'] - data['close_back'].rolling(12, min_periods=1).mean()) / \
+                      data['close_back'].rolling(12, min_periods=1).mean() * 100
+    data['bias_24'] = (data['close_back'] - data['close_back'].rolling(24, min_periods=1).mean()) / \
+                      data['close_back'].rolling(24, min_periods=1).mean() * 100
+    data['bias_6'] = round(data['bias_6'], 2)
+    data['bias_12'] = round(data['bias_12'], 2)
+    data['bias_24'] = round(data['bias_24'], 2)
+
+
+def get_wilr(data):
+    # 威廉指标
+    # 建议用talib库的WILLR方法，亲测有用
+    data['willr'] = ta.WILLR(data['high_back'], data['low_back'], data['close_back'], timeperiod=14)
+
+
+def get_hlfx(data):
+    Trading_signals = 0
+    data_temp = data[['time', 'open_back', 'close_back', 'high_back', 'low_back', 'dif', 'dea', 'macd']]
+    data_temp.columns = ['time', 'open', 'close', 'high', 'low', 'dif', 'dea', 'macd']
+    df_day = pd.DataFrame(columns=['time', 'open', 'close', 'high', 'low', 'volume', 'money', 'HL'])
+    # 先处理去包含
+    for i in data_temp.index:
+        if i == 0 or i == 1:
+            df_day = pd.concat([df_day, data_temp.iloc[[i]]], ignore_index=True)
+        # 不包含
+        elif (df_day.iloc[-1, 3] > data_temp.loc[i, 'high']
+              and df_day.iloc[-1, 4] > data_temp.loc[i, 'low']) \
+                or (df_day.iloc[-1, 3] < data_temp.loc[i, 'high']
+                    and df_day.iloc[-1, 4] < data_temp.loc[i, 'low']):
+            df_day = pd.concat([df_day, data_temp.loc[[i]]], ignore_index=True)
+        # 包含
+        else:
+            # 左高，下降
+            if df_day.iloc[-2, 3] > df_day.iloc[-1, 3]:
+                df_day.iloc[-1, 3] = min(df_day.iloc[-1, 3], data_temp.loc[i, 'high'])
+                df_day.iloc[-1, 4] = min(df_day.iloc[-1, 4], data_temp.loc[i, 'low'])
+            else:
+                # 右高，上升
+                df_day.iloc[-1, 3] = max(df_day.iloc[-1, 3], data_temp.loc[i, 'high'])
+                df_day.iloc[-1, 4] = max(df_day.iloc[-1, 4], data_temp.loc[i, 'low'])
+    # print('111', df_day, data_temp)
+
+    if len(df_day.index) > 2:
+        # 寻找顶底分型
+        for x in range(2, len(df_day.index)):
+            m = x - 1
+            # 底
+            # 符合底分型形态，且第2、3根k线是阳线
+            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'])):
+                # and df_day.loc[x, 'close'] > df_day.loc[x, 'open'] and \
+                #     df_day.loc[x - 1, 'close'] > df_day.loc[x - 1, 'open']:
+
+                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'
+                            # 产生信号，进入hlfx_pool
+                            if x == len(df_day.index) - 1:
+                                Trading_signals = 1
+                        else:
+                            # 不成笔 次级别中枢，保持L* 修订原H为H*
+                            df_day.loc[m, 'HL'] = 'H*'
+                        break
+
+                    elif df_day.loc[m, 'HL'] in ['L', 'LL', 'L*']:
+                        if df_day.loc[m - 1, 'low'] > df_day.loc[x - 1, 'low']:
+                            # 前一个为底更高，且中间不存在更低的底
+                            df_day.loc[x, 'HL'] = 'L'
+                            df_day.loc[m, 'HL'] = '-'
+
+                            # 产生信号，进入hlfx_pool
+                            if x == len(df_day.index) - 1:
+                                Trading_signals = 1
+
+                            # 获得MACD，判断MACD判断背驰
+                            x_macd_dif, x_macd_dea, x_macd_macd = data_temp.loc[x, 'dif'], data_temp.loc[x, 'dea'], \
+                            data_temp.loc[x, 'macd']
+                            m_macd_dif, m_macd_dea, m_macd_macd = data_temp.loc[m, 'dif'], data_temp.loc[m, 'dea'], \
+                            data_temp.loc[m, 'macd']
+
+                            # MACD底背驰
+                            if m_macd_dif < x_macd_dif:
+                                # 次级别背驰底->LL
+                                df_day.loc[x, 'HL'] = 'LL'
+                            break
+                        else:
+                            # 前底更低，本底无效
+                            df_day.loc[x, 'HL'] = '-'
+                            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'])):
+
+                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
+                            if x == len(df_day.index) - 1:
+                                Trading_signals = 2
+                        else:
+                            # 不成笔 次级别中枢，保持H* 修订原L为L*
+                            df_day.loc[m, 'HL'] = 'L*'
+                        break
+
+                    elif df_day.loc[m, 'HL'] in ['H', 'HH', 'H*']:
+                        if df_day.loc[x - 1, 'high'] > df_day.loc[m - 1, 'high']:
+                            # 前一个为顶，且中间存在不包含 or 更高的顶
+                            df_day.loc[x, 'HL'] = 'H'
+                            df_day.loc[m, 'HL'] = '-'
+                            # 产生信号，进入hlfx_pool
+                            if x == len(df_day.index) - 1:
+                                Trading_signals = 2
+
+                            # 获得MACD，判断MACD判断背驰
+                            x_macd_dif, x_macd_dea, x_macd_macd = data_temp.loc[x, 'dif'], data_temp.loc[x, 'dea'], \
+                            data_temp.loc[x, 'macd']
+                            m_macd_dif, m_macd_dea, m_macd_macd = data_temp.loc[m, 'dif'], data_temp.loc[m, 'dea'], \
+                            data_temp.loc[m, 'macd']
+
+                            # MACD顶背驰
+                            if x_macd_dif < m_macd_dif:
+                                # 次级别背驰底->HH
+                                df_day.loc[x, 'HL'] = 'HH'
+                            break
+                        else:
+                            # 前顶更高，本顶无效
+                            df_day.loc[x, 'HL'] = '-'
+                            break
+                    m = m - 1
+                    if m == 0:
+                        df_day.loc[x, 'HL'] = 'H'
+
+            else:
+                df_day.loc[x, 'HL'] = '-'
+    df_temp = df_day[['time', 'HL']]
+
+    return df_temp, Trading_signals
+
+
+def tech_anal(stocks, hlfx_pool, hlfx_pool_daily, err_list):
+    print(f'{dt.now()}开始循环计算！ MyPid is {os.getpid()},池子长度为{len(stocks)}')
+    m = 0
+
+    for stock in stocks:
+        engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/5m_stocks_whole?charset=utf8',
+                               pool_size=1, pool_recycle=7200, max_overflow=1000, pool_timeout=60)
+        engine_tech = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/5m_stocks_tech?charset=utf8',
+                                    pool_size=1, pool_recycle=3600, max_overflow=1000, pool_timeout=60)
+        # print(stock)
+        try:
+            df = pd.read_sql_table('%s_5m' % stock, con=engine.connect())
+            df.dropna(axis=0, how='any')
+            engine.dispose()
+        except BaseException:
+            print(f'{stock}读取有问题')
+            traceback.print_exc()
+            pass
+        else:
+            if len(df) != 0:
+                try:
+                    get_macd_data(df)
+                    get_ris(df)
+                    get_bias(df)
+                    get_wilr(df)
+                    df_temp, T_signals = get_hlfx(df)
+                    df = pd.merge(df, df_temp, on='time', how='left')
+                    df['HL'].fillna(value='-', inplace=True)
+                    df = df.reset_index(drop=True)
+                    # print(stock, '\n', df[['open_front', 'HL']])
+                    df = df.replace([np.inf, -np.inf], np.nan)
+                    df.to_sql('%s_5m' % stock, con=engine_tech, index=False, if_exists='replace')
+                    engine_tech.dispose()
+                # with engine.connect() as con:
+                #     con.execute("ALTER TABLE `%s_5m` ADD PRIMARY KEY (`time`);" % stock)
+                except BaseException as e:
+                    print(f'{stock}存储有问题', e)
+                    traceback.print_exc()
+                    err_list.append(stock)
+                    pass
+                else:
+                    # print(f"{stock} 成功！")
+                    m += 1
+            else:
+                err_list.append(stock)
+                print(f'{stock}数据为空')
+
+            if stock in hlfx_pool and T_signals == 2:
+                hlfx_pool.remove(stock)
+            elif stock not in hlfx_pool and T_signals == 1:
+                hlfx_pool.append(stock)
+                hlfx_pool_daily.append(stock)
+
+    print(f'Pid：{os.getpid()}已经完工了，应处理{len(stocks)},共计算{m}支个股')
+
+
+def ind():
+    sttime = dt.now()
+
+    stocks = xtdata.get_stock_list_in_sector('沪深A股')
+    print(len(stocks))
+    stocks.sort()
+
+    err_list = mp.Manager().list()
+    fre = '5m'
+    engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8',
+                                     pool_size=1, pool_recycle=3600, max_overflow=1000, pool_timeout=60)
+    hlfx_pool = mp.Manager().list()
+    hlfx_pool_daily = mp.Manager().list()
+    hlfx_pool.extend(pd.read_sql_query(
+        text("select value from %s" % fre), engine_hlfx_pool.connect()).iloc[-1, 0].split(","))
+
+    pool = mp.Pool(processes=int(mp.cpu_count()))
+    step = math.ceil(len(stocks) / mp.cpu_count())
+    # pool = mp.Pool(processes=18)
+    # step = math.ceil(len(stocks) / 12)
+    # step = 10000
+    # tech_anal(stocks, hlfx_pool)
+    for i in range(0, len(stocks), step):
+        pool.apply_async(func=tech_anal, args=(stocks[i:i + step], hlfx_pool, hlfx_pool_daily, err_list,),
+                         error_callback=err_call_back)
+    pool.close()
+    pool.join()
+    engine_hlfx_pool.dispose()
+
+    print(f'当日信号：{len(hlfx_pool_daily)}，持续检测为：{len(hlfx_pool)}')
+    print(len(err_list), err_list)
+
+    results_list = ','.join(set(hlfx_pool))
+    results_list_daily = ','.join(set(hlfx_pool_daily))
+
+    # 存档入库
+    db_pool = pymysql.connect(host='localhost',
+                              user='root',
+                              port=3307,
+                              password='r6kEwqWU9!v3',
+                              database='hlfx_pool')
+    cursor_pool = db_pool.cursor()
+    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()
+
+    # 存档入库daily_5m
+    db_pool2 = pymysql.connect(host='localhost',
+                               user='root',
+                               port=3307,
+                               password='r6kEwqWU9!v3',
+                               database='hlfx_pool')
+    cursor_pool2 = db_pool2.cursor()
+    sql2 = "INSERT INTO daily_%s (date,value) VALUES('%s','%s')" % (fre, dt.now().strftime('%Y-%m-%d %H:%M:%S'),
+                                                                    results_list_daily)
+    cursor_pool2.execute(sql2)
+    db_pool2.commit()
+
+    edtime = dt.now()
+    print(edtime - sttime)
+
+
+def to_sql(stock_list):
+    print(f'{dt.now()}开始循环入库！ MyPid is {os.getpid()}')
+    m = 0
+    for stock in stock_list:
+        eng_w = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/5m_stocks_whole?charset=utf8',
+                              pool_recycle=3600, pool_pre_ping=True, pool_size=1)
+        # 后复权数据
+        data_back = xtdata.get_market_data(field, [stock], '5m', end_time='', count=-1, dividend_type='back')
+        df_back = pd.concat([data_back[i].loc[stock].T for i in ['time', 'open', 'high', 'low', 'close', 'volume',
+                                                                 'amount']], axis=1)
+        df_back.columns = ['time', 'open_back', 'high_back', 'low_back', 'close_back', 'volume_back', 'amount_back']
+        df_back['time'] = df_back['time'].apply(lambda x: dt.fromtimestamp(x / 1000.0))
+        df_back.reset_index(drop=True, inplace=True)
+
+        # 前复权数据
+        data_front = xtdata.get_market_data(field, [stock], '5m', end_time='', count=-1, dividend_type='front')
+        df_front = pd.concat([data_front[i].loc[stock].T for i in ['time', 'open', 'high', 'low', 'close', 'volume',
+                                                                   'amount']], axis=1)
+        df_front.columns = ['time', 'open_front', 'high_front', 'low_front', 'close_front', 'volume_front',
+                            'amount_front']
+        df_front['time'] = df_front['time'].apply(lambda x: dt.fromtimestamp(x / 1000.0))
+        df = pd.merge_asof(df_back, df_front, 'time')
+        # print(df)
+        try:
+            # eng_w.connect().execute(text("truncate table `%s_5m`" % stock))
+            df.to_sql('%s_5m' % stock, con=eng_w, index=False, if_exists='replace', chunksize=20000)
+        except BaseException as e:
+            print(stock, e)
+            pass
+        else:
+            m += 1
+
+        eng_w.dispose()
+    print(f'Pid:{os.getpid()}已经完工了.应入库{len(stock_list)},共入库{m}支个股')
+
+
+def download_data():
+    stock_list = xtdata.get_stock_list_in_sector('沪深A股')
+    stock_list.sort()
+    step = math.ceil(len(stock_list) / mp.cpu_count())
+    pool = mp.Pool(processes=mp.cpu_count())
+    for i in range(0, len(stock_list), step):
+        pool.apply_async(func=to_sql, args=(stock_list[i:i+step],), error_callback=err_call_back)
+    pool.close()
+    pool.join()
+    ind()
+
+    print(f'今日数据下载完毕 {dt.now()}')
+
+
+if __name__ == '__main__':
+    field = ['time', 'open', 'close', 'high', 'low', 'volume', 'amount']
+    cpu_count = mp.cpu_count()
+    pus = psutil.Process()
+    # pus.cpu_affinity([12, 13, 14, 15, 16, 17, 18, 19])
+
+    download_data()
+
+    # scheduler = BlockingScheduler()
+    # scheduler.add_job(func=download_data, trigger='cron', day_of_week='0-4', hour='20', minute='05',
+    #                   timezone="Asia/Shanghai", max_instances=10)
+    # try:
+    #     scheduler.start()
+    # except (KeyboardInterrupt, SystemExit):
+    #     pass