Ingest

How statements from institutions enter the ledger, through in-core mappers or connector plugins the core calls, and what the IBKR Flex connector does today.

The ingest context is the only way data gets into OpenFief’s ledger (D-7). A connector reads what an institution reports, translates it into ledger transactions and balance assertions, and lists every row it could not book. The core runs that either with a mapper compiled into it, as for IBKR Flex Query statements today, or by calling a connector plugin over its Connect contract (D-49). This page explains the concepts, shows what internal/ingest actually does, and points to the tests that prove it.

Concepts#

An institution is a broker, bank or registry that holds assets: IBKR, XTB, the treasury bond registry. A connector is the plugin that brings one institution’s data in, by whatever the institution allows: a documented web service, an exported file, or manual entry (D-7). The core project ships only connectors that comply with the institution’s terms (D-8).

What an institution reports about an account is a statement, such as an IBKR Flex Query or an xStation export. The way a statement reaches OpenFief is an ingest path: a file (uploaded, passed to the CLI, or dropped in a watched folder), a pull from the institution, or a sync with a connector plugin. D-7 makes these the only way in. The public read API never writes.

One run of an ingest path is an import batch. It records which institution the statement came from, by which path, when, in which ingest mode, and what came of it.

A mapping is what a connector makes of one statement: transactions, balance assertions and ingest problems. An ingest problem is a statement row the mapping left out of the ledger, with its reason. It is unsupported when OpenFief can’t book the row yet, such as an option trade, and skipped when the row is left out on purpose, such as a summary row. A problem is not an error: the rest of the statement is still imported. An unsupported row’s effect is missing from the ledger, though, so a balance assertion may fail until that row type is supported.

Idempotent re-import. Statements overlap: a daily pull re-reads days that were already imported. So every transaction carries a key: its source plus the institution’s own ID for it, such as IBKR’s tradeID (D-18). Importing the same content again changes nothing. Changed content under the same key becomes a new version that supersedes the old one, which is kept. The ledger does the versioning; the connector’s job is to produce stable keys.

Seeding, then transactions. OpenFief does not replay an account’s full history (D-23). At go-live, an opening batch seeds the account from the institution’s current state: one opening lot per broker lot, with its original open date and cost, plus the opening cash, a deposit (or a withdrawal for a negative balance) against equity, keyed opening-cash:<currency>:<date> (D-33). An account whose current transactions include either is seeded, and the ledger refuses to seed it again unless nothing would change. After that, transactions batches import transactions from go-live on. Both modes import balance assertions.

The connector contract. A connector plugin serves the connector contract, openfief.connector.v1.ConnectorService in connector.proto. Sync pulls what the institution reported since a watermark the connector returned last time. ImportFile maps one statement file. Each call is one import batch, with a batch ID the core generates and reuses on a retry. The answer is a delivery: the mapping, plus any end-of-day prices, FX rates and index readings (CPI and the NBP reference rate, for treasury bonds). The core authenticates each call with a short-lived token whose audience is that plugin instance (D-59). It prefixes every delivered transaction’s key source with plugin:<instance>:, so that no instance can write or supersede another’s transactions (D-48).

Anti-corruption layer. Eric Evans described this pattern in Domain-Driven Design (2003): a translation layer at the boundary, so that a foreign model never leaks into your own. Here there are two boundaries. Each in-core adapter, such as ibkrflex, parses the institution’s format and keeps IBKR words (position, trade, conid, Flex section and attribute names) to itself. The connectrpc adapter translates the contract’s protobuf messages, so no generated type reaches the domain or the use cases (code-organization.md).

IBKR Flex Queries and the Flex Web Service. A Flex Query is a query the account owner sets up in IBKR’s portal: which sections (trades, open positions, cash) and which fields. IBKR returns the result as XML, FlexQueryResponse > FlexStatements > FlexStatement, with one FlexStatement per account and period and one element per row. The Flex Web Service fetches it with a read-only token, in two steps. SendRequest with the token and the query ID returns a reference code. GetStatement with that code returns the statement once IBKR has generated it; until then, it answers with an error code that means “try again”. R-3 lists Flex Queries among IBKR’s access paths.

