文章声明:本内容为个人的业余研究，和任何单位，机构没有关系，文章出现的股票代码，全部只是测试例子，不做投资参考，投资有风险，代码学习使用，不做商业用途
最近在写框架的使用教程，最近接触很多朋友，发行对量化感兴趣，但是不知道怎么样使用软件，编写策略，我坚持每天给大家录2个使用视频教程，ptrade基本上线好了
qmt的视频录制由我录制，结合我的经验说明，直接点击视频看就可以
大qmt目前我上线了2个，还在上线使用视频
今天写一下miniqmt雪球系统使用的教程，下载文件夹使用就可以，前面写了大qmt的使用教程量化研究---全新一代索普大QMT雪球交易系统
文件夹直接网页下载就可以量化研究---强大索普量化交易网页使用教程，点击会员下载最新的版本就可以
选择最新的版本
详细的教程可以参考网页https://gitee.com/quants/miniqmt_xueqiuo_trader
1打开miniqmt，详细的全部内容参考网页那个仔细
模拟盘上不去我用我实盘测试一下
2设置qmt账户路径，点击分析配置.json
3设置组合的id，token登，参考网页，点击策略配置，支持多策略
3点击雪球交易系统运行点击右上角就可以
运行的结果
4调仓测试。登录雪球，随便选择一个组合
里面获取到交易信号分析
持仓数量: 0
索普量化测试 516160 买入不了1000 组合资金50000 调整比例0.06670000000000002，可用金额0.0 小于买入市值3099.0
持仓数量: 0
索普量化测试 159949 买入不了2000 组合资金50000 调整比例0.06280000000000001，可用金额0.0 小于买入市值3028.0
持仓数量: 0
索普量化测试 512480 买入不了2000 组合资金50000 调整比例0.05859999999999999，可用金额0.0 小于买入市值2894.0
持仓数量: 0
索普量化测试 513100 买入不了900 组合资金50000 调整比例0.03900000000000001，可用金额0.0 小于买入市值1777.5
持仓数量: 0
持仓数量: 0
索普量化测试 562310 卖出不了，可用数量0 小于卖出数量6600
持仓数量: 0
索普量化测试 516160 买入不了1200 组合资金50000 调整比例0.0748，可用金额0.0 小于买入市值3718.8
持仓数量: 0
索普量化测试 512480 买入不了1000 组合资金50000 调整比例0.031000000000000014，可用金额0.0 小于买入市值1447.0
持仓数量: 0
持仓数量: 0
索普量化测试 159949 卖出不了，可用数量0 小于卖出数量2000
持仓数量: 0
索普量化测试 513100 买入不了1500 组合资金50000 调整比例0.0629，可用金额0.0 小于买入市值2962.5
委托数量 0
目前没有委托
已经委托的数据*****
Empty DataFrame
Columns: [名称, 证券代码, 调整比例, 交易类型, id, 组合名称, 组合id, 投资备注, 更新时间, 买入黑名单, 股票余额, 可用余额, 交易数量, 成功委托]
Index: []
索普量化测试不能交易的股票等待下次成交交易数量为0
名称 证券代码 调整比例 交易类型 id 组合名称 ... 更新时间 买入黑名单 股票余额 可用余额 交易数量 成功委托
8 新能源ETF 516160 6.67 B 244393981 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
7 创业板50ETF 159949 6.28 B 244393980 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
6 半导体ETF 512480 5.86 B 244393979 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
5 纳指ETF 513100 3.90 B 244393978 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
4 沪深300成长ETF 562310 -13.41 S 244393977 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
3 新能源ETF 516160 7.48 B 244399818 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
2 半导体ETF 512480 3.10 B 244399817 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
1 创业板50ETF 159949 -6.09 S 244399816 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
0 纳指ETF 513100 6.29 B 244399815 索普量化测试 ... 2025-11-13 不是 0 0 0 不是
[9 rows x 14 columns]
索普量化测试 买入股票************
Empty DataFrame
Columns: [名称, 证券代码, 调整比例, 交易类型, id, 组合名称, 组合id, 投资备注, 更新时间, 买入黑名单, 股票余额, 可用余额, 交易数量, 成功委托]
Index: []
索普量化测试 卖出股票数据***********
Empty DataFrame
Columns: [名称, 证券代码, 调整比例, 交易类型, id, 组合名称, 组合id, 投资备注, 更新时间, 买入黑名单, 股票余额, 可用余额, 交易数量, 成功委托]
Index: []
索普量化测试 没有卖出的数据
我的账户没有资金，测试正常没有问题的，这个系统使用的客户多，有问题地方我全部改了，可以挂服务器24小时运行
源代码我全部上传了知识星球可以直接下载使用
不懂问我就可以，加我备注入群可以加入量化群
找我朋友开户可以免费使用网页全部内容
源代码框架学习参考，全部框架我开发的
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):
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 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:
print('*****{}今天有跟单数据*********'.format(now_date))
print(df)
else:
print('*****{}今天没有跟单数据*********'.format(now_date))
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):
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)
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)
工具类策略 · 目录
上一篇量化研究--免费送一套大QMT学习资料代码
作者提示: 个人观点，仅供参考
阅读 512
<iframe src="https://wxa.wxs.qq.com/tmpl/ok/base_tmpl.html" class="iframe_ad_container iframe_adv_ad_container"></iframe>
