Skip to content
Mostly Right

mostlyright.align

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

align as-of joins each source onto the spine and enforces the leakage guard (source.knowledge_time <= spine.decision_time) in exactly ONE place for all domains. Its signature is FROZEN at two concepts — (spine, *sources) — no policy kwargs, ever. All policy (revision authority, pit-fidelity staleness) lives on the SourceContract, never on align. The day align grows a tolerance=/fill= kwarg, the god-function has been relocated, not killed.

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 > via first-seen tie). Reproduces the legacy merge_climate dedup as a CONFORMING spec, NOT a special case here (the falsifier,).
  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 (the <= boundary via allow_exact_matches=True). The join MODE is inferred from the spine’s shape, so align stays frozen at two concepts (no join kwarg).
  3. _audit_source_leakage() — a PER-SOURCE leakage audit (R-22): 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 audit-exempt (knowingly not-PIT).

Returns a plain pd.DataFrame (y columns by convention), stamped attrs["source"] = "align" — no subclass (pandas silently drops subclasses on merge/concat; researchers asked for plain frames).

PARITY FIREWALL: 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 via the spec, it does not modify it.

Functions

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

Classes

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

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 (R-02).

A source builder invoked WITHOUT a window (e.g. weather.observations('KNYC')) returns this instead of eagerly fetching, so the training-table desugaring align(spine, weather.observations('KNYC')) never raises TypeError. 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.

Signature FROZEN at two concepts: (spine, *sources) — no policy kwargs. A source is EITHER a pre-materialized (frame, SourceContract) pair (BYO / rung-1; an optional 3rd 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 the decision_time min/max (R-02).

Per source: refuse latest_only (R-18) → collapse by revision_order authority → namespace non-key columns under the source prefix with a hard collision ContractError (R-09/R-24, never pandas _x/_y) → join (equality or backward as-of, inferred from the spine’s shape) → per-source leakage audit (R-22). 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).