文章声明:本内容为个人的业余研究，和任何单位，机构没有关系，文章出现的股票代码，全部只是测试例子，不做投资参考，投资有风险，代码学习使用，不做商业用途

这几天星期在前面的优化代码的框架，上线了miniqmt，大qmt的工具策略，同时写了详细的教程，在一个一个的代码检查完善，很多代码还是我2年前写的代码，以前写的代码太烂了，我在建立量化网页，把策略全部完善上传量化网页，目前完善了全部的工具策略，明天完善一下日内T0的模块

量化网页http://14.103.193.242:9999/xms_quants.html

http://14.103.193.242:9999/xms_quants.html

使用前需要先注册，邀请码找就可以

普通用户注册

会员账户注册

我目前完善了全部的工具策略，还在弄东方财富发实盘，bigquat等九完成了，目前完成的大qmt策略给了详细的使用教程点击就可以

mini的

mini的比较复杂

明天我会写每一个模块的使用教程，也可以参考网页的教程，我全部上传了，网页的邀请码找我要就可以，全部内容上传了知识星球下载就可以

不懂的问我就可以，网页的邀请码找我要就可以，备注邀请码，找我朋友开户可以申请使用量化系统

感受一下miniqmt的框架，底层是我开发的框架，实盘还是非常复杂的

