适配版本

QMT/download_data_whole.py

@@ -62,10 +62,10 @@ def download_data():
     print(dt.now(), '开始下载！')
     xtdata.download_history_data2(stock_list=stock_list, period='1d', start_time='', end_time='')
     print(dt.now(), '下载完成，准备入库！')
-    # step = math.ceil(len(stock_list) / mp.cpu_count())
-    # pool = mp.Pool(processes=mp.cpu_count())
-    pool = mp.Pool(processes=8)
-    step = math.ceil(len(stock_list) / 8)
+    step = math.ceil(len(stock_list) / mp.cpu_count())
+    pool = mp.Pool(processes=mp.cpu_count())
+    # pool = mp.Pool(processes=8)
+    # step = math.ceil(len(stock_list) / 8)
     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()
@@ -78,7 +78,7 @@ 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])
+    # pus.cpu_affinity([12, 13, 14, 15, 16, 17, 18, 19])
 
     # download_data()
 

QMT/qmt_get_indicators.py

@@ -4,7 +4,7 @@ import numpy as np
 import os
 import pandas as pd
 import time
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, text
 from jqdatasdk import *
 import pymysql
 import multiprocessing as mp
@@ -222,13 +222,14 @@ def get_hlfx(data):
 def tech_anal(stocks, hlfx_pool, hlfx_pool_daily, err_list):
     print(f'{dt.now()}开始循环计算！ MyPid is {os.getpid()},池子长度为{len(stocks)}')
 
-    engine_tech = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_tech?charset=utf8')
+    engine_tech = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_tech?charset=utf8',
+                                pool_recycle=60, max_overflow=-1)
     m = 0
 
     for stock in stocks:
         # print(stock)
         try:
-            df = pd.read_sql_table('%s_1d' % stock, con=engine)
+            df = pd.read_sql_table('%s_1d' % stock, con=engine.connect())
             df.dropna(axis=0, how='any')
         except BaseException:
             print(f'{stock}读取有问题')
@@ -268,6 +269,7 @@ def tech_anal(stocks, hlfx_pool, hlfx_pool_daily, err_list):
                 hlfx_pool.append(stock)
                 hlfx_pool_daily.append(stock)
 
+
     print(f'Pid：{os.getpid()}已经完工了，应处理{len(stocks)},共计算{m}支个股')
 
 
@@ -285,11 +287,11 @@ def ind():
     hlfx_pool = mp.Manager().list()
     hlfx_pool_daily = mp.Manager().list()
     hlfx_pool.extend(pd.read_sql_query(
-        'select value from `%s`' % fre, engine_hlfx_pool).iloc[-1, 0].split(","))
+        text("select value from %s" % fre), engine_hlfx_pool.connect()).iloc[-1, 0].split(","))
 
-    # pool = mp.Pool(processes=int(mp.cpu_count()/2))
+    pool = mp.Pool(processes=int(mp.cpu_count()))
     step = math.ceil(len(stocks) / mp.cpu_count())
-    pool = mp.Pool(processes=18)
+    # pool = mp.Pool(processes=18)
     # step = math.ceil(len(stocks) / 12)
     # step = 10000
     x = 1