mixle.data package

Data layer for mixle – a typed, structure-aware way to get records into the encoder contract.

The core abstraction is DataSource: a lazy, typed (Schema), structured (SampleStructure) reference to data. It is purely additive – seq_encode(list) and seq_encode(rdd) are unchanged; a DataSource is simply a third accepted input that funnels into the same encoder contract. The DataFrame / graph / RDD adapters below remain input/representation helpers (not probability distributions), so they live outside mixle.stats.

exchangeability_check(data, *, alpha=0.01, n_perm=200, seed=0, max_records=10_000)

Test whether row ORDER carries information (see module docstring).

alpha is deliberately strict (0.01): the check should flag clear violations, not manufacture warnings from noise. A dataset too small to have testing power (n < 20), with no numeric field in its record shape (dict/tuple/list/scalar rows are all supported), or whose only numeric field(s) are entirely non-finite (NaN/Inf) is reported inconclusive – untested, never a silent, vacuous “exchangeable”. A field containing SOME non-finite values is reported invalid and excluded from testing; any other, clean numeric field in the same record shape is still tested normally and can still produce a real verdict.

Parameters:

• data (Any)

• alpha (float)

• n_perm (int)

• seed (int)

• max_records (int)

Return type:

ExchangeabilityReport

class ExchangeabilityReport(label, fields=<factory>, n_examined=0, max_records=None, bounded=False, multiplicity=<factory>)

Bases: object

The verdict per numeric field, plus the aggregate label the preconditions record.

Parameters:

• label (str)

• fields (list[dict[str, Any]])

• n_examined (int)

• max_records (int | None)

• bounded (bool)

• multiplicity (dict[str, Any])

multiplicity: dict[str, Any]

How the multiplicity across the primary test family was corrected, and how large that family was. Recorded because the label is only interpretable against it (MXR-080-1887).

property exchangeable: bool

Return True only when the probes ran and did not reject exchangeability.

This is FAILURE TO REJECT, not certification (MXR-080-1887). The probes test for a monotone trend and for a between-halves shift; data can violate exchangeability in ways neither probe looks for, and a real violation can go undetected for want of power. True means “these tests, at this corrected level, found no order signal” – it is not evidence the assumption holds. multiplicity records the correction the verdict was decided under, without which the level is not interpretable.

False both for a detected trend/shift AND for inconclusive (nothing could be tested) – a caller that only inspects this boolean can never mistake “we could not check” for “we checked and found no signal”.

as_dict()

Serialize the report to a JSON-compatible dictionary.

Return type:

dict[str, Any]

class DataSource(*args, **kwargs)

Bases: Protocol

A lazy, typed, structured source of encoder-ready records.

records()

Yield raw records compatible with an encoder’s input type.

Return type:

Iterable[Any]

encode(encoder, num_chunks=1, chunk_size=None)

Partition (structure-aware) and seq_encode -> the same [(count, payload)] shape.

Parameters:

• encoder (Any)

• num_chunks (int)

• chunk_size (int | None)

Return type:

Any

class MaterializedSource(data, structure=EXCHANGEABLE, schema=None)

Bases: object

An in-memory DataSource wrapping a Sequence – what a bare list becomes.

Parameters:

• data (Sequence[Any])

• structure (SampleStructure)

• schema (Schema | None)

records()

Return conformed records from the owned replayable snapshot.

Return type:

Iterable[Any]

materialize()

Return a caller-owned list copy of the conformed snapshot.

Return type:

list[Any]

partition(n, *, by=None)

Split into n structure-safe sub-sources (group-aware for partially-exchangeable data).

Parameters:

• n (int)

• by (Any)

Return type:

list[MaterializedSource]

encode(encoder, num_chunks=1, chunk_size=None)

Encode the materialized records through structure-aware partitioning.

Parameters:

• encoder (Any)

• num_chunks (int)

• chunk_size (int | None)

Return type:

list[tuple[int, Any]]

class LazySource(factory, structure=EXCHANGEABLE, schema=None, length=None)

Bases: object

A DataSource that defers reading to a records factory and materializes on demand.

Connectors (Parquet, SQL, CSV, …) return one of these so open(...) does no I/O until the data is actually encoded; the records are read (and schema-coerced) once and cached.

