Local information screening

See the complete example and qualification scope.

trade_study.FisherContext(parameter_names, parameter_coordinates, parameter_scales, observation_model, design, reference_regime, nuisance_treatment, sensitivity_reference, numerical_reference, qualification_reference, information_reference) dataclass

Caller-owned meaning and qualification of an information matrix.

All references are nonempty caller identifiers whose underlying evidence must be retained. They are declarations, not automatically verified hashes. The supplied matrix is already in parameter_coordinates with the recorded parameter_scales; this adapter applies no further scaling.

Attributes:

Name	Type	Description
parameter_names	tuple[str, ...]	Ordered matrix coordinates, distinct and nonempty.
parameter_coordinates	str	Coordinate/transform definition and units.
parameter_scales	tuple[float, ...]	Positive finite scales already used by the caller.
observation_model	str	Measurement law, noise and observation definitions.
design	str	Experimental design identity.
reference_regime	str	Parameter point or declared sampled regime.
nuisance_treatment	str	Explicit conditioning/profiling convention.
sensitivity_reference	str	Measured-output sensitivity source.
numerical_reference	str	Solver, derivative and numerical settings.
qualification_reference	str	Independent qualification and its scope.
information_reference	str	Identity of the caller's information artifact.

__post_init__()

Detach coordinate sequences and require explicit context.

Raises:

Type	Description
ValueError	If names, scales or context references are invalid.

Source code in src/trade_study/fisher.py

def __post_init__(self) -> None:
    """Detach coordinate sequences and require explicit context.

    Raises:
        ValueError: If names, scales or context references are invalid.
    """
    names, scales = tuple(self.parameter_names), tuple(self.parameter_scales)
    references = (
        self.parameter_coordinates,
        self.observation_model,
        self.design,
        self.reference_regime,
        self.nuisance_treatment,
        self.sensitivity_reference,
        self.numerical_reference,
        self.qualification_reference,
        self.information_reference,
    )
    valid_names = (
        not isinstance(self.parameter_names, str)
        and bool(names)
        and all(_label(name) for name in names)
        and len(set(names)) == len(names)
    )
    valid_scales = len(scales) == len(names) and all(
        _nonnegative(scale) and scale > 0 for scale in scales
    )
    if (
        not valid_names
        or not valid_scales
        or not all(_label(r) for r in references)
    ):
        msg = (
            "Fisher context requires distinct names, positive scales and references"
        )
        raise ValueError(msg)
    object.__setattr__(self, "parameter_names", names)
    object.__setattr__(self, "parameter_scales", tuple(float(s) for s in scales))

to_metadata()

Return detached primitive JSON context.

Returns:

Type	Description
dict[str, object]	Explicit coordinates and caller-owned evidence references.

Source code in src/trade_study/fisher.py

def to_metadata(self) -> dict[str, object]:
    """Return detached primitive JSON context.

    Returns:
        Explicit coordinates and caller-owned evidence references.
    """
    return {
        "parameter_names": list(self.parameter_names),
        "parameter_coordinates": self.parameter_coordinates,
        "parameter_scales": list(self.parameter_scales),
        "observation_model": self.observation_model,
        "design": self.design,
        "reference_regime": self.reference_regime,
        "nuisance_treatment": self.nuisance_treatment,
        "sensitivity_reference": self.sensitivity_reference,
        "numerical_reference": self.numerical_reference,
        "qualification_reference": self.qualification_reference,
        "information_reference": self.information_reference,
    }

trade_study.FisherScreenPolicy(threshold, directions=None, atol=1e-10, rtol=1e-08, max_parameters=128) dataclass

Prespecified local minimum-eigenvalue screen.

Attributes:

Name	Type	Description
threshold	float	Required minimum eigenvalue, finite and nonnegative.
directions	tuple[tuple[float, ...], ...] | None	Optional row-major p-by-k orthonormal direction basis in the matrix coordinates. The tested matrix is B.T @ F @ B. Its complement is fixed; this operation does not profile nuisance parameters. None tests all directions.
atol	float	Absolute symmetry and positive-semidefiniteness tolerance.
rtol	float	Relative symmetry and positive-semidefiniteness tolerance.
max_parameters	int	Dense matrix dimension limit, default 128. This is not a whole-process memory bound.

__post_init__()

Validate the policy before evaluating any design.

Raises:

Type	Description
ValueError	If bounds or the orthonormal basis are invalid.

Source code in src/trade_study/fisher.py

