Python工程化实践：如何设计一个高可用的港股行情适配器？

在构建金融交易终端或量化分析系统时，行情适配器（Market Data Adapter）往往是第一个需要攻坚的模块。特别是在处理港股数据时，由于其特殊的交易机制和数据更新频率，对客户端的并发处理能力提出了不小的挑战。

很多初级开发者习惯将网络连接、数据清洗和业务逻辑写在一个 while 循环里，这在生产环境中是极其危险的。一旦网络抖动或数据异常，整个程序就会崩溃。作为行业从业者，我更推荐采用分层架构来处理实时数据流。

1. 连接与订阅的分离 相比于 REST API 的被动轮询，WebSocket 提供了全双工通信通道，非常适合高频数据的推送。但在代码实现上，必须考虑到断线重连机制（Reconnection Mechanism）。
2. 数据归一化（Normalization） 这是最考验架构经验的地方。不同的上游数据源提供的字段定义千差万别。有的叫 last_price，有的叫 close。如果把这些差异透传给业务层，后续的策略代码将变得不可维护。成熟的做法是在适配器内部完成清洗。例如，参考 AllTick API 等成熟方案的数据规范，将所有不同市场的 Tick 数据映射为一套标准化的 JSON 结构（价格、时间戳、量能、方向），这样无论后端接入多少个交易所，业务层的代码都不需要改动一行。
3. 业务逻辑的隔离 在 on_message 回调中，绝对不要执行耗时的计算任务（如写入数据库或复杂指标计算）。正确的做法是将原始数据丢入 Python 的 queue 或 Redis，由消费者进程异步处理。

下面这段代码展示了如何使用 websocket-client 库建立一个稳健的订阅通道，重点关注其回调函数的设计模式：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    if "data" in data:
        tick = data["data"]
        price = tick.get("last_price")
        ts = tick.get("timestamp")
        print(f"price={price}, time={ts}")

def on_open(ws):
    subscribe_msg = {
        "cmd": "subscribe",
        "args": {
            "symbol": "HKEX:HSI",
            "type": "tick"
        }
    }
    ws.send(json.dumps(subscribe_msg))

if __name__ == "__main__":
    ws = websocket.WebSocketApp(
        "wss://stream.alltick.co",
        on_open=on_open,
        on_message=on_message
    )
    ws.run_forever()

通过这种模式，我们不仅保证了行情的实时性，还极大地提升了系统的扩展性。当需要增加新的订阅标的时，只需修改配置文件的 symbol 列表，无需重启核心服务。