Quickfix Python环境下心跳消息与订单事件的消息序列号重复问题解决问询

问题根源分析 #

你遇到的核心问题在于错误地在toAdmin回调中触发业务消息发送。QuickFix的toAdmin方法是在Admin类消息（比如心跳、登录登出）即将被发送前调用的，此时心跳消息的MsgSeqNum已经被会话分配完成，但还未完成发送和序列号递增操作。这时候你在toAdmin里调用sendToTarget发送订单，会话会复用当前已分配给心跳的序列号，导致两个消息使用相同的34字段值，最终引发协议冲突。

另外，你的线程初始化代码还有个隐藏问题：在run方法的死循环中反复创建新的SocketAcceptor实例，这会导致多个会话同时运行，进一步加剧序列号混乱。

分步解决方案 #

1. 移除toAdmin中的业务逻辑，改用独立轮询线程

不要依赖心跳事件触发数据库查询，单独启动一个后台线程专门处理定期轮询，这样业务消息的发送和心跳消息的处理完全独立，不会抢占序列号。

修改你的Application类，在初始化时启动轮询线程：

class Application(fix.Application):
    def __init__(self):
        super().__init__()
        # 启动数据库轮询后台线程，和心跳间隔保持一致（30秒）
        self.poll_thread = threading.Thread(target=self._db_poll_loop)
        self.poll_thread.daemon = True
        self.poll_thread.start()

    def _db_poll_loop(self):
        while True:
            time.sleep(30)
            # 获取当前所有活跃的会话ID
            active_sessions = fix.Session.getSessions()
            for session_id in active_sessions:
                self.checkDB(session_id)

    def checkDB(self, sessionID):
        print("Start checking for new orders")
        fixtradesdf = pd.read_sql_query("Select * from database where ordstatus = '0'", engine)
        for index, row in fixtradesdf.iterrows():
            message = fix.Message()
            header = message.getHeader()
            # 只需要设置消息类型，其他头部字段（BeginString、SenderCompID等）由Session自动填充
            header.setField(fix.MsgType("D"))
            
            # 填充订单字段
            message.setField(fix.ClOrdID(uuid.uuid4().hex))  # 用UUID生成唯一订单ID
            message.setField(fix.Symbol("N/A"))
            side = fix.Side_BUY if row['side'] == 'BUY' else fix.Side_SELL
            message.setField(fix.Side(side))
            message.setField(fix.TransactTime())
            message.setField(fix.OrderQty(row['orderqty']))
            message.setField(fix.Currency("EUR"))
            message.setField(fix.SecurityID("XXXX"))
            message.setField(fix.SecurityIDSource("4"))
            message.setField(fix.OrdType(fix.OrdType_MARKET))
            message.setField(fix.PriceType(1))
            message.setField(fix.QuoteType(1))
            
            # 填充Party组
            group = fix.Group(453, 448)
            group.setField(fix.PartyID("XXXX"))
            group.setField(fix.PartyIDSource("D"))
            group.setField(fix.PartyRole(7))
            message.addGroup(group)
            
            message.setField(fix.OnBehalfOfCompID("XXXX"))
            # 发送消息，Session会自动处理序列号递增
            fix.Session.sendToTarget(message, sessionID)
            
            # 关键：发送后立即更新数据库订单状态，避免重复发送
            engine.execute("UPDATE database SET ordstatus = '1' WHERE id = %s", (row['id'],))

2. 修复Acceptor的线程初始化逻辑

原来的Threading类在死循环中反复创建SocketAcceptor，这会导致多个会话实例，必须改成只初始化一次：

class Threading(object):
    def __init__(self, interval=1):
        self.interval = interval
        thread = threading.Thread(target=self.run, args=())
        thread.daemon = True
        thread.start()
        
    def run(self):
        settings = fix.SessionSettings(fileName)
        application = Application()
        storeFactory = fix.FileStoreFactory(settings)
        logFactory = fix.ScreenLogFactory(settings)
        acceptor = fix.SocketAcceptor(application, storeFactory, settings, logFactory)
        acceptor.start()
        print("Started acceptor successfully")
        
        # 保持线程存活，不要重复创建Acceptor
        while True:
            time.sleep(self.interval)

额外注意事项 #

• 不要手动设置头部核心字段：BeginString、BodyLength、SenderCompID、TargetCompID、SendingTime这些字段QuickFix会话会自动填充，手动设置会导致BodyLength计算错误或字段冲突。
• 确保ClOrdID唯一：用UUID生成唯一的订单ID，避免重复订单被交易对手拒绝。
• 及时更新数据库状态：发送订单后立刻修改订单状态，防止每次轮询都重复发送同一个订单。

内容的提问来源于stack exchange，提问作者Chris Dagg