原始版本qmt 实盘趋势交易策略，提供源代码

*******0079 · 发表于 2024-7-20 21:44:41

原始版本qmt 实盘趋势交易策略，提供源代码链接https://mp.weixin.qq.com/s/cAdjgtrFSsR7RYuEEBGSeA
源代码

import schedule
from datetime import datetime
import time
import pandas as pd
from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
from xtquant.xttype import StockAccount
from xtquant import xtconstant
import time
import pandas as pd
from xtquant import xtdata
import numpy as np
import schedule
import math
from datetime import datetime
class MyXtQuantTraderCallback(XtQuantTraderCallback):
def on_disconnected(self):
"""
连接断开
:return:
"""
print("connection lost")
def on_stock_order(self, order):
"""
委托回报推送
:param order: XtOrder对象
:return:
"""
print("on order callback:")
print(order.stock_code, order.order_status, order.order_sysid)
def on_stock_asset(self, asset):
"""
资金变动推送
:param asset: XtAsset对象
:return:
"""
print("on asset callback")
print(asset.account_id, asset.cash, asset.total_asset)
def on_stock_trade(self, trade):
"""
成交变动推送
:param trade: XtTrade对象
:return:
"""
print("on trade callback")
print(trade.account_id, trade.stock_code, trade.order_id)
def on_stock_position(self, position):
"""
持仓变动推送
:param position: XtPosition对象
:return:
"""
print("on position callback")
print(position.stock_code, position.volume)
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)
def on_cancel_error(self, cancel_error):
"""
撤单失败推送
:param cancel_error: XtCancelError 对象
:return:
"""
print("on cancel_error callback")
print(cancel_error.order_id, cancel_error.error_id, cancel_error.error_msg)
def on_order_stock_async_response(self, response):
"""
异步下单回报推送
:param response: XtOrderResponse 对象
:return:
"""
print("on_order_stock_async_response")
print(response.account_id, response.order_id, response.seq)
class xg_mean_line_tarder:
def __init__(self,path= r'D:/国金QMT交易端模拟/userdata_mini',
session_id = 123456,account='55009640',account_type='STOCK',
is_slippage=True,slippage=0.01):
'''
均线趋势交易策略
自己开发的小果交易框架
原始版本的qmt框架
网页 https://gitee.com/li-xingguo11111/qmt_trader
'''
if is_slippage==True:
self.slippage=slippage
else:
self.slippage=0
self.path=path
self.session_id=int(time.time())
self.account=account
self.account_type=account_type
self.xt_trader=''
self.acc=''
#避免下单失败多次下单
#买入记录
self.buy_stock_log=[]
#卖出记录
self.sell_stock_log=[]
#买入的目标金额
self.buy_max_value=10000
#卖出的目标金额
self.sell_max_value=0
#买入的分数
self.buy_score=50
#持有的分数
self.hold_score=50
#股票列表
self.code_list=['513100', '159632', '159941', '159502', '159509', '159655',
'513500', '513300', '513400', '513850', '159659', '159660',
'164824', '159866', '513030', '513080', '513730', '159687',
'159937', '159980', '159985', '159981']
'''
证券代码名称
0 513100 纳斯达克ETF
1 159632 纳斯达克ETF华安
2 159941 纳斯达克ETF广发
3 159502 标普生物ETF
4 159509 纳斯达克科技ETf
5 159655 标普ETF
6 513500 标普500ETF
7 513300 纳斯达克ETF华夏
8 513400 道琼斯ETF
9 513850 美国50ETF
10 159659 纳斯达克100招商ETF
11 159660 纳斯达克100添富ETF
12 164824 印度ETF
13 159866 日本ETF
14 513030 德国ETF
15 513080 法国ETF
16 513730 东南亚科技ETF
17 159687 亚太低碳ETF
18 159937 黄金ETF
19 159980 有色ETF
20 159985 豆粕ETF
21 159981 能源化工ETF
'''
self.data=xtdata
#跌破均线卖出
self.down_mean_line_sell=5
#调整股票代码
self.stock_list=[]
for stock in self.code_list:
self.stock_list.append(self.adjust_stock(stock=stock))
#订阅一分钟的数据，需要更快的话可以订阅tick数据
for stock in self.stock_list:
self.data.subscribe_quote(stock_code=stock,period='1d',start_time='20220101',
end_time='20500101',count=-1)
def check_is_trader_date_1(self,trader_time=4,start_date=9,end_date=14,start_mi=0,jhjj='否'):
'''
检测是不是交易时间
'''
if jhjj=='是':
jhjj_time=15
else:
jhjj_time=30
loc=time.localtime()
tm_hour=loc.tm_hour
tm_min=loc.tm_min
wo=loc.tm_wday
if wo<=trader_time:
if tm_hour>=start_date and tm_hour<=end_date:
if tm_hour==9 and tm_min<jhjj_time:
return False
elif tm_min>=start_mi:
return True
else:
return False
else:
return False
else:
print('周末')
return False
def select_data_type(self,stock='600031'):
'''
选择数据类型
'''
if stock[:3] in ['110','113','123','127','128','111','118'] or stock[:2] in ['11','12']:
return 'bond'
elif stock[:3] in ['510','511','512','513','514','515','516','517','518','588','159','501','164'] or stock[:2] in ['16']:
return 'fund'
else:
return 'stock'
def adjust_stock(self,stock='600031.SH'):
'''
调整代码
'''
if stock[-2:]=='SH' or stock[-2:]=='SZ' or stock[-2:]=='sh' or stock[-2:]=='sz':
stock=stock.upper()
else:
if stock[:3] in ['600','601','603','688','510','511',
'512','513','515','113','110','118','501'] or stock[:2] in ['11']:
stock=stock+'.SH'
else:
stock=stock+'.SZ'
return stock
def adjust_amount(self,stock='',amount=''):
'''
调整数量
'''
if stock[:3] in ['110','113','123','127','128','111'] or stock[:2] in ['11','12']:
amount=math.floor(amount/10)*10
else:
amount=math.floor(amount/100)*100
return amount
def select_slippage(self,stock='600031',price=15.01,trader_type='buy'):
'''
选择滑点
安价格来滑点，比如0.01就是一块
etf3位数,股票可转债2位数
'''
stock=self.adjust_stock(stock=stock)
data_type=self.select_data_type(stock=stock)
if data_type=='fund' or data_type=='bond':
slippage=self.slippage/10
if trader_type=='buy' or trader_type==23:
price=price+slippage
else:
price=price-slippage
else:
slippage=self.slippage
if trader_type=='buy' or trader_type==23:
price=price+slippage
else:
price=price-slippage
return price
def connect(self):
'''
连接
path qmt userdata_min是路径
session_id 账户的标志,随便
account账户,
account_type账户内类型
'''
print('链接qmt')
# path为mini qmt客户端安装目录下userdata_mini路径
path = self.path
# session_id为会话编号，策略使用方对于不同的Python策略需要使用不同的会话编号
session_id = self.session_id
xt_trader = XtQuantTrader(path, session_id)
# 创建资金账号为1000000365的证券账号对象
account=self.account
account_type=self.account_type
acc = StockAccount(account_id=account,account_type=account_type)
# 创建交易回调类对象，并声明接收回调
callback = MyXtQuantTraderCallback()
xt_trader.register_callback(callback)
# 启动交易线程
xt_trader.start()
# 建立交易连接，返回0表示连接成功
connect_result = xt_trader.connect()
if connect_result==0:
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = xt_trader.subscribe(acc)
print(subscribe_result)
self.xt_trader=xt_trader
self.acc=acc
return xt_trader,acc
else:
print(connect_result)
print('qmt连接失败')
def position(self):
'''
对接同花顺
持股
'''
try:
positions = self.xt_trader.query_stock_positions(self.acc)
print("持仓数量:", len(positions))
data=pd.DataFrame()
if len(positions) != 0:
for i in range(len(positions)):
df=pd.DataFrame()
df['账号类型']=[positions[i].account_type]
df['资金账号']=[positions[i].account_id]
df['证券代码']=[positions[i].stock_code]
df['证券代码1']=[positions[i].stock_code]
df['证券代码']=df['证券代码'].apply(lambda x:str(x)[:6])
df['股票余额']=[positions[i].volume]
df['可用余额']=[positions[i].can_use_volume]
df['成本价']=[positions[i].open_price]
df['参考成本价']=[positions[i].open_price]
df['市值']=[positions[i].market_value]
df['冻结数量']=[positions[i].frozen_volume]
data=pd.concat([data,df],ignore_index=True)
return data
else:
df=pd.DataFrame()
df['账号类型']=None
df['资金账号']=None
df['证券代码']=None
df['股票余额']=None
df['可用余额']=None
df['成本价']=None
df['市值']=None
df['选择']=None
df['持股天数']=None
df['交易状态']=None
df['明细']=None
df['证券名称']=None
df['冻结数量']=None
df['市价']=None
df['盈亏']=None
df['盈亏比(%)']=None
df['当日买入']=None
df['当日卖出']=None
return df
except:
df=pd.DataFrame()
return df
def balance(self):
'''
对接同花顺
'''
try:
asset = self.xt_trader.query_stock_asset(account=self.acc)
df=pd.DataFrame()
if asset:
df['账号类型']=[asset.account_type]
df['资金账户']=[asset.account_id]
df['可用金额']=[asset.cash]
df['冻结金额']=[asset.frozen_cash]
df['持仓市值']=[asset.market_value]
df['总资产']=[asset.total_asset]
return df
except:
print('获取账户失败，读取上次数据，谨慎使用')
df=pd.DataFrame()
return df
def buy(self,security='600031.SH', order_type=xtconstant.STOCK_BUY,
amount=100,price_type=xtconstant.FIX_PRICE,price=20,strategy_name='',order_remark=''):
'''
单独独立股票买入函数
'''
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = self.xt_trader.subscribe(self.acc)
print(self.xt_trader.query_stock_asset_async(account=self.acc,callback=subscribe_result))
#print(subscribe_result)
stock_code =self.adjust_stock(stock=security)
price=self.select_slippage(stock=security,price=price,trader_type='buy')
order_volume=amount
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
if order_volume>0:
fix_result_order_id = self.xt_trader.order_stock_async(account=self.acc,stock_code=stock_code, order_type=order_type,
order_volume=order_volume, price_type=price_type,
price=price, strategy_name=strategy_name, order_remark=order_remark)
print('交易类型{} 代码{} 价格{} 数量{} 订单编号{}'.format(order_type,stock_code,price,order_volume,fix_result_order_id))
return fix_result_order_id
else:
print('买入标的{} 价格{} 委托数量{}小于0有问题'.format(stock_code,price,order_volume))
def sell(self,security='600031.SH', order_type=xtconstant.STOCK_SELL,
amount=100,price_type=xtconstant.FIX_PRICE,price=20,strategy_name='',order_remark=''):
'''
单独独立股票卖出函数
'''
# 对交易回调进行订阅，订阅后可以收到交易主推，返回0表示订阅成功
subscribe_result = self.xt_trader.subscribe(self.acc)
print(self.xt_trader.query_stock_asset_async(account=self.acc,callback=subscribe_result))
#print(subscribe_result)
stock_code =self.adjust_stock(stock=security)
price=self.select_slippage(stock=security,price=price,trader_type='sell')
order_volume=amount
# 使用指定价下单，接口返回订单编号，后续可以用于撤单操作以及查询委托状态
if order_volume>0:
fix_result_order_id = self.xt_trader.order_stock(account=self.acc,stock_code=stock_code, order_type=order_type,
order_volume=order_volume, price_type=price_type,
price=price, strategy_name=strategy_name, order_remark=order_remark)
print('交易类型{} 代码{} 价格{} 数量{} 订单编号{}'.format(order_type,stock_code,price,order_volume,fix_result_order_id))
return fix_result_order_id
else:
print('卖出标的{} 价格{} 委托数量{}小于0有问题'.format(stock_code,price,order_volume))
def order_target_value(self,stock='501018',value=1000,price=1.33):
'''
目标价值下单
stock: 股票名字
value: 股票价值，value = 最新价 * 手数 * 保证金率（股票为1） * 乘数（股票为100）
prive交易的的价格
'''
hold_stock=self.position()
try:
del hold_stock['Unnamed: 0']
except:
pass
account=self.balance()
try:
del account['Unnamed: 0']
except:
pass
cash=account['可用金额'].tolist()[-1]
frozen_cash=account['冻结金额'].tolist()[-1]
market_value=account['持仓市值'].tolist()[-1]
total_asset=account['总资产'].tolist()[-1]
amount=math.floor(value/price)
if hold_stock.shape[0]>0:
stock=str(stock)
df1=hold_stock[hold_stock['证券代码']==stock]
if df1.shape[0]>0:
#可以使用的数量兼容t0
av_num=df1['可用余额'].tolist()[-1]
#持有数量
hold_num=df1['股票余额'].tolist()[-1]
#持有的价值
hold_value=df1['市值'].tolist()[-1]
else:
av_num=0
hold_num=0
hold_value=0
else:
av_num=0
hold_num=0
hold_value=0
#买卖价值差转成数量
#可转债最新10
amount=self.adjust_amount(stock=stock,amount=amount)
#买卖的差额
buy_sell_num=math.floor(amount-float(hold_num))
buy_sell_num=self.adjust_amount(stock=stock,amount=buy_sell_num)
buy_sell_value=buy_sell_num*price
#存在买入差额,买入的大于持有的
if buy_sell_num>0:
if buy_sell_num>=10:
#可用现金大于买卖的差额
if cash>=buy_sell_value:
print('目标价值下单{} 目标价值{} 可用资金{}大于买卖资金{} 目标数量{} 持有数量{} 可用数量{} 买入差额{}'.format(stock,value,cash,buy_sell_value,amount,hold_num,av_num,buy_sell_num))
return 'buy',buy_sell_num,price
else:
print('目标价值下单{} 目标价值{} 可用资金{}小于买卖资金{} 目标数量{} 持有数量{} 可用数量{} 买入差额{}不下单'.format(stock,value,cash,buy_sell_value,amount,hold_num,av_num,buy_sell_num))
return '',0,price
else:
print('目标价值下单{} 目标价值{} 可用资金{} 买卖资金{} 目标数量{} 持有数量{} 可用数量{} 买入差额{}不下单小于最小单位去10'.format(stock,value,cash,buy_sell_value,amount,hold_num,av_num,buy_sell_num))
return '',0,price
#存在卖出空间,目标数量小于持有数量卖出：
elif buy_sell_num <0:
#可以卖出的数量多
if av_num>=buy_sell_num:
buy_sell_num=abs(buy_sell_num)
buy_sell_num=self.adjust_amount(stock=stock,amount=buy_sell_num)
print('目标价值下单{} 目标价值{} 目标数量{} 持有数量{} 可用数量{} 卖出差额{}'.format(stock,value,amount,hold_num,av_num,buy_sell_num))
return 'sell',buy_sell_num,price
else:
#可以卖出的不足卖出全部可以卖出的
print('目标价值下单{} 目标价值{} 目标数量{} 持有数量{} 可用数量{}小于卖出数量卖出差额{}'.format(stock,value,amount,hold_num,av_num,buy_sell_num))
return 'sell',av_num,price
else:
print('目标价值下单{} 目标价值{} 目标数量{} 等于持有数量{} 可用数量{} 卖出差额{}不下单'.format(stock,value,amount,hold_num,av_num,buy_sell_num))
return '',0,price
def mean_line_models(self,df):
'''
均线模型
趋势模型
5，10，20，30，60
'''
df=df
df1=pd.DataFrame()
df1['5']=df['close'].rolling(window=5).mean()
df1['10']=df['close'].rolling(window=10).mean()
df1['20']=df['close'].rolling(window=20).mean()
df1['30']=df['close'].rolling(window=30).mean()
df1['60']=df['close'].rolling(window=60).mean()
score=0
#加分的情况
mean_5=df1['5'].tolist()[-1]
mean_10=df1['10'].tolist()[-1]
mean_20=df1['20'].tolist()[-1]
mean_30=df1['30'].tolist()[-1]
mean_60=df1['60'].tolist()[-1]
#相邻2个均线进行比较
if mean_5>mean_10:
score+=25
if mean_10>mean_20:
score+=25
if mean_20>mean_30:
score+=25
if mean_30>mean_60:
score+=25
return score
def down_n_mean_line_sell(self,df):
'''
跌破均线卖出
'''
df['n']=df['close'].rolling(window=self.down_mean_line_sell).mean()
if df['close'].tolist()[-1]<=df['n'].tolist()[-1]:
return True
else:
return False
def get_select_stock_data(self):
'''
选股模型
'''
position=self.position()
if position.shape[0]>0:
position['证券代码']=position['证券代码'].apply(lambda x:self.adjust_stock(x))
hold_stock_list=position['证券代码'].tolist()
else:
hold_stock_list=[]
#选股
#读取订阅数据
df=self.data.get_market_data_ex(stock_list=self.stock_list,period='1d',
start_time='20240101',end_time='20500101',count=-1)
#解析数据
buy_stock_list=[]
for stock in self.stock_list:
try:
hist=df[stock]
score=self.mean_line_models(df=hist)
down=self.down_n_mean_line_sell(df=hist)
if stock in hold_stock_list:
print('{}有持股不买入'.format(stock))
elif stock in self.buy_stock_log:
print('{}有买入记录不交易'.format(stock))
elif score<self.buy_score:
print('{}趋势得分小于买入分数{}不交易'.format(stock,score,self.buy_score))
else:
print("{}符合趋势买入模型".format(stock))
if down:
print('{} 跌破均线{}不买入'.format(stock,self.down_mean_line_sell))
else:
print("{}符合趋势,均线买入模型".format(stock))
buy_stock_list.append(stock)
except Exception as e:
print(e)
return buy_stock_list
def get_stock_sell_data(self):
'''
卖出数据
'''
position=self.position()
if position.shape[0]>0:
position['证券代码']=position['证券代码'].apply(lambda x:self.adjust_stock(x))
hold_stock_list=position['证券代码'].tolist()
else:
hold_stock_list=[]
#选股
#解析数据
sell_stock_list=[]
if len(hold_stock_list)>0:
for stock in hold_stock_list:
self.data.subscribe_quote(stock_code=stock,period='1d',start_time='20220101',
end_time='20500101',count=-1)
#读取订阅数据
df=self.data.get_market_data_ex(stock_list=hold_stock_list,period='1d',
start_time='20240101',end_time='20500101',count=-1)
for stock in hold_stock_list:
try:
hist=df[stock]
score=self.mean_line_models(df=hist)
down=self.down_n_mean_line_sell(df=hist)
if stock in self.buy_stock_log:
print('{}有卖出记录不交易'.format(stock))
elif score<self.hold_score:
print('{}趋势得分小于持有分数{}卖出'.format(stock,score,self.hold_score))
sell_stock_list.append(stock)
elif down:
print('{} 跌破均线{}卖出'.format(stock,self.down_mean_line_sell))
sell_stock_list.append(stock)
else:
print("{}符合趋势持有模型模型".format(stock))
except Exception as e:
print(e)
return sell_stock_list
else:
return sell_stock_list
def start_sell_trader(self):
'''
开始卖出
'''
if self.check_is_trader_date_1(trader_time=8,start_date=0,end_date=24,start_mi=0,jhjj='否'):
sell_stock_list=self.get_stock_sell_data()
if len(sell_stock_list)>0:
for stock in sell_stock_list:
try:
price=self.data.get_full_tick(code_list=[stock])[stock]['lastPrice']
trader_type,amount,price=self.order_target_value(stock=stock[:6],price=price,value=self.sell_max_value)
if trader_type=='sell' and amount>=0:
self.sell(security=stock[:6],price=price,amount=amount)
self.sell_stock_log.append(stock)
else:
print('{}卖出不了'.format(stock))
except Exception as e:
print(e)
else:
print('没有卖出的数据')
else:
print('{}不是交易时间'.format(datetime.now()))
def start_buy_trader(self):
'''
开始买入
'''
if self.check_is_trader_date_1(trader_time=8,start_date=0,end_date=24,start_mi=0,jhjj='否'):
buy_stock_list=self.get_select_stock_data()
if len(buy_stock_list)>0:
for stock in buy_stock_list:
try:
price=self.data.get_full_tick(code_list=[stock])[stock]['lastPrice']
trader_type,amount,price=self.order_target_value(stock=stock[:6],price=price,value=self.buy_max_value)
if trader_type=='buy' and amount>=0:
self.buy(security=stock[:6],price=price,amount=amount)
self.buy_stock_log.append(stock)
else:
print('{}买入不了'.format(stock))
except Exception as e:
print(e)
else:
print('没有买入的数据')
else:
print('{}不是交易时间'.format(datetime.now()))
def update_all_data(self):
'''
更新数据
'''
self.get_select_stock_data()
self.get_stock_sell_data()
self.start_sell_trader()
self.start_buy_trader()
self.buy_stock_log=[]
self.sell_stock_log=[]
def show_info(sef):
print('显示数据{} 程序运行正常'.format(datetime.now()))
if __name__=='__main__':
trader=xg_mean_line_tarder(path= r'D:/国金QMT交易端模拟/userdata_mini',
session_id = 123456,account='55011917',account_type='STOCK',
)
trader.connect()
#qmt——taredr 网页 https://gitee.com/li-xingguo11111/qmt_trader
print(trader.balance())
print(trader.position())
#测试强调前面的#去掉,实盘的时候去掉下面这个代码
trader.update_all_data()
schedule.every().day.at('09:45').do(trader.update_all_data)
schedule.every().day.at('14:45').do(trader.update_all_data)
schedule.every(0.05).minutes.do(trader.show_info)
while True:
schedule.run_pending()
time.sleep(1)