Parameters:

• factory (Any)

• structure (SampleStructure)

• schema (Schema | None)

• length (int | None)

materialize()

Return a caller-owned list copy of the immutable, conformed cache.

Return type:

list[Any]

records()

Return an iterator over the cached materialized records.

Return type:

Iterable[Any]

partition(n, *, by=None)

Materialize and split into n structure-aware in-memory sources.

Parameters:

• n (int)

• by (Any)

Return type:

list[MaterializedSource]

encode(encoder, num_chunks=1, chunk_size=None)

Materialize records and encode them through structure-aware partitioning.

Parameters:

• encoder (Any)

• num_chunks (int)

• chunk_size (int | None)

Return type:

list[tuple[int, Any]]

as_source(data, structure=EXCHANGEABLE, schema=None)

Coerce data to a DataSource (pass a source through; wrap a sequence as materialized).

Parameters:

• data (Any)

• structure (SampleStructure)

• schema (Schema | None)

Return type:

DataSource

open_source(kind, *args, **kwargs)

Open a DataSource by connector kind.

Examples: open("csv", path), open("parquet", path, columns=[...]), open("sql", url, query="select ..."). The connector module is imported lazily, and a missing driver raises a clear pip install mixle[extra] message.

Parameters:

• kind (str)

• args (Any)

• kwargs (Any)

Return type:

Any

source_kinds()

Return the registered connector kinds.

Return type:

list[str]

class Schema(fields, container_value=False)

Bases: object

An ordered set of typed fields.

Parameters:

• fields (tuple[Field, ...])

• container_value (bool)

property names: tuple[str, ...]

Return field names in schema order.

conform_record(record)

Coerce one record (a scalar for a 1-field schema, else a tuple/dict) to the schema’s types.

Parameters:

record (Any)

Return type:

Any

conform(records)

Coerce every record in records to this schema (raising clear errors on mismatch).

Parameters:

records (Any)

Return type:

list[Any]

static for_model(model)

Best-effort schema a model expects, from its fields/sources + child distributions.

Formalizes the duck-probe in the DataFrame adapter: a record/composite model exposes fields and sources plus child distributions whose support fixes each field’s logical type; a bare leaf yields a single field typed by its support (discrete -> Count, continuous -> Real, …).

Parameters:

model (Any)

Return type:

Schema

class Field(name, type)

Bases: object

A named, typed column.

Parameters:

• name (str)

• type (FieldType)

class FieldType

Bases: object

Base logical type: a canonical NumPy dtype plus a coercion to the encoder-ready Python value.

numpy_dtype

alias of float64

coerce(value)

Return value in the Python representation expected by the corresponding encoder.

Parameters:

value (Any)

Return type:

Any

class Real

Bases: FieldType

A real-valued scalar.

numpy_dtype

alias of float64

coerce(value)

Coerce a finite, non-Boolean value to a Python float.

Parameters:

value (Any)

Return type:

float

class Count

Bases: FieldType

A non-negative integer count logical type.

numpy_dtype

alias of int64

coerce(value)

Coerce value to a non-negative Python int, validating exactness first.

Bare int(value) used to silently truncate (1.9 -> 1), silently accept True/ False (bool is an int subclass in Python) as 1/0, and silently accept negative values – turning a data problem into a wrong-but-plausible-looking observation instead of an error. A numeric string ("42") is still accepted, since that is common, legitimate connector input (CSV/SQL text columns); a bool, a non-integral value, a negative value, or a non-finite (NaN/Inf) value is rejected instead of silently coerced.

Parameters:

value (Any)

Return type:

int

class Integer

Bases: FieldType

A signed integer logical type.

numpy_dtype

alias of int64

coerce(value)

Coerce an exact, finite, non-Boolean integer.

Parameters:

value (Any)

Return type:

int

class Categorical(categories=None, numpy_dtype=<class 'numpy.object_'>)

Bases: FieldType

A categorical label, optionally over a fixed set of categories.

Parameters:

• categories (tuple[Any, ...] | None)

• numpy_dtype (Any)

numpy_dtype

alias of object_

coerce(value)

