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

reduce need for split data, add load timing data for further research

Browse source
This commit is contained in:
Emil Lerch 2026-08-07 09:54:06 -07:00
parent 5e22e1dccc
commit 87a0447b59
Signed by: lobo
GPG key ID: A7B62D657EF764F8
3 changed files with 197 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
src/PortfolioData.zig
View file

@ -406,7 +406,7 @@ pub fn anchorPath(self: *const PortfolioData) ?[]const u8 {
/// Per-symbol cached candles. Blocks on the candles worker.
pub fn candles(self: *PortfolioData) ?*const std.StringHashMap([]const Candle) {
self.awaitWorker(&self.candles_future);
self.awaitWorkerTimed(&self.candles_future, "candles");
if (self.candles_data) |*m| return m;
return null;
}
@ -416,13 +416,13 @@ pub fn candles(self: *PortfolioData) ?*const std.StringHashMap([]const Candle) {
/// internally awaits the candles worker since snapshot
/// computation depends on candle data.
pub fn snapshots(self: *PortfolioData) ?[HistoricalPeriod.all.len]HistoricalSnapshot {
self.awaitWorker(&self.snapshots_future);
self.awaitWorkerTimed(&self.snapshots_future, "snapshots");
return self.snapshots_data;
}
/// Per-symbol cached dividends. Blocks on the dividends worker.
pub fn dividends(self: *PortfolioData) ?*const std.StringHashMap([]const Dividend) {
self.awaitWorker(&self.dividends_future);
self.awaitWorkerTimed(&self.dividends_future, "dividends");
if (self.dividends_data) |*m| return m;
return null;
}
@ -522,6 +522,40 @@ fn awaitWorker(self: *PortfolioData, fut: *?std.Io.Future(void)) void {
}
}
/// Is startup timing instrumentation requested?
///
/// Gated on an env var and emitted at `info` rather than `debug`, because the
/// release builds people actually install compile debug logging out entirely -
/// which made the first version of this instrumentation invisible in precisely
/// the build that was slow. Off by default, so a normal run stays silent.
fn timingOn(self: *PortfolioData) bool {
const em = self.svc.config.environ_map orelse return false;
const v = em.get("ZFIN_TIMING") orelse return false;
return v.len > 0 and !std.mem.eql(u8, v, "0");
}
fn timing(self: *PortfolioData, comptime fmt: []const u8, args: anytype) void {
if (!self.timingOn()) return;
// `log.info`, not `log.debug`: release builds compile debug logging out, and
// an instrument that vanishes in the build people install is no instrument.
// Gated on the env var, so a normal run stays silent either way.
log.info("timing: " ++ fmt, args);
}
/// Same, but reports how long the caller was blocked.
///
/// These awaits are the wait the user actually feels: `load` returns as soon as
/// prices are in, then the first render calls a blocking accessor and sits
/// there. `load`-phase timing cannot see any of it, which is why a run that
/// felt like ten seconds reported two.
fn awaitWorkerTimed(self: *PortfolioData, fut: *?std.Io.Future(void), name: []const u8) void {
if (fut.* == null) return;
const t0 = std.Io.Timestamp.now(self.io, .real);
self.awaitWorker(fut);
const ms = @divFloor(std.Io.Timestamp.now(self.io, .real).nanoseconds - t0.nanoseconds, std.time.ns_per_ms);
if (ms >= 1) self.timing("blocked on {s} worker: {d}ms", .{ name, ms });
}
// ── Loading ──────────────────────────────────────────────────
/// Overlay live quotes onto the candle-derived price maps before the
@ -628,6 +662,24 @@ pub fn load(
}
self.paths = paths_dup;
// Phase timing for everything that runs BEFORE the TUI paints. These
// phases are synchronous by necessity - positions cannot be aggregated
// until lots are split-adjusted, and the summary cannot be built until
// prices are in - so any network work here is time the user spends looking
// at an unchanged terminal. When that happens (typically the first run of a
// day, as candle and split TTLs lapse), these lines are what identify which
// phase to blame.
const t_start = std.Io.Timestamp.now(self.io, .real);
var t_mark = t_start;
const Phase = struct {
fn done(pd: *PortfolioData, mark: *std.Io.Timestamp, name: []const u8) void {
const now = std.Io.Timestamp.now(pd.io, .real);
const ms = @divFloor(now.nanoseconds - mark.nanoseconds, std.time.ns_per_ms);
mark.* = now;
if (ms >= 1) pd.timing("load phase {s}: {d}ms", .{ name, ms });
}
};
// ── Parse portfolio files ────────────────────────────────
const gpa = self.arena.child_allocator;
const loaded = portfolio_loader.loadPortfolioFromPaths(self.io, gpa, self.paths, today) orelse
@ -645,6 +697,7 @@ pub fn load(
if (loaded.resolved_paths) |rp| rp.deinit();
const pf = self.file.?;
Phase.done(self, &t_mark, "parse");
// ── Compute symbols + positions ──────────────────────────
// Symbols first: the split corpus is fetched per-symbol.
@ -655,6 +708,7 @@ pub fn load(
// TUI's positions and valuation carry effective shares. No-op
// unless `splits_current_through` is set in metadata.srf.
portfolio_loader.enrichLotsSplits(self.svc, gpa, pf.lots, syms, self.paths[0], today, self.fetch_options);
Phase.done(self, &t_mark, "splits");
const positions = pf.positions(today, gpa) catch return error.NoAllocations;
defer gpa.free(positions);
@ -714,6 +768,7 @@ pub fn load(
sym_cb,
);
defer load_all.deinit();
Phase.done(self, &t_mark, "prices");
// Split the unified prices map: portfolio symbols go into
// `prices` (for summary build below); watchlist symbols
@ -811,6 +866,12 @@ pub fn load(
break :blk syms_arena;
};
Phase.done(self, &t_mark, "summary");
{
const total_ms = @divFloor(std.Io.Timestamp.now(self.io, .real).nanoseconds - t_start.nanoseconds, std.time.ns_per_ms);
self.timing("load TOTAL (synchronous, before first paint): {d}ms", .{total_ms});
}
self.candles_future = self.io.async(candlesWorker, .{ self, candles_to_load, opts.delays.candles_ms });
self.snapshots_future = self.io.async(snapshotsWorker, .{ self, today, positions_arena, opts.delays.snapshots_ms });
self.dividends_future = self.io.async(dividendsWorker, .{ self, opts.delays.dividends_ms });
@ -922,11 +983,18 @@ pub fn revalue(self: *PortfolioData, today: Date, overlay: *const std.StringHash
/// cancel the candles future - otherwise we'd race `cancel`
/// against `await` on the same future.
pub fn cancelLoad(self: *PortfolioData) void {
// Cancel DEPENDENCIES BEFORE DEPENDENTS. `snapshotsWorker` opens by
// draining `candles_future`, and `awaitWorker` is a plain await, not a
// cancellable wait - so cancelling snapshots first blocks until the candle
// worker finishes on its own. On the first run of a day every symbol's
// latest candle has expired, which turned quitting the TUI into a wait for
// the whole price refresh. Cancelling candles first lets its per-symbol
// cancel check fire, which unblocks the snapshots await immediately.
if (self.candles_future) |*f| _ = f.cancel(self.io);
self.candles_future = null;
if (self.snapshots_future) |*f| _ = f.cancel(self.io);
self.snapshots_future = null;
self.snapshots_data = null;
if (self.candles_future) |*f| _ = f.cancel(self.io);
self.candles_future = null;
if (self.dividends_future) |*f| _ = f.cancel(self.io);
self.dividends_future = null;
self.dividends_data = null;
@ -978,7 +1046,12 @@ fn snapshotsWorker(self: *PortfolioData, as_of: Date, positions: []const zfin.Po
// Snapshots depend on the candles map. Drain the candles
// future first; if it was canceled (or never produced
// data), the map is null and we abort.
self.awaitWorker(&self.candles_future);
self.awaitWorkerTimed(&self.candles_future, "candles (from snapshots worker)");
// The await above returns as soon as candles is cancelled, which is the
// point of cancelling it first - but that means arriving here mid-teardown
// is normal. Bail rather than spending the compute on a result nobody will
// read.
self.io.checkCancel() catch return;
const candle_map = self.candles_data orelse return;
const summary_ref = self.summary orelse return;

68
src/portfolio_loader.zig
View file

@ -576,7 +576,37 @@ pub fn enrichLotsSplits(
defer freeCutovers(allocator, &cutovers);
if (cutovers.count() == 0) return; // nothing opted in -> skip the corpus fetch
var corpus = svc.loadAllSplits(allocator, syms, opts);
// Fetch splits ONLY for the opted-in symbols. `enrichSplits` looks the
// corpus up strictly by a symbol that is already in `cutovers`, so a split
// record for any other symbol is fetched and then discarded.
//
// That waste is not free and it is not quiet in the way you might hope: it
// is a provider round trip per cache miss, it runs synchronously inside
// `PortfolioData.load` BEFORE the progress UI exists, and split TTLs are
// 14 days with jitter - so on whatever day a few lapse, the TUI sits there
// with no output before first paint. Narrowing the corpus to the opt-in set
// cuts that from every held symbol to the handful that asked for it.
//
// Borrowing `cutovers`' keys is safe: `corpus` is torn down by the defer
// below, which runs before `freeCutovers` above it.
var wanted: std.ArrayList([]const u8) = .empty;
defer wanted.deinit(allocator);
{
var it = cutovers.keyIterator();
while (it.next()) |k| {
// Only symbols actually held - `syms` is the held set, and a
// metadata row can name a symbol the portfolio no longer holds.
for (syms) |s| {
if (std.mem.eql(u8, s, k.*)) {
wanted.append(allocator, k.*) catch return;
break;
}
}
}
}
if (wanted.items.len == 0) return;
var corpus = svc.loadAllSplits(allocator, wanted.items, opts);
defer {
var it = corpus.valueIterator();
while (it.next()) |v| allocator.free(v.*);
@ -1006,6 +1036,42 @@ test "applySplitAdjustment: end-to-end enriches loaded positions from seeded cac
try testing.expectApproxEqAbs(@as(f64, 4000), loaded.positions[0].total_cost, 0.001);
}
test "enrichLotsSplits: a cutover for an unheld symbol does not defeat the held one" {
const allocator = testing.allocator;
const io = testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
// The corpus is now narrowed to the opt-in set intersected with what is
// actually held, because `enrichSplits` can only ever consult a symbol
// present in `cutovers` - fetching the rest was a provider round trip per
// cache miss, spent before the TUI paints, on data that got discarded.
//
// This pins the intersection: metadata legitimately carries rows for
// symbols the portfolio has since sold, and one of those must not stop the
// held symbol from being enriched.
var path_buf: [std.fs.max_path_bytes]u8 = undefined;
const dir = try seedSplitFixture(io, &tmp, &path_buf, "symbol::ZZZZNOTHELD,splits_current_through::2020-01-01\n" ++
"symbol::NVDA,splits_current_through::2024-01-01\n");
var svc = zfin.DataService.init(io, allocator, .{ .cache_dir = dir });
defer svc.deinit();
const pf_path = try std.fs.path.join(allocator, &.{ dir, "zfintest_split_pf.srf" });
defer allocator.free(pf_path);
const paths = try allocator.dupe([]const u8, &.{pf_path});
defer allocator.free(paths);
var loaded = loadPortfolioFromPaths(io, allocator, paths, zfin.Date.fromYmd(2026, 1, 1)) orelse
return error.TestUnexpectedResult;
defer loaded.deinit(allocator);
applySplitAdjustment(&svc, allocator, &loaded, zfin.Date.fromYmd(2026, 1, 1), .{ .skip_network = true });
// NVDA still enriched from the seeded cache despite the unheld row.
try testing.expectApproxEqAbs(@as(f64, 10.0), loaded.portfolio.lots[0].split_factor, 0.001);
try testing.expectApproxEqAbs(@as(f64, 1000), loaded.positions[0].shares, 0.001);
}
test "applySplitAdjustment: no cutover is a no-op (raw shares preserved)" {
const allocator = testing.allocator;
const io = testing.io;

53
src/service.zig
View file

@ -2239,13 +2239,36 @@ pub const DataService = struct {
// against the provider. A full wipe + re-download from scratch
// is reserved for `cache clear`.
// Sub-phase timing. `loadAllPrices` is the whole of the pre-paint wait,
// and its three phases fail in completely different ways: the cache
// scan is local I/O over the largest files in the cache, the server
// sync is one parallel round trip, and the provider fallback is
// rate-limited to a handful of requests per minute. A run that feels
// hung needs to say which of those it was sitting in - "prices: 90s"
// on its own does not distinguish a slow disk from a dead server.
const t0 = std.Io.Timestamp.now(self.io, .real);
var mark = t0;
const timing_on = blk: {
const em = self.config.environ_map orelse break :blk false;
const v = em.get("ZFIN_TIMING") orelse break :blk false;
break :blk v.len > 0 and !std.mem.eql(u8, v, "0");
};
const Sub = struct {
fn done(on: bool, io: std.Io, m: *std.Io.Timestamp, name: []const u8, n: usize) void {
const now = std.Io.Timestamp.now(io, .real);
const ms = @divFloor(now.nanoseconds - m.nanoseconds, std.time.ns_per_ms);
m.* = now;
if (on and ms >= 1) log.info("timing: loadAllPrices {s}: {d}ms ({d} symbols)", .{ name, ms, n });
}
};
// Phase 1: Check local cache (fast path)
var needs_fetch: std.ArrayList([]const u8) = .empty;
defer needs_fetch.deinit(self.allocator);
if (aggregate_progress) |p| p.emit(0, total_count, .cache_check);
for (all_symbols.items) |sym| {
for (all_symbols.items, 0..) |sym, i| {
if (!opts.force_refresh and self.isCandleCacheFresh(sym)) {
if (self.getCachedLastClose(sym)) |close| {
result.prices.put(sym, close) catch |err| log.warn("loadAllPrices cache-hit put({s}): {t}", .{ sym, err });
@ -2255,14 +2278,22 @@ pub const DataService = struct {
} else {
needs_fetch.append(self.allocator, sym) catch |err| log.warn("loadAllPrices needs_fetch append({s}): {t}", .{ sym, err });
}
// Report inside the loop, not just at the ends. This phase reads
// and validates every symbol's candle file - the bulk of the cache
// by size - so on a portfolio of any size it is hundreds of
// milliseconds of the pre-paint wait. Emitting only before and
// after left the display frozen for all of it, which reads as a
// hang rather than as work.
if (aggregate_progress) |p| p.emit(i + 1, total_count, .cache_check);
}
if (aggregate_progress) |p| p.emit(result.cached_count, total_count, .cache_check);
Sub.done(timing_on, self.io, &mark, "cache scan", total_count);
if (needs_fetch.items.len == 0) {
if (aggregate_progress) |p| p.emit(total_count, total_count, .complete);
return result;
}
if (timing_on) log.info("timing: loadAllPrices: {d} of {d} symbols need a fetch", .{ needs_fetch.items.len, total_count });
// Offline mode: skip server sync and provider fetch entirely.
// For symbols without a fresh cache, fall back to stale cache
@ -2300,7 +2331,15 @@ pub const DataService = struct {
}
}
Sub.done(timing_on, self.io, &mark, "server sync", needs_fetch.items.len);
// Phase 3: Sequential provider fallback for server failures
if (server_failures.items.len > 0) {
// The expensive one, and the reason this breakdown exists: the
// provider is rate limited, so this scales in minutes where the
// phases above scale in milliseconds.
if (timing_on) log.info("timing: loadAllPrices: {d} symbols fell through to the PROVIDER (rate limited)", .{server_failures.items.len});
}
if (server_failures.items.len > 0) {
if (aggregate_progress) |p| p.emit(
result.cached_count + result.server_synced_count,
@ -2317,6 +2356,8 @@ pub const DataService = struct {
);
}
Sub.done(timing_on, self.io, &mark, "provider fallback", server_failures.items.len);
if (aggregate_progress) |p| p.emit(total_count, total_count, .complete);
return result;
}
@ -3251,6 +3292,14 @@ pub const DataService = struct {
opts: FetchOptions,
) void {
for (syms) |sym| {
// Per symbol, not once before the loop. Every cache miss here is a
// provider round trip under a 4-request/minute budget, so a batch
// whose TTLs happened to lapse together runs for minutes. The
// dividends worker is cancelled on TUI teardown
// (`PortfolioData.cancelLoad`), and cancelling waits for the
// worker - so without a check inside the loop, quitting blocked
// until the whole batch drained.
self.io.checkCancel() catch return;
const fr = self.getDividends(sym, opts) catch continue;
fr.deinit();
}