lobo/zfin
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

support live streaming on portfolio tab

Browse source
This commit is contained in:
Emil Lerch 2026-06-29 20:21:21 -07:00
parent eadba4696c
commit 051ab35975
Signed by: lobo
GPG key ID: A7B62D657EF764F8
1 changed files with 184 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

186
src/tui/portfolio_tab.zig
View file

@ -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]);
}