from trader_tool.unification_data import unification_data
from trader_tool.trader_frame import trader_frame
import pandas as pd 
import time
from datetime import datetime
import schedule
import json
from trader_tool.base_func import base_func
import os
from trader_tool.xg_tdx import xg_tdx
from trader_tool.seed_trader_info import seed_trader_info
import random
import math
from trader_tool.decode_trader_password import decode_trader_password
class joinquant_trader:
    def __init__(self,
                trader_tool='ths',
                exe='C:/同花顺软件/同花顺/xiadan.exe',
                tesseract_cmd='C:/Program Files/Tesseract-OCR/tesseract',
                qmt_path='D:/国金QMT交易端模拟/userdata_mini',
                qmt_account='55009640',
                qmt_account_type='STOCK',
                data_api='qmt'):
        '''
        索普通达信文本预警交易系统
        作者:索普量化
        微信:xms_quants1
        时间:20251024
        '''
        print('################################################################################################')
        print("""
        风险提示：
        1.以下为量化交易模型，主要内容来源于互联网学习加工，分享的核心是交易思路框架，标的范围和参数仅为学习输出的举例，不能直接运用于交易。
        2.思路框架仅供学习参考，各位投资者朋友需根据自己的需求搭建自己的交易体系和具体策略。
        3.量化交易过程中可能涉及数据准确性、系统BUG、操作不当等风险，交易之前请务必自行充分学习。股市有风险，投资需谨慎和自主决策！
              """)
        print('################################################################################################')
        self.data_api=data_api
        self.exe=exe
        self.tesseract_cmd=tesseract_cmd
        self.trader_tool=trader_tool
        self.qmt_path=qmt_path
        self.qmt_account=qmt_account
        self.qmt_account_type=qmt_account_type
        order_frame=trader_frame(
            trader_tool=self.trader_tool,
            exe=self.exe,
            tesseract_cmd=self.tesseract_cmd,
            qmt_path=self.qmt_path,
            qmt_account=self.qmt_account,
            qmt_account_type=self.qmt_account_type,
            )
        self.trader=order_frame.get_trader_frame()
        data=unification_data(trader_tool=self.trader_tool,data_api=self.data_api)
        self.data=data.get_unification_data()
        self.path=os.path.dirname(os.path.abspath(__file__))
        self.password=decode_trader_password()
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        test=text['是否测试']
        if test=='是':
            print('开启测试模型,实盘记得关闭')
        else:
            self.del_text_data()
    def connact(self):
        '''
        链接交易
        '''
        try:
            self.trader.connect()
            return True
        except Exception as e:
            print("运行错误:",e)
            print('{}连接失败'.format(self.trader_tool))
            return False
    def save_data(self):
        account=self.trader.balance()
        account.to_excel(r'{}/账户数据/账户数据.xlsx'.format(self.path))
        print(account)
        position=self.trader.position()
        position.to_excel(r'{}/持股数据/持股数据.xlsx'.format(self.path))
        print(position)
    def del_text_data(self):
        '''
        清空通达信数据接收最新的数据
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        path=text['通达信警告保存路径']
        try:
            with open(r'{}'.format(path),'r+',encoding='utf-8') as f:
                com=f.truncate(0)
            f.close
        except Exception as e:
            print(e)
    def select_data_type(self,stock='600031'):
        '''
        选择数据类型
        '''
        stock=str(stock)
        if stock[:2] in ['11','12'] or stock[:3] in ['123','110','113','123','127','128','118','132','120']:
            return 'bond'
        elif stock[:2] in ['51','15','50','16','18','52']:
            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','605','688','689',
                ] or stock[:2] in ['11','51','58']:
                stock=stock+'.SH'
            else:
                stock=stock+'.SZ'
        return stock
    def get_price(self,stock='513100'):
        '''
        获取价格
        '''
        price=self.data.get_spot_data(stock=stock)['最新价']
        return price
    def get_hist_tdx_text(self,path='C:/Users/lxg123456/Desktop/预警.txt'):
        '''
        日线预警数据
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        columns_name=text['数据行名称'].split(',')
        try:
            with open(r'{}'.format(path),'r+',encoding='utf-8') as f:
                com=f.readlines()
        except Exception as e:
            with open(r'{}'.format(path),'r+',encoding='gbk') as f:
                com=f.readlines()
        result_list=[]
        if len(com)<0:
            df=pd.DataFrame()
            print('没有警告内容****')
        else:
            for i in com:
                result_list.append(str(i).strip().split('\t'))
            df=pd.DataFrame(result_list)
            if df.shape[0]>0:
                df.columns=columns_name
            else:
                df=pd.DataFrame()
        return df
    def get_tdx_spot_data(self,path='C:/Users/lxg123456/Desktop/实时预警.txt'):
        '''
        读取实时预警数据
        #["证券代码","证券名称","预警时间","价格","涨跌幅","成交量","买卖条件"]
        #['信号日期', '信号时间', '股票代码', '4', '方向', '5', '触发价格', '7', '8', '处理时间']
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        columns_name=text['数据行名称'].split(',')
        try:
            with open(r'{}'.format(path),'r+',encoding='utf-8') as f:
                com=f.readlines()
        except Exception as e:
            with open(r'{}'.format(path),'r+',encoding='gbk') as f:
                com=f.readlines()
        result_list=[]
        if len(com)>0:
            for j in com:
                j=j.strip()
                result=j.split('|')[:-1]
                if len(result)==11:
                    result_list.append(result)
                else:
                    pass
            df=pd.DataFrame(result_list)
            columns=['信号日期',1,'证券代码',3,'买卖条件',5,'价格',7,8,'触发时间','处理时间']
            df.columns=columns
            df['信号日期']=df['信号日期'].apply(lambda x: str(x)[:4]+'-'+str(x)[4:6]+'-'+str(x)[-2:])
            df['信号日期']=df['信号日期']+' '+df['触发时间']
            df['证券名称']=df['证券代码']
            df['涨跌幅']=None
            df['成交量']=None
            df['预警时间']=df['信号日期']
            select=columns_name
            df=df[select]
        else:
            df=pd.DataFrame()
        return df
    def params_tdx_text(self):
        '''
        分析通达信内容
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        path=text['通达信警告保存路径']
        tarder_type=text['交易品种']
        del_list=text['剔除预警标开头'].split(',')
        data_type=text['预警类型']
        st_name=text['策略名称']
        buy_con=text['买入警告条件'].split(',')
        buy_name=text['买入持股同步板块'].split(',')
        buy_name_dict=dict(zip(buy_con,buy_name))
        try:
            if data_type=='实时预警':
                print('实时预警***********************')
                df=self.get_tdx_spot_data(path)
            else:
                print('日线预警***********************')
                df=self.get_hist_tdx_text(path)
        except Exception as e:
            print(e,'有错误，检查预警类型和文件是不是一样')
            df=pd.DataFrame()
        if df.shape[0]>0:
            now_date=str(datetime.now())[:10]
            df['时间']=df['预警时间'].apply(lambda x:str(x)[:10])
            test=text['是否测试']
            if test=='是':
                print('开启测试模式实盘记得关闭**********')
                df=df
            else:
                df=df[df['时间']==now_date]
            df['证券代码']=df['证券代码'].apply(lambda x:str(x)[:6])
            #加上时间参数可以盘中多次交易
            #df['投资备注']=c.st_name+','+df['证券代码']+','+df['买卖条件']+','+df['预警时间']
            df['板块英文']=df['买卖条件'].apply(lambda x: buy_name_dict.get(x,'QT'))
            df['投资备注']=st_name+df['证券代码']+df['板块英文']
            df['同步板块']=df['买卖条件'].apply(lambda x: buy_name_dict.get(x,'QT'))
            df=df.drop_duplicates(subset=['投资备注'],keep='last')
        else:
            df=pd.DataFrame()
        if df.shape[0]>0:
            df['品种']=df['证券代码'].apply(lambda x:self.select_data_type(x))
            df['选择']=df['品种'].apply(lambda x:'是' if x in tarder_type else '不是')
            df=df[df['选择']=='是']
            df['剔除']=df['证券代码'].apply(lambda x: '是' if str(x).startswith(tuple(del_list)) else '不是')
            if df.shape[0]>0:
                df=df[df['剔除']=='不是']
            else:
                df=pd.DataFrame()
        else:
            df=pd.DataFrame()
        return df
    def get_**ysis_buy_sell_data(self):
        '''
        分析买卖数据
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        buy_condi=text['买入警告条件'].split(',')
        sell_condi=text['卖出警告条件'].split(',')
        df=self.params_tdx_text()
        if df.shape[0]>0:
            df['买入']=df['买卖条件'].apply(lambda x: '是' if x in buy_condi else '不是')
            df['卖出']=df['买卖条件'].apply(lambda x: '是' if x in sell_condi else '不是')
        else:
            df=pd.DataFrame()
        return df
    def check_not_trader_data(self):
        '''
        检查单子是不是委托了
        包含了废单避免废单一直下
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        st_name=text['策略名称']
        trader_log=self.trader.today_entrusts()
        if trader_log.shape[0]>0:
            trader_log['策略']=trader_log['委托备注'].apply(lambda x: str(x)[:len(st_name)])
            trader_log=trader_log[trader_log['策略']==st_name]
            if trader_log.shape[0]>0:
                maker_list=trader_log['委托备注'].tolist()
            else:
                maker_list=[]
        else:
            maker_list=[]
        
        return maker_list
        
    def get_trader_stock_data(self):
        '''
        获取股票池数据
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        tarder_type=text['交易品种'].split(',')
        del_list=text['剔除预警标开头'].split(',')
        hold_limit=text['持股限制']
        #读取持股
        hold_stock=self.trader.position()
        if hold_stock.shape[0]>0: