core_data_updater.jobs.prices

prices.py — PricesFlow: all nine+four price tables, the flow of record.

Design record: PRICES_FLOW.md (historical context; executable declarations in this module are the current truth). Every recurrent job splits on its CLOCK: the market families close, crypto never does, so neither island waits on the other’s roster or calendar. UPDATE prices_daily_market market bulks + the coverage pass — close-aware, sensor-fired prices_daily_crypto the 24/7 island — 00:40 UTC daily, weekends included prices_intraday_core_market the CORE market tables (PG), 15m slots prices_intraday_core_crypto the CORE crypto table, 15m slots prices_crypto_aligned DB-derived aligned bars, fired from the canonical crypto-core product prices_intraday_full_market the FULL equity/etf roster → pq/Garage (fail-closed until layout-init) prices_intraday_full_crypto the FULL crypto roster → pq/Garage BUILD prices_build full history, replace (NO stales), manual BACKFILL prices_backfill refresh each island’s coverage FIRST, then coverage-debt ∪ recent-gap fills, upsert, NIGHTLY 09:00 UTC (≈4am ET)

Laws carried here: state-as-params (market rosters, crypto roster, and sessions are separate operations and each consumer requests exactly what it needs); configs read LIVE per step (named fallback; enabled comes back VISIBLE — a disabled row skips, named; the stamped-disabled runs default enabled=False fail-closed); every landing through CoreDataRepo.land (gates + policy op, drops counted); the same-day drop is CLOSE-AWARE (only the session whose close hasn’t passed is unfinished); fulfill-or-name; no dbt anywhere in the landing path. Aligned bars never fetch a vendor: they rehydrate the canonical landed crypto core table and aggregate it against the passed market calendar.

Sensor timings (Sam): market probes begin close+30m (close via the DOUBLE client hit — EODHD exchange_details + FMP exchange_hours, AM/PM honored), re-probe 15m, INDX bulk as the served canary; crypto daily begins 00:40 UTC every day (no market close); intraday = ONE sensor PER JOB on 15m slots (NOT one per family — see the sensor section’s comment: N sensors on one job mint N run_keys for the same slot, so the slot fires N runs). Every sensor also gates on its own island’s prerequisites through the requires asset-key channel (@sensor(requires=[…]) → the body gains upstream_done): a target is not probed until its canonical state products have been bootstrapped. The body and launched job read current table state; requires is deliberately an existence contract, not an age/freshness claim. KNOWN LIMIT (documented, §4): session gates ride the NYSE calendar for every market family until metadata lands multi-exchange calendars.

Classes

`PricesFlow`

Bases: CoreDataFlow

The prices flow — PRICES_FLOW.md §5, method for method.

`backfill_crypto_daily`

backfill_crypto_daily(cfg, fetch_crypto_roster: dict) -> None

`backfill_crypto_intraday_core`

backfill_crypto_intraday_core(cfg, fetch_crypto_roster: dict) -> None

`backfill_equity_daily`

