量化研究--修复雪球交易系统重大的问题

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

结合用户在使用中反馈的问题，我修改了雪球交易系统订单交易id的问题，避免重复下单，网页我也同步更新了可以下载最新的就可以，这个问题发送的原理，因为雪球的调仓会发送2哥订单id，比如A单生成的委托，没有成交的时候的委托123456，这个委托只在不交易的时候存在，比如晚上委托，盘前委托，中午不交易时候的委托调仓，但是在交易的时候，比如盘中成交订单id就会生成其他的id比如678910，这样就会发送一个委托就有2个id，我是利用订单的id识别这个为是不是成交了，2个id我会检查到第一个委托id，下单了就不在下单，担心雪球委托成交了又会给一个id，我会默认这个id没有检查到下单记录，系统就会默认没有下单，就会重新在下单，这样一个委托A就会委托2次，这个委托我修改好了，直接网页下载最新的就可以，这个委托不留意还是很伤脑筋的，网页量化研究---强大索普量化交易网页使用教程

miniqmt下单，网页的授权码中午要就可以

下载最新的时间的就可以

大qmt版本的下载，界面版本的我也更新了建议使用源代码版本

输入账号下载就可以

看一点代码的处理，利用股票代码，交易类型，调整比例剔除一样的委托

可以雪球调仓测试一下

运行代码测试一下

没有问题

会利用订单id检查委托避免多次下单

下单的数据

实时id记录

周末模拟盘不给下单

严格的交易处理

源代码我全部上传了下载，网页的源代码也苦于里面下载

不懂的可以问我朋友就苦于，最近写量化教程有一点忙

