QMT量化交易模型自动申购股票和可转债分享源代码

本文主要介绍，如何通过miniQMT自动申购股票和可转债，主页简介或si我分享源代码

我们直接看源代码怎么样购买，我写的是利用直接开发的框架可能和原来的qmt不太一样，原理是获取新股数据在购买，xtquant本来的源代码

def query_new_purchase_limit_async(self, account, callback):
        """
        :param account: 证券账号
        :return: 返回账户新股申购额度数据
        """
        req = XTQC.QueryNewPurchaseLimitReq()
        req.m_nAccountType = account.account_type
        req.m_strAccountID = account.account_id
        
        seq = self.async_client.nextSeq()
        return self.common_op_async_with_seq(
            seq,
            (self.async_client.queryNewPurchaseLimitWithSeq, seq, req)
            , callback
        )
    
    def query_ipo_data(self):
        """
        :return: 返回新股新债信息
        """
        req = XTQC.QueryIPODataReq()
        req.m_strIPOType = ''
        
        seq = self.async_client.nextSeq()
        ipo_data_list = self.common_op_sync_with_seq(
            seq,
            (self.async_client.queryIPODataWithSeq, seq, req)
        )
        ipo_data_result = dict()
        for item in ipo_data_list:
            ipo_data_result[item.m_strIPOCode] = {
                'name': item.m_strIPOName, 
                'type':  item.m_strIPOType, 
                'maxPurchaseNum':  item.m_nMaxPurchaseNum, 
                'minPurchaseNum':  item.m_nMinPurchaseNum, 
                'purchaseDate':  item.m_strPurchaseDate, 
                'issuePrice': item.m_dIssuePrice,
            }
        return ipo_data_result
        
    def query_ipo_data_async(self, callback):
        """
        :return: 返回新股新债信息
        """
        req = XTQC.QueryIPODataReq()
        req.m_strIPOType = ''
        
        seq = self.async_client.nextSeq()
        return self.common_op_async_with_seq(
            seq,
            (self.async_client.queryIPODataWithSeq, seq, req)
            , callback
        )
    

我利用直接开发的框架封装的交易代码，使用比较简单

def query_stock_ipo_data(self):        '''        获取股票新可转债ipo数据        dict 新股新债信息数据集        { stock1: info1, stock2: info2, ... }        stock - str 品种代码，例如 '301208.SZ'        info - dict 新股信息        name - str 品种名称        type - str 品种类型        STOCK - 股票，BOND - 债券        minPurchaseNum / maxPurchaseNum - int 最小 / 最大申购额度        单位为股（股票）/ 张（债券）        purchaseDate - str 申购日期        issuePrice - float 发行价        '''        try:            data=self.xt_trader.query_ipo_data()            df=pd.DataFrame(data).T            df['证券代码']=df.index            df.columns=['品种名称','品种类型','最大申购额度','最小申购额度','申购日期','发行价','证券代码']            return df        except:            print('没有数据/处理出现错误')            df=pd.DataFrame()            df['品种名称']=None            df['品种类型']=None            df['最小申购额度']=None            df['最大申购额度']=None            df['申购日期']=None            df['发行价']=None            df['证券代码']=None            return df    def get_buy_stock_ipo(self):        '''        可转债新股一起申购        '''        df=self.query_stock_ipo_data()        for trader_type,name,stock,amount,price in zip(df['品种类型'],df['品种名称'],df['证券代码'],df['最大申购额度'],df['发行价']):            self.buy(security=stock,amount=amount,price=price)            print('新股/可转债申购 交易类型{} 名称{} 代码{} 数量{} 价格{}'.format(trader_type,name,stock,amount,price))

出现外部框架测试，我们随便建立一个数据测试文件

运行的效果

下单的结果

测试的源代码

from xgtrader.unification_data_ths import unification_data_ths
from qmt_trader.unification_data_qmt import unification_data_qmt
from xgtrader.stock_data_ths import stock_data_ths
from qmt_trader.stock_data_qmt import stock_data_qmt
from xgtrader.etf_fund_data_ths import etf_fund_data_ths
from qmt_trader.etf_fund_data_qmt import etf_fund_data_qmt
from qmt_trader.bond_cov_data_qmt import bond_cov_data_qmt
from xgtrader.bond_cov_data_ths import bond_cov_data_ths
from xgtrader.tdx_data import tdx_data
from trader_models.bond_cov_rend_strategy.bond_cov_rend_strategy import  bond_cov_rend_strategy
from trader_tool.unification_data import unification_data
from trader_models.bond_cov_rend_strategy.bond_cov_rend_strategy import bond_cov_rend_strategy
from trader_models.etf_trend_strategy.etf_trend_strategy import etf_trend_strategy
from xgtrader.xgtrader import xgtrader
from trader_tool.tdx_data import tdx_data
from trader_tool.analysis_models import analysis_models
from trader_tool.unification_data import unification_data
import  statsmodels.api as sm
from trader_models.micro_stock_cap_trend_trading.micro_stock_cap_trend_trading import micro_stock_cap_trend_trading
from trader_tool.unification_data import unification_data
from trader_tool.ths_rq import ths_rq
from trader_models.micro_stock_cap_trend_trading.micro_stock_cap_trend_trading import micro_stock_cap_trend_trading
from qmt_trader.qmt_trader_ths import qmt_trader_ths
from qmt_trader.qmt_data import qmt_data
#导入交易框架
from qmt_trader.qmt_trader_ths import qmt_trader_ths
models=qmt_trader_ths()
#连接qmt
models.connect()
#获取iop数据
df=models.query_stock_ipo_data()
print(df)
#买入iop
models.get_buy_stock_ipo()

封装在实盘交易模块的模块设置

 "是否自动申购新股可转债":"是",
    "自动申购新股可转债时间":"14:20",
    "是否自动回购国债":"是",
    "国债代码":["131810.SZ"],
    "国债购买比率":0.8,
    "国债购买时间":"14:57",

实盘运行的效果

开启的模块

实盘交易的效果