From 051ab35975c7fca345e0363846dab79d78ebf569 Mon Sep 17 00:00:00 2001 From: Emil Lerch Date: Mon, 29 Jun 2026 20:21:21 -0700 Subject: [PATCH] support live streaming on portfolio tab --- src/tui/portfolio_tab.zig | 186 +++++++++++++++++++++++++++++++++++++- 1 file changed, 184 insertions(+), 2 deletions(-) diff --git a/src/tui/portfolio_tab.zig b/src/tui/portfolio_tab.zig index 761e466..87113e5 100644 --- a/src/tui/portfolio_tab.zig +++ b/src/tui/portfolio_tab.zig @@ -140,6 +140,11 @@ pub const Action = enum { /// on the same symbol; re-target when open and cursor is on /// a different symbol. toggle_overlay, + /// Toggle opt-in live streaming: subscribe all held + watchlist + /// symbols to the websocket feed and re-value the portfolio on + /// each tick. Off by default. See the streaming helpers below for + /// the scope of what updates live. + toggle_stream, }; // ── Tab-private state ───────────────────────────────────────── @@ -256,6 +261,17 @@ pub const State = struct { /// `handleMouse` / `drawContent` / `drawStatusBar` check /// this and route accordingly. modal: Modal = .none, + + /// Opt-in live streaming (default off). Toggled with + /// `toggle_stream`. While true the App polls `tick`, which snapshots + /// the streamed prices and calls `PortfolioData.revalue` to refresh + /// the position table + totals. The stream is owned by + /// `DataService`; torn down on `deactivate` / `deinit`. + streaming: bool = false, + /// Owned, duped symbols subscribed to the stream (held + watchlist), + /// captured when streaming starts so the per-tick snapshot has + /// stable key strings. Freed (and emptied) when streaming stops. + stream_syms: [][]const u8 = &.{}, }; // ── Tab framework contract ──────────────────────────────────── @@ -275,6 +291,7 @@ pub const meta: framework.TabMeta(Action) = .{ // (or re-targets) the symbol-info overlay for the // cursor row's symbol. .{ .action = .toggle_overlay, .key = .{ .codepoint = 'K' } }, + .{ .action = .toggle_stream, .key = .{ .codepoint = 'L' } }, }, .action_labels = std.enums.EnumArray(Action, []const u8).init(.{ .expand_collapse = "Expand/collapse position", @@ -285,6 +302,7 @@ pub const meta: framework.TabMeta(Action) = .{ .clear_account_filter = "Clear account filter", .select_symbol = "Select symbol", .toggle_overlay = "Show symbol details", + .toggle_stream = "Toggle live stream", }), .status_hints = &.{ .sort_col_prev, @@ -292,6 +310,7 @@ pub const meta: framework.TabMeta(Action) = .{ .sort_reverse, .open_account_picker, .toggle_overlay, + .toggle_stream, }, }; @@ -305,6 +324,7 @@ pub const tab = struct { } pub fn deinit(state: *State, app: *App) void { + if (state.streaming) stopStreaming(state, app); state.rows.deinit(app.allocator); if (state.prepared_options) |*opts| opts.deinit(); if (state.prepared_cds) |*cds| cds.deinit(); @@ -328,7 +348,20 @@ pub const tab = struct { loadPortfolioData(state, app); } - pub const deactivate = framework.noopDeactivate(State); + /// Leaving the tab stops any live stream so it doesn't keep a + /// socket + thread alive in the background (and frees the single + /// shared `DataService` stream for another tab). Re-enabled by + /// toggling again on return. + pub fn deactivate(state: *State, app: *App) void { + if (state.streaming) stopStreaming(state, app); + } + + /// Framework poll-tick gate: true only while streaming, so the App + /// keeps a 100ms tick armed to pull fresh prices and re-value. + pub fn wantsPollTick(state: *State, app: *App) bool { + _ = app; + return state.streaming; + } /// Disabled only when the TUI was opened on a specific symbol /// (`zfin AAPL`) with no portfolio loaded: there's nothing to show @@ -404,7 +437,26 @@ pub const tab = struct { loadPortfolioData(state, app); } - pub const tick = framework.noopTick(State); + /// Poll hook: while streaming, snapshot the latest streamed prices + /// for the subscribed symbols, re-value the portfolio against them + /// (cheap - no file I/O), and rebuild the row table. The historical- + /// return block and worker-backed data stay as of the last full + /// load; only the position table + totals reflect the live prices. + pub fn tick(state: *State, app: *App, frame: u64) void { + _ = frame; + if (!state.streaming or state.stream_syms.len == 0) return; + var overlay = std.StringHashMap(f64).init(app.allocator); + defer overlay.deinit(); + app.svc.liveStreamSnapshot(state.stream_syms, &overlay); + if (overlay.count() == 0) return; // no ticks yet; nothing to re-value + if (app.portfolio.revalue(app.today, &overlay)) { + // wall-clock required: stamp the live re-value instant so the + // portfolio footer's "(as of H:MM PM ET)" reflects the stream, + // not the last full load. + app.portfolio.live_quotes_at_s = std.Io.Timestamp.now(app.io, .real).toSeconds(); + loadPortfolioData(state, app); + } + } pub fn handleAction(state: *State, app: *App, action: Action) void { switch (action) { @@ -473,6 +525,20 @@ pub const tab = struct { }; app.toggleOverlay(sym); }, + .toggle_stream => { + if (state.streaming) { + stopStreaming(state, app); + app.setStatus("Live stream: off"); + } else if (app.portfolio.file == null) { + app.setStatus("No portfolio to stream"); + } else if (startStreaming(state, app)) |n| { + var buf: [64]u8 = undefined; + const msg = std.fmt.bufPrint(&buf, "Live stream: on ({d} symbols, Yahoo)", .{n}) catch "Live stream: on"; + app.setStatus(msg); + } else { + app.setStatus("Live stream: nothing to subscribe"); + } + }, } } @@ -759,6 +825,84 @@ fn mapIntent(th: theme.Theme, intent: fmt.StyleIntent) vaxis.Style { /// is `app.portfolio.loaded` which `ensurePortfolioDataLoaded` /// owns. Visiting portfolio after analysis pre-loaded the data /// will still rebuild the row list - cheap.) +// ── Live-stream helpers ─────────────────────────────────────── + +/// Gather held stock symbols + watchlist (deduped), dup them into +/// `state.stream_syms` (owned, so the per-tick snapshot has stable +/// keys), and start the live stream. Returns the subscribed symbol +/// count on success, or null when there's nothing to subscribe or the +/// stream failed to start (in which case `streaming` stays false). +fn startStreaming(state: *State, app: *App) ?usize { + const held_owned: ?[][]const u8 = if (app.portfolio.file) |pf| + (pf.stockSymbols(app.allocator) catch |err| ret: { + std.log.scoped(.tui).debug("stockSymbols for stream: {t}", .{err}); + break :ret null; + }) + else + null; + defer if (held_owned) |h| app.allocator.free(h); // outer slice; gather dups the strings + + const owned = gatherStreamSymbols( + app.allocator, + held_owned orelse &.{}, + app.watchlist orelse &.{}, + ) catch return null; + if (owned.len == 0) { + app.allocator.free(owned); + return null; + } + app.svc.startLiveStream(owned) catch |err| { + std.log.scoped(.tui).warn("portfolio startLiveStream: {t}", .{err}); + for (owned) |s| app.allocator.free(s); + app.allocator.free(owned); + return null; + }; + state.stream_syms = owned; + state.streaming = true; + return owned.len; +} + +/// Build the owned, deduped subscribe list: every held symbol, then any +/// watchlist symbol not already present (held wins). Each string is +/// duped with `allocator`; the caller owns the result and its strings. +/// Pure (allocator + two slices), so it's unit-testable without an App. +fn gatherStreamSymbols( + allocator: std.mem.Allocator, + held: []const []const u8, + watchlist: []const []const u8, +) ![][]const u8 { + var syms: std.ArrayList([]const u8) = .empty; + errdefer { + for (syms.items) |s| allocator.free(s); + syms.deinit(allocator); + } + for (held) |sym| { + try syms.append(allocator, try allocator.dupe(u8, sym)); + } + for (watchlist) |sym| { + var seen = false; + for (syms.items) |existing| { + if (std.mem.eql(u8, existing, sym)) { + seen = true; + break; + } + } + if (seen) continue; + try syms.append(allocator, try allocator.dupe(u8, sym)); + } + return syms.toOwnedSlice(allocator); +} + +/// Stop the live stream (if running) and free the subscribed symbols. +/// Idempotent. +fn stopStreaming(state: *State, app: *App) void { + app.svc.stopLiveStream(); + for (state.stream_syms) |s| app.allocator.free(s); + app.allocator.free(state.stream_syms); + state.stream_syms = &.{}; + state.streaming = false; +} + pub fn loadPortfolioData(state: *State, app: *App) void { // Summary is populated synchronously by pd.load; if it's // null here, no portfolio is loaded (welcome screen). @@ -2557,3 +2701,41 @@ test "ensureCursorVisible: zero visible height is a no-op for the lower bound" { // cursor_row = 0 < 5 -> scroll = 0. Lower-bound branch fires. try testing.expectEqual(@as(usize, 0), scroll); } + +test "gatherStreamSymbols: held first, watchlist deduped against held" { + const a = testing.allocator; + const held = [_][]const u8{ "AAPL", "SPY" }; + const watchlist = [_][]const u8{ "SPY", "QQQ", "AAPL", "VTI" }; + const got = try gatherStreamSymbols(a, &held, &watchlist); + defer { + for (got) |s| a.free(s); + a.free(got); + } + // Held first (AAPL, SPY), then only watchlist entries not already + // present (QQQ, VTI); SPY and AAPL are deduped. + try testing.expectEqual(@as(usize, 4), got.len); + try testing.expectEqualStrings("AAPL", got[0]); + try testing.expectEqualStrings("SPY", got[1]); + try testing.expectEqualStrings("QQQ", got[2]); + try testing.expectEqualStrings("VTI", got[3]); +} + +test "gatherStreamSymbols: empty held + watchlist yields empty list" { + const a = testing.allocator; + const got = try gatherStreamSymbols(a, &.{}, &.{}); + defer a.free(got); + try testing.expectEqual(@as(usize, 0), got.len); +} + +test "gatherStreamSymbols: watchlist-only when no holdings" { + const a = testing.allocator; + const watchlist = [_][]const u8{ "BTC-USD", "ETH-USD" }; + const got = try gatherStreamSymbols(a, &.{}, &watchlist); + defer { + for (got) |s| a.free(s); + a.free(got); + } + try testing.expectEqual(@as(usize, 2), got.len); + try testing.expectEqualStrings("BTC-USD", got[0]); + try testing.expectEqualStrings("ETH-USD", got[1]); +}