alpaca-broker-sse-events

Alpaca Broker API — 实时事件（SSE）

Alpaca 通过 服务器发送事件（SSE） 推送经纪业务生命周期事件：这是一个长连接 HTTP GET，会以流的形式传输 text/event-stream。它不是市场数据 WebSocket（alpaca-broker-market-data）——传输方式、身份验证和可靠性模型都不同。

请先阅读 alpaca-broker-integration，了解基础 URL 和身份验证（客户端凭据 Bearer 令牌；旧版 Basic 仍然可用）。

参考资料

• 指南：https://docs.alpaca.markets/docs/sse-events
• 最新文档：alpaca-docs MCP → search "SSE Events"，然后 fetch us/sse-events

1. 为什么用 SSE（以及为什么它比看起来更简单）

SSE 是普通 HTTP。你不需要特殊客户端：发起 GET 请求，保持连接打开，然后逐行读取响应正文即可。每个事件都是包含 JSON 对象的 data: 行。它支持重放——你可以请求某个过去时间点之后的事件，并无缝追赶到实时数据，因此它在生命周期状态跟踪方面远胜于简单轮询。

2. 事件流

流	路径	内容
账户状态	GET /v1/events/accounts/status	账户属性变更：status/crypto_status（SUBMITTED→ACTIVE、ACTION_REQUIRED、REJECTED），以及 kyc_results、account_blocked、trading_blocked、cash_interest、options
日记账状态	GET /v2/events/journals/status	JNLC/JNLS 生命周期（queued→executed、correct……）
入金/转账状态	GET /v2/events/funding/status	统一流：Transfer、BankRelationship、WireBank、FundingWallet 实体（根据 entity_type 切换处理）
交易更新	GET /v2/events/trades	event 字段中的订单事件：new、fill、partial_fill、canceled、rejected、held、trade_bust、trade_correct……（比订单 status 更丰富）
活动	GET /v2beta1/events/activities	一个流中包含每项金融活动：成交、公司行动、费用、日记账、转账。activity_type（TRD = 交易成交）+ activity_subtype + details 对象；ULID event_id、UUID ref_id
非交易活动（旧版）	GET /v1/events/nta	股息、利息、费用、拆股、ACAT、现金支出。entry_type，例如 JNLC/FEE/INT/DIVNRA/CSD；status ∈ executed/correct/canceled

路径和版本并不统一——请逐一核实。 Alpaca 文档对这种跨流不一致的说明恰恰不够充分：

• 对于 2026-02-11 之后入账的活动，Activity SSE 完全取代 /v1/events/nta 和 GET /v1/accounts/activities；较早的历史记录仍需使用 REST 端点。新合作伙伴不应使用旧版端点。它不会取代 /v2/events/trades：成交、更正和冲销会同时出现在两者中，但非成交生命周期事件（accepted、canceled、expired、replaced）不属于活动，因此不会出现在 Activity SSE 中。
• /v1/events/trades 已移除（已完全弃用，不再可用）。/v1/events/journals/status 是旧版端点——请迁移至 /v2。
• /v1 事件同时包含整数 event_id 和 ULID event_ulid；/v2//v2beta1 事件的 event_id 为 ULID。§4 会说明要传入哪个游标。

不同流的字段结构各不相同——不存在通用信封格式。请检查每个流的架构。

3. 连接

GET /v2/events/journals/status?since_id=<last-ulid-you-saw> HTTP/1.1
Host: broker-api.alpaca.markets
Authorization: Bearer <access-token>
Accept: text/event-stream

Authorization: Basic <base64(key:secret)> 也可用。没有文档说明已打开的流能否在令牌过期后继续工作，因此应将流中途出现的 401 视为普通断连，并使用游标重新连接。

读取响应流，并在 data: {…} 帧到达时解析它们。在大多数语言中，现成的 EventSource/SSE 客户端都能使用——只需确保它支持在初始请求中设置 Authorization 标头（浏览器的 EventSource API 众所周知不支持；请改用服务端 SSE 库）。

