Compare commits

...

3 commits

Author SHA1 Message Date
2876358c5a
quit from interactive mode means quit
All checks were successful
Generic zig build / build (push) Successful in 5m31s
Generic zig build / publish-macos (push) Successful in 13s
Generic zig build / deploy (push) Successful in 20s
2026-10-04 14:02:23 -07:00
4709403eca
switch codeberg to github due to ai 2026-10-04 10:54:47 -07:00
3e3b77ed54
found wells fargo xls-based download, adjust import/reconcile to use it 2026-10-04 10:52:52 -07:00
32 changed files with 2124 additions and 1665 deletions

View file

@ -17,7 +17,7 @@ jobs:
- name: Check out repository code
uses: actions/checkout@v4
- name: Setup Zig
uses: https://codeberg.org/mlugg/setup-zig@v2.2.1
uses: https://github.com/mlugg/setup-zig@v2.2.1
- name: Build project
run: zig build --summary all
- name: Run tests

View file

@ -994,3 +994,4 @@ command.
| [SRF](https://git.lerch.org/lobo/srf) | Cache file format, portfolio/watchlist parsing, serialization |
| [libvaxis](https://github.com/rockorager/libvaxis) (v0.6.0) | Terminal UI rendering |
| [z2d](https://github.com/vancluever/z2d) (v0.11.0) | Pixel chart rendering (Kitty graphics protocol) |
| [biff8](https://git.lerch.org/lobo/biff8) | Legacy `.xls` (BIFF8) reader for the Wells Fargo positions export |

View file

@ -189,7 +189,7 @@ src/
types.zig Normalized BrokeragePosition record + dollar parser
fidelity.zig Fidelity "Download Positions" CSV parser
schwab.zig Schwab export parser
wells_fargo.zig Wells Fargo export parser
wells_fargo.zig Wells Fargo positions spreadsheet (.xls) parser
views/
portfolio_sections.zig Portfolio view model (renderer-agnostic StyleIntent)
compare.zig Compare view model
@ -248,6 +248,7 @@ acknowledgments.srf Acknowledged review-tab findings (app-maintained)
| [libvaxis](https://github.com/rockorager/libvaxis) | Git (v0.6.0) | Terminal UI rendering |
| [z2d](https://github.com/vancluever/z2d) | Git (v0.11.0) | Pixel chart rendering (Kitty graphics protocol) |
| [zeit](https://github.com/rockorager/zeit) | Git (v0.9.0) | Calendar arithmetic + timezone conversion (ET) |
| [biff8](https://git.lerch.org/lobo/biff8) | Git | Wells Fargo `.xls` export reading |
## Building

View file

@ -35,6 +35,13 @@ pub fn build(b: *std.Build) void {
.optimize = optimize,
});
// biff8: read-only legacy .xls (BIFF8) reader, used by the Wells
// Fargo positions-export parser in src/brokerage/wells_fargo.zig.
const biff8_dep = b.dependency("biff8", .{
.target = target,
.optimize = optimize,
});
const srf_mod = srf_dep.module("srf");
const shiller_mod = b.addModule("shiller_year", .{
@ -58,6 +65,7 @@ pub fn build(b: *std.Build) void {
.imports = &.{
.{ .name = "srf", .module = srf_mod },
.{ .name = "zeit", .module = zeit_dep.module("zeit") },
.{ .name = "biff8", .module = biff8_dep.module("biff8") },
.{ .name = "build_info", .module = build_info },
},
});
@ -71,6 +79,7 @@ pub fn build(b: *std.Build) void {
.{ .name = "z2d", .module = z2d_dep.module("z2d") },
.{ .name = "zeit", .module = zeit_dep.module("zeit") },
.{ .name = "websocket", .module = websocket_dep.module("websocket") },
.{ .name = "biff8", .module = biff8_dep.module("biff8") },
.{ .name = "build_info", .module = build_info },
.{ .name = "shiller_year", .module = shiller_mod },
.{ .name = "config_docs", .module = configDocsModule(b) },
@ -140,6 +149,7 @@ pub fn build(b: *std.Build) void {
.imports = &.{
.{ .name = "srf", .module = srf_mod },
.{ .name = "zeit", .module = zeit_dep.module("zeit") },
.{ .name = "biff8", .module = biff8_dep.module("biff8") },
.{ .name = "build_info", .module = build_info },
},
}),

View file

@ -24,6 +24,10 @@
.url = "git+https://github.com/karlseguin/websocket.zig#99df0d3533a41cbcd5ff59b9af68bbfe6169cc62",
.hash = "websocket-0.1.0-ZPISdangBAAgWPap_MVU6Nk4rj3GT3xuJuF0IIv8HN6G",
},
.biff8 = .{
.url = "git+https://git.lerch.org/lobo/biff8#0d1b71a021a9fbd2de2e034b926505c8ec18872d",
.hash = "biff8-0.0.0-Wpwf6n66AQBH5r944iEGTKIVm4Gd0YTcRDER4VQPFkGW",
},
},
.paths = .{
"build",

View file

@ -20,14 +20,15 @@ from a supported broker.
## Supported brokers and how to export
`zfin audit` reconciles against **Fidelity** and **Schwab**. (Wells
Fargo is handled by [`import`](#what-about-wells-fargo), not audit.)
`zfin audit` reconciles against **Fidelity**, **Schwab**, and **Wells
Fargo**.
| Broker | How to export | Flag |
|--------------------------|-------------------------------------------------------------------------------------------------------------|--------------------|
| **Fidelity** | *Positions* tab -> the three-dot (**⋮**) menu -> **Download** (a CSV) | `--fidelity <CSV>` |
| **Schwab** (per-account) | *Accounts -> Positions* -> **Export** (one CSV per account) | `--schwab <CSV>` |
| **Schwab** (summary) | *Accounts -> Summary*: select the accounts table and copy it ([what to copy](#schwab-summary-what-to-copy)) | `--schwab-summary` |
| Broker | How to export | Flag |
|--------------------------|-------------------------------------------------------------------------------------------------------------|-----------------------|
| **Fidelity** | *Positions* tab -> the three-dot (**⋮**) menu -> **Download** (a CSV) | `--fidelity <CSV>` |
| **Schwab** (per-account) | *Accounts -> Positions* -> **Export** (one CSV per account) | `--schwab <CSV>` |
| **Schwab** (summary) | *Accounts -> Summary*: select the accounts table and copy it ([what to copy](#schwab-summary-what-to-copy)) | `--schwab-summary` |
| **Wells Fargo** | Download Type **Portfolio-Expanded Detail**, Portfolio View **Positions**, all brokerage accounts (an `.xls`) | `--wells-fargo <XLS>` |
The two Schwab inputs differ in detail: the **per-account CSV** has full
per-position data (shares, price, value); the **summary paste** carries
@ -260,10 +261,15 @@ What it considers:
`Portfolio_Positions_Jun-19.csv`; the
recency window keeps zfin reconciling *the one you just pulled*, not
last quarter's.
- **Detected by content, not filename.** zfin sniffs the first lines --
Fidelity begins `Account Number`/`Account Name`, a Schwab CSV begins
`"Positions for ...`, a Schwab summary contains `Account number ending
in`. A renamed file still works; an unrelated CSV is skipped.
- **Detected by content, not filename.** zfin sniffs the contents --
Fidelity by its legal footer, a Schwab CSV by its leading `"Positions
for ...`, a Schwab summary by `Account number ending in`, a Wells
Fargo spreadsheet by its `WFA_Positions` sheet. A renamed file still
works; an unrelated file is skipped.
- **A Wells Fargo export with none of your accounts is skipped.** One WF
download covers a whole household, so if you also manage someone
else's WF accounts, their export in your download folder is noted and
skipped rather than reported as eight unmapped accounts.
So with `ZFIN_AUDIT_FILES=~/Downloads`, the workflow collapses to
"download from your broker, run `zfin audit`, done."
@ -277,10 +283,11 @@ portfolio. It does that through
- The export carries an account number -- Schwab's from the "Positions
for account ...1234" title, Fidelity's from the Account Number column,
the summary's from "...ending in 1234".
the summary's from "...ending in 1234", Wells Fargo's from each row's
`*1234`.
- zfin finds the `accounts.srf` entry whose `institution::` (`fidelity`,
`schwab`) **and** `account_number::` match, and compares against that
account's lots.
`schwab`, `wells_fargo`) **and** `account_number::` match, and
compares against that account's lots.
- **No match -> the account is shown as `unmapped`** and flagged as a
discrepancy. Fix it by adding `institution::` and `account_number::`
to that account in `accounts.srf` (a placeholder number you recognize
@ -344,12 +351,13 @@ treatment to track drift. See
## Why it's finicky
- The parsers are **broker-specific and hardcode each export's column
layout** -- if Fidelity or Schwab changes their format, parsing can
break (Fidelity's header is validated to catch this; Schwab's is not).
They are not full RFC-4180 CSV parsers (no escaped quotes or
multi-line fields) -- fine for the real exports, not for arbitrary
CSVs.
- The parsers are **broker-specific**. The CSV parsers hardcode each
export's column layout -- if Fidelity or Schwab changes their format,
parsing can break (both validate their header to catch this). They are
not full RFC-4180 CSV parsers (no escaped quotes or multi-line fields)
-- fine for the real exports, not for arbitrary CSVs. The Wells Fargo
parser finds columns by header name and checks every per-account total
against the lots under it, so a layout change fails loudly.
- Matching is only as good as the `institution::` / `account_number::`
entries you keep in `accounts.srf`.
- Options, CDs, and cash are reconciled separately from share counts.
@ -358,13 +366,24 @@ None of this is a reason to skip it -- it's the single best way to keep
your records honest -- just know it expects some setup and an occasional
manual nudge.
### What about Wells Fargo?
### Wells Fargo: one file, every account, real lots
Wells Fargo's portal has no clean positions export, so it isn't an
`audit` target. Instead, [`zfin import --wells-fargo`](../reference/cli/import.md)
rebuilds a portfolio file from a paste of the WF positions table (copy
the rendered table from the brokerage portal and save it to a file).
Fidelity and Schwab exports can be imported the same way.
The Wells Fargo spreadsheet differs from the CSV exports in two ways
worth knowing:
- **It covers the whole household.** Every brokerage account is in one
file, each row tagged with its account as `*1234`.
- **It lists tax lots, not positions.** Audit sums them per account and
symbol before comparing, so it doesn't matter whether your portfolio
holds the position as one lot or many. It also means
[`zfin import --wells-fargo`](../reference/cli/import.md) can build a
portfolio file with every lot's real trade date and cost, which is the
easy way to keep a WF-managed portfolio current.
Pick **Portfolio-Expanded Detail**, not Collapsed: the collapsed download
has no lot rows, and zfin rejects it with a message saying so. Cash
(balance, sweep, and accrued interest) is compared per account against
your cash lots.
> **Keep brokerage exports private.** They contain real account numbers
> and holdings. Store them outside any git repository and delete them
@ -374,7 +393,7 @@ Fidelity and Schwab exports can be imported the same way.
- [`zfin audit` reference](../reference/cli/audit.md) -- every flag.
- [Map your accounts](set-up-accounts.md) -- the `institution` / `account_number` matching keys.
- [`zfin import`](../reference/cli/import.md) -- build a portfolio file *from* an export (including Wells Fargo).
- [`zfin import`](../reference/cli/import.md) -- build a portfolio file *from* an export.
---

View file

@ -16,17 +16,20 @@ discrepancies.
## Options
| Flag | Effect |
|--------------------|----------------------------------------------------------------------------|
| `--verbose` | Show full reconciliation output even when clean. |
| `--stale-days <N>` | Manual-price staleness threshold (default 3). |
| `--fidelity <CSV>` | Fidelity positions CSV ("All accounts" -> Positions tab -> Download). |
| `--schwab <CSV>` | Schwab per-account positions CSV. |
| `--schwab-summary` | Schwab account summary: paste from the summary page to stdin, then Ctrl-D. |
| Flag | Effect |
|-----------------------|---------------------------------------------------------------------------------------------|
| `--verbose` | Show full reconciliation output even when clean. |
| `--stale-days <N>` | Manual-price staleness threshold (default 3). |
| `--fidelity <CSV>` | Fidelity positions CSV ("All accounts" -> Positions tab -> Download). |
| `--schwab <CSV>` | Schwab per-account positions CSV. |
| `--schwab-summary` | Schwab account summary: paste from the summary page to stdin, then Ctrl-D. |
| `--wells-fargo <XLS>` | Wells Fargo positions spreadsheet (Download Type "Portfolio-Expanded Detail", "Positions"). |
Reconciliation matches export accounts to yours via `institution::` and
`account_number::` in [`accounts.srf`](../config/accounts-srf.md); an
unmatched account is reported as "unmapped."
unmatched account is reported as "unmapped." A discovered Wells Fargo
export none of whose accounts are mapped is skipped with a note instead,
since it is almost certainly another portfolio's.
The hygiene check also flags newly-appeared lots worth at least
$10,000 in a **Large new lots - confirm source** section, so you can

View file

@ -5,40 +5,63 @@ for managed accounts (direct-indexing baskets, accounts you don't track
at lot granularity).
```
Usage: zfin -p <PORTFOLIO> import (--fidelity FILE | --schwab FILE | --wells-fargo FILE [--account NAME]) [-y]
Usage: zfin -p <PORTFOLIO> import (--fidelity FILE | --schwab FILE | --wells-fargo FILE) [-y]
```
Each run **replaces** the target portfolio file with synthetic lots --
one per (account, symbol) -- drawn from the export. Per-buy history is
lost; git serves as the file-level history.
Each run **replaces** the target portfolio file with lots drawn from the
export:
- **Fidelity and Schwab** exports are per-position, so each (account,
symbol) becomes one synthetic lot. Per-buy history is lost; git serves
as the file-level history.
- **Wells Fargo** exports list every tax lot, so each WF lot becomes a
lot with its real trade date and cost. One WF file covers every
account in the household.
**Re-import merge:** when the target already exists, lots still present
in the new export keep their prior `open_date`, `open_price`, and
`note::`, so trailing-return and ST/LT classifications stay stable and
`git diff` flags only genuine brokerage changes. New positions get an
`open_date::1970-01-01` sentinel; disappeared positions are dropped.
Hand-edited fields (`price::`, `ticker::`) are **not** preserved.
in the new export keep their prior `note::` and every hand-edited field
(`ticker::`, `label::`, `price::`, `price_ratio::`, `drip::`, ...), so
`git diff` flags only genuine brokerage changes. A lot whose export row
has no buy date (every Fidelity/Schwab position, and a Wells Fargo
`Intra-Day` fund) also keeps its prior `open_date` and `open_price`; new
ones get an `open_date::1970-01-01` sentinel. Disappeared positions are
dropped.
## Options
| Flag | Effect |
|--------------------------|--------------------------------------------------------------------------------------|
| `-p, --portfolio <FILE>` | Target file (a single concrete path, not a glob). **Required.** |
| `--fidelity <CSV>` | Fidelity positions CSV. |
| `--schwab <CSV>` | Schwab per-account positions CSV. |
| `--wells-fargo <FILE>` | Wells Fargo positions paste (`-` for stdin). |
| `--account <NAME>` | (Wells Fargo only) account to attribute lots to; must match an `accounts.srf` entry. |
| `-y, --yes` | Don't prompt before overwriting an existing file. |
| Flag | Effect |
|--------------------------|-----------------------------------------------------------------|
| `-p, --portfolio <FILE>` | Target file (a single concrete path, not a glob). **Required.** |
| `--fidelity <CSV>` | Fidelity positions CSV. |
| `--schwab <CSV>` | Schwab per-account positions CSV. |
| `--wells-fargo <XLS>` | Wells Fargo positions spreadsheet (`-` for stdin). See below. |
| `-y, --yes` | Don't prompt before overwriting an existing file. |
Account resolution needs an [`accounts.srf`](../config/accounts-srf.md)
next to the target with `institution::` + `account_number::` entries
matching the export; import refuses to write when an export account is
unmapped.
A target that doesn't exist yet is created in the current directory, so
run a first import from the portfolio's own directory.
## Wells Fargo
On the WF site, download the positions spreadsheet with **Download
Type** "Portfolio-Expanded Detail", **Portfolio View** "Positions", for
all brokerage accounts. It saves as `WFA_Portfolio_Positions_*.xls`.
- "Portfolio-Collapsed Detail" has no per-lot rows; import rejects it.
- Each row names its account as `*1234`; map it with
`institution::wells_fargo,account_number::1234`.
- Cash is one lot per account: cash balance, sweep, and accrued
interest combined, matching WF's own cash total.
## Example
```bash
zfin -p portfolio_managed.srf import --fidelity ~/Downloads/Portfolio_Positions.csv
zfin -p portfolio_wells_fargo.srf import --wells-fargo ~/Downloads/WFA_Portfolio_Positions_100326_1247.xls
```
## See also

View file

@ -28,8 +28,8 @@ account::Joint taxable,tax_type::taxable,institution::schwab,account_number::JT0
|-----------------------------|--------|----------|-----------|----------------------------------------------------------------------------------------------------------------------------------|
| `account` | string | Yes | -- | Account name; must match `account::` on lots exactly. |
| `tax_type` | string | Yes | -- | `taxable`, `roth`, `traditional`, or `hsa`. |
| `institution` | string | No | -- | Broker key, e.g. `fidelity`, `schwab`, `vanguard`, `wells_fargo`. Used by [`zfin audit`](../cli/audit.md) to match export files. |
| `account_number` | string | No | -- | Account identifier used with `institution` for audit matching. Use a placeholder, not a full real number. |
| `institution` | string | No | -- | Broker key, e.g. `fidelity`, `schwab`, `vanguard`, `wells_fargo`. Used by `zfin audit` and `zfin import` to match export rows. |
| `account_number` | string | No | -- | Matched with `institution` against export rows (WF: `1234` for `*1234`). Use a placeholder, not a full real number. |
| `update_cadence` | string | No | `weekly` | How often you refresh this account's manual data: `weekly`, `monthly`, `quarterly`, or `none`. Drives the audit staleness nag. |
| `cash_is_contribution` | bool | No | `false` | When `true`, raw cash-balance increases on this account count as real external contributions (see below). |
| `direct_indexing` | bool | No | `false` | Marks an account whose lots track a benchmark with tracking-error drift (loosens contribution/audit tolerances). |

View file

@ -81,6 +81,20 @@ pub fn parse(str: []const u8) !Date {
return fromYmd(y, m, d);
}
/// Parse US "MM/DD/YYYY" format, as brokerage exports print dates.
/// Exactly two-digit month and day and a four-digit year; the month
/// and day are range-checked against the calendar (so `02/30/2026`
/// is rejected rather than rolled into March), because the input
/// comes from external files rather than zfin's own serializer.
pub fn parseMdy(str: []const u8) !Date {
if (str.len != 10 or str[2] != '/' or str[5] != '/') return error.InvalidDateFormat;
const m = std.fmt.parseInt(u8, str[0..2], 10) catch return error.InvalidDateFormat;
const d = std.fmt.parseInt(u8, str[3..5], 10) catch return error.InvalidDateFormat;
const y = std.fmt.parseInt(i16, str[6..10], 10) catch return error.InvalidDateFormat;
if (m < 1 or m > 12 or d < 1 or d > daysInMonth(y, m)) return error.InvalidDateFormat;
return fromYmd(y, m, d);
}
/// Hook for srf coercion via `FieldIterator.to(T, ...)`. Returns a
/// `CoercionResult(Date)` with `require_free_original = true` so SRF
/// frees the consumed source string after parsing.
@ -645,3 +659,29 @@ test "parse error cases" {
try std.testing.expectError(error.InvalidDateFormat, Date.parse("20240115")); // no dashes
try std.testing.expectError(error.InvalidDateFormat, Date.parse("2024/01/15")); // wrong separator
}
test "parseMdy" {
try std.testing.expect(Date.fromYmd(2026, 10, 3).eql(try Date.parseMdy("10/03/2026")));
try std.testing.expect(Date.fromYmd(2024, 2, 29).eql(try Date.parseMdy("02/29/2024"))); // leap day
try std.testing.expect(Date.fromYmd(1999, 12, 31).eql(try Date.parseMdy("12/31/1999")));
}
test "parseMdy error cases" {
const bad = [_][]const u8{
"2026-10-03", // ISO, not US
"10/3/2026", // single-digit day
"10-03-2026", // wrong separator
"10/03/26", // two-digit year
"aa/03/2026",
"10/bb/2026",
"10/03/yyyy",
"00/10/2026", // month 0
"13/01/2026", // month 13
"01/00/2026", // day 0
"02/30/2026", // past month end
"02/29/2026", // not a leap year
"Intra-Day", // what brokerages print when there is no date
"",
};
for (bad) |s| try std.testing.expectError(error.InvalidDateFormat, Date.parseMdy(s));
}

View file

@ -1187,7 +1187,7 @@ fn dividendsWorker(self: *PortfolioData, delay_ms: usize) void {
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_options);
self.svc.loadAllDividends(div_syms, self.fetch_options) catch return;
}
var map = std.StringHashMap([]const Dividend).init(arena_alloc);

View file

@ -6,10 +6,12 @@
//! cdLotAllowance, findAbsentAccounts, ...)
//! - `schwab` Schwab positions CSV + summary reconcilers
//! - `fidelity` Fidelity positions CSV reconciler
//! - `wells_fargo` Wells Fargo positions spreadsheet reconciler
pub const common = @import("reconcile/common.zig");
pub const schwab = @import("reconcile/schwab.zig");
pub const fidelity = @import("reconcile/fidelity.zig");
pub const wells_fargo = @import("reconcile/wells_fargo.zig");
// ── Flat convenience re-exports ──────────────────────────────
pub const value_tolerance = common.value_tolerance;
@ -35,3 +37,4 @@ pub const hasSchwabDiscrepancies = schwab.hasSchwabDiscrepancies;
pub const summaryRatioSuggestions = schwab.summaryRatioSuggestions;
pub const reconcileFidelity = fidelity.reconcile;
pub const reconcileWellsFargo = wells_fargo.reconcile;

View file

@ -0,0 +1,182 @@
//! Wells Fargo reconciler (pure compute).
//!
//! WF's positions export is one spreadsheet for the whole household
//! (see `brokerage/wells_fargo.zig`). This module wires its positions
//! into the shared per-account comparison engine in `common.zig`; the
//! export lists one row per tax lot, which `compareAccounts` sums per
//! (account, symbol) like any multi-row position. The ANSI display
//! lives in `commands/audit/`.
const std = @import("std");
const biff8 = @import("biff8");
const zfin = @import("../../root.zig");
const analysis = @import("../../analytics/analysis.zig");
const Date = @import("../../Date.zig");
const common = @import("common.zig");
const wf_parser = @import("../../brokerage/wells_fargo.zig");
/// Reconcile the positions sheet against the portfolio. Returns owned
/// `AccountComparison` results (free each `.comparisons` slice, then
/// the results slice). String fields in the results borrow from
/// `sheet`, which must outlive them.
pub fn reconcile(
allocator: std.mem.Allocator,
portfolio: zfin.Portfolio,
sheet: *const biff8.Sheet,
account_map: analysis.AccountMap,
prices: std.StringHashMap(f64),
as_of: Date,
) ![]common.AccountComparison {
const positions = try wf_parser.parsePositions(allocator, sheet);
// Result strings borrow from the sheet, not from this slice.
defer allocator.free(positions);
return common.compareAccounts(allocator, portfolio, positions, account_map, wf_parser.institution, prices, as_of);
}
/// A reconciliation of an export file, owning the decoded workbook
/// its results borrow from.
pub const Reconciled = struct {
workbook: biff8.Workbook,
results: []common.AccountComparison,
pub fn deinit(self: *Reconciled, allocator: std.mem.Allocator) void {
for (self.results) |r| allocator.free(r.comparisons);
allocator.free(self.results);
self.workbook.deinit();
}
/// True when at least one of the export's accounts is in
/// `accounts.srf`. An export with none is almost certainly another
/// portfolio's (one household's download sitting in a shared
/// Downloads directory), so auto-discovery skips it rather than
/// reporting every account as a discrepancy.
pub fn mapsAnyAccount(self: Reconciled) bool {
for (self.results) |r| {
if (r.account_name.len > 0) return true;
}
return false;
}
};
/// Decode an export file and reconcile it. Errors are the decode's
/// (`wf_parser.ExportError`) and allocation failure; use
/// `wf_parser.exportHint` to explain the former.
pub fn reconcileXls(
allocator: std.mem.Allocator,
portfolio: zfin.Portfolio,
bytes: []const u8,
account_map: analysis.AccountMap,
prices: std.StringHashMap(f64),
as_of: Date,
) wf_parser.ExportError!Reconciled {
var workbook = try biff8.Workbook.parse(allocator, bytes);
errdefer workbook.deinit();
const sheet = try wf_parser.positionsSheet(&workbook);
const results = try reconcile(allocator, portfolio, sheet, account_map, prices, as_of);
return .{ .workbook = workbook, .results = results };
}
// ---- Tests ----
const Lot = @import("../../models/portfolio.zig").Lot;
const Portfolio = @import("../../models/portfolio.zig").Portfolio;
const Cell = biff8.Cell;
fn txt(s: []const u8) Cell {
return .{ .text = s };
}
fn num(v: f64) Cell {
return .{ .number = v };
}
const header = [_]Cell{ txt("Description"), txt("Symbol"), txt("Account Number"), txt("Market Value"), txt("Shares"), txt("Total Cost"), txt("Trade Date") };
/// Placeholder household: VTI held as two lots in *1234, plus cash.
const rows = [_][]const Cell{
&.{ txt("Description"), txt("Account Number"), txt("Market Value") },
&.{ txt("Cash Balance"), txt("*1234"), num(40) },
&.{ txt("Bank Deposit Sweep"), txt("*1234"), num(60) },
&.{txt("Total Cash/Cash Alternatives & Margin")},
&header,
&.{ txt("VANGUARD TOTAL STOCK MKT"), txt("VTI"), txt("*1234"), num(6000), num(20), num(4500), txt("Detail") },
&.{ txt("VANGUARD TOTAL STOCK MKT"), txt("VTI"), txt("*1234"), num(3600), num(12), num(2400), txt("02/03/2022") },
&.{ txt("VANGUARD TOTAL STOCK MKT"), txt("VTI"), txt("*1234"), num(2400), num(8), num(2100), txt("09/09/2026") },
&.{txt("Total ETFs")},
};
const test_sheet: biff8.Sheet = .{ .name = wf_parser.sheet_name, .rows = &rows };
test "reconcile: per-lot rows and cash rows sum per account and match the portfolio" {
const allocator = std.testing.allocator;
// The portfolio may hold the position as one lot or as lots of its
// own; either way 20 VTI @ $300 plus $100 of cash matches.
var lots = [_]Lot{
.{ .symbol = "VTI", .shares = 20, .open_date = Date.fromYmd(2022, 2, 3), .open_price = 225, .account = "Sample IRA" },
.{ .symbol = "", .shares = 100, .open_date = Date.epoch, .open_price = 1, .account = "Sample IRA", .security_type = .cash },
};
const portfolio = Portfolio{ .lots = &lots, .allocator = allocator };
var entries = [_]analysis.AccountTaxEntry{
.{ .account = "Sample IRA", .tax_type = .traditional, .institution = "wells_fargo", .account_number = "1234" },
};
const acct_map = analysis.AccountMap{ .entries = &entries, .allocator = allocator };
var prices = std.StringHashMap(f64).init(allocator);
defer prices.deinit();
try prices.put("VTI", 300.0);
const results = try reconcile(allocator, portfolio, &test_sheet, acct_map, prices, Date.fromYmd(2026, 10, 3));
defer {
for (results) |r| allocator.free(r.comparisons);
allocator.free(results);
}
try std.testing.expectEqual(@as(usize, 1), results.len);
try std.testing.expectEqualStrings("Sample IRA", results[0].account_name);
try std.testing.expectEqualStrings("*1234", results[0].brokerage_name);
try std.testing.expect(!results[0].has_discrepancies);
// VTI consolidated into one comparison, cash into another.
try std.testing.expectEqual(@as(usize, 2), results[0].comparisons.len);
}
test "Reconciled.mapsAnyAccount: false when no export account is in accounts.srf" {
const allocator = std.testing.allocator;
const portfolio = Portfolio{ .lots = &.{}, .allocator = allocator };
var entries = [_]analysis.AccountTaxEntry{
.{ .account = "Sample Brokerage", .tax_type = .taxable, .institution = "wells_fargo", .account_number = "9012" },
};
const acct_map = analysis.AccountMap{ .entries = &entries, .allocator = allocator };
var prices = std.StringHashMap(f64).init(allocator);
defer prices.deinit();
var rec: Reconciled = .{
.workbook = .{ .arena = std.heap.ArenaAllocator.init(allocator), .sheets = &.{} },
.results = try reconcile(allocator, portfolio, &test_sheet, acct_map, prices, Date.fromYmd(2026, 10, 3)),
};
defer rec.deinit(allocator);
try std.testing.expect(!rec.mapsAnyAccount());
// With the export's account mapped, it is ours.
entries[0].account_number = "1234";
var mapped: Reconciled = .{
.workbook = .{ .arena = std.heap.ArenaAllocator.init(allocator), .sheets = &.{} },
.results = try reconcile(allocator, portfolio, &test_sheet, acct_map, prices, Date.fromYmd(2026, 10, 3)),
};
defer mapped.deinit(allocator);
try std.testing.expect(mapped.mapsAnyAccount());
}
test "reconcileXls: decode errors surface with their names" {
const allocator = std.testing.allocator;
const portfolio = Portfolio{ .lots = &.{}, .allocator = allocator };
var entries = [_]analysis.AccountTaxEntry{};
const acct_map = analysis.AccountMap{ .entries = &entries, .allocator = allocator };
var prices = std.StringHashMap(f64).init(allocator);
defer prices.deinit();
try std.testing.expectError(
error.NotCompoundFile,
reconcileXls(allocator, portfolio, "Symbol,Quantity\nVTI,1\n", acct_map, prices, Date.fromYmd(2026, 10, 3)),
);
}

View file

@ -1,10 +1,11 @@
//! Brokerage export parsers, grouped for downstream consumers.
//!
//! Each broker module parses its export (positions CSV, and for Schwab
//! an account-summary paste) into the normalized `BrokeragePosition`
//! shape in `types.zig`. Pure functions: `(allocator, data) -> parsed`,
//! no IO. The reconciliation layer (`analytics/reconcile`) and finrev
//! consume these.
//! Each broker module parses its export (positions CSV, for Schwab
//! an account-summary paste, for Wells Fargo a positions spreadsheet)
//! into the normalized `BrokeragePosition` shape in `types.zig`. Pure
//! functions: `(allocator, data) -> parsed`, no IO. The
//! reconciliation layer (`analytics/reconcile`) and finrev consume
//! these.
pub const types = @import("brokerage/types.zig");
pub const schwab = @import("brokerage/schwab.zig");

View file

@ -14,9 +14,12 @@
//! keeps this module free of a buffer-lifetime contract.
const std = @import("std");
const biff8 = @import("biff8");
const wells_fargo = @import("wells_fargo.zig");
/// Size ceiling for non-CSV candidates. A CSV is exempt because a positions
/// export legitimately gets large; anything else this big is not an export.
/// Size ceiling for candidates that are not spreadsheet exports. CSV and
/// XLS are exempt because a positions export legitimately gets large;
/// anything else this big is not an export.
const max_size_non_csv = 512 * 1024;
/// Type of a discovered brokerage file.
@ -24,6 +27,7 @@ pub const BrokerFileKind = enum {
fidelity_csv,
schwab_csv,
schwab_summary,
wells_fargo_xls,
};
/// A discovered brokerage file ready for reconciliation.
@ -41,6 +45,14 @@ pub const DiscoveredFile = struct {
/// stable self-identifying marker (rather than the column header) means a
/// format tweak surfaces as a specific parse error, not a misroute.
pub fn detectBrokerFileKind(data: []const u8) ?BrokerFileKind {
// Wells Fargo positions spreadsheet: a binary .xls whose one sheet is
// named `WFA_Positions`. The sheet name is stored as plain ASCII in
// the workbook's sheet directory, so a byte search finds it without
// decoding the file; the compound-file signature keeps a text file
// that merely mentions the name from matching.
if (biff8.isCompoundFile(data) and std.mem.indexOf(u8, data, wells_fargo.sheet_name) != null)
return .wells_fargo_xls;
// Strip optional UTF-8 BOM
const content = if (data.len >= 3 and data[0] == 0xEF and data[1] == 0xBB and data[2] == 0xBF)
data[3..]
@ -104,9 +116,9 @@ pub fn brokerFiles(
const age_s = now_s - mtime_s;
if (age_s > max_age_s) continue;
// Check if it's a CSV (no size limit) or non-CSV (size limit applies)
const is_csv = std.mem.endsWith(u8, entry.name, ".csv") or std.mem.endsWith(u8, entry.name, ".CSV");
if (!is_csv and stat.size > max_size_non_csv) continue;
// Spreadsheet exports (no size limit) vs everything else (size limit applies)
const is_export_format = isSpreadsheetName(entry.name);
if (!is_export_format and stat.size > max_size_non_csv) continue;
// Read and detect content type
const data = dir.readFileAlloc(io, entry.name, allocator, .limited(10 * 1024 * 1024)) catch continue;
@ -124,6 +136,32 @@ pub fn brokerFiles(
return results.toOwnedSlice(allocator);
}
/// `.csv` or `.xls`, either case: the formats brokerages export
/// positions in, exempt from the non-export size ceiling.
fn isSpreadsheetName(name: []const u8) bool {
const ext = std.fs.path.extension(name);
return std.ascii.eqlIgnoreCase(ext, ".csv") or std.ascii.eqlIgnoreCase(ext, ".xls");
}
test "isSpreadsheetName" {
try std.testing.expect(isSpreadsheetName("positions.csv"));
try std.testing.expect(isSpreadsheetName("POSITIONS.CSV"));
try std.testing.expect(isSpreadsheetName("WFA_Portfolio_Positions.xls"));
try std.testing.expect(isSpreadsheetName("export.XLS"));
try std.testing.expect(!isSpreadsheetName("summary.txt"));
try std.testing.expect(!isSpreadsheetName("book.xlsx"));
try std.testing.expect(!isSpreadsheetName("csv"));
}
test "detectBrokerFileKind: wells fargo xls identified by signature + sheet name" {
const sig = [_]u8{ 0xD0, 0xCF, 0x11, 0xE0, 0xA1, 0xB1, 0x1A, 0xE1 };
try std.testing.expectEqual(BrokerFileKind.wells_fargo_xls, detectBrokerFileKind(&sig ++ "....WFA_Positions....").?);
// Some other legacy Office file: compound file, no WF sheet.
try std.testing.expect(detectBrokerFileKind(&sig ++ "....Sheet1....") == null);
// A text file that mentions the sheet name is not a workbook.
try std.testing.expect(detectBrokerFileKind("notes about WFA_Positions") == null);
}
test "detectBrokerFileKind: fidelity csv identified by legal footer" {
// Detection keys on the self-identifying legal-entity footer, not the
// column header (which Fidelity has already re-cased). A realistic

View file

@ -30,6 +30,12 @@
//! shape - exactly what the audit reconciler needs to compare
//! against the portfolio's mapped view.
//!
//! Wells Fargo's positions export is the exception to "no buy date":
//! it lists every tax lot with its trade date, so its parser builds
//! real `Lot`s for import directly (`wells_fargo.parseLots`) rather
//! than going through a position. `lotFromPosition` below is the one
//! conversion rule both paths share.
//!
//! ## Memory & lifetime contract
//!
//! All string fields in records produced by the per-broker parsers
@ -39,6 +45,9 @@
//! allocator.
const std = @import("std");
const Date = @import("../Date.zig");
const Lot = @import("../models/portfolio.zig").Lot;
const LotType = @import("../models/portfolio.zig").LotType;
// ── Brokerage position (normalized from any source) ─────────
@ -56,6 +65,44 @@ pub const BrokeragePosition = struct {
is_cash: bool,
};
/// The lot an export row becomes on import, for portfolio account
/// `account`:
///
/// - cash: `security_type::cash`, shares = the dollar value at a
/// $1.00 open price (the convention audit and snapshot use for
/// cash rows)
/// - otherwise: shares from the row, open price = cost basis per
/// share, falling back to current value per share, else 0
/// - `open_date` is `Date.epoch`, import's "unknown" sentinel; a
/// caller that knows the buy date overwrites it
///
/// A non-cash row without a quantity is malformed export data; it
/// becomes a harmless 0-share lot the user will see in `git diff`.
///
/// Strings are borrowed from `pos` and `account`.
pub fn lotFromPosition(pos: BrokeragePosition, account: []const u8) Lot {
if (pos.is_cash) {
return .{
.symbol = pos.symbol,
.shares = pos.quantity orelse pos.current_value orelse 0,
.open_date = Date.epoch,
.open_price = 1.0,
.account = account,
.security_type = .cash,
};
}
const shares = pos.quantity orelse 0;
const total: ?f64 = pos.cost_basis orelse pos.current_value;
return .{
.symbol = pos.symbol,
.shares = shares,
.open_date = Date.epoch,
.open_price = if (total) |t| (if (shares > 0) t / shares else 0) else 0,
.account = account,
.security_type = .stock,
};
}
// ── Dollar-string parsing ────────────────────────────────────
/// Parse a dollar amount string like "$1,234.56", "+$3,732.40", "-$6,300.00".
@ -132,3 +179,36 @@ test "isUnitPriceCash: unparseable inputs return false" {
try std.testing.expect(!isUnitPriceCash("$1.00", ""));
try std.testing.expect(!isUnitPriceCash("N/A", "$1.00"));
}
fn testPosition(quantity: ?f64, value: ?f64, cost: ?f64, is_cash: bool) BrokeragePosition {
return .{ .account_number = "1234", .account_name = "", .symbol = "SAMPLE", .description = "", .quantity = quantity, .current_value = value, .cost_basis = cost, .is_cash = is_cash };
}
test "lotFromPosition: stock open price is cost basis per share" {
const lot = lotFromPosition(testPosition(100, 17500, 12000, false), "Sample Brokerage");
try std.testing.expectEqualStrings("SAMPLE", lot.symbol);
try std.testing.expectEqualStrings("Sample Brokerage", lot.account.?);
try std.testing.expectEqual(LotType.stock, lot.security_type);
try std.testing.expectEqual(@as(f64, 100), lot.shares);
try std.testing.expectEqual(@as(f64, 120), lot.open_price);
try std.testing.expect(Date.epoch.eql(lot.open_date));
}
test "lotFromPosition: no cost basis falls back to value per share, then 0" {
try std.testing.expectEqual(@as(f64, 400), lotFromPosition(testPosition(10, 4000, null, false), "A").open_price);
try std.testing.expectEqual(@as(f64, 0), lotFromPosition(testPosition(10, null, null, false), "A").open_price);
// No quantity: a harmless zero-share lot rather than a divide by zero.
const no_qty = lotFromPosition(testPosition(null, 4000, 3000, false), "A");
try std.testing.expectEqual(@as(f64, 0), no_qty.shares);
try std.testing.expectEqual(@as(f64, 0), no_qty.open_price);
}
test "lotFromPosition: cash is dollars at a $1 open price" {
const lot = lotFromPosition(testPosition(null, 5000, null, true), "A");
try std.testing.expectEqual(LotType.cash, lot.security_type);
try std.testing.expectEqual(@as(f64, 5000), lot.shares);
try std.testing.expectEqual(@as(f64, 1.0), lot.open_price);
// A cash row that reports a quantity uses it.
try std.testing.expectEqual(@as(f64, 4999), lotFromPosition(testPosition(4999, 5000, null, true), "A").shares);
try std.testing.expectEqual(@as(f64, 0), lotFromPosition(testPosition(null, null, null, true), "A").shares);
}

File diff suppressed because it is too large Load diff

View file

@ -4,9 +4,12 @@
//! per-responsibility modules in the `audit/` directory:
//!
//! - `audit/hygiene.zig` - flagless portfolio hygiene check (no flags)
//! - `audit/fidelity.zig` - `--fidelity` positions-CSV reconciler
//! - `analytics/reconcile/fidelity.zig` - `--fidelity` positions-CSV
//! reconciler
//! - `audit/schwab.zig` - `--schwab` positions-CSV + `--schwab-summary`
//! reconcilers
//! - `analytics/reconcile/wells_fargo.zig` - `--wells-fargo`
//! positions-spreadsheet reconciler
//! - `audit/common.zig` - shared comparison types + per-account display
//!
//! This file sits beside its `audit/` directory (the `tui.zig` +
@ -23,6 +26,8 @@ const framework = @import("framework.zig");
const common = @import("audit/common.zig");
const fidelity = @import("../analytics/reconcile/fidelity.zig");
const wells_fargo = @import("../analytics/reconcile/wells_fargo.zig");
const wf_parser = @import("../brokerage/wells_fargo.zig");
const schwab = @import("audit/schwab.zig");
const hygiene = @import("audit/hygiene.zig");
@ -32,6 +37,7 @@ pub const ParsedArgs = struct {
fidelity_csv: ?[]const u8 = null,
schwab_csv: ?[]const u8 = null,
schwab_summary: bool = false,
wells_fargo_xls: ?[]const u8 = null,
verbose: bool = false,
stale_days: u32 = hygiene.default_stale_days,
};
@ -60,10 +66,13 @@ pub const meta: framework.Meta = .{
\\ --schwab <CSV> Schwab per-account positions CSV export
\\ --schwab-summary Schwab account summary; copy from accounts
\\ summary page, paste to stdin, then ^D
\\ --wells-fargo <XLS> Wells Fargo positions spreadsheet
\\ (Download Type "Portfolio-Expanded Detail",
\\ Portfolio View "Positions")
\\
,
.uppercase_first_arg = false,
.user_errors = error{ UnexpectedArg, EmptyFile, NoAccountsFound, UnexpectedHeader, MissingFlagValue },
.user_errors = error{ UnexpectedArg, EmptyFile, NoAccountsFound, UnexpectedHeader, MissingFlagValue, InvalidExport },
};
pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedArgs {
@ -77,6 +86,8 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr
parsed.schwab_csv = try cli.requireFlagValue(ctx.io, cmd_args, &i, a);
} else if (std.mem.eql(u8, a, "--schwab-summary")) {
parsed.schwab_summary = true;
} else if (std.mem.eql(u8, a, "--wells-fargo")) {
parsed.wells_fargo_xls = try cli.requireFlagValue(ctx.io, cmd_args, &i, a);
} else if (std.mem.eql(u8, a, "--verbose")) {
parsed.verbose = true;
} else if (std.mem.eql(u8, a, "--stale-days")) {
@ -115,6 +126,7 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
const fidelity_csv = parsed.fidelity_csv;
const schwab_csv = parsed.schwab_csv;
const schwab_summary = parsed.schwab_summary;
const wells_fargo_xls = parsed.wells_fargo_xls;
const verbose = parsed.verbose;
const stale_days = parsed.stale_days;
@ -122,7 +134,7 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
// semantics (git blame, commit SHAs, etc.), so resolve the anchor -
// but the large-lot check inside diffs portfolio content and needs
// the merged view, so hand it the whole glob too.
if (fidelity_csv == null and schwab_csv == null and !schwab_summary) {
if (fidelity_csv == null and schwab_csv == null and !schwab_summary and wells_fargo_xls == null) {
const pf = ctx.resolvePortfolioPath();
defer pf.deinit(allocator);
var all = ctx.resolvePortfolioPaths() catch null;
@ -131,7 +143,8 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
return hygiene.runHygieneCheck(io, allocator, ctx.environ_map, svc, pf.path, paths, stale_days, verbose, as_of, now_s, color, ctx.globals.refresh_policy, out);
}
// Reconciliation modes (--fidelity / --schwab / --schwab-summary):
// Reconciliation modes (--fidelity / --schwab / --schwab-summary /
// --wells-fargo):
// load the union of all portfolio files so the comparison sees
// every lot the user holds, even if they're split across multiple
// portfolio_*.srf files.
@ -284,6 +297,38 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
defer allocator.free(absent);
try common.displayAbsentAccounts(absent, color, "this export", out);
}
// Wells Fargo positions spreadsheet (one file, every WF account)
if (wells_fargo_xls) |xls_path| {
const xls_data = std.Io.Dir.cwd().readFileAlloc(io, xls_path, allocator, .limited(10 * 1024 * 1024)) catch |err| {
var msg_buf: [512]u8 = undefined;
const msg = std.fmt.bufPrint(&msg_buf, "Error: Cannot read Wells Fargo export {s}: {s}\n", .{ xls_path, @errorName(err) }) catch "Error: Cannot read Wells Fargo export\n";
cli.stderrPrint(io, msg);
return error.InvalidExport;
};
defer allocator.free(xls_data);
var rec = wells_fargo.reconcileXls(allocator, portfolio, xls_data, account_map, prices, as_of) catch |err| switch (err) {
error.OutOfMemory => return err,
else => |e| {
var msg_buf: [512]u8 = undefined;
const msg = std.fmt.bufPrint(&msg_buf, "Error: Cannot read Wells Fargo export {s}: {s}\n", .{ xls_path, @errorName(e) }) catch "Error: Cannot read Wells Fargo export\n";
cli.stderrPrint(io, msg);
cli.stderrPrint(io, wf_parser.exportHint(e));
return error.InvalidExport;
},
};
defer rec.deinit(allocator);
try common.displayResults(rec.results, color, out);
try common.displayRatioSuggestions(allocator, rec.results, portfolio, prices, account_map, color, out);
const present = try common.presentNumbers(allocator, common.AccountComparison, rec.results);
defer allocator.free(present);
const absent = try common.findAbsentAccounts(allocator, portfolio, account_map, wf_parser.institution, present, prices, as_of);
defer allocator.free(absent);
try common.displayAbsentAccounts(absent, color, "this export", out);
}
}
// ── Tests ────────────────────────────────────────────────────
@ -296,6 +341,7 @@ test "parseArgs: defaults" {
try std.testing.expect(parsed.fidelity_csv == null);
try std.testing.expect(parsed.schwab_csv == null);
try std.testing.expect(!parsed.schwab_summary);
try std.testing.expect(parsed.wells_fargo_xls == null);
try std.testing.expect(!parsed.verbose);
try std.testing.expectEqual(hygiene.default_stale_days, parsed.stale_days);
}
@ -316,6 +362,21 @@ test "parseArgs: --schwab captures CSV path" {
try std.testing.expectEqualStrings("/tmp/sch.csv", parsed.schwab_csv.?);
}
test "parseArgs: --wells-fargo captures the spreadsheet path" {
var ctx: framework.RunCtx = undefined;
ctx.io = std.testing.io;
const args = [_][]const u8{ "--wells-fargo", "wf.xls" };
const parsed = try parseArgs(&ctx, &args);
try std.testing.expectEqualStrings("wf.xls", parsed.wells_fargo_xls.?);
}
test "parseArgs: --wells-fargo without a value is rejected" {
var ctx: framework.RunCtx = undefined;
ctx.io = std.testing.io;
const args = [_][]const u8{"--wells-fargo"};
try std.testing.expectError(error.MissingFlagValue, parseArgs(&ctx, &args));
}
test "parseArgs: --schwab-summary boolean" {
var ctx: framework.RunCtx = undefined;
ctx.io = std.testing.io;

View file

@ -31,6 +31,8 @@ const test_git = @import("../../testutil/git.zig");
const common = @import("common.zig");
const fidelity = @import("../../analytics/reconcile/fidelity.zig");
const wells_fargo = @import("../../analytics/reconcile/wells_fargo.zig");
const wf_parser = @import("../../brokerage/wells_fargo.zig");
const schwab = @import("schwab.zig");
const discover = zfin.brokerage.discover;
const fmt = cli.fmt;
@ -1087,6 +1089,7 @@ pub fn runHygieneCheck(
.fidelity_csv => "fidelity",
.schwab_csv => "schwab csv",
.schwab_summary => "schwab summary",
.wells_fargo_xls => "wells fargo",
};
try out.print(" {s:<52} {s}\n", .{ f.path, kind_label });
}
@ -1131,11 +1134,14 @@ pub fn runHygieneCheck(
// `file_data`, which is freed per loop iteration.
var fidelity_present: std.ArrayList([]const u8) = .empty;
var schwab_present: std.ArrayList([]const u8) = .empty;
var wells_fargo_present: std.ArrayList([]const u8) = .empty;
defer {
for (fidelity_present.items) |s| allocator.free(s);
fidelity_present.deinit(allocator);
for (schwab_present.items) |s| allocator.free(s);
schwab_present.deinit(allocator);
for (wells_fargo_present.items) |s| allocator.free(s);
wells_fargo_present.deinit(allocator);
}
for (all_files.items) |f| {
@ -1211,6 +1217,36 @@ pub fn runHygieneCheck(
try accumulatePresent(allocator, &schwab_present, common.AccountComparison, results);
},
.wells_fargo_xls => {
var rec = wells_fargo.reconcileXls(allocator, portfolio, file_data, account_map, prices, as_of) catch |err| switch (err) {
error.OutOfMemory => return err,
else => |e| {
try cli.printFg(out, color, cli.CLR_WARNING, " {s}: detected as wells fargo export but could not parse ({s}); skipped\n", .{ f.path, @errorName(e) });
continue;
},
};
defer rec.deinit(allocator);
// One WF download covers a whole household. If none of
// its accounts is in this portfolio's accounts.srf, it is
// someone else's export sitting in a shared directory;
// listing every account as a discrepancy would be noise.
if (!rec.mapsAnyAccount()) {
try cli.printFg(out, color, cli.CLR_MUTED, " wells fargo: none of its {d} accounts are in accounts.srf; skipped (another portfolio's export?)\n", .{rec.results.len});
continue;
}
if (verbose or common.hasAccountDiscrepancies(rec.results)) {
try out.print("\n", .{});
try common.displayResults(rec.results, color, out);
try common.displayRatioSuggestions(allocator, rec.results, portfolio, prices, account_map, color, out);
} else {
try cli.printFg(out, color, cli.CLR_POSITIVE, " wells fargo: {d} accounts, no discrepancies\n", .{rec.results.len});
try common.displayRatioSuggestions(allocator, rec.results, portfolio, prices, account_map, color, out);
}
try accumulatePresent(allocator, &wells_fargo_present, common.AccountComparison, rec.results);
},
}
}
@ -1229,6 +1265,11 @@ pub fn runHygieneCheck(
defer allocator.free(absent);
try common.displayAbsentAccounts(absent, color, "any export", out);
}
if (wells_fargo_present.items.len > 0) {
const absent = try common.findAbsentAccounts(allocator, portfolio, account_map, wf_parser.institution, wells_fargo_present.items, prices, as_of);
defer allocator.free(absent);
try common.displayAbsentAccounts(absent, color, "any export", out);
}
}
// ── Section 6: Large new lots - confirm source ──

View file

@ -34,6 +34,17 @@
//! carries these.
//! - `security_type::cash` for cash-classified positions
//!
//! ## Wells Fargo: real lots
//!
//! Wells Fargo's "Portfolio-Expanded Detail" download lists every tax
//! lot with its trade date and cost, for every account in the
//! household, so a WF import writes one lot per WF lot instead of one
//! synthetic lot per position (`wells_fargo.parseLots`). Each lot's
//! `open_date` and `open_price` come from the export; only lots WF
//! shows without a date (`Intra-Day` funds) take the sentinel and fall
//! back to the prior lot as below. Hand-edited fields and the note are
//! inherited the same way for every source.
//!
//! ## Re-import merge
//!
//! When the target portfolio file already exists, `import` reads
@ -46,7 +57,9 @@
//! hand-edited field (`Lot.hand_edited_fields`: `ticker`,
//! `label`, `price`, `price_ratio`, `drip`, ...) from the prior
//! lot; only `shares` and `security_type` come from the new
//! export. A re-import of an unchanged held position produces
//! export. (A lot whose export row carries a real buy date -
//! Wells Fargo - keeps its own `open_date` and `open_price`.)
//! A re-import of an unchanged held position produces
//! byte-identical output, so `git diff` only surfaces actual
//! brokerage changes (lot-size drift, real cost-basis
//! adjustments).
@ -120,11 +133,13 @@
const std = @import("std");
const builtin = @import("builtin");
const biff8 = @import("biff8");
const zfin = @import("../root.zig");
const cli = @import("common.zig");
const framework = @import("framework.zig");
const Date = @import("../Date.zig");
const portfolio_mod = @import("../models/portfolio.zig");
const Lot = @import("../models/portfolio.zig").Lot;
const LotType = @import("../models/portfolio.zig").LotType;
const cache = @import("../cache/store.zig");
const atomic = @import("../atomic.zig");
const fidelity = @import("../brokerage/fidelity.zig");
@ -135,37 +150,24 @@ const analysis = @import("../analytics/analysis.zig");
const BrokeragePosition = brokerage_types.BrokeragePosition;
/// Source brokerage for the import. Fidelity and Schwab carry an
/// account number in the export rows themselves; Wells Fargo's
/// paste does not, so the WF variant carries an optional explicit
/// account-name override (`--account NAME`) that the resolver
/// uses when filename inference fails.
/// Source brokerage for the import, carrying the export's path.
/// Every source's export carries account numbers per row.
pub const Source = union(enum) {
fidelity: []const u8,
schwab: []const u8,
wells_fargo: WellsFargoArgs,
pub const WellsFargoArgs = struct {
path: []const u8,
/// `--account NAME` value, if the user supplied one.
/// Otherwise the resolver falls back to filename
/// inference and finally to a single-WF-entry lookup.
account_override: ?[]const u8,
};
wells_fargo: []const u8,
pub fn label(self: Source) []const u8 {
return switch (self) {
.fidelity => "fidelity",
.schwab => "schwab",
.wells_fargo => "wells_fargo",
.wells_fargo => wells_fargo.institution,
};
}
pub fn path(self: Source) []const u8 {
return switch (self) {
.fidelity => |p| p,
.schwab => |p| p,
.wells_fargo => |a| a.path,
inline else => |p| p,
};
}
};
@ -181,11 +183,13 @@ pub const meta: framework.Meta = .{
.group = .hygiene,
.synopsis = "Synthesize a portfolio file from a brokerage holdings export",
.help =
\\Usage: zfin -p <PORTFOLIO> import (--fidelity FILE | --schwab FILE | --wells-fargo FILE [--account NAME]) [-y]
\\Usage: zfin -p <PORTFOLIO> import (--fidelity FILE | --schwab FILE | --wells-fargo FILE) [-y]
\\
\\Synthesize a portfolio file from a brokerage positions export.
\\Each run REPLACES the target portfolio file with synthetic lots
\\drawn from the export - one lot per (account, symbol).
\\Each run REPLACES the target portfolio file with lots drawn
\\from the export: one synthetic lot per (account, symbol) for
\\Fidelity and Schwab, one lot per tax lot for Wells Fargo
\\(whose export carries each lot's trade date and cost).
\\
\\Designed for managed accounts (direct-indexing baskets, accounts
\\you don't track at lot granularity). Per-buy history is lost;
@ -200,10 +204,9 @@ pub const meta: framework.Meta = .{
\\`git diff` only flags genuine brokerage changes. Newly-introduced
\\positions get `open_date::1970-01-01` (a "we don't know"
\\sentinel; the next import will treat it as the prior anchor).
\\Wells Fargo lots keep the export's own date and cost.
\\Lots that disappear from the export are silently dropped - if
\\you sold a position between imports, it just stops appearing.
\\Only `shares` and `security_type` come from the export; every
\\hand-edited field on a prior lot is preserved.
\\
\\Required:
\\ -p, --portfolio <FILE> Target portfolio file (must be a single
@ -211,19 +214,14 @@ pub const meta: framework.Meta = .{
\\ --fidelity <CSV> Fidelity positions CSV
\\ ("All accounts" -> Positions tab -> Download)
\\ --schwab <CSV> Schwab per-account positions CSV
\\ --wells-fargo <FILE> Wells Fargo paste (copy the rendered
\\ positions table from the WF portal
\\ and save to a file). Pass `-` to
\\ --wells-fargo <XLS> Wells Fargo positions spreadsheet
\\ (Download Type "Portfolio-Expanded
\\ Detail", Portfolio View "Positions",
\\ all brokerage accounts). Covers every
\\ WF account in one file. Pass `-` to
\\ read from stdin.
\\
\\Options:
\\ --account <NAME> (Wells Fargo only) explicit account
\\ name to attribute the lots to. Must
\\ match an entry in accounts.srf. If
\\ omitted, the importer infers the
\\ account from the filename, then
\\ falls back to the single WF entry
\\ in accounts.srf.
\\ -y, --yes Don't prompt before overwriting an
\\ existing file.
\\
@ -246,8 +244,7 @@ pub const meta: framework.Meta = .{
CannotReadCsv,
CannotReadAccountsFile,
UnmappedAccount,
AmbiguousWellsFargoAccount,
UnknownAccount,
InvalidExport,
UserDeclined,
WriteFailed,
},
@ -257,7 +254,6 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr
var fidelity_path: ?[]const u8 = null;
var schwab_path: ?[]const u8 = null;
var wells_fargo_path: ?[]const u8 = null;
var account_override: ?[]const u8 = null;
var yes = false;
var i: usize = 0;
@ -269,8 +265,6 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr
schwab_path = try cli.requireFlagValue(ctx.io, cmd_args, &i, a);
} else if (std.mem.eql(u8, a, "--wells-fargo")) {
wells_fargo_path = try cli.requireFlagValue(ctx.io, cmd_args, &i, a);
} else if (std.mem.eql(u8, a, "--account")) {
account_override = try cli.requireFlagValue(ctx.io, cmd_args, &i, a);
} else if (std.mem.eql(u8, a, "-y") or std.mem.eql(u8, a, "--yes")) {
yes = true;
} else {
@ -292,20 +286,12 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr
return error.ConflictingSources;
}
// `--account` is a Wells-Fargo-only knob; the other sources
// carry per-row account_numbers in the export. Reject up
// front so the user notices early.
if (account_override != null and wells_fargo_path == null) {
cli.stderrPrint(ctx.io, "Error: --account is only meaningful with --wells-fargo (Fidelity/Schwab exports carry account numbers per row)\n");
return error.UnexpectedArg;
}
const source: Source = if (fidelity_path) |p|
.{ .fidelity = p }
else if (schwab_path) |p|
.{ .schwab = p }
else if (wells_fargo_path) |p|
.{ .wells_fargo = .{ .path = p, .account_override = account_override } }
.{ .wells_fargo = p }
else {
cli.stderrPrint(ctx.io, "Error: import requires a source flag (--fidelity FILE, --schwab FILE, or --wells-fargo FILE)\n");
return error.MissingSource;
@ -331,21 +317,32 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
// ── Read the brokerage export ─────────────────────────────
//
// `-` selects stdin (for the paste-from-clipboard workflow);
// any other path reads from disk. Same convention regardless
// of source.
const csv_path = parsed.source.path();
const csv_data = try readSourceData(io, allocator, csv_path);
defer allocator.free(csv_data);
// `-` selects stdin; any other path reads from disk. Same
// convention regardless of source.
const source_path = parsed.source.path();
const source_data = try readSourceData(io, allocator, source_path);
defer allocator.free(source_data);
// ── Parse ─────────────────────────────────────────────────
//
// The Wells Fargo export is a binary workbook; its positions
// borrow strings from the decoded sheet, so the workbook lives
// for the rest of the run.
var workbook: ?biff8.Workbook = null;
defer if (workbook) |*wb| wb.deinit();
var wf_sheet: ?*const biff8.Sheet = null;
const positions: []BrokeragePosition = switch (parsed.source) {
.fidelity => try fidelity.parseCsv(allocator, csv_data),
.fidelity => try fidelity.parseCsv(allocator, source_data),
.schwab => blk: {
const r = try schwab.parseCsv(allocator, csv_data);
const r = try schwab.parseCsv(allocator, source_data);
break :blk r.positions;
},
.wells_fargo => try wells_fargo.parsePaste(allocator, csv_data),
.wells_fargo => blk: {
workbook = biff8.Workbook.parse(allocator, source_data) catch |err| return badWellsFargoExport(io, source_path, err);
wf_sheet = wells_fargo.positionsSheet(&workbook.?) catch |err| return badWellsFargoExport(io, source_path, err);
break :blk wells_fargo.parsePositions(allocator, wf_sheet.?) catch |err| return badWellsFargoExport(io, source_path, err);
},
};
defer allocator.free(positions);
@ -369,21 +366,6 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
};
defer account_map.deinit();
// ── WF: resolve and patch the per-row account_number ──────
//
// Wells Fargo pastes don't carry an account identifier, so
// every position came back with `account_number = ""`. Defer
// to `wells_fargo.applyAccountToPositions` to resolve
// (explicit `--account` -> filename-inferred -> single-WF-entry
// fallback) and rewrite every position's
// account_number/account_name accordingly. The downstream
// `synthesizeLots` lookup then works uniformly across
// brokerages.
if (parsed.source == .wells_fargo) {
const wf_args = parsed.source.wells_fargo;
try wells_fargo.applyAccountToPositions(io, account_map, csv_path, wf_args.account_override, positions);
}
// ── Read the existing target file (if any) for merge ──────
//
// When the target file already exists, we treat its lots as
@ -419,13 +401,12 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
}
// ── Synthesize lots ───────────────────────────────────────
const lots = synthesizeLots(io, allocator, positions, account_map, parsed.source, ctx.today, prior_lookup_opt) catch |err| switch (err) {
error.UnmappedAccount => {
// synthesizeLots already printed the offending account
// numbers to stderr; just propagate as a user-level error.
return err;
},
else => |e| return e,
//
// `UnmappedAccount` arrives with the offending account numbers
// already printed to stderr.
const lots = switch (parsed.source) {
.wells_fargo => try wellsFargoLots(io, allocator, wf_sheet.?, source_path, positions, account_map, ctx.today, prior_lookup_opt),
else => try synthesizeLots(io, allocator, positions, account_map, parsed.source, ctx.today, prior_lookup_opt),
};
defer freeLots(allocator, lots);
@ -536,14 +517,27 @@ fn readSourceData(io: std.Io, allocator: std.mem.Allocator, path: []const u8) ![
};
return data;
}
return std.Io.Dir.cwd().readFileAlloc(io, path, allocator, .limited(10 * 1024 * 1024)) catch {
return std.Io.Dir.cwd().readFileAlloc(io, path, allocator, .limited(10 * 1024 * 1024)) catch |err| {
var msg_buf: [512]u8 = undefined;
const msg = std.fmt.bufPrint(&msg_buf, "Error: Cannot read source file: {s}\n", .{path}) catch "Error: Cannot read source file\n";
const msg = std.fmt.bufPrint(&msg_buf, "Error: Cannot read source file {s}: {s}\n", .{ path, @errorName(err) }) catch "Error: Cannot read source file\n";
cli.stderrPrint(io, msg);
return error.CannotReadCsv;
};
}
/// Explain a Wells Fargo export that could not be read, then return
/// the user-level error.
fn badWellsFargoExport(io: std.Io, path: []const u8, err: wells_fargo.ExportError) error{ InvalidExport, OutOfMemory } {
if (err == error.OutOfMemory) return error.OutOfMemory;
// Quiet under tests, like `requireMapped`; the error is what they check.
if (builtin.is_test) return error.InvalidExport;
var msg_buf: [512]u8 = undefined;
const msg = std.fmt.bufPrint(&msg_buf, "Error: Cannot read Wells Fargo export {s}: {s}\n", .{ path, @errorName(err) }) catch "Error: Cannot read Wells Fargo export\n";
cli.stderrPrint(io, msg);
cli.stderrPrint(io, wells_fargo.exportHint(err));
return error.InvalidExport;
}
/// Prompt for y/N confirmation on stderr, read one line from stdin.
/// Returns true only on a 'y' or 'Y' answer (with optional whitespace);
/// anything else (including EOF / empty) is "no".
@ -582,11 +576,11 @@ fn confirmOverwrite(io: std.Io, path: []const u8) !bool {
/// `Portfolio` and remain valid as long as the source portfolio
/// does.
const PriorLotsLookup = struct {
map: std.StringHashMap(*const portfolio_mod.Lot),
map: std.StringHashMap(*const Lot),
allocator: std.mem.Allocator,
fn init(allocator: std.mem.Allocator, lots: []const portfolio_mod.Lot, today: Date) !PriorLotsLookup {
var map = std.StringHashMap(*const portfolio_mod.Lot).init(allocator);
fn init(allocator: std.mem.Allocator, lots: []const Lot, today: Date) !PriorLotsLookup {
var map = std.StringHashMap(*const Lot).init(allocator);
errdefer {
var it = map.keyIterator();
while (it.next()) |k| allocator.free(k.*);
@ -633,7 +627,7 @@ const PriorLotsLookup = struct {
/// Find the prior `Lot` for `(symbol, account)`, returning
/// null when no match exists. Caller must keep the source
/// portfolio alive while using the returned pointer.
fn find(self: PriorLotsLookup, symbol: []const u8, account: []const u8) !?*const portfolio_mod.Lot {
fn find(self: PriorLotsLookup, symbol: []const u8, account: []const u8) !?*const Lot {
var key_buf: [256]u8 = undefined;
// Fast path: stack-allocate the key. Fall back to heap
// for unusually long symbols/accounts.
@ -654,28 +648,15 @@ const PriorLotsLookup = struct {
}
};
/// Synthesize a `Lot` per `BrokeragePosition`. Resolves each
/// brokerage account_number to a portfolio account name via
/// `account_map`; refuses (with a stderr listing of unmapped
/// numbers) if any can't be resolved.
/// Synthesize a `Lot` per `BrokeragePosition` (Fidelity, Schwab).
/// Resolves each brokerage account_number to a portfolio account
/// name via `account_map`; refuses (with a stderr listing of
/// unmapped numbers) if any can't be resolved.
///
/// `today` stamps the `note::` field on lots that don't have a
/// matching prior entry (`note::imported <broker> YYYY-MM-DD`).
/// Lots that DO match a prior entry inherit that prior lot's
/// note (which carries the original first-seen date), so a
/// re-import of an unchanged held position produces a
/// byte-identical line - the `git diff` only shows genuine
/// brokerage changes.
///
/// `prior_lookup` (if non-null) carries the lots from the
/// existing target file. For each synthesized lot, we look up
/// `(symbol, account)`:
/// - **Match:** preserve `open_date`, `open_price`, and `note`
/// from the prior lot. Brokerage shares/cost-basis still come
/// from the new export.
/// - **No match:** new position. Use `Date.epoch` for
/// `open_date` (no signal to do better) and stamp today's
/// date in the note.
/// Each position becomes a lot by `brokerage_types.lotFromPosition`
/// (cost-basis open price, cash at $1, `Date.epoch` open date), then
/// takes what the export cannot know from the prior file via
/// `inheritFromPrior`.
///
/// Takes `io` so it can print the unmapped-account-number
/// enumeration directly to stderr - easier than threading the list
@ -693,12 +674,85 @@ fn synthesizeLots(
source: Source,
today: Date,
prior_lookup: ?PriorLotsLookup,
) ![]portfolio_mod.Lot {
) ![]Lot {
const institution = source.label();
try requireMapped(io, allocator, positions, account_map, institution);
// First pass: collect any unmapped account numbers so we can
// report all of them at once instead of failing on the first
// and making the user re-run for each.
var lots = std.ArrayList(Lot).empty;
errdefer {
for (lots.items) |lot| freeLot(allocator, lot);
lots.deinit(allocator);
}
var fresh_note_buf: [64]u8 = undefined;
const fresh_note = try freshNote(&fresh_note_buf, institution, today);
for (positions) |pos| {
const acct_name = account_map.findByInstitutionAccount(institution, pos.account_number).?;
var lot = brokerage_types.lotFromPosition(pos, acct_name);
lot.symbol = try allocator.dupe(u8, lot.symbol);
lot.account = allocator.dupe(u8, acct_name) catch |err| {
allocator.free(lot.symbol);
return err;
};
errdefer freeLot(allocator, lot);
try inheritFromPrior(allocator, &lot, prior_lookup, fresh_note);
try lots.append(allocator, lot);
}
return lots.toOwnedSlice(allocator);
}
/// The Wells Fargo lots: real per-lot dates and costs from the
/// export (`wells_fargo.parseLots`), then the same prior-file
/// inheritance as every other source.
///
/// `positions` is the same export's `parsePositions` output, used
/// for the unmapped-account check so the message matches the other
/// sources'. Caller must call `freeLots`.
fn wellsFargoLots(
io: std.Io,
allocator: std.mem.Allocator,
sheet: *const biff8.Sheet,
source_path: []const u8,
positions: []const BrokeragePosition,
account_map: analysis.AccountMap,
today: Date,
prior_lookup: ?PriorLotsLookup,
) ![]Lot {
try requireMapped(io, allocator, positions, account_map, wells_fargo.institution);
const lots = wells_fargo.parseLots(allocator, sheet, account_map) catch |err| switch (err) {
// requireMapped checked these very account numbers.
error.UnmappedAccount => return err,
else => |e| return badWellsFargoExport(io, source_path, e),
};
errdefer freeLots(allocator, lots);
var fresh_note_buf: [64]u8 = undefined;
const fresh_note = try freshNote(&fresh_note_buf, wells_fargo.institution, today);
for (lots) |*lot| try inheritFromPrior(allocator, lot, prior_lookup, fresh_note);
return lots;
}
/// `note::` stamp for lots with no prior match. Held positions
/// inherit their prior note instead, so the `git diff` between two
/// imports of an unchanged held position stays empty.
fn freshNote(buf: *[64]u8, institution: []const u8, today: Date) ![]const u8 {
return std.fmt.bufPrint(buf, "imported {s} {f}", .{ institution, today });
}
/// Refuse the import if any position's account number has no
/// `accounts.srf` entry for `institution`, listing every unmapped
/// number at once rather than failing on the first and making the
/// user re-run for each.
fn requireMapped(
io: std.Io,
allocator: std.mem.Allocator,
positions: []const BrokeragePosition,
account_map: analysis.AccountMap,
institution: []const u8,
) !void {
var unmapped: std.ArrayList([]const u8) = .empty;
defer unmapped.deinit(allocator);
{
@ -713,133 +767,95 @@ fn synthesizeLots(
}
}
}
if (unmapped.items.len > 0) {
// Skip the human-readable error message under tests to keep
// test output clean; the returned error is what the test
// assertions rely on.
if (!builtin.is_test) {
var stderr_buf: [4096]u8 = undefined;
var stderr_writer = std.Io.File.stderr().writer(io, &stderr_buf);
try stderr_writer.interface.print(
"Error: {d} account number{s} from the {s} export {s} not mapped in accounts.srf:\n",
.{
unmapped.items.len,
if (unmapped.items.len == 1) "" else "s",
institution,
if (unmapped.items.len == 1) "is" else "are",
},
);
for (unmapped.items) |num| {
try stderr_writer.interface.print(" - {s}\n", .{num});
}
try stderr_writer.interface.print(
"\nAdd entries to accounts.srf with `institution::{s}` and the matching\n" ++
"`account_number::` value, then rerun the import.\n",
.{institution},
);
try stderr_writer.interface.flush();
if (unmapped.items.len == 0) return;
// Skip the human-readable error message under tests to keep
// test output clean; the returned error is what the test
// assertions rely on.
if (!builtin.is_test) {
var stderr_buf: [4096]u8 = undefined;
var stderr_writer = std.Io.File.stderr().writer(io, &stderr_buf);
try stderr_writer.interface.print(
"Error: {d} account number{s} from the {s} export {s} not mapped in accounts.srf:\n",
.{
unmapped.items.len,
if (unmapped.items.len == 1) "" else "s",
institution,
if (unmapped.items.len == 1) "is" else "are",
},
);
for (unmapped.items) |num| {
try stderr_writer.interface.print(" - {s}\n", .{num});
}
return error.UnmappedAccount;
try stderr_writer.interface.print(
"\nAdd entries to accounts.srf with `institution::{s}` and the matching\n" ++
"`account_number::` value, then rerun the import.\n",
.{institution},
);
try stderr_writer.interface.flush();
}
return error.UnmappedAccount;
}
var lots = std.ArrayList(portfolio_mod.Lot).empty;
errdefer {
for (lots.items) |lot| freeLot(allocator, lot);
lots.deinit(allocator);
/// Fill in what an export cannot know from the prior file's lot for
/// the same `(symbol, account)`:
///
/// - **Open date and price**, only when the export had no buy date
/// (`Date.epoch`, which is every Fidelity/Schwab position and a
/// Wells Fargo `Intra-Day` lot). A lot with a real date keeps it.
/// - **Note**: the prior lot's (carrying its original first-seen
/// date), else `fresh_note`.
/// - **Every hand-edited field** (`Lot.hand_edited_fields`). The
/// set is declared once on `Lot`, so a new hand-edited field is
/// preserved here automatically. None ever appears in an export,
/// so without this a re-import would drop the user's annotations.
///
/// Cash lots are never matched: they have no account-meaningful
/// identity, and the broker's balance is the truth every time.
///
/// Strings are duped against `allocator`; a field is only assigned
/// once its copy exists, so `freeLot` is always safe on `lot`.
fn inheritFromPrior(
allocator: std.mem.Allocator,
lot: *Lot,
prior_lookup: ?PriorLotsLookup,
fresh_note: []const u8,
) !void {
const prior: ?*const Lot = if (lot.security_type == .cash)
null
else if (prior_lookup) |pl|
try pl.find(lot.symbol, lot.account.?)
else
null;
lot.note = try allocator.dupe(u8, if (prior) |prev| prev.note orelse fresh_note else fresh_note);
const p = prior orelse return;
if (lot.open_date.eql(Date.epoch)) {
lot.open_date = p.open_date;
lot.open_price = p.open_price;
}
// Stamp for newly-introduced lots (no prior match). Held
// positions inherit their prior note instead, so the
// `git diff` between two imports of an unchanged held
// position stays empty.
var fresh_note_buf: [64]u8 = undefined;
const fresh_note = try std.fmt.bufPrint(&fresh_note_buf, "imported {s} {f}", .{ institution, today });
for (positions) |pos| {
const acct_name = account_map.findByInstitutionAccount(institution, pos.account_number).?;
const shares = pos.quantity orelse blk: {
// Cash positions have null quantity; the convention
// elsewhere in zfin (audit, snapshot's cash row) is
// shares=current_value, open_price=$1. Mirror that.
if (pos.is_cash) break :blk pos.current_value orelse 0;
// Non-cash without quantity is malformed export data.
// Treat as 0 shares so the lot is harmless; the user
// will see it in `git diff` and either fix the export
// or hand-edit the lot.
break :blk 0;
};
const synthesized_open_price: f64 = if (pos.is_cash)
1.0
else if (pos.cost_basis) |cb|
if (shares > 0) cb / shares else 0
else if (pos.current_value) |cv|
if (shares > 0) cv / shares else 0
else
0;
const security_type: portfolio_mod.LotType = if (pos.is_cash) .cash else .stock;
// Merge the prior lot's open_date / open_price / note if
// we have one. Cash lots are excluded from `prior_lookup`
// (they have no account-meaningful identity) and always
// get the synthesized values.
const prior: ?*const portfolio_mod.Lot = if (prior_lookup) |pl|
(if (security_type == .cash) null else try pl.find(pos.symbol, acct_name))
else
null;
const open_date: Date = if (prior) |p| p.open_date else Date.epoch;
const open_price: f64 = if (prior) |p| p.open_price else synthesized_open_price;
const note_text: []const u8 = if (prior) |p|
(if (p.note) |n| n else fresh_note)
else
fresh_note;
var new_lot = portfolio_mod.Lot{
.symbol = try allocator.dupe(u8, pos.symbol),
.shares = shares,
.open_date = open_date,
.open_price = open_price,
.account = try allocator.dupe(u8, acct_name),
.security_type = security_type,
.note = try allocator.dupe(u8, note_text),
};
// Carry every hand-edited field forward from the prior lot.
// The set is declared once on `Lot.hand_edited_fields`, so a
// new hand-edited field is preserved here automatically.
// String fields are duped into our allocator; value fields
// (numbers, bools, dates) copy directly. None of these are
// ever present in a brokerage export, so without this a
// re-import would silently drop the user's annotations.
if (prior) |p| {
inline for (portfolio_mod.Lot.hand_edited_fields) |fname| {
if (@TypeOf(@field(p, fname)) == ?[]const u8) {
@field(new_lot, fname) = if (@field(p, fname)) |s| try allocator.dupe(u8, s) else null;
} else {
@field(new_lot, fname) = @field(p, fname);
}
}
// String fields are duped into our allocator; value fields
// (numbers, bools, dates) copy directly.
inline for (Lot.hand_edited_fields) |fname| {
if (@TypeOf(@field(p, fname)) == ?[]const u8) {
@field(lot, fname) = if (@field(p, fname)) |s| try allocator.dupe(u8, s) else null;
} else {
@field(lot, fname) = @field(p, fname);
}
try lots.append(allocator, new_lot);
}
return lots.toOwnedSlice(allocator);
}
/// Free per-lot allocator-owned strings + the slice. Mirror of the
/// internal cleanup in `Portfolio.deinit` (which we'd use directly
/// except we don't construct a Portfolio here - `serializePortfolio`
/// takes a bare `[]const Lot`).
fn freeLots(allocator: std.mem.Allocator, lots: []const portfolio_mod.Lot) void {
fn freeLots(allocator: std.mem.Allocator, lots: []const Lot) void {
for (lots) |lot| freeLot(allocator, lot);
allocator.free(lots);
}
fn freeLot(allocator: std.mem.Allocator, lot: portfolio_mod.Lot) void {
fn freeLot(allocator: std.mem.Allocator, lot: Lot) void {
allocator.free(lot.symbol);
if (lot.note) |n| allocator.free(n);
if (lot.label) |l| allocator.free(l);
@ -898,7 +914,7 @@ test "synthesizeLots: stock positions get open_price = cost_basis / quantity" {
// `open_date` is the sentinel and the note carries the
// import date so the user can tell when it was first seen.
try testing.expectEqual(Date.epoch.days, lots[0].open_date.days);
try testing.expectEqual(portfolio_mod.LotType.stock, lots[0].security_type);
try testing.expectEqual(LotType.stock, lots[0].security_type);
try testing.expect(lots[0].note != null);
try testing.expect(std.mem.indexOf(u8, lots[0].note.?, "imported fidelity") != null);
try testing.expect(std.mem.indexOf(u8, lots[0].note.?, "2026-05-21") != null);
@ -957,7 +973,7 @@ test "synthesizeLots: cash positions become security_type=cash with shares=value
defer freeLots(allocator, lots);
try testing.expectEqual(@as(usize, 1), lots.len);
try testing.expectEqual(portfolio_mod.LotType.cash, lots[0].security_type);
try testing.expectEqual(LotType.cash, lots[0].security_type);
try testing.expectApproxEqAbs(@as(f64, 5000), lots[0].shares, 0.01);
try testing.expectApproxEqAbs(@as(f64, 1.0), lots[0].open_price, 0.01);
}
@ -985,7 +1001,7 @@ test "synthesizeLots: lots are byte-identical across imports when prior_lookup m
// by (symbol, account). The merge path should preserve every
// field of this lot (open_date, open_price, note) regardless
// of the import date.
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "AAPL",
.shares = 1,
@ -1126,7 +1142,7 @@ test "synthesizeLots: prior lot for (symbol, account) preserves open_date and op
});
defer account_map.deinit();
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "AAPL",
.shares = 100,
@ -1173,7 +1189,7 @@ test "synthesizeLots: every hand-edited field is preserved on re-import" {
});
defer account_map.deinit();
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "02315N600",
.shares = 100,
@ -1250,7 +1266,7 @@ test "synthesizeLots: new position with no prior match gets sentinel + today's n
// Prior has only AAPL; new export has both AAPL and a new
// GOOG.
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "AAPL",
.shares = 100,
@ -1296,7 +1312,7 @@ test "synthesizeLots: when prior has multiple lots for same (symbol, account), e
});
defer account_map.deinit();
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "AAPL",
.shares = 50,
@ -1350,7 +1366,7 @@ test "synthesizeLots: prior closed lot does NOT anchor a held position" {
});
defer account_map.deinit();
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "AAPL",
.shares = 100,
@ -1391,7 +1407,7 @@ test "synthesizeLots: positions dropped from new export are excluded (closed-lot
});
defer account_map.deinit();
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "AAPL",
.shares = 100,
@ -1432,7 +1448,7 @@ test "PriorLotsLookup: cash lots are excluded from the lookup" {
// every time. Pin that they don't enter the lookup so we
// don't accidentally inherit a stale cash open_price/note.
const allocator = testing.allocator;
const prior_lots = [_]portfolio_mod.Lot{
const prior_lots = [_]Lot{
.{
.symbol = "FZFXX",
.shares = 1000,
@ -1521,20 +1537,200 @@ test "parseArgs: --wells-fargo accepts the lone '-' stdin sentinel" {
ctx.io = testing.io;
const args = [_][]const u8{ "--wells-fargo", "-" };
const parsed = try parseArgs(&ctx, &args);
switch (parsed.source) {
.wells_fargo => |wf| try testing.expectEqualStrings("-", wf.path),
else => try testing.expect(false),
}
try testing.expectEqualStrings("-", parsed.source.wells_fargo);
}
test "parseArgs: --account is gone (the WF export names every account)" {
var ctx: framework.RunCtx = undefined;
ctx.io = testing.io;
const args = [_][]const u8{ "--wells-fargo", "wf.xls", "--account", "Sample IRA" };
try testing.expectError(error.UnexpectedArg, parseArgs(&ctx, &args));
}
test "parseArgs: --wells-fargo conflicts with another source" {
var ctx: framework.RunCtx = undefined;
ctx.io = testing.io;
const args = [_][]const u8{ "--wells-fargo", "wf.xls", "--fidelity", "f.csv" };
try testing.expectError(error.ConflictingSources, parseArgs(&ctx, &args));
}
test "Source.label: returns broker name" {
try testing.expectEqualStrings("fidelity", (Source{ .fidelity = "" }).label());
try testing.expectEqualStrings("schwab", (Source{ .schwab = "" }).label());
try testing.expectEqualStrings("wells_fargo", (Source{ .wells_fargo = .{ .path = "", .account_override = null } }).label());
try testing.expectEqualStrings("wells_fargo", (Source{ .wells_fargo = "" }).label());
}
test "Source.path: returns the CSV path" {
test "Source.path: returns the export path" {
try testing.expectEqualStrings("/x.csv", (Source{ .fidelity = "/x.csv" }).path());
try testing.expectEqualStrings("/y.csv", (Source{ .schwab = "/y.csv" }).path());
try testing.expectEqualStrings("/z.txt", (Source{ .wells_fargo = .{ .path = "/z.txt", .account_override = null } }).path());
try testing.expectEqualStrings("/z.xls", (Source{ .wells_fargo = "/z.xls" }).path());
}
/// An owned lot, as `wells_fargo.parseLots` returns them.
fn ownedLot(allocator: std.mem.Allocator, lot: Lot) !Lot {
var owned = lot;
owned.symbol = try allocator.dupe(u8, lot.symbol);
owned.account = try allocator.dupe(u8, lot.account.?);
return owned;
}
test "inheritFromPrior: a lot with a real buy date keeps it, but takes the prior note and annotations" {
const allocator = testing.allocator;
const prior_lots = [_]Lot{.{
.symbol = "VTI",
.shares = 10,
.open_date = Date.fromYmd(2020, 1, 1),
.open_price = 150,
.account = "Sample IRA",
.note = "imported wells_fargo 2026-01-01",
.label = "Total Market",
.price_ratio = 2.0,
}};
var prior = try PriorLotsLookup.init(allocator, &prior_lots, Date.fromYmd(2026, 10, 3));
defer prior.deinit();
var lot = try ownedLot(allocator, .{ .symbol = "VTI", .shares = 8, .open_date = Date.fromYmd(2026, 9, 9), .open_price = 262.5, .account = "Sample IRA" });
defer freeLot(allocator, lot);
try inheritFromPrior(allocator, &lot, prior, "imported wells_fargo 2026-10-03");
try testing.expect(Date.fromYmd(2026, 9, 9).eql(lot.open_date));
try testing.expectEqual(@as(f64, 262.5), lot.open_price);
try testing.expectEqualStrings("imported wells_fargo 2026-01-01", lot.note.?);
try testing.expectEqualStrings("Total Market", lot.label.?);
try testing.expectEqual(@as(f64, 2.0), lot.price_ratio);
}
test "inheritFromPrior: an undated lot takes the prior date and price" {
const allocator = testing.allocator;
const prior_lots = [_]Lot{.{ .symbol = "SMPLX", .shares = 250, .open_date = Date.fromYmd(2019, 5, 5), .open_price = 12, .account = "Sample IRA" }};
var prior = try PriorLotsLookup.init(allocator, &prior_lots, Date.fromYmd(2026, 10, 3));
defer prior.deinit();
var lot = try ownedLot(allocator, .{ .symbol = "SMPLX", .shares = 250, .open_date = Date.epoch, .open_price = 20, .account = "Sample IRA" });
defer freeLot(allocator, lot);
try inheritFromPrior(allocator, &lot, prior, "imported wells_fargo 2026-10-03");
try testing.expect(Date.fromYmd(2019, 5, 5).eql(lot.open_date));
try testing.expectEqual(@as(f64, 12), lot.open_price);
// The prior lot had no note, so this one gets today's stamp.
try testing.expectEqualStrings("imported wells_fargo 2026-10-03", lot.note.?);
}
test "inheritFromPrior: cash never inherits" {
const allocator = testing.allocator;
const prior_lots = [_]Lot{.{ .symbol = "", .shares = 5, .open_date = Date.fromYmd(2019, 5, 5), .open_price = 1, .account = "Sample IRA", .security_type = .cash, .note = "old" }};
var prior = try PriorLotsLookup.init(allocator, &prior_lots, Date.fromYmd(2026, 10, 3));
defer prior.deinit();
var lot = try ownedLot(allocator, .{ .symbol = "", .shares = 7, .open_date = Date.epoch, .open_price = 1, .account = "Sample IRA", .security_type = .cash });
defer freeLot(allocator, lot);
try inheritFromPrior(allocator, &lot, prior, "fresh");
try testing.expect(Date.epoch.eql(lot.open_date));
try testing.expectEqualStrings("fresh", lot.note.?);
}
test "synthesizeLots: every allocation failure is clean" {
const S = struct {
fn run(allocator: std.mem.Allocator, prior: PriorLotsLookup) !void {
var entries = [_]analysis.AccountTaxEntry{
.{ .account = "Sample Brokerage", .tax_type = .taxable, .institution = "fidelity", .account_number = "Z123" },
};
const account_map: analysis.AccountMap = .{ .entries = &entries, .allocator = allocator };
const positions = [_]BrokeragePosition{
.{ .account_number = "Z123", .account_name = "I", .symbol = "AAPL", .description = "", .quantity = 10, .current_value = 1500, .cost_basis = 1500, .is_cash = false },
.{ .account_number = "Z123", .account_name = "I", .symbol = "02315N600", .description = "", .quantity = 5, .current_value = 500, .cost_basis = 400, .is_cash = false },
};
freeLots(allocator, try synthesizeLots(testing.io, allocator, &positions, account_map, .{ .fidelity = "" }, Date.fromYmd(2026, 5, 21), prior));
}
};
const prior_lots = [_]Lot{.{ .symbol = "02315N600", .shares = 5, .open_date = Date.fromYmd(2024, 6, 1), .open_price = 90, .account = "Sample Brokerage", .note = "n", .ticker = "VTTHX", .label = "TGT2035" }};
var prior = try PriorLotsLookup.init(testing.allocator, &prior_lots, Date.fromYmd(2026, 5, 21));
defer prior.deinit();
try testing.checkAllAllocationFailures(testing.allocator, S.run, .{prior});
}
// ---- Wells Fargo lots ----
fn wfText(s: []const u8) biff8.Cell {
return .{ .text = s };
}
fn wfNum(v: f64) biff8.Cell {
return .{ .number = v };
}
const wf_header = [_]biff8.Cell{ wfText("Description"), wfText("Symbol"), wfText("Account Number"), wfText("Market Value"), wfText("Shares"), wfText("Total Cost"), wfText("Trade Date") };
/// Placeholder export: two VTI lots and an Intra-Day fund in *1234.
const wf_rows = [_][]const biff8.Cell{
&wf_header,
&.{ wfText("VANGUARD TOTAL STOCK MKT"), wfText("VTI"), wfText("*1234"), wfNum(6000), wfNum(20), wfNum(4500), wfText("Detail") },
&.{ wfText("VANGUARD TOTAL STOCK MKT"), wfText("VTI"), wfText("*1234"), wfNum(3600), wfNum(12), wfNum(2400), wfText("02/03/2022") },
&.{ wfText("VANGUARD TOTAL STOCK MKT"), wfText("VTI"), wfText("*1234"), wfNum(2400), wfNum(8), wfNum(2100), wfText("09/09/2026") },
&.{ wfText("SAMPLE GROWTH FUND"), wfText("SMPLX"), wfText("*1234"), wfNum(5000), wfNum(250), wfText("N/A"), wfText("Intra-Day") },
};
const wf_sheet_fixture: biff8.Sheet = .{ .name = wells_fargo.sheet_name, .rows = &wf_rows };
test "wellsFargoLots: real lots, with prior notes, annotations, and undated fallbacks" {
const allocator = testing.allocator;
var entries = [_]analysis.AccountTaxEntry{
.{ .account = "Sample IRA", .tax_type = .traditional, .institution = "wells_fargo", .account_number = "1234" },
};
const account_map: analysis.AccountMap = .{ .entries = &entries, .allocator = allocator };
const prior_lots = [_]Lot{
.{ .symbol = "VTI", .shares = 20, .open_date = Date.fromYmd(2022, 2, 3), .open_price = 200, .account = "Sample IRA", .note = "imported wells_fargo 2026-01-01", .label = "Total Market" },
.{ .symbol = "SMPLX", .shares = 250, .open_date = Date.fromYmd(2018, 8, 8), .open_price = 11, .account = "Sample IRA" },
};
var prior = try PriorLotsLookup.init(allocator, &prior_lots, Date.fromYmd(2026, 10, 3));
defer prior.deinit();
const positions = try wells_fargo.parsePositions(allocator, &wf_sheet_fixture);
defer allocator.free(positions);
const lots = try wellsFargoLots(testing.io, allocator, &wf_sheet_fixture, "wf.xls", positions, account_map, Date.fromYmd(2026, 10, 3), prior);
defer freeLots(allocator, lots);
try testing.expectEqual(@as(usize, 3), lots.len);
// Dated lots keep the export's own date and cost per share...
try testing.expect(Date.fromYmd(2026, 9, 9).eql(lots[1].open_date));
try testing.expectEqual(@as(f64, 262.5), lots[1].open_price);
// ...and inherit the position's note and annotations.
try testing.expectEqualStrings("imported wells_fargo 2026-01-01", lots[1].note.?);
try testing.expectEqualStrings("Total Market", lots[1].label.?);
// The Intra-Day fund falls back to the prior lot's date and price.
try testing.expectEqualStrings("SMPLX", lots[2].symbol);
try testing.expect(Date.fromYmd(2018, 8, 8).eql(lots[2].open_date));
try testing.expectEqual(@as(f64, 11), lots[2].open_price);
try testing.expectEqualStrings("imported wells_fargo 2026-10-03", lots[2].note.?);
}
test "wellsFargoLots: an unmapped account is refused before any lot is built" {
const allocator = testing.allocator;
const account_map: analysis.AccountMap = .{ .entries = &.{}, .allocator = allocator };
const positions = try wells_fargo.parsePositions(allocator, &wf_sheet_fixture);
defer allocator.free(positions);
try testing.expectError(error.UnmappedAccount, wellsFargoLots(testing.io, allocator, &wf_sheet_fixture, "wf.xls", positions, account_map, Date.fromYmd(2026, 10, 3), null));
}
test "wellsFargoLots: a lot layout parsePositions accepts but parseLots rejects is InvalidExport" {
// parsePositions does not read trade dates; parseLots does. A date
// it cannot read is a wrong-download problem, reported as such.
const allocator = testing.allocator;
const rows = [_][]const biff8.Cell{
&wf_header,
&.{ wfText("VANGUARD TOTAL STOCK MKT"), wfText("VTI"), wfText("*1234"), wfNum(3600), wfNum(12), wfNum(2400), wfText("Multiple") },
};
const sheet: biff8.Sheet = .{ .name = wells_fargo.sheet_name, .rows = &rows };
var entries = [_]analysis.AccountTaxEntry{
.{ .account = "Sample IRA", .tax_type = .traditional, .institution = "wells_fargo", .account_number = "1234" },
};
const account_map: analysis.AccountMap = .{ .entries = &entries, .allocator = allocator };
const positions = try wells_fargo.parsePositions(allocator, &sheet);
defer allocator.free(positions);
try testing.expectError(error.InvalidExport, wellsFargoLots(testing.io, allocator, &sheet, "wf.xls", positions, account_map, Date.fromYmd(2026, 10, 3), null));
}
test "badWellsFargoExport: allocation failure stays an allocation failure" {
try testing.expectError(error.OutOfMemory, @as(error{ InvalidExport, OutOfMemory }!void, badWellsFargoExport(testing.io, "wf.xls", error.OutOfMemory)));
try testing.expectError(error.InvalidExport, @as(error{ InvalidExport, OutOfMemory }!void, badWellsFargoExport(testing.io, "wf.xls", error.NotCompoundFile)));
}

View file

@ -212,7 +212,7 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
);
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));
try svc.loadAllDividends(div_syms.items, cli.fetchOptionsFromPolicy(ctx.globals.refresh_policy));
}
for (pf_data.summary.allocations) |a| {
if (svc.getCachedDividends(allocator, a.symbol)) |divs| {

View file

@ -75,28 +75,27 @@ pub fn tryAcquire(self: *RateLimiter) bool {
}
/// Acquire a token, blocking (sleeping) until one is available.
pub fn acquire(self: *RateLimiter) void {
///
/// Returns `error.Canceled` when the calling task is canceled while
/// waiting. Propagate it: std delivers a cancelation to exactly one
/// cancelation point, so a caller that swallows it (this function used
/// to loop back on an interrupted sleep) runs on uncancelable, and on a
/// 4-per-minute bucket that turns a cancel into minutes of blocking.
pub fn acquire(self: *RateLimiter) std.Io.Cancelable!void {
while (!self.tryAcquire()) {
// Sleep for the time needed to generate 1 token. An
// interrupted sleep (cancelation propagating through the
// Io) just loops back to tryAcquire - the next refill
// covers whatever fraction of the wait elapsed.
// Sleep for the time needed to generate 1 token.
const wait_ns: u64 = @intFromFloat(1.0 / self.refill_rate_per_ns);
std.Io.sleep(self.io, .{ .nanoseconds = @intCast(wait_ns) }, .awake) catch |err| {
std.log.scoped(.rate_limiter).debug("acquire sleep interrupted: {t}", .{err});
};
try std.Io.sleep(self.io, .{ .nanoseconds = @intCast(wait_ns) }, .awake);
}
}
/// Sleep until a token is likely available, with a minimum 2-second floor.
/// Use after receiving a server-side 429 to wait before retrying.
pub fn backoff(self: *RateLimiter) void {
/// Returns `error.Canceled` if the task is canceled during the wait; see
/// `acquire`.
pub fn backoff(self: *RateLimiter) std.Io.Cancelable!void {
const wait_ns: u64 = @max(self.estimateWaitNs(), 2 * std.time.ns_per_s);
// Interrupted backoff sleep degrades to a shorter wait; the
// caller's retry may hit 429 again and re-backoff.
std.Io.sleep(self.io, .{ .nanoseconds = @intCast(wait_ns) }, .awake) catch |err| {
std.log.scoped(.rate_limiter).debug("backoff sleep interrupted: {t}", .{err});
};
try std.Io.sleep(self.io, .{ .nanoseconds = @intCast(wait_ns) }, .awake);
}
/// Returns estimated wait time in nanoseconds until a token is available.
@ -153,3 +152,26 @@ test "rate limiter exhaustion" {
// Bucket should be empty now
try std.testing.expect(!rl.tryAcquire());
}
test "acquire: a cancel while waiting for a token returns Canceled" {
// Used to loop back on the interrupted sleep, which made the cancel
// a no-op: on a 1-per-minute bucket the waiter then blocked for the
// full refill.
const io = std.testing.io;
var rl = RateLimiter.perMinute(io, 1);
try rl.acquire(); // spend the only token
var waiter = try io.concurrent(RateLimiter.acquire, .{&rl});
try io.sleep(.fromMilliseconds(20), .awake);
try std.testing.expectError(error.Canceled, waiter.cancel(io));
}
test "backoff: a cancel during the wait returns Canceled" {
const io = std.testing.io;
var rl = RateLimiter.perMinute(io, 1);
try rl.acquire();
var waiter = try io.concurrent(RateLimiter.backoff, .{&rl});
try io.sleep(.fromMilliseconds(20), .awake);
try std.testing.expectError(error.Canceled, waiter.cancel(io));
}

View file

@ -186,19 +186,27 @@ pub const Client = struct {
// "RequestFailed" because the per-attempt error was discarded
// by `catch {}`. Now the caller's `@errorName(err)` reports
// the real cause (e.g., `NoAddressReturned`).
//
// `error.Canceled` is never retried. It is the calling task
// being stopped, not a transport failure, and std delivers a
// cancelation to exactly one cancelation point: a retry would
// run uncancelable and could block indefinitely on a server
// that never answers. That is how quitting the TUI used to
// wait out a whole dividend warm-up.
var attempt: u8 = 0;
var last_err: HttpError = HttpError.RequestFailed;
while (true) : (attempt += 1) {
const response = self.doRequest(method, url, body, extra_headers) catch |err| {
if (err == error.Canceled) return err;
last_err = err;
if (attempt >= self.max_retries) return last_err;
self.backoffSleep(attempt);
try self.backoffSleep(attempt);
continue;
};
return classifyResponse(response) catch |err| {
if (err == HttpError.ServerError and attempt < self.max_retries) {
last_err = err;
self.backoffSleep(attempt);
try self.backoffSleep(attempt);
continue;
}
return err;
@ -206,9 +214,9 @@ pub const Client = struct {
}
}
fn backoffSleep(self: *Client, attempt: u8) void {
fn backoffSleep(self: *Client, attempt: u8) std.Io.Cancelable!void {
const backoff = self.base_backoff_ms * std.math.shl(u64, 1, attempt);
std.Io.sleep(self.io, std.Io.Duration.fromMilliseconds(@intCast(backoff)), .awake) catch |err| std.log.debug("backoff sleep interrupted: {t}", .{err});
try std.Io.sleep(self.io, std.Io.Duration.fromMilliseconds(@intCast(backoff)), .awake);
}
fn doRequest(self: *Client, method: std.http.Method, url: []const u8, body: ?[]const u8, extra_headers: []const std.http.Header) HttpError!Response {
@ -240,14 +248,6 @@ pub const Client = struct {
// doesn't produce nonsense elapsed values.
const t_start = std.Io.Timestamp.now(self.io, .awake).nanoseconds;
var t_stage = t_start;
const stageElapsedMs = struct {
fn f(prev: *i96, io: std.Io) i64 {
const now = std.Io.Timestamp.now(io, .awake).nanoseconds;
const delta_ns = now - prev.*;
prev.* = now;
return @intCast(@divTrunc(delta_ns, std.time.ns_per_ms));
}
}.f;
const uri = std.Uri.parse(url) catch |err| {
log.warn("http {s}: stage=uri_parse err={s} url={f}", .{ @tagName(method), @errorName(err), redactUrl(url) });
@ -284,8 +284,7 @@ pub const Client = struct {
// TLS handshake. Logging at warn level (rather than debug)
// because DNS / connectivity failures are exactly what
// operators need to see immediately.
log.warn("http {s}: stage=connect err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
return self.stageFailed(method, "connect", err, null, &t_stage, url);
};
defer req.deinit();
const ms_connect = stageElapsedMs(&t_stage, self.io);
@ -293,36 +292,18 @@ pub const Client = struct {
if (body) |payload| {
var send_buf: [4096]u8 = undefined;
req.transfer_encoding = .{ .content_length = payload.len };
var bw = req.sendBodyUnflushed(&send_buf) catch |err| {
log.warn("http {s}: stage=send_body_open err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
};
bw.writer.writeAll(payload) catch |err| {
log.warn("http {s}: stage=send_body_write err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
};
bw.end() catch |err| {
log.warn("http {s}: stage=send_body_end err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
};
req.connection.?.flush() catch |err| {
log.warn("http {s}: stage=send_body_flush err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
};
var bw = req.sendBodyUnflushed(&send_buf) catch |err| return self.stageFailed(method, "send_body_open", err, &req, &t_stage, url);
bw.writer.writeAll(payload) catch |err| return self.stageFailed(method, "send_body_write", err, &req, &t_stage, url);
bw.end() catch |err| return self.stageFailed(method, "send_body_end", err, &req, &t_stage, url);
req.connection.?.flush() catch |err| return self.stageFailed(method, "send_body_flush", err, &req, &t_stage, url);
} else {
req.sendBodiless() catch |err| {
log.warn("http {s}: stage=send_bodiless err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
};
req.sendBodiless() catch |err| return self.stageFailed(method, "send_bodiless", err, &req, &t_stage, url);
}
const ms_send = stageElapsedMs(&t_stage, self.io);
// Matches the default redirect capacity in std.http.Client.fetch.
var redirect_buffer: [8 * 1024]u8 = undefined;
var response = req.receiveHead(&redirect_buffer) catch |err| {
log.warn("http {s}: stage=receive_head err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
};
var response = req.receiveHead(&redirect_buffer) catch |err| return self.stageFailed(method, "receive_head", err, &req, &t_stage, url);
const ms_receive_head = stageElapsedMs(&t_stage, self.io);
// Capture the ETag (if any) from the response head BEFORE
@ -358,10 +339,7 @@ pub const Client = struct {
var decompress: std.http.Decompress = undefined;
var decompress_buffer: [64 * 1024]u8 = undefined;
const reader = response.readerDecompressing(&transfer_buffer, &decompress, &decompress_buffer);
_ = reader.streamRemaining(&aw.writer) catch |err| {
log.warn("http {s}: stage=stream_body err={s} elapsed_ms={d} url={f}", .{ @tagName(method), @errorName(err), stageElapsedMs(&t_stage, self.io), redactUrl(url) });
return err;
};
_ = reader.streamRemaining(&aw.writer) catch |err| return self.stageFailed(method, "stream_body", err, &req, &t_stage, url);
const ms_body = stageElapsedMs(&t_stage, self.io);
const resp_body = try aw.toOwnedSlice();
@ -391,6 +369,40 @@ pub const Client = struct {
};
}
/// Report a failed request stage: log it and return the error the
/// caller should see.
///
/// std's reader and writer interfaces turn every transport failure
/// into a generic `ReadFailed` / `WriteFailed` and leave the real
/// cause on the connection's socket stream. When that cause is
/// `error.Canceled`, return it instead: callers must be able to tell
/// "the task was stopped" from "the network failed", and the retry
/// loop in `request` must not retry the former (see there).
///
/// A cancel is logged at debug level. It is the expected result of
/// quitting mid-request, not something an operator needs to see.
/// Other failures log at warn, except under `zig build test`, where
/// the transport tests drive them on purpose (as `classifyResponse`
/// does for statuses).
fn stageFailed(
self: *Client,
method: std.http.Method,
stage: []const u8,
raw: HttpError,
req: ?*const std.http.Client.Request,
t_stage: *i96,
url: []const u8,
) HttpError {
const err = if (req) |r| unmaskCancel(raw, r) else raw;
const elapsed_ms = stageElapsedMs(t_stage, self.io);
if (err == error.Canceled) {
log.debug("http {s}: stage={s} canceled elapsed_ms={d} url={f}", .{ @tagName(method), stage, elapsed_ms, redactUrl(url) });
} else if (!builtin.is_test) {
log.warn("http {s}: stage={s} err={s} elapsed_ms={d} url={f}", .{ @tagName(method), stage, @errorName(err), elapsed_ms, redactUrl(url) });
}
return err;
}
fn classifyResponse(response: Response) HttpError!Response {
switch (response.status) {
.ok => return response,
@ -443,6 +455,37 @@ pub const Client = struct {
}
};
/// Milliseconds since `prev`, advancing `prev` to now. Per-stage
/// request timing; `.awake` (monotonic) so a clock jump mid-request
/// cannot produce a negative stage.
fn stageElapsedMs(prev: *i96, io: std.Io) i64 {
const now = std.Io.Timestamp.now(io, .awake).nanoseconds;
const delta_ns = now - prev.*;
prev.* = now;
return @intCast(@divTrunc(delta_ns, std.time.ns_per_ms));
}
/// `error.Canceled` when a generic `ReadFailed` / `WriteFailed` was a
/// cancelation of the request's socket I/O, else `err` unchanged. See
/// `Client.stageFailed`.
///
/// Reads the socket streams' recorded errors directly rather than
/// through `Connection.getReadError`, which asserts one is set.
fn unmaskCancel(err: HttpError, req: *const std.http.Client.Request) HttpError {
switch (err) {
error.ReadFailed, error.WriteFailed => {},
else => return err,
}
const conn = req.connection orelse return err;
if (conn.stream_reader.err) |e| {
if (e == error.Canceled) return error.Canceled;
}
if (conn.stream_writer.err) |e| {
if (e == error.Canceled) return error.Canceled;
}
return err;
}
// ── URL redaction for logs ───────────────────────────────────
//
// Providers authenticate by query parameter, so a raw request URL is a
@ -849,3 +892,82 @@ test "isCredentialParam: case-insensitive across provider spellings" {
try std.testing.expect(!isCredentialParam("ticker"));
try std.testing.expect(!isCredentialParam("startDate"));
}
// ---- Cancelation ----
const SilentServer = @import("../testutil/silent_server.zig").SilentServer;
fn silentUrl(server: *const SilentServer, buf: []u8) ![]const u8 {
var base_buf: [64]u8 = undefined;
return std.fmt.bufPrint(buf, "{s}/SAMPLE/dividends", .{try server.baseUrl(&base_buf)});
}
test "request: canceling a request blocked on an unresponsive server returns Canceled, with no retry" {
// The bug this pins: std reports a canceled read as a generic
// `ReadFailed`, the retry loop retried it, and the retry ran
// uncancelable (std delivers a cancel only once) - so canceling a
// request to a server that never answers blocked forever.
const io = std.testing.io;
var server: SilentServer = undefined;
try server.start(io, .hold_open);
defer server.stop();
var client = Client.init(io, std.testing.allocator);
defer client.deinit();
var url_buf: [96]u8 = undefined;
const url = try silentUrl(&server, &url_buf);
var pending = try io.concurrent(Client.get, .{ &client, url });
try server.waitForConnections(1, 5000);
try std.testing.expectError(error.Canceled, pending.cancel(io));
// A retry would have connected again within its first backoff.
try io.sleep(.fromMilliseconds(base_backoff_for_tests_ms * 4), .awake);
try std.testing.expectEqual(@as(u32, 1), server.accepted.load(.acquire));
}
test "request: a server that hangs up is still retried" {
// The cancel check must not swallow real transport failures: a
// closed connection is retried up to `max_retries` times.
const io = std.testing.io;
var server: SilentServer = undefined;
try server.start(io, .hang_up);
defer server.stop();
var client = Client.init(io, std.testing.allocator);
defer client.deinit();
client.base_backoff_ms = base_backoff_for_tests_ms;
var url_buf: [96]u8 = undefined;
const url = try silentUrl(&server, &url_buf);
if (client.get(url)) |resp| {
var r = resp;
r.deinit();
return error.TestUnexpectedResult;
} else |err| {
try std.testing.expect(err != error.Canceled);
}
try std.testing.expectEqual(@as(u32, 1 + client.max_retries), server.accepted.load(.acquire));
}
test "request: a cancel during the retry backoff returns Canceled" {
const io = std.testing.io;
var server: SilentServer = undefined;
try server.start(io, .hang_up);
defer server.stop();
var client = Client.init(io, std.testing.allocator);
defer client.deinit();
// Long enough that the cancel lands inside the first backoff sleep.
client.base_backoff_ms = 60_000;
var url_buf: [96]u8 = undefined;
const url = try silentUrl(&server, &url_buf);
var pending = try io.concurrent(Client.get, .{ &client, url });
try server.waitForConnections(1, 5000);
try std.testing.expectError(error.Canceled, pending.cancel(io));
try std.testing.expectEqual(@as(u32, 1), server.accepted.load(.acquire));
}
/// Backoff for tests that exercise retries, so they take milliseconds.
const base_backoff_for_tests_ms = 5;

View file

@ -202,7 +202,7 @@ pub fn deinit(self: *Edgar) void {
/// requires on every request and acquires a rate-limit token before
/// issuing the call.
fn httpGet(self: *Edgar, url: []const u8) !http.Response {
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
var ua_buf: [256]u8 = undefined;
const ua = std.fmt.bufPrint(&ua_buf, "zfin/0.1 ({s})", .{self.user_email}) catch return error.UserEmailTooLong;

View file

@ -41,7 +41,7 @@ pub const Cboe = struct {
allocator: std.mem.Allocator,
symbol: []const u8,
) ![]OptionsChain {
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
// Build URL: {base_url}/{SYMBOL}.json
const url = try buildCboeUrl(allocator, symbol);

View file

@ -65,7 +65,7 @@ pub const Fmp = struct {
allocator: std.mem.Allocator,
symbol: []const u8,
) ![]EarningsEvent {
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
const url = try http.buildUrl(allocator, base_url ++ "/earnings", &.{
.{ "symbol", symbol },

View file

@ -56,7 +56,7 @@ pub const Polygon = struct {
// First request
{
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
var params: [5][2][]const u8 = undefined;
var n: usize = 0;
@ -94,7 +94,7 @@ pub const Polygon = struct {
// Paginate
while (next_url) |cursor_url| {
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
const authed = try appendApiKey(allocator, cursor_url, self.api_key);
defer allocator.free(authed);
@ -123,7 +123,7 @@ pub const Polygon = struct {
allocator: std.mem.Allocator,
symbol: []const u8,
) ![]Split {
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
const url = try http.buildUrl(allocator, base_url ++ "/v3/reference/splits", &.{
.{ "ticker", symbol },

View file

@ -153,7 +153,7 @@ pub const Tiingo = struct {
// Honor Tiingo's 50/hour free-tier cap. Blocks only once a
// single run has spent its 50-token burst within the hour.
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
var response = try self.client.get(url);
defer response.deinit();
@ -229,7 +229,7 @@ pub const Tiingo = struct {
defer allocator.free(url);
// One batched call = one request against the hourly bucket.
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
var response = try self.client.get(url);
defer response.deinit();

View file

@ -48,7 +48,7 @@ pub const TwelveData = struct {
from: Date,
to: Date,
) ![]Candle {
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
var from_buf: [10]u8 = undefined;
var to_buf: [10]u8 = undefined;
@ -82,7 +82,7 @@ pub const TwelveData = struct {
allocator: std.mem.Allocator,
symbol: []const u8,
) !Quote {
self.rate_limiter.acquire();
try self.rate_limiter.acquire();
const url = try http.buildUrl(allocator, base_url ++ "/quote", &.{
.{ "symbol", symbol },

View file

@ -103,8 +103,9 @@ pub const analysis = @import("analytics/analysis.zig");
/// audit command renders these; downstream tools consume them.
pub const reconcile = @import("analytics/reconcile.zig");
/// Brokerage export parsers (Schwab/Fidelity/Wells Fargo positions +
/// summary) and the normalized `BrokeragePosition` shape.
/// Brokerage export parsers (Schwab/Fidelity positions CSVs + Schwab
/// summary, Wells Fargo positions spreadsheet) and the normalized
/// `BrokeragePosition` shape.
pub const brokerage = @import("brokerage.zig");
/// Portfolio loading from srf files: the union-merge loader the CLI uses

View file

@ -82,6 +82,13 @@ pub const DataError = error{
/// user "this symbol isn't in the provider's catalog; mark it
/// manually" instead of an opaque "fetch failed."
NotFound,
/// The calling task was canceled (e.g. the TUI is quitting and
/// `PortfolioData.cancelLoad` stopped its workers). Not a data or
/// provider fault: nothing is cached, negative-cached, or retried,
/// and no fallback provider is tried. Callers return; looping on to
/// the next symbol would run uncancelable, because std delivers a
/// cancelation to exactly one cancelation point.
Canceled,
};
/// Per-call options controlling cache vs network behavior. Drives
@ -717,7 +724,9 @@ pub const DataService = struct {
}
// Try server sync before hitting providers (skipped on force_refresh).
if (!opts.force_refresh and self.syncFromServer(symbol, data_type)) {
// A canceled sync stops here: falling through to the provider would
// be exactly the uncancelable extra work a cancel must prevent.
if (!opts.force_refresh and (self.syncFromServer(symbol, data_type) catch return DataError.Canceled)) {
if (s.read(self.allocator, T, symbol, postProcess, .fresh_only)) |cached| {
// The hook applies here too. Without it a configured
// ZFIN_SERVER defeats the whole mechanism: the sync
@ -738,10 +747,14 @@ pub const DataService = struct {
log.debug("{s}: fetching {s} from provider", .{ symbol, @tagName(data_type) });
self.assertNetworkAllowed("fetchCached fetchFromProvider");
const fetched = self.fetchFromProvider(T, symbol) catch |err| {
// Before anything that caches or retries: a cancel is not a
// statement about the symbol.
if (err == error.Canceled) return DataError.Canceled;
if (err == error.RateLimited) {
// Wait and retry once
self.rateLimitBackoff();
self.rateLimitBackoff() catch return DataError.Canceled;
const retried = self.fetchFromProvider(T, symbol) catch |retry_err| {
if (retry_err == error.Canceled) return DataError.Canceled;
log.warn("{s}: {s} fetch failed after rate-limit retry: {t}", .{ symbol, @tagName(data_type), retry_err });
return DataError.FetchFailed;
};
@ -950,6 +963,8 @@ pub const DataService = struct {
self.allocator.free(triple.candles);
log.warn("{s}: Tiingo full history returned no bars, trying Yahoo", .{symbol});
} else |err| {
// Nor may a cancel: the Yahoo request would run uncancelable.
if (err == error.Canceled) return DataError.Canceled;
// Transient failures must not silently degrade to a
// second provider - the caller needs to know to retry.
if (err == error.RateLimited or isTransientError(err)) return DataError.TransientError;
@ -988,6 +1003,7 @@ pub const DataService = struct {
self.allocator.free(candles);
log.warn("{s}: Yahoo full history returned no bars", .{symbol});
} else |err| {
if (err == error.Canceled) return DataError.Canceled;
log.warn("{s}: Yahoo full history failed: {s}", .{ symbol, @errorName(err) });
// Both providers affirmatively disclaim the symbol.
if (tiingo_not_found and isPermanentProviderFailure(err)) return error.NotFound;
@ -1156,6 +1172,7 @@ pub const DataService = struct {
log.debug("{s}: candles from Yahoo (Tiingo backoff active)", .{symbol});
return .{ .candles = candles, .provider = .yahoo, .tiingo_coverage = .unknown };
} else |err| {
if (err == error.Canceled) return DataError.Canceled;
log.warn("{s}: Yahoo (Tiingo backoff active) failed: {s}", .{ symbol, @errorName(err) });
}
} else |_| {}
@ -1170,6 +1187,9 @@ pub const DataService = struct {
log.debug("{s}: candles from Tiingo", .{symbol});
return .{ .candles = candles, .provider = .tiingo, .tiingo_coverage = .covered };
} else |err| {
// A cancel is not a Tiingo verdict, and every branch below
// would make another (now uncancelable) request.
if (err == error.Canceled) return DataError.Canceled;
log.warn("{s}: Tiingo failed: {s}", .{ symbol, @errorName(err) });
if (err == error.Unauthorized) {
@ -1180,19 +1200,22 @@ pub const DataService = struct {
if (err == error.RateLimited) {
// Rate limited: back off and retry - this is expected, not a failure
log.info("{s}: Tiingo rate limited, backing off", .{symbol});
self.rateLimitBackoff();
self.rateLimitBackoff() catch return DataError.Canceled;
if (tg.fetchCandles(self.allocator, symbol, from, to)) |candles| {
log.debug("{s}: candles from Tiingo (after rate limit backoff)", .{symbol});
return .{ .candles = candles, .provider = .tiingo, .tiingo_coverage = .covered };
} else |retry_err| {
if (retry_err == error.Canceled) return DataError.Canceled;
log.warn("{s}: Tiingo retry after backoff failed: {s}", .{ symbol, @errorName(retry_err) });
if (retry_err == error.RateLimited) {
// Still rate limited after backoff - one more try
self.rateLimitBackoff();
self.rateLimitBackoff() catch return DataError.Canceled;
if (tg.fetchCandles(self.allocator, symbol, from, to)) |candles| {
log.debug("{s}: candles from Tiingo (after second backoff)", .{symbol});
return .{ .candles = candles, .provider = .tiingo, .tiingo_coverage = .covered };
} else |_| {}
} else |final_err| {
if (final_err == error.Canceled) return DataError.Canceled;
}
}
// Exhausted rate limit retries - treat as transient
return DataError.TransientError;
@ -1230,6 +1253,7 @@ pub const DataService = struct {
log.info("{s}: candles from Yahoo (Tiingo fallback)", .{symbol});
return .{ .candles = candles, .provider = .yahoo, .tiingo_coverage = coverage };
} else |err| {
if (err == error.Canceled) return DataError.Canceled;
log.warn("{s}: Yahoo fallback also failed: {s}", .{ symbol, @errorName(err) });
}
} else |_| {
@ -1401,7 +1425,7 @@ pub const DataService = struct {
// Stale - try server sync before incremental fetch.
// (Force-refresh skips server sync too: the user explicitly
// asked for fresh provider data.)
if (!opts.force_refresh and self.syncCandlesFromServer(symbol)) {
if (!opts.force_refresh and (self.syncCandlesFromServer(symbol) catch return DataError.Canceled)) {
// Re-read meta: the sync wrote the server's bytes
// verbatim, so its view of both freshness AND
// adjustment basis is now ours. `serverBarRegression`
@ -1448,6 +1472,9 @@ pub const DataService = struct {
if (self.refetchFullHistory(symbol, today, now_s, now_s < m.tiingo_retry_after_s)) |candles| {
return .{ .data = candles, .source = .fetched, .timestamp = std.Io.Timestamp.now(self.io, .real).toSeconds(), .allocator = self.allocator };
} else |err| {
// A cancel ends the call; falling through would
// start the incremental fetch uncancelable.
if (err == DataError.Canceled) return DataError.Canceled;
// Restatement is best-effort. The existing series
// is untouched and still usable, just understated
// by the missed adjustment - fall through to the
@ -1475,6 +1502,8 @@ pub const DataService = struct {
// Incremental fetch from day after last cached candle
self.assertNetworkAllowed("getCandles incremental fetchCandlesFromProviders");
const result = self.fetchCandlesFromProviders(symbol, fetch_from, today, now_s < m.tiingo_retry_after_s) catch |err| {
// Not a failure of the symbol: no fail_count, no stale fallback.
if (err == DataError.Canceled) return DataError.Canceled;
if (err == DataError.TransientError) {
// Increment fail_count for this symbol
const new_fail_count = m.fail_count +| 1; // saturating add
@ -1540,7 +1569,7 @@ pub const DataService = struct {
}
// No usable cache - try server sync first (skipped on force_refresh).
if (!opts.force_refresh and self.syncCandlesFromServer(symbol)) {
if (!opts.force_refresh and (self.syncCandlesFromServer(symbol) catch return DataError.Canceled)) {
if (s.isCandleMetaFresh(symbol)) {
log.debug("{s}: candles synced from server and fresh (no prior cache)", .{symbol});
if (s.read(self.allocator, Candle, symbol, null, .any)) |r|
@ -1569,6 +1598,8 @@ pub const DataService = struct {
const prior_backoff: i64 = if (meta_result) |mr| mr.meta.tiingo_retry_after_s else 0;
const candles = self.refetchFullHistory(symbol, today, now_s, now_s < prior_backoff) catch |err| {
// Not a failure of the symbol: no fail_count, no negative cache.
if (err == DataError.Canceled) return DataError.Canceled;
if (err == DataError.TransientError) {
// Transient: increment fail_count on existing meta so
// we know to back off if this keeps happening.
@ -2008,7 +2039,7 @@ pub const DataService = struct {
}
// Try server sync before hitting Wikidata.
if (!opts.force_refresh and self.syncFromServer(symbol, .classification)) {
if (!opts.force_refresh and (self.syncFromServer(symbol, .classification) catch return DataError.Canceled)) {
if (s.read(self.allocator, Wikidata.ClassificationRecord, symbol, null, .fresh_only)) |cached| {
log.debug("{s}: classification synced from server", .{symbol});
return .{ .data = cached.data, .source = .cached, .timestamp = cached.timestamp, .allocator = self.allocator };
@ -2021,8 +2052,9 @@ pub const DataService = struct {
const symbols = [_][]const u8{symbol};
const fetched = wd.fetch(self.allocator, &symbols) catch |err| {
if (err == error.Canceled) return DataError.Canceled;
if (err == error.RateLimited) {
self.rateLimitBackoff();
self.rateLimitBackoff() catch return DataError.Canceled;
if (wd.fetch(self.allocator, &symbols)) |retried| {
return self.finalizeClassification(symbol, retried, opts);
} else |_| {}
@ -2327,7 +2359,7 @@ pub const DataService = struct {
return DataError.FetchFailed;
}
if (!opts.force_refresh and self.syncFromServer(cik, .entity_facts)) {
if (!opts.force_refresh and (self.syncFromServer(cik, .entity_facts) catch return DataError.Canceled)) {
if (s.read(self.allocator, Edgar.EntityFactRecord, cik, null, .fresh_only)) |cached| {
log.debug("CIK {s}: entity_facts synced from server", .{cik});
return .{ .data = cached.data, .source = .cached, .timestamp = cached.timestamp, .allocator = self.allocator };
@ -2411,7 +2443,7 @@ pub const DataService = struct {
return DataError.FetchFailed;
}
if (!opts.force_refresh and self.syncFromServer(symbol, .etf_metrics)) {
if (!opts.force_refresh and (self.syncFromServer(symbol, .etf_metrics) catch return DataError.Canceled)) {
if (s.read(self.allocator, Edgar.EtfMetricRecord, symbol, null, .fresh_only)) |cached| {
log.debug("{s}: etf_metrics synced from server", .{symbol});
return .{
@ -3194,7 +3226,7 @@ pub const DataService = struct {
fn run(io: std.Io, svc: *DataService, slot: *ServerSyncResult, done: *AtomicCounter) std.Io.Cancelable!void {
defer _ = done.increment();
try io.checkCancel();
slot.success = svc.syncCandlesFromServer(slot.symbol);
slot.success = try svc.syncCandlesFromServer(slot.symbol);
}
};
@ -3651,11 +3683,13 @@ pub const DataService = struct {
/// Sleep before retrying after a rate limit error.
/// Uses the provider's rate limiter if available, otherwise a fixed 10s backoff.
fn rateLimitBackoff(self: *DataService) void {
/// Wait out a provider's 429. Returns `error.Canceled` if the task
/// is canceled while waiting; callers stop rather than retry.
fn rateLimitBackoff(self: *DataService) std.Io.Cancelable!void {
if (self.td) |*td| {
td.rate_limiter.backoff();
try td.rate_limiter.backoff();
} else {
std.Io.sleep(self.io, std.Io.Duration.fromSeconds(10), .awake) catch |err| log.debug("rate-limit backoff sleep interrupted: {t}", .{err});
try std.Io.sleep(self.io, std.Io.Duration.fromSeconds(10), .awake);
}
}
@ -3665,6 +3699,10 @@ pub const DataService = struct {
/// Returns true if the file was successfully synced, false on any error.
/// Silently returns false if no server is configured.
///
/// `error.Canceled` means the calling task was canceled mid-sync.
/// Callers stop rather than fall back to a provider; see
/// `DataError.Canceled`.
///
/// Applies a single retry with a short delay when the first attempt
/// fails at the HTTP layer OR produces a torn body (integrity
/// mismatch / `looksCompleteSrf` rejection). Motivation: refreshes
@ -3676,7 +3714,7 @@ pub const DataService = struct {
/// from the same refresh are the most valuable diagnostic signal
/// we can produce (same body shape? same byte offset? same time
/// delta? all answers we can't get from a single failure).
fn syncFromServer(self: *DataService, symbol: []const u8, data_type: cache.DataType) bool {
fn syncFromServer(self: *DataService, symbol: []const u8, data_type: cache.DataType) std.Io.Cancelable!bool {
const server_url = self.config.server_url orelse return false;
const endpoint = switch (data_type) {
.candles_daily => "/candles",
@ -3709,18 +3747,21 @@ pub const DataService = struct {
"{s}: retrying {s} server sync (attempt {d}/{d}) after {d}ms delay",
.{ symbol, @tagName(data_type), attempt + 1, max_attempts, retry_delay_ms },
);
std.Io.sleep(self.io, std.Io.Duration.fromMilliseconds(retry_delay_ms), .awake) catch |err| log.debug("syncFromServer retry-delay sleep interrupted: {t}", .{err});
try std.Io.sleep(self.io, std.Io.Duration.fromMilliseconds(retry_delay_ms), .awake);
}
switch (self.tryOneSync(symbol, data_type, full_url)) {
.ok => return true,
// Torn or network error - retry if attempts remain.
.torn, .net_err => {},
.canceled => return error.Canceled,
}
}
return false;
}
const SyncAttempt = enum { ok, torn, net_err };
/// `canceled`: the request was stopped by a cancelation of the
/// calling task, which must not be retried (see `http.Client.request`).
const SyncAttempt = enum { ok, torn, net_err, canceled };
/// One attempt at syncing a file from the server. Archives a torn
/// body when detected but does NOT retry - the caller decides that.
@ -3746,6 +3787,11 @@ pub const DataService = struct {
const extra_headers = serverAuthHeaders(self.config.server_api_key, &hdr_buf);
var response = client.request(.GET, full_url, null, extra_headers) catch |err| {
const elapsed_ms = @divTrunc(std.Io.Timestamp.now(self.io, .awake).nanoseconds - t_start, std.time.ns_per_ms);
// Not a sync failure, so not a warning: the caller is stopping.
if (err == error.Canceled) {
log.debug("{s}: tryOneSync finished ({s}) result=canceled elapsed_ms={d}", .{ symbol, @tagName(data_type), elapsed_ms });
return .canceled;
}
// Operator-visible: surfaces meaningful failures
// (`NoAddressReturned`, `ConnectionRefused`,
// `TlsInitializationFailed`, etc.) instead of swallowing
@ -3983,9 +4029,9 @@ pub const DataService = struct {
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);
fn syncCandlesFromServer(self: *DataService, symbol: []const u8) std.Io.Cancelable!bool {
const daily = try self.syncFromServer(symbol, .candles_daily);
const meta = try self.syncFromServer(symbol, .candles_meta);
return daily and meta;
}
@ -4113,11 +4159,14 @@ pub const DataService = struct {
/// 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.
///
/// The exception is cancelation, which stops the batch and returns
/// `error.Canceled` whether it lands between symbols or mid-fetch.
pub fn loadAllDividends(
self: *DataService,
syms: []const []const u8,
opts: FetchOptions,
) void {
) std.Io.Cancelable!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
@ -4126,8 +4175,14 @@ pub const DataService = struct {
// (`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;
try self.io.checkCancel();
// The check above only catches a cancel that arrived between
// symbols; one that lands mid-fetch comes back as this error,
// and continuing would run every remaining symbol uncancelable.
const fr = self.getDividends(sym, opts) catch |err| switch (err) {
DataError.Canceled => return error.Canceled,
else => continue,
};
fr.deinit();
}
}
@ -4135,6 +4190,8 @@ pub const DataService = struct {
// ── Tests ─────────────────────────────────────────────────────────
const SilentServer = @import("testutil/silent_server.zig").SilentServer;
test "serverAuthHeaders: key present yields one X-API-Key header" {
var buf: [1]std.http.Header = .{.{ .name = "", .value = "" }};
const h = DataService.serverAuthHeaders("s3cret", &buf);
@ -5677,7 +5734,7 @@ test "loadAllDividends: honors skip_network for every symbol, and one miss does
// The whole loop must stay offline, not just the first symbol.
svc.panic_on_network_attempt = true;
svc.loadAllDividends(&.{ "TSTA", "TSTB" }, .{ .skip_network = true });
try svc.loadAllDividends(&.{ "TSTA", "TSTB" }, .{ .skip_network = true });
// The cached symbol survives the pass.
const b = svc.getCachedDividends(allocator, "TSTB") orelse return error.TestUnexpectedResult;
@ -5721,6 +5778,75 @@ test "getCachedSplits: reads cache only and reports absence" {
try std.testing.expect(svc.getCachedSplits(allocator, "TSTA") == null);
}
test "loadAllDividends: a cancel mid-sync stops the batch - no retry, no next symbol, no provider" {
// The TUI exit hang. Quitting cancels the dividend warm; the cancel
// used to surface as a generic read failure that the HTTP and sync
// layers retried, and every retry and every later symbol then ran
// uncancelable, so exit waited out the whole batch against a server
// stuck on rate-limited provider fetches. Here the server never
// answers at all: the batch must still return Canceled after one
// connection.
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 server: SilentServer = undefined;
try server.start(io, .hold_open);
defer server.stop();
var url_buf: [64]u8 = undefined;
var svc = DataService.init(io, allocator, .{ .cache_dir = dir_path, .server_url = try server.baseUrl(&url_buf) });
defer svc.deinit();
// Falling back to a provider after a cancel would panic here.
svc.panic_on_network_attempt = true;
const symbols = [_][]const u8{ "SMPLA", "SMPLB", "SMPLC" };
var warm = try io.concurrent(DataService.loadAllDividends, .{ &svc, &symbols, FetchOptions{} });
try server.waitForConnections(1, 5000);
try std.testing.expectError(error.Canceled, warm.cancel(io));
// A retry (250ms sync delay) or the next symbol would have connected.
try io.sleep(.fromMilliseconds(400), .awake);
try std.testing.expectEqual(@as(u32, 1), server.accepted.load(.acquire));
// Nothing was cached for the canceled symbol, not even a negative entry.
try std.testing.expect(svc.getCachedDividends(allocator, "SMPLA") == null);
}
test "getCandles / getClassification: a cancel during server sync returns Canceled without a provider fallback" {
// Same contract as the dividend warm, through the other entry points
// that sync from the server first. Cold caches, so each goes straight
// to the sync.
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 server: SilentServer = undefined;
try server.start(io, .hold_open);
defer server.stop();
var url_buf: [64]u8 = undefined;
var svc = DataService.init(io, allocator, .{ .cache_dir = dir_path, .server_url = try server.baseUrl(&url_buf) });
defer svc.deinit();
svc.panic_on_network_attempt = true;
var pending_candles = try io.concurrent(DataService.getCandles, .{ &svc, "SMPLA", FetchOptions{} });
try server.waitForConnections(1, 5000);
try std.testing.expectError(DataError.Canceled, pending_candles.cancel(io));
var pending_classification = try io.concurrent(DataService.getClassification, .{ &svc, "SMPLB", FetchOptions{} });
try server.waitForConnections(2, 5000);
try std.testing.expectError(DataError.Canceled, pending_classification.cancel(io));
try io.sleep(.fromMilliseconds(400), .awake);
try std.testing.expectEqual(@as(u32, 2), server.accepted.load(.acquire));
}
test "loadAllDividends: empty symbol list is a no-op" {
const allocator = std.testing.allocator;
const io = std.testing.io;
@ -5736,7 +5862,7 @@ test "loadAllDividends: empty symbol list is a no-op" {
// No symbols means no fetches, so this must hold even with network
// otherwise allowed.
svc.panic_on_network_attempt = true;
svc.loadAllDividends(&.{}, .{});
try svc.loadAllDividends(&.{}, .{});
}
test "getQuote offline mode returns FetchFailed (quotes never cached)" {

View file

@ -0,0 +1,94 @@
//! Test-only: a loopback TCP server that accepts connections and never
//! answers.
//!
//! In `.hold_open` mode a request to it connects, sends, and then blocks
//! in `receiveHead` for as long as the test likes - the shape of a
//! zfin-server holding a request open while it fetches from a
//! rate-limited provider. Tests use it to prove that canceling the task
//! making the request stops it promptly. In `.hang_up` mode every
//! connection is closed as soon as it is accepted, a real transport
//! failure that the HTTP client should still retry.
//!
//! Either way `accepted` counts connections, so a retry, or a move on to
//! the next symbol, shows up as an extra one.
//!
//! Loopback only, ephemeral port, nothing leaves the machine.
const std = @import("std");
const net = std.Io.net;
pub const SilentServer = struct {
pub const Behavior = enum {
/// Accept and hold the connection open without responding.
hold_open,
/// Accept and close immediately without responding.
hang_up,
};
io: std.Io,
server: net.Server,
behavior: Behavior,
/// Connections accepted so far.
accepted: std.atomic.Value(u32) = .init(0),
/// Streams held open in `.hold_open` mode, so a client waits on a
/// live connection instead of seeing it closed. Only the first
/// `accepted` entries (up to `max_streams`) are set.
// SAFETY: entries are written by `acceptLoop` before `accepted` is
// incremented past them, and only that many are read in `stop`.
streams: [max_streams]net.Stream = undefined,
acceptor: ?std.Io.Future(void) = null,
const max_streams = 8;
/// Listen on 127.0.0.1 with an ephemeral port and start accepting.
/// The server must not move after `start`; the accept task holds a
/// pointer to it.
pub fn start(self: *SilentServer, io: std.Io, behavior: Behavior) !void {
const address: net.IpAddress = try .parseIp4("127.0.0.1", 0);
self.* = .{ .io = io, .server = try address.listen(io, .{}), .behavior = behavior };
errdefer self.server.deinit(io);
// `concurrent`, not `async`: accepting blocks, and the test
// thread has to keep running alongside it.
self.acceptor = try io.concurrent(acceptLoop, .{self});
}
/// Stop accepting and close everything.
pub fn stop(self: *SilentServer) void {
if (self.acceptor) |*f| f.cancel(self.io);
if (self.behavior == .hold_open) {
const held = @min(self.accepted.load(.acquire), max_streams);
for (self.streams[0..held]) |*s| s.close(self.io);
}
self.server.deinit(self.io);
}
/// `http://127.0.0.1:<port>`, written into `buf`.
pub fn baseUrl(self: *const SilentServer, buf: []u8) ![]const u8 {
return std.fmt.bufPrint(buf, "http://127.0.0.1:{d}", .{self.server.socket.address.getPort()});
}
/// Wait until at least `n` connections have been accepted, or fail
/// after `timeout_ms`.
pub fn waitForConnections(self: *SilentServer, n: u32, timeout_ms: u32) !void {
var waited: u32 = 0;
while (self.accepted.load(.acquire) < n) : (waited += 5) {
if (waited >= timeout_ms) return error.ConnectionNeverArrived;
try self.io.sleep(.fromMilliseconds(5), .awake);
}
}
fn acceptLoop(self: *SilentServer) void {
while (true) {
// Canceled by `stop`; any other accept failure also ends the
// loop, and the test sees it as a missing connection.
const stream = self.server.accept(self.io) catch return;
const index = self.accepted.load(.acquire);
if (self.behavior == .hold_open and index < max_streams) {
self.streams[index] = stream;
} else {
stream.close(self.io);
}
_ = self.accepted.fetchAdd(1, .release);
}
}
};