How it works#

Packages#

The domain types, as declared in internal/ingest:

NewBatch refuses a batch without an ID, an institution, a source, a known mode or a start date (ErrInvalidBatch); the zero Mode is not a mode. Batch.Complete records the mapping’s counts, the ledger’s summary and a copy of the problems. IndexReading.Validate requires a known index, a date and a source, and a CPI reading dated on the first of its month (ErrInvalidIndexReading). ScopeToConnector returns copies of transactions whose key source is plugin:<instance>:<source>; an empty source stays empty, so the ledger still rejects it.

The ports are declared in app, by their consumer. cmd/core connects them for connector plugins: its ledgerImporter turns the ledger’s summary into ingest.ImportSummary (as cmd/demo-seed’s importer does for Service), and its valuationInputs records prices, FX rates and index readings through valuation’s InputRecorder, each index reading as the reading for the month of its date:

identity/kube.CoreTokens implements CoreTokens with the TokenRequest API (Identity); connectrpc.Client gets each instance’s address from Discovery.BaseURL.

Running an in-core mapper#

Service holds a LedgerImporter, a Mappers map from InstitutionID to Mapper, and injected Today and NewBatchID functions, so it never reads the clock. IngestFile takes an io.Reader and records the source file (SourceFile). Pull checks that the institution has a mapper, calls the StatementFetcher, and ingests the bytes with the source pull (SourcePull). Then both look up the mapper (ErrUnknownInstitution if none), start the batch with NewBatch, and Map the statement. importMapping then imports it by mode:

• An opening batch goes to LedgerImporter.Seed. The ledger refuses to seed an account that is already seeded (ledger.ErrAlreadySeeded), unless every incoming transaction of it is unchanged; then the summary lists it in AlreadySeeded.
• A transactions batch is refused with ErrSeedingOutsideOpening if any transaction seeds (ledger.IsSeeding: an opening lot, or a deposit or withdrawal keyed opening-cash:). Otherwise it goes to LedgerImporter.Import.

Last, Complete fills in the batch, which is returned to the caller. A failed fetch, map or import returns an error and no batch, and nothing reaches the ledger before the map succeeds.

The pull path end to end, with the IBKR adapter:

Calling a connector plugin#

Connectors runs connector plugins’ ingest paths. Its caller supplies the instance name, the batch ID and, for a sync, the last watermark, and owns the schedule and the retries: a failed batch is retried with the same batch ID. connectrpc.Client resolves the instance’s address with a baseURL function, such as its ClusterIP Service, asks CoreTokens for a fresh token, and calls the plugin over HTTP/2 without TLS (H2CClient). ImportFile streams a FileHeader (batch ID, mode, the owner’s file name as a hint, media type) and then the file in 64 KiB chunks (ChunkSize). A sync, end to end:

The translation in mapping.go refuses, with ErrInvalidMessage naming the field: a decimal with an exponent, grouping or a plus sign; a date that isn’t YYYY-MM-DD; a malformed currency; an ISIN with a wrong check digit or an instrument without a symbol (D-35); a missing amount or delivery; any unspecified enum; a price or FX rate that isn’t positive; and a delivery whose batch_id isn’t the one asked for. The client never scopes keys; Connectors does. Every price, FX rate and index reading gets the instance as its source, overwriting what the plugin sent.

Connector bindings#

A connector instance writes only to the accounts the owner bound it to. A connector binding pairs an instance name with a ledger account ID, and app.Bind adds one after checking the name (ValidateInstanceName: a Plugin’s name, a lower-case DNS label of at most 56 characters) and that the account is stored. Before anything is imported, Connectors.deliver loads the instance’s bindings and CheckBound refuses the whole delivery, with an *UnboundAccountsError (ErrAccountNotBound) naming every other account, if any transaction or balance assertion is in an account the instance isn’t bound to. A delivery with only prices, FX rates or index readings, such as a market-data connector’s, needs no binding. Key scoping (D-48) keeps instances from overwriting each other’s transactions; the binding keeps an instance from writing into an account at all.

Scheduling syncs#

The core owns the schedules, retries and import batches (D-49). Scheduler syncs each running connector instance when it is due:

