Skip to content

Ascertainment

Ascertainment models for shared observation-rate structure.

AscertainmentModel 

AscertainmentModel(name: str, signals: tuple[str, ...])

Base class for shared ascertainment structure.

An ascertainment rate is the probability that latent incidence is observed in a particular data stream. Examples include an infection-hospitalization ratio for hospital admissions or an infection-ED-visit ratio for emergency department visits.

AscertainmentModel objects make shared structure explicit in a model specification. A user defines the shared model once, registers it with PyrenewBuilder.add_ascertainment(...), and passes signal-specific accessors into observation processes:

ascertainment = JointAscertainment(...)
builder.add_ascertainment(ascertainment)

PopulationCounts(
    name="hospital",
    ascertainment_rate_rv=ascertainment.for_signal("hospital"),
    ...
)

Subclasses own any NumPyro sites needed for the shared structure. Accessors returned by for_signal() read the sampled values from the active model context and do not sample independently.

Initialize an ascertainment model.

Parameters:

Name Type Description Default
name str

A non-empty string identifying the ascertainment model.

 required
signals tuple[str, ...]

Unique signal names produced by this model.

 required
Source code in pyrenew/ascertainment/base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def __init__(
    self,
    name: str,
    signals: tuple[str, ...],
) -> None:
    """
    Initialize an ascertainment model.

    Parameters
    ----------
    name
        A non-empty string identifying the ascertainment model.
    signals
        Unique signal names produced by this model.
    """
    if not isinstance(name, str) or len(name) == 0:
        raise ValueError(
            f"name must be a non-empty string. Got {type(name).__name__}: {name!r}"
        )
    if not isinstance(signals, tuple) or len(signals) == 0:
        raise ValueError("signals must be a non-empty tuple of strings.")
    if any(not isinstance(signal, str) or len(signal) == 0 for signal in signals):
        raise ValueError("all signals must be non-empty strings.")
    if len(set(signals)) != len(signals):
        raise ValueError("signals must be unique.")

    self.name = name
    self.signals = signals

for_signal 

for_signal(signal_name: str) -> AscertainmentSignal

Return an observation-process accessor for one signal.

Parameters:

Name Type Description Default
signal_name str

Name of the signal produced by this ascertainment model. This name should match the signal name used when the ascertainment model was constructed. It does not have to match the observation process name, but using the same name usually makes model specifications easier to read.

 required

Returns:

Type Description
AscertainmentSignal

RandomVariable-compatible accessor for the signal's sampled ascertainment rate.

Raises:

Type Description
ValueError

If signal_name is not produced by this model.
Source code in pyrenew/ascertainment/base.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def for_signal(self, signal_name: str) -> AscertainmentSignal:
    """
    Return an observation-process accessor for one signal.

    Parameters
    ----------
    signal_name
        Name of the signal produced by this ascertainment model. This name
        should match the signal name used when the ascertainment model was
        constructed. It does not have to match the observation process name,
        but using the same name usually makes model specifications easier
        to read.

    Returns
    -------
    AscertainmentSignal
        RandomVariable-compatible accessor for the signal's sampled
        ascertainment rate.

    Raises
    ------
    ValueError
        If ``signal_name`` is not produced by this model.
    """
    if signal_name not in self.signals:
        raise ValueError(
            f"Unknown signal {signal_name!r} for ascertainment model "
            f"{self.name!r}. Available signals: {self.signals}."
        )
    return AscertainmentSignal(
        ascertainment_name=self.name,
        signal_name=signal_name,
    )

sample abstractmethod 

sample(**kwargs: object) -> Mapping[str, ArrayLike]

Sample all signal-specific ascertainment values owned by this model.

Parameters:

Name Type Description Default
**kwargs object

Additional model-context arguments supplied by MultiSignalModel. Subclasses may ignore unused values.

 {}

Returns:

Type Description
Mapping[str, ArrayLike]

Mapping from signal name to sampled ascertainment rate.
Source code in pyrenew/ascertainment/base.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@abstractmethod
def sample(self, **kwargs: object) -> Mapping[str, ArrayLike]:
    """
    Sample all signal-specific ascertainment values owned by this model.

    Parameters
    ----------
    **kwargs
        Additional model-context arguments supplied by ``MultiSignalModel``.
        Subclasses may ignore unused values.

    Returns
    -------
    Mapping[str, ArrayLike]
        Mapping from signal name to sampled ascertainment rate.
    """
    pass  # pragma: no cover

AscertainmentSignal 

AscertainmentSignal(ascertainment_name: str, signal_name: str)

Bases: RandomVariable

Accessor for one signal's ascertainment value.

Users usually do not instantiate this class directly. It is returned by AscertainmentModel.for_signal(...) and passed to an observation process as ascertainment_rate_rv. During model execution, the parent AscertainmentModel samples the actual rate once, and this accessor retrieves the signal-specific value without creating additional NumPyro sample sites.

Initialize a signal-specific ascertainment accessor.

