Collecting data

A walkthrough of reading a consumer's banking and energy data once you hold a grant. If you have not got that far, start with the README's setup steps.

Everything here uses OpenFeed.Sharing, which has one function per sharing-api endpoint. This guide covers the shape of the work; for the field-by-field schema, use OpenFeed's public reference at https://openfeed.au/developers/api.

What you have, and what you need

A grant gives you an access token. Pass the OpenFeed.Tokens struct rather than the bare string:

{:ok, tokens} = OpenFeed.Auth.exchange_code(config, code, nonce: nonce, pkce_verifier: verifier)

The struct carries the token_type OpenFeed issued, which is the only reliable source of the authorization scheme. OpenFeed issues client-credentials tokens as Bearer even to a private_key_jwt client, so inferring the scheme from your client profile is wrong in exactly one place — and that place returns a bare 401 with no error code to explain it.

Using ash_openfeed? AshOpenFeed.with_token/3 hands you both:

AshOpenFeed.with_token(grant, fn config, tokens ->
  OpenFeed.Sharing.banking_accounts(config, tokens)
end)

Banking

Two levels: accounts, then per-account balance and transactions.

{:ok, accounts} = OpenFeed.Sharing.banking_accounts(config, tokens)

Each account is a map. The keys you will reach for most:

Balances are the expensive call

{:ok, balance} = OpenFeed.Sharing.banking_balance(config, tokens, account_id)

current = OpenFeed.Amount.to_decimal(balance["currentBalance"])

Balances are not mirrored. OpenFeed fetches them from the data holder on demand and caches for around 15 minutes, so they are the slowest calls here and the only ones that routinely fail while everything else is healthy:

{:error, %OpenFeed.Error{kind: :balance_unavailable}}

That is retryable — the data holder was unreachable, not your request being wrong. Do not fetch balances in a loop over hundreds of accounts, and do not let one failure abandon the rest of a sync.

Transactions

{:ok, transactions} =
  OpenFeed.Sharing.banking_transactions(config, tokens, account_id,
    oldest_date: Date.add(Date.utc_today(), -365),
    newest_date: Date.utc_today()
  )

oldest_date and newest_date take a Date or an ISO 8601 string, and are preserved across pagination.

Amounts are strings in ISO 20022 format, signed — positive for credits, negative for debits:

amount = OpenFeed.Amount.to_decimal(transaction["amount"])  # "-52.00" -> #Decimal<-52.00>

description, reference and merchantName are free text and are PAN-scrubbed at ingestion. status is PENDING or POSTED; a pending transaction can change or vanish, so key on transactionId and upsert.

Re-syncing without losing late arrivals

Providers post transactions late. Re-reading from exactly where you left off will miss them, so overlap:

oldest =
  case grant.last_synced_at do
    nil -> Date.add(Date.utc_today(), -365)
    at -> at |> DateTime.to_date() |> Date.add(-7)
  end

Since you are upserting on transactionId, re-reading a week is free of duplicates.

Energy

Three levels: accounts, then meters, then usage — plus invoices and billing at the account level.

{:ok, accounts} = OpenFeed.Sharing.energy_accounts(config, tokens)
{:ok, meters} = OpenFeed.Sharing.energy_meters(config, tokens, account_id)

A meter's detectedReadType matters, because it determines the shape of its usage reads:

Usage, and the trap

{:ok, days} =
  OpenFeed.Sharing.energy_usage(config, tokens, account_id, meter_id,
    oldest_date: Date.add(Date.utc_today(), -90)
  )

Each element is one day, and its reads array is where the value lives. This is the most awkward shape in the API:

%{
  "meterId" => "…",
  "intervalDate" => "2026-03-11",
  "reads" => [
    %{
      "readUType" => "intervalRead",
      "intervalRead" => %{
        "readIntervalLength" => 30,
        "aggregateValue" => 12.5,
        "intervalReads" => [0.2, 0.3, ...]
      }
    }
  ]
}

readUType is a discriminator selecting which sibling is populated. Code that only reads intervalRead.aggregateValue works perfectly against an interval meter and silently reports zero for every daily meter. That is not hypothetical — it is the bug this library's helper was extracted to fix.

Use the helper:

net = OpenFeed.Energy.net_usage(day)   # #Decimal<12.5>

It handles both variants, prefers the data holder's aggregateValue over re-summing intervals (that is what a bill is based on), and skips an unrecognised future variant rather than counting it as zero.

Positive is consumption, negative is export, so a meter with solar can produce a negative net.

"No data" is not "zero usage"

reads can come back empty. Every reading from the grant this library was tested against did:

%{"meterId" => "…", "intervalDate" => "2026-03-11", "reads" => []}

net_usage/1 returns 0 for that, which is convenient for sums but means you cannot tell it apart from a genuinely zero day. If you are charting or reporting, record the difference:

%{
  interval_date: Date.from_iso8601!(day["intervalDate"]),
  net_kwh: OpenFeed.Energy.net_usage(day),
  has_reads: OpenFeed.Energy.has_reads?(day)
}

A gap is more honest than a zero.

Invoices and billing

{:ok, invoices} = OpenFeed.Sharing.energy_invoices(config, tokens, account_id)
{:ok, billing} = OpenFeed.Sharing.energy_billing(config, tokens, account_id)

Here amounts are JSON numbers, not strings — the opposite of banking:

OpenFeed.Amount.to_decimal(invoice["invoiceAmount"])   # 193.8 -> #Decimal<193.8>

OpenFeed.Amount.to_decimal/1 takes either, which is why it exists. Convert at the boundary and keep Decimal from there on; do not accumulate floats and convert at the end.

Large collections

Collection functions fetch every page before returning. One real account produced 3,947 transactions for a single year, so for anything unbounded, stream:

config
|> OpenFeed.Client.stream(tokens, "/v1/banking/accounts/#{account_id}/transactions",
     params: [oldestDate: "2025-01-01"])
|> Stream.chunk_every(500)
|> Enum.each(&bulk_insert!/1)

stream/4 fetches one page at a time and raises OpenFeed.Error on failure, because a lazy stream has nowhere to put an error tuple.

Pagination follows the response's links.next, which is present only when more records exist. Do not reimplement this by incrementing offset until you get a short page: when the record count is an exact multiple of the page size that requests one page too many, and OpenFeed answers offset >= total with 400 no_records_found_at_offset_limit — turning a healthy read into a spurious failure.

Error handling in a real sync

A sync touches many endpoints. One failure should not abandon the run, so collect errors and keep pulling whatever OpenFeed will still give you:

defp sync_banking(config, tokens, grant) do
  case OpenFeed.Sharing.banking_accounts(config, tokens) do
    {:ok, accounts} -> Enum.flat_map(accounts, &sync_account(config, tokens, grant, &1))
    {:error, error} -> [{"banking_accounts", error}]
  end
end

Then decide once, at the end, based on what you collected:

The important line in that table is :subject_mismatch. It shares HTTP 403 with :grant_revoked, and treating them alike will mark healthy grants dead. Use OpenFeed.Error.grant_revoked?/1 rather than matching on the status.

A worked example

examples/openfeed_demo is a Phoenix + Ash app doing all of the above against production OpenFeed — lib/openfeed_demo/open_feed/sync.ex is the file to read.

Next

• Cost and cadence — what syncing costs, and how often is worth it.