Compare commits

...

8 commits

Author SHA1 Message Date
937822f9b0
ensure that new candles do not end up with an assumption of successful refresh
All checks were successful
Generic zig build / build (push) Successful in 5m28s
Generic zig build / publish-macos (push) Successful in 14s
Generic zig build / deploy (push) Successful in 19s
2026-08-11 11:59:22 -07:00
a91e5b0495
allow doctor to see stale cache items 2026-08-11 11:02:21 -07:00
418068a3d1
cleanup timing commit 2026-08-11 09:59:51 -07:00
79326cf41b
add a zfin cache stale command to find out of date candle data 2026-08-11 09:49:39 -07:00
2da30b2474
cacheKeys function in cache.zig 2026-08-10 16:35:29 -07:00
cd053b9a78
watchlist, benchmark symbols, and portfolio symbols refreshed in all the places 2026-08-10 15:22:45 -07:00
3dd72c2e3e
match transfers if moved cash->cash and stock purchased in same period 2026-08-08 17:09:16 -07:00
87a0447b59
reduce need for split data, add load timing data for further research 2026-08-07 09:54:06 -07:00
14 changed files with 2296 additions and 160 deletions

View file

@ -316,6 +316,11 @@ across the codebase so search-and-replace stays trivial):
*5678`
- Filenames: `Sample_IRA_1234.txt`, `Sample_IRA_5678.txt`,
`smpl_1234`, `smpl-ira-1234`
- Holder names: **`Riley` is an approved placeholder holder name** and
appears in existing fixtures (e.g. `<institution> Riley 401(k)`). It is
NOT a real family member. Do not "scrub" it, and do not ask whether it
needs scrubbing - it is already the placeholder. Reuse it rather than
inventing new person names.
- Account numbers: `1234`, `5678`, `9012`, `3456`, `7890`, or
alphanumeric like `Z123`, `Z111`, `Z222`. Do not use real
trailing-digit values from the user's actual accounts file.
@ -386,6 +391,13 @@ Two known classes:
approved placeholder vocabulary above - e.g. the real name
`Inherited IRA` is a substring of the sanctioned fixture value
`Sample Inherited IRA`.
- **Institution names and the placeholder holder name.** Broker brands
(Fidelity, Schwab, Vanguard, ...) appear throughout source legitimately -
in parser names, money-market symbol lists, and doc comments - and are
not PII. Because real account names combine an institution with a
holder, a token list built from `accounts.srf` will match all of those.
`Riley` likewise: it is the approved placeholder holder name, so
`<institution> Riley 401(k)` in a fixture is already scrubbed.
So the rule is: **inspect the context of every hit**, and confirm each
is either coincidental or a generic category before dismissing it. A

View file

@ -77,6 +77,17 @@ server_url: ?[]const u8 = null,
/// requests then go out unauthenticated, which is correct against an open
/// server or one still in its pre-enforcement soft cutover.
server_api_key: ?[]const u8 = null,
/// Is startup/phase timing instrumentation requested (`ZFIN_TIMING`)?
///
/// Resolved once here rather than re-parsed per call site. Two callers -
/// `PortfolioData` and `DataService.loadAllPrices` - each grew their own copy of
/// the same predicate in one commit, which is the drift this field prevents.
///
/// Any non-empty value except "0" enables it. Instrumentation is emitted at
/// `info`, not `debug`, because release builds compile debug logging out - an
/// instrument that vanishes in the build people install is no instrument - so
/// this flag is what keeps a normal run silent.
timing: bool = false,
cache_dir: []const u8,
cache_dir_owned: bool = false, // true when cache_dir was allocated via path.join
zfin_home: ?[]const u8 = null,
@ -101,6 +112,17 @@ environ_map: ?*const std.process.Environ.Map = null,
// Construction / teardown
/// Truthiness of an environment flag: absent, empty and "0" are off, anything
/// else is on.
///
/// Named rather than inlined so the one definition is also the one a test can
/// call. The inline version of this predicate got copied into two call sites in
/// a single commit.
fn envFlag(v: ?[]const u8) bool {
const s = v orelse return false;
return s.len > 0 and !std.mem.eql(u8, s, "0");
}
pub fn fromEnv(io: std.Io, allocator: std.mem.Allocator, environ_map: *const std.process.Environ.Map) @This() {
var self = @This(){
// SAFETY: assigned unconditionally below (the `cache_dir =
@ -143,6 +165,7 @@ pub fn fromEnv(io: std.Io, allocator: std.mem.Allocator, environ_map: *const std
self.user_email = self.resolve("ZFIN_USER_EMAIL");
self.server_url = self.resolve("ZFIN_SERVER");
self.server_api_key = self.resolve("ZFIN_SERVER_API_KEY");
self.timing = envFlag(self.resolve("ZFIN_TIMING"));
const env_cache = self.resolve("ZFIN_CACHE_DIR");
self.cache_dir = env_cache orelse blk: {
@ -1016,3 +1039,20 @@ test "expandGlob: missing directory returns null" {
const result = try expandGlob(io, allocator, "/zfin-test-no-such-dir-xyz", "*.srf", .home_relative);
try testing.expect(result == null);
}
test "envFlag: absent, empty and \"0\" are off; anything else is on" {
// Two callers - PortfolioData and DataService.loadAllPrices - each grew an
// identical copy of this predicate in a single commit. One definition, and
// the test calls it rather than restating it.
try testing.expect(!envFlag(null));
try testing.expect(!envFlag(""));
try testing.expect(!envFlag("0"));
for ([_][]const u8{ "1", "true", "yes", " ", "00" }) |on| {
try testing.expect(envFlag(on));
}
}
test "timing: defaults off so a normal run stays silent" {
const d: @This() = .{ .cache_dir = "unused" };
try testing.expect(!d.timing);
}

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,31 @@ fn awaitWorker(self: *PortfolioData, fut: *?std.Io.Future(void)) void {
}
}
fn timing(self: *PortfolioData, comptime fmt: []const u8, args: anytype) void {
// `Config.timing` resolves ZFIN_TIMING once at init; this used to re-parse
// the env var here, with an identical copy of the predicate in
// `DataService.loadAllPrices`.
if (!self.svc.config.timing) 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 +653,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 +688,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,36 +699,38 @@ 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);
// Compute watchlist symbols
//
// Union of caller-supplied watchlist syms (typically from
// a separate `watchlist.srf` file) and portfolio's own
// `.watch` lots, with held symbols (already in `syms`)
// excluded so we never double-fetch.
var watchlist_set = std.StringHashMap(void).init(gpa);
defer watchlist_set.deinit();
// Union of caller-supplied watchlist syms (typically from a separate
// `watchlist.srf` file) and portfolio's own `.watch` lots, with held
// symbols (already in `syms`) excluded so we never double-fetch.
//
// Shared with the CLI rather than reimplemented here. This was the third
// hand-rolled copy of the same set logic, and the copies had drifted: the
// CLI's omitted `watchlist.srf` entirely, so a watchlist-only symbol was
// displayed from whatever the cache held and never fetched. One
// implementation means that class of divergence cannot recur.
//
// Also now order-stable - watch lots then watchlist entries. The previous
// HashMap-iteration build made the "[5/28] Loading X" progress order vary
// between runs for no reason.
const watch_syms_list = pf.extraPriceSymbols(gpa, syms, opts.watchlist_syms) catch return error.OutOfMemory;
defer gpa.free(watch_syms_list);
// Lookup sets for splitting the unified price map below. Derived FROM the
// two symbol lists rather than rebuilt from the lots, so they cannot
// disagree with what was actually fetched.
var portfolio_set = std.StringHashMap(void).init(gpa);
defer portfolio_set.deinit();
for (syms) |s| portfolio_set.put(s, {}) catch return error.OutOfMemory;
for (opts.watchlist_syms) |sym| {
if (!portfolio_set.contains(sym)) watchlist_set.put(sym, {}) catch return error.OutOfMemory;
}
for (pf.lots) |lot| {
if (lot.security_type == .watch) {
const sym = lot.priceSymbol();
if (!portfolio_set.contains(sym)) watchlist_set.put(sym, {}) catch return error.OutOfMemory;
}
}
var watch_syms_list: std.ArrayList([]const u8) = .empty;
defer watch_syms_list.deinit(gpa);
{
var it = watchlist_set.keyIterator();
while (it.next()) |k| watch_syms_list.append(gpa, k.*) catch return error.OutOfMemory;
}
var watchlist_set = std.StringHashMap(void).init(gpa);
defer watchlist_set.deinit();
for (watch_syms_list) |s| watchlist_set.put(s, {}) catch return error.OutOfMemory;
// Fetch prices
//
@ -708,12 +754,13 @@ pub fn load(
var load_all = self.svc.loadAllPrices(
syms,
watch_syms_list.items,
watch_syms_list,
opts.fetch_options,
opts.aggregate_progress,
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 +858,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 +975,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 +1038,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;

555
src/cache/freshness.zig vendored Normal file
View file

@ -0,0 +1,555 @@
//! Corpus-wide candle-cache staleness detection.
//!
//! `market.candleFreshness` answers "is THIS symbol's newest bar overdue?" from
//! the clock alone. That is the right question for a fetch gate and the wrong
//! one for finding a symbol the cache has quietly frozen, because the clock
//! cannot distinguish three cases that look identical per-symbol: the bar isn't
//! published yet, the market was closed, or this symbol alone got left behind.
//!
//! Peers settle it. Symbols of the same `InstrumentKind` share a publication
//! schedule, so if twenty equities hold Friday's bar and three hold Thursday's,
//! those three are behind - no calendar reasoning required, and an un-modeled
//! closure cannot fool it because a closure moves every symbol together, taking
//! the peer maximum with them and leaving nothing behind it.
//!
//! That self-correction is why there is no calendar clamp on top. A first cut
//! suppressed findings whenever the group's own newest bar was itself overdue,
//! reasoning that a wholly-behind group meant a closure. It meant nothing of
//! the kind: it meant the corpus had not been refreshed yet that day, which is
//! the normal state, and the gate hid three genuinely-lagging symbols in live
//! data. "Is the corpus caught up?" and "is any symbol behind the rest?" are
//! independent questions, and only the second one names a symbol.
//!
//! The observed failure this exists for: three symbols served Thursday's close
//! all weekend while ~19 peers had Friday's. Each had been fetched, missed,
//! stamped `.lagging`, retried 30 minutes later, and by then sat just past the
//! 90-minute provider-lag grace - so `candleFreshness` said `.overdue`,
//! `staleCandleExpiry` routed them to the next session boundary, and the stale
//! bars became authoritative until Monday. The retry machinery worked as
//! designed; the premise was wrong.
//!
//! Pure: dates and `now_s` in, findings out. No I/O, no cache reads, no
//! fetches. The caller supplies the corpus.
const std = @import("std");
const Date = @import("../Date.zig");
const market = @import("../market.zig");
const cache = @import("store.zig");
/// One cached symbol, as the caller found it.
pub const Entry = struct {
symbol: []const u8,
kind: market.InstrumentKind,
/// Newest bar in the cache, or null when there is no candle meta at all -
/// never fetched, negative-cached, or a key that carries no candles (an
/// EDGAR CIK, say). Null entries are never reported stale: absence of
/// evidence is not a stale bar.
last_date: ?Date,
/// Is this symbol in the set zfin intends to keep fresh
/// (`Portfolio.fetchedSymbols`)? An untracked symbol can only ever fall
/// further behind, so reporting it as stale every run would be noise.
tracked: bool,
};
/// Days behind peers that ordinary lag can still explain. Beyond this, a finding
/// is reported as `far_behind` rather than `stale`.
///
/// Four days, and it is deliberately tight: the worst legitimate gap is a Friday
/// bar missed ahead of a Monday holiday, which leaves a symbol four calendar days
/// behind peers holding Tuesday's. More than that is not lag.
///
/// The split exists because the two have different answers. Inside the window a
/// refresh is the remedy. Outside it, a refresh has most likely already been
/// attempted and failed, so offering one as the fix sends the operator in
/// circles.
///
/// This module does NOT infer a cause, and callers must not either. The
/// candidates are many and none of them are visible from here: the ticker
/// changed, the provider dropped coverage, that one symbol is failing auth or
/// being rate-limited at the tail of a fetch, the symbol form is wrong for the
/// provider, whatever refreshes this cache never included it in its set, or it
/// stopped trading. Report the gap and let the operator look.
pub const max_normal_lag_days: i64 = 4;
/// A tracked symbol whose newest bar is behind its peers'.
pub const Finding = struct {
symbol: []const u8,
kind: market.InstrumentKind,
last_date: Date,
/// Newest bar held by any same-kind peer.
peer_date: Date,
/// Calendar days between the two. Calendar rather than trading days
/// because this is a human-facing magnitude, and the distinction between
/// "one session" and "six weeks" is what it has to convey.
days_behind: i64,
fn worstFirst(_: void, a: Finding, b: Finding) bool {
if (a.days_behind != b.days_behind) return a.days_behind > b.days_behind;
return std.mem.order(u8, a.symbol, b.symbol) == .lt;
}
};
/// Per-`InstrumentKind` reference state, so a caller can explain itself rather
/// than just listing symbols.
pub const GroupState = struct {
kind: market.InstrumentKind,
/// Newest bar across every dated entry of this kind; null when the kind
/// has no dated entries.
peer_date: ?Date,
/// Freshness of the group's OWN newest bar - context, NOT a gate.
///
/// Reported so a caller can say "the whole corpus is a session behind,
/// refresh everything" alongside any per-symbol findings. It deliberately
/// does not suppress those findings: an earlier version gated on
/// `.current` and thereby hid three genuinely-lagging symbols every time
/// the corpus as a whole had not been refreshed that day, which is most of
/// the time.
///
/// No clamp is needed, because peer comparison is already immune to the
/// case a clamp was meant to catch: a market closure moves every symbol
/// together, so the peer maximum moves with them and nothing sits behind
/// it. The two questions are independent - "is the corpus caught up?" and
/// "is any symbol behind the rest?" - and only the second one identifies a
/// symbol.
freshness: ?market.CandleFreshness,
/// Dated entries considered for this kind.
dated: usize,
/// Can this kind yield findings at all? Only that a peer exists to compare
/// against; with a single cached symbol there is no comparison to make.
pub fn conclusive(self: GroupState) bool {
return self.dated > 1;
}
};
pub const Report = struct {
/// Tracked symbols behind their peers by no more than
/// `max_normal_lag_days`, worst first. A refresh is the expected remedy.
stale: []Finding,
/// Tracked symbols further behind than lag explains, worst first. Reported
/// separately because a refresh probably is not the answer - see
/// `max_normal_lag_days`.
far_behind: []Finding,
/// Cached, dated, but tracked by nothing. Nothing will ever refresh these,
/// so they are not staleness - they are disk you may want back. Reported
/// separately precisely so they cannot dominate the stale list forever.
orphans: []const []const u8,
/// Tracked but with no cached bar at all. A different finding: not stale,
/// never fetched.
missing: []const []const u8,
groups: []GroupState,
pub fn deinit(self: Report, allocator: std.mem.Allocator) void {
allocator.free(self.stale);
allocator.free(self.far_behind);
allocator.free(self.orphans);
allocator.free(self.missing);
allocator.free(self.groups);
}
};
/// Read the cache and build the corpus `scan` consumes.
///
/// The I/O half, deliberately separate from `scan`: reading is a loop over the
/// store, deciding is the part worth testing exhaustively, and keeping them
/// apart is what lets every branch of the decision be exercised with fixed
/// dates and no filesystem.
///
/// `readCandleMeta` returning null IS the filter for non-candle keys. EDGAR
/// CIKs, negative-cached symbols and anything never fetched all arrive as
/// `last_date = null`, so none of them can be reported stale, and this function
/// needs to know nothing about what any of them are.
///
/// Symbol strings borrow from `keys`; only the slice is allocated.
pub fn collect(
allocator: std.mem.Allocator,
store: *cache.Store,
keys: []const []const u8,
tracked: *const std.StringHashMap(void),
) ![]Entry {
var out = try allocator.alloc(Entry, keys.len);
errdefer allocator.free(out);
for (keys, 0..) |key, i| {
const cm = store.readCandleMeta(key);
out[i] = .{
.symbol = key,
.kind = market.classify(key),
.last_date = if (cm) |m| m.meta.last_date else null,
.tracked = tracked.contains(key),
};
}
return out;
}
/// Classify `entries` as of `now_s`.
///
/// Symbol strings are borrowed from `entries`; only the slices are allocated.
pub fn scan(
allocator: std.mem.Allocator,
entries: []const Entry,
now_s: i64,
) !Report {
const kinds = [_]market.InstrumentKind{ .equity, .mutual_fund };
var groups = try allocator.alloc(GroupState, kinds.len);
errdefer allocator.free(groups);
for (kinds, 0..) |kind, i| {
var newest: ?Date = null;
var dated: usize = 0;
for (entries) |e| {
if (e.kind != kind) continue;
const d = e.last_date orelse continue;
dated += 1;
if (newest == null or newest.?.lessThan(d)) newest = d;
}
groups[i] = .{
.kind = kind,
.peer_date = newest,
// Context only (see GroupState.freshness). Reuses the fetch
// gate's own verdict rather than re-deriving the market calendar.
.freshness = if (newest) |n| market.candleFreshness(now_s, kind, n) else null,
.dated = dated,
};
}
var stale = std.ArrayList(Finding).empty;
errdefer stale.deinit(allocator);
var far_behind = std.ArrayList(Finding).empty;
errdefer far_behind.deinit(allocator);
var orphans = std.ArrayList([]const u8).empty;
errdefer orphans.deinit(allocator);
var missing = std.ArrayList([]const u8).empty;
errdefer missing.deinit(allocator);
for (entries) |e| {
const last = e.last_date orelse {
// No bar at all. Only worth saying when something tracks it.
if (e.tracked) try missing.append(allocator, e.symbol);
continue;
};
if (!e.tracked) {
try orphans.append(allocator, e.symbol);
continue;
}
const g = groups[if (e.kind == .equity) 0 else 1];
if (!g.conclusive()) continue;
const peer = g.peer_date.?;
if (!last.lessThan(peer)) continue;
const days = @divTrunc(peer.toEpoch() - last.toEpoch(), std.time.s_per_day);
const finding = Finding{
.symbol = e.symbol,
.kind = e.kind,
.last_date = last,
.peer_date = peer,
.days_behind = days,
};
if (days > max_normal_lag_days) {
try far_behind.append(allocator, finding);
} else {
try stale.append(allocator, finding);
}
}
// Ownership moves out of the ArrayLists one at a time, so each completed
// handoff needs its own errdefer: the list's errdefer no longer covers a
// slice it has released, and the Report does not exist yet to cover it
// either. An allocation-failure test caught `stale_out` leaking in exactly
// that window.
const stale_out = try stale.toOwnedSlice(allocator);
errdefer allocator.free(stale_out);
std.mem.sort(Finding, stale_out, {}, Finding.worstFirst);
const far_out = try far_behind.toOwnedSlice(allocator);
errdefer allocator.free(far_out);
std.mem.sort(Finding, far_out, {}, Finding.worstFirst);
const orphans_out = try orphans.toOwnedSlice(allocator);
errdefer allocator.free(orphans_out);
const missing_out = try missing.toOwnedSlice(allocator);
return .{
.stale = stale_out,
.far_behind = far_out,
.orphans = orphans_out,
.missing = missing_out,
.groups = groups,
};
}
// tests
const testing = std.testing;
/// Friday 2025-06-13 at 18:00 ET, past the 16:55 equity boundary. Equities
/// should hold Friday's bar; mutual-fund NAVs for Friday are not published
/// until the next morning, so funds should hold Thursday's.
fn fridayEvening() i64 {
return Date.fromYmd(2025, 6, 13).toEpoch() + 22 * std.time.s_per_hour;
}
fn eq(sym: []const u8, d: ?Date, tracked: bool) Entry {
return .{ .symbol = sym, .kind = .equity, .last_date = d, .tracked = tracked };
}
test "scan: a symbol behind its peers is reported" {
const a = testing.allocator;
// The shape of the observed failure: most of the corpus has the latest
// bar, a few do not.
const fri = Date.fromYmd(2025, 6, 13);
const thu = Date.fromYmd(2025, 6, 12);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
eq("NVDA", fri, true),
eq("TSLA", thu, true),
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 1), r.stale.len);
try testing.expectEqualStrings("TSLA", r.stale[0].symbol);
try testing.expectEqual(@as(i64, 1), r.stale[0].days_behind);
try testing.expect(r.stale[0].peer_date.eql(fri));
}
test "scan: a uniformly-behind group reports nothing, and needs no clamp to do it" {
const a = testing.allocator;
// A closure (or simply a corpus nobody refreshed) moves every symbol
// together, so the peer maximum moves too and no symbol sits behind it.
// Peer comparison is self-correcting here - a clock-based check would flag
// all four, and an added calendar clamp would suppress real findings
// elsewhere for no gain. `freshness` is still reported as context so a
// caller can say "refresh everything".
const wed = Date.fromYmd(2025, 6, 11);
const entries = [_]Entry{
eq("AAPL", wed, true),
eq("MSFT", wed, true),
eq("NVDA", wed, true),
eq("TSLA", wed, true),
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 0), r.stale.len);
// Context says the corpus is behind; it does not gate anything.
try testing.expect(r.groups[0].freshness.? != .current);
try testing.expect(r.groups[0].conclusive());
}
test "scan: a laggard is found even when the corpus as a whole is behind" {
const a = testing.allocator;
// THE REGRESSION THIS GUARDS, caught on live data. Monday evening, nothing
// has refreshed since Friday: every symbol is behind the session the
// calendar expects. An earlier version gated findings on the group's own
// bar being `.current` and so reported nothing at all - while three
// symbols sat a full session behind the other twenty-one.
//
// Monday 2025-06-16 22:00 UTC, past the equity boundary, so Monday's bar
// is expected and even the peer maximum (Friday) is overdue.
const monday_evening = Date.fromYmd(2025, 6, 16).toEpoch() + 22 * std.time.s_per_hour;
const fri = Date.fromYmd(2025, 6, 13);
const thu = Date.fromYmd(2025, 6, 12);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
eq("NVDA", fri, true),
eq("AGG", thu, true),
eq("AMZN", thu, true),
eq("NKE", thu, true),
};
var r = try scan(a, &entries, monday_evening);
defer r.deinit(a);
// The group is behind - and that must not silence the three outliers.
try testing.expect(r.groups[0].freshness.? != .current);
try testing.expectEqual(@as(usize, 3), r.stale.len);
for (r.stale) |f| {
try testing.expect(f.last_date.eql(thu));
try testing.expect(f.peer_date.eql(fri));
try testing.expectEqual(@as(i64, 1), f.days_behind);
}
}
test "scan: kinds are judged separately, so fund NAV lag is not staleness" {
const a = testing.allocator;
// Friday evening: equities have Friday, funds still legitimately have
// Thursday. Judged as one corpus, every fund would read a day stale every
// single evening.
const fri = Date.fromYmd(2025, 6, 13);
const thu = Date.fromYmd(2025, 6, 12);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
.{ .symbol = "VBTLX", .kind = .mutual_fund, .last_date = thu, .tracked = true },
.{ .symbol = "VPMAX", .kind = .mutual_fund, .last_date = thu, .tracked = true },
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 0), r.stale.len);
// Both groups are caught up against their own schedules.
try testing.expectEqual(market.CandleFreshness.current, r.groups[0].freshness.?);
try testing.expectEqual(market.CandleFreshness.current, r.groups[1].freshness.?);
}
test "scan: a lone symbol in its kind yields no verdict" {
const a = testing.allocator;
// With one peer there is nothing to compare against, and the clock alone
// cannot tell a frozen cache from a closed market. Silence is the honest
// answer, and this is peer comparison's documented blind spot.
const entries = [_]Entry{
.{ .symbol = "VBTLX", .kind = .mutual_fund, .last_date = Date.fromYmd(2025, 5, 1), .tracked = true },
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 0), r.stale.len);
try testing.expect(!r.groups[1].conclusive());
try testing.expectEqual(@as(usize, 1), r.groups[1].dated);
}
test "scan: an untracked symbol is an orphan, never stale" {
const a = testing.allocator;
// Nothing refreshes it, so it can only fall further behind. Reporting it
// as stale would put a permanent entry at the top of the list and train
// the operator to skip the section.
const fri = Date.fromYmd(2025, 6, 13);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
eq("OLDCO", Date.fromYmd(2025, 1, 2), false),
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 0), r.stale.len);
try testing.expectEqual(@as(usize, 1), r.orphans.len);
try testing.expectEqualStrings("OLDCO", r.orphans[0]);
}
test "scan: no candle meta means missing when tracked, ignored when not" {
const a = testing.allocator;
// A negative-cached symbol or a non-candle key has no bar. That is not a
// stale bar, and inventing an infinitely-old date for it would put it
// permanently at the top of the worst-first list.
const fri = Date.fromYmd(2025, 6, 13);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
eq("DOGE-USD", null, true),
eq("0000320193", null, false),
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 0), r.stale.len);
try testing.expectEqual(@as(usize, 0), r.orphans.len);
try testing.expectEqual(@as(usize, 1), r.missing.len);
try testing.expectEqualStrings("DOGE-USD", r.missing[0]);
try testing.expectEqual(@as(usize, 2), r.groups[0].dated);
}
test "scan: findings are ordered worst first" {
const a = testing.allocator;
// One session behind and three days behind are different magnitudes; the
// larger should not be buried under the smaller.
const fri = Date.fromYmd(2025, 6, 13);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
eq("NEAR", Date.fromYmd(2025, 6, 12), true),
eq("MID", Date.fromYmd(2025, 6, 11), true),
eq("FAR", Date.fromYmd(2025, 6, 9), true),
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 3), r.stale.len);
try testing.expectEqualStrings("FAR", r.stale[0].symbol);
try testing.expectEqualStrings("MID", r.stale[1].symbol);
try testing.expectEqualStrings("NEAR", r.stale[2].symbol);
try testing.expectEqual(@as(i64, 4), r.stale[0].days_behind);
try testing.expectEqual(@as(usize, 0), r.far_behind.len);
}
test "scan: a symbol far behind is separated from one merely lagging" {
const a = testing.allocator;
// Observed live: three symbols 3-4 days behind, which a refresh fixes,
// alongside one 42 days behind that no refresh had moved. In a single
// worst-first list the latter sits permanently on top and buries the
// actionable ones.
const fri = Date.fromYmd(2025, 6, 13);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
eq("LAGGY", Date.fromYmd(2025, 6, 10), true),
eq("STOPPED", Date.fromYmd(2025, 5, 2), true),
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 1), r.stale.len);
try testing.expectEqualStrings("LAGGY", r.stale[0].symbol);
try testing.expectEqual(@as(usize, 1), r.far_behind.len);
try testing.expectEqualStrings("STOPPED", r.far_behind[0].symbol);
try testing.expectEqual(@as(i64, 42), r.far_behind[0].days_behind);
}
test "scan: the boundary is at max_normal_lag_days, exclusive" {
const a = testing.allocator;
const peer = Date.fromYmd(2025, 6, 13);
// Exactly at the limit is still ordinary lag; one day past it is not. The
// limit is a Friday bar missed ahead of a Monday holiday.
const at = peer.addDays(@intCast(-max_normal_lag_days));
const over = peer.addDays(@intCast(-(max_normal_lag_days + 1)));
const entries = [_]Entry{
eq("AAPL", peer, true),
eq("MSFT", peer, true),
eq("AT", at, true),
eq("OVER", over, true),
};
var r = try scan(a, &entries, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 1), r.stale.len);
try testing.expectEqualStrings("AT", r.stale[0].symbol);
try testing.expectEqual(@as(i64, 4), r.stale[0].days_behind);
try testing.expectEqual(@as(usize, 1), r.far_behind.len);
try testing.expectEqualStrings("OVER", r.far_behind[0].symbol);
try testing.expectEqual(@as(i64, 5), r.far_behind[0].days_behind);
}
test "scan: an empty corpus is not an error" {
const a = testing.allocator;
var r = try scan(a, &.{}, fridayEvening());
defer r.deinit(a);
try testing.expectEqual(@as(usize, 0), r.stale.len);
for (r.groups) |g| {
try testing.expectEqual(@as(?Date, null), g.peer_date);
try testing.expect(!g.conclusive());
}
}
/// OOM-path wrapper for `checkAllAllocationFailures`.
fn scanOom(a: std.mem.Allocator, entries: []const Entry, now_s: i64) !void {
var r = try scan(a, entries, now_s);
r.deinit(a);
}
test "scan: every allocation-failure path unwinds cleanly" {
const fri = Date.fromYmd(2025, 6, 13);
const entries = [_]Entry{
eq("AAPL", fri, true),
eq("MSFT", fri, true),
eq("TSLA", Date.fromYmd(2025, 6, 12), true),
eq("OLDCO", Date.fromYmd(2025, 1, 2), false),
eq("DOGE-USD", null, true),
};
try testing.checkAllAllocationFailures(
testing.allocator,
scanOom,
.{ @as([]const Entry, &entries), fridayEvening() },
);
}