4. 重放游标——防止数据丢失的功能

每个流都支持按时间点重放：

参数	含义
since / until	RFC3339 或 YYYY-MM-DD 时间戳（某些 /v1 流仅支持日期）。偏移量中的 + 必须 URL 编码为 %2B。
since_id / until_id	v2 / v2beta1 流——ULID 值。 在 /v1/events/* 中，这些是旧版整数游标：仅供选定的经纪商合作伙伴使用，并将于 2027-02-15 停用。不要基于它们构建功能。
since_ulid / until_ulid	v1 流（accounts/status、nta 和旧版 /v1 journals 流）——ULID 游标，所有合作伙伴均可使用。v1 流请使用这些游标；v2 中不存在这些游标。

规则：设置 until 时必须设置 since；设置 until_id 时必须设置 since_id（since_ulid / until_ulid 也遵循相同规则）；不能混用 since、since_id 和 since_ulid。在 /v2beta1/events/activities 中，依赖关系相反：设置 since 时必须设置 until，且只有 since_id 可以不设上限，用于实时追踪。不提供任何 since 游标时，不会返回历史记录——你只能收到从现在开始的实时推送。到达 until 边界后，流会以 200 结束。

**这是最重要的一条可靠性经验：**持久化保存最后一个成功处理的事件 ID。每次（重新）连接时，将它作为 since 游标传入——v2/v2beta1 使用 since_id，v1 使用 since_ulid——这样 Alpaca 就会重放你在断连期间错过的所有事件。没有游标就重新连接的消费者，会悄无声息地丢弃停机期间发生的每个事件。

5. 顺序注意事项

在同一毫秒内，ULID 包含随机部分，因此同一毫秒内的两个事件可能以任意顺序排序。Alpaca 的建议是：为进行核对，请将流重启到最后一个事件之前几分钟的 since 位置，并依靠幂等处理消化重叠事件。Alpaca 保证单个账户内的顺序；对于跨账户顺序则不作保证。因此，应假定全局顺序大致有序，并进行去重，同时利用单账户内的顺序保证（见 §7）。

6. 可靠性模式（经验之谈）

SSE 连接会断开——网络、负载均衡器、部署和 Alpaca 端的重置都可能导致连接断开。生产环境中的消费者需要：

1. 解析注释行——绝不能丢弃。 任何以 : 开头的行都是 SSE 注释，而 Alpaca 会在其中传递三种含义：:heartbeat（连接正常，无需操作）；: you are reading too slowly, dropped N messages（静默数据丢失——应将其视为数据缺口，并从你的游标重新回放）；: internal server error（服务器即将关闭连接——重新连接；仅适用于 v2/v2beta1 流）。过滤 : 行的消费者会在毫无察觉的情况下丢失数据。
2. 检测静默。 每收到一帧就更新 lastMessageAt，包括心跳；如果流的静默时间达到你观察到的心跳间隔的几倍，就断开并重新连接——失效的套接字往往看起来仍处于“打开”状态。心跳间隔没有文档说明，因此应实际测量，而不是硬编码超时时间。
3. 读取循环中不要执行阻塞 I/O。 Alpaca 的明确建议是：先读入队列，再交由工作线程处理。每个事件都同步写入数据库，就会让你收到 dropped N messages 注释。
4. 使用带上限的指数退避进行重连。 出错或连接关闭时，等待一段时间后重新连接，等待时间逐次加倍，直至达到上限（例如从 1s 开始，上限为 60s）。成功连接后重置等待时间。
5. 防止同时发起多个重连。 使用标志，避免错误风暴导致多个并发重连尝试互相竞争。
6. 始终携带游标重新连接，游标是最后处理的事件（v2 使用 since_id，v1 使用 since_ulid；见 §4）。
7. 以幂等方式处理（见 §7）——回放造成的事件重叠是预期情况，并非异常。
8. 不要让副作用失败导致流中断。 使用 try/catch 包装每个事件的处理；记录日志后继续。单个错误事件（或下游服务中断）都不应阻止你继续消费其余事件。

注意：OpenAPI 无法完整描述 SSE，因此生成的 API 客户端常会在这些端点上挂起（等待一个永远不会结束的响应）。请使用真正的流式 HTTP/SSE 客户端，而不是代码生成的客户端。

7. 幂等处理流水线

每个事件的可靠处理方式如下，应由工作线程执行，而不是在读取循环中处理（§6）：

parse → persist a raw event snapshot (keyed on the dedup id, skip-if-exists)
      → match the local record by Alpaca ID (account_id / journal_id / order_id / transfer_id)
      → update local state guarded by current status, serialized per record
      → fire side effects (notifications, downstream transfers)
      → advance the stored cursor to this event_id

• 区分这两个 ID。 event_id 是回放游标；去重键是业务 ID。对于 Activity SSE，该键是 ref_id（交易的执行 ID，其他情况则是交易 ID）：回放会以至少一次的方式重新传递事件，而更正/撤销会通过 previous_id 关联，因此文档指定的去重键是 ref_id。补录的活动会在 event_id 序列末尾延迟到达，但其 at 保持原值。在旧版状态流中，event_id 同时承担这两种作用。
• 按记录串行处理变更，避免同一笔转账/订单的两个事件发生竞争。按 account_id 对工作线程分区是最简单的默认方案，也能利用每个账户的顺序保证（§5）；SELECT … FOR UPDATE 行锁或乐观版本列则是其他选择。
• 根据当前状态限制状态转换——例如，只处理尚未处于终态的转账，避免较晚到达或重复的“已执行”事件再次触发付款。
• 仅在处理成功后推进游标，这样处理中途崩溃会导致事件被重新回放，而不是被跳过（至少一次投递；幂等性可确保其安全）。

8. 各流的注意事项

• 账户状态： 根据 status_to == ACTIVE 控制引导界面和“启用交易”功能。
• 交易更新： new/accepted/pending_new 表示尚未成交；在 fill/partial_fill/canceled/rejected 时更新本地订单状态。成交时，使任何缓存的投资组合/持仓失效。
• 日志： 请记住，executed 并非最终状态，而 correct 会生成一个新的日志 ID（参见 alpaca-broker-journals）。幂等性和按 ID 索引的快照可同时处理这两种情况。
• 资金存入/转账： 4 种实体类型共用一个流；根据 entity_type 进行分支处理。转账状态均为大写（QUEUED、APPROVAL_PENDING、APPROVED、SENT_TO_CLEARING、COMPLETE、REJECTED、CANCELED、EXPIRED、RETURNED），而资金事件列表与 REST TransferStatus 枚举不同——后者包含 PENDING，但不含 EXPIRED——因此应接受两者的并集。资金钱包中每笔转账的状态可能仍需通过轮询获取（alpaca-broker-funding-transfers）。
• 活动 / NTA： 股息/费用/利息/公司行为和成交——将其作为以 ref_id 为键的活动快照持久化；这些数据用于余额/投资组合对账。活动中的 previous_id 保存更正后活动的 ref_id。

9. SSE 必不可少，但还不够

即便消费者设计得再完善，也可能漏掉事件（停机时间超出保留期限、出现错误或未处理某种类型）。始终将 SSE 与定期对账/修复流程配合使用，从 Alpaca 重新拉取权威状态（活动、日志、转账）并执行 upsert。SSE 用于降低延迟；对账用于确保正确性。请参见 alpaca-broker-reconciliation-idempotency。

相关技能： 正确性保障 → alpaca-broker-reconciliation-idempotency；去重/幂等机制 → alpaca-broker-reconciliation-idempotency；退避详情 → alpaca-broker-rate-limits-resilience；市场数据流（使用 WS，而非 SSE）→ alpaca-broker-market-data。