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

昨天写了一个聚宽策略实盘的原理量化研究--全新一代大QMT聚宽交易系统介绍https://mp.weixin.qq.com/s?__biz=MzI5NTE5NTExMw==&mid=2247495765&idx=1&sn=346d6a8d09599f293d28ed00abd76880&scene=21#wechat_redirect

详细的使用教程可以参考网页https://gitee.com/quants/big_qmt_joinquant_trader

https://gitee.com/quants/big_qmt_joinquant_trader

也可以网页下载就可以

回测的结果

我介绍核心的怎么样对接实盘把下面的代码复杂到聚宽的策略开头就可以

import requests
import json
import pandas as pd
from jqdata import *
url='http://101.34.65.108'
port=8888
#自定义服务器编码
url_code='63d85b6189e42cba63feea36381da615c31ad8e36ae420ed67f60f3598efc9ad'
#记得把这个改成自己的一个策略一个,策略的名称找作者建立
password='国九条后中小板微盘小改'
class joinquant_trader:
    def __init__(self,url='http://124.220.32.224',
                port=8025,
                url_code='63d85b6189e42cba63feea36381da615c31ad8e36ae420ed67f60f3598efc9ad',
                password='123456'):
        '''
        获取服务器数据
        '''
        self.url=url
        self.port=port
        self.url_code=url_code
        self.password=password
    def get_user_data(self,data_type='用户信息'):
        '''
        获取使用的数据
        data_type='用户信息','实时数据',历史数据','清空实时数据','清空历史数据'
        '''
        url='{}:{}/_dash-update-component'.format(self.url,self.port)
        headers={'Content-Type':'application/json'}
        data={"output":"joinquant_trader_table.data@{}".format(self.url_code),
            "outputs":{"id":"joinquant_trader_table","property":"data@{}".format(self.url_code)},
            "inputs":[{"id":"joinquant_trader_password","property":"value","value":self.password},
                    {"id":"joinquant_trader_data_type","property":"value","value":data_type},
                    {"id":"joinquant_trader_text","property":"value","value":"\n               {'状态': 'held', '订单添加时间': 'datetime.datetime(2024, 4, 23, 9, 30)', '买卖': 'False', '下单数量': '9400', '已经成交': '9400', '股票代码': '001.XSHE', '订单ID': '1732208241', '平均成交价格': '10.5', '持仓成本': '10.59', '多空': 'long', '交易费用': '128.31'}\n                "},
                    {"id":"joinquant_trader_run","property":"value","value":"运行"},
                    {"id":"joinquant_trader_down_data","property":"value","value":"不下载数据"}],
                    "changedPropIds":["joinquant_trader_run.value"],"parsedChangedPropsIds":["joinquant_trader_run.value"]}
        res=requests.post(url=url,data=json.dumps(data),headers=headers)
        text=res.json()
        df=pd.DataFrame(text['response']['joinquant_trader_table']['data'])
        return df
    def send_order(self,result):
        '''
        发送交易数据
        '''
        url='{}:{}/_dash-update-component'.format(self.url,self.port)
        headers={'Content-Type':'application/json'}
        data={"output":"joinquant_trader_table.data@{}".format(self.url_code),
            "outputs":{"id":"joinquant_trader_table","property":"data@{}".format(self.url_code)},
            "inputs":[{"id":"joinquant_trader_password","property":"value","value":self.password},
                    {"id":"joinquant_trader_data_type","property":"value","value":'发送信号'},
                    {"id":"joinquant_trader_text","property":"value","value":result},
                    {"id":"joinquant_trader_run","property":"value","value":"运行"},
                    {"id":"joinquant_trader_down_data","property":"value","value":"不下载数据"}],
                    "changedPropIds":["joinquant_trader_run.value"],"parsedChangedPropsIds":["joinquant_trader_run.value"]}
        res=requests.post(url=url,data=json.dumps(data),headers=headers)
        text=res.json()
        df=pd.DataFrame(text['response']['joinquant_trader_table']['data'])
        return df
#继承类
xg_data=joinquant_trader(url=url,port=port,password=password,)
def send_order(result):
    '''
    发送函数
    status: 状态, 一个OrderStatus值
    add_time: 订单添加时间, [datetime.datetime]对象
    is_buy: bool值, 买还是卖，对于期货:
    开多/平空 -> 买
    开空/平多 -> 卖
    amount: 下单数量, 不管是买还是卖, 都是正数
    filled: 已经成交的股票数量, 正数
    security: 股票代码
    order_id: 订单ID
    price: 平均成交价格, 已经成交的股票的平均成交价格(一个订单可能分多次成交)
    avg_cost: 卖出时表示下卖单前的此股票的持仓成本, 用来计算此次卖出的收益. 买入时表示此次买入的均价(等同于price).
    side: 多/空，'long'/'short'
    action: 开/平， 'open'/'close'
    commission交易费用（佣金、税费等）
    '''
    data={}
    data['状态']=str(result.status)
    data['订单添加时间']=str(result.add_time)
    data['买卖']=str(result.is_buy)
    data['下单数量']=str(result.amount)
    data['已经成交']=str(result.filled)
    data['股票代码']=str(result.security)
    data['订单ID']=str(result.order_id)
    data['平均成交价格']=str(result.price)
    data['持仓成本']=str(result.avg_cost)
    data['多空']=str(result.side)
    data['交易费用']=str(result.commission)
    result=str(data)
    xg_data.send_order(result)
    return data