Return value after validating membership in categories when categories are fixed.

Parameters:

value (Any)

Return type:

Any

class Boolean

Bases: FieldType

A boolean flag.

numpy_dtype

alias of bool

coerce(value)

Coerce value to a Python bool.

Parameters:

value (Any)

Return type:

bool

class Vector(dim=None, numpy_dtype=<class 'numpy.float64'>)

Bases: FieldType

A fixed- or free-length real vector.

Parameters:

• dim (int | None)

• numpy_dtype (Any)

numpy_dtype

alias of float64

coerce(value)

Coerce value to a one-dimensional float64 array and validate dim when set.

A multi-dimensional input is flattened in C order ONLY when dim is set and the element count matches it. Declaring the flat length is what makes the reshape explicit rather than silent: sources that emit blocks – PatchSampler hands back a patch_size-shaped patch – have a genuinely rectangular observation whose model-facing form is the flat vector, and with no field kind for N-D arrays the alternative was to refuse them outright. Without a declared dim the shape is still an error, because then nothing states what length the caller expected and a silent flatten could hide a transposed or mis-sliced block.

Parameters:

value (Any)

Return type:

ndarray

class Timestamp

Bases: FieldType

A point in time (datetime / numpy datetime64 / ISO string / POSIX seconds).

coerce(value)

Coerce value to numpy.datetime64 unless it already has that representation.

Parameters:

value (Any)

Return type:

Any

class Text

Bases: FieldType

A free-text string.

numpy_dtype

alias of object_

coerce(value)

Coerce value to str.

Parameters:

value (Any)

Return type:

str

class Optional(inner=<factory>)

Bases: FieldType

A value that may be missing (None passes through; otherwise the inner type coerces).

Parameters:

inner (FieldType)

property numpy_dtype: Any

Return the NumPy dtype of the wrapped field type.

coerce(value)

Pass missing None through; otherwise delegate coercion to inner.

Parameters:

value (Any)

Return type:

Any

class Nested(schema, numpy_dtype=<class 'numpy.object_'>)

Bases: FieldType

A sub-record with its own Schema.

Parameters:

• schema (Schema)

• numpy_dtype (Any)

numpy_dtype

alias of object_

coerce(value)

Conform a nested record with this field’s child schema.

Parameters:

value (Any)

Return type:

Any

class SampleStructure(kind, by=None)

Bases: object

The joint structure of a dataset’s records (an exchangeability class).

Parameters:

• kind (str)

• by (str | GroupingPolicy | None)

property strides_records: bool

True if records may be strided/shuffled across partitions (everything but grouped data).

group_key(record)

Return the group key of record for partial exchangeability (else None).

Parameters:

record (Any)

Return type:

Any

class GroupingPolicy(policy_id, version, function)

Bases: object

A callable grouping rule with durable identity and an explicit semantics version.

Parameters:

• policy_id (str)

• version (str)

• function (Callable[[Any], Any])

partially_exchangeable(by)

Return a PARTIALLY_EXCHANGEABLE structure grouped by field name or key function by.

Parameters:

by (str | GroupingPolicy)

Return type:

SampleStructure

grouping_policy(policy_id, version, function)

Bind a callable grouping rule to a stable, versioned provenance identity.

Parameters:

• policy_id (str)

• version (str)

• function (Callable[[Any], Any])

Return type:

GroupingPolicy

dataset_hash(data, *, sort=False, max_records=None)

Hex SHA-256 fingerprint of data (a sequence of records or a DataSource).

sort=False (default) is order-sensitive (exact training sequence). sort=True combines per-record hashes commutatively for an order-insensitive fingerprint. max_records truncates (the count and whether truncation actually happened are both mixed in, so a hash of a truncated prefix never collides with a hash of a genuinely complete dataset of that same visible length).

Parameters:

• data (Any)

• sort (bool)

• max_records (int | None)

Return type:

str

model_hash(model)

Hex SHA-256 fingerprint of a fitted model’s parameters (its serialized state).

Stable across processes: hashes the canonical form of to_serializable(model), so the same model always yields the same hash and two models hash equal iff their serialized parameters match. Used to fingerprint a checkpoint and chain EM iteration lineage (see mixle.inference.production.provenance).