完整的源代码参考学习，全部都是系统知识星球下载就可以

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.xueqiou_data import xueqie_data
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
        时间:20251023
        '''
        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()
    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):
        try:
            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)
        except Exception as e:
            print(e,'账户路径检测，没有链接成功*************')
    def get_trader_adjust_data(self,name='西蒙斯测试',zh='ZH3368671'):
        '''
        获取交易
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        st_name=text['策略名称']
        test=text['是否测试']
        n=text['测试长度']
        buy_not_list=text['买入黑名单']
        st_name=text['策略名称']
        u=text['雪球U值']
        token=text['雪球token']
        sp_time=text['错误随机等待时间']
        now_date=str(datetime.now())[:10]
        api=xueqie_data(u=u,xq_a_token=token,assembly_id=zh)
        stats,df=api.get_hist_move()
        if stats==False:
            random_n=random.randint(1,sp_time)
            time.sleep(random_n)
            print(name,'速度太快随机等待时间{}****************'.format(random_n))
        else:
            pass
        if df.shape[0]>0:
            df['updated_at']=pd.to_datetime(df['updated_at'],unit='ms')
            df['created_at']=pd.to_datetime(df['created_at'],unit='ms')
            df['updated_at']=df['updated_at'].apply(lambda x:str(x)[:10])
            df['id']=df['id'].astype(str)
            df['更新时间']=df['updated_at']
            df['名称']=df['stock_name']
            df['证券代码']=df['stock_symbol'].apply(lambda x: str(x)[2:])
            df['买入黑名单']=df['证券代码'].apply(lambda x: '是' if x in buy_not_list else '不是')
            df['prev_weight_adjusted']=df['prev_weight_adjusted'].fillna(0)
            df['调整比例']=df['target_weight']-df['prev_weight_adjusted']
            df['交易类型'] =df['调整比例'].apply(lambda x: 'B' if x > 0 else '不变' if x == 0 else 'S')
            df=df.sort_values(by='created_at',ascending=True)
            df['组合名称']=name
            df['组合id']=zh
            df['投资备注']=st_name+df['交易类型']+df['证券代码']+df['id']
            df=df[['名称',"证券代码",'调整比例','交易类型','id',
                    '组合名称','组合id','投资备注','更新时间']]
            df['买入黑名单']=df['证券代码'].apply(lambda x :"是" if x in buy_not_list else '不是')
            df=df[df['买入黑名单']=='不是']
            if test=='是':
                print('开启测试模型,实盘改成是不开启测试')
                df=df[-n:]
            else:
                df=df[df['更新时间']==now_date]
        else:
            df=pd.DataFrame()
        if df.shape[0]>0:
            df['删除id']=df['证券代码'].astype(str)+','+df['调整比例'].astype(str)+','+df['交易类型'].astype(str)
            df=df.drop_duplicates(subset=['删除id'],keep='first')
            print('*****{}今天有跟单数据*********'.format(now_date))
            print(df)
        else:
            print('*****{}今天没有跟单数据*********'.format(now_date))
        df.to_excel(r'数据2.xlsx')
        return df
    def select_data_type(stock='600031.SH'):
        '''
        选择数据类型
        '''
        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 get_price(self,stock='513100'):
        '''
        获取价格
        '''
        price=self.data.get_spot_data(stock=stock)['最新价']
        return price
    def check_is_sell(self,stock='562310',amount=100):
        '''
        检查是否可以卖出
        '''
        position=self.trader.position()
        if position.shape[0]>0:
            position=position[position['证券代码']==stock]
            if position.shape[0]>0:
                position=position[position['股票余额']>=10]
                if position.shape[0]>0:
                    hold_amount=position['股票余额'].tolist()[-1]
                    av_amount=position['可用余额'].tolist()[-1]
                    if av_amount>=amount and amount>=10:
                        return True
                    elif av_amount< amount and av_amount>=10:
                        return True
                else:
                    return False
            else:
                return False
        else:
            return False
    def check_is_buy(self,stock='513100.SH',amount=100,price=1.3):
        '''
        检查是否可以买入
        '''
        account=self.trader.balance()
        #可以使用的现金
        av_cash=account['可用金额'].tolist()[-1]
        value=amount*price
        if av_cash>=value:
            return True
        else:
            return False
    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 user_def_order_percent(self,
        name='西蒙斯测试',
        total=50000,
        trader_type='buy',
        stock='513100.SH',
        price=1.78,
        ratio=0.1):
        '''
        自定义组合总金额百分比调整函数
        '''
        volue=total*ratio
        amount=volue/price
        amount=self.adjust_amount(stock,amount)
        buy_value=amount*price
        if amount>=10:
            account=self.trader.balance()
            position=self.trader.position()
            if position.shape[0]>0:
                position=position[position['证券代码']==stock]
                if position.shape[0]>0:
                    hold_amount=position['股票余额'].tolist()[-1]
                    av_amount=position['可用余额'].tolist()[-1]
                else:
                    hold_amount=0
                    av_amount=0
            else:
                hold_amount=0
                av_amount=0
            av_cash=account['可用金额'].tolist()[-1]
            if trader_type=='B':
                if self.check_is_buy(stock=stock,amount=amount,price=price):
                    print('{}{} 组合资金{} 调整比例{} 买入 {} '.format(name,stock,total,ratio,amount))
                    return trader_type,amount
                else:
                    print('{} {} 买入不了{} 组合资金{} 调整比例{}，可用金额{} 小于买入市值{}'.format(name,stock,amount,total,ratio,av_cash,buy_value))
                    return '',0
            elif  trader_type=='S':
                if self.check_is_sell(stock=stock,amount=amount):
                    print('{} {} 组合资金{} 调整比例{} 卖出{} '.format(name,stock,total,ratio,amount))
                    return trader_type,amount
                else:
                    print('{} {} 卖出不了，可用数量{} 小于卖出数量{}'.format(name,stock,av_amount,amount))
                    return '',0
            elif trader_type=='不变':
                print('{} 没有变化'.format(stock))
                return '',0
            else:
                print('{} 未知道的交易类型'.format(stock))
                return '',0
        else:
            print('{} 低于最低交易数量'.format(stock))
            return '',0
    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_buy_sell_data(self,total=50000,name='西蒙斯测试1',zh='ZH3368671'):
        '''
        获取买卖数据
        '''
        df=self.get_trader_adjust_data(name=name,zh=zh)
        if df.shape[0]>0:
            amount_list=[]
            position=self.trader.position()
            if position.shape[0]>0:
                position=position[position['股票余额']>=10]
                if position.shape[0]>0:
                    hold_stock_dict=dict(zip(position['证券代码'].tolist(),position['股票余额']))
                    av_stock_dict=dict(zip(position['证券代码'].tolist(),position['可用余额']))
                else:
                    hold_stock_dict={}
                    av_stock_dict={}
            else:
                hold_stock_dict={}
                av_stock_dict={}
            df['股票余额']=df['证券代码'].apply(lambda x :hold_stock_dict.get(x,0))
            df['可用余额']=df['证券代码'].apply(lambda x :av_stock_dict.get(x,0))
            for name,stock,trader_type,ratio,av_amount in zip(df['组合名称'],df['证券代码'],
                df['交易类型'],df['调整比例'],df['可用余额']):
                try:
                    ratio=abs(ratio)/100
                    price=self.get_price(stock)
                    trader_type,amount=self.user_def_order_percent(
                        name=name,
                        total=total,
                        trader_type=trader_type,
                        stock=stock,
                        price=price,
                        ratio=ratio)
                    if trader_type=='B' and amount>=10:
                        amount_list.append(amount)
                    elif trader_type=='S' and amount>=10 and av_amount>=10:
                        if av_amount>=amount:
                            amount_list.append(amount)
                        else:
                            amount_list.append(av_amount)
                    else:
                        amount_list.append(0)
                except Exception as e:
                    print(e,stock,'分析有问题')
                    amount_list.append(0)
            df['交易数量']=amount_list
        else:
            df=pd.DataFrame()
        if df.shape[0]>0:
            maker_list=self.check_not_trader_data()
            df['成功委托']=df['投资备注'].apply(lambda x: '是' if x in maker_list else '不是')
            not_order=df[df['成功委托']=='不是']
            order=df[df['成功委托']=='是']
            net_trader=df[df['交易数量']==0]
            trader=not_order[not_order['交易数量']>=10]
            buy_df=trader[trader['交易类型']=='B']
            sell_df=trader[trader['交易类型']=='S']
            print('已经委托的数据*****')
            print(order)
            print('{}不能交易的股票等待下次成交交易数量为0'.format(name))
            print(net_trader)
            print('{} 买入股票************'.format(name))
            print(buy_df)
            print('{} 卖出股票数据***********'.format(name))
            print(sell_df)
        else:
            buy_df=pd.DataFrame()
            sell_df=pd.DataFrame()
        return buy_df,sell_df
    def seed_trader_data(self,msg='test'):
        '''
        发送交易信号
        '''
        with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        seed_type=text['通知发送方式']
        sender_email =text['发送QQ']
        receiver_email =text['接收QQ']
        password =text['QQ掩码']
        dd_token_list=text['钉钉token']
        wx_token_list=text['企业微信token']
        api=seed_trader_info(
            seed_type=seed_type,
            sender_email=sender_email,
            receiver_email=receiver_email,
            password=password,
            dd_token_list=dd_token_list,
            wx_token_list=wx_token_list
            )
        api.seed_trader_info(msg=msg)
    def run_tarder_func(self,total=50000,name='西蒙斯测试',zh='ZH3368671'):
        '''
        运行交易函数
        '''
        #if self.check_is_trader_date_1():
        if True:
            buy_df,sell_df=self.get_buy_sell_data(
                total=total,name=name,zh=zh)
            #先卖出在买入
            if sell_df.shape[0]>0:
                for stock,amount,maker in zip(sell_df['证券代码'],
                    sell_df['交易数量'],sell_df['投资备注']):
                    price=self.get_price(stock)
                    if self.check_is_sell(stock=stock,amount=amount):
                        self.trader.sell(
                            security=stock,
                            amount=amount,
                            price=price,
                            strategy_name=maker,
                            order_remark=maker)
                        print("{} 卖出{} 数量{} 价格{}".format(name,stock,amount,price))
                        msg="""
                        策略:索普量化雪球交易系统,
                        交易状态:成功,
                        交易类型:卖出,
                        时间:"{}",
                        证券代码:"{}",
                        数量:"{}",
                        价格:"{}",
                        """.format(datetime.now(),stock,amount,price)
                        self.seed_trader_data(msg=msg)
                    else:
                        print('{} 不能卖出{}'.format(name,stock))
            else:
                print('{} 没有卖出的数据'.format(name))
            #买入
            if buy_df.shape[0]>0:
                for stock,amount,maker in zip(buy_df['证券代码'],buy_df['交易数量'],buy_df['投资备注']):
                    price=self.get_price(stock)
                    if self.check_is_buy(stock=stock,amount=amount,price=price):
                        self.trader.buy(
                            security=stock,
                            amount=amount,
                            price=price,
                            strategy_name=maker,
                            order_remark=maker)
                        print("{} 买入{} 数量{} 价格{}".format(name,stock,amount,price))
                        msg="""
                        策略:索普量化雪球交易系统,
                        交易状态:成功,
                        交易类型:买入,
                        时间:"{}",
                        证券代码:"{}",
                        数量:"{}",
                        价格:"{}",
                        """.format(datetime.now(),stock,amount,price)
                        self.seed_trader_data(msg=msg)
                    else:
                        print('{} 不能买入{}'.format(name,stock))
            else:
                print('{} 没有买入的数据'.format(name))
        else:
            print(datetime.now(),'{}不是交易时间'.format(name))
    def get_updata_trader(self):
        '''
        更新多策略交易
        '''
        if self.check_is_trader_date_1():
            with open('{}/策略配置.json'.format(self.path),'r+',encoding='utf-8') as f:
                com=f.read()
            text=json.loads(com)
            name_list=text['组合名称']
            zh_list=text['组合代码']
            total_list=text['组合金额']
            ratio_list=text['组合比例']
            n=text['不同组合间隔时间']
            down_type=text['组合金额模式']
            account=self.trader.balance()
            total_cash=account['总资产'].tolist()[-1]
            for name,zh,total,ratio in zip(name_list,zh_list,total_list,ratio_list):
                try:
                    if down_type=='自定义':
                        total=total
                    elif down_type=='比例':
                        total=total_cash*ratio
                    else:
                        total=total_cash
                    print('*************策略：{} 下单模式：{}********************************'.format(name,down_type))
                    print('************************************************************************************************')
                    self.run_tarder_func(total=total,name=name,zh=zh)
                    time.sleep(n)
                except Exception as e:
                    print(e,name,'策略运行有问题*****************')
        else:
            print('{}交易系统目前不是交易时间'.format(datetime.now()))
    
    def check_is_trader_date_1(self):
        '''
        检测是不是交易时间
        '''
        with open('{}/分析配置.json'.format(self.path),'r+',encoding='utf-8') as f:
            com=f.read()
        text=json.loads(com)
        password=text['软件授权码']
        trader_time=text['交易时间段']
        start_date=text['交易开始时间']
        end_date=text['交易结束时间']
        start_mi=text['开始交易分钟']
        jhjj=text['是否参加集合竞价']
        stats=self.password.decode_trader_password()
        if stats==True:
            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
        else:
            print('**************软件授权码不正确联系作者微信xms_quants1**************')
            return False
