经过了2个星期的开发,终于迎来了qm交易框架2.0,超过了3000行源代码,使用类开发,使用方便。
我们看看利用框架写一个简单的实时高频交易策略,很简单
源代码
from qmt_trader.qmt_trader import qmt_trader
from qmt_trader.qmt_data import qmt_data
from qmt_trader.trader_tool import tdx_indicator
import schedule
from datetime import datetime
import time
import pandas as pd
class my_trader:
def __init__(self,path= r'D:/国金QMT交易端模拟/userdata_mini',
session_id = 123456,account='55011917',account_type='STOCK',
is_slippage=True,slippage=0.01,test=True):
'''
实时分钟T0策略利用实盘交易框架2.0
均线金叉买入死差卖出
'''
self.path=path
self.session_id=session_id
self.account=account
self.account_type=account_type
self.is_slippage=is_slippage
self.slippage=slippage
self.test=test
#买入的目标金额
self.buy_max_value=10000
#卖出的目标金额
self.sell_max_value=0
#股票列表
self.code_list=['159937','159980','159985','159981']
#5分,特别提醒这个参数和获取数据的速度有关系默认是3秒一次数据,
# 如果是3秒5分钟就等于3*20*5,short_line=100,这个我后面检验一下
self.short_line=5
#10分钟
self.long_line=10
self.trader=qmt_trader(path=self.path,account=self.account,account_type=self.account_type,
is_slippage=self.is_slippage,slippage=self.slippage)
self.data=qmt_data()
#调整股票代码
self.stock_list=[]
for stock in self.code_list:
self.stock_list.append(self.trader.adjust_stock(stock=stock))
#订阅一分钟的数据,需要更快的话可以订阅tick数据
for stock in self.stock_list:
self.data.subscribe_quote(stock_code=stock,period='1m',start_time='20240101',
end_time='20500101',count=-1)
def connact(self):
'''
链接qmt
'''
try:
self.trader.connect()
print(self.trader.balance())
print(self.trader.position())
return True
except Exception as e:
print("运行错误:",e)
print('{}连接失败'.format('qmt'))
return False
def trader_func(self):
'''
交易函数
'''
#检查是否是交易时间
if self.trader.check_is_trader_date_1(trader_time=8,start_date=9,end_date=14,start_mi=0,jhjj='否'):
#读取订阅数据
df=self.data.get_market_data_ex(stock_list=self.stock_list,period='1m',
start_time='20240101',end_time='20500101',count=-1)
#解析数据
for stock in self.stock_list:
data=pd.DataFrame()
hist=df[stock]
data['date']=hist.index
data['close']=hist['close'].tolist()
data['short_line']=data['close'].rolling(self.short_line).mean()
data['long_line']=data['close'].rolling(self.long_line).mean()
#测试函数
data['test']=data['short_line']>data['long_line']
#金叉
if self.test:
#测试交易
gold_fork=data['test'].tolist()[-1]
else:
gold_fork=tdx_indicator.CROSS_UP(S1=data['short_line'],S2=data['long_line'])[-1]
#死叉
if self.test:
dead_fork=data['test'].tolist()[-1]
else:
dead_fork=tdx_indicator.CROSS_DOWN(S1=data['short_line'],S2=data['long_line'])[-1]
#买入
if gold_fork==True:
#买入
stock=self.trader.adjust_stock(stock=stock)
price=self.data.get_full_tick(code_list=[stock])[stock]['lastPrice']
stock=stock[:6]
trader_type,trader_amount,price=self.trader.order_target_value(stock=stock,price=price,value=self.buy_max_value)
if trader_type=='buy' and trader_amount>=10:
self.trader.buy(security=stock,amount=trader_amount,price=price)
print('{} 死叉 买入 股票{} 数量{} 价格{}'.format(datetime.now(),stock,trader_amount,price))
elif trader_type=='sell' and trader_amount>=10:
self.trader.sell(security=stock,amount=trader_amount,price=price)
print('持有买多了平部分{} 卖出 股票{} 数量{} 价格{}'.format(datetime.now(),stock,trader_amount,price))
else:
print('{} 触发金叉{}执行买入不了'.format(datetime.now(),stock))
else:
print('{} 没有触发金叉{}'.format(datetime.now(),stock))
if dead_fork==True:
#卖出
stock=self.trader.adjust_stock(stock=stock)
price=self.data.get_full_tick(code_list=[stock])[stock]['lastPrice']
stock=stock[:6]
trader_type,trader_amount,price=self.trader.order_target_value(stock=stock,price=price,value=self.sell_max_value)
if trader_type=='sell' and trader_amount>=10:
self.trader.sell(security=stock,amount=trader_amount,price=price)
print('{} 死叉 卖出 股票{} 数量{} 价格{}'.format(datetime.now(),stock,trader_amount,price))
else:
print('{} 触发死叉{}执行卖出不了'.format(datetime.now(),stock))
else:
print('{} 没有触发死叉{}'.format(datetime.now(),stock))
else:
print('{}目前不少交易时间'.format(datetime.now()))
if __name__=='__main__':
trader=my_trader(path= r'D:/国金QMT交易端模拟/userdata_mini',
session_id = 123456,account='55011917',account_type='STOCK',
is_slippage=True,slippage=0.01,test=True)
trader.connact()
#3秒
schedule.every(0.05).minutes.do(trader.trader_func)
while True:
schedule.run_pending()
time.sleep(1)
源代码的非常多,写策略如鱼得水,qmt data
qmt_taredr
超过2000行源代码
运行策略的效果实时t0
实时分析
下单的结果
后面我会继续完善细节,源代码全部上传了后面给教程怎么样当第三方库使用,不懂的内容可以问我,加我备注入群可以进入量化交流群