# coding:utf-8
from jqdatasdk import *
import pandas as pd
import pymysql
from sqlalchemy import create_engine
import threading
from datetime import datetime as dt
from jqdatasdk.technical_analysis import *
from xtquant import xtdata, xtconstant
from xtquant.xttype import StockAccount
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
import time
import math
import multiprocessing as mp
import os
#原始版本
# auth('18616891214', 'Ea?*7f68nD.dafcW34d!')
# auth('18521506014', 'Abc123!@#')
# stocks = list(get_all_securities(['stock'], date=dt.today().strftime('%Y-%m-%d')).index)
# stocks = stocks[0:200]
pd.set_option('display.max_columns', None) # 设置显示最大行
fre = '1d'
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print(datetime.datetime.now(), '连接断开回调')
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print(datetime.datetime.now(), '委托回调', order.order_remark)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print(datetime.datetime.now(), '成交回调', trade.order_remark)
def on_order_error(self, order_error):
"""
委托失败推送
:param order_error:XtOrderError 对象
:return:
"""
# print("on order_error callback")
# print(order_error.order_id, order_error.error_id, order_error.error_msg)
print(f"委托报错回调 {order_error.order_remark} {order_error.error_msg}")
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print(datetime.datetime.now(), sys._getframe().f_code.co_name)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print(f"异步委托回调 {response.order_remark}")
def on_cancel_order_stock_async_response(self, response):
"""
:param response: XtCancelOrderResponse 对象
:return:
"""
print(datetime.datetime.now(), sys._getframe().f_code.co_name)
def on_account_status(self, status):
"""
:param response: XtAccountStatus 对象
:return:
"""
print(datetime.datetime.now(), sys._getframe().f_code.co_name)
def err_call_back(err):
print(f'问题在这里~ error：{str(err)}')
def hlfx(data, stocks, pool_list):
print(f'MyPid is {os.getpid()}, now is {dt.now()}')
engine_stock = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/qmt_stocks_tech?charset=utf8')
print(stocks)
for qmt_stock in stocks:
# 读取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']
except BaseException as e:
print(qmt_stock, '未能读取！', e)
pass
else:
# 获得最新价格信息
get_price = data[qmt_stock]
# 调整time时间格式
get_price['time'] = dt.fromtimestamp(get_price['time'] / 1000.0)
# print('成功判定', get_price['time'])
# 先处理去包含
# 不包含
if (df_day.iloc[-1, 3] > get_price['high']
and df_day.iloc[-1, 4] > get_price['low']) \
or (df_day.iloc[-1, 3] < get_price['high']
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']]],
columns=['time', 'open', 'close', 'high', 'low', 'volume', 'amount'])
# print('qmt_______', qmt_df)
df_day = pd.concat([df_day, qmt_df], ignore_index=True)
# print('不包含，合并完成', df_day)
# 包含
else:
# 左高，下降
if df_day.iloc[-2, 3] > df_day.iloc[-1, 3]:
df_day.iloc[-1, 3] = min(df_day.iloc[-1, 3], get_price['high'])
df_day.iloc[-1, 4] = min(df_day.iloc[-1, 4], get_price['low'])
# 右高，上升
else:
df_day.iloc[-1, 3] = max(df_day.iloc[-1, 3], get_price['high'])
df_day.iloc[-1, 4] = max(df_day.iloc[-1, 4], get_price['low'])
# print('包含', df_day)
# 数合并完成，确认df_day
# print(df_day)
# 寻找顶底分型
x = len(df_day.index)-1
m = x - 1
# 底
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'])):
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'
elif df_day.loc[m, 'HL'] == 'L':
if df_day.loc[m - 1, 'low'] > df_day.loc[x - 1, 'low']:
# pool_list.append(qmt_stock)
# 获得MACD，判断MACD判断背驰
x_macd_dif, x_macd_dea, x_macd_macd = df_day.loc[x, 'dif'], df_day.loc[x, 'dea'], \
df_day.loc[x, 'macd']
m_macd_dif, m_macd_dea, m_macd_macd = df_day.loc[m, 'dif'], df_day.loc[m, 'dea'], \
df_day.loc[m, 'macd']
# 背驰底->LL
if m_macd_dif < x_macd_dif:
df_day.loc[x, 'HL'] = 'LL'
# 产生信号，进入hlfx_pool
pool_list.append(qmt_stock)
# 前一个为底更高，且中间不存在更低的底
else:
df_day.loc[x, 'HL'] = 'L'
# 产生信号，进入hlfx_pool
pool_list.append(qmt_stock)
break
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']) and (qmt_stock in pool_list):
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
pool_list.remove(qmt_stock)
break
elif (df_day.loc[m, 'HL'] == 'H'):
if df_day.loc[x - 1, 'high'] > df_day.loc[m - 1, 'high']:
# 获得MACD，判断MACD判断背驰
x_macd_dif, x_macd_dea, x_macd_macd = df_day.loc[x, 'dif'], df_day.loc[x, 'dea'], \
df_day.loc[x, 'macd']
m_macd_dif, m_macd_dea, m_macd_macd = df_day.loc[m, 'dif'], df_day.loc[m, 'dea'], \
df_day.loc[m, 'macd']
# MACD顶背驰
if x_macd_dif < m_macd_dif:
df_day.loc[x, 'HL'] = 'HH'
# 产生卖出信号，进入hlfx_pool
pool_list.remove(qmt_stock)
# 前一个为顶，且中间存在不包含 or 更高的顶
else:
df_day.loc[x, 'HL'] = 'H'
# 产生卖出信号，进入hlfx_pool
pool_list.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()
db_pool = pymysql.connect(host='localhost',
user='root',
port=3307,
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
def prepare():
engine_hlfx_pool = create_engine('mysql+pymysql://root:r6kEwqWU9!v3@localhost:3307/hlfx_pool?charset=utf8')
results = pd.read_sql_query(
'select value from `%s`' % fre, engine_hlfx_pool).iloc[-1, 0].split(",")
results = [x.replace('XSHG', 'SH').replace('XSHE', 'SZ') for x in results]
print('数据库读取,并转化后缀格式', len(results))
# print(results[0:10])
return results
if __name__ == '__main__':
path = r'c:\\qmt\\userdata_mini'
# 生成session id 整数类型 同时运行的策略不能重复
session_id = int(time.time())
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为 800068 的证券账号对象
acc = StockAccount('920000207040', 'SECURITY')
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
print('建立交易连接，返回0表示连接成功', connect_result)
# 对交易回调进行订阅，订阅后可以收到交易主推，返回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()