Skip to content

Assay choices with an external cost table

This synthetic example compares categorical assay labels and continuous classification thresholds, maximizing accuracy and minimizing unit cost. All costs and response distributions are invented; the assay labels are illustrative and the simulation does not represent measured clinical performance.

Run the complete example from the repository:

uv run --extra examples python examples/assay_study.py

The script is examples/assay_study.py.

Attach external costs

A lookup table associates each categorical assay with a unit cost:

UNIT_COST = {"PCR": 18.0, "ELISA": 8.0, "rapid": 3.0}
annotations = [
    Annotation(name="unit_cost", lookup=UNIT_COST, key="assay")
]

For {"assay": "ELISA", "threshold": 0.5}, Annotation.resolve() reads the assay value and looks up 8.0. Annotations populate a separate matrix alongside the score columns. They can record external costs or other information without changing what the simulator generates. Using a distinct annotation name also keeps DataFrame exports clear.

Make cost part of the decision

An annotation alone does not add a Pareto objective. To optimize cost, the scorer returns a cost score from the same table and the study declares its direction:

observables = [
    Observable("accuracy", Direction.MAXIMIZE),
    Observable("cost", Direction.MINIMIZE),
]

# Inside AssayScorer.score(...):
return {
    "accuracy": float(np.mean(truth == observations)),
    "cost": UNIT_COST[config["assay"]],
}

cost participates in Pareto sorting, while unit_cost remains an externally resolved annotation. Sharing the table avoids maintaining two conflicting cost sources. These are per-assay costs; if total study expense is the objective, compute that quantity explicitly in the scorer instead.

Screen, then refine retained designs

The grid combines three assay labels with three initial thresholds. The categorical factor describes the labels; the threshold factor declares a continuous range, even though this initial grid uses three explicit points.

The screening phase evaluates each design with three replicates of 300 synthetic cases. top_k_pareto_filter(4) retains four designs using their mean objectives. The refinement phase uses grid="carry" and a larger simulator with 3,000 cases per replicate, reducing Monte Carlo variability while keeping both objectives and the cost annotation.

phases = [
    Phase("screen", grid=grid, filter_fn=top_k_pareto_filter(4), n_reps=3),
    Phase("refine", grid="carry", world=AssayWorld(n_cases=3000), n_reps=3),
]

The simulator accepts rep and uses seed + rep. Within a phase, matching replicate ids share the synthetic case/noise stream across assay and threshold choices. The scorer measures classification accuracy against those synthetic truth labels; the external cost stays fixed for an assay.

Inspect annotations and means

Raw phase results retain one row per replicate. Aggregate them for a per-design report and access the annotation matrix separately:

results = study.results("refine").aggregate_replicates()
print(results.observable_names)  # ["accuracy", "cost"]
print(results.annotation_names)  # ["unit_cost"]
print(results.annotations)       # one unit-cost value per design

The runnable script prints each retained assay/threshold, mean accuracy, cost objective and unit-cost annotation. The cost values in both paths agree, while accuracy varies with the simulated readouts and threshold. Retained designs are trade-off alternatives; the front does not specify a single final preference.

Complete source

"""Synthetic assay accuracy/cost study with external cost annotations.

All costs and response distributions are invented for this API example;
assay labels do not imply measured clinical performance.

Run: uv run --extra examples python examples/assay_study.py
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

import numpy as np

from trade_study import (
    Annotation,
    Direction,
    Factor,
    FactorType,
    Observable,
    Phase,
    Study,
    top_k_pareto_filter,
)

if TYPE_CHECKING:
    from numpy.typing import NDArray

UNIT_COST = {"PCR": 18.0, "ELISA": 8.0, "rapid": 3.0}
_NOISE = {"PCR": 0.18, "ELISA": 0.28, "rapid": 0.38}


@dataclass(frozen=True)
class AssayWorld:
    """Invented noisy readouts for binary synthetic truth labels."""

    n_cases: int = 300
    seed: int = 17

    def generate(
        self, config: dict[str, Any], *, rep: int = 0
    ) -> tuple[NDArray[np.bool_], NDArray[np.bool_]]:
        """Generate a replicate with the same case/noise stream across assays.

        Args:
            config: Assay label and classification threshold.
            rep: Replicate id, offsetting the simulation seed.

        Returns:
            Boolean truth and predicted labels for synthetic cases.
        """
        rng = np.random.default_rng(self.seed + rep)
        truth = rng.random(self.n_cases) < 0.3
        noise = rng.standard_normal(self.n_cases)
        signal = truth.astype(float) + _NOISE[config["assay"]] * noise
        observed = signal >= config["threshold"]
        return truth, observed


class AssayScorer:
    """Return accuracy and unit cost as separate optimization objectives."""

    @staticmethod
    def score(
        truth: NDArray[np.bool_],
        observations: NDArray[np.bool_],
        config: dict[str, Any],
    ) -> dict[str, float]:
        """Score the predictions and promote external cost into an objective.

        Args:
            truth: Synthetic reference labels.
            observations: Synthetic predicted labels.
            config: Assay and threshold values for this evaluation.

        Returns:
            Accuracy to maximize and per-assay cost to minimize.
        """
        return {
            "accuracy": float(np.mean(truth == observations)),
            "cost": UNIT_COST[config["assay"]],
        }


def main() -> None:
    """Screen assays and thresholds, then refine retained designs with more cases.

    Raises:
        RuntimeError: If the study unexpectedly omits cost annotations.
    """
    factors = [
        Factor("assay", FactorType.CATEGORICAL, levels=list(UNIT_COST)),
        Factor("threshold", FactorType.CONTINUOUS, bounds=(0.2, 0.8)),
    ]
    grid = [
        {"assay": assay, "threshold": threshold}
        for assay in UNIT_COST
        for threshold in (0.35, 0.5, 0.65)
    ]
    study = Study(
        world=AssayWorld(),
        scorer=AssayScorer(),
        observables=[
            Observable("accuracy", Direction.MAXIMIZE),
            Observable("cost", Direction.MINIMIZE),
        ],
        factors=factors,
        annotations=[Annotation(name="unit_cost", lookup=UNIT_COST, key="assay")],
        phases=[
            Phase("screen", grid=grid, filter_fn=top_k_pareto_filter(4), n_reps=3),
            Phase("refine", grid="carry", world=AssayWorld(n_cases=3000), n_reps=3),
        ],
    )
    study.run()
    results = study.results("refine").aggregate_replicates()
    print("Refined designs: accuracy, cost objective, external unit-cost annotation")
    if results.annotations is None:
        msg = "Expected unit-cost annotations"
        raise RuntimeError(msg)
    for config, scores, external_values in zip(
        results.configs, results.scores, results.annotations, strict=True
    ):
        print(
            f"{config['assay']:5s} threshold={config['threshold']:.2f} "
            f"accuracy={scores[0]:.3f} cost={scores[1]:.2f} "
            f"unit_cost={external_values[0]:.2f}"
        )


if __name__ == "__main__":
    main()