对于开发者而言，最痛苦的不是写不出策略，而是受限于基础设施的性能。如果你还在用 requests 轮询接口获取股票价格，那你基本上已经告别实时性要求较高的金融场景了。

今天我们就从工程化的角度，聊聊如何用 Python 优雅地解决港股实时行情的接入问题。

痛点分析：HTTP vs WebSocket 在传统的 Web 开发中，我们习惯了无状态的 HTTP。但在金融数据领域，高频的握手开销是不可接受的。我们需要全双工通信，Server 端有数据变动直接 Push 给 Client。

技术选型与环境依赖 我们追求的是极致的轻量化。Python 3.9+ 配合 websocket-client 是目前性价比最高的方案。它足够底层，让你能控制每一个字节的流向，又不需要像 asyncio 那样处理复杂的时间循环（当然，如果你需要极高并发，后期可以重构）。

pip install websocket-client

核心代码实现 不管是你是对接交易所直连，还是使用像 AllTick 这样集成的三方 API，核心范式都是一样的：定义 on_message、on_open 等回调函数。

下面的代码片段展示了如何建立一个持久化的 WebSocket 连接。注意看，我们在 on_open 阶段发送了 JSON 格式的订阅 payload，这是目前主流金融 API 的标准交互方式。

import websocket
import json

url = "wss://api.alltick.co/realtime/hk"

def on_message(ws, message):
    data = json.loads(message)
    # 打印最新成交价和涨跌情况
    print(f"{data['symbol']} 最新价格: {data['last_price']} 涨跌: {data['change']}")

def on_open(ws):
    # 订阅恒生指数及指定股票行情
    ws.send(json.dumps({
        "action": "subscribe",
        "symbols": ["HSI", "00700.HK"]
    }))

ws = websocket.WebSocketApp(url, on_message=on_message, on_open=on_open)
ws.run_forever()

数据流的下游处理 原始数据通常是 JSON 字符串，直接解析的开销很小。在生产环境中，我建议你拿到数据后不要直接 print，而是通过消息队列（如 Kafka）或者直接落库。但为了演示方便，我们这里直接用 Pandas 做一个简单的内存化清洗。

import pandas as pd

# 假设我们有一个行情列表
ticks = [
    {"time": "09:30:01", "price": 500, "volume": 100},
    {"time": "09:30:02", "price": 502, "volume": 50},
    {"time": "09:30:03", "price": 501, "volume": 80},
]

df = pd.DataFrame(ticks)
df['time'] = pd.to_datetime(df['time'])
print(df)

经验总结 通过这种方式，我们将数据的获取延迟从“秒级”压缩到了“毫秒级”。在处理港股这种波动剧烈的市场时，这种技术架构的升级，能让你的程序在起跑线上就领先别人一个身位。