dau_core.temporal

Golden models for the temporal-finance hero workflow.

The fused workflow this file is the CPU reference for:

quotes ─┐

├─> backward as-of ─> midpoint/spread/sign ─┐

trades ─┘ ├─> time/vol/$ bars

└─> rolling features

Everything here operates on the encoded integer domain the hardware sees — symbol ids are int32 dictionary codes, timestamps are int64, prices/sizes are int64 fixed-point (the host quantizes at a chosen scale before staging, exactly as dau_polars.ohlcv does with its micro-unit SCALE). The models are therefore scale-agnostic integer arithmetic: a caller that quantizes price and size at scales Sp and Ss gets dollar volume in Sp*Ss units and must express dollar-bar thresholds in the same units. Division is deferred (Q100 discipline): bars and rolling windows emit exact integer accumulators (sum / sum-of-squares / count) and the host divides once, so every value on the wire is exact.

State is bounded and per-symbol: the as-of tile holds one current quote per symbol partition; a bar tile holds one open bar; a rolling tile holds one window. The pre-sort precondition (both streams ordered by (symbol_id, time)) is the same one the sorted-scan tiles already require — the host sorts, or the range-partitioner + sorter path does.

These are semantic references, ahead of the HDL. The int64 timestamp / sum tiles and the as-of / windowed-stat tiles byte-compare against them.

Attributes

_T
QuoteRow
TradeRow
MarketRow

Classes

MarketFeatures	The per-row derived features of one aligned market row.
TemporalBar	One closed bar over aligned market rows. OHLC on price, plus exact
RollingMoment	The exact integer accumulators of one full rolling window — mean and
MarketStream	A deterministic (seeded) quote/trade fixture in the encoded domain, both

Functions