• InstanceLister names the instances that run now. An instance that is uninstalled is no longer listed, so it is no longer synced, and its watermark, batches and bindings stay, so reinstalling it under the same name continues from them (D-58).
• The sync policy (ingest.SyncPolicy) sets when: every Interval after a success, plus a random jitter of up to Jitter, and after a failure again after RetryFirst, doubled for each consecutive failure, up to RetryMax. An instance that was never synced is due at once.
• One sync per instance at a time: in the process, Scheduler tracks the instances it is syncing; across core replicas, SyncStore.ClaimSync claims the instance until the attempt’s timeout plus a minute, so a crashed attempt’s claim lapses.
• A failed attempt leaves its import batch open, and the next attempt retries it with the same batch ID and mode, as the connector contract requires. The first batch of an instance is an opening batch; once one completes, the instance is seeded and later batches are transactions batches (D-23). SyncNow runs one attempt at once and can ask for a mode, which abandons an open batch of the other mode.
• Each attempt has a timeout for the call and the import together. Its outcome is recorded even when the context ended meanwhile.

Two transactions, in a safe order. The ledger import and the valuation inputs are stored in separate database transactions, the ledger first. If storing the inputs fails, the batch fails with the transactions already imported. The retry delivers the same batch again: the ledger finds every transaction unchanged (D-18), and prices, FX rates and index readings replace the rows stored under the same key. The watermark is stored only by CompleteSync, after both steps, in the same database transaction that marks the batch completed, so a sync is never skipped.

Finding the instances. kube.Discovery watches, with client-go informers in the openfief namespace, the Plugin resources (read as unstructured data, so ingest doesn’t depend on the operator’s Go types) and the Services labelled app.kubernetes.io/name: openfief-plugin. An instance runs when its Plugin is not being deleted, its Ready condition is True, its status.effective.capabilities include connector, and its Service, found by the label openfief.io/plugin: <name>, has a ClusterIP and a port named connect. The core calls http://<ClusterIP>:<port>, with no DNS lookup. The names are the operator’s runtime contract; a test checks them against internal/plugins/operator.

In the core. cmd/core serve runs the scheduler when --connectors is on, by default inside a cluster: it starts the informers and the scheduler in the background, ticks every minute, and stops with the process, recording what was in flight. Discovery that can’t reach the API server, or a sync that fails, never affects /readyz. The flags are --sync-every (6h), --sync-jitter (30m), --sync-retry-first (5m), --sync-retry-max (1h), --sync-timeout (5m) and --service-account (openfief-core). core connector bind --instance NAME --account ID adds a connector binding; core connector sync --instance NAME [--mode opening|transactions] syncs one instance now, as the scheduler would, and prints the batch’s counts. Batch IDs are b- and 16 random hex digits.

internal/ingest/postgres keeps this in the ingest schema, migrated by its own runner and advisory lock: connector_instances (watermark, seeded, failures, next sync, claim, open batch), import_batches (instance, mode, state, attempts, and the completed batch’s counts) and connector_bindings. It holds IDs, counts and times only, never an amount or an account number.

The Flex Web Service client#

ibkrflex.Client sends the token only as the t query parameter, next to q and v=3, and only to SendRequestURL and GetStatementURL, which default to IBKR’s documented endpoints. serviceResponse has no field for the statement URL in IBKR’s reply, so the token never follows it elsewhere. Errors name the request path, never the URL that carries the token. SendRequest is not retried. GetStatement is retried only on the codes in retryCode (1001, 1004 to 1009, 1018, 1019, 1021). A SendRequest response is capped at 64 KiB, and a GetStatement response at MaxBytes, by default the parser’s 64 MiB. Wait is injectable, so tests never sleep, and a cancelled context stops the wait. The HTTP field’s comment says it should reach IBKR through the egress proxy (D-32); the client doesn’t check that.

Mapping a Flex statement#

ibkrflex.Mapper has two fields: Accounts, which maps an IBKR account ID to the ledger account it feeds, and Limits. Map refuses an unknown mode (ErrUnknownMode), decodes the XML, then maps each FlexStatement. A statement for an account not in Accounts becomes one skipped problem. A statement without a readable toDate fails the whole map with ErrMalformed. The toDate, the statement’s last day, dates every balance assertion; in opening mode it is also the go-live date. Four sections are read, in this order:

In opening mode every Trades and CashTransactions row becomes a skipped problem, because the opening lots and opening cash already hold its effect. Only instruments that appear in OpenPositions get a holding balance assertion. A zero cash balance still gets a cash assertion but no opening cash transaction.

The keys the adapter writes. Every source is SourcePrefix plus the ledger account ID, such as ibkr-flex:acc-ibkr, never the IBKR account number:

Withholding tax is paired with its dividend by pairWithholding: the same actionID when both rows have one, otherwise the same ISIN, currency and date. Several withholding rows of one dividend add up. A withholding row with no dividend, more than one, or a zero or positive amount becomes an unsupported problem.

Rows the mapper won’t book become problems, never errors. Unsupported: asset categories other than STK (and CASH in Trades); a contract multiplier other than 1; short or zero holdings; transaction types other than ExchTrade, BookTrade, FracShare or none; a side that disagrees with the quantity’s sign; commission or tax credits; commission and taxes in two currencies; cash transaction types other than the four above; fee refunds; dividend reversals; a missing or unreadable attribute (named, its value not kept); and a transaction the ledger rejects, whose reason is cut to “not a valid ledger transaction” because ledger errors can carry amounts. Skipped: an unexpected element in a section (such as Order or Lot in Trades), a level of detail that isn’t read (such as CLOSED_LOT in OpenPositions or SUMMARY in CashTransactions), and opening-mode activity rows. A problem’s Ref names the statement, section, 1-based row and the IBKR ID when there is one, such as FlexStatement#1/Trades#8[tradeID=9000106]. It never holds an amount or an account number (GR-12).

Dates are read as yyyyMMdd or yyyy-MM-dd, optionally followed by ;, ,, space or T and a time, which is dropped without any time zone conversion. Amounts and quantities are exact decimals in plain notation, at most 40 digits. Exponents are refused, and nothing goes through float64.

Hostile statements#

DESIGN §3.7 lists malicious statement files as a threat. decode reads with encoding/xml in strict mode and refuses any DOCTYPE or other directive, undeclared entities, non-UTF-8 input and any root other than FlexQueryResponse. Anything malformed is ErrMalformed. Anything over a limit is ErrTooLarge, and either way the map fails before anything is imported. The limits are fields of Limits; a zero field takes its default: MaxBytes 64 MiB, MaxDepth 16, MaxElements 500,000, MaxAttrs 512 per element, MaxAttrLen 4,096 bytes.

Decisions#

• D-7 Only connector ingest paths: Service (file, pull) and Connectors (sync, file).
• D-11 Connector contract in protobuf, called over Connect RPC.
• D-18 Keys from source plus institution IDs; reported balances become assertions.
• D-19 Double-entry postings, built by the ledger’s constructors.
• D-20 Amounts keep their currency; IBKR’s ConversionRates are ignored.
• D-23 Opening batches seed; a seeded account isn’t seeded twice.
• D-33 Opening cash is booked against the equity ledger account.
• D-34 Trade fees go to the fee ledger account.
• D-35 Instruments need a valid ISIN and a venue symbol.
• D-48 Each instance’s transaction keys are scoped to plugin:<instance>:, and each instance writes only to the accounts it is bound to.
• D-49 The core calls connectors, one batch per call, and Scheduler owns their schedules and retries, with the batches and watermarks in the ingest schema.
• D-58 An uninstalled instance is no longer synced; its state and data stay.
• D-59 A fresh core token per call, with the instance as audience.
• D-32 Partly: the Flex client expects the egress proxy, but doesn’t enforce it.

Where the code differs from a decision, a requirement or the glossary:

• D-23, transactions from go-live. transactions mode books every row in the statement. Nothing compares row dates with go-live, so a statement whose period overlaps the opening statement’s would book those rows on top of the opening lots. problem.go gives “a transaction before go-live” as an example of a skipped row, but no code produces one.
• D-49. File uploads (Connectors.ImportFile) have no entry point and are not recorded as batches in the store.
• Index readings. Valuation keeps one reading per index and month, so a reference rate that takes effect during a month becomes that whole month’s reading, and a later change in the same month replaces it.
• IBKR. The IBKR code stays a package, internal/ingest/ibkrflex; the plugin plugins/ibkr-flex serves it to the core.
• Ingest problems. The glossary and doc.go say rows that aren’t booked become ingest problems. Rows in sections the mapper doesn’t read, and the BASE_SUMMARY cash row, are dropped silently.
• Glossary names.
  • There is no ingest.Connector type.
  • ImportSummary exists twice, in ingest and ledger/app, with different fields.
  • The glossary defines “Index reading” twice, as ingest.IndexReading and valuation.IndexReading.
  • The comment on Batch.Source gives upload and flex-web-service as examples, but the constants are file, pull and sync.

Verification#

Run every test with task test, or only this context with go test ./internal/ingest/.... All of these run offline.

All the ingest requirements above are default firmness and say the author hasn’t reviewed them yet. Of the firm D-49 requirements, REQ-ingest-core-owns-schedules is verified by the scheduler and store tests; REQ-ingest-core-calls-connectors and REQ-ingest-uploads-through-core have no test of their own yet. The ledger side is verified there: the one-transaction import by TestImportHooksRunInTheImportTransaction in hook_test.go (REQ-ledger-app-import-atomic), and the re-seed refusal by TEST-ledger-app-seed (REQ-ledger-seed-refuse-reseed).

The Flex golden files sit next to their fixtures, opening.xml and activity.xml, in testdata. To rewrite them after an intended change, run go test ./internal/ingest/ibkrflex -run TestMapGolden -update and review the diff line by line.

The hypert recordings. TestFetchStatement replays three exchanges from testdata/TestFetchStatement/: a successful SendRequest whose Url points to elsewhere.example, a GetStatement answering 1019 (still generating), and a GetStatement returning a one-trade statement. TestFetchStatementTokenInvalid replays a single SendRequest answering 1015. Replay checks each request’s path, method, query parameters, scheme and headers, including User-Agent; the token is replaced with SANITIZED before comparing. No IBKR paper credentials exist, so these files were written by hand in hypert’s format, with a fake account (U0000000), fake IDs and synthetic ISINs, and recordMode is a false constant. They follow IBKR’s public documentation, not a real response (OQ-22). To record real fixtures, follow the hypert-recording skill and testing.md. The connector client’s tests use an in-process Connect server instead, with synthetic tokens.

Known gaps. These behaviors have no test, and the first two are bugs:

• In pairWithholding (cashtx.go), a dividend’s second withholding row in a different currency resets the dividend’s withholding to zero. The first, valid row then vanishes with no problem of its own, and the dividend is booked as if nothing was withheld.
• Rows in sections the Flex mapper doesn’t read produce no ingest problem, and transactions mode has no go-live cutoff.
• Nothing checks at compile time that ibkrflex.Mapper, *ibkrflex.Client and *connectrpc.Client satisfy the app ports.

Not built yet#

• XTB export parser. Uploaded xStation files (D-9), stocks and ETFs only, with closed CFD results as cash and open CFDs flagged (D-24).
• Treasury bonds. Manual lot entry in the built-in UI, handed to the core (D-9, D-49), and an import of the bond service’s own export if one exists (D-9, OQ-22). The connector contract has only Sync and ImportFile, no call for manual entries.
• More connector plugins. ibkr-flex and nbp-fx exist (Plugins); the XTB parser and the bond imports have none yet.
• Entry points for files. Upload, CLI and watched folder (D-7). Only cmd/demo-seed builds a Service, to load the demo dataset; cmd/core builds Connectors for syncs only, and nothing calls Connectors.ImportFile yet.
• IBKR prices as a cross-check for the market-data connector (D-22). A delivery can carry prices, but the Flex mapper doesn’t read tradePrice or position prices.
• Verification against a real account of the Flex schema, lot-level open dates and cost basis, and how far back one request reaches (OQ-22).
• Full history backfill, which arrives with the tax pack (D-23).

Source: docs/concepts/ingest.md · last changed 2026-10-28 in 837f181