stock/z_jq_his/hlfx.py

import threading
import pymysql
import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime as dt

starttime = dt.now()
# 数据库引擎
# engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qbh_hlfx?charset=utf8')

# 连接数据库
db = pymysql.connect(host='localhost',
                     user='root',
                     port=3307,
                     password='r6kEwqWU9!v3',
                     database='qbh')

fre = '30m'

cursor = db.cursor()
# cursor.execute("select table_name from information_schema.tables where table_schema='qbh_hlfx_backup' and table_name like {}".format('\'%{}\''.format(fre)))
cursor.execute("show tables like '%%%s%%' "% fre)
table_list = [tuple[0] for tuple in cursor.fetchall()]

# print(table_list)
stk = threading.local()


# 主程序
# 找顶底（hdx lfx）分型
def hlfx(table_list, engine, tosql):
    for table in table_list:
        # stk.fxdf = pd.DataFrame(columns=('date', 'open', 'close', 'high', 'low', 'volume', 'money', 'HL'))
        stk.df_day = pd.read_sql_query('select date,open,close,high,low,volume,money,HL from `%s`' % table, engine)

        for i in stk.df_day.index:
            m = i - 1
            if i <= 3:
                # stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
                stk.df_day.loc[i, 'HL'] = '-'
            # 底
            elif ((stk.df_day.loc[i,'high']>stk.df_day.loc[i-1,'high']) and (stk.df_day.loc[i-2,'high']>stk.df_day.loc[i-1,'high'])):
                # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
                    # stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
                stk.df_day.loc[i, 'HL'] = 'L*'
                while m:
                    if m == 1:
                        stk.df_day.loc[i, 'HL'] = 'l'
                    elif stk.df_day.loc[m, 'HL'] == 'H' or stk.df_day.loc[m, 'HL'] == 'h':
                        if(i-m) > 3:
                            stk.df_day.loc[i, 'HL'] = 'L'
                        break
                    elif (stk.df_day.loc[m, 'HL'] == 'L' or stk.df_day.loc[m, 'HL'] == 'l'):
                        if stk.df_day.loc[i-1, 'low'] < stk.df_day.loc[m-1, 'low']:
                            # 前一个为顶，且中间存在不包含 or 更低的底
                            stk.df_day.loc[i, 'HL'] = 'L'
                            break
                        else:
                            break
                    m = m-1

            # 顶
            elif ((stk.df_day.loc[i,'high']<stk.df_day.loc[i-1,'high']) and (stk.df_day.loc[i-2,'high']<stk.df_day.loc[i-1,'high'])):
                # if ((stk.df_day.loc[i-2, 'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-3,'date'] != stk.fxdf.iloc[-1,0]) and (stk.df_day.loc[i-1,'date'] != stk.fxdf.iloc[-1,0])):
                #     stk.fxdf = pd.concat([stk.fxdf, stk.df_day.iloc[[i]]], ignore_index=True)
                stk.df_day.loc[i, 'HL'] = 'H*'
                while m:
                    if m == 1:
                        stk.df_day.loc[i, 'HL'] = 'h'
                    elif stk.df_day.loc[m, 'HL'] == 'L' or stk.df_day.loc[m, 'HL'] == 'l':
                        if i-m > 3:
                            stk.df_day.loc[i, 'HL'] = 'H'
                            stk.df_day.loc[i, 9] = stk.df_day.loc[i, 'close'] - stk.df_day.loc[m, 'close']
                        break
                    elif (stk.df_day.loc[m, 'HL'] == 'H' or stk.df_day.loc[m, 'HL'] == 'h'):
                        if stk.df_day.loc[i-1, 'high'] > stk.df_day.loc[m-1, 'high']:
                            # 前一个为底，且中间存在不包含 or 更高的顶
                            stk.df_day.loc[i, 'HL'] = 'H'
                            break
                        break
                    m = m-1
            else:
                stk.df_day.loc[i, 'HL'] = '-'
        # stk.df_day.to_csv('/Users/daniel/Library/CloudStorage/OneDrive-个人/个人/python_stocks/20220212hlfx2/hlfx%s.csv' % table)
        stk.df_day.to_sql('%s' % table, con=tosql, index=True, if_exists='replace')
        with tosql.connect() as con_backup:
            con_backup.execute('ALTER TABLE `%s` ADD PRIMARY KEY (`date`);' % table)
        print(table, '\n', '**********************************')

# table_list = ['stk002237_1d','stk000004_1d']
# engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qbh_hlfx_backup?charset=utf8')
# tosql = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/bb22?charset=utf8')
# hlfx(table_list, engine, tosql)

step = 100
thread_list = []
engine = []
tosql = []
times_engine = 0
for i in range(0, len(table_list), step):
    engine.append(create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qbh?charset=utf8'))
    tosql.append(create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx?charset=utf8'))
    thread = threading.Thread(target=hlfx, args=(table_list[i:i + step], engine[times_engine], tosql[times_engine]))
    times_engine = times_engine + 1
    thread.start()
    thread_list.append(thread)

for thread in thread_list:
    thread.join()

endtime = dt.now()
print('总时长：', (endtime - starttime).seconds)