I/O

Serialization and deserialization of study results.

save_results() supports replacing a result in an existing directory. It replaces scores and metadata, writes the new annotation array when present, and removes any previous annotations.npz when the replacement is unannotated. Unrelated files remain. These writes are not atomic across the result files; use separate output directories if readers need an unchanged previous bundle during a save.

load_results() also reads versioned result bundles, returning their scored table. Both readers now reject inconsistent parallel rows, names, annotation dimensions and metadata cardinality. Versioned files also require matching digests/shapes/dtypes. For preserved directions and provenance, use load_result_bundle(). Legacy save_results() refuses a committed versioned destination. It also refuses published roots. Readers follow one captured current-generation pointer when present; see the artifact guide for publication and retention.

trade_study.load_results(path)

Load a ResultsTable from disk.

Parameters:

Name	Type	Description	Default
path	str | Path	Directory previously written by save_results.	required

Returns:

Type	Description
ResultsTable	Reconstructed ResultsTable.

Raises:

Type	Description
ValueError	If parallel arrays/metadata or a versioned bundle are inconsistent.

Source code in src/trade_study/io.py

def load_results(path: str | Path) -> ResultsTable:
    """Load a ResultsTable from disk.

    Args:
        path: Directory previously written by ``save_results``.

    Returns:
        Reconstructed ResultsTable.

    Raises:
        ValueError: If parallel arrays/metadata or a versioned bundle are inconsistent.
    """
    try:
        return load_result_bundle(path).results
    except ValueError as error:
        msg = f"Cannot load study results: {error}"
        raise ValueError(msg) from error

trade_study.EvaluationCache(path, *, revision, replicate_namespace, fidelity)

Persistent immutable evaluation evidence for explicitly identified runs.

Create or reopen an evaluation cache.

Parameters:

Name	Type	Description	Default
path	str | Path	SQLite cache file; compatible contexts may share one file.	required
revision	str	Caller-managed simulator/scorer/data/annotation revision. Include every behavior input not identifiable from class code.	required
replicate_namespace	str	Explicit randomness/seed experiment identity. Use a new namespace for genuinely independent replication.	required
fidelity	str	Explicit simulation/evaluation fidelity identity.	required

Raises:

Type	Description
ValueError	If an identity is empty or the cache format is incompatible.

Source code in src/trade_study/cache.py

def __init__(
    self,
    path: str | Path,
    *,
    revision: str,
    replicate_namespace: str,
    fidelity: str,
) -> None:
    """Create or reopen an evaluation cache.

    Args:
        path: SQLite cache file; compatible contexts may share one file.
        revision: Caller-managed simulator/scorer/data/annotation revision.
            Include every behavior input not identifiable from class code.
        replicate_namespace: Explicit randomness/seed experiment identity.
            Use a new namespace for genuinely independent replication.
        fidelity: Explicit simulation/evaluation fidelity identity.

    Raises:
        ValueError: If an identity is empty or the cache format is incompatible.
    """
    if any(
        not value.strip() for value in (revision, replicate_namespace, fidelity)
    ):
        msg = "revision, replicate_namespace and fidelity must be nonempty"
        raise ValueError(msg)
    self.path = Path(path)
    self.revision = revision
    self.replicate_namespace = replicate_namespace
    self.fidelity = fidelity
    self.path.parent.mkdir(parents=True, exist_ok=True)
    with closing(sqlite3.connect(self.path, timeout=30)) as connection, connection:
        connection.execute(
            "CREATE TABLE IF NOT EXISTS cache_format (version INTEGER)"
        )
        stored = connection.execute("SELECT version FROM cache_format").fetchone()
        if stored is None:
            connection.execute("INSERT INTO cache_format VALUES (1)")
        elif stored[0] != 1:
            msg = "Incompatible evaluation cache format"
            raise ValueError(msg)
        connection.execute(
            "CREATE TABLE IF NOT EXISTS evaluations "
            "(key TEXT PRIMARY KEY, payload TEXT NOT NULL)"
        )

clear()

Invalidate all stored evaluations across every context in this file.

Source code in src/trade_study/cache.py

def clear(self) -> None:
    """Invalidate all stored evaluations across every context in this file."""
    with closing(sqlite3.connect(self.path, timeout=30)) as connection, connection:
        connection.execute("DELETE FROM evaluations")

trade_study.save_results(results, path)

Save a ResultsTable to disk.

Uses .npz for score arrays and .json for configs/metadata. Saving into an existing directory replaces the result files and removes previously saved annotations when the new table has none. Unrelated files are retained. The multi-file write is not an atomic publication operation.

Parameters:

Name	Type	Description	Default
results	ResultsTable	The ResultsTable to save.	required
path	str | Path	Directory to write into (created if needed).	required

Raises:

Type	Description
ValueError	If the destination contains a versioned bundle or publication.

Source code in src/trade_study/io.py

def save_results(results: ResultsTable, path: str | Path) -> None:
    """Save a ResultsTable to disk.

    Uses .npz for score arrays and .json for configs/metadata.
    Saving into an existing directory replaces the result files and removes
    previously saved annotations when the new table has none. Unrelated files
    are retained. The multi-file write is not an atomic publication operation.

    Args:
        results: The ResultsTable to save.
        path: Directory to write into (created if needed).

    Raises:
        ValueError: If the destination contains a versioned bundle or publication.
    """
    p = Path(path)
    if (p / "current.json").exists():
        msg = "Legacy export cannot overwrite a published result namespace"
        raise ValueError(msg)
    existing_meta = p / "meta.json"
    if existing_meta.exists():
        try:
            existing = json.loads(existing_meta.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            # Legacy exports can have interrupted metadata writes; retain
            # their established ability to replace an invalid legacy export.
            existing = None
        if isinstance(existing, dict) and "schema_version" in existing:
            msg = "Legacy export cannot overwrite a versioned result bundle"
            raise ValueError(msg)
    p.mkdir(parents=True, exist_ok=True)

    np.savez_compressed(p / "scores.npz", scores=results.scores)
    if results.annotations is not None:
        np.savez_compressed(p / "annotations.npz", annotations=results.annotations)
    else:
        (p / "annotations.npz").unlink(missing_ok=True)

    meta = {
        "observable_names": results.observable_names,
        "annotation_names": results.annotation_names,
        "configs": results.configs,
        "metadata": results.metadata,
    }
    (p / "meta.json").write_text(json.dumps(meta, default=str, indent=2))