if __name__=='__main__':
    '''
    索普量化雪球交易系统
    '''
    with open('分析配置.json','r+',encoding='utf-8') as f:
        com=f.read()
    text=json.loads(com)
    trader_tool=text['交易系统']
    exe=text['同花顺下单路径']
    tesseract_cmd=text['识别软件安装位置']
    qmt_path=text['qmt路径']
    qmt_account=text['qmt账户']
    qmt_account_type=text['qmt账户类型']
    data_api=text['交易数据源']
    trader=joinquant_trader(
        trader_tool=trader_tool,
        exe=exe,
        tesseract_cmd=tesseract_cmd,
        qmt_path=qmt_path,
        qmt_account=qmt_account,
        qmt_account_type=qmt_account_type,
        data_api=data_api)
    trader.connact()
    trader.save_data()
    #3秒运行一次
    schedule.every(0.05).minutes.do(trader.get_updata_trader)
    while True:
        schedule.run_pending()
        time.sleep(1)
   

工具类策略 · 目录

上一篇量化研究--读取读取同花顺自定义板块成分股

作者提示: 个人观点，仅供参考

阅读 415

<iframe src="https://wxa.wxs.qq.com/tmpl/on/base_tmpl.html" class="iframe_ad_container iframe_adv_ad_container"></iframe>