Result artifacts

trade_study.artifacts

Validated result-bundle schemas and primitive JSON input preservation.

ResultsBundle

Bases: NamedTuple

Scored data plus separately preserved scientific interpretation.

Attributes:

Name	Type	Description
results	ResultsTable	Reconstructed parallel score/config/annotation table.
observables	tuple[Observable, ...]	Ordered directions and weights; empty for legacy exports.
provenance	Mapping[str, object]	Caller identities and recorded runtime versions. This does not reconstruct executable simulators/scorers or certify replay.
schema_version	int	One for primitive JSON bundles, zero for legacy export.
generation	str | None	Captured published generation identifier, or None for a directly loaded bundle/legacy export.

load_result_bundle(path)

Load a published generation, direct version-1 bundle or legacy export.

Parameters:

Name	Type	Description	Default
path	str | Path	Published root or direct directory with committed result files.	required

Returns:

Type	Description
ResultsBundle	Scored table and available scientific metadata. Legacy schema version
ResultsBundle	zero has no preserved observable definitions or replayability claim.

Raises:

Type	Description
ValueError	If schema, array digests, parallel rows or metadata disagree.

Source code in src/trade_study/artifacts.py

def load_result_bundle(path: str | Path) -> ResultsBundle:
    """Load a published generation, direct version-1 bundle or legacy export.

    Args:
        path: Published root or direct directory with committed result files.

    Returns:
        Scored table and available scientific metadata. Legacy schema version
        zero has no preserved observable definitions or replayability claim.

    Raises:
        ValueError: If schema, array digests, parallel rows or metadata disagree.
    """
    try:
        return _load_bundle(Path(path))
    except (ValueError, KeyError, TypeError) as error:
        msg = f"Invalid result bundle: {error}"
        raise ValueError(msg) from error

save_result_bundle(results, path, *, observables, provenance=None)

Write a version-1 primitive JSON bundle into a new destination only.

Validate all values before creating the destination. Reserve a new directory, write array snapshots, and atomically publish metadata last. An interrupted write has no committed metadata and cannot read as a complete bundle. This initial schema writer does not overwrite or update existing directories.

Parameters:

Name	Type	Description	Default
results	ResultsTable	Parallel scored data; arrays may contain non-finite score values.	required
path	str | Path	New destination directory, on a filesystem supporting atomic replace.	required
observables	Sequence[Observable]	Ordered scientific directions/weights matching score columns.	required
provenance	dict[str, object] | None	Primitive JSON identities for model/scorer/transforms, external inputs, replication and random streams. Runtime versions are added.	None

Raises:

Type	Description
ValueError	If data or interpretation is inconsistent or values unsupported.
FileExistsError	If the destination already exists, including incomplete data.

Source code in src/trade_study/artifacts.py