def xg_order(func):
    '''
    继承order对象 数据交易函数
    '''
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        if result == None:
            return
        send_order(result)
        return result
    return wrapper
def xg_order_target(func):
    '''
    继承order_target对象 百分比
    '''
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        if result == None:
            return        
        send_order(result)
        return result
    return wrapper

def xg_order_value(func):
    '''
    继承order_value对象 数量
    '''
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        if result == None:
            return        
        send_order(result)
        return result
    return wrapper
def xg_order_target_value(func):
    '''
    继承order_target_value对象 数量
    '''
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        if result == None:
            return        
        send_order(result)
        return result
    return wrapper
order = xg_order(order)
order_target = xg_order_target(order_target)
order_value = xg_order_value(order_value)
order_target_value = xg_order_target_value(order_target_value)

设置策略授权码比如password参数，授权码取一个名字发我就可以

点击一下回测看一下数据是否发送到服务器就可以，数据发送到了服务器没有问题

速度还是非常快的

qmt设置导入策略

源代码版本

点击编辑输入账户，我选择主要的内容说明，其他的全部参考网页教程

输入授权码，和聚宽，服务器，保持一样

名字无所谓随便取

今天没有实时数据，是否测试参数改成是，测试一下下单结果，实盘一点要改成否

点击模型交易导入策略就可以

点击运行

策略目前全部平仓了，没有买入没有问题

没有问题参数是否测试改否挂实盘就可以

点击运行就可以

不懂的问我就可以

"四大搅屎棍策略"深度原理剖析

策略哲学基础

市场无效性假设

该策略基于行为金融学理论，认为A股市场存在系统性行为偏差：

• 散户主导市场导致非理性波动

• 机构跟风效应加剧行业轮动

• 政策敏感性催生结构性机会

小市值溢价理论

策略核心信仰小市值效应在中国市场的持续性：

• 小市值公司成长性更高

• 机构覆盖度低，存在定价错误

• 流动性溢价带来超额收益

行业轮动机制深度解析

行业宽度计算原理

# 关键技术指标：行业宽度df_bias = (df_close.iloc[:, -p_count:] > df_ma20) df_ratio = (df_bias.groupby('industry_code').sum() * 100.0) / df_bias.groupby('industry_code').count()

数学原理：

• 计算每个行业内股价在20日均线以上的股票比例

• 比例越高代表行业整体趋势越强

• 使用滚动窗口计算确保无未来函数

行为金融学解释：

• 当行业宽度高时，说明资金集中流入该行业

• 反映了机构调仓行为和市场共识形成

• 宽度指标比单纯价格涨幅更能反映行业热度

"搅屎棍"行业的心理学基础

四个被排除的行业具有共同特征：

1. 银行(801780)

• 市场地位：权重股，拉升指数但个股不涨

• 资金效应：吸金效应强，挤压小盘股流动性

• 行为模式：银行股大涨通常伴随市场风格切换

2. 有色金属(801050)

• 周期性：强周期行业，波动剧烈

• 外部依赖：受大宗商品价格影响大

• 资金性质：通常为避险或通胀对冲资金

3. 钢铁(801040)

• 政策敏感性：供给侧改革影响显著

• 同质化：个股差异小，板块联动强

• 流动性：大盘股，消耗大量资金

4. 煤炭(801950)

• 传统行业：成长性差，估值天花板低

• 环保压力：长期政策压制

• 季节性：受天气和政策周期影响

核心逻辑：当这些传统大盘股行业走强时，意味着：

• 风险偏好下降，资金避险

• 市场风格向价值股切换

• 小盘股的资金面承压

多因子选股模型详解

基本面因子筛选​​​​​​​

