Introducing Fxmacrodata Sse Streaming

现在FXMacroData提供实时服务器发送事件 (SSE) 流终点. 一次连接,您的应用程序在任何经济发布的时间都会收到即时推送通知,包括所有18种支持货币. 美国联邦储备 政策决策 欧洲央行 其他主要央行的数据.

终点

SSE流可在以下URL上访问.这是一个直接连接到FXMacroData API服务器的 没有使用 fxmacrodata.com/api/... 对于SSE的路径由于该路线通过CDN层,缓冲响应并打破直播流.

https://api.fxmacrodata.com/v1/stream/events

选择性查询参数允许您缩小源:

• currencies 逗号分隔的列表: usd,eur,gbp
• indicators 逗号分隔的列表: inflation,policy_rate
• api_key 需要专业API密钥用于非美元货币

curl -N "https://api.fxmacrodata.com/v1/stream/events?currencies=usd&indicators=inflation,policy_rate"

活动格式

每个发布的经济版本都会产生一个标准的W3C EventSource框架:

id: usd_inflation_1772109000
event: announcement
data: {"event_id": "usd_inflation_1772109000", "currency": "usd", "indicator": "inflation", "records_written": 1, "timestamp": 1772109002}

SSE消息是一个操作触发器,而不是全数据有效载荷.收到后,请在 API数据文件 查找包括新值在内的完整时间序列.

浏览器示例

在浏览器中,原生 EventSource 接下来的例子订阅了 美元非农业工资 现在我 美元通货膨胀 使用免费未经验证层:

const streamUrl = new URL("https://api.fxmacrodata.com/v1/stream/events");
streamUrl.searchParams.set("currencies", "usd");
streamUrl.searchParams.set("indicators", "non_farm_payrolls,inflation");

const source = new EventSource(streamUrl);

source.addEventListener("announcement", async (event) => {
const payload = JSON.parse(event.data);
console.log("Release received:", payload.currency, payload.indicator);

// Fetch the full record now that we know new data is available
const resp = await fetch(
  `https://api.fxmacrodata.com/v1/announcements/${payload.currency}/${payload.indicator}`
);
const records = await resp.json();
console.log("Latest record:", records[records.length - 1]);
});

source.onerror = (err) => console.error("SSE error", err);

Python 的例子

对于服务器端工作者,使用 requests 通过. Last-Event-ID 在重新连接时,重播任何错过的事件:

import json, time, requests

API_KEY = "YOUR_API_KEY"
STREAM_URL = (
  "https://api.fxmacrodata.com/v1/stream/events"
  "?currencies=usd,eur&indicators=inflation,policy_rate&api_key=" + API_KEY
)

def consume():
  last_id = None
  while True:
      headers = {"Accept": "text/event-stream"}
      if last_id:
          headers["Last-Event-ID"] = last_id
      try:
          with requests.get(STREAM_URL, headers=headers, stream=True, timeout=90) as r:
              r.raise_for_status()
              event = {}
              for line in r.iter_lines(decode_unicode=True):
                  if not line:
                      if event.get("event") == "announcement" and event.get("data"):
                          payload = json.loads(event["data"])
                          last_id = event.get("id") or payload["event_id"]
                          print("New release:", payload)
                      event = {}
                      continue
                  if not line.startswith(":"):
                      field, _, value = line.partition(":")
                      event[field] = value.lstrip()
      except requests.RequestException as e:
          print(f"Disconnected: {e}. Retrying...")
          time.sleep(3)

consume()

计划可用性

接下来要建造什么

SSE流最适合作为现有的 REST 工作流的触发层.实用模式包括:

• 实时仪表板卡更新 在新数据到来时更新一个面板,而不是每N分钟进行一次投票.
• 交易警报 将最新版本与您的门进行比较,并立即发出Slack或webhook通知.
• 缓存无效 使用SSE事件作为重新获取和存储更新的指标记录的信号.
• 结合了 发布日程 知道接下来计划什么,并使用SSE作为实时确认,

查看一个完整的步骤执行步骤, 查阅 如何使用SSE流媒体指南现在我们要做什么?

汇率数据团队