如何用Python实现港美股实时套利监控（ADR-H股套利）

很多中国公司在美国（ADR）和香港（H股）两地上市，例如阿里巴巴、京东、百度等。由于两个市场的流动性、投资者结构和交易时间差异，这些资产在两个市场上常常会有价格差异。

我们可以：

• 做空高估的一方
• 做多低估的一方
• 等待价差回归均值，获得无风险利润

这种方式被称为跨市场价差套利（Statistical Arbitrage）。

##价差怎么计算？
以阿里巴巴为例：

• 阿里ADR代码：BABA（美股）
• 阿里港股代码：9988.HK
• 股比关系：1 ADR = 8 股 H股
• 汇率（港币兑美元）：假设为 7.82

计算公式：

H股折算成ADR价格 = 港股价格 × 8 / 7.82
价差 = ADR价格 - 折算后的H股价格

当这个价差超过一定阈值时，就可以考虑做套利操作。

###实战代码
我们通过Infoway API的WebSocket接口订阅实时K线。

实现目标：

• 实时订阅 BABA（ADR）和 9988.HK（港股）1分钟K线
• 计算价格差异（换算后）
• 输出每次更新的价差信息

###步骤一：准备依赖库

pip install websockets asyncio

###步骤二：完整代码实现

import asyncio
import json
import websockets

# 先在官网申请API Key: www.infoway.io
API_KEY = "yourApiKey"

# WebSocket URL
WS_URL = f"wss://data.infoway.io/ws?business=stock&apikey={API_KEY}"

# 股票代码
ADR_SYMBOL = "BABA"
HK_SYMBOL = "9988.HK"

# 汇率与换股比例（真实环境建议用实时汇率）
CONVERSION_RATIO = 8      # 1 ADR = 8 H股
USDHKD = 7.82              # 港币兑美元

# 当前价格缓存
latest_prices = {
    "ADR": None,
    "HK": None
}

# 计算并输出价差
def compute_spread():
    if latest_prices["ADR"] and latest_prices["HK"]:
        hk_equiv = latest_prices["HK"] * CONVERSION_RATIO / USDHKD
        spread = latest_prices["ADR"] - hk_equiv
        print(f"\n[📈 实时套利监控]")
        print(f"ADR: ${latest_prices['ADR']:.2f}")
        print(f"H股折算: ${hk_equiv:.2f}")
        print(f"价差: ${spread:.2f} USD")
        return spread

async def connect_and_receive():
    async with websockets.connect(WS_URL) as websocket:
        # 发送订阅请求
        init_message = {
            "code": 10004,  # K线订阅
            "trace": "adr_hk_arbitrage",
            "data": {
                "arr": [
                    {"type": 1, "codes": ADR_SYMBOL},   # 1分钟K线
                    {"type": 1, "codes": HK_SYMBOL}
                ]
            }
        }
        await websocket.send(json.dumps(init_message))

        # 定时发送 ping 保持连接
        async def send_ping():
            while True:
                await asyncio.sleep(30)
                ping_message = {"code": 10010, "trace": "adr_hk_arbitrage"}
                await websocket.send(json.dumps(ping_message))

        ping_task = asyncio.create_task(send_ping())

        try:
            while True:
                message = await websocket.recv()
                data = json.loads(message)

                # 处理返回的K线数据
                if data.get("code") == 10004 and "data" in data:
                    for item in data["data"]:
                        code = item["code"]
                        kline = item.get("kline", [])
                        if kline:
                            close_price = float(kline[-1][4])  # 收盘价
                            if code == ADR_SYMBOL:
                                latest_prices["ADR"] = close_price
                            elif code == HK_SYMBOL:
                                latest_prices["HK"] = close_price
                            compute_spread()

        except websockets.exceptions.ConnectionClosedOK:
            print("✅ 连接正常关闭")
        finally:
            ping_task.cancel()

# 运行主程序
asyncio.run(connect_and_receive())

###示例输出：

[📈 实时套利监控]
ADR: $80.50
H股折算: $78.20
价差: $2.30 USD

本作品采用《CC 协议》，转载必须注明作者和本文链接