def save_result_bundle(
    results: ResultsTable,
    path: str | Path,
    *,
    observables: Sequence[Observable],
    provenance: dict[str, object] | None = None,
) -> None:
    """Write a version-1 primitive JSON bundle into a new destination only.

    Validate all values before creating the destination. Reserve a new directory,
    write array snapshots, and atomically publish metadata last. An interrupted
    write has no committed metadata and cannot read as a complete bundle. This
    initial schema writer does not overwrite or update existing directories.

    Args:
        results: Parallel scored data; arrays may contain non-finite score values.
        path: New destination directory, on a filesystem supporting atomic replace.
        observables: Ordered scientific directions/weights matching score columns.
        provenance: Primitive JSON identities for model/scorer/transforms, external
            inputs, replication and random streams. Runtime versions are added.

    Raises:
        ValueError: If data or interpretation is inconsistent or values unsupported.
        FileExistsError: If the destination already exists, including incomplete data.
    """
    try:
        _validate_table(results)
        meta = {
            "schema_version": 1,
            "encoding": "primitive-json-v1",
            "observable_names": list(results.observable_names),
            "annotation_names": list(results.annotation_names),
            "configs": _json_value(results.configs),
            "metadata": _json_value(results.metadata),
            "observables": _observable_metadata(observables, results.observable_names),
            "provenance": {
                "runtime": {"trade-study": __version__, "numpy": np.__version__},
                "caller": _json_value(
                    _mapping(
                        {} if provenance is None else provenance, "caller provenance"
                    )
                ),
            },
        }
        scores = results.scores.copy()
        annotations = (
            None if results.annotations is None else results.annotations.copy()
        )
    except (ValueError, TypeError, RecursionError) as error:
        msg = f"Cannot save replayable result inputs: {error}"
        raise ValueError(msg) from error
    directory = Path(path)
    if directory.exists():
        msg = "Result-bundle destination must be new; existing data is not overwritten"
        raise FileExistsError(msg)
    directory.mkdir(parents=True, exist_ok=False)
    np.savez_compressed(directory / "scores.npz", scores=scores)
    annotation_description = None
    if annotations is not None:
        np.savez_compressed(directory / "annotations.npz", annotations=annotations)
        annotation_description = _array_description(
            directory / "annotations.npz", annotations
        )
    meta["arrays"] = {
        "scores": _array_description(directory / "scores.npz", scores),
        "annotations": annotation_description,
    }
    with NamedTemporaryFile(
        mode="w", encoding="utf-8", dir=directory, prefix=".meta-", delete=False
    ) as temporary:
        temporary.write(json.dumps(meta, indent=2, allow_nan=False))
        temporary_path = Path(temporary.name)
    temporary_path.replace(directory / "meta.json")

publish_result_bundle(results, path, *, observables, provenance=None)

Publish an immutable generation with one atomic current-pointer replacement.

Existing generations and direct/legacy files stay intact. A failed update leaves the previous pointer unchanged unless the new complete generation has already been committed. Readers capture one pointer and check all files in that selected generation. Concurrent writers use distinct destinations; the last successful pointer replacement determines the current generation.

Parameters:

Name	Type	Description	Default
results	ResultsTable	Scored data to snapshot into a new version-1 generation.	required
path	str | Path	Publication root on a filesystem supporting atomic file replacement.	required
observables	Sequence[Observable]	Ordered directions/weights matching the score columns.	required
provenance	dict[str, object] | None	Primitive JSON caller identities, separate from runtime versions.	None

Returns:

Type	Description
Path	Retained generation directory, usable for an explicit snapshot read.

Raises:

Type	Description
ValueError	If current publication metadata or new inputs are inconsistent.

Source code in src/trade_study/artifacts.py

def publish_result_bundle(
    results: ResultsTable,
    path: str | Path,
    *,
    observables: Sequence[Observable],
    provenance: dict[str, object] | None = None,
) -> Path:
    """Publish an immutable generation with one atomic current-pointer replacement.

    Existing generations and direct/legacy files stay intact. A failed update
    leaves the previous pointer unchanged unless the new complete generation
    has already been committed. Readers capture one pointer and check all files
    in that selected generation. Concurrent writers use distinct destinations;
    the last successful pointer replacement determines the current generation.

    Args:
        results: Scored data to snapshot into a new version-1 generation.
        path: Publication root on a filesystem supporting atomic file replacement.
        observables: Ordered directions/weights matching the score columns.
        provenance: Primitive JSON caller identities, separate from runtime versions.

    Returns:
        Retained generation directory, usable for an explicit snapshot read.

    Raises:
        ValueError: If current publication metadata or new inputs are inconsistent.
    """
    root = Path(path)
    try:
        generation = _new_generation(root)
        save_result_bundle(
            results, generation, observables=observables, provenance=provenance
        )
        _commit_generation(root, generation)
    except ValueError as error:
        msg = f"Cannot publish result bundle: {error}"
        raise ValueError(msg) from error
    return generation