Parameters:

Name Type Description Default
ascertainment_name str

Name of the parent ascertainment model.

 required
signal_name str

Name of the signal to retrieve.

 required
Source code in pyrenew/ascertainment/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    ascertainment_name: str,
    signal_name: str,
) -> None:
    """
    Initialize a signal-specific ascertainment accessor.

    Parameters
    ----------
    ascertainment_name
        Name of the parent ascertainment model.
    signal_name
        Name of the signal to retrieve.
    """
    if not isinstance(ascertainment_name, str) or len(ascertainment_name) == 0:
        raise ValueError(
            "ascertainment_name must be a non-empty string. "
            f"Got {type(ascertainment_name).__name__}: {ascertainment_name!r}"
        )
    if not isinstance(signal_name, str) or len(signal_name) == 0:
        raise ValueError(
            "signal_name must be a non-empty string. "
            f"Got {type(signal_name).__name__}: {signal_name!r}"
        )
    super().__init__(name=f"{ascertainment_name}_{signal_name}")
    self.ascertainment_name = ascertainment_name
    self.signal_name = signal_name

sample 

sample(**kwargs: object) -> ArrayLike

Return the sampled ascertainment value for this signal.

Parameters:

Name Type Description Default
**kwargs object

Additional keyword arguments, ignored.

 {}

Returns:

Type Description
ArrayLike

Signal-specific ascertainment value from the active context.
Source code in pyrenew/ascertainment/base.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def sample(self, **kwargs: object) -> ArrayLike:
    """
    Return the sampled ascertainment value for this signal.

    Parameters
    ----------
    **kwargs
        Additional keyword arguments, ignored.

    Returns
    -------
    ArrayLike
        Signal-specific ascertainment value from the active context.
    """
    return get_ascertainment_value(
        ascertainment_name=self.ascertainment_name,
        signal_name=self.signal_name,
    )

JointAscertainment 

JointAscertainment(
    name: str,
    signals: tuple[str, ...],
    baseline_rates: ArrayLike,
    scale_tril: ArrayLike | None = None,
    covariance_matrix: ArrayLike | None = None,
    precision_matrix: ArrayLike | None = None,
)

Bases: AscertainmentModel

Joint prior for scalar ascertainment rates across multiple signals.

This model is useful when multiple observation streams have distinct but related probabilities of observing latent incidence. For example, hospital admissions and emergency department visits may have different infection-to-observation ratios, while still being correlated because both depend on care-seeking behavior, testing practices, or reporting systems.

The model samples one logit multivariate normal vector given natural-scale baseline ascertainment rates.

eta ~ MultivariateNormal(logit(baseline_rates), covariance)
ascertainment_rate_j = sigmoid(eta_j)

Each returned rate is scalar and constant over the model time axis.

Initialize a joint scalar ascertainment model.

Parameters:

Name Type Description Default
name str

Name of the ascertainment model.

 required
signals tuple[str, ...]

Unique signal names, such as ("hospital", "ed_visits"). The order corresponds to entries in baseline_rates and the covariance parameter.

 required
baseline_rates ArrayLike

Natural-scale baseline ascertainment rates. Shape (n_signals,). Values must be probabilities in (0, 1). A value of 0.01 centers the corresponding ascertainment rate near 1 percent before accounting for covariance.

 required
scale_tril ArrayLike | None

Lower-triangular scale matrix for the multivariate normal on the logit scale. Exactly one covariance parameter must be supplied.

 None
covariance_matrix ArrayLike | None

Covariance matrix for the multivariate normal on the logit scale. Exactly one covariance parameter must be supplied.

 None
precision_matrix ArrayLike | None

Precision matrix for the multivariate normal on the logit scale. Exactly one covariance parameter must be supplied.

 None
Source code in pyrenew/ascertainment/joint.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    name: str,
    signals: tuple[str, ...],
    baseline_rates: ArrayLike,
    scale_tril: ArrayLike | None = None,
    covariance_matrix: ArrayLike | None = None,
    precision_matrix: ArrayLike | None = None,
) -> None:
    """
    Initialize a joint scalar ascertainment model.

    Parameters
    ----------
    name
        Name of the ascertainment model.
    signals
        Unique signal names, such as ``("hospital", "ed_visits")``. The
        order corresponds to entries in ``baseline_rates`` and the covariance
        parameter.
    baseline_rates
        Natural-scale baseline ascertainment rates. Shape ``(n_signals,)``.
        Values must be probabilities in ``(0, 1)``. A value of ``0.01``
        centers the corresponding ascertainment rate near 1 percent before
        accounting for covariance.
    scale_tril
        Lower-triangular scale matrix for the multivariate normal on the
        logit scale. Exactly one covariance parameter must be supplied.
    covariance_matrix
        Covariance matrix for the multivariate normal on the logit scale.
        Exactly one covariance parameter must be supplied.
    precision_matrix
        Precision matrix for the multivariate normal on the logit scale.
        Exactly one covariance parameter must be supplied.
    """
    super().__init__(name=name, signals=signals)
    baseline_rates_array = jnp.asarray(baseline_rates)
    scale_tril_array = self._optional_array(scale_tril)
    covariance_matrix_array = self._optional_array(covariance_matrix)
    precision_matrix_array = self._optional_array(precision_matrix)
    self._validate_parameters(baseline_rates_array)
    self.distribution: dist.MultivariateNormal = dist.MultivariateNormal(
        loc=logit(baseline_rates_array),
        scale_tril=scale_tril_array,
        covariance_matrix=covariance_matrix_array,
        precision_matrix=precision_matrix_array,
    )