Parameters:

model (Any)

Return type:

str

check_dataset(model, data, *, sample=1000, check_support=True, raise_on_error=False)

Validate data against the schema/support model expects (over the first sample records).

Records both type-coercion failures (wrong shape/dtype for a field) and, when check_support is True, support violations (a value the model assigns probability 0 -> -inf log-density). With raise_on_error the first batch of issues is raised as a ValueError.

sample must be a positive integer. A DataReport is meant to certify that the sampled records were actually inspected, so sample=0 used to check nothing and still return DataReport(ok=True, n_checked=0) – a clean bill of health for data nobody looked at – and a negative sample used to fail deep inside itertools.islice with an error that never mentions sample at all. Both are now rejected clearly, immediately, before any work starts.

Parameters:

• model (Any)

• data (Any)

• sample (int)

• check_support (bool)

• raise_on_error (bool)

Return type:

DataReport

class DataReport(ok, n_checked, schema, issues=<factory>)

Bases: object

Result of checking sample records against a model-derived schema and support.

Parameters:

• ok (bool)

• n_checked (int)

• schema (list[tuple[str, str]])

• issues (list[str])

save_encoded(encoded, path, *, encoder=None)

Write encoded (the output of encoder.seq_encode(...)) to path; return its hex digest.

encoder is required and records the encoder’s versioned structural signature – not just its class name – so a load against a structurally different encoder is flagged. Comparing class names alone would let e.g. a one-field CompositeDataEncoder be accepted for a two-field CompositeDataEncoder request, since both share the same class name. The returned digest (also stored in the header) covers this signature as well as the body – see _envelope_digest().

Parameters:

• encoded (Any)

• path (str)

• encoder (Any)

Return type:

str

load_encoded(path, *, encoder=None, trusted=None)

Load encoded data written by save_encoded(), verifying its integrity digest.

The header is parsed as JSON (never pickle) and the envelope digest – header fields and body together, see _envelope_digest() – is checked BEFORE the body is unpickled, so a truncated, corrupted, or header-tampered file is rejected before any deserialization runs. The required encoder argument’s structural signature must match the one recorded at save time (else ValueError). This deliberately checks more than the encoder’s class: a DataSequenceEncoder subclass is expected to make __str__ structural (mirroring what its __eq__ checks) – e.g. CompositeDataEncoder.__str__ recurses into its component encoders and so reflects field count, and DiagonalGaussianDataEncoder.__str__ includes its dim – so a shape mismatch under the same class name (e.g. a one-field vs. two-field CompositeDataEncoder) is caught, not just a mismatch in class.

trusted=True is required, and is the caller’s statement that path is under their control. The body is pickle, so loading it is equivalent to executing whatever is in the file, and NONE of the checks above change that: the digest is computed from and stored inside the same file, so it detects truncation and header tampering but is worthless against a file replaced wholesale – a forged artifact carries a valid digest for its own contents and a matching encoder signature, and executes on load (MXR-080-1873). The module docstring said as much and the parameter did not exist, so the warning was addressed to whoever read the source rather than to whoever called the function. Making it an argument puts the decision at the call site, where the provenance of path is actually known.

Parameters:

• path (str)

• encoder (Any)

• trusted (bool | None)

Return type:

Any

class GraphDataEncoder(directed=False, fallback_assignments=None)

Bases: DataSequenceEncoder

Encode graph observations as canonical adjacency/assignment objects.

Parameters:

• directed (bool)

• fallback_assignments (Any | None)

seq_encode(x)

Encode graph-like observations into graph observation records.

Parameters:

x (Sequence[Any])

Return type:

tuple[GraphObservation, …]

row_count(x)

Return the number of canonical graph observations in an encoded payload.

Parameters:

x (Any)

Return type:

int

nbytes(x)

Return approximate encoded byte size for graph observations.

Parameters:

x (Any)

Return type:

int

class GraphObservation(adjacency, block_assignments=None)

Bases: object

Canonical binary graph observation used by graph encoders.

Parameters:

• adjacency (ndarray)

• block_assignments (ndarray | None)

