Merge remote-tracking branch 'origin/master'

QMT/download_data_whole.py

@@ -53,7 +53,9 @@ def to_sql(stock_list):
     print(f'Pid:{os.getpid()}已经完工了.应入库{len(stock_list)},共入库{m}支个股')
 
 
-def download_data(stock_list):
+def download_data():
+    stock_list = xtdata.get_stock_list_in_sector('沪深A股')
+    stock_list.sort()
     print(dt.now(), '开始下载！')
     xtdata.download_history_data2(stock_list=stock_list, period='1d', start_time='', end_time='')
     print(dt.now(), '下载完成，准备入库！')
@@ -64,19 +66,19 @@ def download_data(stock_list):
     pool.close()
     pool.join()
 
+    print(f'今日数据下载完毕 {dt.now()}')
+
 
 
 if __name__ == '__main__':
-    stocks = xtdata.get_stock_list_in_sector('沪深A股')
     field = ['time', 'open', 'close', 'high', 'low', 'volume', 'amount']
     cpu_count = mp.cpu_count()
-    stocks.sort()
-    step = math.ceil(len(stocks) / cpu_count)
 
-    # download_data(stocks)
+
+    # download_data()
 
     scheduler = BlockingScheduler()
-    scheduler.add_job(func=download_data, trigger='cron', hour='15', minute='45', args=[stocks],
+    scheduler.add_job(func=download_data, trigger='cron', day_of_week='0-4', hour='15', minute='40',
                       timezone="Asia/Shanghai")
     try:
         scheduler.start()

QMT/qmt_real_hlfx.py

@@ -1,10 +1,11 @@
 # coding:utf-8
-from jqdatasdk import *
+# from jqdatasdk import *
 import pandas as pd
 import pymysql
 from sqlalchemy import create_engine
 import threading
 from datetime import datetime as dt
+import datetime
 from jqdatasdk.technical_analysis import *
 from xtquant import xtdata, xtconstant
 from xtquant.xttype import StockAccount
@@ -13,6 +14,8 @@ import time
 import math
 import multiprocessing as mp
 import os
+import psutil
+import traceback
 #原始版本
 
 # auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
@@ -90,19 +93,29 @@ class MyXtQuantTraderCallback(XtQuantTraderCallback):
 
 def err_call_back(err):
     print(f'问题在这里~ error：{str(err)}')
+    traceback.print_exc()
 
 
-def hlfx(data, stocks, pool_list):
-    print(f'MyPid is {os.getpid()}, now is {dt.now()}')
+def hlfx(data):
+    stock_list = list(data.keys())
+    print(f'def-->hlfx, MyPid is {os.getpid()}, 本次我需要计算{len(stock_list)},now is {dt.now()}')
+
+    # 获得hlfx_pool池子
+    engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')
+    results = []
+    results.extend(pd.read_sql_query(
+        'select value from `%s`' % fre, engine_hlfx_pool).iloc[-1, 0].split(","))
+    print(f'本次hlfx_pool有{len(results)}个个股')
+
     engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_tech?charset=utf8')
-    print(stocks)
-    for qmt_stock in stocks:
+
+    for qmt_stock in stock_list:
         # 读取qmt_stocks_whole表-前复权-信息
         try:
             df_day = pd.read_sql_query(
                 'select time, open_front, close_front, high_front, low_front, volume_front, amount_front, dif, dea, macd,HL from `%s_%s`'
                 % (qmt_stock, fre), engine_stock)
-            df_day.columns=['time', 'open', 'close', 'high', 'low', 'volume', 'amount', 'dif', 'dea', 'macd', 'HL']
+            df_day.columns = ['time', 'open', 'close', 'high', 'low', 'volume', 'amount', 'dif', 'dea', 'macd', 'HL']
         except BaseException as e:
             print(qmt_stock, '未能读取！', e)
             pass
@@ -121,8 +134,9 @@ def hlfx(data, stocks, pool_list):
                         and df_day.iloc[-1, 4] < get_price['low']):
                 # print('lalallala', get_price['open'], get_price['lastPrice'], get_price['high'],
                 #                             get_price['low'], get_price['volume'], get_price['amount'])