baseline_rates property 

baseline_rates: Array

Natural-scale baseline ascertainment rates.

Returns:

Type Description
Array

Distribution location transformed from logit scale to probability scale.

sample 

sample(**kwargs: object) -> Mapping[str, ArrayLike]

Sample jointly distributed scalar ascertainment rates.

Parameters:

Name Type Description Default
**kwargs object

Additional model-context arguments, ignored.

 {}

Returns:

Type Description
Mapping[str, ArrayLike]

Mapping from signal name to sampled scalar ascertainment rate.
Source code in pyrenew/ascertainment/joint.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def sample(self, **kwargs: object) -> Mapping[str, ArrayLike]:
    """
    Sample jointly distributed scalar ascertainment rates.

    Parameters
    ----------
    **kwargs
        Additional model-context arguments, ignored.

    Returns
    -------
    Mapping[str, ArrayLike]
        Mapping from signal name to sampled scalar ascertainment rate.
    """
    eta = numpyro.sample(
        f"{self.name}_eta",
        self.distribution,
    )
    rates = expit(eta)

    result = {}
    for signal, rate in zip(self.signals, rates):
        numpyro.deterministic(f"{self.name}_{signal}", rate)
        result[signal] = rate

    return result

RatioLinkedAscertainment 

RatioLinkedAscertainment(
    name: str,
    base_signal: str,
    linked_signal: str,
    base_rate_rv: RandomVariable,
    ratio_rv: RandomVariable,
)

Bases: AscertainmentModel

Two ascertainment rates expressed as a base rate and a ratio.

The linked ascertainment rate is the sampled base rate multiplied by the sampled ratio.

Initialize a ratio-linked ascertainment model.

Parameters:

Name Type Description Default
name str

Name of the ascertainment model.

 required
base_signal str

Name of the signal whose ascertainment rate is sampled directly.

 required
linked_signal str

Name of the signal whose ascertainment rate is the product of the base rate and ratio.

 required
base_rate_rv RandomVariable

Random variable for the base signal's ascertainment rate.

 required
ratio_rv RandomVariable

Random variable for the ratio of the linked signal's ascertainment rate to the base signal's ascertainment rate.

 required
Source code in pyrenew/ascertainment/linked.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    name: str,
    base_signal: str,
    linked_signal: str,
    base_rate_rv: RandomVariable,
    ratio_rv: RandomVariable,
) -> None:
    """
    Initialize a ratio-linked ascertainment model.

    Parameters
    ----------
    name
        Name of the ascertainment model.
    base_signal
        Name of the signal whose ascertainment rate is sampled directly.
    linked_signal
        Name of the signal whose ascertainment rate is the product of the
        base rate and ratio.
    base_rate_rv
        Random variable for the base signal's ascertainment rate.
    ratio_rv
        Random variable for the ratio of the linked signal's ascertainment
        rate to the base signal's ascertainment rate.
    """
    super().__init__(name=name, signals=(base_signal, linked_signal))
    self.base_signal = base_signal
    self.linked_signal = linked_signal
    self.base_rate_rv = base_rate_rv
    self.ratio_rv = ratio_rv

sample 

sample(**kwargs: object) -> Mapping[str, ArrayLike]

Sample the base rate and ratio and calculate the linked rate.

Parameters:

Name Type Description Default
**kwargs object

Additional model-context arguments, ignored.

 {}

Returns:

Type Description
Mapping[str, ArrayLike]

Mapping from the base and linked signal names to their sampled ascertainment rates.
Source code in pyrenew/ascertainment/linked.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def sample(self, **kwargs: object) -> Mapping[str, ArrayLike]:
    """
    Sample the base rate and ratio and calculate the linked rate.

    Parameters
    ----------
    **kwargs
        Additional model-context arguments, ignored.

    Returns
    -------
    Mapping[str, ArrayLike]
        Mapping from the base and linked signal names to their sampled
        ascertainment rates.
    """
    base_rate = self.base_rate_rv()
    ratio = self.ratio_rv()
    linked_rate = base_rate * ratio

    numpyro.deterministic(f"{self.name}_{self.base_signal}", base_rate)
    numpyro.deterministic(f"{self.name}_{self.linked_signal}", linked_rate)

    return {
        self.base_signal: base_rate,
        self.linked_signal: linked_rate,
    }