stock/QMT/qihuo_get_indicators.py

# coding:utf-8
import time

from datetime import datetime as dt
import socket
import pandas as pd
import numpy as np
from sqlalchemy import create_engine, text
from jqdatasdk import *
import pymysql
import multiprocessing as mp
from multiprocessing import freeze_support
import concurrent.futures
import math
import talib as ta
import os
import traceback
import random
import logging
from myindicator import myind
import psutil
from tqdm import tqdm
from itertools import islice
from func_timeout import func_set_timeout, FunctionTimedOut
from apscheduler.schedulers.blocking import BlockingScheduler

# 显示最大行与列
pd.set_option('display.max_rows', None)
pd.set_option('display.max_columns', None)

def tech_anal(table_list):
    # 创建数据库连接
    engine = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qihuo')
    engine_tech = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qihuo_tech')
    for i in table_list:
        df = pd.read_sql_table(i, con=engine)
        print(i)
        try:
            myind.get_macd_data(df)
            myind.get_ris(df)
            myind.get_bias(df)
            myind.get_wilr(df)
            df = df.round(2)
            df_temp, t_signals = myind.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)
            df = df.replace([np.inf, -np.inf], np.nan)
            df = df.round(2)
        except BaseException as e:
            print(f'{i}计算有问题', e)
        else:
            # 存入数据库
            try:
                # pass
                df.to_sql('%s' % i, con=engine_tech, index=False, if_exists='replace')
                # df.to_sql('%s_1d' % stock, con=engine_tech2, index=False, if_exists='replace')
            except BaseException as e:
                print(f'{i}存储有问题', e)
                traceback.print_exc()
    exit()


# 从qihuo数据库读取所有表数据
def read_sql():
    while True:
        try:
            db = pymysql.connect(host='localhost',
                                 user='root',
                                 port=3307,
                                 password='r6kEwqWU9!v3',
                                 database='qihuo')
            cursor = db.cursor()
            cursor.execute("show tables like '%%%s%%' " % '00.')
            table_list = [tuple[0] for tuple in cursor.fetchall()]
            cursor.close()
            db.close()
        except Exception as e:
            print(e)
            time.sleep(5)
            continue
        finally:
            break
    return table_list


if __name__ == '__main__':
    print('start')
    table_list = read_sql()
    print('已读取期货表')
    tech_anal(table_list)