使用 Python QuickFIX 通过 Stunnel 连接 FIX 服务器

聖光之護

发布时间：2025-09-30 20:22:01

1009人浏览过

来源于php中文网

原创

使用 python quickfix 通过 stunnel 连接 fix 服务器

本文档旨在指导开发者如何使用 Python QuickFIX 库通过 Stunnel 建立安全的 FIX (Financial Information eXchange) 连接。我们将详细介绍 Stunnel 的配置、QuickFIX 的设置，以及如何调试可能出现的问题，确保 FIX 消息能够安全可靠地传输。

Stunnel 配置

Stunnel 是一个通用的 TLS/SSL 加密隧道程序，可以用于为不支持 SSL 的客户端和服务端提供安全的加密连接。以下是一个 Stunnel 客户端配置文件的示例，用于连接到 FIX 服务器：

CAfile=./Roots.crt
cert=./cert.pfx
client=yes
verify=1
sslVersion = TLSv1.2
options = NO_SSLv2
options = NO_SSLv3
debug=7
foreground=yes
output=./stunnel.log
[SESSION]
accept=8080
connect=EXAMPLE_URL:EXAMPLE_IP

配置参数说明：

CAfile: CA 证书文件路径，用于验证服务器证书。
cert: 客户端证书和私钥文件路径。
client: 设置为 yes 表示 Stunnel 作为客户端运行。
verify: 设置为 1 启用证书验证。
sslVersion: 指定 TLS 版本。建议使用 TLSv1.2 或更高版本。
options: 禁用不安全的 SSL 协议版本。
debug: 设置调试级别。7 为最高级别，会输出详细的日志信息。
foreground: 使 Stunnel 在前台运行，方便查看日志。
output: 指定日志文件路径。
accept: Stunnel 监听的本地端口。
connect: FIX 服务器的地址和端口。

注意事项：

立即学习“Python免费学习笔记（深入）”；

确保证书文件 (Roots.crt, cert.pfx) 的权限设置正确，避免权限过大导致安全风险。
使用 checkHost 或 checkIP 选项来限制信任的证书，防止中间人攻击。

QuickFIX 配置

QuickFIX 是一个开源的 FIX 协议引擎，提供多种编程语言的 API。以下是一个 QuickFIX Initiator 配置文件的示例：

[DEFAULT]
ConnectionType=initiator
LogonTimeout=10
ReconnectInterval=30
FileLogPath=./Logs/

[SESSION]
BeginString=FIXT.1.1
DefaultApplVerID=9
SenderCompID=EXAMPLE_COMP_SENDER
TargetCompID=EXAMPLE_COMP_TARGET
StartTime=00:00:00
EndTime=00:00:00
HeartBtInt=30
CheckLatency=N
MaxLatency=240
SocketConnectPort=8080
SocketConnectHost=127.0.0.1
UseDataDictionary=Y
TransportDataDictionary=./FIX50SP2.xml
FileStorePath=./Sessions/

配置参数说明：

ConnectionType: 设置为 initiator 表示 QuickFIX 作为 Initiator (客户端) 运行。
LogonTimeout: 登录超时时间，单位为秒。
ReconnectInterval: 重连间隔时间，单位为秒。
FileLogPath: 日志文件路径。
BeginString: FIX 协议版本。FIXT.1.1 表示 FIXT 1.1 协议。
DefaultApplVerID: 应用消息版本。9 对应 FIX 5.0 SP2。
SenderCompID: 发送方机构 ID。
TargetCompID: 目标机构 ID。
StartTime: 会话开始时间。
EndTime: 会话结束时间。
HeartBtInt: 心跳间隔时间，单位为秒。
CheckLatency: 是否检查延迟。
MaxLatency: 最大延迟时间，单位为秒。
SocketConnectPort: 连接的本地 Stunnel 端口 (与 Stunnel 配置中的 accept 端口一致)。
SocketConnectHost: 连接的本地 Stunnel 地址 (通常为 127.0.0.1)。
UseDataDictionary: 是否使用数据字典。
TransportDataDictionary: 数据字典文件路径。
FileStorePath: 会话数据存储路径。

Python 代码示例：

MuleRun

全球首个AI Agent交易平台

下载

import quickfix as fix

class Application(fix.Application):
    def onCreate(self, sessionID):
        print("onCreate : Session (%s)" % sessionID.toString())

    def onLogon(self, sessionID):
        self.sessionID = sessionID
        print("Successful Logon to session '%s'." % sessionID.toString())

    def onLogout(self, sessionID):
        print("onLogout")

    def toAdmin(self, sessionID, message):
        print("toAdmin")

    def fromAdmin(self, sessionID, message):
        print("fromAdmin")

    def toApp(self, sessionID, message):
        print("Recieved the following message: %s" % message.toString())

    def fromApp(self, message, sessionID):
        print("fromApp")

config_file = "./initiator.cfg"
settings = fix.SessionSettings(config_file)
application = Application()
storeFactory = fix.FileStoreFactory(settings)
logFactory = fix.FileLogFactory(settings)
initiator = fix.SocketInitiator(application, storeFactory, settings, logFactory)
initiator.start()

代码说明：

Application 类继承自 fix.Application，用于处理 FIX 消息的回调函数。
SessionSettings 从配置文件中加载会话设置。
FileStoreFactory 和 FileLogFactory 分别用于创建消息存储和日志记录对象。
SocketInitiator 创建一个 Socket Initiator 对象，用于建立 FIX 连接。
initiator.start() 启动 Initiator，开始尝试连接到 FIX 服务器。

调试

如果在连接过程中遇到问题，可以按照以下步骤进行调试：

检查 Stunnel 日志： Stunnel 日志文件 (stunnel.log) 包含了详细的连接信息，可以帮助你诊断 SSL/TLS 连接问题。
检查 QuickFIX 日志： QuickFIX 日志文件包含了 FIX 消息的发送和接收记录，可以帮助你诊断 FIX 协议问题。
设置 Stunnel 调试级别： 将 Stunnel 配置文件的 debug 选项设置为 7 可以输出更详细的日志信息。
使用 Wireshark 抓包： 使用 Wireshark 可以抓取网络数据包，分析 SSL/TLS 握手过程和 FIX 消息内容。

常见问题及解决方法：

Stunnel 连接断开： 检查 Stunnel 配置是否正确，证书是否有效，以及服务器地址和端口是否可达。查看 Stunnel 日志，确认是否有错误信息。
QuickFIX 无法登录： 检查 QuickFIX 配置是否正确，SenderCompID 和 TargetCompID 是否匹配，以及数据字典文件是否正确。查看 QuickFIX 日志，确认是否有登录失败的错误信息。
心跳超时： 检查 QuickFIX 配置中的 HeartBtInt 参数是否合理，以及网络连接是否稳定。

发送心跳请求

QuickFIX 会自动处理心跳消息。如果需要手动发送心跳请求，可以在 Application 类中添加以下代码：

import time

def onLogon(self, sessionID):
    self.sessionID = sessionID
    print("Successful Logon to session '%s'." % sessionID.toString())
    # 定时发送心跳请求
    while True:
        time.sleep(self.heartbeat_interval)
        try:
            fix.Session.sendHeartbeat(sessionID)
            print("Heartbeat sent")
        except fix.SessionNotFound:
            print("Session not found, stopping heartbeat")
            break

注意事项：

立即学习“Python免费学习笔记（深入）”；