摘要：为什么说“最小”？因为很多同学刚入门量化，想要尝试实盘，但又担心成本太高、流程太复杂。其实没那么难，我们完全可以用 miniQMT + 一个极简策略 来跑起来，低成本、低门槛，验证下自己的想法。
今天想和大家聊一个很实用的话题：怎么用 miniQMT 跑一个最小可用的实盘策略。
为什么说“最小”？因为很多同学刚入门量化，想要尝试实盘，但又担心成本太高、流程太复杂。其实没那么难，我们完全可以用 miniQMT + 一个极简策略 来跑起来，低成本、低门槛，验证下自己的想法。
今天这个策略超级简单，主要就是为了让大家搞明白量化的实盘流程。
策略为：查看是否持有某只股票，如果没有就直接开仓，开仓以后如果盈利5%以上就直接卖出，就这么简单。
为了让大家更直观地理解实盘运行的过程，我整理了一张流程图。
接下来是最小可用实盘策略源码分析：
初始化交易模块
首先是实盘策略的入口函数，这里需要配置上资金账号和QMT软件userdata_mini文件夹的实际安装目录。
if __name__ == '__main__': xtdata.enable_hello = False path = r'G:\QMT交易端\userdata_mini' accountID = '资金帐号' xt_trader,acc = init_account(path,accountID) if xt_trader is None or acc is None: print("初始化失败") else: print("="*20+"初始化成功"+"="*20) stock_list = ['510300.SH'] #要交易的标的 main(stock_list=stock_list,xt_trader=xt_trader,acc=acc)
在init_account方法中对实盘模块进行初始化，返回XtQuantTrader和StockAccount2个对象，后期交易会用到。
初始化的时候我们会配置一个回调函数，它会在不同事件发生时触发，比如订单回报、成交回报、下单失败、撤单失败等。实盘操作中，我们可以基于这些回调做很多常规操作。
# 创建交易回调类对象，并声明接收回调 callback = MyXtQuantTraderCallback xt_trader.register_callback(callback)
1.连接管理
回调函数：on_disconnected常规操作：
记录断线时间和状态，方便后续分析尝试自动重连或触发告警（邮件/钉钉/微信）在断线期间暂停策略下单，避免风险 💡 实盘中断线处理很重要，避免策略在异常情况下疯狂下单def on_disconnected(self): print(datetime.datetime.now,'连接断开回调') # 自动重连 self.trader.reconnect # 发通知 notify_user("交易连接断开，已尝试重连")
2.委托相关
回调函数：
on_stock_order（委托回报）on_order_error（委托失败）on_order_stock_async_response（异步委托回报）常规操作：
更新本地订单状态（已报、已撤、报错等）失败重试逻辑（如网络异常、风控失败等）打印或记录委托日志，方便事后回溯异步回报可以触发策略继续执行下一步操作def on_stock_order(self, order): print(datetime.datetime.now, '委托回调', order.order_remark) # 更新本地订单状态 local_order_manager.update(order) def on_order_error(self, order_error): print(f"委托报错回调 {order_error.order_remark} {order_error.error_msg}") # 可以触发报警或重试逻辑
3. 成交相关
回调函数：on_stock_trade常规操作：
更新本地持仓（持仓数量、成本价等）更新资金账户信息（可用资金、冻结资金）判断策略是否达到止盈止损条件可以触发策略下一步动作（比如分批止盈或反向建仓）def on_stock_trade(self, trade): print(datetime.datetime.now, '成交回调', trade.order_remark, f"方向 {trade.offset_flag} 价格 {trade.traded_price} 数量 {trade.traded_volume}") # 更新持仓 portfolio.update(trade) # 判断是否触发止盈止损 strategy.check_exit(trade)
4.撤单相关
回调函数：
on_cancel_error（撤单失败）on_cancel_order_stock_async_response（异步撤单回报）常规操作：
撤单失败记录日志，判断是否需要人工介入异步撤单回报更新本地订单状态在撤单失败时触发补救措施（如再次撤单或手动处理）5. 账户状态
回调函数：on_account_status常规操作：
定期更新本地账户状态（可用资金、持仓情况、保证金）判断资金是否满足策略下单要求触发资金或持仓告警def on_account_status(self, status): print(datetime.datetime.now, sys._getframe.f_code.co_name) account_manager.update(status) if status.available_cash
下载历史行情
接下来我们需要预先下载历史行情数据，以便在策略运行过程中直接调用。比如我们要做一个双均线策略，就必须依赖历史K线来计算均线指标，因此盘前把所需的历史数据下载好是非常关键的。具体下载什么周期的数据，可以根据策略需求来决定：如果是基于分钟级别的策略，那就需要提前把分钟级别的历史数据落地保存，这样实盘时就能直接进行计算，避免因临时获取数据而影响策略执行效率。
def down_data_before_open(stock_list,period='1d'): # 运行前下载历史数据 stock_list的数量不要超过500个 print("开始下载历史行情数据") for stock in stock_list: xtdata.download_history_data(stock, period=period, incrementally=True)
订阅行情
历史行情下载完成后，我们就需要对当天的行情进行订阅。这里的逻辑和历史行情类似：如果需要日线，就订阅日线；如果需要分钟线，就订阅分钟线。订阅完成后，就可以通过 xtdata.get_market_data_ex 获取行情数据。值得注意的是，通过订阅获取的数据，会自动把历史行情和最新行情拼接在一起，方便我们直接使用。
def subscribe_data(stock_list,period='1d'): # 订阅行情 print(f"开始订阅 {period} 行情数据") for stock in stock_list: xtdata.subscribe_quote(stock,period=period)
有了数据，接下来就是策略的核心部分——策略逻辑的设计与实现了。这一步决定了你的策略能否真正发挥价值，是量化交易中最关键的一环。
策略核心逻辑
今天 Demo 里的策略非常简单，主要目的是帮助大家熟悉实盘操作的流程，了解从数据获取到策略执行的整体过程。
首先，我们定义一个死循环，每隔 100 秒执行一次（具体间隔可根据实际需求调整），在循环中定时检查账户状态、更新策略，并根据策略信号触发下单操作。
while True: try: # 策略核心逻辑 except Exception as e: print(f"运行出错了{e}") finally: time.sleep(100) #休眠100s
在这个最小实盘 Demo 中，我们首先需要判断当前是否是周末。理想情况下，如果交易接口能直接返回当天是否为交易日，那就更方便了。这里我们用 Python 的 weekday 简单判断周末：
weekday = now.weekday if weekday == 5 or weekday == 6: print("今天是周末不交易") continue
接下来是判断要交易的标的是否有持仓，没有持仓的情况下会返回None
p = xt_trader.query_stock_position(acc,stock) if p: # 有持仓 需要判断是否盈利大于5% else: # 没有持仓，可以开始交易了
有持仓的话比较方便，我们通过
tick_data = xtdata.get_market_data_ex(stock_list=[stock],period='tick',count=1)
获取股票最新价格，然后计算下收益率满足5%就调用order_stock卖出持有的股票即可
xt_trader.order_stock(acc,stock,xtconstant.STOCK_SELL, p.can_use_volume,price_type=xtconstant.FIX_PRICE ,price=bidPrice1, strategy_name='hj_001',order_remark='test_001')
没有持仓会复杂一些
首先要检查当前是否存在未成交的委托单。如果还有未完成的订单，就不能重复下单。这是很多新手容易忽略的地方，一不小心就可能导致实盘重复下单的风险。
orders = xt_trader.query_stock_orders(acc) is_ordered=False #股票是否已经下单了但是还没有成交 for o in orders: stock_code = o.stock_code order_status = o.order_status # 委托状态是已撤、废单就继续下单 if order_status in [54,57]: pass else: if stock == stock_code: is_ordered = True #已经下单了
接下来需要判断账户可用资金是否充足。如果现金不足，下单就会失败，这也是实盘操作中必须提前检查的环节。
# 获取可用现金 asset = xt_trader.query_stock_asset(acc) cash = asset.cash print(f"可用现金:{cash}") buycash = bidPrice1*1000 if buycash>cash: print("可用现金不足") continue else: # 没有持仓，买入1000股 # 以买1价格 限价买入1000股 print(f"{stock} 买入 ,报价：{bidPrice1}, 买入数量{1000}") order_id = xt_trader.order_stock(acc,stock,xtconstant.STOCK_BUY,1000,price_type=xtconstant.FIX_PRICE,price=bidPrice1, strategy_name='hj_001',order_remark='test_001') print(f"订单号为{order_id}")
至此，一个简单但完整的收盘 Demo 就完成了。整个流程就是这样，实际应用中可以根据策略需求，增加更多判断条件和操作逻辑。
完整的demo源码已经放到知识星球了。
来源：欢子财经
