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)

Test whether row ORDER carries information (see module docstring). Small n -> exchangeable (no power).

alpha is deliberately strict (0.01): the check should flag clear violations, not manufacture warnings from noise. Non-numeric-only data passes vacuously (order tests need a numeric surface).

Parameters:

• data (Any)

• alpha (float)

• n_perm (int)

• seed (int)

Return type:

ExchangeabilityReport

class ExchangeabilityReport(label, fields=<factory>)

Bases: object

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

Parameters:

• label (str)

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

label: str

fields: list[dict[str, Any]]

property exchangeable: bool

as_dict()

Return type:

dict[str, Any]

class DataSource(*args, **kwargs)

Bases: Protocol

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

schema: Schema | None

structure: SampleStructure

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 type:

Iterable[Any]

materialize()

Return the records as a list, coerced to the schema if one is set.

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)

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 type:

list[Any]

records()

Return type:

Iterable[Any]

partition(n, *, by=None)

Parameters:

• n (int)

• by (Any)

Return type:

list[MaterializedSource]

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

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)

Bases: object

An ordered set of typed fields.

Parameters:

fields (tuple[Field, ...])

fields: tuple[Field, ...]

property names: tuple[str, ...]

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)

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)

Parameters:

value (Any)

Return type:

Any

class Real

Bases: FieldType

A real-valued scalar.

numpy_dtype

alias of float64

coerce(value)

Parameters:

value (Any)

Return type:

float

class Count

Bases: FieldType

A non-negative integer count.

numpy_dtype

alias of int64

coerce(value)

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)

categories: tuple[Any, ...] | None = None

numpy_dtype

alias of object_

coerce(value)

Parameters:

value (Any)

Return type:

Any

class Boolean

Bases: FieldType

A boolean flag.

numpy_dtype

alias of bool

coerce(value)

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)

dim: int | None = None

numpy_dtype

alias of float64

coerce(value)

Parameters:

value (Any)

Return type:

ndarray

class Timestamp

Bases: FieldType

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

numpy_dtype: Any = dtype('<M8[ns]')

coerce(value)

Parameters:

value (Any)

Return type:

Any

class Text

Bases: FieldType

A free-text string.

numpy_dtype

alias of object_

coerce(value)

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)

inner: FieldType

property numpy_dtype: Any

Double-precision floating-point number type, compatible with Python float and C double.

Character code:

'd'

Canonical name:

numpy.double

Alias on this platform (Linux x86_64):

numpy.float64: 64-bit precision floating-point number type: sign bit, 11 bits exponent, 52 bits mantissa.

coerce(value)

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)

schema: Schema

numpy_dtype

alias of object_

coerce(value)

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 | Callable[[Any], Any] | None)

kind: str

by: str | Callable[[Any], Any] | None = 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

partially_exchangeable(by)

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

Parameters:

by (str | Callable[[Any], Any])

Return type:

SampleStructure

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 is mixed in, so a truncated hash never collides with a full one).

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.

Parameters:

• model (Any)

• data (Any)

• sample (int)

• check_support (bool)

• raise_on_error (bool)

Return type:

DataReport

class DataReport(ok: 'bool', n_checked: 'int', schema: 'list[tuple[str, str]]', issues: 'list[str]' = <factory>)

Bases: object

Parameters:

• ok (bool)

• n_checked (int)

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

• issues (list[str])

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 (optional) records the encoder class so a load against a different encoder is flagged.

Parameters:

• encoded (Any)

• path (str)

• encoder (Any)

Return type:

str

load_encoded(path, *, encoder=None)

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

If encoder is given, its class must match the one recorded at save time (else ValueError).

Parameters:

• path (str)

• encoder (Any)

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 the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Any])

Return type:

tuple[GraphObservation, …]

nbytes(x)

Return the approximate in-memory byte size of an encoded payload.

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)

adjacency: ndarray

block_assignments: ndarray | None = None

dataframe_records(df, fields=None, as_dict=False)

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.

Parameters:

• df (Any)

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

• as_dict (bool)

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.

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_seq_as_rdd(sc, dist, seq_len, count_per_split, num_splits, seed=None)

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

Parameters:

• rdd (Any)

• with_replacement (bool)

• n (int)

• seed (int | None)

Subpackages

• mixle.data.sources package
  • Submodules
    • 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.structure module
• mixle.data.validate module