-                qmt_df = pd.DataFrame(data=[[get_price['time'], get_price['open'], get_price['lastPrice'], get_price['high'],
-                                            get_price['low'], get_price['volume'], get_price['amount']]],
+                qmt_df = pd.DataFrame(data=[[get_price['time'], get_price['open'], get_price['lastPrice'],
+                                             get_price['high'], get_price['low'], get_price['volume'],
+                                             get_price['amount']]],
                                       columns=['time', 'open', 'close', 'high', 'low', 'volume', 'amount'])
                 # print('qmt_______', qmt_df)
                 df_day = pd.concat([df_day, qmt_df], ignore_index=True)
@@ -171,12 +185,12 @@ def hlfx(data, stocks, pool_list):
                             if m_macd_dif < x_macd_dif:
                                 df_day.loc[x, 'HL'] = 'LL'
                                 # 产生信号，进入hlfx_pool
-                                pool_list.append(qmt_stock)
+                                results.append(qmt_stock)
                             # 前一个为底更高，且中间不存在更低的底
                             else:
                                 df_day.loc[x, 'HL'] = 'L'
                                 # 产生信号，进入hlfx_pool
-                                pool_list.append(qmt_stock)
+                                results.append(qmt_stock)
                             break
                         break
                     m = m - 1
@@ -186,7 +200,7 @@ def hlfx(data, stocks, pool_list):
             # 顶
 
             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']) and (qmt_stock in pool_list):
+                    df_day.loc[x - 2, 'high'] < df_day.loc[x - 1, 'high']) and (qmt_stock in results):
                 df_day.loc[x, 'HL'] = 'H*'
                 while m:
                     if df_day.loc[m, 'HL'] in ['L', 'LL', 'L*']:
@@ -194,7 +208,7 @@ def hlfx(data, stocks, pool_list):
                             # 成笔->H
                             df_day.loc[x, 'HL'] = 'H'
                             # 产生卖出信号，进入hlfx_pool
-                            pool_list.remove(qmt_stock)
+                            results.remove(qmt_stock)
                             break
 
                     elif (df_day.loc[m, 'HL'] == 'H'):
@@ -209,67 +223,20 @@ def hlfx(data, stocks, pool_list):
                             if x_macd_dif < m_macd_dif:
                                 df_day.loc[x, 'HL'] = 'HH'
                                 # 产生卖出信号，进入hlfx_pool
-                                pool_list.remove(qmt_stock)
+                                results.remove(qmt_stock)
 
                             # 前一个为顶，且中间存在不包含 or 更高的顶
                             else:
                                 df_day.loc[x, 'HL'] = 'H'
                                 # 产生卖出信号，进入hlfx_pool
-                                pool_list.remove(qmt_stock)
+                                results.remove(qmt_stock)
 
                             break
                         break
                     m = m - 1
                     if m == 0:
                         df_day.loc[x, 'HL'] = 'H'
-
-
-
-def bridge(data):
-    # 连接数据库
-    '''
-    db = pymysql.connect(host='localhost',
-                         user='root',
-                         port=3307,
-                         password='r6kEwqWU9!v3',
-                         database='hlfx')
-    cursor = db.cursor()
-    cursor.execute("show tables like '%%%s%%' " % fre)
-    pool_list = [tuple[0] for tuple in cursor.fetchall()]
-    print('取得 table_list %s' % fre)
-    '''
-    '''
-    
-    1.获取hlfx_pool中隔夜的标的
-    2.将本此的data均分，给到进程池
-    3.将data总数据、分配的任务stocklist、hlfx_pool 送入realtime_hlfx中进行计算
-    4.将实时刷新的hlfx存入hlfx_pool 以过滤出现顶分型的标的
-    
-    '''
-
-    # 获得hlfx_pool池子
-    engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')
-    results = mp.Manager().list()
-    results.extend(pd.read_sql_query(
-        'select value from `%s`' % fre, engine_hlfx_pool).iloc[-1, 0].split(","))
-    print(results)
-
-    to_hlfx_list = []
-    keys = list(data.keys())
-    print(len(keys))
-    cpu_count = 4
-    step = math.ceil(len(keys) / cpu_count)
-    for i in range(0, len(keys), step):
-        to_hlfx_list.append([x for x in keys[i:i+step]])
-
-    pool = mp.Pool(processes=cpu_count)
-    for m in range(cpu_count):
-        pool.apply_async(func=hlfx,
-                         args=(data, to_hlfx_list[m], results,), error_callback=err_call_back)
-    pool.close()
-    pool.join()
-
-
+    engine_stock.dispose()
 
     db_pool = pymysql.connect(host='localhost',
                               user='root',
@@ -277,15 +244,19 @@ def bridge(data):
                               password='r6kEwqWU9!v3',
                               database='hlfx_pool')
     cursor_pool = db_pool.cursor()
-
     print(set(results))
     results_list = ','.join(set(results))
     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()
     print(f'{dt.now()}写入新的results，hlfx_pool更新')
-    # hlfx(data, engine_stock, engine_hlfx)
-    pass
+    engine_hlfx_pool.dispose()
+
+
+def bridge(list):
+    print(f'MyPid is {os.getpid()}, now is {dt.now()},我需要负责{len(list)}个个股数据')
+    xtdata.subscribe_whole_quote(list, callback=hlfx)
+    xtdata.run()
 
 
 def prepare():
