# auradefi 0.1.1: the complete documentation Open-source multi-tenant crypto data aggregator for Python. Reads EVM, Bitcoin and Solana balances and history, prices them exactly, keeps tenants isolated, and emits Plaid's wire format. Library first: import it and pay no network cost. The HTTP API is a thin shell over the same core. Version 0.1.1, Apache-2.0, alpha. Source: https://github.com/auracarehq/auradefi Rendered: https://auradefi.info Licence: Apache-2.0 This file is generated from the repository by scripts/build_site.py and contains every prose page, every worked example in full, and a signature listing of the public surface. The twelve notebooks and the changelog are not here; both are linked from llms.txt. Read this first: 1. Start in Sandbox. `Auradefi.sandbox()` needs no key, no network and no configuration: it replays a recording bundled inside the wheel, through the production source, decoder, ledger and pricing code. Going live is one line, `Auradefi.from_env()`. 2. Sandbox answers are constants: one address on `eip155:1`, 2 ETH and 25 USDC worth 5025 USD, seven transactions. Asking it for a different address, chain or page raises `CassetteMissError`, which means the recording does not hold that request. No credential is missing. 3. Amounts are exact. `Quantity` and `Money` wrap `Decimal` and go on the wire as tagged strings, never as JSON numbers. A float anywhere in this arithmetic is a bug. 4. An unpriced asset is never zero. It comes back in `report.holdings` with `price=None` and is named in `report.unpriced`, so a total is either right or visibly incomplete. 5. There are no Bitcoin or Solana prices. The one shipped oracle is DefiLlama: current prices, six EVM chains. Anything else needs your own `prices` port. 6. Chains are CAIP-2 strings. `user.connect_address("eip155:1", "0x…")` works; `"ethereum"` is refused, as is any chain the registry has not been given. 7. Configuration is prefixed. `Settings.from_env()` reads `AURADEFI_ETHERSCAN_API_KEY` and its siblings. A bare `ETHERSCAN_API_KEY` is ignored deliberately, so an unrelated variable cannot become this library's credential. 8. Nothing runs on its own. There is no scheduler, no worker and no background thread. You call `aura.sync(budget=n)` on your own tick, where `budget` caps the source pages that one call may spend; cursors make the next call resume. 9. The default ledger is memory and loses everything at exit. Production means `ledger=SqlModelLedger(session_factory=…)` or your own object. It takes a session factory, not a URL, because your application owns the engine and the migrations. 10. A source is one object satisfying two seams, `balances()` and `fetch_txlist()`. Binding one that has only the first raises at construction time. 11. Ports are structural protocols. There is no base class and no registration: an object with the right methods is the port. Five of them, all optional keyword arguments to `Auradefi.sandbox()` and `Auradefi.from_env()`. 12. Every failure inherits `auradefi.errors.AuradefiError`, so one `except` clause catches this library and nothing else. 13. The gaps are real and documented: no multicall, no on-chain reader for the position adapters, no historical prices, no async surface, no Solana transaction decode. If the reference does not name a symbol, it does not exist. Say so instead of inventing one. ======================================================================== # FILE: docs/quickstart.md ======================================================================== # Quickstart Five lines, no credentials. ```bash pip install auradefi ``` ```python from auradefi import Auradefi aura = Auradefi.sandbox() for holding in aura.holdings()[0].holdings: print(holding.symbol, holding.quantity, holding.value) ``` ``` ETH 2 5000.000000000000000000 USD USDC 25 25.000000 USD ``` That is a complete program. It needs no API key, no database, no network and no configuration file. ## What just happened `Auradefi.sandbox()` opens the Sandbox environment: a recording of one address' real Etherscan and DefiLlama traffic, bundled inside the package and replayed locally. Everything above the transport is the production code path, using the same source, decoder, ledger and pricing arithmetic as a live instance. Sandbox exists so you can write working code before anyone has approved an API key. It also keeps the examples in these docs from drifting away from the library. | | | |---|---| | Address | `0x1111111111111111111111111111111111111111` on `eip155:1` | | Holdings | 2 ETH at 2500 USD and 25 USDC at 1 USD, totalling 5025 USD | | History | seven transactions in blocks 100 to 107 | | Time | frozen at the instant the traffic was recorded | Because Sandbox is a recording, its answers are constants. Ask for something it does not hold, such as a different address, a second chain or a wider page, and you get `CassetteMissError` listing what it does hold. That is the offline guarantee working as intended. ## Sync some history ```python report = aura.sync(budget=10) print(report.pages_fetched, report.transactions_ingested) # 5 7 ``` `budget` caps how many source pages one call may spend. Cursors let the next call resume where this one stopped, so a tick is bounded and keeps its place. Call it again inside `sync_min_interval_s` and it is a no-op that touches no transport: ```python aura.sync(budget=10).no_op # True ``` That is the whole scheduling contract. You call `sync()` on your own schedule, from your own worker. The package starts no background thread, and nothing runs unless you ask for it. ## Go live One line changes: ```python aura = Auradefi.from_env() user = aura.user("your-opaque-user-id") user.connect_address("eip155:1", "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045") aura.sync(budget=5) ``` ```bash export AURADEFI_ETHERSCAN_API_KEY=… # optional; see Authentication ``` The key is optional. Without one, Etherscan's keyless tier applies. [Authentication & keys](authentication.html) lists every service this package talks to, which of them need a credential (almost none do), and what each one covers. Storage still defaults to memory at this point, so `from_env()` on its own loses everything when the process exits. Passing your own database is one keyword: ```python aura = Auradefi.from_env(ledger=SqlModelLedger(session_factory=…)) ``` [Bring your own](bring-your-own.html) has that in full, along with the other four ports. ## Where to go next - [Guides](examples/index.html): one file per task, covering holdings, your own source, your own database, the HTTP API, tenancy, positions, cost basis, webhooks, Bitcoin and Solana. - [Authentication & keys](authentication.html): what you need before pointing this at mainnet, and what happens when a key is wrong. - [Bring your own](bring-your-own.html): every port, its exact methods, and a minimal implementation of each. - [API reference](reference/index.html): signatures, parameters, return fields and exceptions. - [Build with an LLM](llms.html): a prompt to paste into a model before asking it for auradefi code, plus `llms.txt` and the whole documentation as one file. ## What this is not auradefi is alpha, and the [README](index.html) keeps an explicit list of what is absent: no multicall, one price oracle covering six EVM chains, no Bitcoin or Solana prices at all, no scheduler, and no on-chain reader for the DeFi position adapters. Read that list before you budget work against this. Sandbox makes the library easy to try, and the gaps are still there afterwards. ======================================================================== # FILE: README.md ======================================================================== # auradefi Open-source, multi-tenant crypto data aggregator. It takes the tenancy model from Vezgo, the DeFi position depth from DeBank and the transaction decomposition from Zerion, then emits Plaid's wire format, so crypto lands in the same downstream schema as bank and exchange data. It is a library first and a service second. A Python host imports `auradefi` directly and pays no serialisation or network cost. The HTTP API is a thin shell over the same importable core. > Status: alpha, and 0.1.1 is the release to use. Every planned capability > is implemented. The suite is 3,247 tests green offline on a fresh clone with no > API keys, all twelve notebooks execute clean, and every example under > [`examples/`](examples) runs against the published wheel. > > Do not use 0.1.0. A separate adversarial review of it found nineteen > verified defects, of which five were security and four were silent data > loss. None of them failed a test. > [`docs/internal/RELEASE_0.1.1.md`](docs/internal/RELEASE_0.1.1.md) is the > full accounting. 0.1.1 fixes all nineteen and deliberately breaks one id > derivation, so read *Upgrading* in [`CHANGELOG.md`](CHANGELOG.md) before you > move library-ingested data across. > > Alpha means the gaps in *What is not there* are real; read that section > before you budget work against this. > [`STATUS.md`](docs/internal/STATUS.md) carries the live gate state, and > [`docs/internal/SPEC.md`](docs/internal/SPEC.md) is the design contract. **[Documentation site →](https://auradefi.info/)** carries the examples, the twelve executable notebooks and the full reference, rendered with every example's real output. ## Install ```bash pip install auradefi # core; httpx is the only dependency pip install 'auradefi[sql]' # + the SQLModel ledger backend pip install 'auradefi[api]' # + the FastAPI HTTP surface ``` From a clone (if your system python has no pip, `scripts/bootstrap.sh` handles it): ```bash git clone https://github.com/auracarehq/auradefi cd auradefi && bash scripts/bootstrap.sh .venv/bin/pytest # the whole suite, offline, no keys .venv/bin/python examples/quickstart.py # every capability, end to end bash scripts/run_examples.sh # all eleven examples ``` ## Examples [`examples/`](examples) holds one file per question. Each one is self-contained, reads nothing from this repository, and runs offline without API keys. Each asserts its own output, and CI executes all of them, so a stale example fails the build. | Example | What it answers | |---|---| | [`quickstart.py`](examples/quickstart.py) | the whole library in one file; start here | | [`01_holdings_for_an_address.py`](examples/01_holdings_for_an_address.py) | a priced portfolio, exactly, with unpriced assets named | | [`02_embed_in_your_backend.py`](examples/02_embed_in_your_backend.py) | your ports, your tick, your database, and restart resume | | [`03_write_a_source_adapter.py`](examples/03_write_a_source_adapter.py) | point it at your own chain data (two methods) | | [`04_persist_to_your_database.py`](examples/04_persist_to_your_database.py) | host-owned DDL, a resumable cursor feed, a reorg | | [`05_serve_the_http_api.py`](examples/05_serve_the_http_api.py) | Plaid's envelope over HTTP, and batch partial success | | [`06_isolate_two_tenants.py`](examples/06_isolate_two_tenants.py) | one deployment, many customers, attacked four ways | | [`07_read_defi_positions.py`](examples/07_read_defi_positions.py) | an LP and a loan that still add up to net worth | | [`08_report_cost_basis_and_pnl.py`](examples/08_report_cost_basis_and_pnl.py) | FIFO/LIFO/HIFO/ACB, any instant, Plaid `tax_lots[]` | | [`09_deliver_signed_webhooks.py`](examples/09_deliver_signed_webhooks.py) | signing, the pinned retry schedule, replay | | [`10_scan_bitcoin_and_solana.py`](examples/10_scan_bitcoin_and_solana.py) | an xpub that never leaves the process, and Token-2022 | [`examples/README.md`](examples/README.md) annotates the index, and the [documentation site](https://auradefi.info/examples/) renders every example with its real output. ## Using it As a library, where the host owns storage, transport, prices and the tick: ```python from auradefi import Auradefi auradefi = Auradefi( ledger=SqlModelLedger(session_factory=my_session_factory), # your database source=MySource(), # your transport: .balances() + .fetch_txlist() prices=MyPrices(), # your price feed: .usd_prices() ) user = auradefi.user("opaque-host-user-id") # get-or-create, id is derived user.connect_address("eip155:1", "0x…") # validated now, not on a later tick report = auradefi.sync(budget=5) # budgeted, resumable, self-throttling holdings, metrics = auradefi.holdings(), auradefi.scalar_metrics() ``` As a service, where `create_app` takes ports you already built: ```python from auradefi.api.app import create_app from auradefi.api.deps import Deps app = create_app(Deps(tenancy=…, keys=…, ledger=…, webhooks=…, clock=…)) # POST /auth/token, POST /connections, GET /crypto/sync (Plaid's envelope) # GET /coverage, POST /webhooks/endpoints, POST /webhooks/…/replay ``` [`docs/books/`](docs/books) walks both surfaces executably and offline. ## What works today Coverage is published as data (rule #10). Every row below has an executable notebook under [`docs/books/`](docs/books) that runs offline and asserts its own outputs, plus a gate test under `tests/`. | Capability | Limits, and what proves it | |---|---| | `Quantity`/`Money`: exact at 10^77, four-field wire form, `raw` always a JSON string, strict wire grammar | [`02_money`](docs/books/02_money.ipynb) | | CAIP-2/CAIP-19 parse + canonicalize, deterministic `ast_…` ids, both-ways asset registry | 5 seed chains (Ethereum, Polygon, Base, Bitcoin, Solana); [`03_assets_chains`](docs/books/03_assets_chains.ipynb) | | Asset groups (decimals-equality law, `single` fallback) + additive spam scoring (score + numbers, caller threshold) | [`03_assets_chains`](docs/books/03_assets_chains.ipynb) | | Ledger port: idempotent upsert, cursor sync with `has_more` paging, reorg as `removed` + re-`added`, resurrection, tenant isolation | memory and SQLModel backends; [`04_ledger`](docs/books/04_ledger.ipynb) | | Cassette replay harness (`CassetteMissError` offline guarantee) | [`01_foundation`](docs/books/01_foundation.ipynb) | | Style gates: size, structure, placement, layering (`tests/style`) | no allowlist | | EVM balances to holdings, exact-`Decimal` USD totals, unpriced assets named | Etherscan V2 source + DefiLlama prices; [`05_holdings`](docs/books/05_holdings.ipynb) | | Tenancy: org/project/end-user, scoped `adk_` keys, `authEndpoint` JWT mint, three-window quota, audit log | the isolation gate actively tries to leak; [`06_tenancy`](docs/books/06_tenancy.ipynb) | | Rich transactions: `parts[]`/`acts[]`, fees as siblings carrying `borne_by`, derived `type`, ledger bridge, reorg + resurrection | EVM only, one act per transaction; [`07_transactions`](docs/books/07_transactions.ipynb) | | DeFi positions: adapter protocol, drill-down, group totals + health factor, signed synthetic-Holdings projection | Uniswap v2/v3, Aave v3, Lido/Rocket Pool; fixture-driven, see below; [`08_positions`](docs/books/08_positions.ipynb) | | Embedding: `from auradefi import Auradefi`, host-owned session, budgeted two-phase sync, 26-metric scalar projection | chain-scoped connection ids, restart resume enumerated from the state port, one connection's failure contained to its own row (0.1.1 #18/#21/#24/#26); [`09_embedding`](docs/books/09_embedding.ipynb), [`02_embed_in_your_backend.py`](examples/02_embed_in_your_backend.py) | | Bitcoin: pure-Python BIP32 xpub derivation, gap-20 scan, confirmed-only UTXO balances | p2wpkh + Esplora only; the extended key never reaches HTTP; [`10_bitcoin_solana`](docs/books/10_bitcoin_solana.ipynb) | | Solana: SPL + Token-2022 balances, ScaledUiAmount carried both ways, signature history | balances only, no decode; [`10_bitcoin_solana`](docs/books/10_bitcoin_solana.ipynb) | | HTTP API: Plaid `/crypto/sync` envelope, connections, `/coverage` generated as data, nine quota headers, batch holdings | [`12_http_api`](docs/books/12_http_api.ipynb) | | Webhooks: HMAC-SHA256 signed, durable over a pinned retry schedule, dead letter + replay | [`12_http_api`](docs/books/12_http_api.ipynb) | | Accounting: lot ledger, FIFO/LIFO/HIFO/ACB, realised + unrealised PnL, arbitrary-date PnL, Plaid `tax_lots` | 50,000-event gate; [`11_accounting`](docs/books/11_accounting.ipynb) | ### What is not there Rule #10 applies to the absences too. - There are no live network adapters beyond what the cassettes cover. Every I/O path is exercised against committed recordings. Pointing a source at the real Etherscan, Esplora or Solana RPC needs your own keys and endpoints, and CI has not reconciled the output against an incumbent. - Positions are fixture-driven. The `ContractReader` seam ships and every adapter is pinned to block-20,450,000 golden vectors, but no concrete on-chain reader ships. The package has no `eth_call` transport and no multicall batcher, so a host must supply its own reader to run the adapters against a live chain. - There is no multicall anywhere, so token balances cost one request each. - One price oracle (DefiLlama), current prices only. There is no fallback feed and no historical price service: `prices/historian.py` and `prices/store.py` are declared in the module layout and absent, as are the `coingecko`, `manual` and `onchain_amm` oracles, so accounting marks are the caller's. - No `jobs/` package. The layout declares `scheduler.py`, `discover.py`, `refresh.py`, `reprocess.py` and `backfill.py`; none of them ship. There is no scheduler, no background worker and no reprocess path. The host owns every tick, and a backfill is a `sync()` budget you spend yourself. - Five of the nine declared `api/routes/` modules are absent: `accounts.py`, `holdings.py`, `positions.py`, `transactions.py` and `webhooks.py`. What ships is `auth`, `connections`, `sync` and `admin`, so holdings and positions have no HTTP surface of their own, and the webhook admin routes live in `admin.py`. - `project/` ships only `scalar.py`. `project/plaid.py` and `project/native.py` are declared and absent. The Plaid envelope is projected in `api/wire.py` instead, so that projection is not reusable outside the HTTP shell the way the layer contract intends. - Two ledger backends: in-memory and SQLModel/sqlite. Postgres should work through the same port; only sqlite is exercised. The tenancy, keys, quota, audit and webhook stores are in-memory only. - Cosmos is absent, as is every EVM chain the registry does not seed. So are exchange connections, NFTs and protocol-specific decoders (`acts[]` is always one act, and `protocol` is always `None`). - Solana transaction decode is not implemented. Balances and signature history only. - No async surface, no background worker and no scheduler. The host owns the tick. - There is no migration for the 0.1.1 embed id break. Library-ingested embed connection ids, and every `transaction_id` hashed over them, re-derive in 0.1.1, so 0.1.0 rows written through `Auradefi` stop matching. A host either re-derives them itself or accepts the old rows as orphaned history (`CHANGELOG.md`, *Upgrading*). Data written through the HTTP API is unaffected. - `SyncStatePort` is a five-method Protocol in 0.1.1 (`tenants()` was added). A host store written against the 0.1.0 four-method shape is refused at bind time, so it cannot silently sync nothing. ## The rules the code lives by - Money is a tagged decimal string, and a raw amount is never a JSON integer. - Asset ids are deterministic CAIP-19 and permanently stable. - Every movement is a `part[]`; fees are siblings of movements. - Multi-tenancy is designed in. Two tenants can never see each other's data. - `pytest` passes on a fresh clone with no API keys, because the cassettes are committed. - Files cap at 400 lines with no allowlist, and `tests/style/` enforces the layer contract. ## Docker ```bash docker compose run --rm test # full offline suite in a network-less container docker compose run --rm demo # quickstart against the installed wheel ``` ## Docs **[auradefi.info](https://auradefi.info/)** is built from this repository, with every example executed at build time and every signature generated from the code. Start here: - [Quickstart](https://auradefi.info/quickstart.html): five lines, no credentials - [Authentication & keys](https://auradefi.info/authentication.html): what you need before mainnet (at most one key, and it is optional) - [Bring your own](https://auradefi.info/bring-your-own.html): your API, your database, your prices, with every port and its methods - [Guides](https://auradefi.info/examples/index.html): [`examples/`](examples), eleven single files that run offline - [API reference](https://auradefi.info/reference/index.html): signatures, parameters, return fields, exceptions - [HTTP API](https://auradefi.info/http.html): Plaid's wire format, plus `openapi.json` - [Errors](https://auradefi.info/errors.html): every exception and its HTTP status - [Build with an LLM](https://auradefi.info/llms.html): a prompt to paste into a model, plus [llms.txt](https://auradefi.info/llms.txt) and the whole documentation as [one file](https://auradefi.info/llms-full.txt) Also in the repository: - [`docs/books/`](docs/books) holds twelve executable notebooks, run headlessly in CI. - [`CHANGELOG.md`](CHANGELOG.md) records what changed per release, including the 0.1.1 upgrade note. - [`docs/internal/`](docs/internal) covers how this was designed and built rather than how to use it: the [design contract](docs/internal/SPEC.md), the [pinned algorithms](docs/internal/DECISIONS.md), [build status](docs/internal/STATUS.md), the [0.1.0 defect accounting](docs/internal/RELEASE_0.1.1.md), the [release procedure](docs/internal/RELEASING.md) and the [agent loop](docs/internal/AGENT_PROMPTS.md) that wrote most of it. ## Licence Apache-2.0. See [`LICENSE`](LICENSE). ======================================================================== # FILE: docs/authentication.md ======================================================================== # Authentication & keys You need at most one key, and even that one is optional. | Service | What it gives you | Key | Covers | |---|---|---|---| | Sandbox | Everything, recorded | none | one address, one chain, seven transactions | | Etherscan V2 | EVM balances + history | optional | one key, *every* `eip155:*` chain | | DefiLlama | USD prices | none, keyless | 6 EVM chains. No BTC or SOL prices at all | | Blockstream Esplora | Bitcoin UTXO balances | none, keyless | Bitcoin; the network is the base URL | | Solana JSON-RPC | SPL + Token-2022 balances | none for the public endpoint | mainnet-beta | | Webhooks | Delivery to *you* | n/a; we sign, you verify | your endpoints | There is no auradefi account, no dashboard and no credential of ours to obtain. Every key above belongs to a third party, and you bring it. ## Environment variables All configuration is read by `Settings.from_env()`, which `Auradefi.from_env()` calls. Copy [`.env.example`](https://github.com/auracarehq/auradefi/blob/main/.env.example) and fill in what you need. | Variable | Default | Meaning | |---|---|---| | `AURADEFI_ETHERSCAN_API_KEY` | none | Etherscan V2 key. Optional. | | `AURADEFI_HELIUS_API_KEY` | none | Parsed but not yet consumed. See Solana below. | | `AURADEFI_HTTP_TIMEOUT_S` | `10.0` | Timeout for clients the library builds for you. | | `AURADEFI_SYNC_MIN_INTERVAL_S` | `60` | Floor between two ticks for one connection. | | `AURADEFI_PROJECT_ID` | `embed` | Namespace for derived tenant ids. | | `AURADEFI_TRUSTED_PROXY_HOPS` | `0` | How many `X-Forwarded-For` hops *your* proxies add. | The `AURADEFI_` prefix is mandatory. A bare `ETHERSCAN_API_KEY` in your shell is ignored on purpose, so an unrelated variable can never silently become this library's credential. A test pins that behaviour. ## Etherscan V2 This is the only key worth setting. ```bash export AURADEFI_ETHERSCAN_API_KEY=… # https://etherscan.io/apis ``` It is optional. Without it the `apikey` parameter is omitted from the request entirely, rather than sent empty, and Etherscan's keyless tier applies. One key covers every EVM chain. The chain travels in the request as `chainid`, derived from the CAIP-2 id, so Ethereum, Polygon, Base and any other `eip155:N` Etherscan supports all use the same key. The free tier allows 3 requests per second and 100k per day. This package has no retry and no rate limiting anywhere, so a burst surfaces immediately as `SourceError` and you pace your own ticks with `sync(budget=…)`. Token balances cost one request each, because there is no multicall yet, which makes a wide address proportionally expensive. A wrong or revoked key is not a distinct error type. Etherscan answers HTTP 200 with `{"status": "0", "message": "NOTOK", "result": "Invalid API Key"}`, which surfaces as: ``` auradefi.errors.SourceError: etherscan balance error: message='NOTOK' result='Invalid API Key' ``` An empty history is a valid answer rather than an error. `status: "0"` with `"No transactions found"` is an empty page, because a fresh address is a valid address. ## DefiLlama There is no key to set. Two limits matter more than the credential does. It covers six chains: ERC-20 prices resolve on chain ids 1, 10, 56, 137, 8453 and 42161, and native coin prices resolve on the four ETH-native ones. It has no Bitcoin or Solana prices at all. Nothing in this package can price BTC or SOL. Those assets come back held but unpriced: listed in `report.holdings` with `price=None`, named in `report.unpriced`, and never counted as zero. To price them, bind your own `prices` port, described in [Bring your own](bring-your-own.html). ## Bitcoin Esplora needs no key. The base URL *is* the network selector: ```python Esplora(client) # mainnet Esplora(client, base_url="https://blockstream.info/testnet/api") ``` The thing to budget for here is request volume. A gap-20 scan of an empty wallet is about 40 requests, one per derived address, with no throttle. The extended public key never leaves your process. Every request carries a derived `bc1…` address, and the test suite asserts that against recorded traffic. ## Solana The public endpoint needs no key and is aggressively rate-limited upstream; a 429 surfaces as `SourceError: solana rpc HTTP 429`. For a keyed provider, pass the entire URL. `AURADEFI_HELIUS_API_KEY` is parsed by `Settings` and consumed by nothing, because the Helius adapter does not ship: ```python SolanaRpc(client, url="https://mainnet.helius-rpc.com/?api-key=…") ``` Solana transaction decode is not implemented. Balances and signature history only. ## Webhooks Here the direction is reversed: you hold no key of ours, because we sign and you verify. Each endpoint gets a secret, shown once at registration, and every delivery carries `X-Auradefi-Signature` (HMAC-SHA256 over `timestamp.body`) plus `X-Auradefi-Timestamp`. The verifier ships: ```python from auradefi.webhooks.sign import verify_signature verify_signature(secret, timestamp_ms, raw_body, signature, now_ms) ``` It compares in constant time and refuses a stale timestamp, which gives a captured request a shelf life. See [guide 09](examples/09_deliver_signed_webhooks.html). ## The HTTP API's own credentials If you run the [HTTP API](http.html), it has a second, unrelated credential model, and these credentials are yours to issue. `adk_live_…` and `adk_test_…` server keys are created by your backend and scoped (`users:admin`, `accounts:read`, `accounts:write`, `sync:trigger`). They are stored as hashes, so a database dump does not yield working credentials. Short-lived user tokens are minted from a server key for exactly one end user and signed with that project's secret. They are safe for a browser or a mobile app, and a token from one project can never verify under another's secret. See [guide 05](examples/05_serve_the_http_api.html) and [guide 06](examples/06_isolate_two_tenants.html). ## What `CassetteMissError` means If you are in Sandbox and see this, nothing is broken and no credential is missing: ``` auradefi.errors.CassetteMissError: GET https://api.etherscan.io/… is not recorded in sandbox.json. Recorded interactions: … ``` It means you asked for something the recording does not contain, usually a different address, chain or page size. Switch to `from_env()` with a real key, or ask for what the recording holds, which [Quickstart](quickstart.html#what-just-happened) lists. ======================================================================== # FILE: docs/bring-your-own.md ======================================================================== # Bring your own Yes: your own API, your own database, your own prices, your own clock. Every collaborator is a port, and a port is a plain object with one or two methods. There is no base class to inherit and no registration step, because the protocols are structural. Satisfying the shape *is* implementing them. The defaults exist so you do not have to start here. Replace one port and keep the rest: ```python aura = Auradefi.from_env() # all defaults aura = Auradefi.from_env(ledger=MyLedger()) # your database aura = Auradefi.from_env(source=MySource()) # your chain data aura = Auradefi.from_env(prices=MyPrices()) # your price feed aura = Auradefi(ledger=…, source=…, prices=…) # nothing of ours ``` | Port | Methods | Default | Replace it when | |---|---|---|---| | `source` | 2 | `EtherscanSource` | you have your own node, vendor or archive | | `prices` | 1 | DefiLlama via `Inquirer` | you need BTC/SOL, or your own marks | | `ledger` | 4 | `MemoryLedger` | always, in production: the default is not durable | | `sync_state` | 5 | `MemorySyncState` | you want cursors to survive a restart | | `clock` | 1 | `SystemClock` | you are testing, or replaying history | ## Your own database `ledger` is where transactions live. The shipped SQL backend takes a session factory instead of a URL, so that your application keeps ownership of the engine, the connection pool and the migrations. auradefi never opens a connection you did not hand it and never emits DDL, which is also why there is no `AURADEFI_DATABASE_URL` to set. ```python from sqlalchemy import create_engine from sqlmodel import Session from auradefi import Auradefi from auradefi.ledger.backends.models import metadata from auradefi.ledger.backends.sqlmodel import SqlModelLedger engine = create_engine("postgresql+psycopg://user@host/db") metadata.create_all(engine) # your migration, run once, by you aura = Auradefi.from_env( ledger=SqlModelLedger(session_factory=lambda: Session(engine)), ) ``` Install it with `pip install 'auradefi[sql]'`. Postgres and sqlite both go through the same port; only sqlite is exercised in CI. If you would rather own the DDL, [Database schema](schema.html) has both tables as plain SQL for Postgres and SQLite, ready for Alembic, Flyway or a reviewed migration. It also covers two hazards worth knowing about before you hand-write the schema. ### Or write the port yourself Four methods, all of them tenant-scoped. `tenant_id` is the first argument everywhere, and no call may read or write across tenants: ```python class MyLedger: def upsert(self, tenant_id, txns) -> list[SyncEvent]: ... def sync(self, tenant_id, cursor=None, limit=100) -> SyncPage: ... def get(self, tenant_id, txn_id) -> LedgerTransaction: ... def mark_removed(self, tenant_id, txn_ids) -> list[SyncEvent]: ... ``` Callers depend on three behaviours, so a replacement has to copy them: 1. `upsert` is idempotent. Re-ingesting an unchanged transaction emits no event, which is what makes a whole tick safe to retry. 2. A removed row that comes back is re-added rather than mutated: stored with `removed=False`, a bumped sequence, and an `ADDED` event. That is how a reorg stays expressible. 3. `sync` pages by last-modified order rather than by transaction date, so an old row that changes reappears at the end of the feed. Clients page until `has_more` is `False` before persisting the cursor. `get` for another tenant's id must raise `NotFoundError`, which is indistinguishable from a row that never existed. An id therefore cannot be used as an existence oracle across tenants. See [guide 04](examples/04_persist_to_your_database.html). ## Your own chain data `source` is one object with two methods. You may not have to write it, since `EtherscanSource` ships and `from_env()` binds it. ```python class MySource: def balances(self, chain_id: str, address: str) -> list[BalanceRecord]: """What the address holds NOW. Feeds holdings and pricing.""" def fetch_txlist(self, chain_id, address, *, start_block, end_block, page, offset, sort) -> list[dict]: """ONE page of raw history rows for exactly that window.""" ``` The engine owns the window. It chooses the blocks, the page number and the sort order, and it learns that a window has drained by receiving a page shorter than `offset`. Answer the window you were asked for: do not widen it, do not page internally, and do not retry silently. Returning everything at once defeats the budget, and returning an empty page early advances a cursor over data you never read. Rows come back raw, as `list[dict]`, because parsing belongs to the decoder seam. You can replace that too, via `decoder=`. To signal failure, raise `auradefi.errors.SourceError`, or any `AuradefiError`, and `sync()` will contain it to that one connection's report row. Anything else propagates, since a `KeyError` in your adapter is a bug and a loud tick is the better outcome. See [guide 03](examples/03_write_a_source_adapter.html). ## Your own prices One method. Returning nothing for an asset is allowed and is not an error: ```python class MyPrices: def usd_prices(self, caip19s) -> dict[str, Money]: return {asset_id: Money(Decimal("2500"), "USD"), …} ``` An asset you omit comes back held but unpriced: listed, named in `report.unpriced`, and never valued at zero. Bind this port if you need Bitcoin or Solana prices, which the default cannot provide at all. Use `Money` with exact `Decimal` amounts. A float reintroduces the drift this arithmetic exists to avoid. ## Your own cursor store `sync_state` holds connections and their sync cursors. The default is in-process, so a restart forgets every connection. The SQL-backed implementation is not written yet, and this is the port to bind if you want durable cursors before it lands. ```python class MyState: def get_state(self, tenant_id, connection_id) -> SyncState: ... def put_state(self, tenant_id, connection_id, state) -> None: ... def connections(self, tenant_id) -> tuple[ConnectionRecord, ...]: ... def add_connection(self, tenant_id, record) -> None: ... def tenants(self) -> tuple[str, ...]: ... ``` `tenants()` is the one method with no `tenant_id`, and it carries real weight: `sync()` enumerates its work from the store. A worker that read its tenant list from process memory would restart, find nothing, and report a cheerful `no_op` forever. That was a real defect (0.1.1 #21). ## Your own clock ```python class MyClock: def now_ms(self) -> int: ... ``` `FrozenClock(ms)` ships for tests and replays, and `SystemClock` is the default. Because time is a port, quota windows, sync throttling and `as_of_ms` are all testable without sleeping, and Sandbox can hand you reproducible answers. ## What is not pluggable Three edges, stated plainly. The chain registry is per-instance and mutable, so `register()` a chain and `connect_address` will accept it. The seeded set is five chains, and the decoder needs an entry to exist before a connection can be made. The decoder is replaceable via `decoder=`, but the shipped one handles EVM native txlist rows only. Position adapters need a `ContractReader` that you supply. No `eth_call` transport and no multicall ship in this package, which is the largest gap between working and working against mainnet. ======================================================================== # FILE: docs/schema.md ======================================================================== # Database schema There are two tables. Here they are as plain SQL, ready to paste into your own migration: - **[`ledger_postgresql.sql`](https://github.com/auracarehq/auradefi/blob/main/docs/schema/ledger_postgresql.sql)** - **[`ledger_sqlite.sql`](https://github.com/auracarehq/auradefi/blob/main/docs/schema/ledger_sqlite.sql)** Both are generated from the same `metadata` the library uses, and a style gate regenerates and diffs them, so they cannot drift from the code. Regenerate locally with `python scripts/emit_schema.py`. auradefi never emits DDL and never opens a connection you did not hand it. The schema is yours: apply it with Alembic, Flyway, Liquibase, Rails, Prisma, `psql -f`, or whatever reviews your migrations. If you would rather not, one call does it for you: ```python from auradefi.ledger.backends.models import metadata metadata.create_all(engine) # fine for a script; read the warning below ``` ## `auradefi_ledger_transactions` One row per transaction, per tenant. | Column | Type | Null | Meaning | |---|---|---|---| | `tenant_id` | `VARCHAR` | no | PK part 1. The `usr_…` id. Every query is scoped by it. | | `id` | `VARCHAR` | no | PK part 2. The derived `txn_…` id, stable across runs and backends. | | `chain_id` | `VARCHAR` | no | CAIP-2, e.g. `eip155:1`. | | `tx_hash` | `VARCHAR` | no | On-chain hash. Not unique on its own: one hash can touch several accounts. | | `account_id` | `VARCHAR` | no | Which connection this row was ingested for. | | `block_number` | `BIGINT` | yes | `NULL` while pending. | | `initiated_at` | `BIGINT` | no | Millisecond epoch. | | `confirmed_at` | `BIGINT` | yes | Millisecond epoch; `NULL` until confirmed. | | `entries_json` | `VARCHAR` | no | The movements, as canonical JSON. See below. | | `removed` | `BOOLEAN` | no | Reorg tombstone. A removed row is kept, never deleted. | | `last_modified_seq` | `BIGINT` | no | Cursor ordering. Indexed with `tenant_id`. | Index: `ix_auradefi_ledger_transactions_tenant_seq (tenant_id, last_modified_seq)`. `sync()` filters and orders on exactly that pair, so this index is what makes the cursor feed cheap. Keep it. ### `entries_json` Canonical JSON, sorted keys, no whitespace. Each entry looks like this: ```json [{"asset_id":"eip155:1/slip44:60","decimals":18,"direction":"in","raw":"1000000000000000000"}] ``` `raw` is a decimal-int string and never a JSON number. That choice is load-bearing: `json.loads("1e77")` yields a float that is wrong by about `10^60`. Because these rows live in your database, something other than auradefi may write them, so a numeric `raw` is rejected on read instead of being coerced. An error is easier to deal with than a plausible-looking wrong amount. `direction` is `in`, `out` or `self`. ## `auradefi_ledger_seqs` | Column | Type | Null | Meaning | |---|---|---|---| | `tenant_id` | `VARCHAR` | no | PK. | | `seq` | `BIGINT` | no | Monotonic counter for that tenant; first value is 1. | The cursor counter lives in the database rather than in the process, so a restart cannot hand out a sequence number twice. Allocation is documented single-writer: run one ingest worker per tenant, or hold a lock, until Postgres hardening lands. ## Two hazards Every numeric column is `BIGINT`, and it has to be. Python `int` maps to SQLAlchemy `Integer`, which is `int4` on PostgreSQL, and a millisecond epoch (`1_754_000_000_000`) overflows `int4` by 816 times. Until 0.1.2 these columns were `INTEGER`, which meant the SQL ledger could not work on PostgreSQL at all: the first insert would fail with `integer out of range`. SQLite never noticed, because its `INTEGER` affinity is already 8 bytes, and that is how a fully green test suite hid the bug. If you hand-write this schema, use 64-bit integers. `metadata` is the global `SQLModel.metadata`. If your application also uses SQLModel, its tables sit in the same registry, so: ```python metadata.create_all(engine) # creates OUR two tables AND all of yours ``` That is rarely what you want against a production database, and it is the best reason to apply the `.sql` files instead, since they create exactly two tables and touch nothing else. The `auradefi_` prefix on every table name exists for the same reason: these objects land in your database, beside your own. ## What is not here Only the ledger persists. Tenancy, API keys, quota counters, the audit log, webhook endpoints and deliveries, and embed sync-state are all in-memory in this release. They have no tables, and a restart forgets them. That is a real limitation of the release, not an omission from this page. If you need any of it durable, the ports are there: bind your own `sync_state` ([Bring your own](bring-your-own.html)), and keep tenancy in your own schema until a SQL backend for it ships. The audit log in particular is security-relevant and in-memory, so treat it accordingly. ======================================================================== # FILE: examples/README.md ======================================================================== # examples Start here: ```python pip install auradefi from auradefi import Auradefi aura = Auradefi.sandbox() # no keys, no network, no configuration for holding in aura.holdings()[0].holdings: print(holding.symbol, holding.quantity, holding.value) ``` That is a complete program. Sandbox replays a recording bundled inside the package, so you get working code before you hold any credential, and every layer above the transport is the production one. Once you have an Etherscan key, `Auradefi.from_env()` is the only line that changes. Sandbox data is a recording, so its numbers are constants: 5025 USD of holdings and seven transactions. Asking for anything it does not hold raises `CassetteMissError`, which names what it does hold. Ten task-shaped guides follow. Each is a single file that runs offline without keys, asserts its own output, and prints a readable trace. All of them are self-contained, so you can copy a file out, `pip install auradefi`, and run it. ```bash python examples/01_holdings_for_an_address.py bash scripts/run_examples.sh # all of them, from a clone ``` Each file's docstring opens with the question it answers and closes with the change that points it at real infrastructure. | Guide | What it covers | Needs | |---|---|---| | [auradefi in five lines, then the whole library in one file.](quickstart.py) | Every capability in one file, end to end. | core | | [How do I get a priced portfolio for one address?](01_holdings_for_an_address.py) | Exact `Decimal` totals, unpriced assets named instead of zeroed, the wire form, and the one line that points it at mainnet. | core | | [How do I run this inside my own backend, with my own database?](02_embed_in_your_backend.py) | Defaults first, then replacing one port at a time: budgeted sync on your tick, restart resume, one failure contained to one connection, and your own database. | core | | [How do I point this at MY chain data: my RPC, my vendor, my archive?](03_write_a_source_adapter.py) | For when the shipped `EtherscanSource` is not what you want: the two-method seam, the window the engine owns, and how to signal an upstream failure. | core | | [How do I store this in MY database, and stream changes to my clients?](04_persist_to_your_database.py) | Host-owned DDL, idempotent upsert, a resumable cursor feed, and a reorg emitted as `removed` then re-`added`. | `[sql]` | | [How do I expose this over HTTP, the way Plaid clients already expect?](05_serve_the_http_api.py) | Plaid's exact shape: token mint, connections, `/crypto/sync` paging, batch partial success, generated `/coverage`. | `[api]` | | [How do I serve many customers from one deployment without leaking?](06_isolate_two_tenants.py) | Derived tenant ids, project-signed tokens, scoped keys, and per-project quota, attacked four ways. | core | | [How do I get DeFi positions, an LP, a loan, and not lie about them?](07_read_defi_positions.py) | DeFi positions that still add up: raw quantities, one risk group, the projection invariant, re-pricing at zero chain reads. | core | | [How do I answer "what did they make, and what tax lots are open"?](08_report_cost_basis_and_pnl.py) | FIFO/LIFO/HIFO/ACB, any instant you ask about, Plaid `tax_lots[]`, and a visible rounding flag. | core | | [How do I get told when something changes, and trust what arrives?](09_deliver_signed_webhooks.py) | HMAC signing with the shipped verifier, a pinned retry schedule into a dead letter queue, and replay. | core | | [How do I handle a Bitcoin xpub and Solana's token zoo?](10_scan_bitcoin_and_solana.py) | Non-EVM chains: a Bitcoin xpub that never leaves the process, gap-limit scanning, and a Token-2022 mint that breaks `raw / 10**decimals`. | core | Install the extras with `pip install 'auradefi[sql]'` or `pip install 'auradefi[api]'`. `scripts/run_examples.sh` skips an example whose extra is absent and says so, instead of failing. ## How these relate to the rest of the docs - **examples/** answers "how do I do X", one file per task. You are here. - **[`docs/books/`](../docs/books)** holds twelve executable notebooks, one per capability. They go considerably deeper and are run headlessly in CI. CI executes every example here through `scripts/run_examples.sh`, so an example that stops working fails the build. ======================================================================== # FILE: .env.example ======================================================================== ``` # auradefi configuration: copy to .env and fill in what you need. # # cp .env.example .env # # NOTHING here is required. With no configuration at all you still get the # Sandbox environment, which replays a recording bundled in the package: # # from auradefi import Auradefi # aura = Auradefi.sandbox() # no keys, no network # # Every name is read by `Settings.from_env()`, which `Auradefi.from_env()` # calls. The `AURADEFI_` prefix is mandatory: a bare `ETHERSCAN_API_KEY` is # ignored on purpose, so an unrelated variable in your shell can never # silently become this library's credential. # --------------------------------------------------------------- chain data # Etherscan V2, for EVM balances and transaction history. # OPTIONAL: without it the keyless tier applies (slower, rate-limited # sooner). ONE key covers every eip155 chain (Ethereum, Polygon, Base and # the rest) because the chain travels in the request, not in the key. # Get one: https://etherscan.io/apis AURADEFI_ETHERSCAN_API_KEY= # --------------------------------------------------------------------- prices # There is no price key to set. Prices come from DefiLlama, which is # keyless, and cover six EVM chains (Ethereum, Optimism, BSC, Polygon, # Base, Arbitrum). Bitcoin and Solana assets have NO price source in this # package: they are reported held-but-unpriced rather than guessed at. # ------------------------------------------------------------------- Solana # Read by Settings and not yet consumed by any shipped code path: the # Helius adapter is declared in the spec and absent from the tree. To use a # keyed provider today, pass the whole URL: # SolanaRpc(client, url="https://mainnet.helius-rpc.com/?api-key=...") AURADEFI_HELIUS_API_KEY= # ------------------------------------------------------------------ tuning # HTTP timeout in seconds for clients this library builds for you. AURADEFI_HTTP_TIMEOUT_S=10.0 # The floor between two sync ticks for one connection. `sync()` inside this # window is a no-op that touches no transport, so a busy scheduler cannot # hammer an upstream. AURADEFI_SYNC_MIN_INTERVAL_S=60 # Namespace for derived tenant ids. Change it and every `usr_`/`conn_` id # derived under it changes too, so set it once, per environment, and leave # it alone. AURADEFI_PROJECT_ID=embed # How many rightmost X-Forwarded-For hops YOUR proxies append, and so how # far back a client IP may be trusted by the HTTP API. Defaults to 0: no # proxy is trusted and the socket peer is the only verified source, because # an audit row attributed to a caller-supplied header is permanently wrong. AURADEFI_TRUSTED_PROXY_HOPS=0 # ---------------------------------------------------------------- storage # There is deliberately NO database URL. The SQL ledger takes a session # factory because your application owns the engine, the pool and the # migrations. auradefi never opens a connection you did not hand it, and # never emits DDL. Wire it explicitly: # # from sqlalchemy import create_engine # from sqlmodel import Session # from auradefi.ledger.backends.models import metadata # from auradefi.ledger.backends.sqlmodel import SqlModelLedger # # engine = create_engine("postgresql+psycopg://user@host/db") # metadata.create_all(engine) # your migration, once # aura = Auradefi.from_env( # ledger=SqlModelLedger(session_factory=lambda: Session(engine))) # # Without that, `from_env()` stores in memory and loses everything on exit. ``` ======================================================================== # FILE: examples/quickstart.py ======================================================================== ```python """auradefi in five lines, then the whole library in one file. pip install auradefi python examples/quickstart.py No keys. No network. No configuration. `Auradefi.sandbox()` replays a recording bundled inside the package, and every layer above the transport is the production one: the same source, decoder, ledger and pricing a live instance uses. Sandbox data is a RECORDING, so the numbers here are constants, which is what makes them safe to assert. This file is also the smoke test CI, `scripts/release_check.sh` (against a freshly built wheel in a clean venv) and `docker run --network none` all execute, so nothing in it may depend on the repository. Each section maps to one SPEC phase, and to a guide that goes deeper: the five lines examples/01_holdings_for_an_address.py 0 money, chains, assets, ledger docs/books/01_foundation … 04_ledger 1 balances -> holdings examples/01, examples/04 2 tenancy and the token mint examples/06 3 transaction decode and reorg examples/04 4 DeFi positions examples/07 5 embedding in your backend examples/02, examples/03 6 Bitcoin xpub derivation examples/10 7 Solana Token-2022 examples/10 8 webhook signing examples/09 9 cost basis and PnL examples/08 """ from __future__ import annotations import json from decimal import Decimal import auradefi print(f"auradefi {auradefi.__version__}: sandbox quickstart, no keys\n") def section(title: str) -> None: print(f"\n--- {title} " + "-" * max(0, 62 - len(title))) # ===================================================== the whole ask, first section("a priced portfolio, in five lines") from auradefi import Auradefi aura = Auradefi.sandbox() for holding in aura.holdings()[0].holdings: print(f" {holding.symbol:>5} {str(holding.quantity):>4} @ {holding.price}" f" = {holding.value}") (sandbox_report,) = aura.holdings() assert str(sandbox_report.total_value) == "5025.000000000000000000 USD" print(f" total {sandbox_report.total_value}") print(" ^ that is the entire program. Everything below is detail.") # --------------------------------------------------------------- phase 0 section("phase 0: money is exact, and a raw amount is a string") from auradefi.money.decimal_json import quantity_to_wire from auradefi.money.fiat import Money from auradefi.money.quantity import Quantity huge = Quantity(10**77, 18) assert str(huge) == "1" + "0" * 59 # exact, never scientific notation wire = quantity_to_wire(Quantity(4878123456789012345678, 18)) assert isinstance(wire["raw"], str), "rule #2: raw is never a JSON number" assert wire["numeric"] == "4878.123456789012345678" print(f"10^77 at 18 decimals -> {str(huge)[:20]}… ({len(str(huge))} digits, exact)") print(f"wire form: {json.dumps(wire)}") from auradefi.assets.caip import canonical_caip19, parse_caip19 from auradefi.chains.registry import ChainRegistry chains = ChainRegistry() assert [chain.caip2 for chain in chains.chains()][:2] == [ "bip122:000000000019d6689c085ae165831e93", "eip155:1", ] mixed = "eip155:1/erc20:0xA0b86991c6218b36c1D19D4a2e9Eb0cE3606eB48" assert canonical_caip19(mixed) == mixed.lower() assert parse_caip19(mixed).namespace == "erc20" print(f"{len(chains.chains())} chains seeded; CAIP-19 canonicalised: {canonical_caip19(mixed)}") # --------------------------------------------------------------- phase 1 section("phase 1: balances + prices -> holdings, exactly") from auradefi.money.decimal_json import money_to_wire from auradefi.money.fiat import Money # The five lines above already did this. What matters is HOW the number is # built: exact `Decimal` throughout, and an asset nobody prices is named in # `report.unpriced` rather than valued at zero. for holding in sandbox_report.holdings: print(f" {holding.symbol:>5} {str(holding.quantity):>4} @ " f"{str(holding.price):>9} = {holding.value}") assert sandbox_report.total_value == Money(Decimal("5025"), "USD") assert sandbox_report.unpriced == () print(f" total {sandbox_report.total_value} (exact Decimal, never a float)") print(f" on the wire: {json.dumps(money_to_wire(sandbox_report.total_value))}") # The offline guarantee is a guarantee: an unrecorded request fails loudly # rather than reaching the network. from auradefi.errors import CassetteMissError from auradefi.sources import sandbox as recording try: recording.client().get("https://api.etherscan.io/v2/api?chainid=999") except CassetteMissError: print(" an unrecorded request is refused: sandbox cannot reach the network") # --------------------------------------------------------------- phase 2 section("phase 2: two tenants, and one cannot see the other") from auradefi.clock import FrozenClock from auradefi.errors import AuthError from auradefi.tenancy.audit import AuditLog from auradefi.tenancy.keys import ApiKeyStore from auradefi.tenancy.models import Environment, Scope from auradefi.tenancy.store import TenancyStore from auradefi.tenancy.tokens import verify_token clock = FrozenClock(1_767_225_600_000) tenancy = TenancyStore() org = tenancy.create_organisation("Acme", clock) project_a = tenancy.create_project(org.id, "tenant-a", Environment.LIVE, clock) project_b = tenancy.create_project(org.id, "tenant-b", Environment.LIVE, clock) key, plaintext = ApiKeyStore().issue( project_a.id, Environment.LIVE, (Scope.USERS_ADMIN,), clock ) assert plaintext.startswith("adk_live_") and len(plaintext) == 57 token = tenancy.mint_user_token( project_a.id, "host-user-1", ["accounts:read"], 600_000, "203.0.113.7", key.id, clock, AuditLog(), ) claims = verify_token(token, signing_secret=project_a.signing_secret, clock=clock) assert claims.project_id == project_a.id try: verify_token(token, signing_secret=project_b.signing_secret, clock=clock) except AuthError as exc: print(f" A's token under B's secret: {type(exc).__name__}: {exc}") print(f" minted {plaintext[:13]}… -> user token for {claims.external_user_id}, " f"scopes {claims.scopes}, ttl {(claims.exp - claims.iat) // 1000}s") # --------------------------------------------------------------- phase 3 section("phase 3: decode -> parts/fees, bridge -> ledger, reorg") from auradefi.decode.pipeline import decode_account from auradefi.ledger.backends.memory import MemoryLedger from auradefi.ledger.bridge import to_ledger_transaction from auradefi.ledger.models import SyncEventKind from auradefi.ledger.reorg import plan_reorg from auradefi.sources.evm.txlist import NormalTxRecord ME = "0x" + "11" * 20 def row(tx_hash: str, block: int, seconds: int) -> NormalTxRecord: return NormalTxRecord( tx_hash=tx_hash, block_number=block, time_stamp=seconds, from_address="0x" + "99" * 20, to_address=ME, value_wei=10**18, gas_used=21_000, gas_price_wei=10**10, is_error=False, ) HASH_A, HASH_B = "0x" + "aa" * 32, "0x" + "bb" * 32 rich = decode_account("eip155:1", "acct_1", ME, [row(HASH_A, 100, 1_700_000_000), row(HASH_B, 101, 1_700_000_100)], []) first = rich[0] assert [part.direction.value for part in first.parts] == ["in"] assert first.fees[0].borne_by.value == "counterparty" # the sender paid the gas assert to_ledger_transaction(first).entries[0].quantity == Quantity(10**18, 18) print(f" {first.id}: type={first.type.value} parts={len(first.parts)} " f"fees={len(first.fees)} (fee borne_by={first.fees[0].borne_by.value})") ledger = MemoryLedger() bridged = [to_ledger_transaction(txn) for txn in rich] ledger.upsert("tenant-a", bridged) page = ledger.sync("tenant-a", None) assert page.next_cursor == "00000000000000000002" and page.has_more is False reorged = to_ledger_transaction(decode_account( "eip155:1", "acct_1", ME, [row(HASH_B, 105, 1_700_000_500)], [] )[0]) events = ledger.apply_reorg("tenant-a", plan_reorg( [ledger.get("tenant-a", txn.id) for txn in bridged], [reorged], from_block=101 )) assert [event.kind for event in events] == [SyncEventKind.ADDED] delta = ledger.sync("tenant-a", page.next_cursor) print(f" reorg at block 101 -> " + ", ".join( f"{event.kind.value} {event.transaction.id[:12]}… (block {event.transaction.block_number})" for event in delta.events ) + f"; cursor {page.next_cursor} -> {delta.next_cursor}") # --------------------------------------------------------------- phase 4 section("phase 4: positions drill down, and the projection invariant") from auradefi.positions.drill import drill, project_to_synthetic_holdings from auradefi.positions.models import ( MetaType, Position, PositionKind, PositionType, ProtocolModule, Underlying, group_id_for, position_id, ) USDC = "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" ETH_ID = "eip155:1/slip44:60" USDC_ID = f"eip155:1/erc20:{USDC}" AAVE_POOL = "0x87870bca3f3fd6335c3f4ce8392d69350b4fa4e2" AWETH = "0x4d5f47fa6a74757f35c14fd3a6ef8e3c9bc514e8" group = group_id_for("aave-v3", "eip155:1", AAVE_POOL) # SPEC §6.3 verbatim: supply 10 ETH, borrow 5,000 USDC. ONE risk unit. positions = [ Position( id=position_id("aave-v3", "eip155:1", AWETH), adapter_id="aave-v3", chain_id="eip155:1", contract_address=AWETH, kind=PositionKind.APP_TOKEN, position_type=PositionType.DEPOSIT, protocol_module=ProtocolModule.LENDING, group_id=group, underlyings=(Underlying(ETH_ID, Quantity(10 * 10**18, 18), MetaType.SUPPLIED),), ), Position( id=position_id("aave-v3", "eip155:1", AAVE_POOL), adapter_id="aave-v3", chain_id="eip155:1", contract_address=AAVE_POOL, kind=PositionKind.CONTRACT_POSITION, position_type=PositionType.LOAN, protocol_module=ProtocolModule.LENDING, group_id=group, underlyings=(Underlying(USDC_ID, Quantity(5000 * 10**6, 6), MetaType.BORROWED),), ), ] prices = {ETH_ID: Money(Decimal("3584.17"), "USD"), USDC_ID: Money(Decimal("0.999839"), "USD")} drilled = drill(positions, prices) synthetic = project_to_synthetic_holdings(drilled) assert drilled.net_worth.amount == Decimal("30842.505") assert {holding.quantity for holding in synthetic} == {Decimal("10"), Decimal("-5000")} naive_sum = sum((holding.institution_value.amount for holding in synthetic), Decimal("0")) assert naive_sum == drilled.net_worth.amount # THE invariant (SPEC §6.3) SYMBOLS = {ETH_ID: "ETH", USDC_ID: "USDC"} print(f" gross {drilled.gross_assets} − debt {drilled.total_debt} = net {drilled.net_worth}") print(" synthetic holdings: " + ", ".join( f"{holding.quantity.normalize():f} {SYMBOLS[holding.asset_id]}" for holding in synthetic)) print(f" a Plaid-only client summing institution_value gets {naive_sum}: exactly the net worth") # --------------------------------------------------------------- phase 5 section("phase 5: embedding: your ports, your tick, your database") # `sandbox()` and `from_env()` differ by one line and nothing else: # # aura = Auradefi.from_env() # your Etherscan key # aura = Auradefi.from_env(ledger=MyLedger()) # + your database # # `sync(budget=N)` caps the source pages ONE call may spend; cursors make # the next call resume; calling it again inside # `settings.sync_min_interval_s` is a no-op that touches no transport. synced = aura.sync(budget=10) assert (synced.pages_fetched, synced.transactions_ingested) == (5, 7) assert aura.sync(budget=10).no_op is True assert synced.failed_connections == () print(f" sync: {synced.pages_fetched} pages, {synced.transactions_ingested} " f"transactions across {len(synced.connections)} connection(s)") print(f" immediate re-sync: no_op=True, zero requests") print(" one connection's failure lands in report.failed_connections, never") print(" in a lost tick: branch on it every time (examples/02)") metrics = {metric.name: metric.value for metric in aura.scalar_metrics()} assert len(metrics) == 26 print(f" 26 scalar metrics: portfolio_value_usd={metrics['portfolio_value_usd']}, " f"transaction_count={metrics['transaction_count']}") # --------------------------------------------------------------- phase 6/7 section("phase 6+7: Bitcoin derives locally; Solana can break raw/10^d") from auradefi.sources.bitcoin.xpub import derive_addresses from auradefi.sources.solana.spl import ( aggregate_by_mint, build_balances, parse_token_accounts, ) XPUB = ( "xpub661MyMwAqRbcFtXgS5sYJABqqG9YLmC4Q1Rdap9gSE8NqtwybGhePY2gZ29ESFjqJoC" "u1Rupje8YtGqsefD265TMg7usUDFdp6W1EGMcet8" ) addresses = derive_addresses(XPUB, "p2wpkh", 0, 0, 3) assert addresses[0] == "bc1qp5wfcq48h6d63wyy9qz0awtpfqwwv4sma86mhz" print(" BIP32 derived in-process: the extended key never goes near HTTP:") for index, address in enumerate(addresses): print(f" m/0/{index} {address}") T22_MINT = "ScaLedUiAmountMint22222222222222222222222222" OWNER = "9wFFyRfZBsuAha4YcuxcXLKwMxJR43S7fPfQLXMFxbAF" # A Token-2022 account whose mint carries a ScaledUiAmount multiplier of 2: # the node's displayed amount is NOT raw / 10**decimals. accounts = parse_token_accounts([{ "pubkey": "T22AcctC3", "account": {"data": {"program": "spl-token-2022", "parsed": {"type": "account", "info": { "mint": T22_MINT, "owner": OWNER, "state": "initialized", "extensions": [{"extension": "scaledUiAmountConfig", "state": {"multiplier": "2"}}], "tokenAmount": {"amount": "1000000000", "decimals": 9, "uiAmount": 2.0, "uiAmountString": "2"}, }}}}, }]) native, scaled = build_balances(3_500_000_000, aggregate_by_mint(accounts)) assert str(native.quantity) == "3.5" assert str(scaled.quantity) == "1" and scaled.ui_amount_string == "2" and scaled.scaled_ui print(f" Token-2022 ScaledUiAmount: raw/10^decimals = {scaled.quantity}, " f"node says {scaled.ui_amount_string}, scaled_ui={scaled.scaled_ui}: both carried") # --------------------------------------------------------------- phase 8 section("phase 8: webhooks are signed, and verification is shipped") from auradefi.webhooks.sign import sign, verify_signature secret = "ab" * 32 body = '{"type":"connection.created","data":{"connection_id":"conn_demo"}}' at_ms = 1_754_000_000_000 signature = sign(secret, at_ms, body) verify_signature(secret, at_ms, body, signature, at_ms) try: verify_signature(secret, at_ms, body + " ", signature, at_ms) except AuthError as exc: print(f" tampered body rejected: {exc}") print(f" X-Auradefi-Signature: {signature[:34]}…") # --------------------------------------------------------------- phase 9 section("phase 9: four costing methods, four legal answers") from auradefi.accounting.lots import AcquisitionEvent, DisposalEvent from auradefi.accounting.pnl import pnl_at DAY = 86_400_000 T0 = 1_700_000_000_000 trades = ( AcquisitionEvent(T0 + 0 * DAY, ETH_ID, Quantity(1, 0), Money(Decimal("10"), "USD"), "txn_b1"), AcquisitionEvent(T0 + 1 * DAY, ETH_ID, Quantity(1, 0), Money(Decimal("30"), "USD"), "txn_b2"), AcquisitionEvent(T0 + 2 * DAY, ETH_ID, Quantity(1, 0), Money(Decimal("26"), "USD"), "txn_b3"), DisposalEvent(T0 + 3 * DAY, ETH_ID, Quantity(1, 0), Money(Decimal("40"), "USD"), "txn_s1"), ) marks = {ETH_ID: Money(Decimal("50"), "USD")} realised = {method: pnl_at(trades, method, T0 + 3 * DAY, marks).realized for method in ("fifo", "lifo", "hifo", "acb")} assert [str(value) for value in realised.values()] == ["30 USD", "14 USD", "10 USD", "18 USD"] print(" bought at 10, 30, 26; sold one unit for 40:") for method, value in realised.items(): print(f" {method:<5} realised {value}") # Arbitrary date: one millisecond earlier, the sale has not happened yet. before = pnl_at(trades, "fifo", T0 + 3 * DAY - 1, marks) assert before.realized == Money(Decimal("0"), "USD") and len(before.open_lots) == 3 print(f" 1 ms before the sale: realised {before.realized}, {len(before.open_lots)} open lots: " "any instant is answerable, nothing is pre-computed") # --------------------------------------------------------- optional extras section("optional extras (installed only with [sql] / [api])") try: from sqlalchemy import create_engine from sqlalchemy.pool import StaticPool from sqlmodel import Session from auradefi.ledger.backends.models import metadata from auradefi.ledger.backends.sqlmodel import SqlModelLedger engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) metadata.create_all(engine) # the HOST's DDL. The library emits none sql_ledger = SqlModelLedger(session_factory=lambda: Session(engine)) sql_ledger.upsert("tenant-a", bridged) assert len(sql_ledger.sync("tenant-a", None).events) == 2 print(" [sql] SqlModelLedger round-tripped 2 rows through the host's sqlite") except ImportError: print(" [sql] not installed: skipped (pip install 'auradefi[sql]')") try: from fastapi.testclient import TestClient from auradefi.api.app import create_app from auradefi.api.deps import Deps from auradefi.tenancy.quota import QuotaCounter, QuotaLimits from auradefi.tenancy.tokens import RevocationSet from auradefi.webhooks.deliver import WebhookStore api_deps = Deps( tenancy=tenancy, keys=ApiKeyStore(), quota=QuotaCounter(QuotaLimits(1_000, 10_000, 100_000), clock), audit=AuditLog(), revocations=RevocationSet(), ledger=MemoryLedger(), webhooks=WebhookStore(), chains=chains, clock=clock, signing_secret_for={project_a.id: project_a.signing_secret}.get, capabilities={"eip155:1": frozenset({"balances", "transactions", "prices"})}, ) api = TestClient(create_app(api_deps)) coverage = api.get("/coverage").json() ethereum = next(row for row in coverage["chains"] if row["chain_id"] == "eip155:1") assert ethereum["capabilities"]["balances"] is True assert ethereum["capabilities"]["positions"] is False # generated, never prose print(f" [api] GET /coverage: {len(coverage['chains'])} chains, " f"eip155:1 -> {sorted(k for k, v in ethereum['capabilities'].items() if v)}") except ImportError: print(" [api] not installed: skipped (pip install 'auradefi[api]')") print("\nquickstart OK: nothing above touched the network") ``` ======================================================================== # FILE: examples/01_holdings_for_an_address.py ======================================================================== ```python """How do I get a priced portfolio for one address? pip install auradefi python examples/01_holdings_for_an_address.py No keys, no setup: this runs in the Sandbox environment, which replays a recording bundled in the package. Every layer above the transport is the production one, so what you see here is what live code does. The three things worth noticing are the whole design: * the total is exact. It is computed in `Decimal`, never a float: the comparison at the end shows the float answer already wrong at the 17th digit, and that error compounds across a portfolio; * an asset nobody will price is **held, listed and named** in `report.unpriced`, never valued at zero; * going live is one line: `Auradefi.from_env()` instead of `Auradefi.sandbox()`, with `AURADEFI_ETHERSCAN_API_KEY` in your environment. The last section does exactly that if you have a key set. """ from __future__ import annotations import os from decimal import Decimal from auradefi import Auradefi from auradefi.money.decimal_json import money_to_wire, quantity_to_wire from auradefi.money.fiat import Money from auradefi.money.quantity import Quantity # ------------------------------------------------------------- the whole ask aura = Auradefi.sandbox() (report,) = aura.holdings() print(f"holdings for {report.address} on {report.chain_id}\n") print(f" {'asset':<6}{'quantity':>12}{'price':>14} value") for holding in report.holdings: print(f" {holding.symbol:<6}{str(holding.quantity):>12}" f"{str(holding.price):>14} {holding.value}") print(f" {'TOTAL':<6}{'':>12}{'':>14} {report.total_value}") assert report.total_value == Money(Decimal("5025"), "USD") # ------------------------------------------------------------------ exactness # These sandbox numbers are round, so a float would survive them. 5025.0 is # 5025. That is exactly why the guarantee has to be structural rather than # lucky: below is a real wallet-sized balance, and the float answer is wrong # at the 17th digit before anything is even summed. whale = Quantity(4_878_123_456_789_012_345_678, 18) exact = whale.as_decimal() * Decimal("3584.17") lossy = Decimal(str(float(whale.as_decimal()) * 3584.17)) assert exact != lossy print(f"\n 4878.123456789012345678 ETH @ 3584.17") print(f" exact {exact}") print(f" float {lossy} <- 17 significant digits, then guesses") print(f" drift {abs(exact - lossy)} USD on ONE holding") # ----------------------------------------------------------- unpriced assets # Nothing in this recording is unpriced, so here is what it looks like when # something is: the asset stays in `holdings` with price=None and value=None, # and its id is named in `report.unpriced`. It is NEVER counted as zero, and # there is no price source for Bitcoin or Solana assets in this package at # all, so this is the normal case for them, not an edge case. print(f"\n unpriced: {report.unpriced or '(none in the sandbox recording)'}") for holding in report.holdings: if holding.price is None: print(f" {holding.symbol} held, not valued: {holding.quantity}") # -------------------------------------------------------------- on the wire # Rule #2: a raw amount is a tagged decimal STRING, never a JSON number, so # no JavaScript client can quietly round it. The lossy float rides alongside, # clearly labelled, for clients that only want to draw a chart. wire = quantity_to_wire(report.holdings[0].quantity) assert isinstance(wire["raw"], str) print(f"\n quantity on the wire: {wire}") print(f" total on the wire: {money_to_wire(report.total_value)}") # --------------------------------------------------------------- going live # One line different. The key is OPTIONAL, without it Etherscan's keyless # tier applies, and one key covers every eip155 chain. if os.environ.get("AURADEFI_ETHERSCAN_API_KEY"): live = Auradefi.from_env() user = live.user("your-opaque-user-id") user.connect_address("eip155:1", "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045") (live_report,) = live.holdings() print(f"\n LIVE: {live_report.total_value} across " f"{len(live_report.holdings)} assets, {len(live_report.unpriced)} unpriced") else: print("\n live: set AURADEFI_ETHERSCAN_API_KEY and this file will also") print(" run Auradefi.from_env() against mainnet. Same code below it.") print("\nOK: priced exactly, offline, with one line between here and mainnet.") ``` ======================================================================== # FILE: examples/02_embed_in_your_backend.py ======================================================================== ```python """How do I run this inside my own backend, with my own database? pip install auradefi python examples/02_embed_in_your_backend.py `Auradefi.from_env()` gives you working defaults; every collaborator is still a port you can replace one at a time. That is the whole shape: aura = Auradefi.from_env() # defaults aura = Auradefi.from_env(ledger=MyLedger()) # your database aura = Auradefi.from_env(prices=MyPrices()) # your price feed aura = Auradefi(ledger=…, source=…, prices=…) # nothing of ours This file runs in Sandbox so it needs no keys, and demonstrates the five things a host actually has to get right: 1. **validation at CONNECT time**, not on a background tick hours later; 2. a **budgeted** sync you call on your own schedule, and its throttle; 3. a **restart**: a new process over the same stores resumes stored work; 4. one dead chain **failing on its own row** instead of failing the tick; 5. **your database**: the three lines, and why it is not an env var. """ from __future__ import annotations from auradefi import Auradefi from auradefi.clock import FrozenClock from auradefi.embed import bootstrap from auradefi.errors import SourceError, UnknownChainError, ValidationError from auradefi.sources import sandbox as recording # The default port set, as a plain dict. This is what `sandbox()` and # `from_env()` hand to the constructor, and what makes an override a # one-keyword change rather than a fork of the wiring. clock = FrozenClock(recording.SANDBOX_NOW_MS) ports = {**bootstrap.sandbox_ports(), "clock": clock} aura = Auradefi(**ports) print("ports:", ", ".join(sorted(ports))) # ------------------------------------------- 1. validate at connect time user = aura.user("your-opaque-user-id") connection = user.connect_address(recording.SANDBOX_CHAIN, recording.SANDBOX_ADDRESS) print(f"\nconnected {connection.id}") for chain, address, expected in ( ("eip155:99999", recording.SANDBOX_ADDRESS, UnknownChainError), (recording.SANDBOX_CHAIN, "0xnope", ValidationError), (recording.SANDBOX_CHAIN, recording.SANDBOX_ADDRESS, Exception), # duplicate ): try: user.connect_address(chain, address) except expected as exc: print(f" refused now, not later: {type(exc).__name__}: {str(exc)[:58]}") # --------------------------------------------------- 2. sync on your tick # `budget` is the maximum source pages one call may spend. The cursor makes # the next call resume, so a tick is bounded and never loses its place. first = aura.sync(budget=2) print(f"\nsync(budget=2): {first.pages_fetched} pages, " f"{first.transactions_ingested} transactions, no_op={first.no_op}") print(f" same instant again: no_op={aura.sync(budget=2).no_op} " "(throttled by settings.sync_min_interval_s: zero requests)") # ----------------------------------------------------- 3. restart resume # A new process binds fresh objects over the SAME stores. Connections come # from the state port, not from process memory, so a restarted worker # resumes stored work instead of reporting a cheerful no-op forever. clock.advance(60_000) restarted = Auradefi(**ports) resumed = restarted.sync(budget=2) assert [row.connection_id for row in resumed.connections] == [connection.id] print(f"\nafter restart: enumerated {len(resumed.connections)} stored connection(s), " f"ingested {resumed.transactions_ingested} more") # --------------------------------------------- 4. one failure, one row class DeadChain: """Your source, but the RPC is down. Wrap, do not rewrite.""" def __init__(self, real: object) -> None: self._real = real def balances(self, chain_id: str, address: str): return self._real.balances(chain_id, address) def fetch_txlist(self, chain_id, address, **window): # Raise SourceError (any AuradefiError) for an upstream failure and # the library contains it to this ONE connection. Anything else, # a KeyError in your adapter. Propagates, because that is your bug. raise SourceError(f"{chain_id} RPC did not answer") clock.advance(60_000) degraded = Auradefi(**{**ports, "source": DeadChain(ports["source"])}).sync(budget=2) assert degraded.failed_connections == (connection.id,) print(f"\nRPC down: failed_connections={degraded.failed_connections}") print(" branch on report.failed_connections every tick: a partial failure") print(" can never hide behind an aggregate that reads like a clean one") # ------------------------------------------------------ 5. your database # Storage defaults to memory and says so. There is no AURADEFI_DATABASE_URL # on purpose: the SQL ledger takes a SESSION FACTORY because your # application owns the engine, the pool and the migrations: auradefi never # opens a connection you did not hand it, and never emits DDL. # # from sqlalchemy import create_engine # from sqlmodel import Session # from auradefi.ledger.backends.models import metadata # from auradefi.ledger.backends.sqlmodel import SqlModelLedger # # engine = create_engine("postgresql+psycopg://user@host/db") # metadata.create_all(engine) # your migration, run once # aura = Auradefi.from_env( # ledger=SqlModelLedger(session_factory=lambda: Session(engine))) # # See 04_persist_to_your_database.py for that running end to end, and # 03_write_a_source_adapter.py to replace the transport instead. print("\nOK: defaults to start with, ports to replace, your tick throughout.") ``` ======================================================================== # FILE: examples/03_write_a_source_adapter.py ======================================================================== ```python """How do I point this at MY chain data: my RPC, my vendor, my archive? pip install auradefi python examples/03_write_a_source_adapter.py **You may not have to.** `EtherscanSource` ships and satisfies both seams over one Etherscan V2 key, and `Auradefi.from_env()` binds it for you. This guide is for when you want something else behind it: an internal service, a vendor SDK, an archive node, a queue. A source is one object with two methods: balances(chain_id, address) -> list[BalanceRecord] what the address holds NOW. Feeds holdings and pricing. Fetch_txlist(chain_id, address, *, start_block, end_block, page, offset, sort) -> list[dict] ONE page of raw history rows for exactly that window. Both are structural (`typing.Protocol`), no base class, no registration, no import. Get the shape right and the facade accepts it; get it wrong and it refuses at BIND time rather than on a background tick. The rule that matters: **the engine owns the window.** It picks the blocks, the page and the sort order, and it learns that a window drained by getting a page shorter than `offset`. So answer the window you were asked for, never widen it, never page internally, never retry silently. """ from __future__ import annotations from decimal import Decimal from auradefi import Auradefi from auradefi.embed.sync import PageFetcher from auradefi.errors import SourceError, ValidationError from auradefi.ledger.backends.memory import MemoryLedger from auradefi.money.fiat import Money from auradefi.money.quantity import Quantity from auradefi.portfolio.holdings import BalanceSource from auradefi.sources.evm.etherscan import BalanceRecord ETH = "eip155:1/slip44:60" WALLET = "0x1111111111111111111111111111111111111111" class MySource: """Both seams over whatever you already have. This one is a dict.""" #: Pretend history: block -> the rows mined in it. HISTORY = {block: [{"hash": "0x" + f"{block:02x}" * 32, "blockNumber": str(block), "timeStamp": str(1_753_000_000 + block), "from": "0x" + "99" * 20, "to": WALLET, "value": "1000000000000000000", "gasUsed": "21000", "gasPrice": "10000000000", "isError": "0"}] for block in (100, 101, 102)} def __init__(self) -> None: self.windows: list[str] = [] def balances(self, chain_id: str, address: str) -> list[BalanceRecord]: return [BalanceRecord(caip19=ETH, symbol="ETH", quantity=Quantity(3 * 10**18, 18), contract_address=None)] def fetch_txlist(self, chain_id: str, address: str, *, start_block: int, end_block: int, page: int, offset: int, sort: str) -> list[dict]: self.windows.append(f"blocks {start_block}-{end_block} page={page} " f"offset={offset} sort={sort}") blocks = sorted(b for b in self.HISTORY if start_block <= b <= end_block) if sort == "desc": blocks.reverse() rows = [row for block in blocks for row in self.HISTORY[block]] # Answer THIS page of THIS window. The engine did the arithmetic. window = rows[(page - 1) * offset:page * offset] if not window and page == 1 and start_block == 0: raise SourceError("upstream said nothing at all") # never silent return window class MyPrices: """Your price feed: CAIP-19 ids in, `Money` out. Absent is allowed. An asset you cannot price comes back unpriced, never as zero.""" def usd_prices(self, caip19s): return {ETH: Money(Decimal("3000"), "USD")} # ---------------------------------------------------- the seams, checked early source = MySource() assert isinstance(source, BalanceSource) # has balances assert isinstance(source, PageFetcher) # has fetch_txlist print("seams satisfied:", [name for name in ("balances", "fetch_txlist")]) # A source missing a seam is refused HERE, not on tick one. try: Auradefi(MemoryLedger(), object(), MyPrices()) except ValidationError as exc: print(f" a bad source at bind time: {exc}") # ------------------------------------------------------------- drive it aura = Auradefi(MemoryLedger(), source, MyPrices(), sync_page_size=2) user = aura.user("user-42") user.connect_address("eip155:1", WALLET) report = aura.sync(budget=5) (holdings,) = aura.holdings() print("\nwhat the engine asked for, in order:") for index, window in enumerate(source.windows, start=1): print(f" {index}. {window}") print(f"\nsync: {report.pages_fetched} pages, {report.transactions_ingested} " f"transactions; holdings {holdings.total_value}") # The first request is always the cheapest possible one: a single-row probe # at connect time, so a dead endpoint or a bad key fails while your user is # still looking at the screen. assert source.windows[0].endswith("offset=1 sort=desc") print("\nnote the first window: offset=1: the connect-time liveness probe") # ------------------------------------------------------ failure, on purpose # Raise `SourceError` (or any `AuradefiError`) for an upstream problem and # `sync()` contains it to that one connection's report row. Anything else # propagates untouched, because a KeyError in your adapter is your bug and # hiding it would be worse than a loud tick. print("\nOK: two methods, and any data source you already have becomes one.") ``` ======================================================================== # FILE: examples/04_persist_to_your_database.py ======================================================================== ```python """How do I store this in MY database, and stream changes to my clients? pip install 'auradefi[sql]' python examples/04_persist_to_your_database.py The ledger is a port with four methods. This file uses the shipped SQLModel backend against a sqlite file you can open with any client afterwards, and shows the four properties a downstream consumer actually depends on: * **the host owns the schema.** The library emits no DDL and opens no connection: you create the tables and hand over a session factory; * **upsert is idempotent.** Re-ingesting the same transaction produces no second row and no second event; * **`sync(cursor)` is a Plaid-shaped cursor feed.** `added` / `removed` with a `next_cursor` and `has_more`, so a client can resume exactly where it stopped and never has to re-read history; * **a reorg is expressible.** A transaction that leaves the canonical chain is emitted as `removed`, and if it comes back it is re-`added` under the same id, never mutated in place, never quietly deleted. Every id here is derived, so two independent workers ingesting the same transaction write the same row. """ from __future__ import annotations import tempfile from pathlib import Path from sqlalchemy import create_engine from sqlmodel import Session from auradefi.errors import NotFoundError from auradefi.ledger.backends.models import metadata from auradefi.ledger.backends.sqlmodel import SqlModelLedger from auradefi.ledger.models import ( Direction, Entry, LedgerTransaction, SyncEventKind, transaction_id, ) from auradefi.ledger.reorg import plan_reorg from auradefi.money.quantity import Quantity CHAIN = "eip155:1" ETH = "eip155:1/slip44:60" ACCOUNT = "acct_main" ALICE, BOB = "usr_alice", "usr_bob" # two tenants, one database def transaction(index: int, block: int) -> LedgerTransaction: """One inbound 0.1 ETH transfer. The id is DERIVED, never assigned.""" tx_hash = "0x" + f"{index:02x}" * 32 return LedgerTransaction( id=transaction_id(CHAIN, tx_hash, ACCOUNT), chain_id=CHAIN, tx_hash=tx_hash, account_id=ACCOUNT, block_number=block, initiated_at=1_753_000_000_000 + index * 1_000, confirmed_at=1_753_000_000_500 + index * 1_000, entries=(Entry(asset_id=ETH, quantity=Quantity(10**17, 18), direction=Direction.IN),), ) with tempfile.TemporaryDirectory() as tmp: database = Path(tmp) / "host.db" # ------------------------------------------------ 1. your schema, your engine # `metadata.create_all` is the HOST calling it. In production this is # your Alembic migration; the library never runs DDL behind your back. engine = create_engine(f"sqlite:///{database}") metadata.create_all(engine) ledger = SqlModelLedger(session_factory=lambda: Session(engine)) print(f"tables created by the host: {', '.join(sorted(metadata.tables))}") # ------------------------------------------------------- 2. idempotent write batch = [transaction(index, 18_000_000 + index) for index in range(1, 4)] events = ledger.upsert(ALICE, batch) again = ledger.upsert(ALICE, batch) # the same tick runs twice assert [event.kind for event in events] == [SyncEventKind.ADDED] * 3 assert again == [], "a re-ingest is a no-op, not a duplicate" print(f"\nupsert: {len(events)} added, re-upsert: {len(again)} events " "(safe to retry a whole tick)") # -------------------------------------------------------- 3. the cursor feed seen, cursor, pages = [], None, 0 while True: page = ledger.sync(ALICE, cursor, limit=2) pages += 1 seen.extend((event.kind.value, event.transaction.id) for event in page.events) print(f" page {pages}: {len(page.events)} event(s), " f"next_cursor={page.next_cursor} has_more={page.has_more}") cursor = page.next_cursor if not page.has_more: break assert pages == 2 and len(seen) == 3 assert ledger.sync(ALICE, cursor).events == (), "a drained cursor returns nothing" print(f" drained in {pages} pages; the cursor is where a client resumes") # --------------------------------------------------------- 4. tenant isolation # Bob's ledger is empty. There is no filter to forget: the tenant id is # a parameter of every call, and asking for someone else's row raises. # Bob asking for Alice's transaction id gets the same answer as Bob # asking for a transaction that never existed: not found. The id is not # an existence oracle across tenants. assert ledger.sync(BOB).events == () try: ledger.get(BOB, batch[0].id) except NotFoundError as exc: print(f"\nBob asking for Alice's transaction: {type(exc).__name__}: {exc}") # ------------------------------------------------------------- 5. the reorg # Block 18,000,003 is re-mined and transaction 3 lands in a later block. # `plan_reorg` compares what we stored against what the chain now says, # from a block number down. canonical = [transaction(3, 18_000_009)] stored = [ledger.get(ALICE, txn.id) for txn in batch] plan = plan_reorg(stored, canonical, from_block=18_000_003) reorg_events = ledger.apply_reorg(ALICE, plan) delta = ledger.sync(ALICE, cursor) print("\nreorg from block 18,000,003:") for event in delta.events: print(f" {event.kind.value:<8} {event.transaction.id} " f"block={event.transaction.block_number} removed={event.transaction.removed}") # Same id, new block, and the row is NOT removed: it was re-added, which # is exactly what a client replaying the feed needs to see. assert [event.kind for event in reorg_events] == [SyncEventKind.ADDED] resurrected = ledger.get(ALICE, batch[2].id) assert resurrected.block_number == 18_000_009 and resurrected.removed is False print(f" transaction 3 kept its id {resurrected.id}: history stays " "addressable across a reorg") # A transaction that does NOT come back is emitted as removed, with the # row retained so the feed can carry the retraction. dropped = plan_reorg([ledger.get(ALICE, batch[1].id)], [], from_block=18_000_002) (removal,) = ledger.apply_reorg(ALICE, dropped) assert removal.kind is SyncEventKind.REMOVED assert ledger.get(ALICE, batch[1].id).removed is True print(f" transaction 2 orphaned -> {removal.kind.value}, row kept with removed=True") print(f"\nsqlite file: {database.name} " f"({database.stat().st_size} bytes, openable with any client)") # The same code against Postgres is one URL away: # engine = create_engine("postgresql+psycopg://user@host/db") # The port is exercised against sqlite in CI; Postgres should work through # it unchanged, and is not yet covered by a gate (README, *What is not there*). print("\nOK: your schema, your session, derived ids, a resumable feed.") ``` ======================================================================== # FILE: examples/05_serve_the_http_api.py ======================================================================== ```python """How do I expose this over HTTP, the way Plaid clients already expect? pip install 'auradefi[api]' python examples/05_serve_the_http_api.py `create_app(Deps(...))` returns a FastAPI app. It is a shell: it holds no state, opens no connections and invents no stores: you inject the same ports the library uses (`04_persist_to_your_database.py`) and the routes project them onto Plaid's wire format. This file drives the app with FastAPI's `TestClient` so it runs without a server, then shows the one-liner that serves it for real. The journey is the one a client actually makes: POST /auth/token server key -> short-lived user token POST /connections user token -> conn_… (409 names the existing one) GET /crypto/sync user token -> added/modified/removed + cursor POST /batch/holdings server key -> partial success, per-item errors GET /coverage public -> generated capability matrix Ingestion is NOT an HTTP concern: rows arrive in the ledger from your own worker calling the library (`02_embed_in_your_backend.py`). This file seeds them directly so the sync feed has something to page. """ from __future__ import annotations from decimal import Decimal from fastapi.testclient import TestClient from auradefi.api.app import create_app from auradefi.api.deps import Deps from auradefi.chains.registry import ChainRegistry from auradefi.clock import FrozenClock from auradefi.ledger.backends.memory import MemoryLedger from auradefi.ledger.models import Direction, Entry, LedgerTransaction, transaction_id from auradefi.money.fiat import Money from auradefi.money.quantity import Quantity from auradefi.portfolio.models import Holding, HoldingsReport from auradefi.tenancy.audit import AuditLog from auradefi.tenancy.keys import ApiKeyStore from auradefi.tenancy.models import Environment, Scope, end_user_id from auradefi.tenancy.quota import QuotaCounter, QuotaLimits from auradefi.tenancy.store import TenancyStore from auradefi.tenancy.tokens import RevocationSet from auradefi.webhooks.deliver import WebhookStore CHAIN, ETH = "eip155:1", "eip155:1/slip44:60" ADDRESS = "0xAAAAaaaaAAAAaaaaAAAAaaaaAAAAaaaaAAAAaaaa" NOW = 1_754_000_000_000 class StubHoldings: """Whatever answers `holdings(chain_id, address)`. Bind yours, or leave `Deps.holdings=None` and the batch route is not mounted at all. An unbound capability has no endpoint rather than a broken one.""" def holdings(self, chain_id: str, address: str) -> HoldingsReport: return HoldingsReport.assemble(address, chain_id, [ Holding(caip19=ETH, symbol="ETH", quantity=Quantity(2 * 10**18, 18), price=Money(Decimal("3000"), "USD"), value=Money(Decimal("6000"), "USD")), ], NOW) # --------------------------------------------------------- 1. build the app clock = FrozenClock(NOW) tenancy = TenancyStore() organisation = tenancy.create_organisation("acme", clock) project = tenancy.create_project(organisation.id, "main", Environment.LIVE, clock) ledger = MemoryLedger() deps = Deps( tenancy=tenancy, keys=ApiKeyStore(), quota=QuotaCounter(QuotaLimits(1_000, 10_000, 100_000), clock), audit=AuditLog(), revocations=RevocationSet(), ledger=ledger, webhooks=WebhookStore(), chains=ChainRegistry(), clock=clock, signing_secret_for={project.id: project.signing_secret}.get, holdings=StubHoldings(), # `/coverage` is generated from THIS, never from prose. Declare only # what you actually wired up. capabilities={CHAIN: frozenset({"balances", "transactions", "prices"})}, ) client = TestClient(create_app(deps)) print(f"project {project.id}: routes:") schema = client.get("/openapi.json").json() for path in sorted(schema["paths"]): for method in sorted(schema["paths"][path]): print(f" {method.upper():<5} {path}") # ------------------------------------------- 2. server key -> user token # The server key never leaves your backend. It mints a short-lived token # scoped to ONE end user, which is what a browser or mobile app may hold. key, secret = deps.keys.issue( project.id, Environment.LIVE, (Scope.USERS_ADMIN, Scope.ACCOUNTS_READ, Scope.ACCOUNTS_WRITE), clock, ) server_auth = {"Authorization": f"Bearer {secret}"} assert secret.startswith("adk_live_") minted = client.post("/auth/token", json={"external_user_id": "host-user-7"}, headers=server_auth) assert minted.status_code == 200 and list(minted.json()) == ["token"] user_auth = {"Authorization": f"Bearer {minted.json()['token']}"} quota_headers = {name.lower(): value for name, value in minted.headers.items() if name.lower().startswith("x-ratelimit")} assert len(quota_headers) == 9 # 3 windows x limit/remaining/reset print(f"\nminted a user token from {secret[:13]}…") print(f" quota headers: {len(quota_headers)} " f"(second/minute/day x limit/remaining/reset)") print(f" audit: {deps.audit.entries(project.id)[0].event} " f"by {deps.audit.entries(project.id)[0].key_id} " f"from {deps.audit.entries(project.id)[0].ip}") # ---------------------------------------------------- 3. connect a wallet created = client.post("/connections", json={"kind": "address", "descriptor": ADDRESS}, headers=user_auth) assert created.status_code == 201 connection_id = created.json()["id"] # The same wallet again, in different case, is a 409 that NAMES the # connection you already have, so a retrying client can carry on. conflict = client.post("/connections", json={"kind": "address", "descriptor": ADDRESS.lower()}, headers=user_auth) assert conflict.status_code == 409 assert conflict.json()["error"]["existing_connection_id"] == connection_id print(f"\nPOST /connections -> 201 {connection_id}") print(f" again -> {conflict.status_code} {conflict.json()['error']['message']}") # ------------------------------------------------------ 4. the sync feed # Seeded here; in production your worker wrote these rows. tenant = end_user_id(project.id, "host-user-7") ledger.upsert(tenant, [ LedgerTransaction( id=transaction_id(CHAIN, "0x" + f"{index:02x}" * 32, "acct_eth"), chain_id=CHAIN, tx_hash="0x" + f"{index:02x}" * 32, account_id="acct_eth", block_number=18_000_000 + index, initiated_at=NOW - 10_000 + index, confirmed_at=NOW - 9_000 + index, entries=(Entry(asset_id=ETH, quantity=Quantity(index * 10**17, 18), direction=Direction.IN),), ) for index in (1, 2, 3) ]) seen, cursor, pages = [], None, 0 while True: query = "/crypto/sync?limit=2" + (f"&cursor={cursor}" if cursor else "") page = client.get(query, headers=user_auth).json() pages += 1 assert set(page) == {"added", "modified", "removed", "next_cursor", "has_more"} seen += [row["transaction_id"] for row in page["added"]] cursor = page["next_cursor"] if not page["has_more"]: break assert (pages, len(seen)) == (2, 3) quantity = page["added"][0]["entries"][0]["quantity"] assert quantity == {"raw": "300000000000000000", "decimals": 18, "numeric": "0.3", "float": 0.3} print(f"\nGET /crypto/sync: {pages} pages, {len(seen)} transactions, cursor {cursor}") print(f" quantity on the wire: {quantity}") print(" `raw` is a STRING even here: a JS client cannot round it by accident") # A bad cursor is a 422 and costs the caller no quota: a client with a # hard-coded bad parameter cannot drain the project's daily window. bad = client.get("/crypto/sync?cursor=not-a-cursor", headers=user_auth) assert bad.status_code == 422 print(f" bad cursor -> {bad.status_code} {bad.json()['error']['type']}") # ------------------------------------------------------ 5. batch holdings batch = client.post("/batch/holdings", json={"items": [ {"chain": CHAIN, "address": ADDRESS}, {"chain": "eip155:99999", "address": ADDRESS}, # unknown chain ]}, headers=server_auth) assert batch.status_code == 200 # partial success items = batch.json()["items"] assert [item["status"] for item in items] == ["ok", "error"] assert items[0]["result"]["total_value"] == {"amount": "6000", "currency": "USD"} assert items[1]["error"]["type"] == "UnknownChainError" print(f"\nPOST /batch/holdings -> {batch.status_code}, items in request order: " f"{[item['status'] for item in items]}") print(f" item 0: {items[0]['result']['total_value']}") print(f" item 1: {items[1]['error']['type']}: one bad item never fails the request," "\n and index i of the response always answers index i of the request") # ---------------------------------------------------------- 6. coverage coverage = client.get("/coverage").json() # public, no auth row = next(entry for entry in coverage["chains"] if entry["chain_id"] == CHAIN) assert row["capabilities"] == {"balances": True, "transactions": True, "positions": False, "prices": True, "xpub": False} print(f"\nGET /coverage: {len(coverage['chains'])} chains, generated from Deps:") print(f" {CHAIN}: " + ", ".join(sorted(name for name, on in row["capabilities"].items() if on))) print(" (positions=False because no reader is bound: the matrix cannot flatter us)") # ------------------------------------------------------------- 7. serve it # In a file called `main.py`: # # from auradefi.api.app import create_app # from auradefi.api.deps import Deps # app = create_app(Deps(...)) # your stores, your clock # # uvicorn main:app --host 0.0.0.0 --port 8000 # # Behind a proxy that appends one X-Forwarded-For hop, pass # `trusted_proxy_hops=1`; it defaults to 0, so no caller can choose the IP # its own audit row records. print("\nOK: Plaid's envelope over your ports, and nothing else.") ``` ======================================================================== # FILE: examples/06_isolate_two_tenants.py ======================================================================== ```python """How do I serve many customers from one deployment without leaking? pip install auradefi python examples/06_isolate_two_tenants.py Multi-tenancy here is not a `WHERE` clause you must remember to write. The hierarchy is organisation -> project -> end user, and the tenant key is *derived*: `usr_…` is a hash over `project_id | external_user_id`. Two projects using the identical customer id, "user-1", say, cannot collide, because the project id is inside the hash. This file sets up two projects that are as similar as possible, same customer id, same wallet address, and then attacks the boundary between them four ways: 1. replay project A's user token against project B -> refused (signature) 2. use a token beyond the scopes it was minted with -> refused 3. use a token one millisecond after it expires -> refused 4. read the other project's audit log -> empty Then it shows what a caller legitimately gets: scoped keys, a short-lived token, and three quota windows they can see the state of. """ from __future__ import annotations from auradefi.clock import FrozenClock from auradefi.errors import AuthError, QuotaExceededError, ScopeError from auradefi.tenancy.audit import AuditLog from auradefi.tenancy.keys import ApiKeyStore from auradefi.tenancy.models import ConnectionKind, Environment, Scope, end_user_id from auradefi.tenancy.quota import QuotaCounter, QuotaLimits from auradefi.tenancy.store import TenancyStore from auradefi.tenancy.tokens import require_scope, verify_token CUSTOMER = "user-1" # the SAME id in both projects WALLET = "0x1111111111111111111111111111111111111111" clock = FrozenClock(1_767_225_600_000) tenancy = TenancyStore() keys = ApiKeyStore() audit = AuditLog() # ------------------------------------------------------ 1. two tenants organisation = tenancy.create_organisation("Acme", clock) alpha = tenancy.create_project(organisation.id, "alpha", Environment.LIVE, clock) beta = tenancy.create_project(organisation.id, "beta", Environment.LIVE, clock) alpha_user = tenancy.get_or_create_user(alpha.id, CUSTOMER, clock) beta_user = tenancy.get_or_create_user(beta.id, CUSTOMER, clock) # Get-or-create really is: the same external id gives the same row back. assert tenancy.get_or_create_user(alpha.id, CUSTOMER, clock).id == alpha_user.id # Same customer id, same everything else: different tenant, by derivation. assert alpha_user.id != beta_user.id assert alpha_user.id == end_user_id(alpha.id, CUSTOMER) print(f"customer {CUSTOMER!r} in two projects:") print(f" {alpha.id} -> {alpha_user.id}") print(f" {beta.id} -> {beta_user.id}") print(" the project id is INSIDE the hash, so the ids cannot collide") # Both connect the same wallet. Both connections are real, and distinct. alpha_connection = tenancy.create_connection( alpha.id, alpha_user.id, ConnectionKind.ADDRESS, WALLET, clock) beta_connection = tenancy.create_connection( beta.id, beta_user.id, ConnectionKind.ADDRESS, WALLET, clock) assert alpha_connection.id != beta_connection.id print(f"\nthe same wallet connected in both: {alpha_connection.id} vs " f"{beta_connection.id}") # ------------------------------------------ 2. keys are scoped, and per project alpha_key, alpha_secret = keys.issue( alpha.id, Environment.LIVE, (Scope.USERS_ADMIN, Scope.ACCOUNTS_READ), clock) beta_key, beta_secret = keys.issue( beta.id, Environment.LIVE, (Scope.ACCOUNTS_READ,), clock) # The secret is shown once. Only its hash is stored, so a database dump is # not a set of working credentials. assert alpha_secret.startswith("adk_live_") and len(alpha_secret) == 57 assert alpha_secret not in repr(alpha_key) authenticated = keys.authenticate(alpha_secret, clock) assert authenticated.project_id == alpha.id print(f"\nkey {alpha_secret[:13]}… authenticates to {authenticated.project_id} " f"with scopes {sorted(scope.value for scope in authenticated.scopes)}") # Beta's key was never granted users:admin, so it cannot mint tokens even # for its OWN users. Scope is checked, not assumed from possession. beta_authenticated = keys.authenticate(beta_secret, clock) assert Scope.USERS_ADMIN not in beta_authenticated.scopes print(f" beta's key holds {sorted(s.value for s in beta_authenticated.scopes)}: " "it cannot mint a user token at all") # ------------------------------------------------ 3. tokens are project-signed token = tenancy.mint_user_token( alpha.id, CUSTOMER, ["accounts:read"], ttl_ms=600_000, ip="203.0.113.7", key_id=alpha_key.id, clock=clock, audit=audit, ip_source="socket", ) claims = verify_token(token, signing_secret=alpha.signing_secret, clock=clock) assert (claims.project_id, claims.external_user_id) == (alpha.id, CUSTOMER) print(f"\nalpha minted a token for {claims.external_user_id}: " f"scopes {claims.scopes}, ttl {(claims.exp - claims.iat) // 1000}s") # ATTACK 1: replay alpha's token against beta's secret. try: verify_token(token, signing_secret=beta.signing_secret, clock=clock) raise AssertionError("a cross-project token must never verify") except AuthError as exc: print(f" replayed at beta: {type(exc).__name__}: {exc}") # ATTACK 2: use it beyond its scope. try: require_scope(claims, "accounts:write") raise AssertionError("a scope not granted must never pass") except ScopeError as exc: print(f" used to write: {type(exc).__name__}: {exc}") # ATTACK 3: use it after it expires. Time is a port, so this is testable. expired_clock = FrozenClock(claims.exp + 1) try: verify_token(token, signing_secret=alpha.signing_secret, clock=expired_clock) raise AssertionError("an expired token must never verify") except AuthError as exc: print(f" used 1 ms late: {type(exc).__name__}: {exc}") # ATTACK 4: read the other project's audit trail. Every mint is recorded, # under the project that did it, with the IP the SERVER observed: a caller # cannot choose the address its own permanent audit row records. (entry,) = audit.entries(alpha.id) assert audit.entries(beta.id) == () assert (entry.event, entry.key_id, entry.ip) == ("token.minted", alpha_key.id, "203.0.113.7") print(f"\naudit: alpha has {len(audit.entries(alpha.id))} entry " f"({entry.event} by {entry.key_id} from {entry.ip}), beta has " f"{len(audit.entries(beta.id))}") # ------------------------------------------------------ 4. quota, per project # Three windows at once. A project that burns its second does not touch its # day, and beta is not slowed down by alpha at all. quota = QuotaCounter(QuotaLimits(per_second=2, per_day=1_000, per_month=10_000), clock) quota.hit(alpha.id) quota.hit(alpha.id) try: quota.hit(alpha.id) raise AssertionError("the third hit in one second must be refused") except QuotaExceededError as exc: print(f"\nalpha's 3rd request this second: {type(exc).__name__}: {exc}") quota.hit(beta.id) # beta is unaffected by alpha's burst snapshot = quota.snapshot(alpha.id) print(" alpha's windows: " + ", ".join( f"{name} {window.remaining}/{window.limit} left" for name, window in sorted(snapshot.items()))) print(f" beta's second: {quota.snapshot(beta.id)['second'].remaining}/2 left: " "one tenant cannot spend another's budget") clock.advance(1_000) # a new second quota.hit(alpha.id) print(f" one second later alpha is servable again: " f"{quota.snapshot(alpha.id)['second'].remaining}/2 left") print("\nOK: derived tenant ids, project-signed tokens, scoped keys, " "per-project quota.") ``` ======================================================================== # FILE: examples/07_read_defi_positions.py ======================================================================== ```python """How do I get DeFi positions, an LP, a loan, and not lie about them? pip install auradefi python examples/07_read_defi_positions.py A wallet balance is one number. A DeFi position is a claim on other assets, sometimes negative, and the way most tools get it wrong is to flatten it too early. Here the shape is: adapter.discover() -> which contracts matter on this chain adapter.resolve() -> RAW positions: quantities, no prices at all drill(positions, prices) -> value them, group them, net them project_to_synthetic_holdings(drilled) -> the flat Plaid-shaped view Two properties this file proves rather than asserts in prose: * **the projection invariant.** A client that knows nothing about DeFi, sums `institution_value` over the flat holdings and trusts the number, gets EXACTLY the net worth, because debt projects as a negative *quantity* at a positive price, never a negative price; * **re-pricing costs zero chain reads.** Positions carry raw quantities, so a new price is a pure recomputation. Nothing to invalidate, nothing stale. Chain reads go through ONE seam: `call(address, fn, args)`. This file binds a dict of recorded answers, which is also how the shipped adapters are tested. **No concrete on-chain reader ships in the package**: there is no `eth_call` transport and no multicall batcher, so running these adapters against mainnet means writing that `call` yourself (README, *What is not there*). """ from __future__ import annotations from decimal import Decimal from auradefi.money.fiat import Money from auradefi.positions.adapters.lending.aave import AaveV3Adapter, Market from auradefi.positions.drill import drill, project_to_synthetic_holdings from auradefi.positions.models import MetaType, PositionKind, PositionType from auradefi.positions.protocol import ( ContractSet, DiscoveryContext, PositionAdapter, ResolveContext, ) from auradefi.positions.registry import AdapterRegistry from auradefi.positions.resolve import resolve_all CHAIN = "eip155:1" BLOCK = 20_450_000 WALLET = "0x00000000000000000000000000000000000a11ce" POOL = "0x87870bca3f3fd6335c3f4ce8392d69350b4fa4e2" AWETH, DEBT_WETH = "0x4d5f47fa6a74757f35c14fd3a6ef8e3c9bc514e8", "0xea51d7853eefb32b6ee06b1c12e6dcca88be0ffe" AUSDC, DEBT_USDC = "0x98c23e9d8f34fefb1b7bd6a91b7ff122f4e16f5c", "0x72e95b8931767c79ba4eee721354d6e99a61d004" ETH = "eip155:1/slip44:60" USDC = "eip155:1/erc20:0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" class RecordedReader: """The entire chain-read seam: `call(address, fn, args) -> value`. Put your `eth_call` + multicall behind this one method and every shipped adapter works. It is the only thing standing between these adapters and a live chain. """ def __init__(self, answers: dict) -> None: self._answers = dict(answers) self.calls: list[tuple] = [] def call(self, address: str, fn: str, args: tuple = ()) -> object: self.calls.append((address.lower(), fn, args)) return self._answers[(address.lower(), fn, args)] class MainnetAaveV3(AaveV3Adapter): """The shipped adapter, told which markets to look at.""" markets = (Market(AWETH, DEBT_WETH, ETH, 18), Market(AUSDC, DEBT_USDC, USDC, 6)) # SPEC §6.3's worked example: supply 10 ETH, borrow 5,000 USDC. ONE risk unit. reader = RecordedReader({ (AWETH, "balanceOf", (WALLET,)): 10 * 10**18, # 10 ETH supplied (DEBT_WETH, "balanceOf", (WALLET,)): 0, (AUSDC, "balanceOf", (WALLET,)): 0, (DEBT_USDC, "balanceOf", (WALLET,)): 5_000 * 10**6, # 5,000 USDC borrowed (POOL, "getUserAccountData", (WALLET,)): ( 3_584_250_000_000, 500_000_000_000, 2_367_400_000_000, 8250, 8000, 5_812_500_000_000_000_000, # …ltv, health factor ), }) adapter = MainnetAaveV3() assert isinstance(adapter, PositionAdapter) # a Protocol: no base class needed # ------------------------------------------------------- 1. discover + resolve contracts = adapter.discover(DiscoveryContext(chain_id=CHAIN, reader=reader)) supply, borrow = adapter.resolve( ResolveContext(chain_id=CHAIN, address=WALLET, reader=reader, block_number=BLOCK), contracts, ) # Raw quantities only. No price reached this layer, so nothing here can go # stale, and the sign of a debt lives in `meta_type`, not in the number. assert supply.position_type is PositionType.DEPOSIT assert borrow.position_type is PositionType.LOAN assert borrow.kind is PositionKind.CONTRACT_POSITION # no token to hold assert supply.underlyings[0].meta_type is MetaType.SUPPLIED assert borrow.underlyings[0].meta_type is MetaType.BORROWED assert borrow.underlyings[0].quantity.raw > 0 assert all(under.price is None for under in supply.underlyings + borrow.underlyings) # The two are ONE risk unit and say so with a shared group id. You cannot # show the collateral without the debt by accident. assert supply.group_id == borrow.group_id print(f"resolved 2 positions in group {supply.group_id}:") print(f" {supply.position_type.value:<8} {supply.underlyings[0].quantity} ETH " f"({supply.underlyings[0].meta_type.value})") print(f" {borrow.position_type.value:<8} {borrow.underlyings[0].quantity} USDC " f"({borrow.underlyings[0].meta_type.value})") print(f" health factor {supply.group_info.health_factor}, " f"ltv {supply.group_info.ltv}") # Four markets were probed; the two zero balances produced no position at all. probes = [call for call in reader.calls if call[1] == "balanceOf"] assert len(probes) == 4 print(f" {len(probes)} balance probes -> 2 positions (zero balances dropped)") # ------------------------------------------------------------ 2. value it prices = {ETH: Money(Decimal("3584.17"), "USD"), USDC: Money(Decimal("0.999839"), "USD")} drilled = drill([supply, borrow], prices) assert drilled.gross_assets.amount == Decimal("35841.70") assert drilled.total_debt.amount == Decimal("4999.195") assert drilled.net_worth.amount == Decimal("30842.505") print(f"\ngross {drilled.gross_assets}") print(f"debt {drilled.total_debt}") print(f"net {drilled.net_worth}") # --------------------------------------------------- 3. the flat projection holdings = project_to_synthetic_holdings(drilled) by_asset = {holding.asset_id: holding for holding in holdings} assert by_asset[ETH].quantity == Decimal("10") assert by_asset[USDC].quantity == Decimal("-5000") # the QUANTITY is negative assert by_asset[USDC].institution_price.amount > 0 # the PRICE never is naive_total = sum((holding.institution_value.amount for holding in holdings), Decimal("0")) assert naive_total == drilled.net_worth.amount print("\nflat holdings a Plaid-only client sees:") for holding in holdings: print(f" {str(holding.quantity):>7} @ {str(holding.institution_price):>13}" f" = {holding.institution_value}") print(f" {'sum':>7} {'':>13} {naive_total} == net worth: " "the invariant, by exact Decimal equality") # ------------------------------------------------- 4. re-price, zero reads reads_before = len(reader.calls) repriced = drill([supply, borrow], {**prices, ETH: Money(Decimal("3600"), "USD")}) rebuilt = sum((holding.institution_value.amount for holding in project_to_synthetic_holdings(repriced)), Decimal("0")) assert repriced.net_worth.amount == Decimal("31000.805") == rebuilt assert len(reader.calls) == reads_before print(f"\nETH re-priced at 3600 -> net {repriced.net_worth}, " f"{len(reader.calls) - reads_before} extra chain reads") # --------------------------------------------- 5. one broken adapter, contained # Register many adapters and resolve them together. A protocol whose ABI # changed under you fails on its own row: you get every other position plus # a named failure, never a half-built portfolio presented as whole. class BrokenAdapter: id = "some-protocol" chains = frozenset({CHAIN}) def discover(self, ctx): return ContractSet.empty() def resolve(self, ctx, contracts): raise RuntimeError("upstream ABI changed") registry = AdapterRegistry() registry.register(adapter) registry.register(BrokenAdapter()) outcome = resolve_all( registry.adapters(), ResolveContext(chain_id=CHAIN, address=WALLET, reader=reader, block_number=BLOCK), {"aave-v3": contracts, "some-protocol": ContractSet.empty()}, ) assert [position.id for position in outcome.positions] == [supply.id, borrow.id] assert [failure.adapter_id for failure in outcome.failures] == ["some-protocol"] print(f"\nresolve_all over {len(registry.adapters())} adapters: " f"{len(outcome.positions)} positions, failures=" f"{[(f.adapter_id, f.error) for f in outcome.failures]}") print("\nOK: raw positions, one grouping, and a flat view that still adds up.") ``` ======================================================================== # FILE: examples/08_report_cost_basis_and_pnl.py ======================================================================== ```python """How do I answer "what did they make, and what tax lots are open"? pip install auradefi python examples/08_report_cost_basis_and_pnl.py Cost basis is where crypto tools quietly disagree with each other. This package's position is that there is no single right answer. There are four legal ones, and the caller picks: fifo oldest lot first (default nearly everywhere) lifo newest lot first hifo most expensive first (minimises a gain) acb one pooled average cost (Canada) Everything is computed from an event stream, at whatever instant you ask about, so there is no pre-computed state to go stale and no "as of last night's batch". Ask about a millisecond before a sale and the sale has not happened yet. Also here: `tax_lots[]` in Plaid's shape, straight out of the report, and the flag that fires when a `Fraction` had to be rounded into `Money`. The one place exactness cannot survive contact with a currency. """ from __future__ import annotations from decimal import Decimal from auradefi.accounting.lots import AcquisitionEvent, DisposalEvent, derive_events from auradefi.accounting.pnl import pnl_at from auradefi.ledger.models import Direction, Entry, LedgerTransaction, transaction_id from auradefi.money.fiat import Money from auradefi.money.quantity import Quantity ETH = "eip155:1/slip44:60" DAY = 86_400_000 T0 = 1_700_000_000_000 MARKS = {ETH: Money(Decimal("50"), "USD")} # what it is worth today # Three buys and one sale: 1 unit at 10, 1 at 30, 1 at 26, then one sold at 40. TRADES = ( AcquisitionEvent(T0 + 0 * DAY, ETH, Quantity(1, 0), Money(Decimal("10"), "USD"), "txn_b1"), AcquisitionEvent(T0 + 1 * DAY, ETH, Quantity(1, 0), Money(Decimal("30"), "USD"), "txn_b2"), AcquisitionEvent(T0 + 2 * DAY, ETH, Quantity(1, 0), Money(Decimal("26"), "USD"), "txn_b3"), DisposalEvent(T0 + 3 * DAY, ETH, Quantity(1, 0), Money(Decimal("40"), "USD"), "txn_s1"), ) # ------------------------------------------------- 1. four methods, four answers print("bought 1 @ 10, 1 @ 30, 1 @ 26; sold 1 @ 40; mark today 50\n") print(f" {'method':<6}{'realised':>12}{'unrealised':>14} open lots") reports = {} for method in ("fifo", "lifo", "hifo", "acb"): report = pnl_at(TRADES, method, T0 + 3 * DAY, MARKS) reports[method] = report print(f" {method:<6}{str(report.realized):>12}{str(report.unrealized):>14}" f" {len(report.open_lots)}") # Each is right, under its own rule: FIFO sells the 10, LIFO the 26, HIFO the 30. assert [str(reports[method].realized) for method in ("fifo", "lifo", "hifo", "acb")] == [ "30 USD", "14 USD", "10 USD", "18 USD"] print("\n FIFO sold the 10 (gain 30), LIFO the 26 (14), HIFO the 30 (10);") print(" ACB pooled all three to an average 22 (18). Same trades, four legal answers.") # ------------------------------------------------------- 2. any instant, exactly # One millisecond before the sale, the sale has not happened. Nothing is # pre-aggregated, so this is a question you can always ask. before = pnl_at(TRADES, "fifo", T0 + 3 * DAY - 1, MARKS) assert before.realized == Money(Decimal("0"), "USD") assert len(before.open_lots) == 3 after = reports["fifo"] print(f"\n1 ms before the sale: realised {before.realized}, {len(before.open_lots)} open lots") print(f"1 ms after: realised {after.realized}, {len(after.open_lots)} open lots") # A year later with no further trades: realised is unchanged, unrealised moved. later = pnl_at(TRADES, "fifo", T0 + 400 * DAY, {ETH: Money(Decimal("80"), "USD")}) assert later.realized == after.realized assert later.unrealized.amount > after.unrealized.amount print(f"400 days later at 80: realised {later.realized} (unchanged), " f"unrealised {later.unrealized}") # --------------------------------------------------------- 3. Plaid's tax_lots # `open_lots` is already in Plaid's `tax_lots[]` shape, with a DETERMINISTIC # `institution_lot_id`, so the same acquisition reports the same lot id # across runs, across backends and across processes. print("\nopen lots (Plaid tax_lots[] shape, FIFO):") print(f" {'institution_lot_id':<26}{'qty':>5}{'bought at':>11}" f"{'cost basis':>13}{'value now':>12}") for lot in after.open_lots: print(f" {lot.institution_lot_id:<26}{str(lot.quantity):>5}" f"{str(lot.purchase_price):>11}{str(lot.cost_basis):>13}" f"{str(lot.current_value):>12}") repeated = pnl_at(TRADES, "fifo", T0 + 3 * DAY, MARKS) assert [lot.institution_lot_id for lot in repeated.open_lots] == [ lot.institution_lot_id for lot in after.open_lots] assert after.open_lots[0].position_type == "LONG" print(" ids are derived, so a second run reports the same lot ids") # ------------------------------------------------- 4. when rounding happens # A third of a unit has no exact decimal cost. The arithmetic stays in # `Fraction` until the last step, and the report SAYS it rounded rather than # hiding a cent. thirds = ( AcquisitionEvent(T0, ETH, Quantity(3, 0), Money(Decimal("10"), "USD"), "txn_t1"), DisposalEvent(T0 + DAY, ETH, Quantity(1, 0), Money(Decimal("5"), "USD"), "txn_t2"), ) rounded = pnl_at(thirds, "fifo", T0 + DAY, MARKS) print(f"\nsold 1 of a 3-unit lot bought for 10 (basis 10/3):") print(f" realised {rounded.realized} flags={sorted(rounded.flags)}") assert "rounded_basis" in rounded.flags, "a rounded boundary must be visible" # ---------------------------------------- 5. straight from your ledger rows # `derive_events` turns stored transactions into the event stream: IN is an # acquisition, OUT a disposal, SELF nothing (moving your own coins is not # income), and a reorged-away row nothing at all. Costs come out as None, # pricing is deliberately NOT the accounting layer's job, so you attach your # own marks and keep one source of truth for prices. def ledger_row(index: int, direction: Direction) -> LedgerTransaction: tx_hash = "0x" + f"{index:02x}" * 32 return LedgerTransaction( id=transaction_id("eip155:1", tx_hash, "acct_1"), chain_id="eip155:1", tx_hash=tx_hash, account_id="acct_1", block_number=18_000_000 + index, initiated_at=T0 + index * DAY, confirmed_at=T0 + index * DAY + 500, entries=(Entry(asset_id=ETH, quantity=Quantity(10**18, 18), direction=direction),), ) rows = [ledger_row(1, Direction.IN), ledger_row(2, Direction.OUT), ledger_row(3, Direction.SELF)] events = derive_events(rows) assert [type(event).__name__ for event in events] == ["AcquisitionEvent", "DisposalEvent"] assert events[0].cost is None and events[1].proceeds is None print(f"\nderive_events over {len(rows)} ledger rows -> " f"{[type(event).__name__.removesuffix('Event') for event in events]} " "(the SELF transfer is not income)") print("\nOK: four methods, any instant, deterministic lot ids, honest rounding.") ``` ======================================================================== # FILE: examples/09_deliver_signed_webhooks.py ======================================================================== ```python """How do I get told when something changes, and trust what arrives? pip install auradefi python examples/09_deliver_signed_webhooks.py Webhooks are the part of an integration that fails silently. Three things have to be true or you cannot rely on them, and this file exercises all three against a receiver it controls: * **signed.** HMAC-SHA256 over `timestamp.body` with a per-endpoint secret, compared in constant time. The verifier ships in the package, so the receiving side is not left to improvise it; * **durable.** A receiver that is down is retried on a schedule that is pinned in code, not "eventually", and ends in a dead letter queue you can list rather than in a log line nobody reads; * **replayable.** A dead letter can be re-sent as a NEW delivery row. The original is never mutated, so the history of what you attempted survives. Delivery is driven by `Deliverer.tick(now_ms)`: you call it from your own worker. There is no background thread in this package. """ from __future__ import annotations import httpx from auradefi.clock import FrozenClock from auradefi.errors import AuthError from auradefi.webhooks.deliver import Deliverer, WebhookStore from auradefi.webhooks.models import RETRY_SCHEDULE_MS, EventName from auradefi.webhooks.replay import replay from auradefi.webhooks.sign import sign, verify_signature PROJECT = "proj_demo" HOOK_URL = "https://hooks.example.com/auradefi" class Receiver: """The other end. Records what arrives and answers one status code.""" def __init__(self, status_code: int) -> None: self.status_code = status_code self.requests: list[httpx.Request] = [] def __call__(self, request: httpx.Request) -> httpx.Response: self.requests.append(request) return httpx.Response(self.status_code) def deliverer_for(store: WebhookStore, receiver: Receiver) -> Deliverer: """A Deliverer over an httpx client you own: here, a mock transport.""" return Deliverer(store, httpx.Client(transport=httpx.MockTransport(receiver))) clock = FrozenClock(1_754_000_000_000) store = WebhookStore() # ------------------------------------------------------- 1. register, once endpoint, secret = store.register_endpoint(PROJECT, HOOK_URL, clock=clock) assert len(secret) == 64 # The secret is returned once, here, and is not readable off the endpoint # afterwards: list your endpoints and it is absent. assert secret not in repr(endpoint) print(f"endpoint {endpoint.id} -> {endpoint.url}") print(f" secret shown once at creation ({len(secret)} hex chars), never again") # --------------------------------------------------------- 2. emit and sign clock.advance(1_000) queued_at = clock.now_ms() (delivery,) = store.emit(PROJECT, EventName.CONNECTION_CREATED, {"connection_id": "conn_demo"}, clock) assert delivery.status.value == "pending" event = store.get_event(PROJECT, delivery.event_id) print(f"\nqueued {delivery.id}: {event.name} " f"(attempts={delivery.attempts}, due at {delivery.next_attempt_at_ms})") receiver = Receiver(200) (delivered,) = deliverer_for(store, receiver).tick(queued_at) assert delivered.status.value == "delivered" (sent,) = receiver.requests body = sent.content.decode("utf-8") assert sent.method == "POST" and str(sent.url) == HOOK_URL assert sent.headers["X-Auradefi-Timestamp"] == str(queued_at) print(f" POST -> 200, headers X-Auradefi-Timestamp + X-Auradefi-Signature") print(f" body: {body[:88]}…") # --------------------------------------- 3. the receiving side, done properly # This is the code YOUR endpoint runs. It is four lines because the verifier # ships: constant-time compare, and the timestamp is inside the signed # preimage so a captured request cannot be replayed later. verify_signature(secret, queued_at, body, sent.headers["X-Auradefi-Signature"], clock.now_ms()) print(f"\nverified with the shipped verifier: {sent.headers['X-Auradefi-Signature'][:34]}…") for label, corruption in (("body altered by one space", (secret, queued_at, body + " ")), ("wrong secret", ("ff" * 32, queued_at, body)), ("timestamp moved", (secret, queued_at + 1, body))): try: verify_signature(*corruption, sent.headers["X-Auradefi-Signature"], clock.now_ms()) raise AssertionError(f"{label} must not verify") except AuthError as exc: print(f" {label:<26} -> {type(exc).__name__}: {exc}") # An old-but-genuine delivery is refused too: the timestamp is signed, so a # captured request has a shelf life. stale = sign(secret, queued_at, body) try: verify_signature(secret, queued_at, body, stale, queued_at + 10 * 60 * 1_000) raise AssertionError("a 10-minute-old signature must not verify") except AuthError as exc: print(f" {'10 minutes late':<26} -> {type(exc).__name__}: {exc}") # ------------------------------------------- 4. a receiver that is down # The schedule is pinned in `RETRY_SCHEDULE_MS`, so what happens next is a # fact you can plan around rather than a vendor behaviour you discover. clock.advance(1_000) born_at = clock.now_ms() (pending,) = store.emit(PROJECT, EventName.CONNECTION_CREATED, {"connection_id": "conn_unlucky"}, clock) down = Receiver(500) worker = deliverer_for(store, down) print(f"\nretry schedule (ms after queueing): {RETRY_SCHEDULE_MS}") for attempt, offset in enumerate(RETRY_SCHEDULE_MS, start=1): (row,) = worker.tick(born_at + offset) assert row.attempts == attempt and row.last_status_code == 500 print(f" attempt {attempt} at +{offset:>8} ms -> 500, " f"next at {row.next_attempt_at_ms}") assert len(down.requests) == len(RETRY_SCHEDULE_MS) assert row.status.value == "dead_letter" and row.next_attempt_at_ms is None (dead,) = store.dead_letter(PROJECT) assert dead.id == pending.id and dead.attempts == 6 print(f" {row.attempts} attempts over 24h -> dead_letter, and it is LISTED: " f"store.dead_letter() has {len(store.dead_letter(PROJECT))}") # -------------------------------------------------------------- 5. replay # A new row, a new id, the original left exactly as it was. clock.advance(1_000) replayed = replay(store, PROJECT, dead.id, clock) assert replayed.id != dead.id and replayed.replay_ordinal == 1 assert replayed.status.value == "pending" assert store.get_delivery(PROJECT, dead.id).status.value == "dead_letter" back_up = Receiver(200) (settled,) = deliverer_for(store, back_up).tick(clock.now_ms()) assert settled.id == replayed.id and settled.status.value == "delivered" print(f"\nreplayed {dead.id}") print(f" -> {settled.id} {settled.status.value} (replay_ordinal=" f"{replayed.replay_ordinal}); the dead row is still dead, on purpose") print("\nfinal delivery log: " + ", ".join( f"{row.id[:14]}…={row.status.value}" for row in store.deliveries(PROJECT))) # Over HTTP the same three operations are routes: register, list the dead # letter queue, replay one, so an operator does not need database access: # POST /webhooks/endpoints # GET /webhooks/dead_letter # POST /webhooks/deliveries/{id}/replay print("\nOK: signed, retried on a pinned schedule, dead-lettered, replayable.") ``` ======================================================================== # FILE: examples/10_scan_bitcoin_and_solana.py ======================================================================== ```python """How do I handle a Bitcoin xpub and Solana's token zoo? pip install auradefi python examples/10_scan_bitcoin_and_solana.py Two chains that are not EVM, and each breaks an assumption: **Bitcoin has no account.** One wallet is an unbounded set of addresses derived from an extended public key, and the balance is the sum over the used ones. BIP32 derivation here is pure Python, no `secp256k1` C library, no `bitcoinlib`, and, crucially, the xpub **never leaves the process**: every HTTP request carries a derived `bc1…` address, which this file asserts against the recorded traffic rather than promising in prose. The scan stops after `gap` consecutive unused addresses (BIP44's gap limit, 20 by default), and an address that was used and then swept still counts as used. **Solana can lie about `raw / 10**decimals`.** A Token-2022 mint carrying the ScaledUiAmount extension displays a multiple of the raw amount, so the identity every other chain relies on does not hold. Both numbers are carried, and the divergence is flagged: a wallet that only stores one of them shows the wrong balance and cannot tell. Both sections replay synthesised HTTP, so the file runs offline. """ from __future__ import annotations import functools import json import tempfile from pathlib import Path import httpx from auradefi.money.quantity import Quantity from auradefi.sources.bitcoin.esplora import Esplora, scan from auradefi.sources.bitcoin.xpub import derive_addresses, parse_xpub from auradefi.sources.solana.rpc import ( TOKEN_2022_PROGRAM, TOKEN_PROGRAM, SolanaBalances, SolanaRpc, ) from auradefi.testing.cassettes import load # BIP32 test vector 1's master public key: a published, keyless fixture. XPUB = ("xpub661MyMwAqRbcFtXgS5sYJABqqG9YLmC4Q1Rdap9gSE8NqtwybGhePY2gZ29ESFjqJoC" "u1Rupje8YtGqsefD265TMg7usUDFdp6W1EGMcet8") ESPLORA = "https://blockstream.info/api" GAP = 20 # ============================================================ bitcoin parsed = parse_xpub(XPUB) assert parsed.depth == 0 and len(parsed.chain_code) == 32 and len(parsed.pubkey) == 33 print(f"parsed xpub: depth={parsed.depth}, pubkey {parsed.pubkey.hex()[:20]}…") # `derive(chain, start, count)` is the seam `scan` drives. Binding the key # with partial() means the scanner never receives it. derive = functools.partial(derive_addresses, XPUB, "p2wpkh") external = derive(0, 0, 3) assert external[0] == "bc1qp5wfcq48h6d63wyy9qz0awtpfqwwv4sma86mhz" print("derived in-process, no network, no C dependency:") for index, address in enumerate(external): print(f" m/0/{index} {address}") # This wallet: two used receive addresses (one swept), one used change # address, everything else empty. `funded - spent` is the confirmed balance. FUNDED = { (0, 0): (100_000_000, 0, 3), # 1 BTC, never spent (0, 1): (60_000_000, 60_000_000, 2), # used and fully swept -> 0, still used (0, 2): (25_000, 0, 1), (1, 0): (999_000_000, 0, 12), # change } def address_row(chain: int, index: int) -> dict: """One `GET /address/{addr}` response. mempool_stats is ignored by design: an unconfirmed balance is not a balance.""" funded, spent, count = FUNDED.get((chain, index), (0, 0, 0)) address = derive(chain, index, 1)[0] return {"request": {"method": "GET", "url": f"{ESPLORA}/address/{address}"}, "response": {"status": 200, "json": { "address": address, "chain_stats": {"funded_txo_sum": funded, "spent_txo_sum": spent, "tx_count": count}, "mempool_stats": {"funded_txo_sum": 7_777, "spent_txo_sum": 0, "tx_count": 1}}}} # Exactly the addresses a gap-20 scan can reach: the used ones plus the 20 # empties that stop each chain. CASSETTE = {"interactions": [address_row(0, index) for index in range(23)] + [address_row(1, index) for index in range(21)]} with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "esplora.json" path.write_text(json.dumps(CASSETTE), encoding="utf-8") cassette = load(path) requested: list[str] = [] def recording(request: httpx.Request) -> httpx.Response: requested.append(str(request.url)) return cassette.handle(request) client = httpx.Client(transport=httpx.MockTransport(recording)) result = scan(Esplora(client, base_url=ESPLORA), derive, gap=GAP) # THE security property, asserted against the traffic: every request is a # derived address, and the xpub is in none of them. assert all(url.rsplit("/", 1)[-1].startswith("bc1") for url in requested) assert not any("xpub" in url for url in requested) assert len(requested) == 44 # 23 external + 21 change: the stop rule print(f"\n{len(requested)} address lookups; not one carried the extended key") for row in result.addresses: print(f" m/{row.chain}/{row.index} {row.address} " f"{row.balance_sats:>11} sats ({row.tx_count} tx)") assert [(row.chain, row.index, row.balance_sats) for row in result.addresses] == [ (0, 0, 100_000_000), (0, 1, 0), (0, 2, 25_000), (1, 0, 999_000_000)] assert result.total == Quantity(1_099_025_000, 8) and str(result.total) == "10.99025" print(f" {'TOTAL':>28} {result.total_sats:>17} sats = {result.total} BTC") print(f" asset id: {result.caip19}") print(" m/0/1 was swept to zero and is still reported: it is part of the wallet") # ============================================================= solana OWNER = "9wFFyRfZBsuAha4YcuxcXLKwMxJR43S7fPfQLXMFxbAF" USDC_MINT = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" T22_MINT = "ScaLedUiAmountMint22222222222222222222222222" RPC_URL = "https://api.mainnet-beta.solana.com" def token_account(pubkey: str, program: str, mint: str, amount: str, decimals: int, ui: str, extensions: list | None = None) -> dict: info = {"mint": mint, "owner": OWNER, "state": "initialized", "tokenAmount": {"amount": amount, "decimals": decimals, "uiAmount": float(ui), "uiAmountString": ui}} if extensions is not None: info["extensions"] = extensions return {"pubkey": pubkey, "account": {"data": {"program": program, "parsed": {"type": "account", "info": info}}}} def rpc_reply(method: str, params: list, result: object) -> dict: return {"request": {"method": "POST", "url": RPC_URL, "json": {"jsonrpc": "2.0", "id": 1, "method": method, "params": params}}, "response": {"status": 200, "json": {"jsonrpc": "2.0", "id": 1, "result": result}}} SOLANA = {"interactions": [ rpc_reply("getBalance", [OWNER], {"context": {"slot": 1}, "value": 3_500_000_000}), # Two SPL accounts of the SAME mint: they sum, they do not shadow. rpc_reply("getTokenAccountsByOwner", [OWNER, {"programId": TOKEN_PROGRAM}, {"encoding": "jsonParsed"}], {"context": {"slot": 1}, "value": [ token_account("Acct1", "spl-token", USDC_MINT, "250000000", 6, "250"), token_account("Acct2", "spl-token", USDC_MINT, "750000000", 6, "750"), ]}), # A Token-2022 mint with a x2 ScaledUiAmount multiplier. rpc_reply("getTokenAccountsByOwner", [OWNER, {"programId": TOKEN_2022_PROGRAM}, {"encoding": "jsonParsed"}], {"context": {"slot": 1}, "value": [ token_account("Acct3", "spl-token-2022", T22_MINT, "1000000000", 9, "2", extensions=[{"extension": "scaledUiAmountConfig", "state": {"multiplier": "2"}}]), ]}), ]} with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "solana.json" path.write_text(json.dumps(SOLANA), encoding="utf-8") posted: list[str] = [] solana_cassette = load(path) def recording_post(request: httpx.Request) -> httpx.Response: posted.append(json.loads(request.content)["method"]) return solana_cassette.handle(request) rpc = SolanaRpc(httpx.Client(transport=httpx.MockTransport(recording_post)), url=RPC_URL) balances = SolanaBalances(rpc).balances(OWNER) # Both token programs are asked, in a pinned order: a wallet that only knows # the original SPL program silently misses every Token-2022 balance. assert posted == ["getBalance", "getTokenAccountsByOwner", "getTokenAccountsByOwner"] native, usdc, scaled = balances print(f"\n{len(posted)} RPC calls -> {len(balances)} balances") print(f" {'raw/10^decimals':>16}{'node says':>12} scaled_ui asset") for balance in balances: print(f" {str(balance.quantity):>16}{balance.ui_amount_string:>12}" f" {str(balance.scaled_ui):<9} {balance.caip19.split('/')[-1][:34]}") assert str(native.quantity) == "3.5" and native.mint is None assert usdc.quantity == Quantity(1_000_000_000, 6) # 250M + 750M, summed assert usdc.ui_amount_string == "1000" and usdc.scaled_ui is False # The interesting row: 1 by the usual identity, 2 by the mint's own rule. assert str(scaled.quantity) == "1" and scaled.ui_amount_string == "2" assert scaled.scaled_ui is True print("\n two token accounts of one mint summed to " f"{usdc.quantity.as_decimal():f} USDC") print(f" Token-2022: raw/10^9 = {scaled.quantity}, the mint displays " f"{scaled.ui_amount_string}, scaled_ui={scaled.scaled_ui}") print(" both numbers are carried, so a caller can never quietly show the wrong one") # Solana transaction DECODE is not implemented: balances and signature # history only (README, *What is not there*). `rpc.get_signatures(address)` # pages history if you want to build on it. print("\nOK: an xpub that never left the process, and a token that breaks the identity.") ``` ======================================================================== # API REFERENCE (generated from the code) ======================================================================== ## Getting started The two factories and the object they return. ### Auradefi (auradefi.embed.facade) The library's public surface (SPEC §8). `source` must structurally satisfy BOTH `portfolio.holdings.BalanceSource` (`balances`, for holdings) and `embed.sync.PageFetcher` (`fetch_txlist`, for history): one object, two seams, so a host writes one adapter. __init__(self, ledger: LedgerPort, source: BalanceSource | PageFetcher, prices: PriceOracle, clock: Clock | None = None, settings: Settings | None = None, *, sync_state: SyncStatePort | None = None, decoder: Decoder | None = None, sync_page_size: int = 1000) -> None Bind the host's ports. ZERO I/O happens here. sandbox(cls, *, connect: bool = True, **overrides: object) -> Auradefi A working instance over a bundled recording, no keys, no network. from_env(cls, **overrides: object) -> Auradefi A live instance wired from the environment (SPEC §8). user(self, external_user_id: str) -> UserHandle Get-or-create the handle for one opaque host user id. sync(self, budget: int = 5) -> SyncReport One tick across every known connection, in creation order. holdings(self) -> tuple[HoldingsReport, ...] One priced `HoldingsReport` per connection, creation order. scalar_metrics(self) -> tuple[scalar_projection.Metric, ...] `(name, ms, float)` triples per connection, concatenated. ### UserHandle (auradefi.embed.handle) One host user's slice of the library (SPEC §7.1, §8). Created by `user`; `tenant_id` is derived, never supplied. Every operation is scoped to that tenant (rule #6). __init__(self, facade: Auradefi, external_user_id: str, tenant_id: str) -> None Bind the facade and the derived tenant id. No I/O. connect_address(self, chain: str, address: str) -> ConnectionRecord Watch one address on one chain: validated NOW, not later. connections(self) -> tuple[ConnectionRecord, ...] This tenant's connections, in creation order. sync(self, budget: int = 5) -> SyncReport `sync` restricted to THIS user's connections. holdings(self) -> tuple[HoldingsReport, ...] `holdings` for THIS user's connections. scalar_metrics(self) -> tuple[scalar_projection.Metric, ...] `scalar_metrics` for THIS user's connections. ### Settings (auradefi.config) Settings(etherscan_api_key: 'str | None' = None, helius_api_key: 'str | None' = None, http_timeout_s: 'float' = 10.0, sync_min_interval_s: 'int' = 60, project_id: 'str' = 'embed', trusted_proxy_hops: 'int' = 0) Fields: etherscan_api_key: str | None, helius_api_key: str | None, http_timeout_s: float, sync_min_interval_s: int, project_id: str, trusted_proxy_hops: int from_env(cls, env: Mapping[str, str] | None = None) -> "Settings" __init__(self, etherscan_api_key: str | None = None, helius_api_key: str | None = None, http_timeout_s: float = 10.0, sync_min_interval_s: int = 60, project_id: str = embed, trusted_proxy_hops: int = 0) -> None ## Ports you implement Bring your own: each is a structural Protocol. ### LedgerPort (auradefi.ledger.port) Structural contract for ledger persistence backends. Tenant-scoped throughout (rule #6). Sync events are ordered by ascending last-modified sequence, SPEC §6.4: last-modified order, NOT transaction date, and clients page until `has_more` is `False` before persisting the cursor. upsert(self, tenant_id: str, txns: Sequence[LedgerTransaction]) -> list[SyncEvent] Insert or update transactions inside one tenant's ledger. sync(self, tenant_id: str, cursor: str | None = None, limit: int = 100) -> SyncPage Page of changes for one tenant since `cursor` (SPEC §6.4). get(self, tenant_id: str, txn_id: str) -> LedgerTransaction Fetch one transaction within the caller's tenant scope. mark_removed(self, tenant_id: str, txn_ids: Sequence[str]) -> list[SyncEvent] Mark transactions removed (reorg semantics) inside one tenant. __init__(self, *args, **kwargs) ### SyncStatePort (auradefi.embed.state) Structural contract for embed sync-state persistence. Tenant-scoped throughout (rule #6): `tenant_id` is the first argument of every scoped method, and no call may read or write across tenants. `tenants` is the one enumeration, and it yields ids only, never another tenant's records. get_state(self, tenant_id: str, connection_id: str) -> SyncState Return the stored `SyncState` for one connection. put_state(self, tenant_id: str, connection_id: str, state: SyncState) -> None Store `state` for one connection; last write wins. connections(self, tenant_id: str) -> tuple[ConnectionRecord, ...] All of one tenant's connection records, in creation order. add_connection(self, tenant_id: str, record: ConnectionRecord) -> None Register a connection record under one tenant. tenants(self) -> tuple[str, ...] Every tenant this store holds anything for, first-seen order. __init__(self, *args, **kwargs) ### BalanceSource (auradefi.portfolio.holdings) Structural seam: typed balances for one (chain × address). Any object with a conforming `balances` method is a source. `EtherscanV2` conforms WITHOUT this module importing anything from it beyond the `BalanceRecord` record type. balances(self, chain_id: str, address: str) -> Sequence[BalanceRecord] Return typed balance records for `address` on `chain_id`. __init__(self, *args, **kwargs) ### PageFetcher (auradefi.embed.sync) Structural seam: ONE page of raw explorer rows for one window. A host satisfies it by shape (rule #12), no base class, no registration. Rows are the explorer's RAW dicts (Etherscan txlist rows); parsing is the decoder's job, not this seam's. fetch_txlist(self, chain_id: str, address: str, *, start_block: int, end_block: int, page: int, offset: int, sort: str) -> list[dict] Raw rows in `[start_block, end_block]`, `sort` asc|desc. __init__(self, *args, **kwargs) ### PriceOracle (auradefi.prices.inquirer) Structural interface: current USD prices for CAIP-19 asset ids. usd_prices(self, caip19s: Sequence[str]) -> dict[str, Money] Return USD prices for the ids this oracle can price. __init__(self, *args, **kwargs) ## Ports we ship Defaults, so you only implement what you want to. ### EtherscanSource (auradefi.sources.evm.source) Both source seams over one Etherscan V2 client. `balances(chain_id, address)` -> `list[BalanceRecord]` and `fetch_txlist(chain_id, address, *, start_block, end_block, page, offset, sort)` -> `list[dict]` of RAW rows for the decoder seam. The client is injected: this constructor performs no I/O and opens no connection. `from_key` is the convenience that builds one. __init__(self, client: httpx.Client, api_key: str | None = None, base_url: str = https://api.etherscan.io/v2/api, page_size: int = 1000) -> None Bind the client and credentials. ZERO I/O happens here. from_key(cls, api_key: str | None = None, *, timeout_s: float = 10.0, base_url: str = https://api.etherscan.io/v2/api, page_size: int = 1000) -> EtherscanSource Build a source owning its own `httpx.Client`. balances(self, chain_id: str, address: str) -> list[BalanceRecord] What `address` holds on `chain_id` now (SPEC §6.1). fetch_txlist(self, chain_id: str, address: str, *, start_block: int, end_block: int, page: int, offset: int, sort: str) -> list[dict] One page of raw history rows for exactly the window asked for. ### EtherscanV2 (auradefi.sources.evm.etherscan) Etherscan V2 balance source over an injected `httpx.Client`. The client is REQUIRED and injected so cassettes plug in; the constructor performs no I/O. `api_key=None` omits the `apikey` query param entirely. __init__(self, client: httpx.Client, api_key: str | None = None, base_url: str = https://api.etherscan.io/v2/api, page_size: int = 1000) -> None Bind the injected client and request parameters. No I/O. balances(self, chain_id: str, address: str) -> list[BalanceRecord] All non-zero balances for `address` on `chain_id`. ### DefiLlamaOracle (auradefi.prices.oracles.defillama) Current USD prices from DefiLlama's keyless `coins.llama.fi`. `client` is REQUIRED and injected: the oracle never constructs a transport of its own, never retries, never rate-limits. Structurally a `prices.inquirer.PriceOracle`; does not import it. __init__(self, client: httpx.Client, base_url: str = https://coins.llama.fi) -> None Bind the injected client and base URL. Performs no I/O. usd_prices(self, caip19s: Sequence[str]) -> dict[str, Money] Current USD price for each priceable input CAIP-19. ### Inquirer (auradefi.prices.inquirer) First-wins USD price aggregation over an ordered oracle sequence. __init__(self, oracles: Sequence[PriceOracle]) -> None Hold `oracles`; query order is construction order. usd_prices(self, caip19s: Sequence[str]) -> dict[str, Money] Merged first-wins USD prices for `caip19s`. ### MemoryLedger (auradefi.ledger.backends.memory) Dict-backed `LedgerPort` with hard per-tenant isolation. Constructed empty with no arguments: `MemoryLedger()`, no tenants, every per-tenant seq counter starting from 0 (first write gets 1). Every method validates `tenant_id` first: anything that is not a non-empty, non-whitespace `str` raises `auradefi.errors.TenantIsolationError`. One tenant's ids are indistinguishable from nonexistent ids for every other tenant. __init__(self) -> None upsert(self, tenant_id: str, txns: Sequence[LedgerTransaction]) -> list[SyncEvent] Insert or update `txns` in one tenant's store. sync(self, tenant_id: str, cursor: str | None = None, limit: int = 100) -> SyncPage Page of changes since `cursor`, ascending last-modified seq. get(self, tenant_id: str, txn_id: str) -> LedgerTransaction Fetch one transaction from this tenant's store. mark_removed(self, tenant_id: str, txn_ids: Sequence[str]) -> list[SyncEvent] Mark transactions removed (reorg semantics), one tenant. apply_reorg(self, tenant_id: str, plan: ReorgPlan) -> list[SyncEvent] Apply a `ReorgPlan`: `mark_removed` then `upsert`. ### SqlModelLedger (auradefi.ledger.backends.sqlmodel) `LedgerPort` over host-owned SQLModel sessions (SPEC §8). Every public method validates `tenant_id` FIRST (non-empty, non-whitespace `str`, else `auradefi.errors.TenantIsolationError`) before touching any session, then runs one session/commit per call. Per-tenant monotonic seqs come from `TenantSeqRow` (first value 1). The counter lives in the DB, so a second binding over the same engine continues the sequence. __init__(self, session_factory: Callable[[], Session]) -> None Bind the HOST's session factory. ZERO I/O happens here. upsert(self, tenant_id: str, txns: Sequence[LedgerTransaction]) -> list[SyncEvent] Insert or update `txns` in one tenant's store (SPEC §6.4). sync(self, tenant_id: str, cursor: str | None = None, limit: int = 100) -> SyncPage Page of changes since `cursor`, ascending seq (SPEC §6.4). get(self, tenant_id: str, txn_id: str) -> LedgerTransaction Fetch one transaction within THIS tenant (rule #6). mark_removed(self, tenant_id: str, txn_ids: Sequence[str]) -> list[SyncEvent] Mark transactions removed (reorg semantics), one tenant. apply_reorg(self, tenant_id: str, plan: ReorgPlan) -> list[SyncEvent] Apply a `ReorgPlan`: mark_removed then upsert, atomically. ### MemorySyncState (auradefi.embed.state) Dict-backed `SyncStatePort` with hard per-tenant isolation. Constructed empty with no arguments: `MemorySyncState()`. Every method validates `tenant_id` first: anything that is not a non-empty, non-whitespace `str` raises `auradefi.errors.TenantIsolationError` (MemoryLedger's tenant hygiene, copied). One tenant's connection ids are indistinguishable from nonexistent ids for every other tenant. __init__(self) -> None get_state(self, tenant_id: str, connection_id: str) -> SyncState Stored state for one connection; `SyncState()` when absent. put_state(self, tenant_id: str, connection_id: str, state: SyncState) -> None Store `state` under (tenant, connection); last write wins. connections(self, tenant_id: str) -> tuple[ConnectionRecord, ...] This tenant's connection records, in creation order. add_connection(self, tenant_id: str, record: ConnectionRecord) -> None Register `record`; duplicate id → `ConflictError` with tenants(self) -> tuple[str, ...] Tenants known through a record OR a cursor, first-seen order. ### SystemClock (auradefi.clock) The wall clock, and the default when a host binds no other. Time is a port precisely so this class is replaceable: quota windows, sync throttling and `as_of_ms` are all derived from `now_ms()`, so swapping it is what makes those testable without sleeping. now_ms(self) -> int Current time as integer milliseconds since the Unix epoch. ### FrozenClock (auradefi.clock) Deterministic clock for tests; moves only when advance() is called. __init__(self, now_ms: int) -> None now_ms(self) -> int advance(self, ms: int) -> None ## Values on the wire What you get back, field by field. ### Quantity (auradefi.money.quantity) An exact base-unit amount: `raw * 10**-decimals`. Equality is strict on `(raw, decimals)`: two quantities of equal numeric value but different scales are NOT equal. Arithmetic and ordering are defined only between quantities of equal `decimals`; mixing scales raises `DecimalsMismatchError` (auradefi.errors). Fields: raw: int, decimals: int as_decimal(self) -> Decimal The exact `Decimal` value `raw * 10**-decimals`. __init__(self, raw: int, decimals: int) -> None ### Money (auradefi.money.fiat) An exact amount denominated in one currency. `amount` is a `Decimal` (never a float); `currency` is validated at construction: 3-letter uppercase code or CAIP-19 (contains '/'). Fields: amount: Decimal, currency: str __init__(self, amount: Decimal, currency: str) -> None ### HoldingsReport (auradefi.portfolio.models) All holdings of one (address × chain) with an exact USD total. `as_of_ms` is an integer ms-epoch timestamp (SPEC §4.4: ms epoch, everywhere, always). `unpriced` lists the CAIP-19 ids of holdings that carry no value, in input order, so a consumer knows exactly what the total omits (SPEC §4.4 data_quality spirit: incompleteness is first-class, never silent). Fields: address: str, chain_id: str, holdings: tuple[Holding, ...], total_value: Money, unpriced: tuple[str, ...], as_of_ms: int assemble(cls, address: str, chain_id: str, holdings: Iterable[Holding], as_of_ms: int) -> HoldingsReport Assemble a report from holdings; pinned algorithm. __init__(self, address: str, chain_id: str, holdings: tuple[Holding, ...], total_value: Money, unpriced: tuple[str, ...], as_of_ms: int) -> None ### Holding (auradefi.portfolio.models) One asset balance on one account (≡ Plaid Holding). `price` is the unit USD price; `value` is the position USD value. Pricing is all-or-nothing: both set (priced) or both `None` (unpriced). Exactly one of the two being `None` raises `ValidationError` at construction. Fields: caip19: str, symbol: str | None, quantity: Quantity, price: Money | None, value: Money | None __init__(self, caip19: str, symbol: str | None, quantity: Quantity, price: Money | None, value: Money | None) -> None ### SyncReport (auradefi.embed.models) Aggregate result of one `sync()` tick (SPEC §8). Same invariants as `ConnectionSyncReport`: negative counts raise `auradefi.errors.ValidationError`, as does `no_op=True` with any non-zero count and any split where `pages_fetched != live_pages + backfill_pages`. The tick's whole budget is the sum of the two phases it was spent on. `connections` carries the per-connection breakdown, defaulting to `()`. When it is non-empty it PINS the aggregate: each of the four counts must equal the sum over the rows and `no_op` must be True exactly when every row is a no-op. An aggregate that contradicts its own breakdown raises `ValidationError` rather than reporting two different truths. Prefer `assemble`, which derives all five from the rows so they cannot disagree. `failed_connections` is likewise derived from the rows, so a partial failure is nameable rather than hidden behind an aggregate that reads like a clean tick (RELEASE_0.1.1 §5 #24). Fields: no_op: bool, pages_fetched: int, live_pages: int, backfill_pages: int, transactions_ingested: int, connections: tuple[ConnectionSyncReport, ...] assemble(cls, connections: Iterable[ConnectionSyncReport]) -> SyncReport Assemble a tick aggregate from its rows; pinned algorithm. __init__(self, no_op: bool, pages_fetched: int, live_pages: int, backfill_pages: int, transactions_ingested: int, connections: tuple[ConnectionSyncReport, ...] = ()) -> None ### ConnectionSyncReport (auradefi.embed.models) What one connection's slice of a `sync()` call did (SPEC §8). `pages_fetched` is this connection's share of the one shared budget, partitioned into the two phases it was spent on: `pages_fetched == live_pages + backfill_pages`, always. Raises `auradefi.errors.ValidationError` when any count (`pages_fetched`, `live_pages`, `backfill_pages`, `transactions_ingested`) is negative, when `no_op` is True and ANY of those four is non-zero, a no-op that fetched pages or ingested transactions is a lie, or when the two halves do not sum to `pages_fetched`. `failed` defaults to False, so every row written before the field existed still means "this went fine". It is True when the connection's sync raised an `auradefi.errors.AuradefiError` that the tick contained (RELEASE_0.1.1 §5 #24), and it is MUTUALLY EXCLUSIVE with `no_op`: "nothing needed doing" and "I could not do it" are different answers, and a row claiming both raises `ValidationError` rather than being believed. A failed row still obeys every count invariant. Failure is not a licence to emit an incoherent partition. Fields: connection_id: str, no_op: bool, pages_fetched: int, live_pages: int, backfill_pages: int, transactions_ingested: int, live_cursor: int, backfill_cursor: int | None, backfill_complete: bool, failed: bool failure(cls, connection_id: str, state: SyncState) -> ConnectionSyncReport The row for a connection whose sync RAISED; pinned shape. __init__(self, connection_id: str, no_op: bool, pages_fetched: int, live_pages: int, backfill_pages: int, transactions_ingested: int, live_cursor: int, backfill_cursor: int | None, backfill_complete: bool, failed: bool = False) -> None ### ConnectionRecord (auradefi.embed.models) One watched address bound to a tenant (SPEC §8, §3.1). `id` is deterministic. See `derive_connection_id`. `chain_id` is a CAIP-2 string; `created_at_ms` is ms-epoch. Fields: id: str, chain_id: str, address: str, created_at_ms: int __init__(self, id: str, chain_id: str, address: str, created_at_ms: int) -> None ### LedgerTransaction (auradefi.ledger.models) A persisted transaction: identity, timing, movements, bookkeeping. `id` is deterministic. See `transaction_id`. `initiated_at` is a ms-epoch int; `confirmed_at` is `None` until confirmation. `removed` and `last_modified_seq` are backend bookkeeping and are excluded from `payload_equal`. Fields: id: str, chain_id: str, tx_hash: str, account_id: str, block_number: int | None, initiated_at: int, confirmed_at: int | None, entries: tuple[Entry, ...], removed: bool, last_modified_seq: int __init__(self, id: str, chain_id: str, tx_hash: str, account_id: str, block_number: int | None, initiated_at: int, confirmed_at: int | None, entries: tuple[Entry, ...], removed: bool = False, last_modified_seq: int = 0) -> None ### Entry (auradefi.ledger.models) One movement of a single asset inside a transaction. Fields: asset_id: str, quantity: Quantity, direction: Direction __init__(self, asset_id: str, quantity: Quantity, direction: Direction) -> None ### SyncPage (auradefi.ledger.models) One page of sync events, ordered by ascending last-modified seq. Clients page until `has_more` is `False` before persisting `next_cursor` (SPEC §6.4). Fields: events: tuple[SyncEvent, ...], next_cursor: str, has_more: bool __init__(self, events: tuple[SyncEvent, ...], next_cursor: str, has_more: bool) -> None ## Accounting Cost basis and PnL at any instant. ### pnl_at (auradefi.accounting.pnl) PnL at an ARBITRARY date. The thing Zerion cannot do (SPEC §9). Replays only the events at or before `at_ms`, in ONE pass, and reports as of that cutoff. Exactly equivalent to filtering the stream by time, calling `process` and then `report`, no marks are pre-computed at fixed dates, and no cutoff is privileged over any other, so a date 3,000 transactions from the nearest month end costs what any other date costs. The filter is a generator, so events after the cutoff are skipped without being materialised. `marks` prices the units still held at the cutoff and follows the same rules as in `report`. pnl_at(events: Iterable[AccountingEvent], method: str, at_ms: int, marks: Mapping[str, Money]) -> PnLReport ### derive_events (auradefi.accounting.lots) Distil ledger transactions into the taxable event stream (SPEC §9). Per entry: `Direction.IN` yields an `AcquisitionEvent` with `cost=None`, `Direction.OUT` a `DisposalEvent` with `proceeds=None`. This module carries totals only when a caller supplies them, so pricing stays out of the accounting layer. Three things produce NO event at all: * `Direction.SELF` entries, moving your own coins is not income; * every entry of a transaction whose `id` is in `internal_transfer_ids`, the whole transaction is skipped, both legs, which is what `is_internal_transfer` is for; * every entry of a `removed=True` transaction, a reorged-away transaction never happened. `at_ms` is `confirmed_at` when it is not `None`, else `initiated_at` (DECISIONS "Accounting event time"). The result is sorted by `at_ms` STABLY, so entries sharing a timestamp keep the order they arrived in. Replay is deterministic. derive_events(transactions: Sequence[LedgerTransaction], internal_transfer_ids: frozenset[str] = frozenset()) -> tuple[AccountingEvent, ...] ### PnLReport (auradefi.accounting.report) PnL as of one instant, under one costing method. `as_of_ms` is the instant the caller asked about, carried through verbatim; it is not read from a clock and need not coincide with any event. `method` is the method the state was replayed under. A report cannot be re-costed, because the lot ledger behind it has already been consumed one particular way. `realized` is the exact sum of the disposals whose realised amount is KNOWN, and `missing_realized_count` is how many were left out of that sum; a caller that ignores the count will silently read an understated total, which is why the count is not optional. `unrealized` is `None` if ANY held asset is uncertain, so it is the conservative whole-portfolio figure rather than a partial sum: the per-asset detail in `per_asset` is where the known parts stay visible. `open_lots` is sorted by `(asset_id, opened_at_ms, lot_id)`: a total order with no ties, so the wire output is byte-stable across runs even when two lots were opened in the same millisecond. UNDER `method="acb"`, `unrealized` AND `TaxLot.cost_basis` DO NOT AGREE, and that is correct. ACB costs from a per-asset running POOL, so a disposal consumes `pool_cost × take/pool` and leaves the pool reduced proportionally; the lots behind it are untouched and keep reporting their own remaining basis, because they stay ground truth for lot-level reporting (docs/internal/DECISIONS.md, "ACB pooling"). Buy 1 at 10, 1 at 20 and 1 at 15, sell one, and the pool holds 30 while the surviving lots sum to 35: a permanent, intended gap of 5. Summing `TaxLot.cost_basis` and comparing it with what `unrealized` implies is therefore the wrong check, and it is an easy one to reach for. `basis_source` names which cost `unrealized` actually subtracted, and `unrealized_basis` and `open_lots_basis` expose both figures, so the difference is inspectable rather than something a caller has to reverse-engineer and mistake for a bug. Fields: as_of_ms: int, method: str, realized: Money, missing_realized_count: int, unrealized: Money | None, per_asset: Mapping[str, AssetPnL], open_lots: tuple[TaxLot, ...], flags: tuple[str, ...], basis_source: str, unrealized_basis: Money | None __init__(self, as_of_ms: int, method: str, realized: Money, missing_realized_count: int, unrealized: Money | None, per_asset: Mapping[str, AssetPnL], open_lots: tuple[TaxLot, ...], flags: tuple[str, ...] = (), basis_source: str = lots, unrealized_basis: Money | None = None) -> None ### TaxLot (auradefi.accounting.report) One open lot in Plaid's `tax_lots[]` shape (SPEC §6.2, DECISIONS "Plaid TaxLot mapping"). `institution_lot_id` is the deterministic id from `lot_id`, so the same acquisition reports the same lot across runs and across backends. `quantity`, `cost_basis` and `current_value` all describe what is LEFT of the lot: units remaining, the exact un-disposed part of the basis rounded once into `Money`, and those units at the mark. `purchase_price` is the exception. It is `cost_total / quantity_original`, a fact of the ACQUISITION that does not move as the lot is drawn down, so a half-consumed lot still reports the price it was bought at. `purchase_price` and `cost_basis` are `None` together when the acquisition was unpriced; `current_value` is `None` when the asset has no mark. Only lots with units remaining are ever emitted. Fields: institution_lot_id: str, original_purchase_datetime: int, quantity: Decimal, purchase_price: Money | None, cost_basis: Money | None, current_value: Money | None, position_type: str, flags: tuple[str, ...] __init__(self, institution_lot_id: str, original_purchase_datetime: int, quantity: Decimal, purchase_price: Money | None, cost_basis: Money | None, current_value: Money | None, position_type: str = LONG, flags: tuple[str, ...] = ()) -> None ## Webhooks Signing, delivery and replay. ### sign (auradefi.webhooks.sign) Return the `X-Auradefi-Signature` value for `body`. `f"v1={hmac_sha256(secret, f'{timestamp_ms}.{body}')}"`: 67 characters: the `v1=` prefix plus 64 lowercase hex. Both the secret and the signed string are encoded UTF-8. sign(secret: str, timestamp_ms: int, body: str) -> str ### verify_signature (auradefi.webhooks.sign) Return `None` iff `signature` is valid AND fresh. Raises plain `AuthError`: one class, one message, for every failure: a mismatched signature, a missing or wrong `v1=` prefix, a mutated body, a wrong secret, and a timestamp outside the window are INDISTINGUISHABLE to the caller. Freshness is `abs(now_ms - timestamp_ms) <= tolerance_ms` (inclusive at both edges, past and future). Comparison is `hmac.compare_digest`, never `str.__eq__`. verify_signature(secret: str, timestamp_ms: int, body: str, signature: str, now_ms: int, tolerance_ms: int = 300000) -> None ### Deliverer (auradefi.webhooks.deliver) Drains due deliveries through an injected `httpx.Client`. The client is the host's: it owns timeouts, proxies, TLS. Tests inject `httpx.Client(transport=httpx.MockTransport(handler))`. __init__(self, store: WebhookStore, client: httpx.Client) -> None Bind the store to drain and the client to POST through. tick(self, now_ms: int) -> tuple[models.Delivery, ...] Attempt every delivery due at `now_ms`; return the updates. ### replay (auradefi.webhooks.replay) Re-arm one delivery; return the NEW PENDING row. Any status may be replayed: dead-lettered, delivered, or still pending. An unknown or cross-project `delivery_id` raises `NotFoundError`, indistinguishably. replay(store: ReplayStore, project_id: str, delivery_id: str, clock: Clock) -> Delivery