def __post_init__(self) -> None:
    """Validate the policy before evaluating any design.

    Raises:
        ValueError: If bounds or the orthonormal basis are invalid.
    """
    if (
        not all(_nonnegative(v) for v in (self.threshold, self.atol, self.rtol))
        or isinstance(self.max_parameters, bool)
        or not isinstance(self.max_parameters, Integral)
        or self.max_parameters < 1
    ):
        msg = (
            "Fisher policy requires nonnegative finite bounds and a positive limit"
        )
        raise ValueError(msg)
    if self.directions is not None:
        basis = np.asarray(self.directions)
        valid_shape = (
            basis.ndim == 2
            and 0 < basis.shape[1] <= basis.shape[0] <= self.max_parameters
            and basis.dtype.kind in "iuf"
        )
        if not valid_shape:
            msg = "directions must be a finite real p-by-k orthonormal basis"
            raise ValueError(msg)
        # Validate norms in float64, avoiding integer Gram-matrix overflow.
        with np.errstate(over="ignore", invalid="ignore"):
            basis = np.asarray(basis, dtype=np.float64)
            gram = basis.T @ basis
        if (
            not np.isfinite(basis).all()
            or not np.isfinite(gram).all()
            or not np.allclose(gram, np.eye(basis.shape[1]), rtol=1e-10, atol=1e-10)
        ):
            msg = "directions must be a finite real p-by-k orthonormal basis"
            raise ValueError(msg)
        object.__setattr__(
            self, "directions", tuple(tuple(float(v) for v in row) for row in basis)
        )

constraint(observable='fisher_min_eigenvalue')

Build the existing deterministic feasibility constraint.

Parameters:

Name	Type	Description	Default
observable	str	Scored column name.	'fisher_min_eigenvalue'

Returns:

Type	Description
Constraint	A >= constraint without an invented confidence level.

Raises:

Type	Description
ValueError	If the observable name is empty.

Source code in src/trade_study/fisher.py

def constraint(self, observable: str = "fisher_min_eigenvalue") -> Constraint:
    """Build the existing deterministic feasibility constraint.

    Args:
        observable: Scored column name.

    Returns:
        A >= constraint without an invented confidence level.

    Raises:
        ValueError: If the observable name is empty.
    """
    if not _label(observable):
        msg = "observable must be nonempty"
        raise ValueError(msg)
    return Constraint("local_information", observable, ">=", float(self.threshold))

to_metadata()

Return detached primitive JSON policy values.

Returns:

Type	Description
dict[str, object]	Threshold, basis, numerical tolerances and dense dimension limit.

Source code in src/trade_study/fisher.py

def to_metadata(self) -> dict[str, object]:
    """Return detached primitive JSON policy values.

    Returns:
        Threshold, basis, numerical tolerances and dense dimension limit.
    """
    return {
        "threshold": float(self.threshold),
        "directions": None
        if self.directions is None
        else [list(row) for row in self.directions],
        "atol": float(self.atol),
        "rtol": float(self.rtol),
        "max_parameters": int(self.max_parameters),
    }

trade_study.FisherScreenResult(status, reason, eigenvalues, matrix_digest, context, policy) dataclass

A local screen outcome retaining its context and numerical evidence.

Attributes:

Name	Type	Description
status	ScreenStatus	Feasible, infeasible, unreadable or failed.
reason	str	Machine-readable outcome explanation or upstream failure text.
eigenvalues	tuple[float, ...]	Actual tested eigenvalues; empty for unavailable results.
matrix_digest	str | None	SHA256 of the supplied host float64 little-endian matrix bytes in C order, before symmetrization; None for missing inputs.
context	FisherContext	Caller-owned scientific and numerical declarations.
policy	FisherScreenPolicy	The prespecified screen.

as_scores(observable='fisher_min_eigenvalue')

Expose a readable result to an existing scorer/constraint.

Parameters:

Name	Type	Description	Default
observable	str	Scored column name.	'fisher_min_eigenvalue'

Returns:

Type	Description
dict[str, float]	The actual minimum tested eigenvalue in a single scored column.

Raises:

Type	Description
ValueError	If the result is unavailable or the name is empty.

Source code in src/trade_study/fisher.py

def as_scores(self, observable: str = "fisher_min_eigenvalue") -> dict[str, float]:
    """Expose a readable result to an existing scorer/constraint.

    Args:
        observable: Scored column name.

    Returns:
        The actual minimum tested eigenvalue in a single scored column.

    Raises:
        ValueError: If the result is unavailable or the name is empty.
    """
    if self.status not in {"feasible", "infeasible"} or not _label(observable):
        msg = (
            "Only readable Fisher results with a nonempty observable can be scored"
        )
        raise ValueError(msg)
    return {observable: min(self.eigenvalues)}

to_metadata()

Return a result-bundle-compatible record without nonfinite sentinels.

Returns:

Type	Description
dict[str, object]	Versioned primitive JSON with actual values and preserved context.

Source code in src/trade_study/fisher.py

def to_metadata(self) -> dict[str, object]:
    """Return a result-bundle-compatible record without nonfinite sentinels.

    Returns:
        Versioned primitive JSON with actual values and preserved context.
    """
    return {
        "schema_version": 1,
        "scope": "local_directional_information_screen",
        "status": self.status,
        "reason": self.reason,
        "eigenvalues": list(self.eigenvalues),
        "matrix_digest": self.matrix_digest,
        "context": self.context.to_metadata(),
        "policy": self.policy.to_metadata(),
    }

trade_study.screen_fisher(information, *, context, policy, failure=None)

Screen a caller-qualified host information matrix in declared directions.

No derivatives, Fisher model, scaling, nuisance profile, statistical uncertainty or global identifiability conclusion is constructed here. The caller explicitly transfers native backend artifacts to host NumPy.

Parameters:

