data_updater_v2.data_client
DataClient — the one front door for the raw data.
It combines the vendor clients’ methods, fills one vendor’s gaps from another, applies the
fixes that need more than one dataset, and lands the result in the raw tables
(tables/) through common-db; it reads them back for anyone who needs to look. The jobs
call it; dbt builds everything adjusted, total-return and aligned from what it lands.
Sections, built one at a time. Each section’s main methods are build_* (history) and
update_* (the daily run); they return the section’s table rows as the table MODELS
(tables/: a MarketSession per session …) with what the run found beside them, and
write nothing. Writing and reading are their own methods, added once every section’s logic
is settled; the one checked landing path they will use (_land) and the read path
(_read) are below, tested on a disposable Postgres.
1 · CALENDAR market_holidays · market_sessions2 · UNIVERSE etf / equity / index / crypto / commodity universes, our symbol and each vendor's, what each listing IS (nc-feeds' readers)3 · PRICES {family}_daily_prices (update_daily) · {family}_intraday_prices4 · CORPORATE ACTIONS dividends · splits (sections 4 and 5)6 · SYMBOL CHANGES symbol_changes: every rename told, with our ids7 · LISTINGS listings · delistings: every new listing and delisting told8 · FACTS & ANALYTICS what dbt builds from the raw, read back: ``read_prices`` (the aligned table, a family's daily or intraday facts, the aligned crypto table), ``read_catalog`` (securities, eras), ``read_events`` (splits, dividends), ``read_metric_facts`` (the metric facts, the daily metrics view) — four verbs, filtered the same way9 · INDEX CONSTITUENTS index_constituents: what the S&P 500, Nasdaq 100 and Dow 30 held and when it changed, as FMP tells it, with our ids10 · METRICS equity_metrics · etf_metrics: what a security is counted and valued at day by day — the SEC's cover counts and FMP's float and market cap for a stock, FMP's market cap, count and NAV for a fundEvery dataset row carries the listing id of the universe row it was fetched for
(identity: our ids, minted once, carried across renames); a bare symbol’s rows carry none.
What it is handed or builds: the database (common-db DBClient.from_env: POSTGRES_* with
the configs and monitoring databases beside the primary), one Postgres-backed rate limiter
shared by every vendor (FMP’s key is shared with production: every pod draws from the same
bucket), and the vendor clients with their keys from conf.
Classes
DataClient
The raw data’s front door.
Carried state: the DBClient for the pipeline’s own database — the data and the issues
(owned when built here, borrowed when handed in) — and monitoring, the one carrying the
central monitoring database as its monitoring target (common-db’s routing; the same
client by default, DBClient.from_env routes both), the
shared limiter (Postgres-backed on that DBClient unless one is handed in), and the
vendor clients it builds and owns — fmp (FMPClient), eodhd (EODHDClient), sec
(SECClient, named to the SEC by Settings.sec_user_agent) and general
(GeneralClient: Bitstamp, NYSE, Nasdaq, OpenFIGI — keyed when conf has its key, keyless
otherwise) — or borrows when handed in. One thing is kept across calls: the columns of each dbt table
read through section 8, learned once when first asked (_dbt_known). Nothing else
about any call is kept.
The fetch-and-combine methods (build_* / update_*) never touch the database:
what they need beyond their own vendors comes in as a parameter, read by the job from
our tables — the trading days as sessions (read_calendar: OUR calendar, never a
vendor’s or mcal’s; only build_calendar reads the sources it is built from), the
stored rows, the checked splits, the stored daily bars. The read_* / write_*
methods are the only ones that do, one table each; section 8’s three readers are the
doors to what dbt builds (facts, analytics), which the client reads and never writes.
Every section’s write_* goes through one path (_write): the rows are checked against the table
first (every row carries exactly the table’s columns, no null where the table forbids
one, no key twice, at least one row, none outside the scope), then common-db writes
them in ONE transaction, and anything short of a whole landing raises WriteError.
aclose
aclose() -> Noneaopen
aopen() -> DataClientbuild_borrow
build_borrow(listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None, slugs: Mapping[str, Mapping[str, Any]] | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The borrow fee and the shares available per day before our own collection began:
companiesmarketcap’s republished IB history (general.borrow_history, from
2023-01-21, companies only: an ETF’s history starts with IB’s file), as Borrow
models in_companiesmarketcap. slugs: the site’s slug map
(general.borrow_slugs, two minutes: a build in batches reads it once and hands it
to each); None reads it here. Returns (result, gaps): empty names a listing the
site does not rank or has no borrow data for.
build_calendar
build_calendar(end: dt.date | None = None) -> dict[str, Any]NYSE’s calendar from 1900 through end (None = the end of next year), as the
table models: holidays (MarketHoliday, every weekday closed or shortened)
and sessions (MarketSession, every day traded, with its hours), beside every
day the sources disagree on (disagreements, each source’s verdict named), the
source that could not be read (unread: NYSE’s page, FMP — its years fall to
mcal) and the counts. Writes nothing.
build_constituents
build_constituents(universe: Mapping[str, Sequence[Any]] | None = None) -> dict[str, Any]Every line of the three indices’ membership stories FMP tells — each add and
remove in their histories (S&P 500 from 1957, Nasdaq 100 from 1985, Dow 30 from
1994), and today’s members as the anchor (member, dated today) — as
IndexConstituent rows, oldest first; with universe (family -> its rows, every
era) our ids on the index and on each constituent the catalog holds
(resolve_constituent_ids). A change line that names no ticker (a removal FMP
records by name alone) is dropped, named. Writes nothing. Returns rows,
indices (index -> changes, members), dropped, and what the ids left
unresolved.
build_crypto_metrics
build_crypto_metrics(table: str, listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]One Coin Metrics table (supply, valuation, network, exchange) for the
coins over [start, end] (None: each metric’s first day; None: yesterday UTC, the last
complete day), as that table’s models (tables.CRYPTO_METRICS) with complete
and final as the client reads them. Returns (result, gaps): empty a coin that
carries none of the table’s metrics, error a coin whose call failed.
build_daily
build_daily(kind: Kind, listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None, splits: Iterable[Any] | Mapping[str, Iterable[Any]] | None = None, sessions: Iterable[Any] | None = None, resolve_gaps: bool = True) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The daily history of each listing (our symbol, a list of them, or the family’s
universe rows; a row inactive or with fetch false is skipped and named) from start
(None = each source’s first day) through end (None = the last finished day), as
the family’s daily-price models, with the rows each source gave (sources) and the
gap days filled (filled). Returns (result, gaps): gaps.daily_bar_gaps after
the refill, issue first, failed added (gaps.daily_bar_disqualifiers).
resolve_gaps False asks the main source only and returns its gaps. Whole histories are large (11,000 rows for
AAPL): hand the listings in batches.
splits (ETFs, equities): the checked split table (build_splits’ models, or
symbol -> rows) — what lets FMP fill an ETF’s or equity’s days as traded, and EODHD’s
early-adjusted pre-split sessions be replaced. Without it FMP is not asked for them.
result["early_adjusted"] names the sessions taken out of EODHD’s answer.
sessions: our calendar’s trading days (read_calendar) — the days a listing
owes and the last finished one; every family but crypto needs them.
build_delisted
build_delisted(kind: Literal['etf', 'equity'], stored: Sequence[Any], elsewhere: Sequence[Any] = ()) -> dict[str, Any]The listings that have LEFT the US venues, as EODHD’s delisted list and FMP’s delisted list tell them, minted as inactive eras of one family — the inactive universe a history without survivorship bias is built on (the daily build then runs on them: EODHD serves a delisted ticker’s bars, FMP its own list’s).
Per ticker (our spelling): EODHD’s line read for what it IS (papers.eodhd_kind,
the venue in FMP’s code, EODHD_VENUES), FMP’s line for its delisting day and IPO
day (FMP’s list carries no kind: a name reading as an ETF joins the ETF table, else
the equity table; a ticker EODHD’s list types as the other family is that table’s,
whatever FMP’s name reads). Identifiers: EODHD’s ISIN, and OpenFIGI asked with
unlisted instruments included (GeneralClient.figis); what it IS settles as the
active build settles it. A ticker stored already holds inactive is carried as
stored (never re-minted); one a stored ACTIVE era holds today is a former holder of
a reused ticker: minted apart on its identifiers, else on its delisting day when FMP
dates it (identity.seed’s reused_on: the day the ticker was freed), else
named ambiguous and left out — the day its last bar lands can mint it. One
security sits in one catalog once: a line that mints to a security a LIVE era of
this family already is (a vendor’s delisted list naming a listing still trading:
EODHD’s did for ~900 on 2026-10-18) is named still_listed and left out, never
written over the live era; one that mints to a security the other family’s catalog
holds (elsewhere: its rows, every era) is named elsewhere and left out.
Returns rows (the new inactive eras, fetch true, on_fmp / on_eodhd
by which list tells it), delistings (FMP’s dated ones, in_fmp, for
write_delistings), and what was decided: carried, ambiguous,
still_listed, elsewhere, refused, the counts.
build_delistings
build_delistings(start: dt.date | None = None, end: dt.date | None = None, universe: Mapping[str, Sequence[Any]] | None = None, nasdaq_start: dt.date | None = None) -> dict[str, Any]Every US delisting dated in [start, end] (None: all FMP and Nasdaq keep; a future
end: what is announced), as Delisting models, oldest first; with universe
our ids on the ones the catalog holds. nasdaq_start: as build_listings’.
build_dividends
build_dividends(listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None, resolve_gaps: bool = True) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The dividend history of each listing (our symbol, a list of them, or ETF / equity universe rows) with its ex-date in [start, end] (None = no bound), as Dividend models.
Returns (result, gaps): result has the models (rows), the payouts each source
stands behind (sources), each listing’s route and count (listings), the rows
not fetched (skipped); gaps is gaps.dividend_gaps after the combine, issue
first — error (a source that did not answer), differs, dates,
only_fmp / only_eodhd / only_nasdaq, rejected, duplicate,
cleaned (FMP’s folded repeats), invalid, announced, expected — with
failed added (gaps.dividend_disqualifiers: what no third source settles).
build_fails
build_fails(listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The SEC’s fails to deliver for the listings over the half months in [start, end]
(None: from July 2009, the half-month files’ start; None: today), as
FailsToDeliver models (sec.fails a half at a time; a half not published is
empty). A row joins a listing by CUSIP where the catalog holds one, else by the
file’s spelling of the ticker; every row of ours is kept, its id set. Returns
(result, gaps) as build_short_interest.
build_intraday
build_intraday(kind: Kind, listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None, daily: Iterable[Any] | Mapping[str, Iterable[Any]] | None = None, sessions: Iterable[Any] | None = None, interval: Width = '1min', check: Check | None = None, splits: Iterable[Any] | Mapping[str, Iterable[Any]] | None = None, resolve_gaps: bool = True) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The intraday history of each listing (our symbol, a list of them, or the family’s
universe rows) from start through end (None = the last finished session).
splits (ETFs, equities: the checked split table, as build_daily takes it):
FMP’s split-adjusted minutes go back to the raw scale only by a factor those splits
explain (all applied, all but one, one twice, none); a day whose measured scale is
none of them is refused. Without it, the nearest clean split ratio is taken.
start None = each listing’s whole history: from its first daily bar (when
daily is given), never before INTRADAY_START; the daily history older than the
main vendor’s first minute is then asked of the other vendor. daily: the listings’
raw daily bars (the family’s daily models, or symbol -> bars) — the anchor that puts
FMP’s history on the raw scale and checks every day; without it the bars are kept as
served, which is only right for days no split has adjusted yet (the latest).
sessions: the calendar (MarketSession models, or GeneralClient.sessions rows);
None = mcal’s sessions. Minutes are many (2 million a listing since 2003): hand a
few listings, or a window, at a time. interval: an ETF’s, equity’s or index’s in
1min or 5min (an hour does not cut its session whole: _whole_session).
resolve_gaps False: the gaps are found and returned, the alternate is not asked
(a listing no other vendor carries is never asked either way).
Returns (result, gaps): result has the family’s intraday models (rows), the
bars each vendor gave (sources), each listing’s days, fills and refused days
(days), the listings not fetched (skipped); gaps is
gaps.intraday_bar_gaps after the refill, issue first, with failed (the
disqualifiers) — the input of a later targeted run.
build_listings
build_listings(start: dt.date, end: dt.date, universe: Mapping[str, Sequence[Any]] | None = None, nasdaq_start: dt.date | None = None) -> dict[str, Any]Every US listing dated in [start, end] (a future end: what is expected), as
Listing models, oldest first; with universe (family -> its rows, every era)
our ids on the ones the catalog holds. nasdaq_start: the first day Nasdaq’s status
history is asked from (None: the window’s own start) — its site serves about a dozen
pages before a minute’s rest, so a build asks it for months, not its whole history.
build_metrics
build_metrics(kind: MetricKind, listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None, sessions: Iterable[Any] | None = None, sec: bool = True, splits: Iterable[Any] | Mapping[str, Iterable[Any]] | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]A family’s metric history per listing (our symbol, a list of them, or the family’s
universe rows; a row inactive or with fetch false is skipped and named), as the
family’s metric model, one row per listing and day in [start, end] (None: each
source’s reach, today). Returns (result, gaps): rows (symbol then day),
sources (the rows each source told), skipped; the gaps as the daily prices
shape them — failed (listing -> why nothing came: no source carries it, or none
answered), error (listing -> source -> why), empty (a listing every source
answered with nothing: no count told, no issue) — what issues.fetch_failures and
gap_findings read.
An equity: FMP’s as-reported statement counts (fmp.shares windowed: the count
as filed per quarter, dated its period end, in its day’s units, STATEMENTS_FLOOR
on) under in_fmp, kept only when the latest one agrees with today’s count within
FMP’s UNITS_TOLERANCE once the splits since it are applied (the stored split
rows, as read_splits returns them, or symbol -> rows; without them a name split
since its last statement goes to the cap route) — a class or an ADS whose statement counts another class (BRK-B’s class A
equivalents, TSM’s ordinary shares) fails the check, as does a fund, and takes the
market-cap route: FMP’s cap a day as told (fmp.market_cap_history), the count
behind it dbt’s; and with sec the SEC’s cover counts by the catalog row’s CIK
(sec.shares: each 10-Q and 10-K’s count as of its cover day, filed the day
it was known, source the form and accession) under in_sec; a day both told
carries the SEC count (the anchor), both flags. A fund: FMP’s market
cap a day (fmp.market_cap_history) and, dated the session they value, today’s
NAV, net assets, expense ratio and holdings (fmp.navs) and count (fmp.shares).
A listing FMP does not spell (an EODHD-only era) is asked of the SEC alone. Only a
universe row’s metrics can land (the id); a bare symbol is fetched for a look.
build_short_interest
build_short_interest(listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]FINRA’s short interest for the listings at every settlement date in [start, end]
(None: from FINRA’s floor, 2017-12-29; None: today), as ShortInterest models
(general.short_interest: every venue’s rows, ours kept by symbol). Returns
(result, gaps): rows, told (FINRA’s rows over the window), dropped
(its spellings that are no US ticker), skipped; the gaps empty (a listing
FINRA reported nothing for) and error (the sweep unread: a stop, raised).
build_short_volume
build_short_volume(listings: str | Sequence[Any], start: dt.date, end: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]FINRA’s daily short-sale volume for the listings over the weekdays in [start, end]
(None: start alone), as ShortVolume models (general.short_volume: one
file a day). Returns (result, gaps) as build_short_interest; a day with no file
(a holiday, today before the cut) is in dropped, named.
build_splits
build_splits(listings: str | Sequence[Any], start: dt.date | None = None, end: dt.date | None = None, daily: Iterable[Any] | Mapping[str, Iterable[Any]] | None = None, sessions: Iterable[Any] | None = None, resolve_gaps: bool = True) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The split history of each listing (our symbol, a list of them, or ETF / equity
universe rows; an inactive or fetch false row is skipped and named) dated in
[start, end] (None = no bound), announced ones included, as Split models.
daily: the listings’ raw daily bars as traded (the prices tables’ models, or
symbol -> bars) — the judge of every split the sources do not all tell alike.
Returns (result, gaps): result has the models (rows), the splits each source
set stands behind (sources), each listing’s route and count (listings), the
rows not fetched (skipped); gaps is gaps.split_gaps issue first with
failed added (gaps.split_disqualifiers).
build_symbol_changes
build_symbol_changes(universe: Mapping[str, Sequence[Any]] | None = None) -> dict[str, Any]Every rename both vendors tell, as SymbolChange models, oldest first, beside
what each vendor’s list took out (dropped), the dates they tell apart
(disagree) and the days two renames name one ticker (conflict). With
universe (family -> its rows, every era) each rename our catalog holds gets our
ids (resolve_symbol_changes); without, the ids are left null.
build_universe
build_universe(kind: Kind, stored: Sequence[Any] = (), changes: Sequence[Any] = (), since: dt.date | None = None) -> dict[str, Any]One family’s universe with every listing in today’s pull identified afresh; the
stored rows (the table’s models, every era) carry our ids and fetch — a rebuild
never re-mints a security. ETFs and equities: about 6,000 profile calls each, plus
etf/info or sec-profile, batched under the shared limiter (~17 and ~16 minutes at
FMP’s 750 a minute). changes / since as update_universe.
check_dividends_sec
check_dividends_sec(listings: str | Sequence[Any], payouts: Iterable[Any]) -> dict[str, Any]The filers’ own word on our dividend record: for each listing with a CIK (universe
rows), the SEC’s cash dividends per share per fiscal period (sec.dividends)
against the payouts handed in (the stored Dividend rows, or any rows with
symbol / id, ex_date, amount, declaration_date, payment_date),
summed into the filer’s quarters (sec.check_dividends). Returns checks
(symbol -> the quarters, each with declared, declared_sum, declared_agrees, paid,
paid_sum, paid_agrees, payouts), disagreements (the quarters whose sum is not
what the filer declared: a payout missed, doubled or mis-sized), agreed (quarters
agreeing), unchecked (listings with no CIK, or whose filer reports no dividend per
share: funds, most ADRs), errors (a filer unread). COVER_CONCURRENCY filers
at a time; nothing is written.
check_splits_sec
check_splits_sec(listings: str | Sequence[Any], splits: Iterable[Any]) -> dict[str, Any]The filers’ own word on our split record: for each listing with a CIK, the
ratios the filer reported (sec.splits) against the splits handed in (the stored
Split rows, or rows with symbol / id, date, split_to,
split_from): a filed ratio equal to ours (sec.same_ratio: the filer’s
rounding allowed) dated within SPLIT_SEC_DAYS of our day confirms it. Returns
confirmed (symbol -> [{date, sec_date, split_to, split_from}]), unconfirmed
(our splits no filing shows: the vendors’ word alone), sec_only (ratios the
filer reported that no split of ours matches: a split the vendors lack, or one they
dated far off — the arbiter’s finding), unchecked, errors. Nothing is written.
daily_backfill
daily_backfill(kind: Kind, issues: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, limit: int | None = None, through: dt.date | None = None, prices_table: str | None = None, sessions_table: str = f'{SCHEMA}.market_sessions', exchange: str = 'NYSE') -> list[dict[str, Any]]What the backfill job fetches for one family, from the issue table and the price
table: the live eras with an open daily issue that asks work (issues: which, one or
a list; default every one with an action) or an open universe new_listing, each
checked again now (daily_issues), so the work is what the tables owe today — never
what an old issue once said. Per era: its whole history (no bars yet), the history
before its first bar (late), or the days missing, behind or invalid, merged into
windows (days closer than GAP_MERGE_DAYS ask together).
Returns the work grouped by window, the listings sharing one asked together:
[{"start", "end", "rows", "issue_ids", "why"}] — start/end None is a whole
history (build_daily’ own defaults), rows the catalog rows to hand
build_daily, issue_ids the open issues the window answers. limit: at most
this many listings. After build_daily and write_prices, daily_issues for
the same ids closes what landed.
daily_issues
daily_issues(kind: Kind, issues: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, start: dt.date | None = None, end: dt.date | None = None, through: dt.date | None = None, prices_table: str | None = None, sessions_table: str = f'{SCHEMA}.market_sessions', exchange: str = 'NYSE') -> dict[str, Any]What one family’s stored daily bars owe, read from the tables alone — the catalog,
the price table (prices_table, default the family’s), our calendar
(sessions_table, exchange’s sessions) and the event tables — whenever it runs,
in whatever job. Per listing era (the catalog’s rows, ids narrowing them):
no_history a live era with no bar (a new listing not yet built) — build_historylate_history a listing's (ETF, equity) first bar after the first day it owed: its inception or IPO, or the day a rename went into it (META owes from 2022-06-09, never FB's years) — build_historybehind a live era's last bar before the last day owed (``through``, default the last finished day): the update missed it, or did not run — fill_days; an index or commodity (its calendar is not ours) only when more than one session behindrun_missed a day every live era with bars before it lacks: the run for it did not happen (whole-family checks only; error)missing_days days owed between a live era's first and last bar with none (ETFs, equities: NYSE sessions; a coin: every UTC day) — fill_daysinvalid_bars bars that cannot be real (a price <= 0, high under low, an open or close outside the range) — fill_daysextra_days bars on a day that is no session (ETFs, equities)bars_after_end an ended era's bars after its last day (its delisting's day, or the day before the rename out of it)issues: which (one or a list; default all). start / end bound the bars
looked at for missing, invalid and extra days (None: the whole history; late history
is checked only then). Live eras are the ones fetched (active, fetch); an ended
era is checked only for what its stored bars show (invalid, extra, after its end).
Returns issues (observations for observe_issues, keyed by the era and the
kind: the days and counts in evidence), resolved (the ids of the open issues
this check covered that no longer hold — and of the universe’s open new_listing
issues whose era now has bars — for resolve_issues), superseded (the open
issues of what a live era owes — history, behind, missing days — whose era ended or is
no longer fetched: for resolve_issues(closure="superseded")), through and
checked.
What a vendor was found to have nothing for (an issue the job closed as
source_assessment) is not raised again: no history, or none before the first bar,
never; days only those the closing evidence named ({"days": [...]}).
dividend_backfill
dividend_backfill(ids: str | Sequence[str] | None = None, limit: int | None = None, table_name: str | None = None) -> list[dict[str, Any]]The dividends fetches to retry: _retry_backfill for the dividends section.
intraday_backfill
intraday_backfill(kind: Kind, issues: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, limit: int | None = None, max_days: int = 250, through: dt.date | None = None, interval: Width = '1min', prices_table: str | None = None, daily_table: str | None = None, sessions_table: str = f'{SCHEMA}.market_sessions', exchange: str = 'NYSE', store: Store = 'hot') -> list[dict[str, Any]]What the intraday backfill job fetches for one family, from the issue table and the
tables: the live eras with an open intraday issue that asks work (issues: which,
one or a list; default every one with an action), narrowed to ids (the job’s
intraday set), each checked again now (intraday_issues) so the work is what the
tables owe today — never what an old issue once said. Per era: its whole history (no
bars yet), the history before its first bar (late), or the days behind, missing,
short, at the edges or invalid, merged into windows (days closer than
GAP_MERGE_DAYS ask together).
Every window is at most max_days days long (sessions; a coin’s UTC days), a
long history cut into consecutive windows, so one build never holds more than that
(minutes are many: ~98,000 1min bars a year per listing).
Returns the work grouped by window, the listings sharing one asked together:
[{"start", "end", "rows", "issue_ids", "why"}] — rows the catalog rows to hand
build_intraday, issue_ids the open issues the window answers. limit: at
most this many listings. After build_intraday and write_prices,
intraday_issues for the same ids closes what landed.
intraday_issues
intraday_issues(kind: Kind, issues: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, start: dt.date | None = None, end: dt.date | None = None, through: dt.date | None = None, interval: Width = '1min', prices_table: str | None = None, daily_table: str | None = None, sessions_table: str = f'{SCHEMA}.market_sessions', exchange: str = 'NYSE', store: Store = 'hot') -> dict[str, Any]What one family’s stored interval bars owe, read from the tables alone — the
catalog, the intraday table (prices_table, default the family’s), the daily table
(daily_table: the days owed), our calendar (sessions_table, exchange’s
sessions and hours) and the event tables — whenever it runs, in whatever job. Bars are
counted in the database (per listing per day), never read out. ids: the listings
checked — the job’s intraday set from its config (None: every era of the family).
A day is owed when the listing’s daily bar traded it (an index: printed it; a coin:
every UTC day its daily bar has), on or after INTRADAY_START; a session’s bars are
its minutes ÷ the width (a half day’s fewer; a coin’s 1,440 a day at 1min). The same
lines as the build (REFILL_FILL, EDGE_MINUTES). Per listing era:
no_history a live era owing days with no bar (owed_first / owed_last / days) — build_historylate_history the first bar after the first day owed — build_historybehind a live era's last bar before the last day owed (``through``, default the last finished day) — fill_daysrun_missed a day every checked era with bars before it lacks (2 or more eras: one alone is behind) — the run did not happen (error); closed once a checked era that owed it has bars on or past itmissing_days days owed between the first and last bar with none — fill_daysshort_days days with under ``REFILL_FILL`` of their bars — fill_daysedge_days days whose first bar is not within ``EDGE_MINUTES`` of the open or whose last does not reach within it of the close — fill_daysinvalid_bars bars that cannot be real (a price <= 0, high under low, an open or close outside the range) — fill_daysextra_days bars on a day that is no session, or (ETFs, equities) outside the session's hoursbars_after_end an ended era's bars after its last day (its delisting's day, or the day before the rename out of it)issues: which (one or a list; default all). start / end bound the days
looked at for the day checks (missing, short, edge, invalid, extra; None: the whole
history; late history is checked only then). store: hot (Postgres); cold
raises until common-db’s cold leg exists.
Returns issues (observations for observe_issues, keyed by the era and the kind,
the days and counts in evidence), resolved (the open intraday issues this check
covered that no longer hold, for resolve_issues), superseded (the open issues of
what a live era owes whose era ended or is no longer fetched: for
resolve_issues(closure="superseded")), through and checked.
What a vendor was found to have nothing for (an issue the job closed as
source_assessment) is not raised again: no history, or none before the first bar,
never; days only those the closing evidence named ({"days": [...]}).
last_landing
last_landing(model: Any) -> dt.date | NoneThe session of the last complete write of model’s table (record’s mirror:
the receipts in the monitoring database, with the day the run was about) — what an
update takes as the day the table was last updated. None: never landed. Monitoring
unreadable raises: an update must not take that for a first build.
observe_issues
observe_issues(observations: Sequence[Mapping[str, Any]], run_id: str | None = None, stage_id: str | None = None, observed_at: dt.datetime | None = None, actor: str = 'dagster') -> list[dict[str, Any]]Open what was observed (issues.observation mappings), in ONE transaction: a new
identity opens an issue (detected); one open already is seen again — its count,
last seen and revision move, an error never softens (reobserved); one closed
before opens a new occurrence linked to it (recurrence_detected). Every change
writes its issue_event. Returns each issue as it now stands, with event.
read_calendar
read_calendar(what: Literal['sessions', 'holidays'] = 'sessions', start: dt.date | None = None, end: dt.date | None = None, exchange: str | None = None, closed: bool | None = None, half_days: bool | None = None, disagree: bool | None = None, columns: Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'polars') -> AnyOne stored calendar table: what — the trading days (market_sessions) or
the closed and shortened weekdays (market_holidays) — with dates in [start, end]
(None: unbounded), oldest first (descending newest first; limit the first
that many), for exchange (None: EXCHANGE, NYSE).
Narrowed by: closed (holidays: True the full closures, False the shortened
days), half_days (sessions: True the shortened sessions, False the full ones),
disagree (True: only the days the sources tell differently; False: only the
rest), columns (e.g. (“date”,)). output: polars, pandas, arrow, dicts, rows
(tuples), or models (the table models). A failed read raises the error common-db
carried.
read_catalog
read_catalog(table: CatalogTable = 'securities', symbols: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, families: str | Sequence[str] | None = None, active: bool | None = None, status: str | Sequence[str] | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'polars') -> AnyThe catalog dbt builds: securities (one row per security under its current id
and ticker, with its price references) or eras (one row per span of a ticker era,
with its bounds). active narrows on is_active; status (active, delisted,
dormant) is the securities table’s alone. The rest as read_prices.
read_constituents
read_constituents(index: str | Sequence[str] | None = None, start: dt.date | None = None, end: dt.date | None = None, table: ConstituentTable = 'daily', event: str | Sequence[str] | None = None, symbols: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, ours: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output | None = None) -> AnyWhat each index held, index (GSPC, NDX, DJI) and the days [start, end] narrowing,
one table at a time:
daily facts.index_constituents — every session's members, one row per index, ``date`` and member under its current id and ticker: the point-in-time universe (the default; a frame)members facts.index_members — the membership spans (``start`` / ``end`` bound the day a span began)changes facts.index_changes — the additions and removals as told, under the current ids (``event``: added, removed)raw raw.index_constituents — the stored lines as told (``event``: added, removed, member; ``ours``: the lines the catalog holds the ticker of; models by default, what the jobs take back)symbols are current tickers (the raw lines’: as told), ids current era ids,
global_ids securities; as output (polars by default, models for raw).
read_delistings
read_delistings(start: dt.date | None = None, end: dt.date | None = None, symbols: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, exchange: str | Sequence[str] | None = None, ours: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'models') -> AnyThe stored delistings dated in [start, end], oldest first, narrowed as
read_listings; as output.
read_dividends
read_dividends(start: dt.date | None = None, end: dt.date | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, symbols: str | Sequence[str] | None = None, announced: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] = ('id', 'ex_date'), descending: bool = False, limit: int | None = None, output: Output = 'models', table: str | None = None) -> AnyThe stored payouts with their ex-date in [start, end] (None: unbounded), narrowed by
ids (listing ids), global_ids (securities: every era’s payouts, each under its
own id, through the ETF and equity catalogs), symbols (the ticker as fetched),
announced (True: ex-date after today; False: today or before), where (any
other column); as output (models by default: what a job hands back to a build).
table: another schema.table of the same shape.
read_events
read_events(table: EventTable, start: dt.date | None = None, end: dt.date | None = None, symbols: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, families: str | Sequence[str] | None = None, effective: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'polars') -> AnyThe events dbt builds under the current ids: splits (one row per security and
effective date) or dividends (one per security, ex_date and amount).
start / end bound that day; effective narrows to the events whose day has
traded (false: the announced ones ahead). The rest as read_prices.
read_issues
read_issues(section: str | Sequence[str] | None = None, issue_type: str | Sequence[str] | None = None, status: str | Sequence[str] | None = 'open', action: str | Sequence[str] | None = None, table_name: str | Sequence[str] | None = None, listing_ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, symbols: str | Sequence[str] | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] = 'last_observed_at', descending: bool = False, limit: int | None = None, output: Output = 'dicts') -> AnyThe stored issues (open by default; None: every status), narrowed by section, kind,
status, action, table, listing ids, security ids, symbols (each one or a list) and
where; oldest seen first; as output. What a backfill job asks: “the open
build_history issues of the universe”.
read_ledger
read_ledger(since: dt.datetime) -> dict[str, list[dict[str, Any]]]The monitoring ledger since since (an instant, naive = UTC): runs — every run
started then, oldest first (run_id, job_name, job_type, run_date: the session it was
about, status, error, started_at, finished_at) — and receipts — every landing
recorded then (target_name, target_key, status, rows_written, rows_rejected,
rows_total_after, recorded_at, the run’s job_name and run_date, and the errors and
warnings its findings counted). What the night’s report reads. Monitoring
unreadable raises.
read_listings
read_listings(start: dt.date | None = None, end: dt.date | None = None, symbols: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, status: str | Sequence[str] | None = None, exchange: str | Sequence[str] | None = None, ours: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'models') -> AnyThe stored listings dated in [start, end], oldest first, narrowed by symbols /
global_ids / status / exchange (one or a list), ours (True: the
ones our catalog holds), where; as output.
read_metric_facts
read_metric_facts(table: MetricTable = 'daily_metrics', start: dt.date | None = None, end: dt.date | None = None, symbols: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, families: str | Sequence[str] | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'polars') -> AnyThe metrics dbt builds under the current ids: daily_metrics (the view: every
ETF, equity and coin on the aligned calendar — a listing’s latest count carried in the
day’s units, our market cap, a fund’s NAV and premium; a coin’s day of Coin Metrics’
tables — families narrows); the told days, nearly raw: equity_metrics and
etf_metrics (with the day’s close and derivations), short_interest,
short_volume, fails_to_deliver, borrow and the coins’ crypto_supply,
crypto_valuation, crypto_network, crypto_exchange. start / end
bound the day (session on the view, date on the facts). The rest as
read_prices.
read_metrics
read_metrics(kind: MetricKind, start: dt.date | None = None, end: dt.date | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, symbols: str | Sequence[str] | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] = ('id', 'date'), descending: bool = False, limit: int | None = None, output: Output = 'polars', table: str | None = None) -> AnyOne family’s stored metric rows dated in [start, end], narrowed by ids,
global_ids (every era of the security), symbols and where, as output
(polars by default; models for what a build takes back). table: another table
of the same shape (a scratch target).
read_prices
read_prices(table: PriceTable = 'aligned', kind: Kind | None = None, start: dt.date | dt.datetime | None = None, end: dt.date | dt.datetime | None = None, interval: str = '1min', symbols: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, families: str | Sequence[str] | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'polars') -> AnyThe prices dbt builds, one table at a time, as output (polars by default):
aligned analytics.market_aligned_daily_prices — every security on the NYSE calendar, one row per security per ``session``, every family; a coin's row carries the market's marks beside its own daydaily facts.{kind}_daily_prices — one family (``kind``), one row per security per ``date``: the bar as traded, the day's split and dividend, the two forward factors, ``tr``intraday facts.{kind}_intraday_prices — one family's minutes of ``interval`` (1min by default) under their security, with the day's factors; ``start`` / ``end`` as dates bound the ``day``, as datetimes (naive = UTC) the bar's instantcrypto_aligned facts.crypto_daily_aligned — every coin per ``session``: its own day, the same day from its minutes, the market's markssymbols are current tickers, ids current era ids, global_ids securities,
families the aligned table’s families; each narrows (given together, all must
hold). where: any other column, as common-db’s predicates. order defaults
to the symbol and the table’s day.
read_raw_prices
read_raw_prices(kind: Kind, start: dt.date | dt.datetime | None = None, end: dt.date | dt.datetime | None = None, interval: str = 'daily', source_interval: str | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, symbols: str | Sequence[str] | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] = ('id', 'date'), descending: bool = False, limit: int | None = None, output: Output = 'polars', table: str | None = None, store: Store = 'hot') -> AnyOne family’s stored bars, narrowed, as output (polars by default; models for
what a build takes back, e.g. build_intraday’s daily).
interval: daily (the daily table) or a bar width — 1min … 59min,
1hour … 24hour. source_interval: the stored bars it is read from
(default 1min). The same width as interval reads those stored rows as they are
(1min by default; interval="5min", source_interval="5min" the stored 5min
rows). A coarser width is built in the database from them, which must divide it: open
the first bar’s, high the highest, low the lowest, close the last bar’s, volume (and
coin volume) the sum — null when any bar’s is null — and bars how many went in.
Buckets start at the session open (09:30 ET) for listings and 00:00 UTC for a coin,
never across a day; a bucket is stamped with its start, as the stored bars are, and
the session’s last one may be short (15:30-16:00 for an hour).
start / end: a date is the venue’s day (the ET session date for listings, the
UTC day for a coin); a datetime is the bar’s instant (naive = UTC), for an intraday
window. ids: listing ids; global_ids: securities — every era’s rows, each under
its own id and the symbol it was fetched as (FB’s and META’s), through the catalog;
symbols: the ticker as fetched. Each narrows (given together, all must hold).
where: any other column, as common-db’s predicates (a value, None for null, a
(lo, hi) range, a list); on built buckets, it narrows the bars that go in.
table: another schema.table of the same shape (default the family’s).
store: hot (Postgres); cold (the intraday parquet store) raises until
common-db’s cold leg exists.
read_splits
read_splits(start: dt.date | None = None, end: dt.date | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, symbols: str | Sequence[str] | None = None, announced: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] = ('id', 'date'), descending: bool = False, limit: int | None = None, output: Output = 'models', table: str | None = None) -> AnyThe stored splits dated in [start, end] (None: unbounded), narrowed by ids,
global_ids (every era’s, through the ETF and equity catalogs), symbols,
announced (True: dated after today; False: today or before), where; as
output (models by default: what build_daily / build_intraday /
build_splits take as splits). table: another schema.table of the same
shape.
read_symbol_changes
read_symbol_changes(start: dt.date | None = None, end: dt.date | None = None, old_symbols: str | Sequence[str] | None = None, new_symbols: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, ours: bool | None = None, conflict: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, descending: bool = False, limit: int | None = None, output: Output = 'models') -> AnyThe stored renames dated in [start, end] (None: unbounded), oldest first,
narrowed by old_symbols / new_symbols / global_ids (one or a list),
ours (True: the ones our catalog holds the security of; False: the rest),
conflict, where (any other column); as output.
read_universe
read_universe(kind: Kind, active: bool | None = True, fetch: bool | None = True, symbols: str | Sequence[str] | None = None, ids: str | Sequence[str] | None = None, global_ids: str | Sequence[str] | None = None, on_fmp: bool | None = None, on_eodhd: bool | None = None, on_nasdaq: bool | None = None, on_bitstamp: bool | None = None, asset_class: str | Sequence[str] | None = None, security_type: str | Sequence[str] | None = None, exchange: str | Sequence[str] | None = None, is_leveraged: bool | None = None, where: Mapping[str, Any] | None = None, columns: Sequence[str] | None = None, order: str | Sequence[str] = 'symbol', descending: bool = False, limit: int | None = None, output: Output = 'models') -> AnyOne family’s stored catalog (kind), narrowed — what every price and event
method takes as its listings. By default today’s listings to fetch (active and
fetch True; None for every row either way), as the table models.
symbols, ids, global_ids, asset_class, security_type,
exchange: one value or a list; on_fmp / on_eodhd / on_nasdaq /
on_bitstamp / is_leveraged: True or False; where: any other column, as
common-db’s predicates (a value, None for null, a (lo, hi) range, a list). A narrowing
the family’s table has no column for (on_bitstamp on ETFs) is refused. columns
(not with models), order (columns), descending, limit; output: models,
polars, pandas, arrow, dicts, rows. A failed read raises the error common-db carried.
record
record(run_id: str, stage_id: str, table: str, written: Any = None, findings: Sequence[Mapping[str, Any]] = (), target_key: str | None = None, partial: bool | None = None, run_date: dt.date | None = None, message: str | None = None) -> strWrite one monitoring receipt for one table a step wrote (common-dagster’s
target_receipt, idempotent per run, stage, table and target_key): the rows
written (written: a write’s result — inserted and updated), refused, and the
step’s findings counts (issues.universe_findings …) beside it. partial:
the step did not do all it was asked (None: partial when a finding is an error).
Monitoring never fails the work: an outage is logged and the receipt id still comes
back. Returns the receipt id.
resolve_constituent_ids
resolve_constituent_ids(rows: Sequence[Any], universe: Mapping[str, Sequence[Any]]) -> list[dict[str, Any]]Our ids on each line, set in place: the index’s era (the index catalog’s row for
GSPC, NDX, DJI), and the constituent’s — the one era of its ticker the ETF and equity
catalogs hold; a ticker several securities have held resolves a member line to
the era live today and a dated change to the era live on that day where the catalog
knows it (a stored first day), else is left unresolved and named. Returns the
unresolved, each with why.
resolve_event_ids
resolve_event_ids(rows: Sequence[Any], universe: Mapping[str, Sequence[Any]]) -> dict[str, Any]Our ids on each listing or delisting (build_* rows, or stored ones read back)
our catalog holds the ticker of — the one era of that ticker in universe (family
-> its rows, every era), set in place. A ticker several securities have held (a
reused ticker: Sizzle’s SZZL, CAMP) resolves only its latest listing among rows,
to the era live under it today (one ticker, one live era); the rest of it is left
unresolved and named (ambiguous). With one era, an event that era cannot be is
named the same way, never given its ids: one the ticker was listed again after (a
listing among rows more than LISTING_DAYS later — an earlier holder’s), or one
more than LISTING_DAYS before the era’s first day of trading where its row knows
it (ipo_date / inception_date). The job runs it again after write_universe,
so the day’s events carry the day’s eras.
resolve_issues
resolve_issues(issue_ids: Sequence[str], closure: Literal['raw_postcondition', 'source_assessment', 'manual_verification', 'superseded'] = 'raw_postcondition', evidence: Mapping[str, Any] | None = None, run_id: str | None = None, stage_id: str | None = None, closed_at: dt.datetime | None = None, actor: str = 'dagster') -> list[dict[str, Any]]Close open issues as resolved (closure: how it is known — the data proves it,
a verdict, a person, a newer issue), in ONE transaction, each with its resolved
event. An id not open is left as it is. Returns the issues closed.
resolve_symbol_changes
resolve_symbol_changes(changes: Sequence[Any], universe: Mapping[str, Sequence[Any]]) -> dict[str, Any]Our ids on each rename our catalog holds the security of (identity .resolve_changes, walked back from every era in universe: family -> its rows),
set on the models in place: global_id, id (the new era), old_id (the era
before); the others’ left null. Returns how many were resolved and the renames two
securities would claim (clashes: left unresolved).
split_backfill
split_backfill(ids: str | Sequence[str] | None = None, limit: int | None = None, table_name: str | None = None) -> list[dict[str, Any]]The splits fetches to retry: _retry_backfill for the splits section. A
split_changed issue is the price jobs’ (read_issues(section="splits", issue_type="split_changed"): the listing and the day to rebuild prices up to).
universe_issues
universe_issues(kind: Kind, start: dt.date, end: dt.date | None = None) -> list[dict[str, Any]]What the kind universe saw in [start, end] (end None: start alone), read from
the tables the update wrote — the three events, each one our catalog holds (the
event’s id is one of the family’s eras):
new_listing a listing (not withdrawn) whose era is live and no rename went into — action ``build_history`` (the backfill job builds its datasets)rename a rename into one of our eras (the old ticker in ``evidence``); its series is dbt's, stitched by ``global_id``delisting a delisting whose era the catalog has endedEach is keyed by the era (its listing id): one era is listed, renamed into or
delisted once, however many rows tell it (FMP’s IPO day beside the update’s own row
days later); the day and who told it ride in evidence. The window may reach back
before the last update — a vendor lists a ticker days after it trades — so an era
the issue table already holds, open or closed, is never given back again. Returns
the observations, for observe_issues.
update_borrow
update_borrow(listings: str | Sequence[Any]) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]Today’s borrow fee and shares available from IB’s own file (general.ib_shortable:
the whole file, ours kept by symbol), as Borrow models in_ib dated the file’s
day. Returns (result, gaps): empty names a listing IB does not lend.
update_calendar
update_calendar() -> dict[str, Any]The daily run: the same build (one NYSE page, one FMP call, mcal locally) — the whole calendar is cheap, and rebuilding it is what catches a newly announced day.
update_constituents
update_constituents(universe: Mapping[str, Sequence[Any]] | None = None) -> dict[str, Any]The daily run: the whole stories again (FMP’s lists are small and a late line can
land anywhere in them), today’s members as today’s anchor — build_constituents.
update_crypto_metrics
update_crypto_metrics(table: str, listings: str | Sequence[Any], day: dt.date | None = None, days: int | None = None, bars: Mapping[str, Mapping[dt.date, Any]] | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]day (None: the UTC day closed last) and the days (CRYPTO_REASK_DAYS)
before it, as build_crypto_metrics: an upsert of these rows replaces what was
provisional or filled. With bars (our symbol -> UTC day -> our daily bar: its
close, volume in dollars and coin_volume; a bare number is a close) what
Coin Metrics has not told yet for a day of the window is filled (CRYPTO_FILL:
Coin Metrics completes a day 01:25-04:16 UTC the next morning, so the night’s run
fills most of its newest day) from the coin’s last told value within
CRYPTO_REASK_DAYS — Coin Metrics’ answer reaches that much further back, and the
stored told rows stand in for a coin whose call failed — the row estimated,
final false; the next night’s re-ask lands Coin Metrics’ own row over it. Returns
(result, gaps) as build_crypto_metrics, the result’s estimated the rows
filled.
update_daily
update_daily(kind: Kind, listings: str | Sequence[Any], sessions: Iterable[Any] | None = None, resolve_gaps: bool = True, day: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The last finished day’s bar for each listing (our symbol, a list of them, or the
family’s universe rows; a row inactive or with fetch false is skipped and named), as the
family’s daily-price model, with where each came from. Returns (result, gaps): the
gaps of that one day, issue first, failed added (gaps.daily_bar_disqualifiers).
resolve_gaps False skips the last pass (FMP for whatever is still missing).
sessions: our calendar’s trading days (read_calendar) — which day is the last
finished one; every family but crypto needs them. day: that day given (the session
a run’s sensor stamped: a run launched late still fetches the day it is about).
update_delistings
update_delistings(start: dt.date, end: dt.date | None = None, universe: Mapping[str, Sequence[Any]] | None = None, nasdaq_start: dt.date | None = None) -> dict[str, Any]The daily run: build_delistings from start, with Nasdaq’s deletes for today;
the stored rows’ in_universe / in_sec kept (_sticky).
update_dividends
update_dividends(start: dt.date, end: dt.date, listings: str | Sequence[Any] | None = None, resolve_gaps: bool = True, eodhd_end: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]Every payout with its ex-date in [start, end] (a future end lists what is
announced) from the three calendars, combined as build_dividends does, for the
listings asked (None = every US symbol the calendars carry). The calendars answer a
whole window: a writer can replace that window, so a cancelled announcement drops.
eodhd_end: the last day EODHD’s files are asked for (None: end) — the
announced payouts ahead are FMP’s calendar’s in one call and Nasdaq’s a free call a
day, EODHD’s a 100-credit file a day.
Returns (result, gaps) as build_dividends; result["unanswered"] names a
calendar that did not answer (its listings are combined from the others, and each
names that calendar’s error under the gaps’ error).
update_fails
update_fails(listings: str | Sequence[Any], halves: int = 2) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The last halves half months (the SEC publishes a half two to three weeks after
it ends: the newest asked may not be out yet), as build_fails.
update_intraday
update_intraday(kind: Kind, listings: str | Sequence[Any], day: dt.date | None = None, daily: Iterable[Any] | Mapping[str, Iterable[Any]] | None = None, sessions: Iterable[Any] | None = None, interval: Width = '1min', check: Check | None = None, resolve_gaps: bool = True) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]One finished session’s minutes (day None = the last finished one) for each
listing, as build_intraday for that day alone: the main vendor, its gaps that day
asked of the other. The day pulled is raw on both vendors; daily (that day’s bar)
still checks it when given.
update_listings
update_listings(start: dt.date, end: dt.date, universe: Mapping[str, Sequence[Any]] | None = None, nasdaq_start: dt.date | None = None) -> dict[str, Any]The daily run: build_listings over the window, with Nasdaq’s adds for today; the
stored rows’ in_universe / in_sec kept (_sticky).
update_metrics
update_metrics(kind: MetricKind, listings: str | Sequence[Any], sessions: Iterable[Any] | None = None, sec: bool = True, day: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The last finished session’s rows for each listing (day: that session, else the
newest NYSE session closed by now from sessions), as the family’s metric model,
with what the run found — (result, gaps) as build_metrics shapes them. FMP’s
sweep of counts (fmp.shares_all, ten calls for the world) and its batch of
market caps (fmp.market_caps) for every listing FMP spells: either unread raises
FetchError (the job opens an issue on the table and skips). A count is dated the
session its stamp values (_valued_on: the last closed at FMP’s refresh clock, as
the build dates it; a count FMP has not refreshed for weeks lands on its own old day,
again, unchanged; one valuing a session after day is that session’s run’s, left
out here); a cap is dated FMP’s own day — the session, or an earlier one FMP
has not moved past — and a cap FMP dates after the session (read in the next session:
the forming day’s) is left out, named in late. A fund adds its etf/info day
(fmp.navs), the NAV dated the session its stamp values (it lags one or two). An equity with sec adds the SEC’s covers of the
filers of COVER_FORMS in EDGAR’s daily index for the last COVER_LOOKBACK_DAYS
days (the index of a day fills through its evening), each filer’s whole cover
history re-read (idempotent under the key); a day’s index unread is a finding
(sec_index), never a stop.
update_short_interest
update_short_interest(listings: str | Sequence[Any], since: dt.date) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The settlement dates after since (the last one stored), as build_short_interest;
general.short_interest_dates says when a new one exists.
update_short_volume
update_short_volume(listings: str | Sequence[Any], day: dt.date | None = None, sessions: Iterable[Any] | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]The last finished session’s file (day: that session), as build_short_volume.
update_splits
update_splits(start: dt.date, end: dt.date, listings: str | Sequence[Any], daily: Iterable[Any] | Mapping[str, Iterable[Any]] | None = None, sessions: Iterable[Any] | None = None, resolve_gaps: bool = True, eodhd_end: dt.date | None = None) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]Every split dated in [start, end] (a future end lists the announced ones) for
the listings asked, from FMP’s calendar, EODHD’s daily files (one call per day, 100
credits) and Nasdaq’s calendar, combined as build_splits does. eodhd_end: the
last day EODHD’s files are asked for (None: end) — the announced splits ahead are
FMP’s and Nasdaq’s calendars’ in one call each, EODHD’s a file a day. Returns
(result, gaps) as build_splits; result["unanswered"] names a calendar that
failed.
update_symbol_changes
update_symbol_changes(universe: Mapping[str, Sequence[Any]] | None = None) -> dict[str, Any]The daily run: the same build (two calls) — the whole lists are cheap, and reading
them whole is what catches a vendor’s late or corrected line — and, with a
universe, the SEC’s covers on the last SEC_CHANGES_DAYS’ renames we hold
(_sec_renames: in_sec where the filer’s own filings show the change); the
stored rows’ in_universe / in_sec kept (_sticky: the SEC checks only the
last SEC_CHANGES_DAYS, and a rename once confirmed stays confirmed).
update_universe
update_universe(kind: Kind, stored: Sequence[Any], changes: Sequence[Any] = (), listings: Sequence[Any] = (), delistings: Sequence[Any] = (), since: dt.date | None = None) -> dict[str, Any]One family’s universe today against what is stored (the table’s models, every
era): today’s pull, only the new listings, any never profiled and every ticker a
symbol change since since names identified, the rest carried as stored, the ones
gone kept inactive. About 30 calls plus 3 to 4 per listing identified.
changes: the symbol changes (SymbolChange models or rows) — the ones dated
after since (the day the stored catalog was last updated) and by today carry a
stored security onto its new ticker (identity.assign); without since none
is applied. listings / delistings: the stored listing events (read_listings
/ read_delistings since since), which say why a ticker came or went
(catalog.identify_catalog_changes: result["catalog"], the unexplained named).
result["identity"] names what was decided: renamed, twins, reused, refused,
ambiguous; result["changed"] is every row new or changed against stored —
what write_universe writes. result["listings"] / result["symbol_changes"]
/ result["delistings"]: what the update found that no vendor told (an era opened
that no listing names, a rename found by the identifiers alone, an era ended that no
delisting or rename explains), as rows for write_listings /
write_symbol_changes / write_delistings (in_universe, dated today; ETFs
and equities, the families the event tables cover).
write_calendar
write_calendar(data: Any, what: Literal['sessions', 'holidays'], target_schema: str | None = None, target_table: str | None = None, exchange: str | None = None, mode: Mode = 'reconcile', keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite one calendar table (the one write path, _write), scoped to exchange
(None: EXCHANGE): what — the trading days (market_sessions) or the closed
and shortened weekdays (market_holidays). data: that table’s rows in any form
_write takes (build_calendar’s sessions or holidays). target_schema
/ target_table: where it lands (None: the table’s own). The rest is _write’s;
returns its result.
write_constituents
write_constituents(data: Any, target_schema: str | None = None, target_table: str | None = None, mode: Mode = 'reconcile', keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite the constituents table (_write): build_constituents’ rows, or any
form _write takes. A reconcile’s scope is the indices the rows tell (a story told
whole each time); the rest is _write’s.
write_delistings
write_delistings(data: Any, target_schema: str | None = None, target_table: str | None = None, window: tuple[dt.date | None, dt.date | None] | None = None, mode: Mode = 'reconcile', keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite delistings (_write), window the dates a reconcile makes exact (None:
the whole table). The rest is _write’s.
write_dividends
write_dividends(data: Any, target_schema: str | None = None, target_table: str | None = None, mode: Mode = 'upsert', ids: str | Sequence[str] | None = None, window: tuple[dt.date | None, dt.date | None] | None = None, keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite payouts (_write) in the dividends table’s shape: build_dividends /
update_dividends’ rows, or any form _write reads (rows, a frame, a csv /
parquet path). target_schema / target_table: where they land (None: ours).
mode: upsert (default: new payouts added, changed ones updated, nothing
removed — a corrected amount is a new key, ex-date and amount, so the old row stays)
· append · reconcile / replace (a rebuild: the slice made exactly the
data — the listings ids the run answered for, required, over the ex-dates
window, None all; a reconcile removes what the data no longer has only with
stale archive (to dividends__retired, dated) or delete, stale_cap
None = no cap). A rebuild after a calendar did not answer (update_dividends’
result["unanswered"]) would remove what only that calendar had: the job writes
an upsert then.
No payouts is a normal answer: an upsert or a rebuild keeping the rest writes nothing and returns None; a rebuild that would remove with no payouts in the data is refused (an empty answer never wipes a record). Returns common-db’s result.
write_listings
write_listings(data: Any, target_schema: str | None = None, target_table: str | None = None, window: tuple[dt.date | None, dt.date | None] | None = None, mode: Mode = 'reconcile', keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite listings (_write), window the dates a reconcile makes exact (the
build’s window; None: the whole table). The rest is _write’s.
write_metrics
write_metrics(data: Any, kind: MetricKind | None = None, target_schema: str | None = None, target_table: str | None = None, mode: Mode = 'upsert', ids: str | Sequence[str] | None = None, window: tuple[dt.date | None, dt.date | None] | None = None, keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite metric rows (_write): a build’s or update’s models, or any form _write
reads — then kind names the table’s shape. mode: upsert (default: new days
inserted, changed ones updated, nothing removed — the daily update, a new listing’s
history) · append · reconcile / replace (the slice made exactly the data: a
rebuild). The slice is the listings ids (default: the data’s own) over the days
window (None: all of them); a reconcile removes the slice’s rows the data no
longer has only with stale delete or archive (default keep), refused
past stale_cap of them, and refused outright when a listing of the slice has no
row in the data. An upsert of a stock’s or a fund’s rows is merged with the stored
row of each day first (_merge_metric): one day is told by several sources on
different nights (the SEC’s cover, FMP’s count by its stamp, FMP’s cap, the NAV a
session or two late), so a value told before and not now is kept, the SEC’s count
stays its day’s anchor, and the in_* flags name everyone who told the day. The
rest is _write’s.
write_prices
write_prices(data: Any, kind: Kind | None = None, interval: str = 'daily', target_schema: str | None = None, target_table: str | None = None, mode: Mode = 'upsert', ids: str | Sequence[str] | None = None, window: tuple[dt.date | None, dt.date | None] | None = None, keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False, store: Store = 'hot') -> AnyWrite bars (_write): a build’s or update’s models, or any form _write reads
(a frame, rows, a csv / parquet path) — then kind and interval (daily, or
any intraday width) name the table’s shape. target_schema / target_table: where
the bars land (None: the family’s table); any other gets the same shape.
store: hot (Postgres). cold is the intraday tables’ parquet store, which
needs common-db’s cold leg: until it exists, asking for it raises.
mode: upsert (default: new bars inserted, changed ones updated, nothing
removed — the daily update, a refill, a new listing’s history) · append ·
reconcile / replace (the slice made exactly the data: a rebuild). The slice is
the listings ids (default: the data’s own — never a listing the data does not
carry) over the venue days window (None: all of them), and for intraday the
data’s bar widths: another listing, era, day or width is never touched. A reconcile
removes the slice’s bars the data no longer has only with stale delete or
archive (default keep: nothing removed), refused past stale_cap of them,
and refused outright when a listing of the slice has no bar in the data — an empty
answer never wipes a history. coerce: coerce (safe conversions), strict
(any mismatch refuses). dry_run: the counts, nothing written.
write_splits
write_splits(data: Any, target_schema: str | None = None, target_table: str | None = None, mode: Mode = 'upsert', ids: str | Sequence[str] | None = None, window: tuple[dt.date | None, dt.date | None] | None = None, keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite splits (_write) in the splits table’s shape: build_splits /
update_splits’ rows, or any form _write reads. target_schema /
target_table: where they land (None: ours).
mode: upsert (default: new splits added, a changed ratio updated in place,
nothing removed) · append · reconcile / replace (a rebuild: the listings
ids the run answered for, required, over the dates window, None all; removed
only with stale archive (to splits__retired, dated) or delete,
stale_cap None = no cap). A split moved to another day is a new key: only a
rebuild retires the old one.
What a fetch changes against the stored splits (events.split_changes: read the
stored ones first with read_splits) is what the price jobs rebuild around: the
daily days FMP filled before a changed split, and EODHD’s pre-adjusted sessions
beside it, were put on the as-traded scale with the old list.
No splits is a normal answer: an upsert or a rebuild keeping the rest writes nothing and returns None; a rebuild that would remove with no splits in the data is refused. Returns common-db’s result.
write_symbol_changes
write_symbol_changes(data: Any, target_schema: str | None = None, target_table: str | None = None, window: tuple[dt.date | None, dt.date | None] | None = None, mode: Mode = 'reconcile', keep: Sequence[str] = (), stale: Stale = 'keep', stale_cap: int | None = None, coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite renames (_write): build_symbol_changes’ rows, or any form _write
takes. window (dates; None: the whole table) is the slice a reconcile makes
exact. The rest is _write’s.
write_universe
write_universe(data: Any, kind: Kind | None = None, target_schema: str | None = None, target_table: str | None = None, mode: Mode = 'upsert', keep: Sequence[str] = (), coerce: Coerce = 'coerce', dry_run: bool = False) -> AnyWrite a family’s catalog rows (_write): update_universe’s changed (the
new rows, and every stored row whose columns changed — an era ended is one value,
is_active), or any form _write takes (kind names the table then). An
upsert by id by default: rows are added or changed, never removed — every
dataset row points at its era’s id, so a delisted or renamed era stays, inactive.
The eras that end are written first, then the rest (one active row per symbol). The
rest is _write’s; returns the counts of both writes.
ShortListError
Bases: FetchError
A vendor’s catalog answered whole but short: it would end more of the active eras at
once than real churn does (DataClient.SHRINK_LIMIT). A fetch failure to the jobs (an
unread issue and a skip; the repair asks again), carrying no vendor response.
WriteError
Bases: RuntimeError
A landing that did not land whole — refused before the write (rows that do not fit
the table) or by common-db (a failed or refused op: its report rides on report).
The table is as it was: every op is one transaction.
Functions
merge_calendar
merge_calendar(sessions: list[dict[str, Any]], start: dt.date, end: dt.date, nyse: Mapping[str, Any] | None, fmp: list[dict[str, Any]] | None, record: Mapping[dt.date, Mapping[str, Any]] = NYSE_RECORD, exchange: str = 'NYSE') -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]The calendar from mcal’s sessions (GeneralClient: every session, its hours and
whether it is shortened), NYSE’s announced calendar (nyse_calendar; None = unread)
and FMP’s holiday rows (None = unread), over [start, end].
Who decides a day: NYSE in the years it announces; record on its days; mcal
otherwise. FMP never decides. Each source covering the day gives its verdict — closed,
or open with a shortened close / late open, or a regular day — and the day is marked
disagree when the verdicts differ. Returns (market_holidays rows,
market_sessions rows, the disagreeing days with every source’s verdict).
settle
settle(profile: Mapping[str, Any] | None, records: Sequence[Mapping[str, Any]], eodhd: Mapping[str, Any] | None, sec: Mapping[str, Any] | None = None) -> tuple[str | None, str | None, bool, dict[str, Any]]What a listing IS from the four readings (each the vendor’s line read by nc-feeds’
readers, papers): FMP’s profile line, OpenFIGI’s record (only when its records agree),
EODHD’s ticker-list row, and the SEC’s directories (sec: SECClient.ids’ answer —
a ticker in its fund directory is a 40-Act fund, an ETF; the company directory says
nothing of the paper). The (kind, paper) two of them say; else OpenFIGI’s, else the
SEC’s, else EODHD’s, else FMP’s. Returns (kind, paper, disagree, what each said). On the
615-name sample of 2026-10-05 this settles the 64 notes and preferreds FMP’s line calls
common (STRC, TMUSL) to what OpenFIGI and EODHD say, keeps the BDCs closed-end funds, and
the 529 plans funds; the SEC’s word settles the 64 ETFs EODHD typed common stock on
2026-10-18.