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

computation of the overall portfolio summary in live mode

Browse source
This commit is contained in:
Emil Lerch 2026-06-29 19:35:41 -07:00
parent b2128bfcdd
commit 123bd3eb46
Signed by: lobo
GPG key ID: A7B62D657EF764F8
1 changed files with 168 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

168
src/PortfolioData.zig
View file

@ -293,6 +293,29 @@ live_quotes_at_s: ?i64 = null,
/// render). Allocated in pd's arena.
watchlist_prices: ?std.StringHashMap(f64) = null,
// ── Re-value (live-streaming) state ───────────────────────────
//
// Captured by `load` and reused by `revalue` for the cheap
// price-overlay re-value used during live streaming: apply streamed
// prices over the base candle-close prices and recompute only the
// summary - no file re-read, no worker respawn. Both fields live in
// the main `arena` (reaped on reload), so they are valid only between
// loads; `revalue` bails when they are empty/null.
/// Parsed positions from the last load (shallow arena copy; symbol
/// slices point into `file`, which outlives them). Empty before the
/// first load.
revalue_positions: []const zfin.Position = &.{},
/// Base (candle-close) prices from the last load, keyed by held symbol
/// with arena-duped keys. Null before the first load. A `revalue`
/// overlays streamed prices onto a copy of this.
revalue_base_prices: ?std.StringHashMap(f64) = null,
/// Dedicated arena for `revalue`'s working price map + recomputed
/// summary, reset at the start of each call so a long streaming
/// session can't grow memory unbounded. Lives across reloads; released
/// on `deinit`.
revalue_arena: ArenaAllocator,
// ── Async data (access via methods) ──────────────────────────
//
// Each datum has a (future, data) pair. The future is what
@ -327,6 +350,7 @@ pub fn init(opts: InitOptions) PortfolioData {
return .{
.arena = .init(opts.gpa),
.candles_arena = .init(opts.gpa),
.revalue_arena = .init(opts.gpa),
.io = opts.io,
.svc = opts.svc,
};
@ -341,6 +365,7 @@ pub fn deinit(self: *PortfolioData) void {
// slice values - single bulk free is enough; no need to
// walk the map.
self.candles_arena.deinit();
self.revalue_arena.deinit();
self.arena.deinit();
self.* = undefined;
}
@ -558,6 +583,11 @@ pub fn load(
self.file = null;
_ = self.arena.reset(.retain_capacity);
self.paths = &.{};
// The reset above reaped the previous load's re-value state (it
// lived in this arena). Null it now so a stray `revalue` between
// here and the recapture below can't read freed memory.
self.revalue_positions = &.{};
self.revalue_base_prices = null;
self.summary = null;
self.latest_quote_date = null;
self.live_prices_applied = false;
@ -716,6 +746,14 @@ pub fn load(
}
self.watchlist_prices = wp;
// Capture re-value state for live streaming BEFORE buildFallbackPrices
// mutates `prices` (so the base is the candle-close + any startup
// overlay, without the manual/avg-cost fallbacks that revalue
// re-derives each call). Keys are duped into the arena since the
// load_all-borrowed keys don't outlive this function.
self.revalue_positions = dupePositions(arena_alloc, positions);
self.revalue_base_prices = dupePricesArena(arena_alloc, &prices);
// ── Build summary ─────────────────────────────────────────
var manual_price_set = zfin.valuation.buildFallbackPrices(arena_alloc, pf.lots, positions, &prices) catch
return error.OutOfMemory;
@ -797,6 +835,56 @@ pub fn reload(self: *PortfolioData, today: Date, opts: LoadOptions) LoadError!Lo
return self.load(saved, today, opts);
}
/// Cheap live re-value: apply a streamed-price `overlay` over the base
/// (candle-close) prices captured by the last `load`, then recompute
/// ONLY the summary. No file I/O, no worker respawn - safe to call on a
/// sub-second tick while streaming.
///
/// `overlay` maps held symbol -> latest price; its keys must match the
/// summary's price keys (held symbols), and non-matching keys are
/// ignored. Symbols absent from the overlay keep their base price, so
/// non-streaming assets (mutual funds, CDs, cash) stay put.
///
/// Returns true when the summary was recomputed; false (a no-op) before
/// the first successful load, or if the previous load captured no
/// re-value state. On any internal failure the previous summary is left
/// intact and false is returned.
pub fn revalue(self: *PortfolioData, today: Date, overlay: *const std.StringHashMap(f64)) bool {
const pf = self.file orelse return false;
const base = if (self.revalue_base_prices) |*b| b else return false;
const positions = self.revalue_positions;
if (positions.len == 0) return false;
// Reset bounds memory across a long streaming session: each call
// reuses the same arena for its working price map and the new
// summary; the previous call's summary is reaped here.
_ = self.revalue_arena.reset(.retain_capacity);
const ra = self.revalue_arena.allocator();
// Working prices = base candle-close prices, with the streamed
// overlay applied over matching held symbols.
var prices = std.StringHashMap(f64).init(ra);
var overrides: usize = 0;
var bit = base.iterator();
while (bit.next()) |e| {
const px = if (overlay.get(e.key_ptr.*)) |live_px| blk: {
overrides += 1;
break :blk live_px;
} else e.value_ptr.*;
prices.put(e.key_ptr.*, px) catch return false;
}
const manual_price_set = zfin.valuation.buildFallbackPrices(ra, pf.lots, positions, &prices) catch return false;
const sum = zfin.valuation.portfolioSummary(today, ra, pf, positions, prices, manual_price_set) catch return false;
if (sum.allocations.len == 0) return false;
self.summary = sum;
// Flip the "as of" wording when at least one held position was
// re-priced from the live overlay (mirrors the load-overlay path).
self.live_prices_applied = overrides > 0;
return true;
}
/// Cancel any in-flight load and pending background workers.
/// Safe to call at any time including when nothing is in-flight.
/// After cancel, snapshots / dividends / account_map data is
@ -929,6 +1017,20 @@ fn dupePositions(arena: Allocator, src: []const zfin.Position) []const zfin.Posi
return dup;
}
/// Copy a price map into `arena` with arena-duped keys, so it outlives
/// the (load_all-borrowed) source keys. Used to capture the base prices
/// for `revalue`. Best-effort: a failed entry is skipped (a missing
/// base price just degrades that position to its fallback in revalue).
fn dupePricesArena(arena: Allocator, src: *std.StringHashMap(f64)) std.StringHashMap(f64) {
var out = std.StringHashMap(f64).init(arena);
var it = src.iterator();
while (it.next()) |e| {
const k = arena.dupe(u8, e.key_ptr.*) catch continue;
out.put(k, e.value_ptr.*) catch continue;
}
return out;
}
/// Adapter that captures the first 8 failed symbols into pd's
/// arena while still forwarding to the caller's progress callback.
const FailCapture = struct {
@ -1378,3 +1480,69 @@ test "PortfolioData.candles_data: soft load preserves the candle map" {
try testing.expect(pd.candles_data != null);
try testing.expect(pd.candles_data.?.contains("VTI"));
}
test "PortfolioData.revalue: overlays streamed prices and recomputes the total" {
var svc: DataService = .{
.allocator = testing.allocator,
.io = testing.io,
.config = .{ .cache_dir = "./.tmp/zfin-pd-test-cache" },
};
var pd = PortfolioData.init(.{ .gpa = testing.allocator, .io = testing.io, .svc = &svc });
defer {
// `file` below is an empty stack Portfolio, not heap-owned; clear
// it so deinit doesn't try to free it.
pd.file = null;
pd.deinit();
}
var positions = [_]zfin.Position{
.{ .symbol = "AAPL", .shares = 10, .avg_cost = 150.0, .total_cost = 1500.0, .open_lots = 1, .closed_lots = 0, .realized_gain_loss = 0 },
.{ .symbol = "SPY", .shares = 5, .avg_cost = 400.0, .total_cost = 2000.0, .open_lots = 1, .closed_lots = 0, .realized_gain_loss = 0 },
};
pd.file = .{ .lots = &.{}, .allocator = testing.allocator };
pd.revalue_positions = &positions;
var base = std.StringHashMap(f64).init(testing.allocator);
defer base.deinit();
try base.put("AAPL", 200.0);
try base.put("SPY", 700.0);
pd.revalue_base_prices = base;
const today = Date.fromYmd(2026, 5, 8);
// No overlay: total = 10*200 + 5*700 = 5500; nothing overridden.
var no_overlay = std.StringHashMap(f64).init(testing.allocator);
defer no_overlay.deinit();
try testing.expect(pd.revalue(today, &no_overlay));
try testing.expectApproxEqAbs(@as(f64, 5500.0), pd.summary.?.total_value, 0.01);
try testing.expect(!pd.live_prices_applied);
// Overlay AAPL -> 210: total = 10*210 + 5*700 = 5600.
var overlay = std.StringHashMap(f64).init(testing.allocator);
defer overlay.deinit();
try overlay.put("AAPL", 210.0);
try testing.expect(pd.revalue(today, &overlay));
try testing.expectApproxEqAbs(@as(f64, 5600.0), pd.summary.?.total_value, 0.01);
try testing.expect(pd.live_prices_applied);
// A repeat call reuses the revalue arena (memory stays bounded) and
// yields the same result - exercises the per-call arena reset.
try testing.expect(pd.revalue(today, &overlay));
try testing.expectApproxEqAbs(@as(f64, 5600.0), pd.summary.?.total_value, 0.01);
}
test "PortfolioData.revalue: no-op before any load" {
var svc: DataService = .{
.allocator = testing.allocator,
.io = testing.io,
.config = .{ .cache_dir = "./.tmp/zfin-pd-test-cache" },
};
var pd = PortfolioData.init(.{ .gpa = testing.allocator, .io = testing.io, .svc = &svc });
defer pd.deinit();
var overlay = std.StringHashMap(f64).init(testing.allocator);
defer overlay.deinit();
// No positions / base prices captured -> false, and summary stays null.
try testing.expect(!pd.revalue(Date.fromYmd(2026, 5, 8), &overlay));
try testing.expect(pd.summary == null);
}