Name	Type	Description	Default
information	NDArray[float64] | None	Real host NumPy p-by-p matrix, or None when unavailable.	required
context	FisherContext	Ordered coordinates and retained qualification references.	required
policy	FisherScreenPolicy	Prespecified threshold, directions and numerical limits.	required
failure	str | None	Upstream failure reason, supplied only with information=None.	None

Returns:

Type	Description
FisherScreenResult	A readable local result, an unreadable invalid/unsupported input, or
FisherScreenResult	a failed upstream/eigensolver record. No input matrix is retained.

Raises:

Type	Description
ValueError	If failure is empty or supplied alongside a matrix.

Source code in src/trade_study/fisher.py

def screen_fisher(  # ruff: ignore[too-many-return-statements]
    information: NDArray[np.float64] | None,
    *,
    context: FisherContext,
    policy: FisherScreenPolicy,
    failure: str | None = None,
) -> FisherScreenResult:
    """Screen a caller-qualified host information matrix in declared directions.

    No derivatives, Fisher model, scaling, nuisance profile, statistical
    uncertainty or global identifiability conclusion is constructed here.
    The caller explicitly transfers native backend artifacts to host NumPy.

    Args:
        information: Real host NumPy p-by-p matrix, or None when unavailable.
        context: Ordered coordinates and retained qualification references.
        policy: Prespecified threshold, directions and numerical limits.
        failure: Upstream failure reason, supplied only with information=None.

    Returns:
        A readable local result, an unreadable invalid/unsupported input, or
        a failed upstream/eigensolver record. No input matrix is retained.

    Raises:
        ValueError: If failure is empty or supplied alongside a matrix.
    """

    def result(
        status: ScreenStatus, reason: str, values: tuple[float, ...] = ()
    ) -> FisherScreenResult:
        return FisherScreenResult(status, reason, values, digest, context, policy)

    digest = None
    if failure is not None:
        if information is not None or not _label(failure):
            msg = "An upstream failure needs a reason and no information matrix"
            raise ValueError(msg)
        return result("failed", failure)
    if information is None:
        return result("unreadable", "missing_information")
    matrix, reason, digest, tolerance = _prepare_information(
        information, context, policy
    )
    if matrix is None:
        return result("unreadable", reason)
    p = len(context.parameter_names)
    basis = None if policy.directions is None else np.asarray(policy.directions)
    if basis is not None and basis.shape[0] != p:
        return result("unreadable", "direction_coordinate_mismatch")
    try:
        full_values = np.linalg.eigvalsh(matrix)
        with np.errstate(over="ignore", invalid="ignore"):
            values = (
                full_values
                if basis is None
                else np.linalg.eigvalsh(basis.T @ matrix @ basis)
            )
    except np.linalg.LinAlgError:
        return result("failed", "eigensolver_failed")
    if not np.isfinite(full_values).all() or not np.isfinite(values).all():
        return result("unreadable", "nonfinite_spectrum")
    if float(full_values[0]) < -tolerance:
        return result("unreadable", "indefinite_information")
    status: ScreenStatus = (
        "feasible" if float(values[0]) >= policy.threshold else "infeasible"
    )
    return result(status, "threshold_comparison", tuple(float(v) for v in values))

trade_study.fisher_screen_counts(records, *, planned_ids)

Count outcomes against a fixed declared cohort without dropping failures.

Parameters:

Name	Type	Description	Default
records	Mapping[str, FisherScreenResult]	At most one outcome per planned identifier.	required
planned_ids	Sequence[str]	Complete, distinct, nonempty planned replicate/task IDs.	required

Returns:

Type	Description
dict[str, int]	Planned, feasible, infeasible, unreadable, failed and missing counts.
dict[str, int]	These counts alone establish no probability, interval or gate verdict.

Raises:

Type	Description
ValueError	If identities or result statuses are invalid.

Source code in src/trade_study/fisher.py

def fisher_screen_counts(
    records: Mapping[str, FisherScreenResult], *, planned_ids: Sequence[str]
) -> dict[str, int]:
    """Count outcomes against a fixed declared cohort without dropping failures.

    Args:
        records: At most one outcome per planned identifier.
        planned_ids: Complete, distinct, nonempty planned replicate/task IDs.

    Returns:
        Planned, feasible, infeasible, unreadable, failed and missing counts.
        These counts alone establish no probability, interval or gate verdict.

    Raises:
        ValueError: If identities or result statuses are invalid.
    """
    ids = tuple(planned_ids)
    if (
        isinstance(planned_ids, str)
        or not ids
        or not all(_label(identity) for identity in ids)
        or len(set(ids)) != len(ids)
        or set(records) - set(ids)
    ):
        msg = "Screen counts require a complete distinct planned-ID cohort"
        raise ValueError(msg)
    counts = dict.fromkeys(("feasible", "infeasible", "unreadable", "failed"), 0)
    for record in records.values():
        if not isinstance(record, FisherScreenResult) or record.status not in counts:
            msg = "Screen counts require valid FisherScreenResult statuses"
            raise ValueError(msg)
        counts[record.status] += 1
    return {"planned": len(ids), **counts, "missing": len(ids) - len(records)}