外汇行情还在更新吗？用 Python 订阅 EURUSD 并记录双时间
TickDB Research ·
适合正在开发外汇报价监控、需要确认脚本持续收到行情的 Python 开发者。使用 TickDB 的 WebSocket 行情订阅，记录 EURUSD 报价、数据时间和本机接收时间，生成原始消息与 CSV。通过目标报价数量、接收间隔和数据时间推进情况，区分连接仍在、消息到达与数据更新。本次30秒订阅观察收到38条目标报价；代码支持有限运行和自动保存结果，方便在自己的环境复查。
先把“还在更新”变成一份记录
你正在做一个外汇监控页面，EURUSD价格已经显示出来，终端也没报错。几秒后价格不动了：市场报价没变，还是脚本已经收不到消息？如果只留屏幕上最后一个价格，排查时很难还原发生了什么。
做法是给每条目标报价留下价格、数据时间和接收时间，再检查观察期内消息是否继续到达、数据时间是否推进。TickDB在这条流程中提供持续的外汇行情推送，Python脚本负责筛选EURUSD、记录本机时间并保存文件。这样能把页面上一个静止的数字，追溯到具体的接收记录。
TickDB在这里提供什么
TickDB为开发者提供统一的市场数据接口。本篇使用其中的WebSocket持续订阅：连接一次，订阅EURUSD的ticker频道，接收后续报价消息。一次性的REST或MCP查询可以取得快照；需要观察连续更新时，应使用对应的持续订阅通道。
官方ticker文档列出EURUSD消息中的last_price和毫秒级Unix时间戳timestamp，以及按数据可用性返回的买卖报价。数据时间来自消息，本机接收时间由脚本在收到消息后立即记录；文件保存和后面的判断由应用代码实现。
这次实际收到了什么
2026年10月28日，脚本在UTC 02:10:30至02:11:02之间完成连接和观察；连接后设置30秒接收窗口，收到38条EURUSD报价。相邻目标消息的数据时间变化了37次，没有观察到时间倒退。以下为前三条记录：
| 数据时间 UTC | 接收时间 UTC | last_price |
|---|---|---|
| 2026-10-28T02:10:33.305+00:00 | 2026-10-28T02:10:33.405+00:00 | 1.11242 |
| 2026-10-28T02:10:33.706+00:00 | 2026-10-28T02:10:33.816+00:00 | 1.11245 |
| 2026-10-28T02:10:33.803+00:00 | 2026-10-28T02:10:33.905+00:00 | 1.11240 |
38条是这次短窗口的实际计数，不是固定推送频率。再次运行时，记录数和价格会变化。
两个时间分别怎么用
| 保存字段 | 来源 | 用途 |
|---|---|---|
last_price | TickDB目标报价消息 | 保留当条消息的价格，不先转成浮点数 |
data_time_ms、data_time_utc | 消息timestamp及其UTC显示 | 检查数据时间是否继续变化或出现倒退 |
received_time_ms、received_time_utc | 本机收到消息后的系统时钟 | 回查脚本何时读到这条消息 |
receive_gap_seconds | 本机单调时钟 | 观察相邻目标报价的接收间隔，不受系统时钟校正直接影响 |
两个UTC时间相减可能包含源端时间定义、传输、缓冲、调度和本机时钟差异，不能直接当成网络延迟。这里只记录它们，不生成延迟指标。
运行代码，拿到自己的CSV
需要Python 3.11及以上版本。安装依赖：
python3 -m pip install websockets certifi
将下面代码保存为eurusd_monitor.py。运行时隐藏输入自己的TickDB API Key；已有TICKDB_API_KEY环境变量也可以直接读取。不要把Key写进代码或截图，账户是否具备目标标的权限以实际账户为准。
python3 eurusd_monitor.py --seconds 30 --out eurusd_run_01
输出目录必须尚不存在，以免覆盖上次记录。完整代码：
"""Python 3.11+; pip install websockets certifi. Bounded observation, not trading."""
import argparse
import asyncio
import csv
import getpass
import json
import os
import ssl
import time
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlencode
import certifi
import websockets
def utc(ms):
return datetime.fromtimestamp(ms / 1000, timezone.utc).isoformat(timespec='milliseconds')
async def observe(seconds, out):
key = os.environ.get('TICKDB_API_KEY') or getpass.getpass('TickDB API key (hidden): ')
if not key.strip():
raise ValueError('API key required')
out.mkdir(parents=True, exist_ok=False)
fields = ['symbol', 'last_price', 'bid_price', 'ask_price', 'data_time_ms',
'data_time_utc', 'received_time_ms', 'received_time_utc', 'receive_gap_seconds']
summary = {'started_utc': utc(time.time_ns() // 1000000), 'requested_seconds': seconds,
'symbol': 'EURUSD', 'status': 'starting', 'quote_count': 0,
'data_time_changes': 0, 'backward_data_times': 0, 'last_receive_age_seconds': None,
'max_receive_gap_seconds': None, 'handshake_http_status': None}
previous_mono = previous_data = last_receive_mono = None
gaps = []
try:
uri = 'wss://api.tickdb.ai/v1/realtime?' + urlencode({'api_key': key})
context = ssl.create_default_context(cafile=certifi.where())
async with websockets.connect(uri, ssl=context, open_timeout=10, close_timeout=3,
ping_interval=20, ping_timeout=20) as ws:
summary['status'] = 'connected'
summary['handshake_http_status'] = 101
await ws.send(json.dumps({'cmd': 'subscribe', 'data': {
'channel': 'ticker', 'symbols': ['EURUSD'], 'type': 'forex'}}))
deadline = time.monotonic() + seconds
with (out / 'messages.jsonl').open('w') as raw, (out / 'quotes.csv').open('w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=fields)
writer.writeheader()
while time.monotonic() < deadline:
try:
message = await asyncio.wait_for(ws.recv(), timeout=min(1, max(.001, deadline-time.monotonic())))
except asyncio.TimeoutError:
continue
received_ms = time.time_ns() // 1000000
received_mono = time.monotonic()
try:
decoded = json.loads(message)
except (ValueError, UnicodeDecodeError):
summary['invalid_json_count'] = summary.get('invalid_json_count', 0) + 1
continue
raw.write(json.dumps({'received_time_ms': received_ms, 'message': decoded}).replace(key, '[REDACTED]')+'\n')
raw.flush()
for item in decoded if isinstance(decoded, list) else [decoded]:
if not isinstance(item, dict):
continue
if item.get('cmd') == 'error':
summary['api_error_code'] = item.get('code')
raise RuntimeError('API subscription error')
data = item.get('data')
if item.get('cmd') != 'ticker' or not isinstance(data, dict) or data.get('symbol') != 'EURUSD':
continue
stamp = data.get('timestamp')
if isinstance(stamp, bool) or not isinstance(stamp, int) or not isinstance(data.get('last_price'), str):
summary['invalid_quote_count'] = summary.get('invalid_quote_count', 0) + 1
continue
gap = None if previous_mono is None else round(received_mono-previous_mono, 6)
if gap is not None:
gaps.append(gap)
if previous_data is not None:
summary['data_time_changes'] += int(stamp != previous_data)
summary['backward_data_times'] += int(stamp < previous_data)
row = dict(symbol='EURUSD', last_price=data['last_price'],
bid_price=data.get('bid_price'), ask_price=data.get('ask_price'),
data_time_ms=stamp, data_time_utc=utc(stamp),
received_time_ms=received_ms, received_time_utc=utc(received_ms),
receive_gap_seconds=gap)
writer.writerow(row)
f.flush()
summary['quote_count'] += 1
previous_mono = last_receive_mono = received_mono
previous_data = stamp
summary['status'] = 'observed_quotes' if summary['quote_count'] else 'connected_without_target_quotes'
except Exception as exc:
summary['status'] = 'failed'
summary['error_type'] = type(exc).__name__
response = getattr(exc, 'response', None)
summary['handshake_http_status'] = getattr(response, 'status_code', None)
# Never print exception text: it may contain the authenticated URL.
finally:
summary['ended_utc'] = utc(time.time_ns() // 1000000)
summary['last_receive_age_seconds'] = None if last_receive_mono is None else round(time.monotonic()-last_receive_mono, 6)
summary['max_receive_gap_seconds'] = max(gaps) if gaps else None
(out / 'summary.json').write_text(json.dumps(summary, indent=2)+'\n')
print(json.dumps(summary, indent=2))
return summary['status'] == 'observed_quotes'
if __name__ == '__main__':
p = argparse.ArgumentParser()
p.add_argument('--seconds', type=int, default=30)
p.add_argument('--out', type=Path, default=Path('eurusd_observation'))
a = p.parse_args()
if not 1 <= a.seconds <= 300:
p.error('--seconds must be 1..300')
raise SystemExit(0 if asyncio.run(observe(a.seconds, a.out)) else 1)
程序最多接收指定时长，然后关闭连接。它使用WebSocket协议层的Ping/Pong检查连接；没有沿用旧示例中每秒发送应用ping的方式。当前官方连接建议也区分了协议层心跳与附加应用心跳。
运行后有三份文件：messages.jsonl保留带接收时间的消息，quotes.csv保存有效EURUSD报价，summary.json保存计数、状态和运行时间。握手失败时可能只生成摘要；查看status和错误类型，不要把没有CSV当成“收到零条正常行情”。脚本不会输出带Key的连接URL。
用结果判断下一步
observed_quotes表示观察期内确实读到了目标报价。结合quote_count、data_time_changes和CSV看更新是否持续出现；消息时间变化而价格不变仍可能是有效更新，不必要求每次价格都变化。
connected_without_target_quotes表示连接建立了，但窗口内没有有效EURUSD报价。先确认订阅对象、权限和原始控制消息，再考虑该时段是否有报价活动。连接心跳正常只帮助判断连接，不能替代目标报价更新的检查。
failed表示连接、订阅或接收过程出现错误。若有部分报价保存，仍应保留失败状态并定位中断。握手权限问题先处理权限；不要靠无限重试掩盖错误。
短窗口内连续收到消息，也不能据此确定中间没有遗漏。正式告警还需要结合自己的场景设置静默阈值、检查时间推进、处理重连和重新订阅；本脚本提供的是排查入口，不承接自动恢复或交易执行。
先到TickDB获取自己的接入凭据，用这段代码保存一份EURUSD记录，再检查CSV中最后几条数据时间是否推进、接收时间是否接近观察结束。把这一步接到你的监控页面，页面最后价格就有了可回查的来源。