diff --git a/build.zig b/build.zig index d9e669f..2de1310 100644 --- a/build.zig +++ b/build.zig @@ -26,6 +26,15 @@ pub fn build(b: *std.Build) void { .optimize = optimize, }); + // karlseguin/websocket.zig: blocking-socket + std.Thread websocket + // client used by the live-price stream (src/net/LiveStream.zig). + // Internal-only, so it's wired into the unified module's imports + // below but NOT into the public `zfin` library module. + const websocket_dep = b.dependency("websocket", .{ + .target = target, + .optimize = optimize, + }); + const srf_mod = srf_dep.module("srf"); const shiller_mod = b.addModule("shiller_year", .{ @@ -61,6 +70,7 @@ pub fn build(b: *std.Build) void { .{ .name = "vaxis", .module = vaxis_dep.module("vaxis") }, .{ .name = "z2d", .module = z2d_dep.module("z2d") }, .{ .name = "zeit", .module = zeit_dep.module("zeit") }, + .{ .name = "websocket", .module = websocket_dep.module("websocket") }, .{ .name = "build_info", .module = build_info }, .{ .name = "shiller_year", .module = shiller_mod }, }; diff --git a/build.zig.zon b/build.zig.zon index 6517060..4d9df62 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -20,6 +20,10 @@ .url = "git+https://github.com/rockorager/zeit?ref=v0.9.0#b1c1c2fcbc71fd7799a316bbcf0ff88d06d80ccc", .hash = "zeit-0.9.0-5I6bk2m9AgBSMH8-L6rYJkwuQAyhXplnfxnvTSGzVHUR", }, + .websocket = .{ + .url = "git+https://github.com/karlseguin/websocket.zig#99df0d3533a41cbcd5ff59b9af68bbfe6169cc62", + .hash = "websocket-0.1.0-ZPISdangBAAgWPap_MVU6Nk4rj3GT3xuJuF0IIv8HN6G", + }, }, .paths = .{ "build", diff --git a/src/commands/quote.zig b/src/commands/quote.zig index 2e20feb..ca2555f 100644 --- a/src/commands/quote.zig +++ b/src/commands/quote.zig @@ -26,6 +26,11 @@ pub const ParsedArgs = struct { /// "2025-01-15"), kept only to label the "Change ()" detail /// row. Null when `--since` was omitted. since_raw: ?[]const u8 = null, + /// When set (`--live`), stream live prices for the symbol + /// continuously (Yahoo websocket) instead of printing a one-shot + /// quote. Runs until interrupted (Ctrl-C) or killed (e.g. + /// `timeout`). Ignores the chart / --since / --export-chart options. + live: bool = false, }; pub const meta: framework.Meta = .{ @@ -34,7 +39,7 @@ pub const meta: framework.Meta = .{ .synopsis = "Show latest quote with chart and 20-day history", .uppercase_first_arg = true, .help = - \\Usage: zfin quote [--since ] [--export-chart ] + \\Usage: zfin quote [--since ] [--export-chart ] [--live] \\ \\Show the latest real-time quote for a symbol (Yahoo / TwelveData) \\plus a price chart over a recent window (an inline Kitty image @@ -58,12 +63,21 @@ pub const meta: framework.Meta = .{ \\ (1920x1080) and exit. No text output \\ is emitted. Uses the TUI's default \\ theme. + \\ --live Stream live prices for SYMBOL continuously + \\ (free Yahoo websocket) instead of a + \\ one-shot quote. Prints the latest price + \\ once per second until interrupted (Ctrl-C) + \\ or killed. Ignores the chart options above. + \\ Outside market hours, US equities stream a + \\ closing snapshot then go quiet; 24/7 + \\ symbols (e.g. BTC-USD) keep ticking. \\ \\Examples: \\ zfin quote AAPL \\ zfin quote spy # symbols are case-insensitive \\ zfin quote AAPL --since 1Y \\ zfin quote AAPL --export-chart aapl.png + \\ timeout 12 zfin quote --live BTC-USD # watch live ticks for 12s \\ , .user_errors = error{ MissingSymbol, UnexpectedArg, MissingFlagValue, InvalidDate }, @@ -85,6 +99,7 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr var export_chart: ?[]const u8 = null; var since: ?zfin.Date = null; var since_raw: ?[]const u8 = null; + var live: bool = false; var i: usize = 0; while (i < cmd_args.len) : (i += 1) { @@ -95,6 +110,8 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr const value = try cli.requireFlagValue(ctx.io, cmd_args, &i, a); since = cli.parseRequiredDateOrStderr(ctx.io, value, ctx.today, "--since") catch return error.InvalidDate; since_raw = value; + } else if (std.mem.eql(u8, a, "--live")) { + live = true; } else if (a.len > 0 and a[0] == '-') { // Reject ANY leading-dash token we don't recognize, // including single-dash ones like `-x`. Previously only @@ -117,11 +134,14 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr cli.stderrPrint(ctx.io, "Error: 'quote' requires a symbol argument\n"); return error.MissingSymbol; } - return .{ .symbol = symbol.?, .export_chart = export_chart, .since = since, .since_raw = since_raw }; + return .{ .symbol = symbol.?, .export_chart = export_chart, .since = since, .since_raw = since_raw, .live = live }; } pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void { const svc = ctx.svc orelse return error.MissingDataService; + // `--live` streams continuously and shares none of the one-shot + // quote's candle/chart/name machinery, so branch before any of it. + if (parsed.live) return runLive(ctx, svc, parsed.symbol); const opts = cli.fetchOptionsFromPolicy(ctx.globals.refresh_policy); // Fetch candle data for chart and history const candle_result = svc.getCandles(parsed.symbol, opts) catch |err| switch (err) { @@ -233,6 +253,51 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void { try display(ctx.allocator, candles, quote, parsed.symbol, name, ctx.today, ctx.color, ctx.out, display_count, window_label, chart_render); } +/// `--live`: open the live-price websocket stream for `symbol` and +/// print the latest price once per second until the process is +/// interrupted (Ctrl-C) or killed (e.g. `timeout 12 zfin quote --live +/// SPY`). Each line is flushed immediately so output appears in real +/// time and a SIGTERM mid-loop doesn't swallow buffered lines. +/// +/// This is the streaming counterpart to the one-shot quote above and +/// the simplest end-to-end exercise of the Yahoo stream transport. +/// Note: outside market/extended hours, US equities stream their +/// closing snapshot once and then go quiet; 24/7 symbols (e.g. +/// `BTC-USD`) keep ticking. +fn runLive(ctx: *framework.RunCtx, svc: *zfin.DataService, symbol: []const u8) !void { + const syms = [_][]const u8{symbol}; + svc.startLiveStream(&syms) catch |err| { + cli.stderrPrint(ctx.io, "Error: failed to start live stream for "); + cli.stderrPrint(ctx.io, symbol); + cli.stderrPrint(ctx.io, ": "); + cli.stderrPrint(ctx.io, @errorName(err)); + cli.stderrPrint(ctx.io, "\n"); + return; + }; + defer svc.stopLiveStream(); + + try ctx.out.print("Streaming live quotes for {s} - Ctrl-C (or timeout) to stop.\n", .{symbol}); + try ctx.out.flush(); + + var prices = std.StringHashMap(f64).init(ctx.allocator); + defer prices.deinit(); + + var elapsed: usize = 0; + while (true) { + prices.clearRetainingCapacity(); + svc.liveStreamSnapshot(&syms, &prices); + if (prices.get(symbol)) |p| { + try ctx.out.print(" [{d:>4}s] {s} {f}\n", .{ elapsed, symbol, Money.from(p) }); + } else { + try ctx.out.print(" [{d:>4}s] {s} (waiting for first tick...)\n", .{ elapsed, symbol }); + } + try ctx.out.flush(); + // Cancelable; on cancel, break so `defer stopLiveStream` runs. + ctx.io.sleep(.fromSeconds(1), .real) catch break; + elapsed += 1; + } +} + /// Copy `s` (clamped to `buf`'s capacity) into `buf` and return the /// written slice. Fund/security names fit easily in 256 bytes. fn clampName(buf: []u8, s: []const u8) []const u8 { @@ -505,6 +570,23 @@ test "parseArgs: --since with an invalid value is rejected" { try std.testing.expectError(error.InvalidDate, parseArgs(&ctx, &args)); } +test "parseArgs: --live sets the live flag" { + var ctx: framework.RunCtx = undefined; + ctx.io = std.testing.io; + const args = [_][]const u8{ "SPY", "--live" }; + const parsed = try parseArgs(&ctx, &args); + try std.testing.expectEqualStrings("SPY", parsed.symbol); + try std.testing.expect(parsed.live); +} + +test "parseArgs: live defaults to false when --live is omitted" { + var ctx: framework.RunCtx = undefined; + ctx.io = std.testing.io; + const args = [_][]const u8{"SPY"}; + const parsed = try parseArgs(&ctx, &args); + try std.testing.expect(!parsed.live); +} + test "display with candles only" { var buf: [8192]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); diff --git a/src/net/LiveStream.zig b/src/net/LiveStream.zig new file mode 100644 index 0000000..8038887 --- /dev/null +++ b/src/net/LiveStream.zig @@ -0,0 +1,658 @@ +//! Live-price websocket stream (push-based intraday quotes). +//! +//! A `LiveStream` holds a single persistent websocket on a background +//! thread, parses incoming price ticks, and exposes the latest price +//! per ticker via `snapshotInto`. The TUI/CLI poll the snapshot; they +//! never block on the socket. +//! +//! ## Transport-ready +//! +//! The stream is parameterized by a `Transport` enum so a second +//! provider slots in without reshaping the scaffolding. Only `.yahoo` +//! is implemented today: +//! +//! - **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. +//! +//! 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. +//! +//! ## Why a raw background thread (not std.Io tasks) +//! +//! The read loop is a single long-lived blocking loop (`readLoop`), +//! so it runs on its own OS thread that coexists with the app's +//! `Io.Threaded` event loop. The pinned `websocket.zig` client is +//! itself Io-aware (connect, TLS, reads, writes go through `std.Io`, +//! which is safe to call from this thread); we thread `io` in for it. +//! +//! ## Threading and shutdown +//! +//! `prices` and `client` are guarded by `mutex` (a `std.atomic.Mutex` +//! spinlock, since the read loop runs outside the Io task model and +//! critical sections are single hashmap ops held for microseconds). +//! Shutdown: `stop()` sets `should_stop` and calls `client.close()` +//! from the caller's thread, which unblocks the read thread's +//! `readLoop`, after which we `join`. The blocking connect (DNS + TCP +//! + TLS) runs WITHOUT holding `mutex`, so a quit during a slow connect +//! isn't blocked. + +const std = @import("std"); +const websocket = @import("websocket"); +const json_utils = @import("../providers/json_utils.zig"); +const jsonStr = json_utils.jsonStr; + +const log = std.log.scoped(.live_stream); + +const LiveStream = @This(); + +const tls_port: u16 = 443; + +const yahoo_host = "streamer.finance.yahoo.com"; +const yahoo_path = "/?version=2"; +const yahoo_origin = "https://finance.yahoo.com"; + +/// Tickers are short; 24 bytes covers stocks, class suffixes, index +/// (`^GSPC`), crypto (`BTC-USD`), and futures (`ES=F`) shapes. +const max_ticker_len = 24; +const handshake_timeout_ms = 10_000; +const max_message_size = 512 * 1024; +const read_buffer_size = 32 * 1024; +const reconnect_backoff_ms = 2_000; +/// Yahoo pricing protobufs are ~60-90 bytes decoded; 1 KiB is ample. +/// 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 }; + +/// A single parsed price update. The ticker is stored inline so the +/// value owns no heap memory. +pub const StreamTick = struct { + ticker_buf: [max_ticker_len]u8 = @splat(0), + ticker_len: usize = 0, + price: f64 = 0, + + pub fn ticker(self: *const StreamTick) []const u8 { + return self.ticker_buf[0..self.ticker_len]; + } +}; + +const TransportConfig = struct { + host: []const u8, + port: u16, + path: []const u8, + /// Raw handshake header block (CRLF-separated, no trailing CRLF). + /// The library does not auto-add Host, so it is included here. + headers: []const u8, +}; + +// ── Fields ──────────────────────────────────────────────────── + +allocator: std.mem.Allocator, +/// Threaded into the websocket client (the pinned `websocket.zig` is +/// Io-aware: connect, TLS, reads, and writes all go through it). +/// `std.Io` is safe to use from the read thread. +io: std.Io, +transport: Transport, +/// Owned copy of the API token, or null for keyless transports +/// (Yahoo). Decoupled from Config's lifetime. +api_key: ?[]u8 = null, +/// Owned, duped tickers to subscribe. Set in `start`, freed in `stop`. +tickers: [][]u8 = &.{}, +/// Guards `prices` and `client`. Spinlock (not an Io mutex) because the +/// read loop is a raw OS thread outside the Io task model; critical +/// sections are one hashmap op or a short pointer read. +mutex: std.atomic.Mutex = .unlocked, +/// Latest price per ticker, keyed by UPPERCASED ticker (owned dup). +/// Written by the read thread, read via `snapshotInto`. +prices: std.StringHashMap(f64), +/// The live client, published so `stop()` can close it cross-thread. +/// Heap-owned by the read thread for the duration of one connection. +client: ?*websocket.Client = null, +thread: ?std.Thread = null, +should_stop: std.atomic.Value(bool) = .init(false), + +// ── Lifecycle ───────────────────────────────────────────────── + +/// Allocate and initialize a stream. The returned pointer is stable +/// (the background thread captures it), so callers must keep it +/// heap-allocated and call `destroy` (not move it). `api_key` is copied +/// when non-null; pass null for keyless transports. +pub fn create(allocator: std.mem.Allocator, io: std.Io, transport: Transport, api_key: ?[]const u8) !*LiveStream { + const self = try allocator.create(LiveStream); + errdefer allocator.destroy(self); + const key_dup: ?[]u8 = if (api_key) |k| try allocator.dupe(u8, k) else null; + self.* = .{ + .allocator = allocator, + .io = io, + .transport = transport, + .api_key = key_dup, + .prices = std.StringHashMap(f64).init(allocator), + }; + return self; +} + +/// Stop the stream (if running) and free everything. +pub fn destroy(self: *LiveStream) void { + self.stop(); + var it = self.prices.keyIterator(); + while (it.next()) |k| self.allocator.free(k.*); + self.prices.deinit(); + if (self.api_key) |k| self.allocator.free(k); + self.allocator.destroy(self); +} + +/// Start streaming `symbols`. Idempotent: a no-op if already running +/// (the symbol set is fixed for the stream's lifetime; change it with +/// `stop` + `start`). `symbols` is copied. +pub fn start(self: *LiveStream, symbols: []const []const u8) !void { + if (self.thread != null) return; + + const tk = try self.allocator.alloc([]u8, symbols.len); + errdefer self.allocator.free(tk); + var n: usize = 0; + errdefer for (tk[0..n]) |t| self.allocator.free(t); + for (symbols) |s| { + tk[n] = try self.allocator.dupe(u8, s); + n += 1; + } + self.tickers = tk; + + self.should_stop.store(false, .release); + self.thread = try std.Thread.spawn(.{}, threadMain, .{self}); +} + +/// Signal the read thread to stop, close the connection to unblock it, +/// join, and free the subscribed-ticker list. Safe to call when not +/// running. +pub fn stop(self: *LiveStream) void { + self.should_stop.store(true, .release); + // Close the socket so the read thread's blocking readLoop returns. + // close() is thread-safe per the library. + self.lock(); + if (self.client) |c| c.close(.{}) catch |err| log.debug("close on stop: {t}", .{err}); + self.unlock(); + + if (self.thread) |t| { + t.join(); + self.thread = null; + } + + for (self.tickers) |t| self.allocator.free(t); + self.allocator.free(self.tickers); + self.tickers = &.{}; +} + +pub fn isRunning(self: *LiveStream) bool { + return self.thread != null; +} + +/// Spin-acquire `mutex`. Critical sections are O(1) hashmap ops (or an +/// O(symbols) snapshot scan), so the spin is effectively uncontended. +fn lock(self: *LiveStream) void { + while (!self.mutex.tryLock()) std.atomic.spinLoopHint(); +} + +fn unlock(self: *LiveStream) void { + self.mutex.unlock(); +} + +/// Copy the latest streamed prices for `symbols` into `out`, keyed by +/// the caller's original slices (so `out`'s keys borrow `symbols`). +/// Matching is case-insensitive (we store UPPERCASED tickers). Symbols +/// with no streamed price yet are left absent (caller falls back to the +/// cached close). +pub fn snapshotInto(self: *LiveStream, symbols: []const []const u8, out: *std.StringHashMap(f64)) void { + self.lock(); + defer self.unlock(); + var upper: [max_ticker_len]u8 = undefined; + for (symbols) |s| { + if (s.len > upper.len) continue; + const key = std.ascii.upperString(upper[0..], s); + if (self.prices.get(key)) |p| { + out.put(s, p) catch |err| log.warn("snapshot put({s}): {t}", .{ s, err }); + } + } +} + +/// Read-thread handler hook (called by `websocket.Client.readLoop`). +/// Parses one frame and updates `prices`. Never propagates an error (a +/// bad frame must not tear down the stream); failures are swallowed so +/// the connection stays up. +pub fn serverMessage(self: *LiveStream, data: []const u8) !void { + const tick = switch (self.transport) { + .yahoo => parseYahooMessage(self.allocator, data), + } orelse return; + var upper: [max_ticker_len]u8 = undefined; + if (tick.ticker().len > upper.len) return; + const key = std.ascii.upperString(upper[0..], tick.ticker()); + + self.lock(); + defer self.unlock(); + const gop = self.prices.getOrPut(key) catch return; + if (!gop.found_existing) { + gop.key_ptr.* = self.allocator.dupe(u8, key) catch { + // Roll back the entry whose key is still the stack slice. + _ = self.prices.remove(key); + return; + }; + } + gop.value_ptr.* = tick.price; +} + +// ── Read-thread internals ───────────────────────────────────── + +fn threadMain(self: *LiveStream) void { + while (!self.should_stop.load(.acquire)) { + self.connectAndRun() catch |err| log.debug("stream connection ended: {t}", .{err}); + if (self.should_stop.load(.acquire)) break; + self.backoffSleep(reconnect_backoff_ms); + } +} + +fn transportConfig(self: *LiveStream) TransportConfig { + return switch (self.transport) { + .yahoo => .{ + .host = yahoo_host, + .port = tls_port, + .path = yahoo_path, + .headers = "Host: " ++ yahoo_host ++ "\r\nOrigin: " ++ yahoo_origin, + }, + }; +} + +fn connectAndRun(self: *LiveStream) !void { + const cfg = self.transportConfig(); + + // Blocking connect (DNS + TCP + TLS) happens BEFORE publishing the + // client, so a concurrent stop() isn't blocked waiting on it. + const cptr = try self.allocator.create(websocket.Client); + cptr.* = websocket.Client.init(self.io, self.allocator, .{ + .host = cfg.host, + .port = cfg.port, + .tls = true, + .max_size = max_message_size, + .buffer_size = read_buffer_size, + }) catch |err| { + self.allocator.destroy(cptr); + return err; + }; + // From here cptr owns a live connection; always tear down + unpublish. + defer { + self.lock(); + self.client = null; + self.unlock(); + cptr.deinit(); + self.allocator.destroy(cptr); + } + + // Publish so stop() can close the socket to unblock readLoop. + self.lock(); + self.client = cptr; + self.unlock(); + + try cptr.handshake(cfg.path, .{ + .timeout_ms = handshake_timeout_ms, + .headers = cfg.headers, + }); + try self.sendSubscribe(cptr); + // Blocks until the connection closes (or stop() closes it). The + // library auto-handles ping/pong and close frames; our + // serverMessage handles the text data frames. + try cptr.readLoop(self); +} + +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); + // 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 { + switch (transport) { + .yahoo => { + // {"subscribe":["SPY","AAPL"]} + // Tickers are alphanumeric plus '.'/'-'/'^'/'='/'=X', so no + // JSON string escaping is required. + try buf.appendSlice(allocator, "{\"subscribe\":["); + 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, "]}"); + }, + } +} + +/// Sleep `ms` in small chunks, bailing early if asked to stop, so a +/// quit during reconnect backoff doesn't wait the full interval. Uses +/// the `.awake` (monotonic) clock so a wall-clock jump can't skip or +/// extend the backoff. +fn backoffSleep(self: *LiveStream, ms: u64) void { + var remaining = ms; + while (remaining > 0 and !self.should_stop.load(.acquire)) { + const chunk = @min(remaining, 200); + self.io.sleep(.fromMilliseconds(@intCast(chunk)), .awake) catch return; + remaining -= chunk; + } +} + +// ── Yahoo frame parsing ─────────────────────────────────────── + +/// Parse one Yahoo streamer frame. Returns a tick only for +/// `{"type":"pricing","message":""}` frames whose +/// protobuf carries field 1 (ticker) and field 2 (price). Any other +/// type, or a malformed/oversized frame, yields null. Pure and +/// allocator-scoped (parse memory freed before return), so fully +/// fixture-testable. +fn parseYahooMessage(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 "pricing" frames carry a price payload. + const typ = jsonStr(obj.get("type")) orelse return null; + if (!std.mem.eql(u8, typ, "pricing")) return null; + const b64 = jsonStr(obj.get("message")) orelse return null; + + // Base64-decode the protobuf into a stack buffer. Standard alphabet + // with padding (Yahoo emits '+', '/', '='). + const dec = std.base64.standard.Decoder; + const decoded_len = dec.calcSizeForSlice(b64) catch return null; + if (decoded_len > max_decoded_message) return null; + var buf: [max_decoded_message]u8 = undefined; + dec.decode(buf[0..decoded_len], b64) catch return null; + + return parsePricingProto(buf[0..decoded_len]); +} + +/// Walk a Yahoo PricingData protobuf for field 1 (ticker, length- +/// delimited string) and field 2 (price, fixed32 little-endian float), +/// skipping every other field. Returns null if either is absent or the +/// buffer is malformed. +fn parsePricingProto(pb: []const u8) ?StreamTick { + var i: usize = 0; + var ticker: ?[]const u8 = null; + var price: ?f64 = null; + + while (i < pb.len) { + const tag = readVarint(pb, &i) orelse break; + const field = tag >> 3; + const wire: u3 = @intCast(tag & 0x7); + switch (wire) { + 0 => _ = readVarint(pb, &i) orelse break, // varint + 1 => { // fixed64 + if (i + 8 > pb.len) break; + i += 8; + }, + 2 => { // length-delimited + const len = readVarint(pb, &i) orelse break; + if (i + len > pb.len) break; + const bytes = pb[i .. i + len]; + i += len; + if (field == 1) ticker = bytes; + }, + 5 => { // fixed32 + if (i + 4 > pb.len) break; + const raw = std.mem.readInt(u32, pb[i..][0..4], .little); + i += 4; + if (field == 2) price = @as(f64, @as(f32, @bitCast(raw))); + }, + else => break, // groups (3/4) are deprecated/unsupported + } + if (ticker != null and price != null) break; + } + + const t = ticker orelse return null; + const p = price orelse return null; + var tick: StreamTick = .{ .price = p }; + const n = @min(t.len, tick.ticker_buf.len); + @memcpy(tick.ticker_buf[0..n], t[0..n]); + tick.ticker_len = n; + return tick; +} + +/// Decode a base-128 protobuf varint starting at `i.*`, advancing `i` +/// past it. Returns null on truncation or an over-long (> 64-bit) +/// encoding. +fn readVarint(buf: []const u8, i: *usize) ?u64 { + var result: u64 = 0; + var shift: usize = 0; + while (i.* < buf.len) { + const byte = buf[i.*]; + i.* += 1; + if (shift >= 64) return null; // malformed: more than 10 bytes + result |= @as(u64, byte & 0x7f) << @intCast(shift); + if (byte & 0x80 == 0) return result; + shift += 7; + } + return null; // truncated +} + +// ── Tests ───────────────────────────────────────────────────── + +test "parseYahooMessage: AAPL pricing frame yields ticker + price" { + const body = + \\{"type":"pricing","message":"CgRBQVBMFUgBjUMY4JKAz+JnKgNOTVMwCDgBRQisH79InK3UKmUAj+K/sAFQ2AEE"} + ; + const tick = parseYahooMessage(std.testing.allocator, body) orelse return error.TestExpectedTick; + try std.testing.expectEqualStrings("AAPL", tick.ticker()); + // field 2 is a fixed32 float: exactly 282.010009765625. + try std.testing.expectApproxEqAbs(@as(f64, 282.010009765625), tick.price, 0.0001); +} + +test "parseYahooMessage: SPY pricing frame yields ticker + price" { + const body = + \\{"type":"pricing","message":"CgNTUFkVUhg5RBiwooDP4mcqA1BDWDAUOAFF9v3HP0jQ+bonZYA9NkGwAVDYAQQ="} + ; + const tick = parseYahooMessage(std.testing.allocator, body) orelse return error.TestExpectedTick; + try std.testing.expectEqualStrings("SPY", tick.ticker()); + try std.testing.expectApproxEqAbs(@as(f64, 740.3800048828125), tick.price, 0.0001); +} + +test "parseYahooMessage: non-pricing and malformed frames yield null" { + const a = std.testing.allocator; + // Not JSON. + try std.testing.expect(parseYahooMessage(a, "not json") == null); + // JSON array root, not object. + try std.testing.expect(parseYahooMessage(a, "[1,2,3]") == null); + // Pricing-shaped but wrong type. + try std.testing.expect(parseYahooMessage(a, + \\{"type":"heartbeat","message":"CgRBQVBM"} + ) == null); + // Missing message field. + try std.testing.expect(parseYahooMessage(a, + \\{"type":"pricing"} + ) == null); + // message present but not valid base64. + try std.testing.expect(parseYahooMessage(a, + \\{"type":"pricing","message":"!!!not-base64!!!"} + ) == null); +} + +test "parsePricingProto: missing field 1 or field 2 yields null" { + // Only field 1 (ticker "AB"), no price field. + const ticker_only = [_]u8{ 0x0a, 0x02, 'A', 'B' }; + try std.testing.expect(parsePricingProto(&ticker_only) == null); + // Only field 2 (price), no ticker. + var price_only: [5]u8 = undefined; + price_only[0] = 0x15; // field 2, wire type 5 (fixed32) + std.mem.writeInt(u32, price_only[1..5], @bitCast(@as(f32, 12.5)), .little); + try std.testing.expect(parsePricingProto(&price_only) == null); +} + +test "parsePricingProto: skips intervening fields to reach price" { + // field 1 = "X", field 3 varint (skipped), field 2 = 9.5 fixed32. + var pb: std.ArrayList(u8) = .empty; + defer pb.deinit(std.testing.allocator); + const a = std.testing.allocator; + try pb.appendSlice(a, &.{ 0x0a, 0x01, 'X' }); // field 1, len 1, "X" + try pb.appendSlice(a, &.{ 0x18, 0x96, 0x01 }); // field 3 varint = 150 + try pb.append(a, 0x15); // field 2, wire 5 + var price_bytes: [4]u8 = undefined; + std.mem.writeInt(u32, &price_bytes, @bitCast(@as(f32, 9.5)), .little); + try pb.appendSlice(a, &price_bytes); + + const tick = parsePricingProto(pb.items) orelse return error.TestExpectedTick; + try std.testing.expectEqualStrings("X", tick.ticker()); + try std.testing.expectApproxEqAbs(@as(f64, 9.5), tick.price, 0.0001); +} + +test "parsePricingProto: skips a fixed64 field before reaching the price" { + var pb: std.ArrayList(u8) = .empty; + defer pb.deinit(std.testing.allocator); + const a = std.testing.allocator; + try pb.appendSlice(a, &.{ 0x0a, 0x01, 'Y' }); // field 1, "Y" + try pb.append(a, 0x21); // field 4, wire type 1 (fixed64) + try pb.appendSlice(a, &.{ 0, 0, 0, 0, 0, 0, 0, 0 }); // 8 bytes, skipped + try pb.append(a, 0x15); // field 2, wire type 5 (fixed32) + var price_bytes: [4]u8 = undefined; + std.mem.writeInt(u32, &price_bytes, @bitCast(@as(f32, 3.25)), .little); + try pb.appendSlice(a, &price_bytes); + + const tick = parsePricingProto(pb.items) orelse return error.TestExpectedTick; + try std.testing.expectEqualStrings("Y", tick.ticker()); + try std.testing.expectApproxEqAbs(@as(f64, 3.25), tick.price, 0.0001); +} + +test "parsePricingProto: unsupported wire type (group) stops the walk" { + // field 1 = "Z", then a group-start tag (wire type 3) -> break. + // Price is never seen, so the result is null. + const pb = [_]u8{ 0x0a, 0x01, 'Z', 0x0b }; // 0x0b = field 1, wire type 3 + try std.testing.expect(parsePricingProto(&pb) == null); +} + +test "parsePricingProto: over-long ticker is truncated to capacity" { + var pb: std.ArrayList(u8) = .empty; + defer pb.deinit(std.testing.allocator); + const a = std.testing.allocator; + const long = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; // 36 chars + try pb.append(a, 0x0a); // field 1, wire 2 + try pb.append(a, @intCast(long.len)); + try pb.appendSlice(a, long); + try pb.append(a, 0x15); // field 2, wire 5 + var price_bytes: [4]u8 = undefined; + std.mem.writeInt(u32, &price_bytes, @bitCast(@as(f32, 1.0)), .little); + try pb.appendSlice(a, &price_bytes); + + const tick = parsePricingProto(pb.items) orelse return error.TestExpectedTick; + try std.testing.expectEqual(@as(usize, max_ticker_len), tick.ticker().len); + try std.testing.expectEqualStrings("ABCDEFGHIJKLMNOPQRSTUVWX", tick.ticker()); +} + +test "readVarint: multi-byte and truncated encodings" { + var i: usize = 0; + // 150 = 0x96 0x01 + const ok = [_]u8{ 0x96, 0x01 }; + try std.testing.expectEqual(@as(?u64, 150), readVarint(&ok, &i)); + try std.testing.expectEqual(@as(usize, 2), i); + // Truncated (continuation bit set, no follow-on byte). + i = 0; + const trunc = [_]u8{0x80}; + try std.testing.expect(readVarint(&trunc, &i) == null); +} + +test "LiveStream: create/destroy lifecycle, keyless" { + const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null); + try std.testing.expect(!s.isRunning()); + try std.testing.expect(s.api_key == null); + s.destroy(); // testing.allocator would flag any leak +} + +test "LiveStream: create dupes and frees an api key" { + // Yahoo ignores the key, but the dupe/free branches must be exercised. + const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, "secret-token"); + try std.testing.expect(s.api_key != null); + try std.testing.expectEqualStrings("secret-token", s.api_key.?); + s.destroy(); +} + +test "LiveStream: serverMessage stores prices; snapshotInto matches case-insensitively" { + const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null); + defer s.destroy(); + + try s.serverMessage( + \\{"type":"pricing","message":"CgRBQVBMFUgBjUMY4JKAz+JnKgNOTVMwCDgBRQisH79InK3UKmUAj+K/sAFQ2AEE"} + ); + try s.serverMessage( + \\{"type":"pricing","message":"CgNTUFkVUhg5RBiwooDP4mcqA1BDWDAUOAFF9v3HP0jQ+bonZYA9NkGwAVDYAQQ="} + ); + // A non-pricing frame is ignored (early return, no entry, no crash). + try s.serverMessage("not a frame"); + + var out = std.StringHashMap(f64).init(std.testing.allocator); + defer out.deinit(); + // lowercase request hits the UPPERCASED stored key; an unseen symbol + // is absent; an over-long symbol is skipped without overflowing. + const syms = [_][]const u8{ "aapl", "spy", "TSLA", "THIS_TICKER_IS_TOO_LONG_TO_FIT_IN_BUF" }; + s.snapshotInto(&syms, &out); + try std.testing.expectApproxEqAbs(@as(f64, 282.010009765625), out.get("aapl").?, 0.0001); + try std.testing.expectApproxEqAbs(@as(f64, 740.3800048828125), out.get("spy").?, 0.0001); + try std.testing.expect(out.get("TSLA") == null); +} + +test "LiveStream: serverMessage overwrites an existing ticker (found_existing path)" { + const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null); + defer s.destroy(); + const frame = + \\{"type":"pricing","message":"CgRBQVBMFUgBjUMY4JKAz+JnKgNOTVMwCDgBRQisH79InK3UKmUAj+K/sAFQ2AEE"} + ; + try s.serverMessage(frame); + try s.serverMessage(frame); // second time reuses the existing key + var out = std.StringHashMap(f64).init(std.testing.allocator); + defer out.deinit(); + const syms = [_][]const u8{"AAPL"}; + s.snapshotInto(&syms, &out); + try std.testing.expectEqual(@as(usize, 1), out.count()); +} + +test "LiveStream: yahoo transportConfig carries the Origin header and version path" { + const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null); + defer s.destroy(); + const cfg = s.transportConfig(); + try std.testing.expectEqualStrings(yahoo_host, cfg.host); + try std.testing.expectEqualStrings(yahoo_path, cfg.path); + try std.testing.expectEqual(@as(u16, 443), cfg.port); + try std.testing.expect(std.mem.indexOf(u8, cfg.headers, "Origin: https://finance.yahoo.com") != null); + try std.testing.expect(std.mem.indexOf(u8, cfg.headers, "Host: " ++ yahoo_host) != null); +} + +test "buildSubscribe: yahoo message for one and many tickers" { + const a = std.testing.allocator; + var buf: std.ArrayList(u8) = .empty; + defer buf.deinit(a); + + try buildSubscribe(.yahoo, a, &.{"SPY"}, &buf); + try std.testing.expectEqualStrings("{\"subscribe\":[\"SPY\"]}", buf.items); + + buf.clearRetainingCapacity(); + try buildSubscribe(.yahoo, a, &.{ "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 std.testing.expectEqualStrings("{\"subscribe\":[]}", buf.items); +} diff --git a/src/service.zig b/src/service.zig index dc82b75..ae9daa7 100644 --- a/src/service.zig +++ b/src/service.zig @@ -41,6 +41,10 @@ const performance = @import("analytics/performance.zig"); const http = @import("net/http.zig"); const atomic = @import("atomic.zig"); const market = @import("market.zig"); +// File-as-struct: the import binds the whole file struct, so its inline +// tests are discovered by the test runner (see AGENTS.md "Test +// discovery"). +const LiveStream = @import("net/LiveStream.zig"); // ── Wall-clock policy ──────────────────────────────────────── // @@ -328,6 +332,12 @@ pub const DataService = struct { wikidata: ?Wikidata = null, edgar: ?Edgar = null, + /// Live-price websocket stream (push-based intraday quotes), lazily + /// created by `startLiveStream`. Heap-owned (the stream's background + /// thread captures the pointer, so it must be stable). Torn down by + /// `stopLiveStream` / `deinit`. + live_stream: ?*LiveStream = null, + /// Test-only guard: when true, any code path that would touch /// the network panics with a clear message. Used by offline-mode /// tests to verify that `FetchOptions.skip_network = true` @@ -379,6 +389,7 @@ pub const DataService = struct { } pub fn deinit(self: *DataService) void { + if (self.live_stream) |s| s.destroy(); if (self.td) |*td| td.deinit(); if (self.pg) |*pg| pg.deinit(); if (self.fmp) |*fmp| fmp.deinit(); @@ -737,6 +748,54 @@ 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(). + + /// Start (or ensure running) the live-price stream for `symbols`. + /// Idempotent while running: the symbol set is fixed for the + /// stream's lifetime, so to change it call `stopLiveStream` first. + /// `symbols` is copied by the stream, so the caller need not keep it + /// alive. + /// + /// 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. + 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; + } + self.live_stream.?.start(symbols) catch |err| { + log.warn("startLiveStream: {t}", .{err}); + return DataError.FetchFailed; + }; + } + + /// Stop and tear down the live stream if running. + pub fn stopLiveStream(self: *DataService) void { + if (self.live_stream) |s| { + s.destroy(); + self.live_stream = null; + } + } + + /// True if the live stream is currently running. + pub fn liveStreamActive(self: *DataService) bool { + return if (self.live_stream) |s| s.isRunning() else false; + } + + /// Copy the latest streamed prices for `symbols` into `out` (keys + /// borrow `symbols`, mirroring a one-shot quote map). No-op when the + /// stream isn't running; symbols not yet seen are left absent so the + /// caller falls back to the last cached close. + pub fn liveStreamSnapshot(self: *DataService, symbols: []const []const u8, out: *std.StringHashMap(f64)) void { + if (self.live_stream) |s| s.snapshotInto(symbols, out); + } + /// Fetch daily candles for a symbol (10+ years for trailing returns). /// Checks cache first; fetches from Tiingo (primary) or Yahoo (fallback) if stale/missing. /// Uses incremental updates: when the cache is stale, only fetches @@ -3118,6 +3177,28 @@ test "DataService init/deinit lifecycle" { try std.testing.expect(svc.tg == null); } +test "DataService live-stream accessors no-op when no stream is running" { + const allocator = std.testing.allocator; + const config = Config{ + .cache_dir = "/tmp/zfin-test-cache", + }; + var svc = DataService.init(std.testing.io, allocator, config); + defer svc.deinit(); + + // No stream created yet: active is false, snapshot leaves `out` + // empty, and stop is a safe no-op (must not crash on null). + try std.testing.expect(!svc.liveStreamActive()); + + var out = std.StringHashMap(f64).init(allocator); + defer out.deinit(); + const syms = [_][]const u8{ "SPY", "AAPL" }; + svc.liveStreamSnapshot(&syms, &out); + try std.testing.expectEqual(@as(usize, 0), out.count()); + + svc.stopLiveStream(); + try std.testing.expect(!svc.liveStreamActive()); +} + test "DataService store helper creates valid store" { const allocator = std.testing.allocator; const config = Config{