dataframe_records(df, fields=None, as_dict=False, *, _dict_keys='logical')

Convert DataFrame columns into observation records for seq_encode.

A single selected field becomes scalar observations. Multiple selected fields become tuple observations in the requested field order, matching the data shape expected by composite distributions. When as_dict=True, each row is returned as a mapping keyed by the selected source field names.

Missing cells are canonicalized by what the column holds, not by its dtype backend: a gap in a column of numbers (any int/uint/float dtype – numpy, nullable or pyarrow – and an object column whose present values are all numbers) becomes NaN; a gap in any other column becomes None. So NaN, None, pd.NA and pd.NaT all collapse onto the one marker that column’s numpy-backed spelling would have produced; df and df.convert_dtypes() mark every gap the same way, and the OptionalDistribution fitted from either frame carries the same missing_value and scores both (campaign four, T2-02). Present values are NOT respelled – an Int64 column still yields int where its float64 twin yields float – so the two frames can still select different families.

That covers both directions of the same defect: the numeric one was loud (a ContractError on the whole batch), the string one silent (pandas 3 spells a plain str column’s gap NaN and its nullable twin’s pd.NA, and the mismatched sentinel scored -inf per row with no error). Gaps nested INSIDE a cell – a tuple/list/mapping observation – are left as the caller wrote them apart from pandas’ own sentinels; they are the wrapped field’s data, not this column’s gap.

Parameters:

• df (Any)

• fields (str | Sequence[Any] | None)

• as_dict (bool)

• _dict_keys (str)

Return type:

list[Any]

seq_encode_dataframe(df, fields=None, encoder=None, estimator=None, model=None, num_chunks=1, chunk_size=None)

Sequence-encode selected DataFrame columns with the ordinary stats API.

The frame-shaped counterpart of mixle.stats.seq_encode(), which consumes RECORDS and refuses a frame outright (“expected a sequence of 2-tuples, got DataFrame”). optimize/ fit accept a DataFrame, so the two entry points disagree about what a dataset is and the conversion is the caller’s job: either encode the frame here, or convert it once with dataframe_records() and hand the records to seq_encode yourself. Gaps are canonicalized exactly as dataframe_records() canonicalizes them.

Parameters:

• df (Any)

• fields (str | Sequence[Any] | None)

• encoder (DataSequenceEncoder | None)

• estimator (ParameterEstimator | None)

• model (SequenceEncodableProbabilityDistribution | None)

• num_chunks (int)

• chunk_size (int | None)

sample_rdd(sc, dist, count_per_split, num_splits, seed=None)

Sample independent draws from a distribution into a Spark RDD.

sample_seq_as_rdd(sc, dist, seq_len, count_per_split, num_splits, seed=None)

Sample fixed-length sequences from a distribution into a Spark RDD.

take_sample(rdd, with_replacement, n, seed=None)

Take a deterministic-order random sample from a Spark RDD.

Raises ValueError if with_replacement is False and rdd has fewer than n elements: Spark’s own RDD.takeSample(False, n, ...) silently caps the result at the RDD’s true cardinality in that case (returning every element instead of the requested count) rather than raising, so this checks the returned length itself and fails loudly with the shortfall instead of silently handing the caller fewer records than they asked for (MXR-080-0066). with_replacement sampling always returns exactly n elements and is unaffected.

Parameters:

• rdd (Any)

• with_replacement (bool)

• n (int)

• seed (int | None)

Subpackages

• mixle.data.sources package
  • Submodules
    • mixle.data.sources.array_source module
    • mixle.data.sources.arrow_source module
    • mixle.data.sources.graph_source module
    • mixle.data.sources.hadoop_source module
    • mixle.data.sources.mongo_source module
    • mixle.data.sources.pandas_source module
    • mixle.data.sources.spark_source module
    • mixle.data.sources.sql_source module
    • mixle.data.sources.text_source module

Submodules

• mixle.data.core module
• mixle.data.encoded_io module
• mixle.data.exchangeability module
• mixle.data.hashing module
• mixle.data.partition module
• mixle.data.schema module
• mixle.data.stream_token_source module
• mixle.data.streaming_corpus module
• mixle.data.structure module
• mixle.data.validate module