asof_backward(trades, quotes, *[, tolerance]) → list[MarketRow | None]	Backward as-of join: each trade attached to the most recent quote for
matched_rows(aligned) → list[MarketRow]	Drop the as-of misses — the aligned market rows the downstream fused
midpoint(bid, ask) → int	(bid + ask) // 2 (floor). The quote midpoint in the price scale.
spread(bid, ask) → int	ask - bid in the price scale.
trade_sign(price, bid, ask) → int	The stateless quote-rule trade sign: +1 above the midpoint, -1
signed_volume(price, size, bid, ask) → int	trade_sign * size — size credited to the initiating side.
dollar_volume(price, size) → int	price * size — notional in price_scale * size_scale units (the
vwap_numerator(price, size) → int	Alias of dollar_volume(): VWAP = sum(price*size) / sum(size),
market_features(rows) → list[MarketFeatures]	Derive the fused per-row market features for a stream of aligned market
_open_bar(symbol_id, time, price, size, signed) → TemporalBar
time_bars(rows, *, bar_width) → list[TemporalBar]	Fixed-width time bars: trades bucket by time // bar_width; a bar spans
_accumulator_bars(rows, *, threshold, metric) → list[TemporalBar]	Threshold-clock bars: accumulate metric per symbol and close the bar
volume_bars(rows, *, threshold) → list[TemporalBar]	Volume bars: close when cumulative traded size reaches threshold.
dollar_bars(rows, *, threshold) → list[TemporalBar]	Dollar bars: close when cumulative notional (sum price*size) reaches
imbalance_bars(rows, *, threshold) → list[TemporalBar]	Imbalance bars: close when |cumulative signed volume| reaches
_rolling(values, *, window, reduce) → list[_T | None]
rolling_sum(values, *, window) → list[int | None]	Per-symbol rolling sum of the last window values (exact int).
rolling_min(values, *, window) → list[int | None]	Per-symbol rolling minimum of the last window values.
rolling_max(values, *, window) → list[int | None]	Per-symbol rolling maximum of the last window values.
rolling_moments(values, *, window) → list[RollingMoment | None]	Per-symbol rolling (count, sum, sum_of_squares) over the last
moment_mean(moment) → float	Mean from a rolling moment.
moment_var(moment, *[, ddof]) → float	Variance from a rolling moment (ddof=1 sample variance, matching
moment_std(moment, *[, ddof]) → float	Sample standard deviation from a rolling moment.
ewma(values, *, alpha) → list[float]	Per-symbol exponentially weighted moving average, non-adjusted recursion
deltas(values) → list[int | None]	Per-symbol first difference (value[i] - value[i-1]; None for the
gaps(times, *, max_gap) → list[bool]	Per-symbol gap flags: True where time[i] - time[i-1] > max_gap
generate_market_stream(*, seed[, num_symbols, ...]) → MarketStream	Generate a deterministic quote/trade fixture: per symbol, a random-walk

Module Contents

dau_core.temporal._T

dau_core.temporal.QuoteRow

dau_core.temporal.TradeRow

dau_core.temporal.MarketRow

dau_core.temporal.asof_backward(trades: collections.abc.Iterable[TradeRow], quotes: collections.abc.Iterable[QuoteRow], *, tolerance: int | None = None) → list[MarketRow | None]

Backward as-of join: each trade attached to the most recent quote for the same symbol whose quote_time <= trade_time (and, when tolerance is set, trade_time - quote_time <= tolerance). Emits one entry per input trade, in input order — a matched (symbol_id, time, price, size, bid, ask) market row, or None when no quote qualifies (the miss is a typed residue; matched_rows() drops them).

Matches polars.DataFrame.join_asof(strategy="backward", by="symbol", tolerance=...): on equal timestamps the last qualifying quote wins. Both streams must be pre-sorted by (symbol_id, time). The tile holds one current quote per symbol partition and advances it monotonically — bounded state, one pass.

dau_core.temporal.matched_rows(aligned: collections.abc.Iterable[MarketRow | None]) → list[MarketRow]

Drop the as-of misses — the aligned market rows the downstream fused stages consume.

dau_core.temporal.midpoint(bid: int, ask: int) → int

(bid + ask) // 2 (floor). The quote midpoint in the price scale.

dau_core.temporal.spread(bid: int, ask: int) → int

ask - bid in the price scale.

dau_core.temporal.trade_sign(price: int, bid: int, ask: int) → int

The stateless quote-rule trade sign: +1 above the midpoint, -1 below, 0 at it. (The tick-test tie-break needs prior state and is a later windowed variant.)

dau_core.temporal.signed_volume(price: int, size: int, bid: int, ask: int) → int

trade_sign * size — size credited to the initiating side.

dau_core.temporal.dollar_volume(price: int, size: int) → int

price * size — notional in price_scale * size_scale units (the VWAP numerator; exact, host rescales).

dau_core.temporal.vwap_numerator(price: int, size: int) → int

Alias of dollar_volume(): VWAP = sum(price*size) / sum(size), accumulated over a bar and divided once by the host.

class dau_core.temporal.MarketFeatures

The per-row derived features of one aligned market row.

symbol_id: int

time: int

price: int

size: int

midpoint: int

spread: int

trade_sign: int

signed_volume: int

dollar_volume: int

dau_core.temporal.market_features(rows: collections.abc.Iterable[MarketRow]) → list[MarketFeatures]

Derive the fused per-row market features for a stream of aligned market rows (the derive stage of the workflow plan).

class dau_core.temporal.TemporalBar

One closed bar over aligned market rows. OHLC on price, plus exact integer accumulators (volume, dollar/vwap-numerator, signed volume, count) so VWAP and averages divide once on the host — int64 on the wire (see INT64_MAX). open_time/close_time carry the ordering keys for associative cross-batch merge.

symbol_id: int

open: int

high: int

low: int

close: int

volume: int

dollar: int

signed_volume: int

count: int

open_time: int

close_time: int

_extend(time: int, price: int, size: int, signed: int) → TemporalBar

dau_core.temporal._open_bar(symbol_id: int, time: int, price: int, size: int, signed: int) → TemporalBar

dau_core.temporal.time_bars(rows: collections.abc.Iterable[MarketRow], *, bar_width: int) → list[TemporalBar]

Fixed-width time bars: trades bucket by time // bar_width; a bar spans one (symbol_id, bucket) and closes when the bucket or symbol changes. Rows must be pre-sorted by (symbol_id, time). Matches group_by_dynamic(every=bar_width, closed="left") per symbol.

dau_core.temporal._accumulator_bars(rows: collections.abc.Iterable[MarketRow], *, threshold: int, metric: str) → list[TemporalBar]

Threshold-clock bars: accumulate metric per symbol and close the bar on the trade that reaches threshold (the crossing trade is included — the López de Prado convention). A symbol change closes the open bar. Emits a trailing partial bar at end-of-stream.

dau_core.temporal.volume_bars(rows: collections.abc.Iterable[MarketRow], *, threshold: int) → list[TemporalBar]

Volume bars: close when cumulative traded size reaches threshold.

dau_core.temporal.dollar_bars(rows: collections.abc.Iterable[MarketRow], *, threshold: int) → list[TemporalBar]

Dollar bars: close when cumulative notional (sum price*size) reaches threshold (same units as dollar_volume()).

dau_core.temporal.imbalance_bars(rows: collections.abc.Iterable[MarketRow], *, threshold: int) → list[TemporalBar]

Imbalance bars: close when |cumulative signed volume| reaches threshold — a bar per unit of net directional flow.

dau_core.temporal._rolling(values: collections.abc.Iterable[tuple[int, int]], *, window: int, reduce: collections.abc.Callable[[list[int]], _T]) → list[_T | None]

dau_core.temporal.rolling_sum(values: collections.abc.Iterable[tuple[int, int]], *, window: int) → list[int | None]

Per-symbol rolling sum of the last window values (exact int).

dau_core.temporal.rolling_min(values: collections.abc.Iterable[tuple[int, int]], *, window: int) → list[int | None]

Per-symbol rolling minimum of the last window values.

dau_core.temporal.rolling_max(values: collections.abc.Iterable[tuple[int, int]], *, window: int) → list[int | None]

Per-symbol rolling maximum of the last window values.

class dau_core.temporal.RollingMoment

The exact integer accumulators of one full rolling window — mean and variance divide once on the host (moment_mean(), moment_var()).

count: int

total: int

total_sq: int

dau_core.temporal.rolling_moments(values: collections.abc.Iterable[tuple[int, int]], *, window: int) → list[RollingMoment | None]

Per-symbol rolling (count, sum, sum_of_squares) over the last window values — the deferred-division form of rolling mean/variance.

REFUSES a window whose sum of squares leaves a signed 64-bit accumulator. That is the tile’s arithmetic precondition, and it binds inside the int32 domain: (2**31-1)**2 is already 2**61, so a full-range int32 column overflows sum_sq from window 3. The tile cannot check it — it sees one row at a time and has no bound to compare against — and it wraps silently into a wrong variance rather than erroring. Python integers do not wrap, so a golden that computed the exact value would DISAGREE with the hardware over exactly the inputs the contract excludes, and the bench comparing the two would be the only thing that ever noticed. Raising instead keeps the reference and the datapath from ever quietly parting company: inside the precondition they are equal, and outside it neither claims an answer.

The proof properly lives further upstream — dau-polars refuses to route a rolling scan whose value column declares no magnitude_bits — so this is a backstop for a direct caller, not the primary check.

dau_core.temporal.moment_mean(moment: RollingMoment) → float

Mean from a rolling moment.

dau_core.temporal.moment_var(moment: RollingMoment, *, ddof: int = 1) → float

Variance from a rolling moment (ddof=1 sample variance, matching polars.rolling_var). Raises on an under-determined window.

The numerator count*sum_sq - sum*sum is formed in exact integers before the single float divide — dividing sum*sum first (float) would cancel the variance away at fixed-point magnitudes (e.g. three consecutive values near 1e9).

dau_core.temporal.moment_std(moment: RollingMoment, *, ddof: int = 1) → float

Sample standard deviation from a rolling moment.

The root of moment_var(), and deliberately defined in terms of it rather than independently: polars’ rolling_std is the root of its rolling_var, so any other formulation would disagree with the reference in the last place for no benefit.

On device this is where the divide and sqrt tiles land – variance divides before it roots – which is why the numerator stays exact integers right up to the single divide.

dau_core.temporal.ewma(values: collections.abc.Iterable[tuple[int, int]], *, alpha: float) → list[float]

Per-symbol exponentially weighted moving average, non-adjusted recursion (e[0] = x[0]; e[i] = alpha*x[i] + (1-alpha)*e[i-1]) — matches pl.col(x).ewm_mean(alpha=alpha, adjust=False).over("symbol"). Float: the reference real-valued semantics the fixed-point tile approximates under its declared error budget.

dau_core.temporal.deltas(values: collections.abc.Iterable[tuple[int, int]]) → list[int | None]

Per-symbol first difference (value[i] - value[i-1]; None for the first row of each symbol) — matches pl.col(x).diff().over("symbol").

dau_core.temporal.gaps(times: collections.abc.Iterable[tuple[int, int]], *, max_gap: int) → list[bool]

Per-symbol gap flags: True where time[i] - time[i-1] > max_gap (the first row of each symbol is never a gap).

class dau_core.temporal.MarketStream

A deterministic (seeded) quote/trade fixture in the encoded domain, both streams sorted by (symbol_id, time).

quotes: tuple[QuoteRow, ...]

trades: tuple[TradeRow, ...]

dau_core.temporal.generate_market_stream(*, seed: int, num_symbols: int = 4, quotes_per_symbol: int = 24, trades_per_symbol: int = 40, price_scale: int = 1000000) → MarketStream

Generate a deterministic quote/trade fixture: per symbol, a random-walk two-sided quote (micro-dollar prices) and interleaved trades in integer share sizes priced near the prevailing midpoint — so dollar_volume accumulators stay comfortably in int64. Same seed ⇒ byte-identical output (replay determinism). Uses only the stdlib random — zero-dependency, like the rest of dau-core.

Both paths are exercised by construction: one trade per symbol is placed just before that symbol’s first quote, so asof_backward() always produces at least one miss as well as matches.