From bb04d7babd116a4b8d72efd11515ad373d308244 Mon Sep 17 00:00:00 2001 From: Emil Lerch Date: Tue, 30 Jun 2026 07:38:32 -0700 Subject: [PATCH] Tiingo as live provider message parsing - working end to end --- src/net/LiveStream.zig | 169 +++++++++++++++++++++++++++++++++++------ src/service.zig | 22 ++++-- 2 files changed, 163 insertions(+), 28 deletions(-) diff --git a/src/net/LiveStream.zig b/src/net/LiveStream.zig index 858412e..f12804f 100644 --- a/src/net/LiveStream.zig +++ b/src/net/LiveStream.zig @@ -5,25 +5,34 @@ //! per ticker via `snapshotInto`. The TUI/CLI poll the snapshot; they //! never block on the socket. //! -//! ## Transport-ready +//! ## Dual transport //! -//! The stream is parameterized by a `Transport` enum so a second -//! provider slots in without reshaping the scaffolding. Only `.yahoo` -//! is implemented today: +//! The stream is parameterized by a `Transport` enum; both arms share +//! all the scaffolding: //! //! - **Yahoo** (`streamer.finance.yahoo.com`): keyless, free, works on //! every tier. Subscribe with `{"subscribe":[...]}`; each frame is //! `{"type":"pricing","message":""}` whose protobuf //! carries field 1 = ticker (string) and field 2 = price (fixed32 //! float). It is an UNOFFICIAL feed (the browser uses it) and can -//! break without notice - hence "could die any time", which is why a -//! future Tiingo transport is planned as the paid, official backup. +//! break without notice - hence the official Tiingo backup. +//! - **Tiingo** (`api.tiingo.com/iex`): official IEX feed at +//! `thresholdLevel: 6` (derived `tngoLast` reference price; needs no +//! IEX license agreement - unlike raw levels 0/5 - so it works on ANY +//! Tiingo account, free tier included). Subscribe carries the API +//! key; each frame is a JSON `"A"` message whose `data` is +//! `[datetime, ticker, price]`. Each connect spends one request +//! against the hourly quota (a stable stream is then free), so it's +//! the default for Power subscribers (reconnect headroom) but any +//! keyed account can opt in. //! -//! The per-transport surface is exactly three things: the connect -//! target + handshake headers (`transportConfig`), the subscribe -//! message (`sendSubscribe`), and the frame parser (`serverMessage`). -//! Everything else - thread, lock, snapshot, reconnect/backoff, client -//! lifecycle - is shared. +//! Which one runs is chosen by `DataService.startLiveStream` from +//! `Config.effectiveLiveQuoteProvider()`, mirroring the REST live-quote +//! provider. The per-transport surface is exactly three things: the +//! connect target + handshake headers (`transportConfig`), the +//! subscribe message (`buildSubscribe`), and the frame parser +//! (`serverMessage`). Everything else - thread, lock, snapshot, +//! reconnect/backoff, client lifecycle - is shared. //! //! ## Why a raw background thread (not std.Io tasks) //! @@ -54,6 +63,7 @@ const std = @import("std"); const websocket = @import("websocket"); const json_utils = @import("../providers/json_utils.zig"); const jsonStr = json_utils.jsonStr; +const optFloat = json_utils.optFloat; const log = std.log.scoped(.live_stream); @@ -65,6 +75,9 @@ const yahoo_host = "streamer.finance.yahoo.com"; const yahoo_path = "/?version=2"; const yahoo_origin = "https://finance.yahoo.com"; +const tiingo_host = "api.tiingo.com"; +const tiingo_path = "/iex"; + /// Tickers are short; 24 bytes covers stocks, class suffixes, index /// (`^GSPC`), crypto (`BTC-USD`), and futures (`ES=F`) shapes. const max_ticker_len = 24; @@ -81,11 +94,18 @@ const reconnect_backoff_ms = 2_000; /// A frame whose decoded size exceeds this is skipped (returns null). const max_decoded_message = 1024; -/// Which live-quote feed the stream connects to. Only `.yahoo` is -/// wired today; the enum exists so a paid Tiingo transport is a -/// fill-in (add the arm to `transportConfig`/`sendSubscribe`/the -/// `serverMessage` parser) rather than a refactor. -pub const Transport = enum { yahoo }; +/// Which live-quote feed the stream connects to. +/// +/// - `.yahoo`: keyless, free, all tiers; base64+protobuf "pricing" +/// frames. Unofficial - can break without notice. +/// - `.tiingo`: official IEX feed via `wss://api.tiingo.com/iex` at +/// `thresholdLevel: 6` (the derived `tngoLast` reference price, which +/// - unlike raw levels 0/5 - needs no IEX license agreement, so it +/// works on ANY Tiingo account incl. the free tier). Requires an API +/// key in the subscribe message; each connect spends one request +/// against the hourly quota, so it's the default for Power +/// subscribers but any keyed account can opt in. +pub const Transport = enum { yahoo, tiingo }; /// A single parsed price update. The ticker is stored inline so the /// value owns no heap memory. @@ -238,6 +258,7 @@ pub fn snapshotInto(self: *LiveStream, symbols: []const []const u8, out: *std.St pub fn serverMessage(self: *LiveStream, data: []const u8) void { const tick = switch (self.transport) { .yahoo => parseYahooMessage(self.allocator, data), + .tiingo => parseTiingoMessage(self.allocator, data), } orelse return; var upper: [max_ticker_len]u8 = undefined; if (tick.ticker().len > upper.len) return; @@ -274,6 +295,12 @@ fn transportConfig(self: *LiveStream) TransportConfig { .path = yahoo_path, .headers = "Host: " ++ yahoo_host ++ "\r\nOrigin: " ++ yahoo_origin, }, + .tiingo => .{ + .host = tiingo_host, + .port = tls_port, + .path = tiingo_path, + .headers = "Host: " ++ tiingo_host, + }, }; } @@ -338,15 +365,17 @@ fn readUntilStopped(self: *LiveStream, c: *websocket.Client) !void { fn sendSubscribe(self: *LiveStream, c: *websocket.Client) !void { var buf: std.ArrayList(u8) = .empty; defer buf.deinit(self.allocator); - try buildSubscribe(self.transport, self.allocator, self.tickers, &buf); + try buildSubscribe(self.transport, self.allocator, self.api_key, self.tickers, &buf); // write() masks the buffer in place, so it must be a mutable buffer // we own; we discard it right after. try c.write(buf.items); } /// Build the transport's subscribe message into `buf`. Pure (no I/O), -/// so it's unit-testable without a live connection. -fn buildSubscribe(transport: Transport, allocator: std.mem.Allocator, tickers: []const []const u8, buf: *std.ArrayList(u8)) !void { +/// so it's unit-testable without a live connection. `api_key` is used +/// only by the Tiingo transport (its subscribe carries authorization); +/// Yahoo ignores it. +fn buildSubscribe(transport: Transport, allocator: std.mem.Allocator, api_key: ?[]const u8, tickers: []const []const u8, buf: *std.ArrayList(u8)) !void { switch (transport) { .yahoo => { // {"subscribe":["SPY","AAPL"]} @@ -361,6 +390,22 @@ fn buildSubscribe(transport: Transport, allocator: std.mem.Allocator, tickers: [ } try buf.appendSlice(allocator, "]}"); }, + .tiingo => { + // {"eventName":"subscribe","authorization":"", + // "eventData":{"thresholdLevel":6,"tickers":["spy",...]}} + // The key and tickers are alphanumeric (plus '.'/'-'), so no + // JSON string escaping is required. + try buf.appendSlice(allocator, "{\"eventName\":\"subscribe\",\"authorization\":\""); + try buf.appendSlice(allocator, api_key orelse ""); + try buf.appendSlice(allocator, "\",\"eventData\":{\"thresholdLevel\":6,\"tickers\":["); + for (tickers, 0..) |t, i| { + if (i != 0) try buf.append(allocator, ','); + try buf.append(allocator, '"'); + try buf.appendSlice(allocator, t); + try buf.append(allocator, '"'); + } + try buf.appendSlice(allocator, "]}}"); + }, } } @@ -473,6 +518,43 @@ fn readVarint(buf: []const u8, i: *usize) ?u64 { return null; // truncated } +// ── Tiingo frame parsing ────────────────────────────────────── + +/// Parse one Tiingo IEX text frame. Returns a tick only for +/// `thresholdLevel: 6` "A" (new-data) reference-price messages, whose +/// `data` array is `[datetime, ticker, referencePrice]`. Heartbeats +/// ("H"), subscription acks ("I"), and any malformed/unknown frame +/// yield null. Pure and allocator-scoped (parse memory freed before +/// return), so fully fixture-testable. +fn parseTiingoMessage(allocator: std.mem.Allocator, body: []const u8) ?StreamTick { + const parsed = std.json.parseFromSlice(std.json.Value, allocator, body, .{}) catch return null; + defer parsed.deinit(); + + const obj = switch (parsed.value) { + .object => |o| o, + else => return null, + }; + + // Only "A" carries price data; "H" (heartbeat) / "I" (ack) do not. + const mt = jsonStr(obj.get("messageType")) orelse return null; + if (!std.mem.eql(u8, mt, "A")) return null; + + const data = switch (obj.get("data") orelse return null) { + .array => |a| a.items, + else => return null, + }; + // Reference format: data[0]=datetime, data[1]=ticker, data[2]=price. + if (data.len < 3) return null; + const tkr = jsonStr(data[1]) orelse return null; + const price = optFloat(data[2]) orelse return null; + + var tick: StreamTick = .{ .price = price }; + const n = @min(tkr.len, tick.ticker_buf.len); + @memcpy(tick.ticker_buf[0..n], tkr[0..n]); + tick.ticker_len = n; + return tick; +} + // ── Tests ───────────────────────────────────────────────────── test "parseYahooMessage: AAPL pricing frame yields ticker + price" { @@ -666,15 +748,58 @@ test "buildSubscribe: yahoo message for one and many tickers" { var buf: std.ArrayList(u8) = .empty; defer buf.deinit(a); - try buildSubscribe(.yahoo, a, &.{"SPY"}, &buf); + try buildSubscribe(.yahoo, a, null, &.{"SPY"}, &buf); try std.testing.expectEqualStrings("{\"subscribe\":[\"SPY\"]}", buf.items); buf.clearRetainingCapacity(); - try buildSubscribe(.yahoo, a, &.{ "SPY", "AAPL", "BTC-USD" }, &buf); + try buildSubscribe(.yahoo, a, null, &.{ "SPY", "AAPL", "BTC-USD" }, &buf); try std.testing.expectEqualStrings("{\"subscribe\":[\"SPY\",\"AAPL\",\"BTC-USD\"]}", buf.items); // No tickers -> empty array (degenerate but well-formed). buf.clearRetainingCapacity(); - try buildSubscribe(.yahoo, a, &.{}, &buf); + try buildSubscribe(.yahoo, a, null, &.{}, &buf); try std.testing.expectEqualStrings("{\"subscribe\":[]}", buf.items); } + +test "buildSubscribe: tiingo message carries auth + thresholdLevel 6" { + const a = std.testing.allocator; + var buf: std.ArrayList(u8) = .empty; + defer buf.deinit(a); + + try buildSubscribe(.tiingo, a, "secret-key", &.{ "spy", "aapl" }, &buf); + try std.testing.expectEqualStrings( + "{\"eventName\":\"subscribe\",\"authorization\":\"secret-key\",\"eventData\":{\"thresholdLevel\":6,\"tickers\":[\"spy\",\"aapl\"]}}", + buf.items, + ); +} + +test "parseTiingoMessage: threshold-6 A message yields ticker + price" { + const body = + \\{"messageType":"A","service":"iex","data":["2026-06-29T13:33:45.383-05:00","spy",612.595]} + ; + const tick = parseTiingoMessage(std.testing.allocator, body) orelse return error.TestExpectedTick; + try std.testing.expectEqualStrings("spy", tick.ticker()); + try std.testing.expectApproxEqAbs(@as(f64, 612.595), tick.price, 0.0001); +} + +test "parseTiingoMessage: heartbeat / ack / malformed frames yield null" { + const a = std.testing.allocator; + // Heartbeat "H" and ack "I" carry no price data. + try std.testing.expect(parseTiingoMessage(a, + \\{"messageType":"H","response":{"code":200,"message":"HeartBeat"}} + ) == null); + try std.testing.expect(parseTiingoMessage(a, + \\{"messageType":"I","response":{"code":200},"data":{"subscriptionId":1}} + ) == null); + // "A" but data too short. + try std.testing.expect(parseTiingoMessage(a, + \\{"messageType":"A","data":["2026-06-29","spy"]} + ) == null); + // "A" but price isn't a number. + try std.testing.expect(parseTiingoMessage(a, + \\{"messageType":"A","data":["2026-06-29","spy","oops"]} + ) == null); + // Not JSON / array root. + try std.testing.expect(parseTiingoMessage(a, "not json") == null); + try std.testing.expect(parseTiingoMessage(a, "[1,2,3]") == null); +} diff --git a/src/service.zig b/src/service.zig index 8f9884c..bef84c2 100644 --- a/src/service.zig +++ b/src/service.zig @@ -759,9 +759,10 @@ pub const DataService = struct { // ── Live-price stream (push-based intraday quotes) ────────── // - // Phase A wires the Yahoo transport (keyless, free, all tiers). A - // later phase adds Tiingo as the paid/official transport, selected - // via Config.effectiveLiveQuoteProvider(). + // The transport mirrors the REST live-quote provider + // (Config.effectiveLiveQuoteProvider): a paid Tiingo subscriber + // streams via Tiingo's official IEX feed, everyone else via Yahoo's + // free (unofficial) feed. /// Start (or ensure running) the live-price stream for `symbols`. /// Idempotent while running: the symbol set is fixed for the @@ -771,12 +772,21 @@ pub const DataService = struct { /// /// This is the push counterpart to a one-shot quote fetch: once /// running, `liveStreamSnapshot` returns continuously-updated prices - /// with no further requests. The Yahoo transport is keyless and - /// free, so it runs on any tier. + /// with no further requests. Transport follows + /// `effectiveLiveQuoteProvider`: `.tiingo` (official real-time IEX, + /// key-guarded) for a Power subscriber or anyone who set + /// `ZFIN_LIVE_QUOTE_PROVIDER=tiingo` (IEX level-6 works on the free + /// tier too - one request per connect), else `.yahoo` (keyless, all + /// tiers, but ~15-min delayed/unofficial). pub fn startLiveStream(self: *DataService, symbols: []const []const u8) DataError!void { self.assertNetworkAllowed("startLiveStream"); if (self.live_stream == null) { - self.live_stream = LiveStream.create(self.allocator, self.io, .yahoo, null) catch return DataError.OutOfMemory; + // `effectiveLiveQuoteProvider` already key-guards Tiingo, so + // `.tiingo` here implies `tiingo_key != null`. + const use_tiingo = self.config.effectiveLiveQuoteProvider() == .tiingo; + const transport: LiveStream.Transport = if (use_tiingo) .tiingo else .yahoo; + const key: ?[]const u8 = if (use_tiingo) self.config.tiingo_key else null; + self.live_stream = LiveStream.create(self.allocator, self.io, transport, key) catch return DataError.OutOfMemory; } self.live_stream.?.start(symbols) catch |err| { log.warn("startLiveStream: {t}", .{err});