205
src/cache/store.zig vendored
View file

@ -275,6 +275,18 @@ pub const Store = struct {
};
}
/// Is `name` a data cache key, or one of this store's own synthetic ones?
///
/// `_edgar` holds the EDGAR ticker indexes and `_torn` archived torn-body
/// forensics. Both are created by this store and neither is a symbol.
///
/// Shared by `cacheKeys` and `diskStats` because they disagreed: `doctor`
/// reported 41 symbols where `zfin cache stats` reported 40, the difference
/// being `_edgar`.
fn isDataKey(name: []const u8) bool {
return name.len > 0 and name[0] != '_';
}
/// Aggregate on-disk cache statistics.
pub const DiskStats = struct { symbols: usize = 0, files: usize = 0, bytes: u64 = 0 };
@ -300,7 +312,9 @@ pub const Store = struct {
stats.bytes += st.size;
},
.directory => {
stats.symbols += 1;
// Synthetic keys are not symbols, but their bytes are still
// on the disk, so they count toward files/bytes below.
if (isDataKey(entry.name)) stats.symbols += 1;
const subpath = std.fs.path.join(self.allocator, &.{ self.cache_dir, entry.name }) catch continue;
defer self.allocator.free(subpath);
var sub = std.Io.Dir.cwd().openDir(io, subpath, .{ .iterate = true }) catch continue;
@ -321,8 +335,66 @@ pub const Store = struct {
return stats;
}
// Generic typed API
/// Every cache key on disk, sorted. Caller owns the strings and the outer
/// slice.
///
/// Deliberately dumb: it lists directory names and makes no judgement about
/// what a key means. A key may be a ticker, a CUSIP, or an EDGAR CIK - this
/// is a generic SRF store and classifying them is not its business.
///
/// Callers filter by what they actually need, which is usually cheaper and
/// always more accurate than guessing from the name. A candle-staleness
/// sweep, for instance, wants keys that have candle meta, so
/// `readCandleMeta() != null` is both its filter and its data - and it
/// excludes CIK keys, negative-cached symbols and the EDGAR indexes for
/// free, without this function knowing any of those exist.
///
/// The one exclusion is the store's own `_`-prefixed synthetic keys
/// (`_edgar` for the EDGAR ticker indexes, `_torn` for archived torn-body
/// forensics). That is self-knowledge, not domain knowledge: this store
/// created them.
///
/// Read-only. Returns empty (not an error) when the cache directory does
/// not exist yet.
pub fn cacheKeys(self: *Store, allocator: std.mem.Allocator) ![][]const u8 {
const io = self.io;
var out = std.ArrayList([]const u8).empty;
errdefer {
for (out.items) |k| allocator.free(k);
out.deinit(allocator);
}
var dir = std.Io.Dir.cwd().openDir(io, self.cache_dir, .{ .iterate = true }) catch
return out.toOwnedSlice(allocator);
defer dir.close(io);
var iter = dir.iterate();
while (iter.next(io) catch null) |entry| {
if (entry.kind != .directory) continue;
if (!isDataKey(entry.name)) continue;
const owned = try allocator.dupe(u8, entry.name);
{
errdefer allocator.free(owned);
try out.append(allocator, owned);
}
}
const keys = try out.toOwnedSlice(allocator);
std.mem.sort([]const u8, keys, {}, struct {
fn lt(_: void, a: []const u8, b: []const u8) bool {
return std.mem.order(u8, a, b) == .lt;
}
}.lt);
return keys;
}
/// Free a `cacheKeys` result.
pub fn freeCacheKeys(allocator: std.mem.Allocator, keys: [][]const u8) void {
for (keys) |k| allocator.free(k);
allocator.free(keys);
}
// Generic typed API
/// Map a model type to its cache DataType.
pub fn dataTypeFor(comptime T: type) DataType {
return switch (T) {
@ -1903,7 +1975,18 @@ pub fn deserializePortfolio(allocator: std.mem.Allocator, data: []const u8) !Por
var skipped: usize = 0;
while (try it.next()) |fields| {
const line = it.state.line;
var lot = fields.to(Lot, .{}) catch {
// `strings_to_numbers` because these are HUMAN-EDITED files, which is
// exactly the case srf's default strict coercion is not for - its own
// doc says "if you want to use this for human-edited files, turn this
// on". Strict mode assumes the writer was a machine, so a numeric
// field spelled with a string separator (`close_price::200.00` instead
// of `close_price:num:200.00`) reaches an unchecked `val.?.number` and
// takes the whole process down. One such typo was enough to panic every
// `zfin portfolio` run.
//
// The `catch` below still handles genuinely unparseable values; this
// only stops a hand-typed separator from being fatal.
var lot = fields.to(Lot, .{ .strings_to_numbers = true }) catch {
std.log.warn("portfolio: could not parse record at line {d}", .{line});
skipped += 1;
continue;
@ -3720,3 +3803,119 @@ test "appendRaw atomicity: concurrent readers see either pre- or post-append, ne
try testing.expect(total > 0);
try testing.expectEqual(@as(u32, 0), bad);
}
test "deserializePortfolio: a hand-typed string separator on a numeric field is not fatal" {
// THE CRASH THIS GUARDS. A numeric field spelled with a string separator -
// `close_price::200.00` instead of `close_price:num:200.00`, one character -
// panicked with "access of union field 'number' while field 'string' is
// active", because srf's default strict coercion is built for
// machine-written cache files and reaches an unchecked `val.?.number`.
// These files are hand-maintained, so leniency is the documented answer.
const data =
"#!srfv1\n" ++
"symbol::AAPL,shares:num:100,open_date::2024-01-15,open_price:num:140.00,close_date::2024-06-03,close_price::200.00\n";
var p = try deserializePortfolio(std.testing.allocator, data);
defer p.deinit();
try std.testing.expectEqual(@as(usize, 1), p.lots.len);
// Parsed, not skipped, and to the right value.
try std.testing.expectApproxEqAbs(@as(f64, 200.00), p.lots[0].close_price.?, 0.001);
}
test "deserializePortfolio: underscore digit separators parse (they always did)" {
// Recorded because they look suspicious and are not the bug: Zig's
// std.fmt.parseFloat accepts `_` natively, so `shares:num:1_234_567` has
// always been fine. Chased this before finding the real cause.
const data =
"#!srfv1\n" ++
"security_type::illiquid,symbol::Sample Asset,shares:num:1_234_567,open_date::2024-01-15,open_price:num:1.00\n";
var p = try deserializePortfolio(std.testing.allocator, data);
defer p.deinit();
try std.testing.expectEqual(@as(usize, 1), p.lots.len);
try std.testing.expectApproxEqAbs(@as(f64, 1_234_567), p.lots[0].shares, 0.5);
}
test "cacheKeys: directory names only, sorted, store-internal keys excluded" {
const allocator = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
// Unsorted on disk, salted with the shapes a real cache holds. Note
// `0000320193` (a CIK) IS returned - classifying keys is the caller's job,
// and a candle sweep drops it for free because it has no candle meta.
for ([_][]const u8{ "NVDA", "AAPL", "_edgar", "_torn", "0000320193", "BRK-B" }) |name|
try tmp.dir.createDir(io, name, std.Io.File.Permissions.default_dir);
(try tmp.dir.createFile(io, "cusip_tickers.srf", .{})).close(io);
var s = Store.init(io, allocator, dir_path);
const keys = try s.cacheKeys(allocator);
defer Store.freeCacheKeys(allocator, keys);
try std.testing.expectEqual(@as(usize, 4), keys.len);
try std.testing.expectEqualStrings("0000320193", keys[0]);
try std.testing.expectEqualStrings("AAPL", keys[1]);
try std.testing.expectEqualStrings("BRK-B", keys[2]);
try std.testing.expectEqualStrings("NVDA", keys[3]);
}
test "cacheKeys: a missing cache directory is empty, not an error" {
const allocator = std.testing.allocator;
const io = std.testing.io;
// First run on a fresh machine: callers must report "nothing cached"
// rather than failing.
var s = Store.init(io, allocator, "/nonexistent/zfin-cache-path");
const keys = try s.cacheKeys(allocator);
defer Store.freeCacheKeys(allocator, keys);
try std.testing.expectEqual(@as(usize, 0), keys.len);
}
/// OOM-path wrapper for `checkAllAllocationFailures`.
fn cacheKeysOom(a: std.mem.Allocator, s: *Store) !void {
const keys = try s.cacheKeys(a);
Store.freeCacheKeys(a, keys);
}
test "cacheKeys: every allocation-failure path unwinds cleanly" {
// The same check on `Portfolio.fetchedSymbols` found a real double-free in
// this exact shape of code - an inner errdefer left armed past the point
// the list took ownership.
const allocator = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
for ([_][]const u8{ "AAPL", "NVDA", "MSFT" }) |name|
try tmp.dir.createDir(io, name, std.Io.File.Permissions.default_dir);
var s = Store.init(io, allocator, dir_path);
try std.testing.checkAllAllocationFailures(allocator, cacheKeysOom, .{&s});
}
test "diskStats and cacheKeys agree on the symbol count" {
const allocator = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
// They disagreed: `doctor` reported 41 symbols where `zfin cache stats`
// reported 40, because only one of them excluded `_edgar`.
for ([_][]const u8{ "AAPL", "NVDA", "_edgar", "_torn" }) |name|
try tmp.dir.createDir(io, name, std.Io.File.Permissions.default_dir);
(try tmp.dir.createFile(io, "cusip_tickers.srf", .{})).close(io);
var s = Store.init(io, allocator, dir_path);
const keys = try s.cacheKeys(allocator);
defer Store.freeCacheKeys(allocator, keys);
const ds = s.diskStats();
try std.testing.expectEqual(keys.len, ds.symbols);
try std.testing.expectEqual(@as(usize, 2), ds.symbols);
// Synthetic keys are excluded from the symbol tally but their bytes are
// still on the disk, so the top-level file still counts toward `files`.
try std.testing.expectEqual(@as(usize, 1), ds.files);
}

View file

@ -3,11 +3,12 @@ const zfin = @import("../root.zig");
const cli = @import("common.zig");
const framework = @import("framework.zig");
const srf = @import("srf");
const freshness = @import("../cache/freshness.zig");
const Store = zfin.cache.Store;
const DataType = zfin.cache.DataType;
pub const Subcommand = enum { stats, clear };
pub const Subcommand = enum { stats, stale, clear };
pub const ParsedArgs = struct {
sub: Subcommand,
@ -18,13 +19,19 @@ pub const meta: framework.Meta = .{
.group = .infra,
.synopsis = "Inspect or clear the local provider-data cache",
.help =
\\Usage: zfin cache <stats|clear>
\\Usage: zfin cache <stats|stale|clear>
\\
\\Subcommands:
\\ stats List every cached symbol with per-data-type size,
\\ age, and freshness state. Stale entries (past TTL)
\\ are flagged. Includes the cusip_tickers.srf file
\\ if present.
\\ stale Find symbols whose newest candle is behind their
\\ peers'. Compares each symbol against others of the
\\ same kind (equity vs mutual fund) rather than against
\\ the clock, so an un-modeled market closure - which
\\ moves every symbol together - is not mistaken for a
\\ frozen cache. Read-only; reports, never fetches.
\\ clear Delete every file under the cache directory.
\\ No confirmation; the next provider call will
\\ re-fetch everything.
@ -55,7 +62,7 @@ const display_labels = [_][]const u8{
pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedArgs {
if (cmd_args.len < 1) {
cli.stderrPrint(ctx.io, "Error: 'cache' requires a subcommand (stats, clear)\n");
cli.stderrPrint(ctx.io, "Error: 'cache' requires a subcommand (stats, stale, clear)\n");
return error.MissingSubcommand;
}
if (cmd_args.len > 1) {
@ -66,18 +73,22 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr
if (std.mem.eql(u8, sub_str, "stats")) {
return .{ .sub = .stats };
}
if (std.mem.eql(u8, sub_str, "stale")) {
return .{ .sub = .stale };
}
if (std.mem.eql(u8, sub_str, "clear")) {
return .{ .sub = .clear };
}
cli.stderrPrint(ctx.io, "Error: unknown cache subcommand '");
cli.stderrPrint(ctx.io, sub_str);
cli.stderrPrint(ctx.io, "'. Use 'stats' or 'clear'.\n");
cli.stderrPrint(ctx.io, "'. Use 'stats', 'stale' or 'clear'.\n");
return error.UnknownSubcommand;
}
pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
switch (parsed.sub) {
.stats => try runStats(ctx),
.stale => try runStale(ctx),
.clear => try runClear(ctx),
}
}
@ -95,46 +106,26 @@ fn runStats(ctx: *framework.RunCtx) !void {
const now_s = std.Io.Timestamp.now(io, .real).toSeconds();
try out.print("Cache directory: {s}\n\n", .{config.cache_dir});
var dir = std.Io.Dir.cwd().openDir(io, config.cache_dir, .{ .iterate = true }) catch {
// Enumerate via the store rather than walking the directory here. The
// staleness sweep needs the same list, and two hand-rolled walks would
// drift the way the watch-symbol unions did.
var store = Store.init(io, allocator, config.cache_dir);
const symbols = store.cacheKeys(allocator) catch {
try out.print(" (empty -- no cached data)\n", .{});
return;
};
defer dir.close(io);
defer Store.freeCacheKeys(allocator, symbols);
// Collect and sort symbol names
var symbols: std.ArrayList([]const u8) = .empty;
defer {
for (symbols.items) |s| allocator.free(s);
symbols.deinit(allocator);
}
var iter = dir.iterate();
while (iter.next(io) catch null) |entry| {
if (entry.kind == .directory) {
const name = allocator.dupe(u8, entry.name) catch continue;
symbols.append(allocator, name) catch {
allocator.free(name);
continue;
};
}
}
if (symbols.items.len == 0) {
if (symbols.len == 0) {
try out.print(" (empty -- no cached data)\n", .{});
return;
}
std.mem.sort([]const u8, symbols.items, {}, struct {
fn cmp(_: void, a: []const u8, b: []const u8) bool {
return std.mem.order(u8, a, b) == .lt;
}
}.cmp);
// Track totals
var total_size: u64 = 0;
var total_files: usize = 0;
for (symbols.items) |symbol| {
for (symbols) |symbol| {
try out.print("{s}\n", .{symbol});
// Print header
@ -205,12 +196,124 @@ fn runStats(ctx: *framework.RunCtx) !void {
var total_buf: [10]u8 = undefined;
try out.print("{d} symbol(s), {d} file(s), {s} total\n", .{
symbols.items.len,
symbols.len,
total_files,
formatSize(&total_buf, total_size),
});
}
/// Peer-comparison staleness sweep over the cached corpus.
///
/// Read-only by design: it reports and never fetches, so it is safe to run
/// against a shared cache and safe to put in front of an operator who has not
/// decided what to do yet. `zfin cache refresh` is the acting half.
fn runStale(ctx: *framework.RunCtx) !void {
const io = ctx.io;
const allocator = ctx.allocator;
const out = ctx.out;
// wall-clock required: the sweep compares each symbol's newest bar against
// the market calendar's notion of what should be published by now.
// Captured once so every symbol is judged against the same instant.
const now_s = std.Io.Timestamp.now(io, .real).toSeconds();
// Arena for the transient string work below. Not a nicety: the tracked-set
// keys are borrowed by a hashmap that outlives the block that builds them,
// and a scoped `defer free` there dangles every key before the sweep reads
// it. One arena at function scope removes that hazard and the two smaller
// leaks around it.
var arena_state = std.heap.ArenaAllocator.init(allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var store = Store.init(io, allocator, ctx.config.cache_dir);
const keys = store.cacheKeys(arena) catch {
try out.print("Cache is empty; nothing to check.\n", .{});
return;
};
if (keys.len == 0) {
try out.print("Cache is empty; nothing to check.\n", .{});
return;
}
// The set zfin intends to keep fresh. Absent a portfolio every cached
// symbol counts as untracked, which is honest - without one there is
// nothing to be tracked BY.
var tracked = std.StringHashMap(void).init(arena);
if (cli.loadPortfolio(ctx, ctx.today)) |loaded_pf| {
var l = loaded_pf;
defer l.deinit(allocator);
tracked = try cli.trackedSymbols(ctx, arena, l.portfolio, l.anchor());
}
const entries = try freshness.collect(arena, &store, keys, &tracked);
var report = try freshness.scan(arena, entries, now_s);
defer report.deinit(arena);
for (report.groups) |g| {
const label = switch (g.kind) {
.equity => "equity/ETF",
.mutual_fund => "mutual fund",
};
if (g.peer_date) |pd| {
try out.print("{s:<12} {d} cached, newest bar {f}", .{ label, g.dated, pd });
if (!g.conclusive()) {
try out.print(" - only one cached, no peer to compare against", .{});
} else if (g.freshness) |f| switch (f) {
// Advice about the corpus, NOT a reason to withhold findings.
// The per-symbol comparison below stands on its own.
.lagging => try out.print(" - the latest session's bar is not published yet", .{}),
.overdue => try out.print(" - the whole group is a session or more behind; a refresh is due", .{}),
.current => {},
};
try out.print("\n", .{});
} else {
try out.print("{s:<12} none cached\n", .{label});
}
}
if (report.stale.len > 0) {
try out.print("\nBehind their peers ({d}):\n", .{report.stale.len});
for (report.stale) |f| {
try out.print(" {s:<10} {f} {d}d behind {f}\n", .{ f.symbol, f.last_date, f.days_behind, f.peer_date });
}
try out.print("\n zfin --refresh-data=force portfolio re-fetches the tracked set\n", .{});
} else if (report.far_behind.len == 0) {
try out.print("\nNothing behind its peers.\n", .{});
}
if (report.far_behind.len > 0) {
// Separated because a refresh is probably not the remedy. Naming no
// cause on purpose: none of the candidates is visible from here, and the
// first draft of this asserted "delisted", which is one of the least
// likely of them.
try out.print("\nFurther behind than lag explains ({d}):\n", .{report.far_behind.len});
for (report.far_behind) |f| {
try out.print(" {s:<10} {f} {d}d behind {f}\n", .{ f.symbol, f.last_date, f.days_behind, f.peer_date });
}
try out.print(" A refresh has probably already been tried on these. Worth checking:\n", .{});
try out.print(" - whether whatever refreshes this cache includes the symbol at all\n", .{});
try out.print(" (a shared server refreshes ITS symbol set, not yours)\n", .{});
try out.print(" - whether the ticker changed, or the provider wants a different form\n", .{});
try out.print(" - whether that one symbol is failing auth or being rate-limited\n", .{});
try out.print(" - whether the provider still covers it, or it stopped trading\n", .{});
try out.print(" `zfin cache stats` shows the per-symbol fetch state.\n", .{});
}
if (report.missing.len > 0) {
try out.print("\nTracked but never cached ({d}): ", .{report.missing.len});
for (report.missing, 0..) |sym, i| try out.print("{s}{s}", .{ if (i == 0) "" else ", ", sym });
try out.print("\n", .{});
}
if (report.orphans.len > 0) {
try out.print("\nCached but tracked by nothing ({d}): ", .{report.orphans.len});
for (report.orphans, 0..) |sym, i| try out.print("{s}{s}", .{ if (i == 0) "" else ", ", sym });
try out.print("\n Nothing will refresh these. `zfin cache clear` is the blunt option.\n", .{});
}
}
fn runClear(ctx: *framework.RunCtx) !void {
var store = Store.init(ctx.io, ctx.allocator, ctx.config.cache_dir);
try store.clearAll();
@ -412,3 +515,11 @@ test "formatSize: megabytes" {
try std.testing.expectEqualStrings("1.0 MB", formatSize(&buf, 1024 * 1024));
try std.testing.expectEqualStrings("2.5 MB", formatSize(&buf, 2 * 1024 * 1024 + 512 * 1024));
}
test "parseArgs: 'stale' resolves to .stale" {
var ctx: framework.RunCtx = undefined;
ctx.io = std.testing.io;
const args = [_][]const u8{"stale"};
const parsed = try parseArgs(&ctx, &args);
try std.testing.expectEqual(Subcommand.stale, parsed.sub);
}

View file

@ -8,6 +8,7 @@ const framework = @import("framework.zig");
const stderr = @import("../stderr.zig");
pub const fmt = @import("../format.zig");
const theme = @import("../tui/theme.zig");
const projections = @import("../analytics/projections.zig");
// Active CLI text palette
// RGB foreground colors for ALL CLI (non-TUI) text output, emitted as
@ -389,6 +390,75 @@ fn printLoadSummaryImpl(io: std.Io, color: bool, s: LoadSummaryStats) !void {
const portfolio_loader = @import("../portfolio_loader.zig");
/// The set of symbols zfin intends to keep fresh, for a loaded portfolio.
///
/// Holdings, `security_type::watch` lots, `watchlist.srf`, and the projections
/// benchmark pair - the union `Portfolio.fetchedSymbols` defines, with the
/// watchlist and benchmark inputs resolved from `ctx`.
///
/// Keys borrow from `arena`; pass an arena that outlives every lookup. The
/// scoped-`defer`-frees version of this dangled its keys before the caller read
/// them, which made every cached symbol look untracked.
pub fn trackedSymbols(
ctx: *framework.RunCtx,
arena: std.mem.Allocator,
portfolio: zfin.Portfolio,
anchor: []const u8,
) !std.StringHashMap(void) {
var set = std.StringHashMap(void).init(arena);
const wl = ctx.resolveWatchlistPath();
defer wl.deinit(ctx.allocator);
const wl_syms: ?[][]const u8 = if (ctx.globals.watchlist_path != null or wl.resolved != null)
loadWatchlist(ctx.io, arena, wl.path)
else
null;
const syms = try portfolio.fetchedSymbols(arena, .{
.watchlist_syms = if (wl_syms) |w| w else &.{},
.benchmarks = benchmarkPair(ctx.io, arena, anchor),
});
for (syms) |sym| try set.put(sym, {});
return set;
}
/// A path to `name` in the same directory as `anchor`.
///
/// The portfolio's siblings - `accounts.srf`, `metadata.srf`,
/// `transaction_log.srf`, `projections.srf` - all live beside whichever
/// portfolio file was resolved, so "next to the anchor" is the one rule that
/// finds them regardless of `ZFIN_HOME`, `-p` patterns, or cwd.
///
/// Shared rather than per-command: it was private to `doctor` and the second
/// caller promptly hand-rolled dirname + join, which is how the two would have
/// drifted on trailing-separator handling.
pub fn siblingPath(arena: std.mem.Allocator, anchor: []const u8, name: []const u8) ![]const u8 {
const dir_end = if (std.mem.lastIndexOfScalar(u8, anchor, std.fs.path.sep)) |idx| idx + 1 else 0;
return std.fmt.allocPrint(arena, "{s}{s}", .{ anchor[0..dir_end], name });
}
/// The projections benchmark pair, read from `projections.srf` beside the
/// portfolio. Returns the SPY/AGG defaults when the file is absent or silent.
///
/// These belong in the tracked set even though they are held nowhere: they are
/// fetched for the benchmark comparison, which is why AGG went stale unnoticed
/// while SPY - which doubles as a `ticker::` alias on a real holding - stayed
/// current.
///
/// `views/projections.zig` reads the same file for the same config, but keeps
/// its `UserConfig` alive and reads the fields in place. This exists only
/// because the tracked set outlives the config: an overridden symbol lives in a
/// `[16]u8` field inside `UserConfig`, so the slices must be copied out before
/// that struct dies.
pub fn benchmarkPair(io: std.Io, arena: std.mem.Allocator, anchor: []const u8) []const []const u8 {
const path = siblingPath(arena, anchor, "projections.srf") catch return &.{};
const data = std.Io.Dir.cwd().readFileAlloc(io, path, arena, .limited(64 * 1024)) catch null;
const cfg = projections.parseProjectionsConfig(data);
const pair = arena.alloc([]const u8, 2) catch return &.{};
pair[0] = arena.dupe(u8, cfg.benchmark_stock) catch return &.{};
pair[1] = arena.dupe(u8, cfg.benchmark_bond) catch return &.{};
return pair;
}
pub const LoadedPortfolio = portfolio_loader.LoadedPortfolio;
pub const PortfolioData = portfolio_loader.PortfolioData;
pub const loadPortfolioFromConfig = portfolio_loader.loadPortfolioFromConfig;
@ -1475,3 +1545,55 @@ test "buildPortfolioData: builds summary + candle_map for stock positions" {
try std.testing.expect(pf_data.summary.allocations.len > 0);
try std.testing.expectApproxEqAbs(@as(f64, 20_000), pf_data.summary.total_value, 1.0);
}
test "siblingPath: joins a filename onto the anchor's directory" {
var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena.deinit();
const a = arena.allocator();
try std.testing.expectEqualStrings("/home/u/data/accounts.srf", try siblingPath(a, "/home/u/data/portfolio.srf", "accounts.srf"));
// Bare filename (no separator) -> sibling is just the name.
try std.testing.expectEqualStrings("accounts.srf", try siblingPath(a, "portfolio.srf", "accounts.srf"));
}
test "benchmarkPair: defaults to SPY/AGG when projections.srf is absent" {
// These two must be in the tracked set even though they are held nowhere.
// AGG went stale unnoticed for exactly this reason - it is fetched only for
// the benchmark comparison, while SPY looked fine because it doubles as a
// `ticker::` alias on a real holding.
var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena_state.deinit();
const pair = benchmarkPair(std.testing.io, arena_state.allocator(), "/nonexistent/portfolio.srf");
try std.testing.expectEqual(@as(usize, 2), pair.len);
try std.testing.expectEqualStrings("SPY", pair[0]);
try std.testing.expectEqualStrings("AGG", pair[1]);
}
test "benchmarkPair: an override in projections.srf is honoured" {
const allocator = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
{
const f = try tmp.dir.createFile(io, "projections.srf", .{});
defer f.close(io);
var buf: [256]u8 = undefined;
var w = f.writer(io, &buf);
try w.interface.writeAll("#!srfv1\ntype::config,benchmark_stock::VTI,benchmark_bond::BND\n");
try w.interface.flush();
}
var arena_state = std.heap.ArenaAllocator.init(allocator);
defer arena_state.deinit();
const anchor = try std.fs.path.join(allocator, &.{ dir_path, "portfolio.srf" });
defer allocator.free(anchor);
// Duped out of the config's stack-local override buffers - borrowing them
// would dangle the moment benchmarkPair returned.
const pair = benchmarkPair(io, arena_state.allocator(), anchor);
try std.testing.expectEqual(@as(usize, 2), pair.len);
try std.testing.expectEqualStrings("VTI", pair[0]);
try std.testing.expectEqualStrings("BND", pair[1]);
}

View file

@ -2088,6 +2088,13 @@ fn matchTransfers(
// order. Underflow past tolerance -> unmatched_transfer.
var cash_budget: std.StringHashMap(f64) = .init(allocator);
defer cash_budget.deinit();
// Shortfalls carried out of the cash-destination path, per destination
// account, drawn down against new lots once every record has been seen.
// Deferred to the end so a second transfer into the same account can add to
// the same budget before any of it is spent.
var transfer_funding: std.StringHashMap(FundingShortfall) = .init(allocator);
defer transfer_funding.deinit();
for (changes.items) |c| {
const v = c.value();
switch (c.kind) {
@ -2125,12 +2132,16 @@ fn matchTransfers(
try matchLotDestination(allocator, changes, &consumed_lot_idx, rec, dl);
},
.cash => {
try matchCashDestination(allocator, changes, &cash_budget, cash_attributed_by_account, rec);
try matchCashDestination(allocator, changes, &cash_budget, cash_attributed_by_account, &transfer_funding, rec);
},
}
tryMatchFromSide(changes, rec);
}
// After every record, so multiple transfers into one account pool their
// funding before it is drawn against that account's purchases.
try matchTransferFundedPurchases(allocator, changes, &transfer_funding);
}
/// Append a synthetic `unmatched_transfer` Change carrying the record's
@ -2292,31 +2303,44 @@ fn matchCashDestination(
changes: *std.ArrayList(Change),
cash_budget: *std.StringHashMap(f64),
cash_attributed_by_account: *std.StringHashMap(f64),
transfer_funding: *std.StringHashMap(FundingShortfall),
rec: transaction_log.TransferRecord,
) !void {
const budget_entry = cash_budget.getPtr(rec.to);
const available = if (budget_entry) |p| p.* else 0.0;
if (available < rec.amount - transfer_amount_tolerance) {
const buf = try std.fmt.allocPrint(
allocator,
"destination cash increase ${d:.2} insufficient for transfer ${d:.2}",
.{ available, rec.amount },
);
try appendUnmatchedWithOwnedNote(allocator, changes, rec, buf);
return;
const available = @max(0.0, if (budget_entry) |p| p.* else 0.0);
// Credit whatever cash actually showed up, and carry the rest as a funding
// budget for the destination's new lots.
//
// This used to bail out entirely when the cash increase fell short, which
// meant a transfer whose cash was invested inside the same window credited
// nothing at all and every purchase it funded read as new money. The cash
// is genuinely absent from the snapshot in that case - it arrived and left
// between two commits - so the shortfall is expected, not a discrepancy.
// See `matchTransferFundedPurchases`, which draws it down and flags only
// what the account's new lots cannot absorb.
const credited = @min(available, rec.amount);
const shortfall = rec.amount - credited;
if (shortfall > transfer_amount_tolerance) {
const gop = try transfer_funding.getOrPut(rec.to);
if (!gop.found_existing) gop.value_ptr.* = .{ .rec = rec };
gop.value_ptr.*.amount += shortfall;
gop.value_ptr.*.declared += rec.amount;
}
// Draw from the budget. Running remainder stays on the budget
// so later records on the same account see the correct
// capacity.
if (budget_entry) |p| p.* -= rec.amount;
if (budget_entry) |p| p.* -= credited;
// Accumulate into per-account attribution bucket. The per-
// account totals pass subtracts this from cash-side totals so
// transferred cash doesn't double-count.
const gop = try cash_attributed_by_account.getOrPut(rec.to);
if (!gop.found_existing) gop.value_ptr.* = 0;
gop.value_ptr.* += rec.amount;
// Only the cash that was actually observed; the rest is attributed to the
// new lots instead, so adding the full amount here would double-count.
gop.value_ptr.* += credited;
// Distribute the record amount across the destination account's
// cash-side Changes by accumulating into each Change's
@ -2335,7 +2359,7 @@ fn matchCashDestination(
// pass continues to use `cash_attributed_by_account` for its
// per-account math - the two views agree because the same
// amount is subtracted on both sides.
var remaining = rec.amount;
var remaining = credited;
for (changes.items) |*c| {
if (remaining <= 0) break;
if (!std.mem.eql(u8, c.account, rec.to)) continue;
@ -2576,6 +2600,79 @@ fn matchInKindTransfer(
/// several purchase lots; the total netted is the same regardless of
/// order, but which specific lot shows a residual can vary. This
/// mirrors `matchCashDestination`'s order-dependent draw.
/// Draw `budget` down against the new purchase lots in `account`, marking the
/// funded portion on each. Returns whatever the account's lots could not
/// absorb.
///
/// Shared by the two things that can fund a purchase without it being new
/// money: cash that visibly left the same account, and a declared transfer
/// whose cash was spent before it could be observed. The drawdown is identical;
/// only the meaning of a leftover differs, which is why the callers handle the
/// return value differently rather than this function deciding.
fn drawDownAgainstNewLots(changes: *std.ArrayList(Change), account: []const u8, budget: f64) f64 {
var remaining = budget;
for (changes.items) |*c| {
if (remaining <= 0) break;
switch (c.kind) {
.new_stock, .new_cd => {},
else => continue,
}
if (!std.mem.eql(u8, c.account, account)) continue;
const unattributed = c.attributedValue();
if (unattributed <= 0) continue;
const draw = @min(unattributed, remaining);
c.internal_funded += draw;
remaining -= draw;
}
return remaining;
}
/// Attribute purchases funded by a declared transfer whose cash never appeared
/// in a snapshot.
///
/// A `transfer` record says money moved from A to B. `matchCashDestination`
/// credits it against an observed cash increase in B - but when the cash is
/// invested inside the same reconcile window, no snapshot ever contains it: the
/// diff sees new security lots in B and a few dollars of leftover cash. The
/// transfer then failed its cash check and the purchases counted as fresh
/// money, which is how one 401(k)-to-BrokerageLink move reported $738,814 of
/// contributions that were nothing of the kind.
///
/// So the shortfall becomes a funding budget for that account's new lots -
/// exactly what `matchIntraAccountPurchases` does with an observed cash
/// decrease, seeded from the operator's declaration instead of from an
/// observation. Anything the lots cannot absorb is still flagged: a transfer
/// claiming more than the destination gained is a real discrepancy and must not
/// be silently swallowed.
fn matchTransferFundedPurchases(
allocator: std.mem.Allocator,
changes: *std.ArrayList(Change),
funding: *std.StringHashMap(FundingShortfall),
) !void {
var it = funding.iterator();
while (it.next()) |entry| {
const account = entry.key_ptr.*;
const sf = entry.value_ptr.*;
if (sf.amount <= transfer_amount_tolerance) continue;
const leftover = drawDownAgainstNewLots(changes, account, sf.amount);
if (leftover <= transfer_amount_tolerance) continue;
const buf = try std.fmt.allocPrint(
allocator,
"transfer of ${d:.2} exceeds the destination's cash increase and new lots by ${d:.2}",
.{ sf.declared, leftover },
);
try appendUnmatchedWithOwnedNote(allocator, changes, sf.rec, buf);
}
}
/// Per-account transfer shortfall, plus the record it came from so an
/// unabsorbed remainder can be reported against the right transfer.
const FundingShortfall = struct {
amount: f64 = 0,
declared: f64 = 0,
rec: transaction_log.TransferRecord,
};
fn matchIntraAccountPurchases(
allocator: std.mem.Allocator,
changes: *std.ArrayList(Change),
@ -2606,19 +2703,12 @@ fn matchIntraAccountPurchases(
}
if (outflow.count() == 0) return;
// Draw each account's outflow down against its new purchase lots.
for (changes.items) |*c| {
switch (c.kind) {
.new_stock, .new_cd => {},
else => continue,
}
const budget = outflow.getPtr(c.account) orelse continue;
if (budget.* <= 0) continue;
const unattributed = c.attributedValue(); // value() minus any prior attribution
if (unattributed <= 0) continue;
const draw = @min(unattributed, budget.*);
c.internal_funded += draw;
budget.* -= draw;
// Draw each account's outflow down against its new purchase lots. A
// leftover here is unremarkable - cash can leave an account for reasons
// other than buying something - so it is simply discarded.
var oit = outflow.iterator();
while (oit.next()) |e| {
_ = drawDownAgainstNewLots(changes, e.key_ptr.*, e.value_ptr.*);
}
}
@ -5068,7 +5158,56 @@ test "matchTransfers: amount exceeds lot value emits unmatched" {
try std.testing.expectEqual(@as(usize, 1), n_unmatched);
}
test "matchTransfers: cash insufficient emits unmatched" {
test "matchTransfers: a transfer spent on securities before any snapshot is not new money" {
var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena_state.deinit();
const allocator = arena_state.allocator();
var prices = std.StringHashMap(f64).init(allocator);
defer prices.deinit();
try prices.put("VOO", 100.0);
// The real 2026-08 shape, scaled down. $50k moved into an account and was
// invested the same week, so no snapshot ever contains the cash: the diff
// sees a new security lot plus the few dollars that did not get spent.
const before = [_]Lot{};
const after = [_]Lot{
.{ .symbol = "VOO", .shares = 499, .open_date = Date.fromYmd(2026, 5, 2), .open_price = 100.0, .account = "Acct B" },
.{ .symbol = "cash", .shares = 100, .open_date = Date.fromYmd(2026, 5, 2), .open_price = 1.0, .security_type = .cash, .account = "Acct B" },
};
const tlog = try transaction_log.parseTransactionLogFile(allocator,
\\#!srfv1
\\transfer::2026-05-02,type::cash,amount:num:50000,from::Acct A,to::Acct B,dest_lot::cash
\\
);
const report = try computeReport(allocator, &before, &after, &prices, Date.fromYmd(2026, 5, 4), .{
.transfer_log = tlog.transfers,
});
// Nothing new entered the portfolio: $49,900 of VOO plus $100 of leftover
// cash is exactly the $50,000 that moved. Before this was handled, the
// purchase counted as a fresh contribution - the mechanism that reported
// $738,814 of contributions for a 401(k)-to-BrokerageLink move.
const t = report.account_totals.get("Acct B").?;
try std.testing.expectApproxEqAbs(@as(f64, 0.0), t.new_money, 0.01);
// And it is not reported as a discrepancy either, because it is not one:
// the destination gained precisely what the record declared.
for (report.changes) |c| {
try std.testing.expect(c.kind != .unmatched_transfer);
}
// The purchase is attributed as internally funded rather than being
// dropped, so it still shows under "Internal purchases".
var funded: f64 = 0;
for (report.changes) |c| {
if (c.kind == .new_stock) funded += c.internal_funded;
}
try std.testing.expectApproxEqAbs(@as(f64, 49900.0), funded, 0.01);
}
test "matchTransfers: a short cash increase credits what arrived and flags the gap" {
var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena_state.deinit();
const allocator = arena_state.allocator();
@ -5091,14 +5230,20 @@ test "matchTransfers: cash insufficient emits unmatched" {
.transfer_log = tlog.transfers,
});
// The $2k the destination never gained is still a real discrepancy.
var n_unmatched: usize = 0;
for (report.changes) |c| if (c.kind == .unmatched_transfer) {
n_unmatched += 1;
};
try std.testing.expectEqual(@as(usize, 1), n_unmatched);
// new_cash stays unchanged; $3k still counts as new_money.
// But the $3k that DID arrive is transferred money, not new money.
// Previously the whole record was abandoned when the amounts disagreed,
// so a correct partial attribution was discarded and the $3k was reported
// as a fresh contribution - which it demonstrably is not, since a transfer
// record says where it came from.
const t = report.account_totals.get("Acct B").?;
try std.testing.expectApproxEqAbs(@as(f64, 3000.0), t.new_money, 0.01);
try std.testing.expectApproxEqAbs(@as(f64, 0.0), t.new_money, 0.01);
}
test "matchTransfers: same-day multi-cash records drain a single cash_delta" {

View file

@ -34,6 +34,7 @@ const Config = zfin.Config;
const Lot = @import("../models/portfolio.zig").Lot;
const Date = @import("../Date.zig");
const cache = @import("../cache/store.zig");
const freshness = @import("../cache/freshness.zig");
const classification = @import("../models/classification.zig");
const analysis = @import("../analytics/analysis.zig");
const transaction_log = @import("../models/transaction_log.zig");
@ -160,6 +161,59 @@ fn joinCapped(arena: std.mem.Allocator, items: []const []const u8, cap: usize) !
return buf.items;
}
/// Are any tracked symbols' candles behind their same-kind peers'?
///
/// Reads the cache only - no fetches, no writes - which keeps `doctor`'s
/// read-only contract intact. `zfin cache stale` is the detailed view and
/// `zfin cache refresh` the acting one; this exists so the problem surfaces
/// without being asked about, since the failure it catches is silent by nature.
///
/// Always `.warn` or better, never `.fail`: a `.fail` makes `doctor` exit
/// non-zero, and a stale cache is a thing to fix rather than a broken install.
/// Cron and CI stay green.
///
/// Takes the tracked set and `now_s` rather than a `RunCtx` so it can be tested
/// against a temp cache directory. Building the tracked set is the caller's job.
fn checkCandleFreshness(
arena: std.mem.Allocator,
store: *cache.Store,
tracked: *const std.StringHashMap(void),
now_s: i64,
) !Check {
const label = "Candle freshness";
const keys = store.cacheKeys(arena) catch
return .{ .status = .info, .label = label, .detail = "cache not readable" };
if (keys.len == 0) return .{ .status = .info, .label = label, .detail = "nothing cached" };
const entries = try freshness.collect(arena, store, keys, tracked);
var report = try freshness.scan(arena, entries, now_s);
defer report.deinit(arena);
if (report.stale.len == 0 and report.far_behind.len == 0) {
return .{ .status = .ok, .label = label, .detail = "no symbol is behind its peers" };
}
// Both buckets warn, but the detail keeps them apart: one is fixed by a
// refresh and the other probably is not.
var names: std.ArrayList([]const u8) = .empty;
for (report.stale) |f| {
try names.append(arena, try std.fmt.allocPrint(arena, "{s} ({d}d)", .{ f.symbol, f.days_behind }));
}
for (report.far_behind) |f| {
// No cause named - see `freshness.max_normal_lag_days`. "!" marks it as
// the bucket a refresh probably will not fix.
try names.append(arena, try std.fmt.allocPrint(arena, "{s} ({d}d!)", .{ f.symbol, f.days_behind }));
}
return .{
.status = .warn,
.label = label,
.detail = try std.fmt.allocPrint(arena, "{d} behind peers: {s} - see `zfin cache stale`", .{
report.stale.len + report.far_behind.len,
try joinCapped(arena, names.items, 6),
}),
};
}
/// Cross-reference check: every name in `needed` should appear in
/// `known`. OK when all are present (or `needed` is empty); WARN listing
/// the missing ones otherwise. Operates on plain string slices so it's
@ -457,11 +511,6 @@ fn checkSrfFile(
}
/// Join a sibling filename onto the anchor portfolio's directory.
fn siblingPath(arena: std.mem.Allocator, anchor: []const u8, name: []const u8) ![]const u8 {
const dir_end = if (std.mem.lastIndexOfScalar(u8, anchor, std.fs.path.sep)) |idx| idx + 1 else 0;
return std.fmt.allocPrint(arena, "{s}{s}", .{ anchor[0..dir_end], name });
}
// run
pub fn run(ctx: *framework.RunCtx, _: ParsedArgs) !void {
@ -473,8 +522,13 @@ pub fn run(ctx: *framework.RunCtx, _: ParsedArgs) !void {
var sections: std.ArrayList(Section) = .empty;
// Collected for cross-reference (Section B).
// Collected for cross-reference (Section B) and the cache checks
// (Section C), so both are hoisted out of Section A's block.
var all_lots: std.ArrayList(Lot) = .empty;
// First resolved portfolio path. Names the directory the portfolio's
// siblings live in - `projections.srf` among them, which Section C needs
// for the benchmark pair.
var anchor: ?[]const u8 = null;
var account_map: ?analysis.AccountMap = null;
var class_map: ?classification.ClassificationMap = null;
var transfer_log: ?transaction_log.TransactionLog = null;
@ -485,7 +539,6 @@ pub fn run(ctx: *framework.RunCtx, _: ParsedArgs) !void {
const source: []const u8 = if (config.zfin_home) |h| h else "cwd";
// Portfolio file(s) - globbed, union-merged. Parse-check each.
var anchor: ?[]const u8 = null;
const pf = config.resolveUserFiles(io, arena, Config.default_portfolio_filename) catch
Config.ResolvedPaths{ .paths = &.{}, .allocator = arena };
if (pf.paths.len == 0) {
@ -505,10 +558,10 @@ pub fn run(ctx: *framework.RunCtx, _: ParsedArgs) !void {
if (anchor) |a| {
// Accounts - parsed + kept for cross-reference.
{
const r = checkSrfFile(io, arena, "accounts.srf", try siblingPath(arena, a, "accounts.srf"), .optional, vAccounts);
const r = checkSrfFile(io, arena, "accounts.srf", try cli.siblingPath(arena, a, "accounts.srf"), .optional, vAccounts);
try checks.append(arena, r);
if (r.status == .ok) {
const path = try siblingPath(arena, a, "accounts.srf");
const path = try cli.siblingPath(arena, a, "accounts.srf");
if (std.Io.Dir.cwd().readFileAlloc(io, path, arena, .limited(16 * 1024 * 1024))) |b| {
account_map = analysis.parseAccountsFile(arena, b) catch null;
} else |_| {}
@ -516,10 +569,10 @@ pub fn run(ctx: *framework.RunCtx, _: ParsedArgs) !void {
}
// Metadata - parsed + kept.
{
const r = checkSrfFile(io, arena, "metadata.srf", try siblingPath(arena, a, "metadata.srf"), .optional, vMetadata);
const r = checkSrfFile(io, arena, "metadata.srf", try cli.siblingPath(arena, a, "metadata.srf"), .optional, vMetadata);
try checks.append(arena, r);
if (r.status == .ok) {
const path = try siblingPath(arena, a, "metadata.srf");
const path = try cli.siblingPath(arena, a, "metadata.srf");
if (std.Io.Dir.cwd().readFileAlloc(io, path, arena, .limited(16 * 1024 * 1024))) |b| {
class_map = classification.parseClassificationFile(arena, b) catch null;
} else |_| {}
@ -527,16 +580,16 @@ pub fn run(ctx: *framework.RunCtx, _: ParsedArgs) !void {
}
// Transaction log - parsed + kept.
{
const r = checkSrfFile(io, arena, "transaction_log.srf", try siblingPath(arena, a, "transaction_log.srf"), .optional, vTransfers);
const r = checkSrfFile(io, arena, "transaction_log.srf", try cli.siblingPath(arena, a, "transaction_log.srf"), .optional, vTransfers);
try checks.append(arena, r);
if (r.status == .ok) {
const path = try siblingPath(arena, a, "transaction_log.srf");
const path = try cli.siblingPath(arena, a, "transaction_log.srf");
if (std.Io.Dir.cwd().readFileAlloc(io, path, arena, .limited(16 * 1024 * 1024))) |b| {
transfer_log = transaction_log.parseTransactionLogFile(arena, b) catch null;
} else |_| {}
}
}
try checks.append(arena, checkSrfFile(io, arena, "projections.srf", try siblingPath(arena, a, "projections.srf"), .optional, validateSrf));
try checks.append(arena, checkSrfFile(io, arena, "projections.srf", try cli.siblingPath(arena, a, "projections.srf"), .optional, validateSrf));
// imported_values.srf and the snapshots both live under
// <portfolio_dir>/history/, NOT directly beside the
// portfolio file.
@ -610,6 +663,18 @@ pub fn run(ctx: *framework.RunCtx, _: ParsedArgs) !void {
.label = "Cache",
.detail = try std.fmt.allocPrint(arena, "{d} symbols, {d} files, {s} ({s})", .{ ds.symbols, ds.files, cache_cmd.formatSize(&size_buf, ds.bytes), config.cache_dir }),
});
// Without a portfolio there is no tracked set, so every symbol
// would read as an orphan and nothing as stale - an empty set is
// the honest input.
var tracked = std.StringHashMap(void).init(arena);
if (anchor) |a| {
tracked = cli.trackedSymbols(ctx, arena, .{ .lots = all_lots.items, .allocator = arena }, a) catch
std.StringHashMap(void).init(arena);
}
// wall-clock required: peer freshness is judged against the market
// calendar, which needs the real instant rather than `ctx.today`.
const now_s = std.Io.Timestamp.now(io, .real).toSeconds();
try checks.append(arena, try checkCandleFreshness(arena, &store, &tracked, now_s));
}
// Hand-maintained data staleness.
@ -1359,15 +1424,6 @@ test "cross-ref end to end: missing account surfaces as a warn" {
try testing.expect(std.mem.indexOf(u8, c.detail, "Sample HSA") != null);
}
test "siblingPath: joins a filename onto the anchor's directory" {
var arena = std.heap.ArenaAllocator.init(testing.allocator);
defer arena.deinit();
const a = arena.allocator();
try testing.expectEqualStrings("/home/u/data/accounts.srf", try siblingPath(a, "/home/u/data/portfolio.srf", "accounts.srf"));
// Bare filename (no separator) -> sibling is just the name.
try testing.expectEqualStrings("accounts.srf", try siblingPath(a, "portfolio.srf", "accounts.srf"));
}
test "validateSrf: accepts a valid stream, rejects a headerless one" {
var arena = std.heap.ArenaAllocator.init(testing.allocator);
defer arena.deinit();
@ -1439,3 +1495,119 @@ test "trimTrailingSlash: drops a single trailing slash" {
try testing.expectEqualStrings("https://h", trimTrailingSlash("https://h/"));
try testing.expectEqualStrings("https://h", trimTrailingSlash("https://h"));
}
/// Seed a cache directory with `candles_meta.srf` for each symbol/date pair.
fn seedCandleMeta(io: std.Io, tmp: *std.testing.TmpDir, pairs: []const struct { []const u8, []const u8 }) !void {
for (pairs) |p| {
try tmp.dir.createDir(io, p[0], std.Io.File.Permissions.default_dir);
const rel = try std.fmt.allocPrint(testing.allocator, "{s}/candles_meta.srf", .{p[0]});
defer testing.allocator.free(rel);
const f = try tmp.dir.createFile(io, rel, .{});
defer f.close(io);
var buf: [256]u8 = undefined;
var w = f.writer(io, &buf);
try w.interface.print(
"#!srfv1\n#!expires=99999999999\nlast_close:num:100.00,last_date::{s},provider::tiingo\n",
.{p[1]},
);
try w.interface.flush();
}
}
test "checkCandleFreshness: warns and names the laggards, never fails" {
const allocator = testing.allocator;
const io = testing.io;
var arena_state = std.heap.ArenaAllocator.init(allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
try seedCandleMeta(io, &tmp, &.{
.{ "AAPL", "2025-06-13" },
.{ "MSFT", "2025-06-13" },
.{ "NVDA", "2025-06-13" },
.{ "TSLA", "2025-06-12" },
});
var tracked = std.StringHashMap(void).init(arena);
for ([_][]const u8{ "AAPL", "MSFT", "NVDA", "TSLA" }) |s| try tracked.put(s, {});
var store = cache.Store.init(io, arena, dir_path);
// Friday 2025-06-13, 22:00 UTC - past the equity boundary.
const now_s = zfin.Date.fromYmd(2025, 6, 13).toEpoch() + 22 * std.time.s_per_hour;
const c = try checkCandleFreshness(arena, &store, &tracked, now_s);
// WARN, not FAIL: `doctor` exits non-zero on any fail, and a stale cache is
// a thing to fix rather than a broken install. Cron stays green.
try testing.expectEqual(Status.warn, c.status);
try testing.expect(std.mem.indexOf(u8, c.detail, "TSLA (1d)") != null);
try testing.expect(std.mem.indexOf(u8, c.detail, "zfin cache stale") != null);
// The three current symbols are not named.
try testing.expect(std.mem.indexOf(u8, c.detail, "AAPL") == null);
}
test "checkCandleFreshness: a caught-up corpus is OK" {
const allocator = testing.allocator;
const io = testing.io;
var arena_state = std.heap.ArenaAllocator.init(allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
try seedCandleMeta(io, &tmp, &.{ .{ "AAPL", "2025-06-13" }, .{ "MSFT", "2025-06-13" } });
var tracked = std.StringHashMap(void).init(arena);
for ([_][]const u8{ "AAPL", "MSFT" }) |s| try tracked.put(s, {});
var store = cache.Store.init(io, arena, dir_path);
const now_s = zfin.Date.fromYmd(2025, 6, 13).toEpoch() + 22 * std.time.s_per_hour;
const c = try checkCandleFreshness(arena, &store, &tracked, now_s);
try testing.expectEqual(Status.ok, c.status);
}
test "checkCandleFreshness: an empty cache is informational, not a warning" {
const allocator = testing.allocator;
const io = testing.io;
var arena_state = std.heap.ArenaAllocator.init(allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// First run on a fresh machine must not look like a problem.
var tracked = std.StringHashMap(void).init(arena);
var store = cache.Store.init(io, arena, "/nonexistent/zfin-cache");
const c = try checkCandleFreshness(arena, &store, &tracked, 0);
try testing.expectEqual(Status.info, c.status);
}
test "checkCandleFreshness: an untracked laggard does not warn" {
const allocator = testing.allocator;
const io = testing.io;
var arena_state = std.heap.ArenaAllocator.init(allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
try seedCandleMeta(io, &tmp, &.{
.{ "AAPL", "2025-06-13" },
.{ "MSFT", "2025-06-13" },
.{ "OLDCO", "2025-01-02" },
});
// OLDCO is cached but tracked by nothing - an orphan. Nothing refreshes it,
// so warning about it every run would train the operator to skip the line.
var tracked = std.StringHashMap(void).init(arena);
for ([_][]const u8{ "AAPL", "MSFT" }) |s| try tracked.put(s, {});
var store = cache.Store.init(io, arena, dir_path);
const now_s = zfin.Date.fromYmd(2025, 6, 13).toEpoch() + 22 * std.time.s_per_hour;
const c = try checkCandleFreshness(arena, &store, &tracked, now_s);
try testing.expectEqual(Status.ok, c.status);
}

View file

@ -126,23 +126,35 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
var fail_count: usize = 0;
// Also collect watch symbols that need fetching
var watch_syms: std.ArrayList([]const u8) = .empty;
defer watch_syms.deinit(allocator);
{
var seen = std.StringHashMap(void).init(allocator);
defer seen.deinit();
for (syms) |s| try seen.put(s, {});
for (portfolio.lots) |lot| {
if (lot.security_type == .watch and !seen.contains(lot.priceSymbol())) {
try seen.put(lot.priceSymbol(), {});
try watch_syms.append(allocator, lot.priceSymbol());
}
}
}
// Loaded BEFORE the fetch below, not after, and that ordering is the
// fix for a real bug: these symbols used to be read only for display
// and priced from `getCachedLastClose`, so nothing on any CLI path ever
// put them in the fetch set. A watchlist-only symbol therefore went
// arbitrarily stale - SPCX sat 39 days out of date - while the TUI,
// which does pass `watchlist_syms` into its load, showed it current.
//
// Lifetime is unchanged: these slices must outlive the `display` call
// at the end of `run`, because `watch_list` and `watch_prices`' keys
// borrow them. Freeing earlier rendered them as freed-memory garbage.
const wl_syms: ?[][]const u8 = if (watchlist_path) |wl_path|
cli.loadWatchlist(io, allocator, wl_path)
else
null;
defer cli.freeWatchlist(allocator, wl_syms);
// Symbols to price that aren't stock positions: `security_type::watch`
// lots in the portfolio file, plus everything in `watchlist.srf`. The set
// logic is a Portfolio method so it is testable - `run` needs a live
// context, so anything embedded here can only be covered by hand.
const watch_syms = try portfolio.extraPriceSymbols(
allocator,
syms,
if (wl_syms) |l| l else &.{},
);
defer allocator.free(watch_syms);
// All symbols to fetch (stock positions + watch)
const all_syms_count = syms.len + watch_syms.items.len;
const all_syms_count = syms.len + watch_syms.len;
if (all_syms_count > 0) {
// Use consolidated parallel loader
@ -150,7 +162,7 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
io,
svc,
syms,
watch_syms.items,
watch_syms,
ctx.globals.refresh_policy,
color,
);
@ -183,21 +195,11 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
// Separate watchlist file (backward compat). Loaded here at the run
// scope - NOT inside the collection block below - so its symbol
// strings outlive the `display` call at the end of `run`.
// `watch_list` (and `watch_prices`' keys) borrow these slices;
// freeing them before display - the previous behavior, where the
// load + `defer freeWatchlist` lived inside the collection block -
// rendered the file's watch symbols as freed-memory garbage on the
// CLI. The TUI is unaffected: it keeps watchlist symbols in its
// long-lived arena.
const wl_syms: ?[][]const u8 = if (watchlist_path) |wl_path|
cli.loadWatchlist(io, allocator, wl_path)
else
null;
defer cli.freeWatchlist(allocator, wl_syms);
// Collect watch symbols and their prices for display.
// Includes watch lots from portfolio + symbols from the watchlist file.
// `wl_syms` was loaded above, before the fetch, so these symbols are
// now in the fetch set rather than being priced from a cache nothing
// refreshes.
var watch_list: std.ArrayList([]const u8) = .empty;
defer watch_list.deinit(allocator);
var watch_prices = std.StringHashMap(f64).init(allocator);

View file

@ -908,6 +908,144 @@ pub const Portfolio = struct {
}
return result.toOwnedSlice(allocator);
}
/// Every symbol zfin fetches candles for: holdings, watch lots, the
/// separate `watchlist.srf`, and the benchmark pair.
///
/// This exists because "what do we keep fresh?" had **five** disjoint
/// answers, and things fell through the gaps between them. Holdings came
/// from `Portfolio.stockSymbols`, watch lots from a hand-rolled loop in
/// each caller, `watchlist.srf` from `cli.loadWatchlist` (TUI only - the
/// CLI loaded it for display and priced it from cache, so a
/// watchlist-only symbol went arbitrarily stale), and the benchmark pair
/// from two hardcoded `getCandles(sym, .{})` calls on a lazy path that
/// only ran when someone opened projections. Observed consequences: SPCX
/// sat 39 days out of date while sitting in `watchlist.srf`, and AGG was
/// unreachable by `--refresh-data=force` entirely.
///
/// One answer, shared by the CLI, the TUI and zfin-server, so a symbol
/// cannot be tracked by one and invisible to another.
///
/// **Every returned string is duplicated into `allocator`.** Unlike
/// `stockSymbols`, which borrows from the portfolio, the inputs here have
/// mixed and shorter lifetimes - notably a benchmark override lives in a
/// `[16]u8` field inside a stack `UserConfig`, so borrowing it would
/// dangle the moment that config went out of scope. Caller owns the
/// result; free the slices and the outer slice, or use an arena.
pub fn fetchedSymbols(
self: Portfolio,
allocator: std.mem.Allocator,
opts: struct {
/// Symbols from a separate `watchlist.srf`.
watchlist_syms: []const []const u8 = &.{},
/// Benchmark symbols (e.g. the projections stock/bond pair).
/// Passed as plain strings so this stays free of any dependency
/// on the projections config.
benchmarks: []const []const u8 = &.{},
},
) ![][]const u8 {
var seen = std.StringHashMap(void).init(allocator);
defer seen.deinit();
var result = std.ArrayList([]const u8).empty;
errdefer {
for (result.items) |s| allocator.free(s);
result.deinit(allocator);
}
// Owns nothing until the dupe succeeds, so `seen` keys borrow from
// `result` and stay valid for the whole build.
const add = struct {
fn f(
a: std.mem.Allocator,
set: *std.StringHashMap(void),
list: *std.ArrayList([]const u8),
sym: []const u8,
) !void {
if (sym.len == 0) return;
if (set.contains(sym)) return;
const owned = try a.dupe(u8, sym);
// The errdefer is scoped to the append and no further, on purpose.
// Left armed across the `set.put` below it would double-free:
// `list` already owns `owned` by then, and the caller's errdefer
// frees everything in `list`. An allocation-failure test caught
// exactly that as a segfault.
{
errdefer a.free(owned);
try list.append(a, owned);
}
try set.put(owned, {});
}
}.f;
// Holdings. Skips options, CDs, cash, and manual-price-only lots -
// see `stockSymbols` for why each is excluded.
const held = try self.stockSymbols(allocator);
defer allocator.free(held);
for (held) |s| try add(allocator, &seen, &result, s);
// `security_type::watch` lots inside the portfolio file.
for (self.lots) |lot| {
if (lot.security_type != .watch) continue;
try add(allocator, &seen, &result, lot.priceSymbol());
}
for (opts.watchlist_syms) |s| try add(allocator, &seen, &result, s);
for (opts.benchmarks) |s| try add(allocator, &seen, &result, s);
return result.toOwnedSlice(allocator);
}
/// Symbols to price that are NOT stock positions: `security_type::watch` lots
/// in the portfolio file, plus every entry from a separate `watchlist.srf`,
/// excluding anything already in `held`.
///
/// Separate from `fetchedSymbols` because the price loader takes holdings and
/// extras as two slices - it derives progress totals from the two counts - so a
/// single flat union does not fit there.
///
/// It lives here rather than inline in the command for a testability reason
/// that bit once already: `commands/portfolio.zig`'s `run` needs a live
/// `RunCtx`, a `DataService` and the network, so its tests only ever exercise
/// `display`. Set logic embedded in `run` is untestable by construction, and
/// the version that was embedded there had a bug - it never included
/// `watchlist.srf` at all, leaving SPCX 39 days stale.
///
/// Returned slices BORROW from `portfolio` and `watchlist_syms`; only the outer
/// slice is owned by the caller.
pub fn extraPriceSymbols(
self: Portfolio,
allocator: std.mem.Allocator,
held: []const []const u8,
watchlist_syms: []const []const u8,
) ![][]const u8 {
var seen = std.StringHashMap(void).init(allocator);
defer seen.deinit();
for (held) |s| try seen.put(s, {});
var out = std.ArrayList([]const u8).empty;
errdefer out.deinit(allocator);
for (self.lots) |lot| {
if (lot.security_type != .watch) continue;
const sym = lot.priceSymbol();
if (sym.len == 0 or seen.contains(sym)) continue;
try seen.put(sym, {});
try out.append(allocator, sym);
}
for (watchlist_syms) |sym| {
if (sym.len == 0 or seen.contains(sym)) continue;
try seen.put(sym, {});
try out.append(allocator, sym);
}
return out.toOwnedSlice(allocator);
}
/// Free a `fetchedSymbols` result.
pub fn freeFetchedSymbols(allocator: std.mem.Allocator, syms: [][]const u8) void {
for (syms) |s| allocator.free(s);
allocator.free(syms);
}
};
/// Check if a string looks like a CUSIP (9 alphanumeric characters).
@ -1299,9 +1437,9 @@ test "positions separates lots with different price_ratio" {
var lots = [_]Lot{
// Direct SPY holding, price_ratio = 1.0 (default)
.{ .symbol = "SPY", .shares = 717.34, .open_date = Date.fromYmd(2025, 2, 25), .open_price = 461.24, .account = "Tax Loss" },
.{ .symbol = "SPY", .shares = 100.0, .open_date = Date.fromYmd(2025, 2, 25), .open_price = 400.00, .account = "Sample Account" },
// Institutional S&P 500 CIT, uses SPY as ticker with a ratio
.{ .symbol = "NON40OR52", .shares = 5070.866, .open_date = Date.fromYmd(2026, 2, 26), .open_price = 97.24, .ticker = "SPY", .price_ratio = 0.2381, .account = "Fidelity Riley 401(k)" },
.{ .symbol = "NON40OR52", .shares = 5000.0, .open_date = Date.fromYmd(2026, 2, 26), .open_price = 90.00, .ticker = "SPY", .price_ratio = 0.25, .account = "Fidelity Riley 401(k)" },
};
var portfolio = Portfolio{ .lots = &lots, .allocator = allocator };
@ -1316,12 +1454,12 @@ test "positions separates lots with different price_ratio" {
for (pos) |p| {
if (p.price_ratio == 1.0) {
found_direct = true;
try std.testing.expectApproxEqAbs(@as(f64, 717.34), p.shares, 0.01);
try std.testing.expectApproxEqAbs(@as(f64, 100.0), p.shares, 0.01);
try std.testing.expectEqualStrings("SPY", p.lot_symbol);
} else {
found_institutional = true;
try std.testing.expectApproxEqAbs(@as(f64, 5070.866), p.shares, 0.01);
try std.testing.expectApproxEqAbs(@as(f64, 0.2381), p.price_ratio, 0.0001);
try std.testing.expectApproxEqAbs(@as(f64, 5000.0), p.shares, 0.01);
try std.testing.expectApproxEqAbs(@as(f64, 0.25), p.price_ratio, 0.0001);
try std.testing.expectEqualStrings("NON40OR52", p.lot_symbol);
}
}
@ -1752,3 +1890,226 @@ test "positionsAsOf reflects split_factor: effective shares, invariant basis, ef
// (the whole point - raw 100 * 120 would undercount 10x).
try std.testing.expectApproxEqAbs(@as(f64, 120000), positions[0].marketValue(120.0, false), 0.01);
}
// fetchedSymbols
/// Build a Portfolio from lots for the union tests. Lots borrow from the
/// caller; `fetchedSymbols` dupes everything it keeps, so that is safe.
fn testPortfolio(lots: []Lot) Portfolio {
return .{ .lots = lots, .allocator = std.testing.allocator };
}
test "fetchedSymbols: unions all four sources and dedups across them" {
const a = std.testing.allocator;
var lots = [_]Lot{
.{ .symbol = "AMZN", .shares = 10, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 200, .security_type = .stock },
// A ticker alias: the price symbol is what gets fetched, which is
// why SPY stayed fresh while AGG did not.
.{ .symbol = "NON40OR52", .ticker = "SPY", .shares = 5, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 90, .security_type = .stock },
.{ .symbol = "QTUM", .shares = 0, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 0, .security_type = .watch },
// Excluded by stockSymbols: manual price, no ticker alias.
.{ .symbol = "ORCBI", .shares = 3, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 10, .price = 11, .security_type = .stock },
// Excluded: not a stock or watch lot.
.{ .symbol = "CASHX", .shares = 1, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 1, .security_type = .cash },
};
const wl = [_][]const u8{ "SPCX", "AMZN" }; // AMZN duplicates a holding
const bm = [_][]const u8{ "SPY", "AGG" }; // SPY duplicates the alias above
const syms = try testPortfolio(&lots).fetchedSymbols(a, .{
.watchlist_syms = &wl,
.benchmarks = &bm,
});
defer Portfolio.freeFetchedSymbols(a, syms);
// AMZN, SPY, QTUM, SPCX, AGG - five distinct, no duplicates.
try std.testing.expectEqual(@as(usize, 5), syms.len);
for ([_][]const u8{ "AMZN", "SPY", "QTUM", "SPCX", "AGG" }) |want| {
var found = false;
for (syms) |s| if (std.mem.eql(u8, s, want)) {
found = true;
};
try std.testing.expect(found);
}
// Manual-price-only and cash lots stay out.
for (syms) |s| {
try std.testing.expect(!std.mem.eql(u8, s, "ORCBI"));
try std.testing.expect(!std.mem.eql(u8, s, "CASHX"));
}
}
test "fetchedSymbols: a watchlist-only symbol is included" {
// THE SPCX REGRESSION. It sat in watchlist.srf 39 days out of date
// because no CLI path ever put it in the fetch set - the CLI loaded
// the file for display and priced it from cache.
const a = std.testing.allocator;
var lots = [_]Lot{
.{ .symbol = "AMZN", .shares = 10, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 200, .security_type = .stock },
};
const wl = [_][]const u8{"SPCX"};
const syms = try testPortfolio(&lots).fetchedSymbols(a, .{ .watchlist_syms = &wl });
defer Portfolio.freeFetchedSymbols(a, syms);
var found = false;
for (syms) |s| if (std.mem.eql(u8, s, "SPCX")) {
found = true;
};
try std.testing.expect(found);
}
test "fetchedSymbols: a benchmark symbol held nowhere is still included" {
// THE AGG REGRESSION. AGG is not held and not watched - it is the bond
// half of the benchmark comparison, fetched only from a lazy
// projections path with hardcoded default FetchOptions, so
// `--refresh-data=force` could never reach it.
const a = std.testing.allocator;
var lots = [_]Lot{
.{ .symbol = "AMZN", .shares = 10, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 200, .security_type = .stock },
};
const bm = [_][]const u8{ "SPY", "AGG" };
const syms = try testPortfolio(&lots).fetchedSymbols(a, .{ .benchmarks = &bm });
defer Portfolio.freeFetchedSymbols(a, syms);
try std.testing.expectEqual(@as(usize, 3), syms.len);
var found_agg = false;
for (syms) |s| if (std.mem.eql(u8, s, "AGG")) {
found_agg = true;
};
try std.testing.expect(found_agg);
}
test "fetchedSymbols: empty and blank inputs produce no entries" {
const a = std.testing.allocator;
var lots = [_]Lot{};
const wl = [_][]const u8{""}; // blank line in watchlist.srf
const syms = try testPortfolio(&lots).fetchedSymbols(a, .{ .watchlist_syms = &wl });
defer Portfolio.freeFetchedSymbols(a, syms);
try std.testing.expectEqual(@as(usize, 0), syms.len);
}
test "fetchedSymbols: result outlives a stack-allocated benchmark override" {
// A projections override lives in a [16]u8 INSIDE the UserConfig
// struct, so borrowing it would dangle as soon as that config went out
// of scope. This is why the union dupes rather than borrows.
const a = std.testing.allocator;
var lots = [_]Lot{};
var syms: [][]const u8 = undefined;
{
var buf: [16]u8 = undefined;
@memcpy(buf[0..4], "VBIL");
const bm = [_][]const u8{buf[0..4]};
syms = try testPortfolio(&lots).fetchedSymbols(a, .{ .benchmarks = &bm });
@memset(&buf, 0xAA); // scribble over the source
}
defer Portfolio.freeFetchedSymbols(a, syms);
try std.testing.expectEqual(@as(usize, 1), syms.len);
try std.testing.expectEqualStrings("VBIL", syms[0]);
}
test "watchSymbols: watchlist.srf entries are included, holdings excluded" {
// THE SPCX BUG, at the layer where it actually lived. The version embedded
// in `commands/portfolio.zig` never looked at watchlist.srf at all, so a
// watchlist-only symbol was displayed from whatever the cache happened to
// hold - 39 days old, in SPCX's case - and never fetched.
const a = std.testing.allocator;
var lots = [_]Lot{
.{ .symbol = "AMZN", .shares = 10, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 200, .security_type = .stock },
.{ .symbol = "QTUM", .shares = 0, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 0, .security_type = .watch },
};
const held = [_][]const u8{"AMZN"};
const wl = [_][]const u8{ "SPCX", "QTUM", "AMZN", "" };
const out = try testPortfolio(&lots).extraPriceSymbols(a, &held, &wl);
defer a.free(out);
// QTUM once (watch lot, deduped against the watchlist), SPCX from the
// file. AMZN is held so it belongs to the other slice, and the blank
// line is dropped.
try std.testing.expectEqual(@as(usize, 2), out.len);
try std.testing.expectEqualStrings("QTUM", out[0]);
try std.testing.expectEqualStrings("SPCX", out[1]);
}
test "watchSymbols: a ticker alias on a watch lot is priced by its alias" {
const a = std.testing.allocator;
var lots = [_]Lot{
.{ .symbol = "NON40OR52", .ticker = "SPY", .shares = 0, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 0, .security_type = .watch },
};
const out = try testPortfolio(&lots).extraPriceSymbols(a, &.{}, &.{});
defer a.free(out);
try std.testing.expectEqual(@as(usize, 1), out.len);
try std.testing.expectEqualStrings("SPY", out[0]);
}
test "watchSymbols: no watch lots and no watchlist yields an empty set" {
const a = std.testing.allocator;
var lots = [_]Lot{
.{ .symbol = "AMZN", .shares = 10, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 200, .security_type = .stock },
};
const held = [_][]const u8{"AMZN"};
const out = try testPortfolio(&lots).extraPriceSymbols(a, &held, &.{});
defer a.free(out);
try std.testing.expectEqual(@as(usize, 0), out.len);
}
/// OOM-path wrapper for `checkAllAllocationFailures`.
fn fetchedSymbolsOom(a: std.mem.Allocator, lots: []Lot, wl: []const []const u8, bm: []const []const u8) !void {
const syms = try (Portfolio{ .lots = lots, .allocator = a }).fetchedSymbols(a, .{
.watchlist_syms = wl,
.benchmarks = bm,
});
Portfolio.freeFetchedSymbols(a, syms);
}
fn watchSymbolsOom(a: std.mem.Allocator, lots: []Lot, held: []const []const u8, wl: []const []const u8) !void {
const out = try (Portfolio{ .lots = lots, .allocator = a }).extraPriceSymbols(a, held, wl);
a.free(out);
}
test "fetchedSymbols/watchSymbols: every allocation-failure path unwinds cleanly" {
// Covers the errdefer arms, which are otherwise unreachable: a partial
// build must free the strings it already duped, and the inner arm must
// free a dupe whose append then failed.
var lots = [_]Lot{
.{ .symbol = "AMZN", .shares = 10, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 200, .security_type = .stock },
.{ .symbol = "QTUM", .shares = 0, .open_date = Date.fromYmd(2026, 1, 2), .open_price = 0, .security_type = .watch },
};
const wl = [_][]const u8{"SPCX"};
const bm = [_][]const u8{ "SPY", "AGG" };
const held = [_][]const u8{"AMZN"};
try std.testing.checkAllAllocationFailures(
std.testing.allocator,
fetchedSymbolsOom,
.{ &lots, @as([]const []const u8, &wl), @as([]const []const u8, &bm) },
);
try std.testing.checkAllAllocationFailures(
std.testing.allocator,
watchSymbolsOom,
.{ &lots, @as([]const []const u8, &held), @as([]const []const u8, &wl) },
);
}
test "extraPriceSymbols: order is stable - watch lots first, then the watchlist file" {
// `PortfolioData.load` used to build this through a StringHashMap, so
// iteration order - and therefore the "[5/28] Loading X" progress order -
// varied run to run for no reason. Callers may now rely on the order.
const a = std.testing.allocator;
var lots = [_]Lot{
.{ .symbol = "AAPL", .shares = 10, .open_date = Date.fromYmd(2024, 1, 2), .open_price = 100, .security_type = .stock },
.{ .symbol = "TSLA", .shares = 0, .open_date = Date.fromYmd(2024, 1, 2), .open_price = 0, .security_type = .watch },
.{ .symbol = "NVDA", .shares = 0, .open_date = Date.fromYmd(2024, 1, 2), .open_price = 0, .security_type = .watch },
};
const held = [_][]const u8{"AAPL"};
const wl = [_][]const u8{ "MSFT", "QTUM" };
// Run twice: a hash-order build would be free to differ between calls.
for (0..2) |_| {
const out = try testPortfolio(&lots).extraPriceSymbols(a, &held, &wl);
defer a.free(out);
try std.testing.expectEqual(@as(usize, 4), out.len);
try std.testing.expectEqualStrings("TSLA", out[0]);
try std.testing.expectEqualStrings("NVDA", out[1]);
try std.testing.expectEqualStrings("MSFT", out[2]);
try std.testing.expectEqualStrings("QTUM", out[3]);
}
}

View file

@ -576,7 +576,36 @@ 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 `syms`' entries 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);
for (syms) |s| {
// Only symbols actually held - `syms` is the held set, and a metadata
// row can name a symbol the portfolio no longer holds.
//
// Driven from `syms` with a hash lookup, not from the map's keys with a
// linear scan of `syms` per key: same result, O(n) instead of O(n*m),
// and the output order follows the held set instead of hash iteration
// order.
// Bailing on OOM rather than propagating: split adjustment is opt-in
// enrichment, and the un-adjusted portfolio is still correct.
if (cutovers.contains(s)) wanted.append(allocator, s) catch return;
}
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 +1035,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;

View file

@ -630,8 +630,8 @@ pub const DataService = struct {
// wall-clock required: stamp the candle-meta freshness boundary
// (market-aware next post-close / NAV-availability time).
const now_s = std.Io.Timestamp.now(self.io, .real).toSeconds();
const expires = market.nextCandleExpiry(now_s, market.classify(symbol));
s.cacheCandles(symbol, triple.candles, .tiingo, 0, expires);
const kind = market.classify(symbol);
s.cacheCandles(symbol, triple.candles, .tiingo, 0, expiryAfterFetch(now_s, kind, triple.candles));
}
// Dividends and splits use the supplement write path: Tiingo's
// view merges into existing (typically Polygon-sourced) records
@ -965,8 +965,11 @@ pub const DataService = struct {
if (s.read(self.allocator, Candle, symbol, null, .any)) |r|
return .{ .data = r.data, .source = .cached, .timestamp = std.Io.Timestamp.now(self.io, .real).toSeconds(), .allocator = self.allocator };
} else {
// Append new candles to existing file + update meta, reset fail_count
s.appendCandles(symbol, new_candles, result.provider, 0, expires);
// Append new candles to existing file + update meta, reset fail_count.
// TTL via `expiryAfterFetch`, NOT the precomputed
// next-boundary `expires`: getting a bar back does not
// mean getting the RIGHT bar back.
s.appendCandles(symbol, new_candles, result.provider, 0, expiryAfterFetch(now_s, kind, new_candles));
if (s.read(self.allocator, Candle, symbol, null, .any)) |r| {
self.allocator.free(new_candles);
return .{ .data = r.data, .source = .fetched, .timestamp = std.Io.Timestamp.now(self.io, .real).toSeconds(), .allocator = self.allocator };
@ -2239,13 +2242,32 @@ 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 = self.config.timing;
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 +2277,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 +2330,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 +2355,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;
}
@ -3104,6 +3144,39 @@ pub const DataService = struct {
// Write to local cache
var s = self.store();
// Never let the shared cache move a symbol BACKWARDS. The server's
// bytes are written verbatim, `#!expires=` included, so its view of
// freshness becomes the client's - and if the server's copy carries an
// older bar than the one already here, an unconditional write replaces
// good local data with worse and stamps it authoritative.
//
// Observed: the server's cron fetched at 17:00 ET, some symbols got that
// session's bar and some did not, and every one of them was stamped
// fresh until the next boundary. A client that had already fetched the
// newer bar would have had it overwritten and then believed the older
// one for a full day.
//
// Only candle data can be ordered this way, so only candle data is
// guarded; everything else falls through unchanged.
if (isCandleType(data_type)) {
if (serverBarRegression(&s, symbol, response.body)) |reg| {
// WARN, not debug. A shared cache should never be behind a
// client that draws from it: it is the tier with the refresh
// cron and the provider budget. When it happens, something on
// the server side has stopped keeping up, and every other client
// is being handed the same stale bar - so this is an operator
// event, not a diagnostic detail. Logged at debug, it was
// invisible in exactly the builds people install.
log.warn(
"{s}: shared cache is BEHIND this client for {s} - it offered {f}, local copy has {f}. Refused the sync rather than move the cache backwards; the server needs a refresh (see `zfin cache stale`).",
.{ symbol, @tagName(data_type), reg.incoming, reg.local },
);
log.debug("{s}: tryOneSync finished ({s}) result=ok elapsed_ms={d}", .{ symbol, @tagName(data_type), @divTrunc(std.Io.Timestamp.now(self.io, .awake).nanoseconds - t_start, std.time.ns_per_ms) });
return .ok;
}
}
s.writeRaw(symbol, data_type, response.body) catch |err| {
log.debug("{s}: failed to write synced {s} to cache: {s}", .{ symbol, @tagName(data_type), @errorName(err) });
log.debug("{s}: tryOneSync finished ({s}) result=net_err elapsed_ms={d}", .{ symbol, @tagName(data_type), @divTrunc(std.Io.Timestamp.now(self.io, .awake).nanoseconds - t_start, std.time.ns_per_ms) });
@ -3115,6 +3188,88 @@ pub const DataService = struct {
}
/// Sync candle data (both daily and meta) from the server.
/// Do these bytes describe candle data, i.e. data with a newest-bar date
/// that can be compared for age?
fn isCandleType(data_type: cache.DataType) bool {
return data_type == .candles_daily or data_type == .candles_meta;
}
/// Both dates, when writing `body` would replace the local candle data with
/// an OLDER bar. Null otherwise.
///
/// Returns the pair rather than a bool because the caller has to report
/// them: "the shared cache is behind you" is only actionable with the two
/// dates attached.
///
/// Null whenever the question cannot be answered - no local copy, an
/// unparseable body, no dates on either side - so an unknown never blocks a
/// sync. The guard only fires on a definite regression.
fn serverBarRegression(
s: *cache.Store,
symbol: []const u8,
body: []const u8,
) ?struct { local: Date, incoming: Date } {
const local = s.readCandleMeta(symbol) orelse return null;
const incoming = newestDateIn(body) orelse return null;
if (!incoming.lessThan(local.meta.last_date)) return null;
return .{ .local = local.meta.last_date, .incoming = incoming };
}
/// Newest `last_date::` or `date::` value in an SRF body.
///
/// Deliberately a scan for the maximum rather than a parse: `candles_meta`
/// carries one `last_date`, `candles_daily` carries a `date` per bar, and
/// the ordering of the latter is not something this check should assume.
fn newestDateIn(body: []const u8) ?Date {
var best: ?Date = null;
for ([_][]const u8{ "last_date::", "date::" }) |key| {
var rest = body;
while (std.mem.indexOf(u8, rest, key)) |idx| {
const start = idx + key.len;
rest = rest[start..];
const end = std.mem.indexOfAny(u8, rest, ",\n\r") orelse rest.len;
if (Date.parse(rest[0..end])) |d| {
if (best == null or best.?.lessThan(d)) best = d;
} else |_| {}
}
}
return best;
}
/// TTL to stamp after a fetch that DID return candles.
///
/// A successful fetch is not necessarily a caught-up one. An incremental
/// request asks for everything after the last cached bar, so it can return
/// the bar missed in a PREVIOUS session while the current session's bar is
/// still unposted - a success by every measure the fetch itself can see.
///
/// Stamping the next-boundary TTL there writes the session off after a
/// partial catch-up, and for a symbol whose provider posts later than
/// `provider_lag_grace_s` that never converges: each session it recovers the
/// previous session's bar and is immediately a session behind again.
/// Observed on AMZN and NKE, which sat exactly one session behind for days
/// while the bar had been sitting at the provider the whole time -
///
/// Fri 18:36 zero bars, 99min past grace -> next boundary (Mon 16:55)
/// Mon 17:00 returned FRIDAY's bar -> next boundary (Tue 16:55) <- here
/// Tue 16:55 returns MONDAY's bar -> next boundary (Wed 16:55)
///
/// `staleCandleExpiry` asks the question the zero-bar branch already asks -
/// is the newest bar we now hold the one that should be available? - and
/// returns the next boundary when it is, so a genuinely caught-up fetch
/// behaves exactly as before.
///
/// Takes the maximum rather than the last element: provider ordering is not
/// something this decision should depend on.
fn expiryAfterFetch(now_s: i64, kind: market.InstrumentKind, candles: []const Candle) i64 {
if (candles.len == 0) return market.nextCandleExpiry(now_s, kind);
var newest = candles[0].date;
for (candles[1..]) |c| {
if (newest.lessThan(c.date)) newest = c.date;
}
return market.staleCandleExpiry(now_s, kind, newest);
}
fn syncCandlesFromServer(self: *DataService, symbol: []const u8) bool {
const daily = self.syncFromServer(symbol, .candles_daily);
const meta = self.syncFromServer(symbol, .candles_meta);
@ -3251,6 +3406,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();
}
@ -4862,3 +5025,116 @@ test "earningsNeedsRefresh: chase window is inclusive at the boundary" {
const past_window = [_]EarningsEvent{.{ .date = Date.fromYmd(2026, 6, 11), .estimate = 1.0 }};
try std.testing.expect(!DataService.earningsNeedsRefresh(&past_window, today, 14));
}
test "newestDateIn: picks the maximum across candles_meta and candles_daily shapes" {
// `candles_meta` carries one `last_date`; `candles_daily` carries a `date`
// per bar, in an order this check must not assume.
const meta = "#!srfv1\n#!expires=1\nlast_close:num:100.00,last_date::2026-08-07,provider::tiingo\n";
try std.testing.expect(DataService.newestDateIn(meta).?.eql(Date.fromYmd(2026, 8, 7)));
// Deliberately out of order.
const daily = "#!srfv1\ndate::2026-08-05,close:num:1\ndate::2026-08-07,close:num:3\ndate::2026-08-06,close:num:2\n";
try std.testing.expect(DataService.newestDateIn(daily).?.eql(Date.fromYmd(2026, 8, 7)));
// Nothing parseable -> null, so the guard cannot fire on garbage.
try std.testing.expectEqual(@as(?Date, null), DataService.newestDateIn("#!srfv1\n"));
try std.testing.expectEqual(@as(?Date, null), DataService.newestDateIn("last_date::not-a-date\n"));
}
test "isCandleType: only candle data is age-comparable" {
try std.testing.expect(DataService.isCandleType(.candles_daily));
try std.testing.expect(DataService.isCandleType(.candles_meta));
// Dividends, splits, options and the rest have no single newest-bar date,
// so they sync unguarded.
try std.testing.expect(!DataService.isCandleType(.dividends));
try std.testing.expect(!DataService.isCandleType(.splits));
try std.testing.expect(!DataService.isCandleType(.classification));
}
test "serverBarRegression: blocks a regression and reports both dates" {
const allocator = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const dir_path = try tmp.dir.realPathFileAlloc(io, ".", allocator);
defer allocator.free(dir_path);
var s = cache.Store.init(io, allocator, dir_path);
// Local copy holds Monday's bar.
s.updateCandleMeta("AAPL", 100.0, Date.fromYmd(2026, 8, 10), .tiingo, 0, 9999999999);
const older = "#!srfv1\nlast_close:num:99.00,last_date::2026-08-07,provider::tiingo\n";
const same = "#!srfv1\nlast_close:num:99.00,last_date::2026-08-10,provider::tiingo\n";
const newer = "#!srfv1\nlast_close:num:99.00,last_date::2026-08-11,provider::tiingo\n";
// THE REGRESSION THIS GUARDS. The shared cache's cron fetched at 17:00 ET;
// some symbols got that session's bar and some did not, and all were
// stamped fresh until the next boundary. Written unconditionally, the older
// body replaces good local data and is then believed for a full day.
const reg = DataService.serverBarRegression(&s, "AAPL", older) orelse
return error.ExpectedRegression;
// Both dates come back, because the warning is only actionable with them.
try std.testing.expect(reg.local.eql(Date.fromYmd(2026, 8, 10)));
try std.testing.expect(reg.incoming.eql(Date.fromYmd(2026, 8, 7)));
try std.testing.expectEqual(@as(?@TypeOf(reg), null), DataService.serverBarRegression(&s, "AAPL", same));
try std.testing.expectEqual(@as(?@TypeOf(reg), null), DataService.serverBarRegression(&s, "AAPL", newer));
// Unknowns never block: no local copy, and an unparseable body.
try std.testing.expectEqual(@as(?@TypeOf(reg), null), DataService.serverBarRegression(&s, "NOLOCAL", older));
try std.testing.expectEqual(@as(?@TypeOf(reg), null), DataService.serverBarRegression(&s, "AAPL", "#!srfv1\n"));
}
test "expiryAfterFetch: a partial catch-up keeps retrying instead of writing off the session" {
// THE REGRESSION THIS GUARDS, with Monday's real numbers. AMZN was fetched
// Mon 2026-08-10 at 17:00 ET holding Thursday's bar, and the incremental
// request returned FRIDAY's. One new candle - a success - so the old code
// stamped the next-boundary TTL and stopped looking until Tue 16:55, while
// Monday's bar showed up at the provider minutes later. Repeat daily and the
// symbol is permanently one session behind.
const kind = market.InstrumentKind.equity;
// Mon 2026-08-10 17:00 ET, five minutes past the 16:55 equity target.
const mon_1700 = Date.fromYmd(2026, 8, 10).toEpoch() + 21 * std.time.s_per_hour;
var friday_only = [_]Candle{.{ .date = Date.fromYmd(2026, 8, 7), .open = 1, .high = 1, .low = 1, .close = 1, .adj_close = 1, .volume = 1 }};
const partial = DataService.expiryAfterFetch(mon_1700, kind, &friday_only);
try std.testing.expectEqual(mon_1700 + market.short_retry_s, partial);
try std.testing.expect(partial != market.nextCandleExpiry(mon_1700, kind));
// And the case that must NOT change: a fetch that actually caught up gets
// the full next-boundary TTL exactly as before.
var through_monday = [_]Candle{
.{ .date = Date.fromYmd(2026, 8, 7), .open = 1, .high = 1, .low = 1, .close = 1, .adj_close = 1, .volume = 1 },
.{ .date = Date.fromYmd(2026, 8, 10), .open = 1, .high = 1, .low = 1, .close = 1, .adj_close = 1, .volume = 1 },
};
try std.testing.expectEqual(
market.nextCandleExpiry(mon_1700, kind),
DataService.expiryAfterFetch(mon_1700, kind, &through_monday),
);
}
test "expiryAfterFetch: takes the maximum, not the last element" {
// Provider ordering is not something this decision should depend on.
const kind = market.InstrumentKind.equity;
const mon_1700 = Date.fromYmd(2026, 8, 10).toEpoch() + 21 * std.time.s_per_hour;
var descending = [_]Candle{
.{ .date = Date.fromYmd(2026, 8, 10), .open = 1, .high = 1, .low = 1, .close = 1, .adj_close = 1, .volume = 1 },
.{ .date = Date.fromYmd(2026, 8, 7), .open = 1, .high = 1, .low = 1, .close = 1, .adj_close = 1, .volume = 1 },
};
// Newest is 08-10 even though it is first, so this is caught up.
try std.testing.expectEqual(
market.nextCandleExpiry(mon_1700, kind),
DataService.expiryAfterFetch(mon_1700, kind, &descending),
);
}
test "expiryAfterFetch: an empty slice falls back to the next boundary" {
// Both call sites are guarded by a non-empty check; this keeps the helper
// safe if a third one is ever added.
const kind = market.InstrumentKind.equity;
const mon_1700 = Date.fromYmd(2026, 8, 10).toEpoch() + 21 * std.time.s_per_hour;
try std.testing.expectEqual(
market.nextCandleExpiry(mon_1700, kind),
DataService.expiryAfterFetch(mon_1700, kind, &.{}),
);
}

View file

@ -420,10 +420,9 @@ pub const tab = struct {
for (wl) |sym| watch_syms.append(a, sym) catch |err| std.log.debug("watch_syms append: {t}", .{err});
}
// Symbols to quote live = watchlist + held stock symbols. Held
// symbols are duped into the arena (see above).
// Symbols to quote live = held + everything on either watch source.
// Held symbols are duped into the arena (see above).
var quote_syms: std.ArrayList([]const u8) = .empty;
for (watch_syms.items) |sym| quote_syms.append(a, sym) catch |err| std.log.debug("quote_syms append: {t}", .{err});
if (app.portfolio.file) |pf| {
if (pf.stockSymbols(app.allocator)) |hs| {
defer app.allocator.free(hs); // outer slice; strings duped into arena
@ -431,6 +430,18 @@ pub const tab = struct {
const dup = a.dupe(u8, sym) catch continue;
quote_syms.append(a, dup) catch |err| std.log.debug("quote_syms append: {t}", .{err});
}
// The same union the price load uses, so live quotes and
// candles cover the same set. Previously this took only
// `watchlist.srf` and skipped the portfolio's own `.watch`
// lots, so those rows silently fell back to the prior close
// while watchlist rows updated intraday - the same
// two-sources-of-watch-symbols split that left a
// watchlist-only symbol unfetched on the CLI, mirrored.
// Borrowed, not duped: these point into `pf.lots` and
// `app.watchlist`, both stable across this call.
if (pf.extraPriceSymbols(a, hs, watch_syms.items)) |extra| {
for (extra) |sym| quote_syms.append(a, sym) catch |err| std.log.debug("quote_syms append: {t}", .{err});
} else |err| std.log.debug("extraPriceSymbols for live quotes: {t}", .{err});
} else |err| std.log.debug("stockSymbols for live quotes: {t}", .{err});
}