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

json feeds do not publish NAV for Friday until Monday :(. This will derive that value temporarily
All checks were successful
Generic zig build / build (push) Successful in 1m16s
Details
Generic zig build / deploy (push) Successful in 16s
Details

Browse source
This commit is contained in:
Emil Lerch 2026-08-31 10:47:24 -07:00
parent e18253436a
commit 5f47c95339
Signed by: lobo
GPG key ID: A7B62D657EF764F8
5 changed files with 1157 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

202
README.md
View file

@ -7,6 +7,7 @@ portfolios held in the finance repo: **ORCBI** and **ORC42**.
zfin-vestwell reconstruct # rebuild data/ORCBI.srf and data/ORC42.srf
zfin-vestwell verify # check them against every evidence source
zfin-vestwell record # append today's published value to the log
zfin-vestwell project # fill sessions the feed has not published yet
zfin-vestwell assemble --out DIR # write zfin cache files for it to serve
```
@ -102,6 +103,22 @@ prior session's value.
Those values are rounded to 2dp at source, which is already +/-0.005 of noise.
**One session forward, unpinned.** Both figures above measure *interpolation* --
error in the middle of a gap whose far end is anchored. `project` has no far end,
so neither bounds it. Walking from a published value one session forward and
comparing against the next published value:
| Fund | Projected | Published | Error |
|-------|------------|------------|---------|
| ORCBI | 19.092910 | 19.092721 | 0.0010% |
| ORC42 | 16.808301 | 16.808106 | 0.0012% |
About eighty times tighter than the interior interpolation bound, and two orders
of magnitude under the 0.1% reporting threshold. That is unsurprising -- one day of
tracking error on a basket with exactly known weights has very little room to go
wrong -- but it is measured rather than assumed, and `assemble` re-measures it
every time a published value lands on a date that was projected.
**Implied drag** is reported for information, not enforced. It should sit near
the plan's published cost table (0.234%/yr for ORCBI, 0.239% for ORC42); a fee
is not identifiable over a short gap, so per-gap figures are noisy by nature.
@ -113,6 +130,7 @@ something puts files in its candle cache. That is what `assemble` does.
```
zfin-vestwell record # feed -> data/observed.srf (idempotent)
zfin-vestwell project # funds -> data/observed.srf (weekend fill)
zfin-vestwell assemble --out "$CACHE" # series + log + feed -> cache files
```
@ -182,34 +200,130 @@ against all eight Internet Archive captures of the feed, 8/8 consistent:
value.** The Sunday capture is the informative one: a three-day-old value only
fits if weekends do not publish, so Friday's value does not appear until Monday.
The practical consequence is that anything scheduled before 12:00 UTC reads the
*previous* day's publish and is therefore two business days behind. 3:30am ET --
07:30 UTC in EDT, 08:30 in EST -- is on the wrong side of that line.
Confirmed directly from three consecutive job runs at 13:00 UTC:
| Run (UTC) | dow | navDate served | outcome |
|---------------------|-----|----------------|----------------------------|
| 2026-08-29 13:00:00 | Sat | 2026-08-27 | already recorded, no-op |
| 2026-08-30 13:00:00 | Sun | 2026-08-27 | already recorded, no-op |
| 2026-08-31 13:00:00 | Mon | 2026-08-28 | recorded |
So the weekend blackout is real and is not a scheduling artefact. **Friday's unit
value does not exist in this feed until Monday around 12:00 UTC**, and no cron
time can change that:
```
Mon 12:00 -> Fri Thu 12:00 -> Wed
Tue 12:00 -> Mon Fri 12:00 -> Thu
Wed 12:00 -> Tue Sat (none) -> Thu
Sun (none) -> Thu
```
From Friday's publish until Monday's, the newest value the feed can offer is
Thursday's. `project` is what fills that; see below. The plan's own logged-in UI
does have Friday's value on Saturday, so the recordkeeper is ahead of this
marketing export -- but that path needs an authenticated session and this one does
not.
Two lesser hazards, neither of which has bitten yet:
- The object ships **no `cache-control` and no `expires`**, so CloudFront falls
back to a distribution default TTL. A POP serving a pre-publish copy at 13:00
would look exactly like the weekend blackout.
- That copy **cannot be forced**. A query-string cache-buster, `Cache-Control:
no-cache` and `Pragma: no-cache` all return a byte-identical cached response
with the `date` header frozen and `age` still climbing. The only available
response is to detect and retry later.
The practical consequence for scheduling is that anything before 12:00 UTC reads
the *previous* day's publish and is therefore two business days behind. 3:30am ET
-- 07:30 UTC in EDT, 08:30 in EST -- is on the wrong side of that line.
### Steady state
Schedule at **13:00 UTC**, expressed in UTC rather than a local zone: the feed's
schedule is a UTC cron, so DST never enters into it. That leaves about an hour of
margin after the publish.
margin after the publish -- measured at 59m23s against an observed `last-modified`
of 12:00:39.
```sh
zfin-vestwell record --data-dir /data && \
zfin-vestwell project --data-dir /data && \
zfin-vestwell assemble --data-dir /data --out /cache --offline
```
`record` does the one feed fetch and persists it; `assemble --offline` then merges
purely from disk. Two steps rather than one so a failure is attributable, and
`assemble` stays independently useful -- after a `reconstruct` on a workstation
syncs new `data/*.srf` across, `assemble` alone refreshes the cache.
`record` does the one feed fetch and persists it; `project` fills any session the
underlying funds already have but the feed has not published; `assemble --offline`
then merges purely from disk. Three steps rather than one so a failure is
attributable, and `assemble` stays independently useful -- after a `reconstruct` on
a workstation syncs new `data/*.srf` across, `assemble` alone refreshes the cache.
Running all seven days is slightly better than weekdays only. `record` is
idempotent per navDate so weekend runs add nothing, but `assemble` re-stamps
`#!expires` and keeps clients from falling through to the provider path.
Running all seven days is what makes `project` worth having: Saturday is the run
that fills Friday. `record` is idempotent per navDate and `project` per
`(date, value)`, so the extra runs append nothing they should not, and `assemble`
re-stamps `#!expires` and keeps clients from falling through to the provider path.
Observed values always beat reconstructed ones for the same date, and `assemble`
reports any date where they disagreed by more than 0.1% -- a sustained count there
means the model has drifted.
### Filling the weekend: `project`
The feed publishes on weekdays; the underlying Vanguard funds do not have that
gap. zfin's NAV refresh runs at 3:30am ET, so by Saturday 13:00 UTC the funds'
Friday closes are in its cache while the feed still reads Thursday. `project`
walks that difference: it takes the newest **feed** value and compounds the
weighted total return forward, exactly as the reconstruction does between anchors
but with nothing at the far end to pin against.
The trigger is a comparison, not a date calculation:
| Sat 13:00 UTC | funds newest | feed newest | outcome |
|---------------|--------------|-------------|--------------------------------|
| Tue-Fri | D-1 | D-1 | no gap, no-op |
| **Sat** | **Fri** | **Thu** | **project Fri** |
| Sun | Fri | Thu | same value recomputed, no-op |
| Mon | Fri | Fri | no gap; the publish supersedes |
Nothing in the code knows what a weekend is. A market holiday or an unscheduled
closure needs no handling either: a day that did not trade has no bar, so it never
enters the calendar. And if zfin's refresh fails, the funds do not advance, no gap
is seen, and behaviour degrades to what it was before this existed rather than to
a wrong number.
Projections live in `data/observed.srf` alongside the feed values, marked
`source::projected`. An absent `source` means `feed`, so every row written before
this existed still means what it did. Three things depend on the distinction, and
each would be silently wrong without it: `record` must not treat a projected date
as already recorded, or Monday's real value would be skipped; a projection must
not be walked forward from another projection; and `verify`'s live-feed check must
not read a projection as the feed being stale.
Precedence in `assemble` is `observed` > `reconstructed` > `projected` -- a value
pinned at both ends beats one pinned at a single end. So Monday's publish replaces
Saturday's estimate automatically, and `assemble` reports the realised error when
it does. That is the only measurement of extrapolation accuracy there is; the
cross-validation figures above bound *interpolation*.
Measured against a published anchor, one session forward is accurate to about
**0.001%** for both portfolios -- roughly eighty times tighter than the interior
interpolation bound, and well under the 0.1% reporting threshold.
Two deliberate limits:
- **A horizon past five sessions is refused**, and `project` exits non-zero. One
session is normal, two means the feed missed a weekday publish, and a week means
the feed is broken rather than closed. Failing stops the `&&` chain, so the last
good cache keeps being served instead of an ever-longer unpinned walk, and the
job goes red.
- **`project` alone depends on zfin's NAV refresh** having run. `record` and
`assemble` still do not, so the independence claimed under *What runs where* is
narrowed to this one command rather than given up.
Every other failure -- a fund missing from the cache, a navDate the funds have no
bar for, a walk that cannot be computed -- warns and exits 0. A projection is an
enrichment, and it must never stop the cache being written.
### Running it on the NAS
The container is `FROM scratch` with a static musl binary, ~10MB. Built and pushed
@ -217,8 +331,14 @@ by `.forgejo/workflows/zig-build.yaml` to
`git.lerch.org/lobo/zfin-vestwell:{<shortsha>,latest}`.
Two mounts. `/data` is the syncthing-mirrored repo, which is where the SRF inputs
live and where `record` appends. `/cache` is zfin-server's cache directory, which
is what `assemble` populates.
live and where `record` and `project` append. `/cache` is zfin-server's cache
directory: `assemble` populates it, and `project` reads the underlying funds out of
it.
`ZFIN_CACHE_DIR` is required rather than optional here. `project` resolves the
cache the way zfin does -- `$ZFIN_CACHE_DIR`, else `$XDG_CACHE_HOME/zfin`, else
`$HOME/.cache/zfin` -- and a `scratch` image has no `/etc/passwd`, so `$HOME` is
unset and the fallback chain has nothing to land on.
```sh
IMAGE=git.lerch.org/lobo/zfin-vestwell:<shortsha>
@ -227,10 +347,23 @@ CACHE=/data/zfin/cache
docker run --rm -v "$DATA:/data" -v "$CACHE:/cache" "$IMAGE" \
record --data-dir /data && \
docker run --rm -e ZFIN_CACHE_DIR=/cache -v "$DATA:/data" -v "$CACHE:/cache" "$IMAGE" \
project --data-dir /data && \
docker run --rm -v "$DATA:/data" -v "$CACHE:/cache" "$IMAGE" \
assemble --data-dir /data --out /cache --offline
```
Only the `project` invocation carries `-e ZFIN_CACHE_DIR=/cache`, because it is the
only one of the three that reads the candle cache. Passing it to all three would be
harmless, but naming it once makes it obvious which command would break without it.
`project` reads the funds from the same `/cache` it hands `assemble` as `--out`;
they are different symbols under one root, so no extra mount is needed. It reads
that cache directly rather than asking ZFIN_SERVER over HTTP: this job already runs
on the host that owns the cache, so a network call would fetch data off the disk it
has mounted while adding a live-service dependency and an API key to a job that
otherwise needs neither.
The Cronicle event pins `<shortsha>` rather than tracking `:latest`, because the
workflow moves `:latest` on a push to any branch and this writes into a live
cache. The deploy job prints the tag to copy.
@ -241,7 +374,14 @@ cache. The deploy job prints the tag to copy.
Writing into the cache while zfin-server serves from it is safe: zfin's `writeRaw`
is atomic (tmp + fsync + rename), so a concurrent reader sees either the old file
or the new one, never a torn one.
or the new one, never a torn one. That same guarantee is what makes `project`
reading the funds out from under a concurrent NAV refresh safe.
ORC42 needs **VIPIX** and **VTIFX** tracked on the NAS in addition to VSMPX, VTPSX
and VBMPX; they are 2% of its allocation each. ORCBI has been a single static era
of `VSMPX 36 / VTPSX 24 / VBMPX 40` since 2018-09-01, so it needs nothing beyond
what was already there. Without the two extra funds `project` declines ORC42 and
says so, rather than projecting from a partial basket.
### What runs where, and why
@ -250,23 +390,29 @@ or the new one, never a torn one.
| `reconstruct` | yes, five Vanguard funds | workstation |
| `verify` | yes, cross-validation needs them | workstation |
| `record` | no | NAS |
| `project` | yes, the in-force era's funds | NAS |
| `assemble` | no | NAS |
`reconstruct` and `verify` call the zfin-cache reader; `record` and `assemble` do
not. That is not incidental -- it is why the NAS job needs no zfin cache of its
own, which matters because the NAS tracks VSMPX, VTPSX and VBMPX but not VIPIX or
VTIFX.
`reconstruct`, `verify` and `project` call the zfin-cache reader; `record` and
`assemble` do not. `record` and `assemble` between them are still enough to
maintain a correct cache, which is why a failure in `project` is not allowed to
stop them.
There is therefore **no ordering dependency on the zfin NAV-refresh cron.** The
only timing constraint is the Vestwell publish above.
`project` is therefore **the one command with an ordering dependency on the zfin
NAV-refresh cron**, and it needs VIPIX and VTIFX tracked as well as VSMPX, VTPSX
and VBMPX. The dependency is comfortable rather than tight: the refresh runs at
3:30am ET and the job at 13:00 UTC, four and a half hours later. If it has not run,
the funds have not advanced, `project` sees no gap and does nothing.
For everything else the only timing constraint is the Vestwell publish above.
### Single writer
**Only the NAS runs `record`.** `data/observed.srf` lives in the syncthing-mirrored
tree so that it stays version-controlled and visible to the workstation, which
means two writers would produce `.sync-conflict` files. Syncthing carries the
appended log back for committing; avoid `git checkout`, `stash` or `reset` on that
file while a run could be in flight.
**Only the NAS runs `record` and `project`.** `data/observed.srf` lives in the
syncthing-mirrored tree so that it stays version-controlled and visible to the
workstation, which means two writers would produce `.sync-conflict` files.
Syncthing carries the appended log back for committing; avoid `git checkout`,
`stash` or `reset` on that file while a run could be in flight.
`.stignore` keeps the build artifacts out of the mirror. Without it syncthing would
carry ~2GB (`.zig-cache` alone is 1.9GB) against 208K of actual data.
@ -277,6 +423,12 @@ carry ~2GB (`.zig-cache` alone is 1.9GB) against 208K of actual data.
- **The series is derived, not observed**, except at the nine anchors. Every row
carries `source::anchor` or `source::reconstructed` so no consumer can confuse
the two.
- **Between Friday's publish and Monday's, the newest bar is an estimate.** It
carries `source::projected` in `data/observed.srf`, `assemble` says so in its
output, and the published value replaces it when it arrives. The zfin cache
itself has no source column, so a consumer reading only the cache cannot tell a
projected bar from a published one -- check the log or `assemble`'s report if
that distinction matters to you.
- **It cannot be regenerated from this repo alone.** `reconstruct` needs about ten
years of dividend-adjusted daily prices for five institutional-class Vanguard
funds, read from zfin's cache -- which lives outside this repo, is not

237
src/assemble.zig
View file

@ -1,10 +1,17 @@
//! Merge the reconstructed series with observed feed values into the bar series
//! Merge the reconstructed series with logged unit values into the bar series
//! zfin's cache wants.
//!
//! Two sources, and a clear precedence: an observed value always beats a
//! reconstructed one for the same date. Reconstruction exists only to cover
//! Three sources, and a clear precedence: a value observed from the feed always
//! beats a derived one for the same date. Reconstruction exists only to cover
//! dates nobody recorded, so as the observation log grows the derived portion
//! shrinks and eventually stops mattering for anything recent.
//!
//! The third source is a projection: a session the feed has not published yet,
//! walked forward from the newest observation. It ranks last because it is the
//! weakest evidence available -- a reconstruction is pinned at both ends and a
//! projection is pinned at one -- and it is always superseded by the feed's own
//! value once that arrives. That supersession is also the only way the projection
//! ever gets marked: see `Counts.projection_corrected`.
const std = @import("std");
const cache_files = @import("cache_files.zig");
@ -18,12 +25,34 @@ pub const Counts = struct {
observed: usize = 0,
/// Dates that exist only in the reconstruction.
reconstructed: usize = 0,
/// Dates present in both, where the observed value differed enough to be
/// worth reporting. A large number here means the reconstruction has drifted
/// or the model is wrong.
/// Dates carried only by a projection, because the feed has not published
/// them and the reconstruction does not reach them.
projected: usize = 0,
/// Dates present in both the log and the reconstruction, where the observed
/// value differed enough to be worth reporting. A large number here means the
/// reconstruction has drifted or the model is wrong.
corrected: usize = 0,
/// Largest relative disagreement seen on an overlapping date.
worst_correction: f64 = 0,
/// Dates where an observation displaced a PROJECTION and disagreed materially.
///
/// Kept apart from `corrected` on purpose. Reconstruction drift is an
/// interpolation error and means something is wrong; projection error is a
/// one-session extrapolation with nothing downstream to pin it, and some is
/// expected. Folding them together would make the drift alarm fire every
/// weekend and so stop meaning anything.
projection_corrected: usize = 0,
/// Largest relative disagreement seen between a projection and the feed value
/// that replaced it. This is the realised accuracy of the projection, which is
/// otherwise unmeasured -- the cross-validation figures bound interpolation,
/// not extrapolation.
worst_projection_error: f64 = 0,
/// Whether the NEWEST bar is a projection rather than a published value.
///
/// Worth surfacing on its own: it is the difference between a cache whose tip
/// is a fact and one whose tip is an estimate, and every other count here is
/// an aggregate over history that cannot answer that.
newest_projected: bool = false,
};
pub const Result = struct {
@ -39,8 +68,8 @@ const correction_threshold = 0.001;
/// Where a candidate value came from. The numeric order IS the precedence: an
/// observation is a published fact and always beats a derived value for the same
/// date.
const Rank = enum(u8) { observed = 0, derived = 1 };
/// date, and a two-end-pinned reconstruction beats a one-end projection.
const Rank = enum(u8) { observed = 0, derived = 1, projected = 2 };
/// One candidate value for one date, before precedence is resolved.
const Entry = struct {
@ -95,7 +124,10 @@ pub fn merge(
try entries.append(arena, .{
.date = o.date,
.unit_value = o.unit_value,
.rank = .observed,
.rank = switch (o.source) {
.feed => .observed,
.projected => .projected,
},
.seq = entries.items.len,
});
}
@ -124,24 +156,53 @@ pub fn merge(
const winner = group[0];
// An observation that displaced a derived value for the same date is
// worth reporting when they disagree materially: a rising count there
// means the model has drifted.
// worth reporting when they disagree materially. The two kinds of derived
// value are counted separately: a rising `corrected` means the model has
// drifted, whereas `projection_corrected` is just the realised error of a
// one-session extrapolation and is expected to be non-zero.
//
// At most one of each kind is compared. The group is sorted by rank then
// by descending sequence, so the first entry of a kind is the freshest of
// it -- which matters because a restated fund price can leave two
// projected rows for the same date, and counting both would inflate the
// error tally with a value that never reached the cache.
if (winner.rank == .observed) {
var did_derived = false;
var did_projected = false;
for (group[1..]) |e| {
if (e.rank != .derived) continue;
const rel = @abs(winner.unit_value - e.unit_value) / winner.unit_value;
if (rel > correction_threshold) {
counts.corrected += 1;
counts.worst_correction = @max(counts.worst_correction, rel);
switch (e.rank) {
.observed => continue,
.derived => {
if (did_derived) continue;
did_derived = true;
if (rel > correction_threshold) {
counts.corrected += 1;
counts.worst_correction = @max(counts.worst_correction, rel);
}
},
.projected => {
if (did_projected) continue;
did_projected = true;
if (rel > correction_threshold) counts.projection_corrected += 1;
// Recorded unconditionally, unlike the drift counter: this
// is the only measurement of projection accuracy that
// exists, so discarding the small values would leave
// nothing to calibrate against.
counts.worst_projection_error = @max(counts.worst_projection_error, rel);
},
}
break;
}
counts.observed += 1;
} else if (winner.rank == .projected) {
counts.projected += 1;
} else {
counts.reconstructed += 1;
}
try bars.append(arena, .{ .date = winner.date, .unit_value = winner.unit_value });
// Bars come out ascending, so the last group decided is the newest.
counts.newest_projected = winner.rank == .projected;
}
counts.total = bars.items.len;
@ -158,6 +219,150 @@ fn ob(symbol: []const u8, date: []const u8, v: f64) data.Observation {
return .{ .symbol = symbol, .date = date, .unit_value = v, .recorded_at = "2026-08-28" };
}
fn proj(symbol: []const u8, date: []const u8, v: f64) data.Observation {
return .{ .symbol = symbol, .date = date, .unit_value = v, .recorded_at = "2026-08-29", .source = .projected };
}
test "a projection carries a date nothing else reaches" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// The weekend state: the series stops at the last anchor, the feed published
// Thursday, and Friday exists only because it was projected.
const r = try merge(
arena,
"ORCBI",
&.{pt("2026-08-27", 19.092721)},
&.{ ob("ORCBI", "2026-08-27", 19.092721), proj("ORCBI", "2026-08-28", 19.02) },
null,
);
try testing.expectEqual(@as(usize, 2), r.bars.len);
try testing.expectEqualStrings("2026-08-28", r.bars[1].date);
try testing.expectApproxEqAbs(@as(f64, 19.02), r.bars[1].unit_value, 1e-9);
try testing.expectEqual(@as(usize, 1), r.counts.projected);
try testing.expectEqual(@as(usize, 1), r.counts.observed);
}
test "an observation beats a projection for the same date" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Monday: the feed finally publishes Friday, and it must win over Saturday's
// estimate no matter that the estimate was appended first.
const r = try merge(
arena,
"ORCBI",
&.{},
&.{ proj("ORCBI", "2026-08-28", 19.02), ob("ORCBI", "2026-08-28", 19.021989) },
null,
);
try testing.expectEqual(@as(usize, 1), r.bars.len);
try testing.expectApproxEqAbs(@as(f64, 19.021989), r.bars[0].unit_value, 1e-12);
try testing.expectEqual(@as(usize, 1), r.counts.observed);
try testing.expectEqual(@as(usize, 0), r.counts.projected);
}
test "precedence holds even when the projection is appended last" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Rank must dominate sequence. Ordering alone would hand this to the
// projection, which is exactly the bug that expressing precedence as a sort
// key rather than as overwrite order is meant to prevent.
const r = try merge(
arena,
"ORCBI",
&.{},
&.{ ob("ORCBI", "2026-08-28", 19.021989), proj("ORCBI", "2026-08-28", 19.02) },
null,
);
try testing.expectEqual(@as(usize, 1), r.bars.len);
try testing.expectApproxEqAbs(@as(f64, 19.021989), r.bars[0].unit_value, 1e-12);
}
test "projection error is counted apart from model drift" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// A projection that missed by 1% must not read as the model having drifted:
// that alarm watches interpolation, and firing it every weekend would retire
// it. 19.02 -> 19.21 is about 1%.
const r = try merge(
arena,
"ORCBI",
&.{},
&.{ proj("ORCBI", "2026-08-28", 19.02), ob("ORCBI", "2026-08-28", 19.21) },
null,
);
try testing.expectEqual(@as(usize, 0), r.counts.corrected);
try testing.expectApproxEqAbs(@as(f64, 0), r.counts.worst_correction, 1e-12);
try testing.expectEqual(@as(usize, 1), r.counts.projection_corrected);
try testing.expect(r.counts.worst_projection_error > 0.009);
}
test "projection error is measured even when it is small" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// The realised extrapolation error is the only calibration figure that
// exists, so a miss below the reporting threshold still has to be recorded --
// otherwise a well-behaved projection leaves no evidence that it worked.
const r = try merge(
arena,
"X",
&.{},
&.{ proj("X", "2026-08-28", 19.020), ob("X", "2026-08-28", 19.021) },
null,
);
try testing.expectEqual(@as(usize, 0), r.counts.projection_corrected);
try testing.expect(r.counts.worst_projection_error > 0);
try testing.expect(r.counts.worst_projection_error < correction_threshold);
}
test "a revised projection is counted once, against the value that survived" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// A restated fund price appends a second projection for a date already
// projected. Only the freshest reached the cache, so only it may contribute to
// the error tally; counting the superseded one too would invent a miss.
const r = try merge(
arena,
"X",
&.{},
&.{
proj("X", "2026-08-28", 17.0),
proj("X", "2026-08-28", 19.02),
ob("X", "2026-08-28", 19.021),
},
null,
);
try testing.expectEqual(@as(usize, 1), r.bars.len);
try testing.expectEqual(@as(usize, 0), r.counts.projection_corrected);
try testing.expect(r.counts.worst_projection_error < correction_threshold);
}
test "a reconstruction outranks a projection for the same date" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// They should not overlap in practice, since a projection starts after the
// newest observation and the reconstruction stops at the last anchor. If they
// ever do, the value pinned at both ends is the better one.
const r = try merge(arena, "X", &.{pt("2026-08-28", 19.05)}, &.{proj("X", "2026-08-28", 19.02)}, null);
try testing.expectEqual(@as(usize, 1), r.bars.len);
try testing.expectApproxEqAbs(@as(f64, 19.05), r.bars[0].unit_value, 1e-12);
try testing.expectEqual(@as(usize, 1), r.counts.reconstructed);
try testing.expectEqual(@as(usize, 0), r.counts.projected);
}
test "merge prefers an observed value over a reconstructed one" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();

308
src/data.zig
View file

@ -290,21 +290,54 @@ const ObservationRow = struct {
date: []const u8,
unit_value: f64,
recorded_at: []const u8 = "",
/// Absent means `feed`; see `ObservationSource`.
///
/// A string rather than an `ObservationSource` directly because SRF coerces
/// enum fields through an unchecked `stringToEnum(...).?` no matter what
/// `CoercionOptions` say -- `srf_opts` documents that hole -- so an unknown
/// value would panic rather than error. Mapped explicitly in `parseObserved`,
/// the same way `series.read` handles its own `source` column.
source: ?[]const u8 = null,
};
/// One unit value read straight from the plan's feed.
/// Where a logged value came from.
///
/// These are observations, not derivations, and they accumulate forever. Once
/// this log is dense the reconstruction stops mattering for recent dates: `assemble`
/// overlays observations on top of the reconstructed series, so an observed value
/// always wins.
/// Absent on disk means `feed`. Every row written before projection existed
/// carries no `source::` field, and an append-only log is not something to
/// rewrite for cosmetics. `parseObserved` normalises the absence away, so no
/// caller downstream has to know that "feed" has two spellings.
pub const ObservationSource = enum {
/// Read from the plan's published feed. A fact.
feed,
/// Walked forward from the newest feed value using the underlying funds'
/// returns, because the feed had not published this session yet. An estimate,
/// and always superseded by the feed's own value once that arrives.
projected,
};
/// One unit value for one date, with where it came from.
///
/// Mostly these are observations read straight from the plan's feed, and they
/// accumulate forever. Once this log is dense the reconstruction stops mattering
/// for recent dates: `assemble` overlays the log on top of the reconstructed
/// series, so a logged value wins.
///
/// A `projected` row is the exception, and the reason `source` exists: it is
/// derived, not observed, and several things must treat it differently. See
/// `hasFeedObservation` and `newestFeedObservation` -- using the wrong one of
/// those is silently wrong rather than loud.
pub const Observation = struct {
symbol: []const u8,
/// The feed's own navDate, which is the prior business day.
/// For a feed row, the feed's own navDate, which is the prior business day.
/// For a projected row, the session the value was walked forward to.
date: []const u8,
unit_value: f64,
/// When the fetch happened. Provenance only; nothing keys off it.
/// When the fetch or the projection happened. Provenance only; nothing keys
/// off it.
recorded_at: []const u8,
/// Defaults to `feed`, so every existing row and every `record` write means
/// exactly what it always did.
source: ObservationSource = .feed,
};
pub fn parseObserved(arena: std.mem.Allocator, data: []const u8) ![]const Observation {
@ -318,32 +351,97 @@ pub fn parseObserved(arena: std.mem.Allocator, data: []const u8) ![]const Observ
const row = try fields.to(ObservationRow, srf_opts.user_edited);
if (!civil.isValidIso(row.date)) return error.InvalidObservedDate;
if (!(row.unit_value > 0)) return error.InvalidObservedValue;
const source: ObservationSource = if (row.source) |s|
std.meta.stringToEnum(ObservationSource, s) orelse return error.UnknownObservationSource
else
.feed;
try out.append(arena, .{
.symbol = try arena.dupe(u8, row.symbol),
.date = try arena.dupe(u8, row.date),
.unit_value = row.unit_value,
.recorded_at = try arena.dupe(u8, row.recorded_at),
.source = source,
});
}
return out.toOwnedSlice(arena);
}
/// Whether this log already holds a value for `symbol` on `date`.
/// Whether the log already holds a FEED value for `symbol` on `date`.
///
/// Makes `record` idempotent: the feed republishes the same navDate all weekend
/// and across holidays, so a daily job re-reads the same value repeatedly and
/// must not append it repeatedly.
pub fn hasObservation(obs: []const Observation, symbol: []const u8, date: []const u8) bool {
///
/// Projected rows are deliberately invisible here. A projection exists precisely
/// for a date the feed has not published yet, so counting one would make `record`
/// skip the real value when it finally arrives -- silently, since "already
/// recorded" is a success path. That is the whole reason `source` exists.
pub fn hasFeedObservation(obs: []const Observation, symbol: []const u8, date: []const u8) bool {
for (obs) |o| {
if (o.source != .feed) continue;
if (std.mem.eql(u8, o.symbol, symbol) and std.mem.eql(u8, o.date, date)) return true;
}
return false;
}
/// The newest observation for `symbol`, or null when there is none.
pub fn newestObservation(obs: []const Observation, symbol: []const u8) ?Observation {
/// Whether the log already holds this exact projected value for `symbol` on
/// `date`.
///
/// Makes `project` idempotent the way `hasFeedObservation` does for `record`, so a
/// weekend's second and third runs add nothing without anything having to know
/// what a weekend is.
///
/// Compared at the log's own precision rather than on raw `f64` bits. The file
/// stores six decimals, so a value read back is the rounded one while a value
/// just recomputed carries all of its bits; exact equality between the two is
/// never true, and using it appended a duplicate row on every run. The question
/// this needs to answer is "would writing this produce the row already there",
/// and that is a question about the rendered value.
///
/// Keyed on the value as well as the date because a restated fund price changes
/// the projection for a date already projected, and that revision is worth
/// appending; an unchanged re-run is not.
pub fn hasProjection(obs: []const Observation, symbol: []const u8, date: []const u8, unit_value: f64) bool {
for (obs) |o| {
if (o.source != .projected) continue;
if (!std.mem.eql(u8, o.symbol, symbol) or !std.mem.eql(u8, o.date, date)) continue;
if (sameAtLogPrecision(o.unit_value, unit_value)) return true;
}
return false;
}
/// Decimals the observation log keeps for a unit value.
///
/// The feed publishes six and the plan's own config declares four, so six stores
/// the feed's values verbatim.
const log_decimals = 6;
/// `10 ^ log_decimals`, derived so the two cannot drift apart.
const log_scale: f64 = blk: {
var s: f64 = 1;
for (0..log_decimals) |_| s *= 10;
break :blk s;
};
/// Whether two unit values would be written to the log as the same number.
fn sameAtLogPrecision(a: f64, b: f64) bool {
return @round(a * log_scale) == @round(b * log_scale);
}
/// The newest FEED observation for `symbol`, or null when there is none.
///
/// Feed-only, and both callers need it that way:
///
/// - `project` walks forward from here. Anchoring on a projected row instead
/// would extrapolate from an extrapolation, compounding its own error every
/// run rather than recomputing from the last known fact.
/// - `verify`'s live-feed check asks "did the feed publish something we failed
/// to capture". A projection is ahead of the feed by design, so counting one
/// would report staleness every weekend.
pub fn newestFeedObservation(obs: []const Observation, symbol: []const u8) ?Observation {
var best: ?Observation = null;
for (obs) |o| {
if (o.source != .feed) continue;
if (!std.mem.eql(u8, o.symbol, symbol)) continue;
if (best == null or civil.lessThan(best.?.date, o.date)) best = o;
}
@ -355,10 +453,27 @@ pub fn newestObservation(obs: []const Observation, symbol: []const u8) ?Observat
const ObservationOut = struct {
symbol: []const u8,
date: []const u8,
unit_value: srf_num.Fixed(6),
unit_value: srf_num.Fixed(log_decimals),
recorded_at: []const u8,
};
/// Write-side row for a value that is not a plain feed reading.
///
/// A second struct rather than an optional `source` on `ObservationOut`, because
/// SRF renders a null optional as an empty field instead of omitting it: its
/// `formatField` falls through to `formatValue(name, null, ...)` when the
/// optional is absent. One struct with `source: ?[]const u8` would therefore
/// stamp a bare `source::` onto every feed row written from now on, for no
/// reader's benefit. Two structs keep feed rows byte-identical to the rows
/// already in the log, and confine the new column to the rows that need it.
const ObservationOutSourced = struct {
symbol: []const u8,
date: []const u8,
unit_value: srf_num.Fixed(log_decimals),
recorded_at: []const u8,
source: []const u8,
};
/// Render one observation as an SRF record line, newline included.
///
/// Goes through `srf.fmt` rather than a format string. That is not tidiness: SRF
@ -367,17 +482,30 @@ const ObservationOut = struct {
/// line silently produced a corrupt record for any value with a comma in it --
/// the same mistake that broke `anchors.srf` when a comment was written into an
/// `evidence` field.
///
/// `emit_directives = false` on both paths: this line is appended to a file that
/// already has its `#!srfv1` header.
pub fn formatObservation(arena: std.mem.Allocator, o: Observation) ![]const u8 {
const rows = [_]ObservationOut{.{
if (o.source == .feed) {
const rows = [_]ObservationOut{.{
.symbol = o.symbol,
.date = o.date,
.unit_value = .init(o.unit_value),
.recorded_at = o.recorded_at,
}};
return std.fmt.allocPrint(arena, "{f}", .{
srf.fmt(ObservationOut, &rows, .{ .emit_directives = false }),
});
}
const rows = [_]ObservationOutSourced{.{
.symbol = o.symbol,
.date = o.date,
.unit_value = .init(o.unit_value),
.recorded_at = o.recorded_at,
.source = @tagName(o.source),
}};
// `emit_directives = false`: this line is appended to a file that already has
// its `#!srfv1` header.
return std.fmt.allocPrint(arena, "{f}", .{
srf.fmt(ObservationOut, &rows, .{ .emit_directives = false }),
srf.fmt(ObservationOutSourced, &rows, .{ .emit_directives = false }),
});
}
@ -634,20 +762,72 @@ test "parseObserved reads the feed log and rejects unusable rows" {
try testing.expectError(error.InvalidObservedValue, parseObserved(arena, "#!srfv1\nsymbol::X,date::2026-01-01,unit_value:num:0\n"));
}
test "hasObservation makes recording idempotent per symbol and date" {
test "hasFeedObservation makes recording idempotent per symbol and date" {
const obs: []const Observation = &.{
.{ .symbol = "ORCBI", .date = "2026-08-27", .unit_value = 19.09, .recorded_at = "" },
.{ .symbol = "ORC42", .date = "2026-08-27", .unit_value = 16.81, .recorded_at = "" },
};
try testing.expect(hasObservation(obs, "ORCBI", "2026-08-27"));
try testing.expect(hasObservation(obs, "ORC42", "2026-08-27"));
try testing.expect(hasFeedObservation(obs, "ORCBI", "2026-08-27"));
try testing.expect(hasFeedObservation(obs, "ORC42", "2026-08-27"));
// Same date, different symbol, and same symbol, different date: both misses.
try testing.expect(!hasObservation(obs, "OTHER", "2026-08-27"));
try testing.expect(!hasObservation(obs, "ORCBI", "2026-08-28"));
try testing.expect(!hasObservation(&.{}, "ORCBI", "2026-08-27"));
try testing.expect(!hasFeedObservation(obs, "OTHER", "2026-08-27"));
try testing.expect(!hasFeedObservation(obs, "ORCBI", "2026-08-28"));
try testing.expect(!hasFeedObservation(&.{}, "ORCBI", "2026-08-27"));
}
test "newestObservation picks the latest date, not the last row" {
test "a projection never suppresses the feed value for the same date" {
// The weekend case, and the reason `source` exists at all. Saturday projects
// Friday; Monday the feed finally publishes Friday. If the projection counted
// as "already recorded" here, `record` would skip the real value and say
// nothing -- "already recorded" is a success path, so the day would simply
// never be captured.
const obs: []const Observation = &.{
.{ .symbol = "ORCBI", .date = "2026-08-27", .unit_value = 19.092721, .recorded_at = "2026-08-28" },
.{ .symbol = "ORCBI", .date = "2026-08-28", .unit_value = 19.02, .recorded_at = "2026-08-29", .source = .projected },
};
try testing.expect(!hasFeedObservation(obs, "ORCBI", "2026-08-28"));
try testing.expect(hasFeedObservation(obs, "ORCBI", "2026-08-27"));
}
test "hasProjection is keyed on the value so a revision still appends" {
const obs: []const Observation = &.{
.{ .symbol = "X", .date = "2026-08-28", .unit_value = 19.02, .recorded_at = "", .source = .projected },
};
// Sunday recomputes Saturday's projection from the same inputs: nothing new.
try testing.expect(hasProjection(obs, "X", "2026-08-28", 19.02));
// A restated fund price moves the projection, which is worth recording.
try testing.expect(!hasProjection(obs, "X", "2026-08-28", 19.03));
try testing.expect(!hasProjection(obs, "X", "2026-08-31", 19.02));
try testing.expect(!hasProjection(obs, "OTHER", "2026-08-28", 19.02));
// A feed row for the same date is not a projection.
const feed_row: []const Observation = &.{
.{ .symbol = "X", .date = "2026-08-28", .unit_value = 19.02, .recorded_at = "" },
};
try testing.expect(!hasProjection(feed_row, "X", "2026-08-28", 19.02));
}
test "hasProjection compares at the precision the log actually stores" {
// The regression that mattered. What comes back from the file is rounded to
// six decimals; what `projectForward` just computed has all of its bits. Exact
// f64 equality between the two is essentially never true, so every run
// appended another row for a date it had already projected.
const stored = try std.fmt.parseFloat(f64, "19.092910");
const obs: []const Observation = &.{
.{ .symbol = "X", .date = "2026-08-27", .unit_value = stored, .recorded_at = "", .source = .projected },
};
// Differs far below the stored precision: the same row, so no append.
try testing.expect(hasProjection(obs, "X", "2026-08-27", 19.0929104271));
try testing.expect(hasProjection(obs, "X", "2026-08-27", 19.09290961));
// A difference the log can actually represent is a different row.
try testing.expect(!hasProjection(obs, "X", "2026-08-27", 19.092911));
try testing.expect(!hasProjection(obs, "X", "2026-08-27", 19.092909));
// And the values are genuinely unequal as f64, which is the whole point.
try testing.expect(stored != @as(f64, 19.0929104271));
}
test "newestFeedObservation picks the latest date, not the last row" {
// An append-only log can be out of order if a backfill was pasted in.
const obs: []const Observation = &.{
.{ .symbol = "X", .date = "2026-08-27", .unit_value = 3.0, .recorded_at = "" },
@ -655,11 +835,52 @@ test "newestObservation picks the latest date, not the last row" {
.{ .symbol = "X", .date = "2026-08-31", .unit_value = 4.0, .recorded_at = "" },
.{ .symbol = "X", .date = "2026-08-20", .unit_value = 1.0, .recorded_at = "" },
};
const n = newestObservation(obs, "X").?;
const n = newestFeedObservation(obs, "X").?;
try testing.expectEqualStrings("2026-08-31", n.date);
try testing.expectApproxEqAbs(@as(f64, 4.0), n.unit_value, 1e-12);
try testing.expectEqual(@as(?Observation, null), newestObservation(obs, "MISSING"));
try testing.expectEqual(@as(?Observation, null), newestObservation(&.{}, "X"));
try testing.expectEqual(@as(?Observation, null), newestFeedObservation(obs, "MISSING"));
try testing.expectEqual(@as(?Observation, null), newestFeedObservation(&.{}, "X"));
}
test "newestFeedObservation ignores projections so one cannot anchor another" {
// Without this, Sunday would walk forward from Saturday's projection instead
// of from Thursday's published value, compounding the estimate's own error
// every run rather than recomputing it from the last known fact.
const obs: []const Observation = &.{
.{ .symbol = "X", .date = "2026-08-27", .unit_value = 19.092721, .recorded_at = "" },
.{ .symbol = "X", .date = "2026-08-28", .unit_value = 19.02, .recorded_at = "", .source = .projected },
};
const n = newestFeedObservation(obs, "X").?;
try testing.expectEqualStrings("2026-08-27", n.date);
try testing.expectEqual(ObservationSource.feed, n.source);
}
test "parseObserved defaults an absent source to feed and rejects an unknown one" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Every row already in the log looks like this: no `source` column at all.
const back = try parseObserved(
arena,
"#!srfv1\nsymbol::X,date::2026-08-27,unit_value:num:19.092721,recorded_at::2026-08-28\n",
);
try testing.expectEqual(@as(usize, 1), back.len);
try testing.expectEqual(ObservationSource.feed, back[0].source);
const proj = try parseObserved(
arena,
"#!srfv1\nsymbol::X,date::2026-08-28,unit_value:num:19.02,recorded_at::2026-08-29,source::projected\n",
);
try testing.expectEqual(ObservationSource.projected, proj[0].source);
// Must error, not panic. SRF reaches an unchecked `stringToEnum(...).?` for
// enum fields regardless of coercion options, which is why the column is
// read as a string and mapped by hand.
try testing.expectError(error.UnknownObservationSource, parseObserved(
arena,
"#!srfv1\nsymbol::X,date::2026-08-28,unit_value:num:19.02,source::invented\n",
));
}
test "formatObservation round-trips through parseObserved" {
@ -684,6 +905,41 @@ test "formatObservation round-trips through parseObserved" {
try testing.expectApproxEqAbs(@as(f64, 16.808106), back[0].unit_value, 1e-9);
}
test "a feed row gains no source column and a projected row carries one" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Feed rows must stay byte-identical in shape to the rows already in the log.
// SRF emits a null optional as an empty field rather than omitting it, so a
// single output struct with `source: ?[]const u8` would have put a bare
// `source::` on every row from here on.
const fed = try formatObservation(arena, .{
.symbol = "ORC42",
.date = "2026-08-27",
.unit_value = 16.808106,
.recorded_at = "2026-08-28",
});
try testing.expect(std.mem.indexOf(u8, fed, "source") == null);
const projected = try formatObservation(arena, .{
.symbol = "ORC42",
.date = "2026-08-28",
.unit_value = 16.738955,
.recorded_at = "2026-08-29",
.source = .projected,
});
try testing.expect(std.mem.indexOf(u8, projected, "source::projected") != null);
// Both shapes must read back out of the same file.
const body = try std.fmt.allocPrint(arena, "#!srfv1\n{s}{s}", .{ fed, projected });
const back = try parseObserved(arena, body);
try testing.expectEqual(@as(usize, 2), back.len);
try testing.expectEqual(ObservationSource.feed, back[0].source);
try testing.expectEqual(ObservationSource.projected, back[1].source);
try testing.expectApproxEqAbs(@as(f64, 16.738955), back[1].unit_value, 1e-9);
}
test "formatObservation survives a comma in a string value" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();

204
src/main.zig
View file

@ -7,6 +7,9 @@
//!
//! reconstruct rebuild data/<SYMBOL>.srf from data/model.srf + data/anchors.srf
//! verify check the series on disk against every available evidence source
//! record append the feed's newly published value to data/observed.srf
//! project fill sessions the funds have but the feed has not published
//! assemble merge everything into the cache files zfin serves
const std = @import("std");
@ -31,6 +34,7 @@ const usage =
\\ zfin-vestwell reconstruct [--data-dir DIR]
\\ zfin-vestwell verify [--data-dir DIR] [--offline]
\\ zfin-vestwell record [--data-dir DIR]
\\ zfin-vestwell project [--data-dir DIR]
\\ zfin-vestwell assemble --out DIR [--data-dir DIR] [--offline]
\\
\\Commands:
@ -45,6 +49,11 @@ const usage =
\\ record Fetch the feed and append anything new to
\\ data/observed.srf. Idempotent per navDate, so running it
\\ daily is safe.
\\ project Fill sessions the underlying funds already have but the feed
\\ has not published, by walking forward from the newest
\\ recorded value. The feed skips weekends, so Friday's value
\\ would otherwise be missing until Monday. Needs zfin's candle
\\ cache. A no-op whenever the feed is level with the funds.
\\ assemble Merge data/<SYMBOL>.srf with data/observed.srf and the live
\\ feed, then write candles_daily.srf + candles_meta.srf into
\\ <out>/<SYMBOL>/ for zfin to serve. Needs no zfin cache.
@ -58,7 +67,7 @@ const usage =
\\
;
const Command = enum { reconstruct, verify, record, assemble, help };
const Command = enum { reconstruct, verify, record, project, assemble, help };
const Options = struct {
cmd: Command,
@ -77,6 +86,8 @@ fn parseArgs(args: []const []const u8) !Options {
.verify
else if (std.mem.eql(u8, args[1], "record"))
.record
else if (std.mem.eql(u8, args[1], "project"))
.project
else if (std.mem.eql(u8, args[1], "assemble"))
.assemble
else if (std.mem.eql(u8, args[1], "-h") or
@ -305,15 +316,18 @@ fn cmdVerify(
const fund_id = feed.fundIdFor(sym) orelse return error.NoFundIdForSymbol;
const q = try feed.quoteFrom(arena, body, fund_id);
// Compare the feed against the newest value ANY source has, not just
// the committed series. Once `record` is running daily the series
// Compare the feed against the newest value any FEED source has, not
// just the committed series. Once `record` is running daily the series
// deliberately stops at the last anchor and `observed.srf` carries
// everything since, so checking the series alone would report a
// false staleness every day.
//
// Feed-only on purpose: a projected row is ahead of the feed by
// construction, so counting one would report staleness every weekend.
var newest_date = points[points.len - 1].date;
var newest_value = points[points.len - 1].unit_value;
var newest_from: []const u8 = "series";
if (data.newestObservation(l.observed, sym)) |obs| {
if (data.newestFeedObservation(l.observed, sym)) |obs| {
if (civil.lessThan(newest_date, obs.date)) {
newest_date = obs.date;
newest_value = obs.unit_value;
@ -377,7 +391,7 @@ fn cmdRecord(arena: std.mem.Allocator, w: *std.Io.Writer, io: std.Io, o: Options
for (symbols) |sym| {
const fund_id = feed.fundIdFor(sym) orelse return error.NoFundIdForSymbol;
const q = try feed.quoteFrom(arena, body, fund_id);
if (data.hasObservation(obs, sym, q.date)) {
if (data.hasFeedObservation(obs, sym, q.date)) {
try w.print("{s:<6} {s} already recorded ({d:.6})\n", .{ sym, q.date, q.unit_value });
continue;
}
@ -409,6 +423,151 @@ fn cmdRecord(arena: std.mem.Allocator, w: *std.Io.Writer, io: std.Io, o: Options
return 0;
}
/// Fill sessions the underlying funds already have but the feed has not published.
///
/// ## Why this exists
///
/// The plan's feed publishes on weekdays only, around 12:00 UTC, carrying the
/// previous business day. So Friday's unit value does not appear until Monday, and
/// across a weekend the newest bar anyone can serve is Thursday's. The underlying
/// Vanguard funds have no such gap: Friday's closes are in zfin's cache by
/// Saturday morning. This walks the difference.
///
/// ## Why there is no calendar in here
///
/// The trigger is a comparison, not a date calculation: project when the funds'
/// newest session is newer than the newest recorded value, and otherwise do
/// nothing. On a weekday both sides top out at the same session and this is a
/// no-op. Saturday is the case that fires. Sunday recomputes the identical value
/// and `hasProjection` drops it. Holidays and unscheduled closures need no code at
/// all, because a day the market did not trade has no bar and so never enters the
/// calendar. Nothing here knows what a weekend is.
///
/// ## Why it almost never fails the job
///
/// A projection is an enrichment, and `assemble` must still run without one. So a
/// missing fund, a navDate the funds have no bar for, or a walk that cannot be
/// computed all warn and exit 0. The single exception is a horizon past
/// `recon.max_projected_sessions`: that means the feed is broken rather than
/// closed, and it exits non-zero on purpose, which stops the `&&` chain and
/// leaves the last good cache in place rather than extending an unpinned
/// extrapolation for a second week.
fn cmdProject(
arena: std.mem.Allocator,
w: *std.Io.Writer,
io: std.Io,
env: *const std.process.Environ.Map,
o: Options,
) !u8 {
const l = try load(arena, io, env, o.data_dir);
const path = try std.fs.path.join(arena, &.{ o.data_dir, "observed.srf" });
var today_buf: [10]u8 = undefined;
const now = try civil.today(io, &today_buf);
var appended: std.ArrayList(u8) = .empty;
var added: usize = 0;
var stale = false;
for (symbols) |sym| {
const in = try l.inputsFor(sym);
const cal = try recon.calendar(arena, in.model, in.tickers);
const newest = data.newestFeedObservation(l.observed, sym) orelse {
// Nothing recorded yet means there is nothing to walk forward from.
// `reconstruct` covers history; this only ever extends the log's tip.
try w.print("{s:<6} no recorded value to project from\n", .{sym});
continue;
};
const from = switch (recon.planProjection(cal, newest.date)) {
.no_sessions => {
std.log.warn("{s}: no priceable sessions in zfin's cache; not projecting", .{sym});
continue;
},
.up_to_date => {
try w.print("{s:<6} funds have nothing newer than {s}, nothing to project\n", .{ sym, newest.date });
continue;
},
.date_not_a_session => {
std.log.warn(
"{s}: recorded date {s} is not a session the funds all priced; not projecting",
.{ sym, newest.date },
);
continue;
},
.walk_from => |i| i,
};
const out = recon.projectForward(arena, in.model, in.tickers, cal, from, newest.unit_value) catch |err| switch (err) {
error.ProjectionGapTooLarge => {
std.log.err(
"{s}: funds have reached {s} but the newest recorded value is {s}, " ++
"more than {d} sessions behind. The feed is not merely closed; " ++
"refusing to extrapolate that far.",
.{ sym, cal[cal.len - 1], newest.date, recon.max_projected_sessions },
);
stale = true;
continue;
},
else => {
std.log.warn("{s}: cannot project from {s} ({s})", .{ sym, newest.date, @errorName(err) });
continue;
},
};
if (out.len > 1) {
std.log.warn(
"{s}: projecting {d} sessions from {s}; one is normal, so the feed has " ++
"missed a publish it should have made",
.{ sym, out.len, newest.date },
);
}
for (out) |p| {
if (data.hasProjection(l.observed, sym, p.date, p.unit_value)) {
try w.print("{s:<6} {s} already projected ({d:.6})\n", .{ sym, p.date, p.unit_value });
continue;
}
try appended.appendSlice(arena, try data.formatObservation(arena, .{
.symbol = sym,
.date = p.date,
.unit_value = p.unit_value,
.recorded_at = now,
.source = .projected,
}));
added += 1;
try w.print("{s:<6} {s} projected {d:.6} from {s} {d:.6}\n", .{
sym, p.date, p.unit_value, newest.date, newest.unit_value,
});
}
}
if (added > 0) {
// Read-modify-write rather than an append-mode handle, matching `record`:
// the file is tiny, and rewriting the whole thing keeps a partial write
// from leaving a torn record.
const existing = std.Io.Dir.cwd().readFileAlloc(io, path, arena, .limited(32 * 1024 * 1024)) catch |err| switch (err) {
error.FileNotFound => "#!srfv1\n",
else => return err,
};
var full: std.ArrayList(u8) = .empty;
try full.appendSlice(arena, existing);
if (existing.len > 0 and existing[existing.len - 1] != '\n') try full.append(arena, '\n');
try full.appendSlice(arena, appended.items);
try std.Io.Dir.cwd().writeFile(io, .{ .sub_path = path, .data = full.items });
try w.print("\nappended {d} projection(s) to {s}\n", .{ added, path });
} else {
try w.print("\nnothing new to project\n", .{});
}
if (stale) {
try w.print("\nprojection REFUSED: the feed is too far behind to extrapolate\n", .{});
return 1;
}
return 0;
}
/// Merge reconstruction + observations into zfin cache files.
fn cmdAssemble(
arena: std.mem.Allocator,
@ -445,7 +604,7 @@ fn cmdAssemble(
try cache_files.writeSymbol(arena, io, out_dir, sym, merged.bars, now_s);
const newest = merged.bars[merged.bars.len - 1];
try w.print("{s:<6} {d:>5} bars {s} .. {s} last {d:.6} ({d} observed, {d} reconstructed)\n", .{
try w.print("{s:<6} {d:>5} bars {s} .. {s} last {d:.6} ({d} observed, {d} reconstructed, {d} projected)\n", .{
sym,
merged.counts.total,
merged.bars[0].date,
@ -453,13 +612,29 @@ fn cmdAssemble(
newest.unit_value,
merged.counts.observed,
merged.counts.reconstructed,
merged.counts.projected,
});
if (merged.counts.newest_projected) {
// The tip of the cache is an estimate, not a published value. Normal
// between Friday's publish and Monday's, and worth saying out loud
// rather than leaving to be inferred from the counts.
try w.print(" newest bar {s} is PROJECTED, not yet published by the feed\n", .{newest.date});
}
if (merged.counts.corrected > 0) {
try w.print(" {d} date(s) where an observation corrected the reconstruction, worst {d:.3}%\n", .{
merged.counts.corrected,
merged.counts.worst_correction * 100,
});
}
if (merged.counts.worst_projection_error > 0) {
// The realised one-session extrapolation error. Reported whenever a
// published value has landed on a date that was projected, because
// this is the only measurement of projection accuracy there is.
try w.print(" projection vs published: {d} past 0.1%, worst {d:.4}%\n", .{
merged.counts.projection_corrected,
merged.counts.worst_projection_error * 100,
});
}
}
try w.print("\nwrote candles_daily.srf + candles_meta.srf under {s}\n", .{out_dir});
@ -507,6 +682,7 @@ fn run(arena: std.mem.Allocator, w: *std.Io.Writer, io: std.Io, init: std.proces
.reconstruct => cmdReconstruct(arena, w, io, init.environ_map, o),
.verify => cmdVerify(arena, w, io, init.environ_map, o),
.record => cmdRecord(arena, w, io, o),
.project => cmdProject(arena, w, io, init.environ_map, o),
.assemble => cmdAssemble(arena, w, io, o),
};
}
@ -530,6 +706,22 @@ test "parseArgs recognises both commands and their flags" {
try testing.expectEqualStrings("/tmp/x", o.data_dir);
try testing.expect(o.offline);
}
{
const o = try parseArgs(&.{ "zfin-vestwell", "project" });
try testing.expectEqual(Command.project, o.cmd);
try testing.expectEqualStrings("data", o.data_dir);
}
{
const o = try parseArgs(&.{ "zfin-vestwell", "project", "--data-dir", "/data" });
try testing.expectEqual(Command.project, o.cmd);
try testing.expectEqualStrings("/data", o.data_dir);
}
{
// `project` reads a cache and writes the log; there is nothing for
// `--out` to mean, and unlike `assemble` it must not require one.
const o = try parseArgs(&.{ "zfin-vestwell", "project" });
try testing.expectEqual(@as(?[]const u8, null), o.out);
}
}
test "parseArgs defaults to help and rejects bad input" {

279
src/recon.zig
View file

@ -219,6 +219,71 @@ pub fn build(arena: std.mem.Allocator, in: Inputs) ![]const Point {
return out.toOwnedSlice(arena);
}
/// One session the feed has not published yet, and the value walked forward to it.
///
/// Deliberately not a `Point`: a `Point` carries a `PointSource` and lands in
/// `data/<SYMBOL>.srf`, which `reconstruct` owns and which must contain only
/// anchored and interpolated values. A projection's home is the observation log,
/// so it gets its own type and never risks being written to the wrong file.
pub const Projection = struct {
date: []const u8,
unit_value: f64,
};
/// Most sessions a projection will walk before it refuses.
///
/// The normal gap is one: the feed skips weekends, so on Saturday the underlying
/// funds have Friday and the feed does not. Two means the feed also missed a
/// weekday. Past a week's worth of sessions the feed is not closed, it is broken,
/// and a long unpinned walk is the wrong thing to paper over it with -- an
/// unpinned walk across ORCBI's 6.5-year gap is off by -1.29%, and while five
/// sessions is nowhere near that, the error grows with no downstream anchor to
/// absorb it. Refusing turns a silent slow drift into a job that fails and mails
/// someone.
pub const max_projected_sessions = 5;
/// Walk forward from a known value to the end of the calendar.
///
/// This is `pinGap` with the pinning removed, because there is nothing to pin to:
/// the closing anchor of a projection has not been published yet. That makes it
/// strictly weaker than the reconstruction -- step 2 of the module comment above
/// is what buys the accuracy, and a projection does not get it -- which is why
/// `assemble` ranks a projected value below a reconstructed one, and why the
/// horizon is capped.
///
/// `from` is the calendar index of the day `v0` belongs to; the result covers
/// `cal[from + 1 ..]` and is empty when `from` is already the last session. The
/// caller is expected to have obtained `from` by exact date match, not by
/// snapping: snapping would silently apply the wrong day's return.
pub fn projectForward(
arena: std.mem.Allocator,
model: data.Model,
tickers: []const candles.Ticker,
cal: []const []const u8,
from: usize,
v0: f64,
) ![]const Projection {
if (from >= cal.len) return error.GapOutOfRange;
if (!(v0 > 0)) return error.NonPositiveStart;
const n = cal.len - 1 - from;
if (n == 0) return &.{};
if (n > max_projected_sessions) return error.ProjectionGapTooLarge;
const out = try arena.alloc(Projection, n);
var v = v0;
for (0..n) |k| {
const r = try dayReturn(model, tickers, cal[from + k], cal[from + k + 1]);
v *= 1.0 + r;
// A projection that has gone non-positive or non-finite is not a value to
// publish. `!(v > 0)` also rejects NaN; the finite check catches infinity,
// which is positive and would otherwise pass.
if (!(v > 0) or !std.math.isFinite(v)) return error.WalkCollapsed;
out[k] = .{ .date = cal[from + k + 1], .unit_value = v };
}
return out;
}
/// Value at an exact date, or null when the series has no entry for it.
pub fn valueAt(points: []const Point, date: []const u8) ?f64 {
for (points) |p| {
@ -227,6 +292,51 @@ pub fn valueAt(points: []const Point, date: []const u8) ?f64 {
return null;
}
/// Calendar index of an exact date, or null when the date is not a session.
///
/// Exact rather than snapping, and that matters: `projectForward` walks using
/// `dayReturn(cal[i], cal[i+1])`, so an index that is merely near the value's own
/// date would compound the wrong day's return into the result.
fn indexOfDate(cal: []const []const u8, date: []const u8) ?usize {
for (cal, 0..) |d, i| {
if (std.mem.eql(u8, d, date)) return i;
}
return null;
}
/// What, if anything, a projection should do.
pub const ProjectionPlan = union(enum) {
/// zfin's cache has no day this portfolio can be priced on at all.
no_sessions,
/// Nothing the funds have is newer than the value already recorded. The
/// ordinary weekday answer, and also the answer on a market holiday and on a
/// Sunday, without any of those being named anywhere.
up_to_date,
/// The recorded value's own date is not a session every fund in its era
/// priced, so there is no index to walk from that would compound the right
/// returns. Decline rather than guess which side of it to start on.
date_not_a_session,
/// Walk forward from this calendar index.
walk_from: usize,
};
/// Decide whether there is a gap to fill, given the funds' calendar and the date
/// of the newest recorded value.
///
/// This is the whole trigger, and it is a comparison rather than a date
/// calculation. There is no weekday arithmetic and no holiday table: a day the
/// market did not trade has no bar, so it never reaches the calendar, and a feed
/// that is level with the funds produces `up_to_date` whatever the reason.
pub fn planProjection(cal: []const []const u8, from_date: []const u8) ProjectionPlan {
if (cal.len == 0) return .no_sessions;
// Also covers the feed running AHEAD of the funds, which happens whenever
// zfin's refresh is behind. `up_to_date` is the right answer there too, and
// reporting it as a missing session would be actively misleading.
if (!civil.lessThan(from_date, cal[cal.len - 1])) return .up_to_date;
if (indexOfDate(cal, from_date)) |i| return .{ .walk_from = i };
return .date_not_a_session;
}
/// Value on the last day strictly before `date`, or null when there is none.
///
/// This is the lag=1 alignment the hand-typed values need: the portal publishes
@ -620,3 +730,172 @@ test "weights are inert when anchors are one session apart" {
try testing.expectApproxEqAbs(@as(f64, 10.25), vals[1], 1e-12);
}
}
test "projectForward walks forward with no downstream pin" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
const f = try fixture(arena);
const cal = try calendar(arena, f.model, f.tickers);
try testing.expectEqual(@as(usize, 3), cal.len);
// AA rises 10% a day, BB is flat, weights are 50/50, so the basket returns
// 5% a day and the walk is a plain compounding with nothing to pin it to.
const out = try projectForward(arena, f.model, f.tickers, cal, 0, 100.0);
try testing.expectEqual(@as(usize, 2), out.len);
try testing.expectEqualStrings("2020-01-02", out[0].date);
try testing.expectApproxEqRel(@as(f64, 105.0), out[0].unit_value, 1e-12);
try testing.expectEqualStrings("2020-01-03", out[1].date);
try testing.expectApproxEqRel(@as(f64, 110.25), out[1].unit_value, 1e-12);
}
test "projectForward yields nothing when no session is newer than the value" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// The weekday case, and the common one: the feed and the funds both top out
// at the same session, so there is no gap and no work. No date arithmetic
// decides this -- the calendar and the log simply agree.
const f = try fixture(arena);
const cal = try calendar(arena, f.model, f.tickers);
const out = try projectForward(arena, f.model, f.tickers, cal, cal.len - 1, 100.0);
try testing.expectEqual(@as(usize, 0), out.len);
}
test "projectForward reproduces the source value when the funds did not move" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
const models = try data.parseModel(arena,
\\#!srfv1
\\symbol::T,era_start::2020-01-01,ticker::AA,weight:num:100
\\
);
const aa = try candles.parse(arena, "AA", "#!srfv1\n" ++
"date::2020-01-01,adj_close:num:50\n" ++
"date::2020-01-02,adj_close:num:50\n");
const model = data.findModel(models, "T").?;
const tickers = try arena.dupe(candles.Ticker, &.{aa});
const cal = try calendar(arena, model, tickers);
const out = try projectForward(arena, model, tickers, cal, 0, 19.092721);
try testing.expectEqual(@as(usize, 1), out.len);
try testing.expectApproxEqRel(@as(f64, 19.092721), out[0].unit_value, 1e-12);
}
/// A single-fund model over `days` sessions, for horizon tests.
fn runFixture(arena: std.mem.Allocator, days: usize, scale: f64) !struct {
model: data.Model,
tickers: []const candles.Ticker,
cal: []const []const u8,
} {
const models = try data.parseModel(arena,
\\#!srfv1
\\symbol::T,era_start::2020-01-01,ticker::AA,weight:num:100
\\
);
var buf: std.ArrayList(u8) = .empty;
try buf.appendSlice(arena, "#!srfv1\n");
for (1..days + 1) |d| {
const fd: f64 = @floatFromInt(d);
try buf.print(arena, "date::2020-01-{d:0>2},adj_close:num:{d}\n", .{ d, (100.0 + fd) * scale });
}
const aa = try candles.parse(arena, "AA", buf.items);
const model = data.findModel(models, "T").?;
const tickers = try arena.dupe(candles.Ticker, &.{aa});
return .{ .model = model, .tickers = tickers, .cal = try calendar(arena, model, tickers) };
}
test "projectForward refuses a horizon longer than a week of sessions" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// A gap this wide means the feed is broken rather than closed. Refusing is
// what turns that into a job that fails loudly instead of a cache quietly
// serving an ever-longer unpinned extrapolation.
const f = try runFixture(arena, max_projected_sessions + 2, 1.0);
try testing.expectError(
error.ProjectionGapTooLarge,
projectForward(arena, f.model, f.tickers, f.cal, 0, 10.0),
);
// Exactly at the cap is still allowed.
const ok = try runFixture(arena, max_projected_sessions + 1, 1.0);
const out = try projectForward(arena, ok.model, ok.tickers, ok.cal, 0, 10.0);
try testing.expectEqual(@as(usize, max_projected_sessions), out.len);
}
test "projectForward rejects a start value that is not a positive price" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
const f = try fixture(arena);
const cal = try calendar(arena, f.model, f.tickers);
try testing.expectError(error.NonPositiveStart, projectForward(arena, f.model, f.tickers, cal, 0, 0));
try testing.expectError(error.GapOutOfRange, projectForward(arena, f.model, f.tickers, cal, cal.len, 10.0));
}
test "projectForward is invariant to rescaling the input price series" {
var arena_state: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Same property the reconstruction has, and it must hold here too: zfin
// re-bases a fund's whole adjusted series whenever adjustments are restated,
// and the walk consumes only per-day ratios, so the rescaling has to cancel.
const base = try runFixture(arena, 4, 1.0);
const scaled = try runFixture(arena, 4, 37.912345);
const a = try projectForward(arena, base.model, base.tickers, base.cal, 0, 10.0);
const b = try projectForward(arena, scaled.model, scaled.tickers, scaled.cal, 0, 10.0);
try testing.expectEqual(a.len, b.len);
try testing.expect(a.len > 0);
for (a, b) |x, y| {
try testing.expectEqualStrings(x.date, y.date);
try testing.expectApproxEqRel(x.unit_value, y.unit_value, 1e-12);
}
}
test "indexOfDate matches exactly rather than snapping" {
// Snapping would be actively wrong here: `projectForward` compounds
// `dayReturn(cal[i], cal[i+1])`, so an index that merely brackets the value's
// date would fold in a return the value already includes, or skip one.
const cal: []const []const u8 = &.{ "2020-01-01", "2020-01-03" };
try testing.expectEqual(@as(?usize, 0), indexOfDate(cal, "2020-01-01"));
try testing.expectEqual(@as(?usize, 1), indexOfDate(cal, "2020-01-03"));
// A date the funds have no bar for is not a session; the caller must decline
// to project rather than guess which side of it to start from.
try testing.expectEqual(@as(?usize, null), indexOfDate(cal, "2020-01-02"));
try testing.expectEqual(@as(?usize, null), indexOfDate(&.{}, "2020-01-01"));
}
test "planProjection fires only when the funds are ahead of the record" {
const cal: []const []const u8 = &.{ "2026-08-26", "2026-08-27", "2026-08-28" };
// Saturday: the funds have Friday, the feed stopped at Thursday.
try testing.expectEqual(@as(usize, 1), planProjection(cal, "2026-08-27").walk_from);
try testing.expectEqual(@as(usize, 0), planProjection(cal, "2026-08-26").walk_from);
// Weekday: both sides top out together, so there is nothing to do.
try testing.expectEqual(ProjectionPlan.up_to_date, planProjection(cal, "2026-08-28"));
// The feed running ahead of a lagging fund refresh is also "up to date", not
// a missing session: there is still nothing to project.
try testing.expectEqual(ProjectionPlan.up_to_date, planProjection(cal, "2026-08-31"));
// No priceable day at all.
try testing.expectEqual(ProjectionPlan.no_sessions, planProjection(&.{}, "2026-08-27"));
}
test "planProjection declines a recorded date the funds did not all price" {
// A published navDate with no bar behind it means some fund in the era is
// missing that day. Snapping to a neighbour would compound a return the
// recorded value already contains, so the only safe answer is to decline.
const cal: []const []const u8 = &.{ "2026-08-26", "2026-08-28" };
try testing.expectEqual(ProjectionPlan.date_not_a_session, planProjection(cal, "2026-08-27"));
}