asof-store

A versioned store for retrieving the latest value written at or before a specified timestamp.

Install

uv add asof-store

Usage

from asof_store import AsOfStore

store = AsOfStore.from_memory()
store.put(as_of=10, key="price", value=100)
store.put(as_of=20, key="price", value=125)

assert store.get(as_of=15, key="price") == 100

with store.as_of(20) as snapshot:
    assert snapshot.get("price") == 125

Both get methods return None when no value exists at or before the requested timestamp. For each key, writes must use strictly increasing timestamps. put returns False without recording a new version when the value equals that key's latest value, or when the exact latest (timestamp, key, value) is repeated. Otherwise, it records the value and returns True. Reusing a timestamp with a different value or going back to an earlier timestamp raises ValueError. Passing None as as_of to get or store.as_of(None) selects the latest available value, matching get_all(None) and get_unique_set(field, None).

SQL storage (optional)

Install the optional extra matching your database:

python -m pip install "asof-store[sql-sqlite]"
# or, for PostgreSQL:
python -m pip install "asof-store[sql-postgres]"

Pass the table name and timestamp, key, and value types when creating a SQL-backed store:

from asof_store import AsOfStore

sqlite_store = AsOfStore.from_sql(
    "sqlite:///asof.db", "price_versions", int, str, dict
)
postgres_store = AsOfStore.from_sql(
    "postgresql+psycopg://user:password@localhost/database",
    "price_versions",
    int,
    str,
    dict,
)

Native SQL types are used where supported, including JSONB for dict, list, tuple, and Pydantic BaseModel subclasses on PostgreSQL (JSON on SQLite). Tuples are stored as JSON arrays and restored as tuples. Pydantic models are serialized with model_dump(mode="json", exclude_unset=True) and restored to their declared model type when read. Other Python types use pickle. SQL stores validate writes against the declared types. SQL-backed datetime values must be timezone-aware. When value_type is NoneType, SQL tables omit the value column; the memory backend does not allocate per-version value storage for None values. Use a distinct table name for each type combination; an existing table is not altered if its schema differs. Other Python values are stored with pickle, so only use SQL stores with databases you trust, since loading a database containing untrusted pickle data can execute code. Call store.close() when finished to release SQLAlchemy's pooled connections. SQL timestamps must use sortable scalar types supported by the database (such as integers, strings, dates, datetimes, decimals, and UUIDs); compound JSON and arbitrary pickle-backed timestamps are rejected. Boolean timestamps are not supported because PostgreSQL does not provide ordering comparisons for them. SQL tables use (key, timestamp) as a composite primary key, which also provides the index used for efficient latest-version lookups. All stores provide get_all(as_of=None) to return each key's latest value and get_unique_set(field, as_of=None) to return distinct values of a mapping or Pydantic model field from those latest values. Supplying as_of limits each query to versions at or before that timestamp. Missing or null fields appear as None in the set.

Use get_all_matching(filters, as_of=None, exact=False) to select mapping or Pydantic values by string fields. Filters are combined with AND, matching is a case-insensitive substring by default, and dotted names select nested mapping fields. Set exact=True for case-insensitive exact matching. SQL stores apply these filters before decoding result values; memory stores use the same matching rules in Python. AsOfView.get_all_matching(filters, exact=False) applies the view's fixed timestamp.

get_keys_matching(filters, as_of=None, exact=False) applies the same filtering rules to JSON-compatible keys. This supports stores that retain records as keys with NoneType values. AsOfView.get_keys_matching(filters, exact=False) uses the view's fixed timestamp.

For fuzzy lookups, use get_all_matching_terms(terms, fields, as_of=None, exact=False) or get_keys_matching_terms(...). Every term must match at least one listed field, and all terms must match. SQL backends combine these checks in the database before decoding matching values or keys.

Interactive Brokers data (optional)

Install the Zapros extra to query the Interactive Brokers UK web API:

uv add "asof-store[scraper]"

Functions in asof_store.ib create and close a Zapros client when one is not provided. Pass an existing sync or async client to reuse its connection. get_exchanges fetches the exchange catalogue, while get_instrument_summary fetches product counts by type. get_products_by_filters fetches one page, and scrape_instruments fetches every reported page and yields instruments. get_funds_by_filters and scrape_funds use IB's dedicated funds endpoint. Async equivalents have an _async suffix; scrape_instruments_async is an async iterator.

The API returns Pydantic models: ExchangeResponse, InstrumentSummaryItem, ProductsResponse, Instrument, and Fund. Request models InstrumentSummaryRequest, ProductsByFiltersRequest, and FundProductsRequest serialize field names to the format expected by IB. Instrument and fund models allow additional IB fields so new or product-specific response fields are retained. Instrument.conid and Fund.conid are IB contract IDs and are used as their primary keys. Instruments or funds without a contract ID have primary_key == None. Exchanges use the composite primary key (id, country_code).

from asof_store.ib import InstrumentScrapeSummary, get_exchanges, scrape_instruments

exchanges = get_exchanges()
for item in scrape_instruments(page_size=500):
    if isinstance(item, InstrumentScrapeSummary):
        print("Summary:", item.products)
    else:
        print(item)

us_stocks = scrape_instruments(
    page_size=500,
    product_type=["STK"],
    product_country=["US"],
    new_product="T",
)

import asyncio

from asof_store.ib import (
    InstrumentScrapeSummary,
    get_exchanges_async,
    scrape_instruments_async,
)


async def main():
    exchanges = await get_exchanges_async()
    async for item in scrape_instruments_async(page_size=500):
        if isinstance(item, InstrumentScrapeSummary):
            print("Summary:", item.products)
        else:
            print(item)

    async for instrument in scrape_instruments_async(
        product_type=["STK"],
        product_country=["US"],
        new_product="F",
    ):
        print(instrument)


asyncio.run(main())

product_type and product_country accept lists of IB product types and countries. If both lists are empty, the scraper uses all keys from product_id2name (including CFD) as product types; otherwise, the provided filters are passed through. new_product accepts "all" (default), "T", or "F" to select all, new, or non-new products. These filters are sent to both the summary used to determine pages and the products requests. Requests use the live IB API and are not cached or persisted. Product page sizes must be 100, 200, 300, 400, or 500. Both instrument scrape iterators yield an InstrumentScrapeSummary tagged event first (event == "summary", with products containing InstrumentSummaryItem values), before requesting product pages, followed by Instrument objects and a page-complete event after each page. The persistence API invokes on_summary before processing, then calls on_progress once for each completed page with that page's instrument ordinal range.

Live integration tests

Opt-in integration tests are kept in tests/integration/. The IB tests call the real IB UK API, including each endpoint and the synchronous and asynchronous scrapers. Install the scraper and test extras, then enable those tests explicitly:

uv sync --extra scraper --extra dev
$env:ASOF_STORE_RUN_IB_INTEGRATION = "1"
uv run pytest -m integration tests/integration/test_ib_live.py

The PostgreSQL tests use postgresql://louis@fre.local/louis. Install the PostgreSQL and test extras, then enable them explicitly:

uv sync --extra sql-postgres --extra dev
$env:ASOF_STORE_RUN_POSTGRES_INTEGRATION = "1"
uv run pytest -m integration tests/integration/test_asof_store_postgres.py

Without the corresponding environment variable, live integration tests are skipped and the default test suite makes no live IB or PostgreSQL requests. The live IB tests also exercise the dedicated funds endpoint.

Scraping instruments into an as-of store

Install the scraper and SQL extras, then run the explicit scraper command:

uv sync --extra scraper --extra sql-sqlite
uv run ass scrape-instruments --sql-uri "sqlite:///ib-instruments.db"

Use --sql-uri with --table-name (default ib_instruments) to choose the destination. Optional repeated --product-type and --product-country arguments filter the scrape; --new-product accepts all, T, or F. --sort-field selects the product-page sort field (currency, country, symbol, or exchange_id; default symbol). --sort-direction selects asc or desc (default asc). --start-page-number resumes pagination at a given page and requires exactly one --product-type; pages before it are not requested. --end-page-number sets an inclusive last page and also requires exactly one --product-type. --print-new prints each new instrument as JSON, and --print-changes prints field-level JSON diffs for changed instruments. Progress is reported after every page by default; --progress-every changes the number of page reports between progress messages. Each message includes the product type and instrument ordinal range, calculated from the page number and page size. The final summary reports total and disjoint counts for new, changed, and unchanged instruments with primary keys, plus new and unchanged instruments without primary keys. Instruments without an IB conid are stored in a second table, ib_instrument_templates by default, keyed by the complete Instrument model; choose another table with --missing-primary-key-table-name. Use --dry-run to request and report only the filtered instrument summary; this mode does not require --sql-uri, fetch product pages, or write to a store.

Reconciling missing instruments

Use scrape-reconcile to fill gaps in an existing instrument store while avoiding pages whose sorted intervals are already complete:

uv run ass scrape-reconcile --sql-uri "sqlite:///ib-instruments.db"

It uses the same table names, page size, product filters, sort options, progress reporting, and print options as scrape-instruments. The command checks each product type independently, fetches the first and last pages, then probes only intervals whose remote position count exceeds the matching local instrument count. If the matching local count already equals IB's reported total, no product pages are requested. Sort-key runs that span page boundaries are checked as a whole, so duplicate symbols or other repeated sort values are handled without assuming a unique ordering key. Instruments from fetched pages are persisted immediately, including instruments without a conid. Reconciliation diagnostics are written to standard error for each completed region and each region split for further probing, including page ranges, boundary sort keys, and local versus remote counts where available. Page requests report their first and last sort values, and duplicate-run boundary searches report each probe and the count calculation used to decide whether the run is complete or must be fetched in full. Duplicate-boundary probes more than one page from the reconciliation boundary are marked with a warning so a wide or unexpected repeated-key region is visible during diagnosis. Local counts are deduplicated by conid, and a fetched version replaces that conid's local sort value for subsequent interval checks, so probing an already-populated page does not inflate the completeness count.

Reconciliation assumes the selected local instruments are a subset of the current IB results. It adds or updates fetched instruments but does not delete local records that are no longer returned by IB; use scrape-instruments when you need a full scan.

Scraping funds into an as-of store

IB uses a dedicated funds scanner endpoint and returns a total count with each page of funds. Scrape and persist those pages with:

uv run ass scrape-funds --sql-uri "sqlite:///ib-funds.db"

The primary table defaults to ib_funds, keyed by each fund's CONID; funds without a contract ID are stored in ib_fund_templates by default. Use --table-name and --missing-primary-key-table-name to select different tables. --product-country can be repeated, and --product-symbol, --new-product, --page-size, --start-page-number, and --end-page-number filter or limit the scrape. Progress is reported after each page by default; --progress-every selects how many pages to skip between reports. New records and changes can be printed with --print-new and --print-changes.

Scrape and store the exchange catalogue with:

uv run ass scrape-exchanges --sql-uri "sqlite:///ib-exchanges.db"

Exchanges use (id, country_code) as their composite primary key. Exchanges missing either component are stored in ib_exchange_templates by default; use --missing-primary-key-table-name to select another table. The exchange scraper also supports --print-new, --print-changes, and --progress-every.

The Python API provides scrape_and_store_instruments(store, ...) and async_scrape_and_store_instruments(store, ...). Both accept an instrument store, plus a missing_primary_key_store of type AsOfStore[datetime, Instrument, NoneType], and return a ScrapeReport. Pass start_page_number and end_page_number to resume or limit a single-product-type scrape. Use scrape_and_store_exchanges(store, ...) or async_scrape_and_store_exchanges(store, ...) for exchanges, with an AsOfStore[datetime, tuple[str, str], Exchange] and a separate AsOfStore[datetime, Exchange, NoneType] for exchanges missing key components. Use scrape_and_store_funds(store, ...) or async_scrape_and_store_funds(store, ...) with an AsOfStore[datetime, int, Fund] and a separate AsOfStore[datetime, Fund, NoneType] for funds without a contract ID.

Publishing

Releases published on GitHub are built and published to PyPI by .github/workflows/publish.yml. Before the first release, configure a PyPI Trusted Publisher for this GitHub repository with workflow publish.yml and environment pypi. Create the matching pypi environment in the repository's GitHub Actions settings.

To publish a new version, update the version in pyproject.toml, push the change, and publish a GitHub Release for that version. The workflow runs the test suite before building and publishing the package.

Development

Install the development tools, tests, and optional dependencies with uv:

uv sync --all-extras --locked
uv run pytest

Install the Git commit hooks with uv run prek install. The hooks run Ruff on changed Python files and check project types with ty when Python files change. Run them manually with uv run prek run --all-files.