load dividends with portfolio
This commit is contained in:
parent
d434395daa
commit
ef97842807
3 changed files with 161 additions and 4 deletions
|
|
@ -87,6 +87,7 @@ const Dividend = zfin.Dividend;
|
|||
const AccountMap = zfin.analysis.AccountMap;
|
||||
const ClassificationMap = zfin.classification.ClassificationMap;
|
||||
const DataService = zfin.DataService;
|
||||
const FetchOptions = zfin.FetchOptions;
|
||||
|
||||
// ── Public types ──────────────────────────────────────────────
|
||||
|
||||
|
|
@ -258,6 +259,13 @@ candles_arena: ArenaAllocator,
|
|||
io: std.Io,
|
||||
svc: *DataService,
|
||||
|
||||
/// Cache policy for the current load. Candles and splits are warmed
|
||||
/// synchronously inside `load()`, where `LoadOptions` is still in scope, so
|
||||
/// only the dividends warm needs this: it runs inside a worker (to keep
|
||||
/// Polygon's 4/min bucket off the first-paint path) and a worker cannot see
|
||||
/// `opts`. Reset on every load/reload.
|
||||
fetch_opts: FetchOptions = .{},
|
||||
|
||||
/// Parsed portfolio file path(s). Arena-owned `[]const u8`
|
||||
/// strings; arena-owned outer slice. `paths[0]` is the anchor
|
||||
/// for sibling-file derivation. Empty before the first load.
|
||||
|
|
@ -598,6 +606,12 @@ pub fn load(
|
|||
self.account_map_data = null;
|
||||
self.classification_map_data = null;
|
||||
|
||||
// Capture the cache policy for the workers that fetch (dividends).
|
||||
self.fetch_opts = .{
|
||||
.force_refresh = opts.force_refresh,
|
||||
.skip_network = opts.skip_network,
|
||||
};
|
||||
|
||||
// candles_data lives in candles_arena and survives across
|
||||
// reloads - kept entries are reused, only new symbols hit
|
||||
// the cache. force_refresh wipes it wholesale.
|
||||
|
|
@ -988,10 +1002,40 @@ fn snapshotsWorker(self: *PortfolioData, as_of: Date, positions: []const zfin.Po
|
|||
);
|
||||
}
|
||||
|
||||
/// Warm the dividend cache, then read it into the map.
|
||||
///
|
||||
/// The warm lives here rather than in `load()` on purpose. Polygon serves
|
||||
/// dividends from a 4/min bucket, so a cold cache with no `ZFIN_SERVER`
|
||||
/// could take minutes - unacceptable in front of a first paint. Inside the
|
||||
/// worker, that cost only lands when a consumer calls `dividends()` and
|
||||
/// blocks on the future, which today is the review tab alone. `load()`
|
||||
/// still initiates it, matching every other datum.
|
||||
///
|
||||
/// Unconditional: unlike split adjustment, dividends have no per-symbol
|
||||
/// opt-in to gate on.
|
||||
fn dividendsWorker(self: *PortfolioData, delay_ms: usize) void {
|
||||
self.io.sleep(.fromMilliseconds(@intCast(delay_ms)), .real) catch return;
|
||||
const summary_ref = self.summary orelse return;
|
||||
const arena_alloc = self.allocator();
|
||||
|
||||
// Warm first: `getCachedDividends` below cannot populate the cache, so
|
||||
// without this the map silently omits every symbol never fetched by a
|
||||
// per-symbol command. The warm is best-effort - on allocation failure
|
||||
// we skip it and still read whatever is already cached, because a
|
||||
// failed optimization must not cost us the data we already have.
|
||||
//
|
||||
// Skipped entirely under `skip_network`: there is nothing to warm, the
|
||||
// read loop below already covers the cache, and going through
|
||||
// `getDividends` would emit a "stale-cached returned (skip_network)"
|
||||
// info line per symbol for no benefit.
|
||||
warm: {
|
||||
if (self.fetch_opts.skip_network) break :warm;
|
||||
const div_syms = arena_alloc.alloc([]const u8, summary_ref.allocations.len) catch break :warm;
|
||||
for (summary_ref.allocations, div_syms) |alloc, *dst| dst.* = alloc.symbol;
|
||||
self.io.checkCancel() catch return;
|
||||
self.svc.loadAllDividends(div_syms, self.fetch_opts);
|
||||
}
|
||||
|
||||
var map = std.StringHashMap([]const Dividend).init(arena_alloc);
|
||||
for (summary_ref.allocations) |alloc| {
|
||||
self.io.checkCancel() catch return;
|
||||
|
|
|
|||
|
|
@ -191,9 +191,12 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
|
|||
var acct_map_opt: ?zfin.analysis.AccountMap = svc.loadAccountMap(allocator, anchor_path);
|
||||
defer if (acct_map_opt) |*am| am.deinit();
|
||||
|
||||
// Per-symbol cached dividends so total-return windows include
|
||||
// dividend reinvestment when available. Cached-only - no
|
||||
// network - to keep the command fast on large portfolios.
|
||||
// Per-symbol dividends so total-return windows include dividend
|
||||
// reinvestment. Warmed first: `getCachedDividends` cannot populate the
|
||||
// cache, so reading it alone silently omitted every symbol the user had
|
||||
// never inspected with `divs`/`perf` - and an omitted symbol degrades to
|
||||
// a price-only return with no indication. Honors --refresh-data, so
|
||||
// `never` still reads cache only.
|
||||
var dividend_map = std.StringHashMap([]const zfin.Dividend).init(allocator);
|
||||
defer {
|
||||
var it = dividend_map.iterator();
|
||||
|
|
@ -202,6 +205,15 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
|
|||
}
|
||||
dividend_map.deinit();
|
||||
}
|
||||
{
|
||||
var div_syms = try std.ArrayList([]const u8).initCapacity(
|
||||
allocator,
|
||||
pf_data.summary.allocations.len,
|
||||
);
|
||||
defer div_syms.deinit(allocator);
|
||||
for (pf_data.summary.allocations) |a| div_syms.appendAssumeCapacity(a.symbol);
|
||||
svc.loadAllDividends(div_syms.items, cli.fetchOptionsFromPolicy(ctx.globals.refresh_policy));
|
||||
}
|
||||
for (pf_data.summary.allocations) |a| {
|
||||
if (svc.getCachedDividends(allocator, a.symbol)) |divs| {
|
||||
try dividend_map.put(a.symbol, divs.data);
|
||||
|
|
|
|||
103
src/service.zig
103
src/service.zig
|
|
@ -510,7 +510,8 @@ pub const DataService = struct {
|
|||
if (err == error.RateLimited) {
|
||||
// Wait and retry once
|
||||
self.rateLimitBackoff();
|
||||
const retried = self.fetchFromProvider(T, symbol) catch {
|
||||
const retried = self.fetchFromProvider(T, symbol) catch |retry_err| {
|
||||
log.warn("{s}: {s} fetch failed after rate-limit retry: {t}", .{ symbol, @tagName(data_type), retry_err });
|
||||
return DataError.FetchFailed;
|
||||
};
|
||||
s.writeWithSource(T, symbol, retried, data_type.ttl(), sourceHintFor(T));
|
||||
|
|
@ -521,8 +522,18 @@ pub const DataService = struct {
|
|||
// Transient failures (network, 5xx, auth misconfig, parse
|
||||
// error) propagate as FetchFailed without poisoning the
|
||||
// cache, so the next call retries naturally.
|
||||
//
|
||||
// Log the provider's own error either way: the typed return
|
||||
// collapses to FetchFailed, so this line is the only place the
|
||||
// distinction between RateLimited, Unauthorized and NotFound
|
||||
// survives. Callers that swallow failures per symbol (see
|
||||
// `loadAllDividends`) depend on it.
|
||||
if (isPermanentProviderFailure(err)) {
|
||||
// The normal "this symbol has no data of this type" outcome.
|
||||
log.info("{s}: {s} unavailable: {t}", .{ symbol, @tagName(data_type), err });
|
||||
s.writeNegative(symbol, data_type);
|
||||
} else {
|
||||
log.warn("{s}: {s} fetch failed: {t}", .{ symbol, @tagName(data_type), err });
|
||||
}
|
||||
return DataError.FetchFailed;
|
||||
};
|
||||
|
|
@ -3238,6 +3249,44 @@ pub const DataService = struct {
|
|||
}
|
||||
return corpus;
|
||||
}
|
||||
|
||||
/// Warm the dividend cache for each symbol. Warm-only: nothing is
|
||||
/// returned, because the sole consumer (`PortfolioData`'s dividends
|
||||
/// worker) reads the cache immediately afterward. That is the whole
|
||||
/// difference from `loadAllSplits`, whose corpus feeds `enrichSplits`
|
||||
/// inline.
|
||||
///
|
||||
/// Why this exists: dividends were the only per-symbol data type with
|
||||
/// a portfolio-wide reader and no portfolio-wide writer. Candles are
|
||||
/// warmed by `loadAllPrices` and splits by `loadAllSplits`, but nothing
|
||||
/// warmed dividends, so `getCachedDividends` read a cache that only
|
||||
/// per-symbol commands (`divs`, `perf`) had ever populated. The visible
|
||||
/// symptom was `views/review.zig` silently reporting price-only
|
||||
/// trailing returns for every symbol the user had never inspected
|
||||
/// individually.
|
||||
///
|
||||
/// Sequential on purpose. The expensive phase is the provider, and
|
||||
/// Polygon serves both dividends and splits from one 4/min bucket, so
|
||||
/// concurrency cannot speed that up; the server-sync phase is a
|
||||
/// sub-second serial cost for a normal portfolio. Rate limiting needs
|
||||
/// no wiring here - it lives in the provider, so every `getDividends`
|
||||
/// call is already throttled.
|
||||
///
|
||||
/// Failures are swallowed per symbol: a missing dividend history
|
||||
/// degrades a total return to price-only, which is not worth failing
|
||||
/// a whole portfolio load over. `fetchCached` logs the provider's own
|
||||
/// error (rate limit vs auth vs no-such-data) before collapsing it to
|
||||
/// `FetchFailed`, so a swallowed failure is still diagnosable.
|
||||
pub fn loadAllDividends(
|
||||
self: *DataService,
|
||||
syms: []const []const u8,
|
||||
opts: FetchOptions,
|
||||
) void {
|
||||
for (syms) |sym| {
|
||||
const fr = self.getDividends(sym, opts) catch continue;
|
||||
fr.deinit();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// ── Tests ─────────────────────────────────────────────────────────
|
||||
|
|
@ -3642,6 +3691,58 @@ test "fetchCached offline mode returns stale-cached data" {
|
|||
try std.testing.expectEqual(Source.cached, result.source);
|
||||
}
|
||||
|
||||
test "loadAllDividends: honors skip_network for every symbol, and one miss does not abort the rest" {
|
||||
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 config = Config{ .cache_dir = dir_path };
|
||||
var svc = DataService.init(io, allocator, config);
|
||||
defer svc.deinit();
|
||||
|
||||
// TSTB is cached; TSTA is absent. TSTA comes first, so if a per-symbol
|
||||
// failure aborted the loop, TSTB would never be reached.
|
||||
var divs = [_]Dividend{
|
||||
.{ .ex_date = Date.fromYmd(2026, 3, 15), .amount = 0.50, .type = .regular },
|
||||
};
|
||||
var store = svc.store();
|
||||
store.write(Dividend, "TSTB", divs[0..], cache.DataType.dividends.ttl());
|
||||
|
||||
// The whole loop must stay offline, not just the first symbol.
|
||||
svc.panic_on_network_attempt = true;
|
||||
svc.loadAllDividends(&.{ "TSTA", "TSTB" }, .{ .skip_network = true });
|
||||
|
||||
// The cached symbol survives the pass.
|
||||
const b = svc.getCachedDividends(allocator, "TSTB") orelse return error.TestUnexpectedResult;
|
||||
defer b.deinit();
|
||||
try std.testing.expectEqual(@as(usize, 1), b.data.len);
|
||||
|
||||
// The absent symbol stays absent - a warm must not leave a negative or
|
||||
// empty entry behind that would mask a later real fetch.
|
||||
try std.testing.expect(svc.getCachedDividends(allocator, "TSTA") == null);
|
||||
}
|
||||
|
||||
test "loadAllDividends: empty symbol list is a no-op" {
|
||||
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 config = Config{ .cache_dir = dir_path };
|
||||
var svc = DataService.init(io, allocator, config);
|
||||
defer svc.deinit();
|
||||
|
||||
// No symbols means no fetches, so this must hold even with network
|
||||
// otherwise allowed.
|
||||
svc.panic_on_network_attempt = true;
|
||||
svc.loadAllDividends(&.{}, .{});
|
||||
}
|
||||
|
||||
test "getQuote offline mode returns FetchFailed (quotes never cached)" {
|
||||
const allocator = std.testing.allocator;
|
||||
const io = std.testing.io;
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue