Metadata-Version: 2.5
Name: asof-store
Version: 0.6.1
Summary: asof store provides a simple and efficient way to store and retrieve data with versioning support. It allows users to save different versions of their data and access them as needed.
Author-email: Louis Zhuang <yzhuang@outlook.com>
Classifier: Development Status :: 3 - Alpha
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.14
Provides-Extra: dev
Requires-Dist: pytest>=9; extra == 'dev'
Provides-Extra: scraper
Requires-Dist: pydantic>=2.0; extra == 'scraper'
Requires-Dist: zapros>=0.20.0; extra == 'scraper'
Provides-Extra: sql-postgres
Requires-Dist: psycopg[binary]>=3.3; extra == 'sql-postgres'
Requires-Dist: sqlalchemy>=2.1; extra == 'sql-postgres'
Provides-Extra: sql-sqlite
Requires-Dist: sqlalchemy>=2.1; extra == 'sql-sqlite'
Description-Content-Type: text/markdown

# asof-store

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

## Install

```sh
uv add asof-store
```

## Usage

```python
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:

```sh
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:

```python
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:

```sh
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)`.

```python
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",
)
```

```python
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:

```powershell
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:

```powershell
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:

```powershell
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:

```powershell
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:

```powershell
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:

```powershell
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:

```sh
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`.