@@ -296,10 +267,15 @@ def prepare():
     results = [x.replace('XSHG', 'SH').replace('XSHE', 'SZ') for x in results]
     print('数据库读取,并转化后缀格式', len(results))
     # print(results[0:10])
+    engine_hlfx_pool.dispose()
     return results
 
 
 if __name__ == '__main__':
+    mp.freeze_support()
+    pus = psutil.Process()
+    pus.cpu_affinity([0, 1, 2, 3])
+
     path = r'c:\\qmt\\userdata_mini'
     # 生成session id 整数类型 同时运行的策略不能重复
     session_id = int(time.time())
@@ -317,9 +293,19 @@ if __name__ == '__main__':
     # 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
     subscribe_result = xt_trader.subscribe(acc)
     print('对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功', subscribe_result)
-
     stocks = xtdata.get_stock_list_in_sector('沪深A股')
-    xtdata.subscribe_whole_quote(stocks, callback=bridge)
 
-    xtdata.run()
+    cpu_count = 4
+    pool = mp.Pool(processes=cpu_count, maxtasksperchild=8)
+    step = math.ceil(len(stocks) / cpu_count)
+    to_hlfx_list = []
+
+    for i in range(0, len(stocks), step):
+        to_hlfx_list.append([x for x in stocks[i:i+step]])
+
+    for m in range(cpu_count):
+        pool.apply_async(func=bridge,
+                         args=(to_hlfx_list[m],), error_callback=err_call_back)
+    pool.close()
+    pool.join()
 

QMT/real_time.py

@@ -11,6 +11,7 @@ from jqdatasdk import *
 import pymysql
 import multiprocessing as mp
 import math
+import psutil
 
 auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 db_pool = pymysql.connect(host='localhost',
@@ -19,7 +20,7 @@ db_pool = pymysql.connect(host='localhost',
                           password='r6kEwqWU9!v3',
                           database='hlfx_pool')
 cursor_pool = db_pool.cursor()
-engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_front?charset=utf8')
+engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_whole?charset=utf8')
 
 
 def real_price(datas):
@@ -36,11 +37,11 @@ def ma(stock, num, data):
     try:
         i = (num - 1) * -1
         df = pd.read_sql_query(
-            'select close from `%s_1d`' % stock, engine_stock)
+            'select close_front from `%s_1d`' % stock, engine_stock)
     except:
         return 9999999
     else:
-        ma_num = (sum(df['close'][i:]) + data[stock]['lastPrice'])/num
+        ma_num = (sum(df['close_front'][i:]) + data[stock]['lastPrice'])/num
         return ma_num
 
 
@@ -49,11 +50,11 @@ def ma_1(stock, num):
     i = (num) * -1
     try:
         df = pd.read_sql_query(
-            'select close from `%s_1d`' % stock, engine_stock)
+            'select close_front from `%s_1d`' % stock, engine_stock)
     except BaseException:
         return 9999999
     else:
-        ma_num_1 = df['close'][i:].mean()
+        ma_num_1 = df['close_front'][i:].mean()
         return ma_num_1
 
 
@@ -62,16 +63,18 @@ def his_vol(stock, num):
     num = num * -1
     try:
         df = pd.read_sql_query(
-            'select volume from `%s_1d`' % stock, engine_stock)
+            'select volume_front from `%s_1d`' % stock, engine_stock)
     except BaseException:
         return 9999999
     else:
-        return df['volume'].iloc[num]
+        return df['volume_front'].iloc[num]
 
 
 def ma_judge(data, stock_list, results):
-    print('这个ma_judge的PID为：', os.getpid())
-    for stock in data:
+    print(f'这个ma_judge的PID为:{os.getpid()},收到的data数据为:{len(data.keys())},stock_pool长度为{len(stock_list)},now is {dt.now()}')
+    list_judge = list(set(data.keys()) & set(stock_list))
+    print(f'本轮计算:{len(list_judge)}个股')
+    for stock in list_judge:
         i = stock.replace('XSHG', 'SH').replace('XSHE', 'SZ')
         current_price, open_price = data[i]['lastPrice'], data[i]['open']
         MA5, MA10, MA20 = ma(i, 5, data), ma(i, 10, data), ma(i, 20, data)
@@ -88,26 +91,28 @@ def sell_trader(data, positions_dict):
     # for m in data:
     #     print(m, data[m]['lastPrice'])
     print('卖出函数:', dt.now())
-    # positions = xt_trader.query_stock_positions(acc)
-    # print('持仓总数：', len(positions_list))
+    positions = xt_trader.query_stock_positions(acc)
+    print('持仓总数：', len(positions))
 
     for stock, volume in positions_dict.items():
         if stock in data:
-            print('持仓', stock, volume)
             current_price = data[stock]['lastPrice']
             open_price = data[stock]['open']
             MA5 = ma(stock, 5, data)
             MA5_1 = ma_1(stock, 5)
+            print(f'{stock},持仓量为{volume}当前价:{current_price},MA5:{MA5},昨日MA5：{MA5_1},开始判断:')
             if current_price < MA5 or MA5 < MA5_1 or current_price > MA5 * 1.07:
                 print('卖出信号！！！！！！', stock, current_price)
                 order_id = xt_trader.order_stock(acc, stock, xtconstant.STOCK_SELL, volume,
                                                  xtconstant.LATEST_PRICE, 0, 'strategy1', 'order_test')
                 print('价格：', current_price, open_price, MA5, MA5_1)
                 print(order_id, stock, volume)
+        else:
+            print(f'本轮没有持仓股票信息！')
 
 
-def buy_trader(data):
-    print('买入函数:', dt.now())
+def buy_trader(data, positions):
+    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')
@@ -117,7 +122,7 @@ def buy_trader(data):
             'select value from `%s`' % '1d', engine_hlfx_pool)
         stock_pool = stock_pool.iloc[-1, 0].split(",")
         stock_pool.sort()
-        print('stock_pool', stock_pool)
+        print('stock_pool',len(stock_pool))
     except BaseException:
         pass
     '''
@@ -134,10 +139,9 @@ def buy_trader(data):
                     print('append')
     '''
 
-    step = math.ceil(len(stock_pool) / mp.cpu_count())
+    step = math.ceil(len(stock_pool) / (mp.cpu_count()/2))
     print('step:', step)
-    print('cpu_count =', mp.cpu_count())
-    for i in range(0, len(stock_pool), math.ceil(len(stock_pool) / mp.cpu_count())):
+    for i in range(0, len(stock_pool), step):
         p = mp.Process(target=ma_judge, args=(data, stock_pool[i:i + step], results))
         mp_list.append(p)
         p.start()
@@ -178,20 +182,23 @@ def buy_trader(data):
 
         keep_stocks = results_industry.split(",")
         new_keep_stock = [stock.replace('XSHG', 'SH').replace('XSHE', 'SZ') for stock in keep_stocks]
-        print(new_keep_stock)
+        print(f'new_keep_stock is:{len(new_keep_stock)},{new_keep_stock}')
 
-        for stock in data:
+        max_pos = 15
+        for stock in new_keep_stock:
             asset = xt_trader.query_stock_asset(acc)
             cash = asset.cash
-            print(cash)
-            if stock in new_keep_stock:
-                current_price = data[stock]['lastPrice']
-                if cash > 2000:
-                    volume = int((cash / 3 / current_price) // 100 * 100)
-                    print('volume：', volume)
+            if cash > 2000 and len(positions) < max_pos:
+                if stock in new_keep_stock:
+                    current_price = data[stock]['lastPrice']
+                    volume = int((cash / 2 / 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, 'strategy1', 'order_test')
+                    order_id = xt_trader.order_stock(acc, stock, xtconstant.STOCK_BUY, volume, xtconstant.LATEST_PRICE,
+                                                     current_price, 'strategy1', 'order_test')
                     print(order_id)
+            else:
+                print(f'Cash只有：{cash} 或者 现有持仓{len(positions)} 超过了{max_pos}')
+    engine_hlfx_pool.dispose()
     print('一轮结束了，现在时间是：', dt.now())
 
 
@@ -206,7 +213,7 @@ def trader(data):
         sell_trader(data, positions_dict)
 
     # 买入条件
-    buy_trader(data)
+    buy_trader(data, positions)
 
 
 class MyXtQuantTraderCallback(XtQuantTraderCallback):
@@ -277,6 +284,11 @@ class MyXtQuantTraderCallback(XtQuantTraderCallback):
 if __name__ == '__main__':
     auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
 
+    mp.freeze_support()
+    print('cpu_count =', mp.cpu_count())
+    pus = psutil.Process()
+    pus.cpu_affinity([4, 5, 6, 7])
+
     print("start")
     # 指定客户端所在路径
     path = r'c:\\qmt\\userdata_mini'

docker.start.txt

@@ -2,4 +2,7 @@ docker run -itd --name mysql8 -p 3307:3306 -e MYSQL_ROOT_PASSWORD=r6kEwqWU9!v3 -
 
 docker run -itd --name mysql82 -p 3312:3306 -e MYSQL_ROOT_PASSWORD=r6kEwqWU9!v3 mysql:8.0.28-oracle
 
+
+mysqldump -uroot -pr6kEwqWU9!v3 -P3307 --databases qmt_stocks_whole >d:/qmt_stocks_whole.sql
+
 # version order 1