Skip to content

mostlyright.align

mr.align(spine, *sources) — the composition operator.

align as-of joins each source onto the spine and enforces the leakage guard (source.knowledge_time <= spine.decision_time) in one place for every domain. Its signature stays at two concepts — (spine, *sources) — with no policy keyword arguments. All policy (revision authority, point-in-time fidelity) is declared on the SourceContract, never passed to align.

A “source” is a (frame, SourceContract) pair: a pre-materialized DataFrame plus its bitemporal contract. The operator runs three independently testable pure stages per source:

  1. _collapse_revisions() — collapse append-only revisions to one row per (entity_key, event_time_col) by the spec’s revision_order authority (strict >, first-seen on a tie). This reproduces the merge_climate dedup as a conforming spec rather than as a special case inside align.
  2. join — either a materialized equality join (when the spine already carries the source’s event_time_col, the reconstructed-label path) or a backward as-of on knowledge_time (allow_exact_matches=True is what makes the comparison <=). The join mode is inferred from the spine’s columns, so align needs no join keyword argument. Those are the columns the caller passed in, snapshotted once before the first join — never the accumulating result. A source restoring its own event_time_col onto the result must not turn a later co-source declaring that same name into an equality join keyed on the earlier source’s timestamps.
  3. _audit_source_leakage() — a per-source leakage audit: each source’s knowledge_time is checked against the spine decision_time in isolation, so a raised LeakageError names which source leaked and never blames a clean co-aligned source. point_in_time_fidelity="reconstructed" sources are exempt from the audit because they are knowingly not point-in-time.

Returns a plain pd.DataFrame (y columns by convention), stamped attrs["source"] = "align". It is not a DataFrame subclass, because pandas drops subclasses on merge/concat.

align is source-blind for the label aggregation and never imports or touches the four CWOP firewall files or _internal/merge/climate.py: it reproduces the climate dedup through the spec, it does not modify it.

Functions

Section titled “Functions”
FunctionDescription
align(spine, *sources)As-of join each source onto spine.

Classes

Section titled “Classes”
ClassDescription
DeferredSource(contract, entity, builder)A source whose fetch window is deferred until align runs.

class mostlyright.align.DeferredSource(contract, entity, builder)

Section titled “class mostlyright.align.DeferredSource(contract, entity, builder)”

Bases: object

A source whose fetch window is deferred until align runs.

An internal source builder (e.g. the private _deferred_observations that training_table() uses) returns this instead of eagerly fetching, so a source can be handed to align before its window is known. align is the single place the fetch window is inferred — from the spine’s copy of the source’s equality-join key (contract.event_time_col, e.g. the day key local_standard_date) when present, else the decision_time min/max — and it calls materialize() to turn the deferred source into a concrete (frame, SourceContract) source.

contract

Section titled “contract”

The SourceContract the materialized frame conforms to (prefix / point_in_time_fidelity / entity_key).

entity

Section titled “entity”

The entity (station) the deferred builder will fetch.

builder

Section titled “builder”

A callable (entity, from_date, to_date) -> pd.DataFrame that re-enters the eager fetch path with the align-inferred window. The from_date/to_date are inclusive ISO YYYY-MM-DD strings.

builder: Callable[[str | list[str] | tuple[str, ...], str, str], DataFrame]

Section titled “builder: Callable[[str | list[str] | tuple[str, ...], str, str], DataFrame]”

contract: SourceContract

Section titled “contract: SourceContract”

entity: str | list[str] | tuple[str, ...]

Section titled “entity: str | list[str] | tuple[str, ...]”

materialize(from_date, to_date)

Section titled “materialize(from_date, to_date)”

Run the deferred builder over [from_date, to_date] (inclusive).

mostlyright.align.align(spine, *sources)

Section titled “mostlyright.align.align(spine, *sources)”

As-of join each source onto spine.

The signature stays at two concepts: (spine, *sources), with no policy keyword arguments. A source is either a pre-materialized (frame, SourceContract) pair (an optional third tuple element overrides the prefix) or a DeferredSource — a window-deferred builder call that align materializes over the spine’s local-standard-day key (the source’s event_time_col) when the spine carries it, else over the decision_time min/max.

Per source: refuse latest_only → collapse by revision_order authority → namespace non-key columns under the source prefix, raising a ContractError on collision rather than emitting pandas _x / _y → join (equality or backward as-of, inferred from the caller’s spine columns, snapshotted before the first join) → per-source leakage audit.

Join keys are never namespaced, so each source’s declared event_time_col/knowledge_time_col is restored onto the output under its declared name only when that name has a single owner; a name TWO sources declare is dropped for both (one source’s timestamps are not the frame’s), and a name the spine owns is never overwritten. Returns a plain pd.DataFrame stamped attrs["source"] = "align", plus per-column attrs["provenance"] (source id) and attrs["coverage"] (source valid_from/coverage edge) so an all-NaN column prefix is explainable via mr.provenance(frame).