query(    valuation.code,).filter(    valuation.code.in_(choice),    indicator.roe > 0.15,      # 净资产收益率 > 15%    indicator.roa > 0.10,      # 总资产收益率 > 10%).order_by(    valuation.market_cap.asc() # 按市值升序排列).limit(g.stock_num)

ROE因子(>15%)

• 经济意义：衡量股东权益的回报率

• 阈值设定：15%确保公司具备持续盈利能力

• 排除效应：过滤掉盈利能力差的僵尸企业

ROA因子(>10%)

• 经济意义：衡量总资产的使用效率

• 双重验证：与ROE结合，防止高杠杆扭曲

• 质量保证：确保盈利来自真实经营能力

小市值因子

• 学术基础：Fama-French三因子模型

• A股特性：小盘股溢价在中国市场显著

• 逆向思维：在机构关注度低的领域寻找机会

技术面过滤机制

流动性过滤​​​​​​​

def filter_paused_stock(stock_list):    return [stock for stock in stock_list if not current_data[stock].paused]

• 停牌风险：避免资金被锁定

• 信息不对称：复牌后可能补跌

风险警示过滤

python

deffilter_st_stock(stock_list):return[stock for stock in stock_list            ifnot current_data[stock].is_st            and'ST'notin current_data[stock].name            and'*'notin current_data[stock].name            and'退'notin current_data[stock].name]

• 退市风险：ST/*ST/退市预警股票

• 极端事件：避免黑天鹅事件冲击

特殊板块过滤

python

deffilter_kcbj_stock(stock_list):for stock in stock_list[:]:if stock[0]=='4'or stock[0]=='8'or stock[:2]=='68'or stock[0]=='3':

• 科创板(68开头)：波动率大，定价机制不同

• 北交所(4/8开头)：流动性差，机构参与度低

• 创业板(3开头)：虽保留但通过其他条件严格控制

价格限制过滤

python

deffilter_limitup_stock(context, stock_list):return[stock for stock in stock_list if stock in context.portfolio.positions.keys()or last_prices[stock][-1]< current_data[stock].high_limit]

• 涨停限制：避免追高被套

• 持仓例外：已持仓的涨停股继续持有享受溢价

上市时间过滤

python

deffilter_new_stock(context, stock_list):return[stock for stock in stock_list ifnot yesterday - get_security_info(stock).start_date < datetime.timedelta(days=375)]

• 次新股风险：估值泡沫，波动剧烈

• 稳定期要求：上市满375天，财务数据完整

交易执行算法深度分析

仓位管理策略

等权重分配

python

value = context.portfolio.cash / buy_numfor stock in target_B:if stock notinlist(context.portfolio.positions.keys()):if open_position(stock, value):

• 风险分散：避免单个股票过度影响组合

• 再平衡机制：每周调仓实现动态平衡

• 资金利用：满仓操作，追求收益最大化

调仓逻辑

python

for stock in g.hold_list:if(stock notin target_B)and(stock notin g.yesterday_HL_list):         position = context.portfolio.positions[stock]         close_position(position)

• 严格纪律：不在目标列表立即卖出

• 涨停例外：昨日涨停股享受持有特权

• 动量延续：承认涨停股的短期动量效应

涨停板特殊处理机制

行为金融学基础

python

defcheck_limit_up(context):if current_data.iloc[0,0]< current_data.iloc[0,1]:         log.info("[%s]涨停打开，卖出"%(stock))else:         log.info("[%s]涨停，继续持有"%(stock))

涨停板现象的心理学解释：

• 注意力效应：涨停吸引市场关注

• 羊群行为：投资者跟风买入

• 处置效应：获利了结压力在涨停打开时集中释放

策略应对逻辑：

• 尾盘观察：14:00检查，避免盘中波动误判

• 严格止损：涨停打开立即卖出，锁定收益

• 动量延续：连续涨停享受超额收益

风险控制体系

系统性风险控制

市场环境识别

通过行业宽度和"搅屎棍"行业监控市场状态：

• 牛市环境：小盘股普涨，行业宽度扩散

• 熊市环境：资金避险，"搅屎棍"行业走强

• 震荡环境：行业轮动加快，需要严格筛选

风格切换预警

当银行、有色等大盘股行业进入强势前列时：

• 降低仓位：直接空仓，避免风格切换损失

• 等待时机：直到小盘股友好环境再现

个股风险控制

多层过滤机制

1. 行业层面：避开不利行业环境

2. 基本面层面：ROE/ROA质量要求

3. 技术面层面：均线位置、涨跌停状态

4. 流动性层面：停牌、ST风险排除

交易执行风险控制

python

set_slippage(FixedSlippage(0))set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.0003, close_commission=0.0003))

• 零滑点假设：理想化条件，实际需要调整

• 真实成本：考虑印花税和佣金影响

• 最小佣金：避免小金额交易成本过高

策略优势与局限

核心优势

1. 多维度验证：行业、基本面、技术面三重筛选

2. 行为金融应用：充分利用市场非有效性

3. 严格风控：多层过滤，系统性风险识别

4. 中国特色：涨停板策略贴合A股实际

这个策略体现了深度市场理解与严格量化纪律的结合，是典型的基于中国特色的多因子选股策略。