backfill_equity_daily(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`backfill_equity_intraday_core`

backfill_equity_intraday_core(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`backfill_etf_daily`

backfill_etf_daily(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`backfill_etf_intraday_core`

backfill_etf_intraday_core(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`backfill_index_daily`

backfill_index_daily(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`backfill_index_intraday_core`

backfill_index_intraday_core(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_aligned`

build_aligned(cfg, fetch_history_sessions: dict) -> None

Full aligned history from OWN landed intraday + the full calendar.

`build_crypto_daily`

build_crypto_daily(cfg, fetch_crypto_roster: dict) -> None

`build_crypto_intraday_core`

build_crypto_intraday_core(cfg, fetch_crypto_roster: dict) -> None

`build_crypto_intraday_full`

build_crypto_intraday_full(cfg, fetch_crypto_roster: dict) -> None

`build_equity_daily`

build_equity_daily(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_equity_intraday_core`

build_equity_intraday_core(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_equity_intraday_full`

build_equity_intraday_full(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_etf_daily`

build_etf_daily(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_etf_intraday_core`

build_etf_intraday_core(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_etf_intraday_full`

build_etf_intraday_full(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_index_daily`

build_index_daily(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`build_index_intraday_core`

build_index_intraday_core(cfg, fetch_market_rosters: dict, fetch_history_sessions: dict) -> None

`crypto_daily_ready`

crypto_daily_ready(ctx, upstream_done)

The 24/7 clock: fires once per UTC day from 00:40, weekends and holidays included — the just-finished UTC day is the target. Gated on the crypto catalog: the island’s roster IS its universe, so the job cannot run before that product has been bootstrapped.

`daily_ready`

daily_ready(ctx, upstream_done)

Fire the newest completed session missing from any daily table.

Calendar close truth comes from the canonical session table. Distinct landed table dates, not min/max coverage, expose an interior missed day. One run repairs the newest hole; the next sensor tick advances to the next one. The chosen date rides both run key and tag so retries and reexecution fetch the same vendor partition deterministically.

`fetch_crypto_roster`

fetch_crypto_roster(cfg) -> dict

Read only the 24/7 island’s catalog; no market table is touched.

`fetch_history_sessions`

fetch_history_sessions(cfg) -> dict

Read the complete configured calendar span for builds/backfills.

The historical jobs can fetch rows years outside the operational heartbeat. Feeding their landing gates the 30-day payload silently discarded valid old bars; this dedicated state product uses explicit start/end bounds and is selected only by historical workloads.

`fetch_market_rosters`

fetch_market_rosters(cfg) -> dict

Read only the equity/ETF/index state used by market-clock jobs.

`fetch_sessions`

fetch_sessions(cfg) -> dict

Read the bounded NYSE calendar used by operational heartbeats.

`intraday_full_crypto_ready`

intraday_full_crypto_ready(ctx, upstream_done)

The island’s 15m slot probe → the FULL crypto sweep.

`intraday_full_market_ready`

intraday_full_market_ready(ctx, upstream_done)

The market clock’s 15m slot probe → the FULL equity/etf sweep.

`intraday_ready_crypto`

intraday_ready_crypto(ctx, upstream_done)

The island’s 15m slot probe → the core crypto intraday job.

`intraday_ready_market`

intraday_ready_market(ctx, upstream_done)

The market clock’s 15m slot probe → the core market intraday job.

`refresh_crypto_coverage`

refresh_crypto_coverage(cfg, fetch_crypto_roster: dict) -> None

`refresh_market_coverage`

refresh_market_coverage(cfg, fetch_market_rosters: dict) -> None

`sweep_crypto_full`

sweep_crypto_full(cfg, fetch_crypto_roster: dict) -> None

`sweep_equity_full`

sweep_equity_full(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None

`sweep_etf_full`

sweep_etf_full(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None

`update_crypto_aligned`

update_crypto_aligned(cfg, fetch_sessions: dict) -> None

Derive the recent aligned window from landed core bars only.

This is a separate product-triggered job: the 24/7 ingestion island never waits on the market calendar, while every successful writer of the canonical core product can request a deterministic re-derivation.

`update_crypto_daily`

update_crypto_daily(cfg, fetch_crypto_roster: dict) -> None

The 24/7 island — fired at 00:40 UTC every day; the running UTC day is ALWAYS the unfinished one.

`update_crypto_intraday_core`

update_crypto_intraday_core(cfg, fetch_crypto_roster: dict) -> None

`update_equity_daily`

update_equity_daily(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None

`update_equity_intraday_core`

update_equity_intraday_core(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None

`update_etf_daily`

update_etf_daily(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None

`update_etf_intraday_core`

update_etf_intraday_core(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None

`update_index_daily`

update_index_daily(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None

`update_index_intraday_core`

update_index_intraday_core(cfg, fetch_market_rosters: dict, fetch_sessions: dict) -> None