外汇三角套利：EUR/USD、GBP/USD、EUR/GBP 的实时套利监控
阅读时间：约 15 分钟 | 技术门槛：中高级 | 代码复杂度：⭐⭐⭐⭐
“在外汇市场，三角套利是极少数能在理论上消除风险、让数学站在你这边的策略。但正是这种'无风险'的光环，让它成为散户最难复现的机构游戏。”
大多数交易者对三角套利的认知停留在教科书层面：EUR/USD、GBP/USD、EUR/GBP 三个货币对之间存在一个数学恒等式，当实际汇率偏离这个恒等式时，理论上可以买入被低估的、卖出被高估的，实现无风险利润。
问题在于：教科书没有告诉你的是——这个"无风险"的前提是毫秒级的执行速度和无限流动性的市场。当你在散户账户上按下回车键时，机构算法已经完成了 200 次这样的操作。
本文拆解三角套利的数学本质，给出生产级的实时监控代码，让你至少能先"看见"套利窗口的开启与关闭——至于能否在它关闭前入场，那是另一个问题。
一、三角套利的数学原理
1.1 什么是三角套利
三角套利（Triangular Arbitrage）的核心逻辑藏在一个简单的恒等式里：
EUR/USD × GBP/USD = EUR/GBP
这个等式意味着：先用美元买入欧元，再用欧元买入英镑，最后把英镑换回美元。如果汇率完全精确，这个闭环不会产生任何利润——1 美元无论怎么周转，最终还是 1 美元。
但市场不是数学。市场报价由无数参与者共同决定，存在微小的时间差和信息差。因此：
EUR/USD × GBP/USD ≠ EUR/GBP
这个偏离就是套利机会。
1.2 套利利润的计算
假设三个货币对的报价如下：
| 货币对 | 报价 | 说明 |
|---|---|---|
| EUR/USD | 1.0850 / 1.0852 | 买入价 1.0850，卖出价 1.0852 |
| GBP/USD | 1.2650 / 1.2652 | 买入价 1.2650，卖出价 1.2652 |
| EUR/GBP | 0.8570 / 0.8573 | 买入价 0.8570，卖出价 0.8573 |
三角套利的关键在于买卖方向的组合。我们以 1,000,000 美元为初始资金，演示两种路径：
路径 A：美元 → 欧元 → 英镑 → 美元
1. 卖出 USD，买入 EUR：1,000,000 / 1.0852 = 921,497.98 EUR
2. 卖出 EUR，买入 GBP：921,497.98 × 0.8570 = 789,723.41 GBP
3. 卖出 GBP，买入 USD：789,723.41 × 1.2650 = 999,000.18 USD
路径 B：美元 → 英镑 → 欧元 → 美元
1. 卖出 USD，买入 GBP：1,000,000 / 1.2652 = 790,384.68 GBP
2. 卖出 GBP，买入 EUR：790,384.68 × (1/0.8573) = 922,042.15 EUR
3. 卖出 EUR，买入 USD：922,042.15 × 1.0850 = 1,000,015.72 USD
路径 B 产生了 15.72 美元的无风险利润。
1.3 汇率偏离度指标
在实际监控中，我们用隐含汇率 vs 实际汇率的偏离度来判断套利窗口：
# 三角套利偏离度公式
implied_EUR_GBP = (EUR_USD_bid / GBP_USD_ask) # 隐含 EUR/GBP
deviation = (implied_EUR_GBP - EUR_GBP_bid) / EUR_GBP_bid * 10000 # 单位：bp
# 偏离度 > 交易成本 → 存在套利窗口
# 假设单边交易成本为 1bp，三角总成本约 3bp
# 当 deviation > 3bp 时，理论上存在利润空间
偏离度的单位通常用基点（Basis Point, bp）——1 bp = 0.01%，1 万点等于 100%。
二、为什么散户难以捕捉套利窗口
2.1 窗口的寿命以毫秒计
在外汇这个日均交易量超过 7.5 万亿美元的市场，机构算法每秒扫描数千次价格。它们使用专用的光纤线路、直连交易所的服务器、以及 FPGA 硬件加速——价格偏离的瞬间，它们的订单已经在路上了。
散户面临的是三重障碍：
| 障碍 | 散户 | 机构 |
|---|---|---|
| 信息延迟 | 100-500ms | <1ms |
| 执行延迟 | 500ms-2s | <10ms |
| 交易成本 | 固定点差 + 佣金 | 银行间市场零佣金 |
理论上存在的套利窗口，在散户能够行动之前，往往已经关闭。
2.2 流动性与滑点
教科书假设你可以"无限量"地买卖而不影响价格。现实是：
- 你的订单量越大，市场冲击成本越高
- 非农就业数据发布时段，流动性急剧萎缩
- 银行报价可能在毫秒内撤回
因此，即使检测到套利机会，还需要计算净收益 = 理论利润 - 交易成本 - 滑点 - 延迟损耗。
2.3 散户的最优策略
这不是让你放弃。散户有两种可行的路径：
路径一：做套利窗口的"记录者"而非"参与者"
监控套利窗口的出现频率、持续时间、幅度分布，建立对市场微观结构的深度认知。这本身就是有价值的数据产品。
路径二：关注套利窗口与宏观事件的关联
在特定时段（如央行决议、非农数据发布前后），套利窗口出现的频率和幅度会显著增加。找到这种关联，本身就是 alpha。
无论哪条路，第一步都是构建可靠的价格监控基础设施——这正是本文的核心。
三、实时监控架构设计
3.1 为什么需要 WebSocket 而非轮询
三角套利需要三个货币对的同一时刻价格。轮询模式下：
- 你请求 EUR/USD，得到 10:00:00.100 的价格
- 你请求 GBP/USD，得到 10:00:00.350 的价格
- 你请求 EUR/GBP，得到 10:00:00.600 的价格
三个时间戳不同步，计算出的偏离度存在严重的时间偏差——上一个时刻的 EUR/USD + 当前时刻的 GBP/USD，这已经不是三角套利，而是随机赌博。
WebSocket 推送解决了这个问题：所有价格更新在同一数据流中携带时间戳，你可以缓存最新报价，计算窗口时使用同一时刻的数据。
3.2 系统架构图
┌─────────────────────────────────────────────────────────────────┐
│ TickDB WebSocket 实时数据流 │
└────────────────────┬─────────────────────────────┬───────────────┘
│ │
┌──────────▼──────────┐ ┌───────────▼───────────┐
│ EUR/USD 频道 │ │ GBP/USD 频道 │
└──────────┬──────────┘ └───────────┬───────────┘
│ │
└──────────┬──────────────────┘
│
┌──────────▼──────────┐
│ 价格缓存层（Redis） │
│ - 保留最近 100 条 │
│ - 自动过期：500ms │
└──────────┬──────────┘
│
┌──────────▼──────────┐
│ 套利计算引擎 │
│ - 偏离度计算 │
│ - 交易成本估算 │
│ - 窗口识别 │
└──────────┬──────────┘
│
┌──────────▼──────────┐
│ 告警层 │
│ - 飞书 WebHook │
│ - Slack │
│ - 日志持久化 │
└─────────────────────┘
3.3 套利窗口识别算法
class TriangularArbitrageDetector:
"""
三角套利窗口检测器
检测逻辑：
1. 维护三个货币对的最新买卖价
2. 计算两种路径的隐含利润
3. 当利润 > 交易成本时，触发告警
"""
def __init__(self, cost_bp=3.0):
self.prices = {
'EUR_USD': {'bid': None, 'ask': None, 'timestamp': None},
'GBP_USD': {'bid': None, 'ask': None, 'timestamp': None},
'EUR_GBP': {'bid': None, 'ask': None, 'timestamp': None},
}
self.cost_bp = cost_bp # 三角总交易成本（基点）
self.window_history = []
def update_price(self, pair, bid, ask):
"""更新单个货币对的价格"""
self.prices[pair] = {
'bid': bid,
'ask': ask,
'timestamp': time.time()
}
def check_data_freshness(self):
"""
检查三个货币对的数据是否足够新鲜
超过 500ms 的数据视为过期
"""
if any(p['bid'] is None for p in self.prices.values()):
return False
now = time.time()
max_age = 0.5 # 500ms
for pair, data in self.prices.items():
age = now - data['timestamp']
if age > max_age:
logger.warning(f"{pair} 数据过期: {age*1000:.1f}ms")
return False
return True
def calculate_path_profit(self, initial_usd=1_000_000):
"""
计算两种套利路径的利润
返回:
dict: {
'path_A_profit': 美元金额,
'path_B_profit': 美元金额,
'best_profit': 最大利润,
'deviation_bp': 偏离度基点
}
"""
eur_usd = self.prices['EUR_USD']
gbp_usd = self.prices['GBP_USD']
eur_gbp = self.prices['EUR_GBP']
# 路径 A: USD -> EUR -> GBP -> USD
# 卖 USD 买 EUR（用 ask），卖 EUR 买 GBP（用 bid），卖 GBP 买 USD（用 bid）
step1_A = initial_usd / eur_usd['ask'] # USD -> EUR
step2_A = step1_A * eur_gbp['bid'] # EUR -> GBP
step3_A = step2_A * gbp_usd['bid'] # GBP -> USD
# 路径 B: USD -> GBP -> EUR -> USD
# 卖 USD 买 GBP（用 ask），卖 GBP 买 EUR（用 bid），卖 EUR 买 USD（用 bid）
step1_B = initial_usd / gbp_usd['ask'] # USD -> GBP
step2_B = step1_B / eur_gbp['ask'] # GBP -> EUR (汇率反向)
step3_B = step2_B * eur_usd['bid'] # EUR -> USD
profit_A = step3_A - initial_usd
profit_B = step3_B - initial_usd
# 计算偏离度
implied_eur_gbp = eur_usd['bid'] / gbp_usd['ask']
deviation_bp = (implied_eur_gbp - eur_gbp['bid']) / eur_gbp['bid'] * 10000
return {
'path_A_profit': profit_A,
'path_B_profit': profit_B,
'best_profit': max(profit_A, profit_B),
'deviation_bp': deviation_bp,
'net_profit': max(profit_A, profit_B) - (initial_usd * self.cost_bp / 10000)
}
def detect_window(self, min_profit=10.0):
"""
检测是否存在可交易的套利窗口
参数:
min_profit: 最小利润阈值（美元）
返回:
dict 或 None: 如果存在窗口，返回窗口详情；否则返回 None
"""
if not self.check_data_freshness():
return None
result = self.calculate_path_profit()
if result['net_profit'] >= min_profit:
window = {
'timestamp': time.time(),
'profit': result['best_profit'],
'net_profit': result['net_profit'],
'deviation_bp': result['deviation_bp'],
'prices': {k: {'bid': v['bid'], 'ask': v['ask']}
for k, v in self.prices.items()}
}
self.window_history.append(window)
return window
return None
四、生产级 WebSocket 实时监控代码
以下代码实现了完整的三角套利监控系统，包含：
- 三个货币对的并行 WebSocket 连接
- 心跳保活机制
- 指数退避重连
- 限频自适应处理
- 价格缓存与偏离度计算
- 飞书告警集成
#!/usr/bin/env python3
"""
外汇三角套利实时监控工具
支持: EUR/USD, GBP/USD, EUR/GBP
数据源: TickDB WebSocket
告警: 飞书 WebHook
⚠️ 本代码仅用于监控和学习，不构成投资建议
"""
import os
import json
import time
import logging
import asyncio
import websockets
import requests
from datetime import datetime
from typing import Optional, Dict, Any
# ============================================================
# 配置区
# ============================================================
# TickDB API 配置
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
TICKDB_WS_URL = "wss://stream.tickdb.ai/v1/ws"
# 监控的货币对
SYMBOLS = {
"EUR/USD": "EUR_USD.FX", # 注意：需要确认 TickDB 的符号格式
"GBP/USD": "GBP_USD.FX",
"EUR/GBP": "EUR_GBP.FX",
}
# 告警配置
FEISHU_WEBHOOK_URL = os.environ.get("FEISHU_WEBHOOK_URL")
ALERT_THRESHOLD_BP = 2.0 # 偏离度阈值（基点）
MIN_PROFIT_USD = 1.0 # 最小利润阈值（美元）
# 重连配置
MAX_RETRIES = 10
BASE_DELAY = 1
MAX_DELAY = 60
# 日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)
# ============================================================
# 工具函数
# ============================================================
def calculate_triangular_arb(profits: Dict[str, Dict]) -> Dict[str, float]:
"""
计算三角套利利润
参数:
profits: 三个货币对的最新报价 {'symbol': {'bid': float, 'ask': float}}
返回:
包含各路径利润和偏离度的字典
"""
eur_usd = profits.get('EUR/USD', {})
gbp_usd = profits.get('GBP/USD', {})
eur_gbp = profits.get('EUR/GBP', {})
if not all([eur_usd.get('bid'), gbp_usd.get('bid'), eur_gbp.get('bid')]):
return {'valid': False}
initial_usd = 1_000_000
# 路径 A: USD -> EUR -> GBP -> USD
# 买 EUR 用 ask，卖 EUR 用 bid；买 GBP 用 ask，卖 GBP 用 bid
step1_A = initial_usd / eur_usd['ask'] # USD → EUR (用 ask 价格买入)
step2_A = step1_A * eur_gbp['bid'] # EUR → GBP (用 bid 价格卖出)
step3_A = step2_A * gbp_usd['bid'] # GBP → USD (用 bid 价格卖出)
profit_A = step3_A - initial_usd
# 路径 B: USD -> GBP -> EUR -> USD
step1_B = initial_usd / gbp_usd['ask'] # USD → GBP
step2_B = step1_B / eur_gbp['ask'] # GBP → EUR (除以 ask，等效于买入 EUR)
step3_B = step2_B * eur_usd['bid'] # EUR → USD
profit_B = step3_B - initial_usd
# 计算偏离度
# 隐含 EUR/GBP = EUR/USD_bid / GBP/USD_ask
# 如果隐含值 > 实际 EUR/GBP_bid，则存在套利机会
implied_eur_gbp = eur_usd['bid'] / gbp_usd['ask']
deviation_bp = (implied_eur_gbp - eur_gbp['bid']) / eur_gbp['bid'] * 10000
return {
'valid': True,
'path_A_profit': profit_A,
'path_B_profit': profit_B,
'best_profit': max(profit_A, profit_B),
'deviation_bp': deviation_bp,
'best_path': 'A' if profit_A >= profit_B else 'B'
}
def send_feishu_alert(window: Dict[str, Any], profits: Dict[str, Dict]):
"""
发送飞书告警
⚠️ 告警不保证送达，存在网络延迟，不应作为交易依据
"""
if not FEISHU_WEBHOOK_URL:
return
timestamp = datetime.fromtimestamp(window['timestamp']).strftime('%H:%M:%S.%f')[:-3]
message = {
"msg_type": "interactive",
"card": {
"header": {
"title": {"tag": "plain_text", "content": f"🚨 三角套利窗口预警 {timestamp}"},
"template": "red"
},
"elements": [
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": (
f"**偏离度**: {window['deviation_bp']:.2f} bp\n"
f"**理论利润**: ${window['best_profit']:.2f}\n"
f"**最优路径**: 路径 {window['best_path']}"
)
}
},
{"tag": "hr"},
{
"tag": "div",
"text": {
"tag": "lark_md",
"content": (
f"**EUR/USD**: {profits['EUR/USD']['bid']:.5f} / {profits['EUR/USD']['ask']:.5f}\n"
f"**GBP/USD**: {profits['GBP/USD']['bid']:.5f} / {profits['GBP/USD']['ask']:.5f}\n"
f"**EUR/GBP**: {profits['EUR/GBP']['bid']:.5f} / {profits['EUR/GBP']['ask']:.5f}"
)
}
},
{"tag": "hr"},
{
"tag": "note",
"elements": [{"tag": "plain_text", "content": "⚠️ 本告警仅供监控研究，不构成交易建议"}]
}
]
}
}
try:
response = requests.post(
FEISHU_WEBHOOK_URL,
headers={"Content-Type": "application/json"},
data=json.dumps(message),
timeout=5
)
if response.status_code != 200:
logger.warning(f"飞书告警发送失败: {response.status_code}")
except requests.RequestException as e:
logger.warning(f"飞书告警发送异常: {e}")
# ============================================================
# WebSocket 客户端
# ============================================================
async def subscribe_symbol(websocket, symbol: str):
"""
订阅单个货币对的实时报价
订阅消息格式参考 TickDB WebSocket 文档
"""
subscribe_msg = {
"cmd": "subscribe",
"params": {
"symbol": symbol,
"channels": ["ticker"] # ticker 频道包含最新买卖价
}
}
await websocket.send(json.dumps(subscribe_msg))
logger.info(f"已订阅: {symbol}")
async def keepalive(websocket, interval=30):
"""
WebSocket 心跳保活
防止连接因空闲被服务器断开
"""
while True:
try:
await asyncio.sleep(interval)
await websocket.send(json.dumps({"cmd": "ping"}))
logger.debug("心跳保活已发送")
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"心跳异常: {e}")
break
async def receive_messages(websocket, profits: Dict, last_update: Dict):
"""
接收并处理 WebSocket 消息
"""
async for message in websocket:
try:
data = json.loads(message)
# 处理心跳响应
if data.get('type') == 'pong':
logger.debug("心跳响应正常")
continue
# 处理 ticker 数据
if 'data' in data:
for item in data['data']:
symbol = item.get('symbol', '')
# 格式化货币对名称
if 'EUR' in symbol and 'USD' in symbol and 'GBP' not in symbol:
pair_name = 'EUR/USD'
elif 'GBP' in symbol and 'USD' in symbol:
pair_name = 'GBP/USD'
elif 'EUR' in symbol and 'GBP' in symbol:
pair_name = 'EUR/GBP'
else:
continue
# 更新价格缓存
profits[pair_name] = {
'bid': float(item.get('bid', 0)),
'ask': float(item.get('ask', 0)),
}
last_update[pair_name] = time.time()
logger.debug(f"更新 {pair_name}: {item.get('bid')} / {item.get('ask')}")
except json.JSONDecodeError:
logger.warning(f"收到无效 JSON: {message[:100]}")
except Exception as e:
logger.error(f"消息处理异常: {e}")
async def monitor_loop(profits: Dict, last_update: Dict, alert_cooldown: float = 60):
"""
主监控循环：定时检查套利窗口
参数:
alert_cooldown: 告警冷却时间（秒），防止告警轰炸
"""
last_alert_time = 0
while True:
await asyncio.sleep(0.5) # 每 500ms 检查一次
# 检查数据新鲜度
now = time.time()
stale_pairs = []
for pair, timestamp in last_update.items():
if now - timestamp > 2.0: # 超过 2 秒未更新视为过期
stale_pairs.append(pair)
if stale_pairs:
logger.debug(f"数据过期: {stale_pairs}")
continue
# 检查是否所有货币对都有数据
if len(profits) < 3:
continue
# 计算三角套利
result = calculate_triangular_arb(profits)
if not result['valid']:
continue
# 检查是否触发告警条件
if result['deviation_bp'] >= ALERT_THRESHOLD_BP or result['best_profit'] >= MIN_PROFIT_USD:
# 检查告警冷却
if now - last_alert_time < alert_cooldown:
continue
window = {
'timestamp': now,
'deviation_bp': result['deviation_bp'],
'best_profit': result['best_profit'],
'best_path': result['best_path']
}
logger.warning(
f"套利窗口! 偏离度={result['deviation_bp']:.2f}bp, "
f"利润=${result['best_profit']:.2f}, 路径={result['best_path']}"
)
# 发送告警
send_feishu_alert(window, profits)
last_alert_time = now
async def connect_with_retry(symbols: list):
"""
WebSocket 连接，支持指数退避重连
重连策略:
- 初始等待 1 秒
- 每次失败后等待时间翻倍
- 最大等待时间 60 秒
- 添加随机抖动避免惊群效应
"""
if not TICKDB_API_KEY:
raise ValueError("请设置环境变量 TICKDB_API_KEY")
uri = f"{TICKDB_WS_URL}?api_key={TICKDB_API_KEY}"
profits = {}
last_update = {}
retry_count = 0
while retry_count < MAX_RETRIES:
try:
logger.info(f"尝试连接 WebSocket (第 {retry_count + 1} 次)")
async with websockets.connect(
uri,
ping_interval=30,
ping_timeout=10,
close_timeout=10
) as websocket:
logger.info("WebSocket 连接成功")
retry_count = 0 # 连接成功后重置计数
# 订阅所有货币对
for symbol in symbols:
await subscribe_symbol(websocket, symbol)
# 并行运行心跳、接收消息、监控循环
await asyncio.gather(
keepalive(websocket),
receive_messages(websocket, profits, last_update),
monitor_loop(profits, last_update),
return_exceptions=True
)
except websockets.exceptions.ConnectionClosed as e:
logger.warning(f"连接关闭: {e}")
except asyncio.CancelledError:
logger.info("任务取消，退出连接")
break
except Exception as e:
logger.error(f"连接异常: {e}")
# 计算退避时间
retry_count += 1
delay = min(BASE_DELAY * (2 ** (retry_count - 1)), MAX_DELAY)
jitter = delay * 0.1 * (0.5 - (retry_count % 2)) # 添加抖动
total_delay = delay + jitter
logger.info(f"{total_delay:.1f} 秒后重连...")
await asyncio.sleep(total_delay)
logger.error("达到最大重试次数，退出程序")
async def main():
"""主函数"""
logger.info("=" * 50)
logger.info("外汇三角套利实时监控系统启动")
logger.info(f"监控货币对: {list(SYMBOLS.keys())}")
logger.info(f"告警阈值: {ALERT_THRESHOLD_BP} bp / ${MIN_PROFIT_USD}")
logger.info("=" * 50)
# ⚠️ 注意: 实际符号需要根据 TickDB 的数据格式调整
# 以下为示意，请登录 tickdb.ai 确认具体符号格式
symbols = list(SYMBOLS.values())
try:
await connect_with_retry(symbols)
except KeyboardInterrupt:
logger.info("收到中断信号，正在关闭...")
if __name__ == "__main__":
# ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio
# 本代码使用 websockets 库，适合中小规模监控
asyncio.run(main())
4.1 环境配置与依赖
# 安装依赖
pip install websockets requests
# 设置环境变量
export TICKDB_API_KEY="your_api_key_here"
export FEISHU_WEBHOOK_URL="https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxx"
# 运行监控
python forex_triangular_arb.py
4.2 代码核心设计说明
| 设计要素 | 实现方式 | 原因 |
|---|---|---|
| 心跳保活 | keepalive() 每 30 秒发送 ping |
防止空闲连接被服务端断开 |
| 指数退避 | delay = min(BASE_DELAY * 2^retry, MAX_DELAY) |
避免频繁重连对服务造成压力 |
| 随机抖动 | jitter = delay * 0.1 * (0.5 - retry%2) |
多个客户端同时重连时分散请求 |
| 数据新鲜度检查 | 超过 2 秒未更新的货币对不参与计算 | 避免用过期价格计算套利 |
| 告警冷却 | 60 秒内不重复告警 | 防止套利窗口持续期间收到轰炸 |
| 异步架构 | asyncio + websockets | 高并发、低资源占用的 IO 密集型任务 |
4.3 关于 TickDB 数据格式的说明
重要提示：本文代码中使用的货币对符号（如 EUR_USD.FX）为示意。TickDB 的实际符号格式需要登录后在控制台确认。获取可用符号列表的 API 调用示例：
import os
import requests
headers = {"X-API-Key": os.environ.get("TICKDB_API_KEY")}
# 获取所有可用交易品种
response = requests.get(
"https://api.tickdb.ai/v1/symbols/available",
headers=headers,
timeout=(3.05, 10)
)
data = response.json()
# 筛选外汇相关品种（符号中包含 FX 或货币代码）
fx_symbols = [s for s in data.get('data', []) if 'FX' in s or 'EUR' in s or 'GBP' in s]
print("可用外汇品种:", fx_symbols)
五、监控数据的深度分析
5.1 偏离度的统计特征
仅靠单个套利窗口的信息无法形成有效策略。以下是你应该在积累足够数据后分析的指标：
| 指标 | 含义 | 应用 |
|---|---|---|
| 偏离度均值 | 正常市场下，三个货币对的平均偏离度 | 判断是否存在系统性偏差 |
| 偏离度标准差 | 偏离度的波动程度 | 设置动态阈值 |
| 窗口持续时间 | 从出现到消失的平均时长 | 判断是否有执行窗口 |
| 窗口频率 | 每小时/每天出现的次数 | 评估策略容量 |
5.2 偏离度分布图（示意）
偏离度分布（假设数据）
频率
200 |▐▌
150 |▐▐▌
100 |▐▐▐▌
50 |▐▐▐▐▌
0 +---+---+---+---+---+---
-5 -2.5 0 2.5 5 偏离度 (bp)
↑ ↑ ↑
负偏离 正常范围 正偏离
(套利空间) (无机会) (套利空间)
在实际数据中，大多数偏离度会集中在 ±1bp 的范围内，这部分被交易成本消耗殆尽。真正的机会出现在尾部区域。
5.3 套利窗口与宏观事件的关联
根据历史经验，以下时段套利窗口出现的频率和幅度会显著增加：
| 时段 | 原因 | 频率增幅 |
|---|---|---|
| 央行利率决议前后 30 分钟 | 货币政策不确定性导致报价分歧 | 3-5x |
| 非农就业数据发布 | 美元短期剧烈波动 | 5-10x |
| 英国退欧公投、美国大选 | 地缘政治风险 | 10x+ |
| 美股开盘/收盘 | 流动性结构变化 | 2-3x |
注意：这些时段虽然机会更多，但流动性风险也更高。限价单可能无法成交，市价单的滑点可能吞噬理论利润。
六、TickDB 在三角套利监控中的定位
6.1 为什么选择 TickDB
外汇三角套利监控的核心需求是：多货币对并行接入 + 实时价格推送 + 低延迟。
| 能力维度 | 传统方案（如 Tushare） | TickDB |
|---|---|---|
| 数据推送方式 | 轮询（最小间隔 1 秒） | WebSocket 推送 |
| 多品种并行 | 需要多个请求，串行获取 | 单连接订阅多个品种 |
| 数据新鲜度 | 上一秒的历史快照 | 实时tick |
| 延迟 | 1-5 秒 | <100ms |
| 连接稳定性 | 无自动重连 | 心跳 + 指数退避重连 |
6.2 功能限制说明
外汇市场在 TickDB 中的定位是高流动性现货市场，适合实时监控和趋势分析。需要注意：
- TickDB 不提供订单簿深度数据（depth），这是外汇市场的技术限制
- 外汇交易存在隔夜利息（swap），本文的三角套利策略假设为日内操作
- 不同经纪商的报价存在差异，实际套利需要考虑经纪商间的报价差
6.3 与其他数据源的协同
如果你的策略需要更高的数据频率或更深的订单簿信息，可以将 TickDB 作为实时监控层，结合其他数据源：
| 数据需求 | 推荐数据源 | TickDB 的角色 |
|---|---|---|
| 高频订单流 | L2 经纪商 API | 价格监控 |
| 历史回测 | TickDB / Broker API | 提供清洗过的历史数据 |
| 事件标记 | 财经日历 API | 事件时间同步 |
七、常见问题与避坑指南
7.1 为什么我看到的机会成交不了？
原因 1：报价延迟
你看到的报价可能是 500ms 前的价格。机构算法用直连线路，可能在 5ms 内完成交易，你看到机会时它已经消失了。
原因 2：流动性枯竭
大单进入市场后，报价会向不利方向移动。模拟盘显示的利润在实盘中会被滑点吞噬。
原因 3：经纪商限制
某些经纪商对套利行为有限制条款，可能冻结账户或拒绝成交。
7.2 如何验证代码逻辑正确性？
步骤 1：用历史数据回测
用 TickDB 的历史 K 线数据验证公式计算是否正确。将历史数据代入 calculate_triangular_arb() 函数，输出每日的理论套利机会，与理论值对比。
步骤 2：用模拟盘验证
先用模拟账户（小资金、限价单）测试 2 周以上，统计"看到的信号"与"实际成交的信号"之间的比例。
步骤 3：对比多个数据源
同时接入两个数据源（如 TickDB + 另一个经纪商 API），对比同一时刻的价格差异。如果差异持续存在，可能是数据源本身的问题。
7.3 代码有哪些工程风险点？
| 风险点 | 表现 | 处理方式 |
|---|---|---|
| 网络抖动导致数据丢失 | 某货币对数据突然中断 | 数据新鲜度检查 + 缓存过期 |
| WebSocket 假连接 | 服务端已断开但客户端未感知 | 心跳 + 超时检测 |
| 告警轰炸 | 套利窗口持续存在时每秒告警 | 冷却时间机制 |
| 数据竞争 | 多线程同时写入价格缓存 | asyncio 单线程协程，无锁 |
八、结语
三角套利是外汇市场中最接近"数学确定"的策略。但正因为它的"无风险"属性，它也是最难被散户复现的机构游戏——不是策略本身有问题，而是执行基础设施的天壤之别。
本文提供的监控代码，能让你至少看见这些窗口的存在。看见，是理解的前提；理解，是改进的开始。
如果你对三角套利的数学细节还有疑问，建议从一手数据开始：用 TickDB 积累 1 周的三个货币对实时价格，用 Python 分析偏离度的统计分布。你的第一手发现，比任何教科书都更有价值。
下一步行动
如果你是外汇新手，想了解货币对间的数学关系：
先从 EUR/USD 的价格形成机制开始，理解"什么是点、什么是手、什么是点差"。
如果你是量化开发者，想验证本文的监控逻辑：
- 访问 tickdb.ai 注册（免费，无需信用卡）
- 在控制台生成 API Key
- 设置环境变量
TICKDB_API_KEY，复制本文代码即可运行 - 观察偏离度的统计分布，验证策略可行性
如果你对订单簿数据有需求：
TickDB 当前支持港股和加密货币的深度数据（depth）。如果你同时关注港股 ETF 的三角套利（如 02800.HK、02801.HK、02828.HK），可以使用 depth 频道获取更完整的流动性视图。联系 [email protected] 获取详情。
如果你想用 AI 辅助开发：
在 Claude、Cursor 等 AI 助手中，搜索安装 tickdb-market-data SKILL，可以用自然语言查询 TickDB 的数据能力。
风险提示：本文不构成任何投资建议。三角套利策略存在执行延迟、流动性风险、交易成本等现实约束，历史机会不代表未来可复制。市场有风险，投资需谨慎。