Skip to content

Ascertainment

Ascertainment models for shared observation-rate structure.

AscertainmentModel 

AscertainmentModel(
    name: str,
    signals: tuple[str, ...],
    temporal_processes: Mapping[str, TemporalProcess] | None = None,
)

An AscertainmentModel is a component of a PyRenew renewal process model. It provides an ascertainment rate for one or more observation signals.

Each signal has a scalar baseline rate. If a temporal process is specified for that signal, the component combines the baseline rate with the temporal process to produce a rate trajectory over the model period. Otherwise, the baseline rate applies throughout the model period.

A baseline rate may lie in [0, 1] when no temporal process is configured. A signal with a temporal process requires a baseline rate strictly inside (0, 1), because the temporal deviation is added on the logit scale. At either endpoint, the logit is infinite and finite deviations cannot change the rate.

The model samples one scalar baseline rate per signal and can add an optional signal-specific temporal deviation on the logit scale. Subclasses implement _sample_baseline_rates() to define relationships among the scalar baselines. The base class owns validation, temporal sampling, and the standard deterministic sites.

Concrete subclasses determine how the baseline rates are specified. IndependentAscertainment specifies each baseline separately, JointAscertainment assigns the baselines a joint distribution, and RatioLinkedAscertainment defines one baseline relative to another.

Register the component with PyrenewBuilder.add_ascertainment(...). Use for_signal(...) to obtain the signal-specific accessor passed to an observation process as ascertainment_rate_rv. Accessors returned by for_signal() read the final sampled values from the active model context and do not sample independently.

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

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

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
temporal_processes Mapping[str, TemporalProcess] | None

Optional temporal processes keyed by signal name. Signals without a process retain their scalar baseline rate.

 None

Raises:

Type Description
TypeError

If temporal_processes is not a mapping or a configured process does not satisfy the TemporalProcess protocol.
ValueError

If a temporal process is configured for an unknown signal.
Source code in pyrenew/ascertainment/base.py 
129
130
131
132
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def __init__(
    self,
    name: str,
    signals: tuple[str, ...],
    temporal_processes: Mapping[str, TemporalProcess] | None = None,
) -> None:
    """
    Initialize an ascertainment model.

    Parameters
    ----------
    name
        A non-empty string identifying the ascertainment model.
    signals
        Unique signal names produced by this model.
    temporal_processes
        Optional temporal processes keyed by signal name. Signals without
        a process retain their scalar baseline rate.

    Raises
    ------
    TypeError
        If ``temporal_processes`` is not a mapping or a configured process
        does not satisfy the ``TemporalProcess`` protocol.
    ValueError
        If a temporal process is configured for an unknown signal.
    """
    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

    if temporal_processes is None:
        temporal_processes = {}
    if not isinstance(temporal_processes, Mapping):
        raise TypeError(
            "temporal_processes must be a mapping, "
            f"got {type(temporal_processes).__name__}."
        )

    expected_signals = set(signals)
    unknown_signals = tuple(
        signal for signal in temporal_processes if signal not in expected_signals
    )
    if unknown_signals:
        raise ValueError(
            f"temporal_processes contains unknown signals {unknown_signals} for "
            f"ascertainment model {name!r}. Available signals: {signals}."
        )

    ordered_processes: dict[str, TemporalProcess] = {}
    for signal in signals:
        if signal not in temporal_processes:
            continue
        process = temporal_processes[signal]
        if not isinstance(process, TemporalProcess):
            raise TypeError(
                f"temporal process for signal {signal!r} must satisfy the "
                f"TemporalProcess protocol, got {type(process).__name__}."
            )
        ordered_processes[signal] = process
    self.temporal_processes = ordered_processes

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 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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,
    )

requires_calendar_anchor 

requires_calendar_anchor() -> bool

Return whether any temporal process requires a calendar anchor.

Returns:

Type Description
bool

True when the model needs a calendar anchor; otherwise False.
Source code in pyrenew/ascertainment/base.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
def requires_calendar_anchor(self) -> bool:
    """Return whether any temporal process requires a calendar anchor.

    Returns
    -------
    bool
        ``True`` when the model needs a calendar anchor; otherwise
        ``False``.
    """
    return any(
        getattr(process, "requires_calendar_anchor", False)
        for process in self.temporal_processes.values()
    )

sample 

sample(
    n_timepoints: int, first_day_dow: int | None = None
) -> Mapping[str, ArrayLike]

Sample baseline rates and optional temporal rate trajectories.

Parameters:

Name Type Description Default
n_timepoints int

Positive number of timepoints on the shared model axis.

 required
first_day_dow int | None

Day of week for the first model-axis timepoint. Calendar-aligned temporal processes use this value.

 None

Returns:

Type Description
Mapping[str, ArrayLike]

Final scalar rates or full-axis trajectories in signal order.

Raises:

Type Description
TypeError

If n_timepoints is not a built-in integer.
ValueError

If n_timepoints is less than one, baseline rates are invalid, or a temporal process returns an unexpected shape.
Source code in pyrenew/ascertainment/base.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def sample(
    self,
    n_timepoints: int,
    first_day_dow: int | None = None,
) -> Mapping[str, ArrayLike]:
    """Sample baseline rates and optional temporal rate trajectories.

    Parameters
    ----------
    n_timepoints
        Positive number of timepoints on the shared model axis.
    first_day_dow
        Day of week for the first model-axis timepoint. Calendar-aligned
        temporal processes use this value.

    Returns
    -------
    Mapping[str, ArrayLike]
        Final scalar rates or full-axis trajectories in signal order.

    Raises
    ------
    TypeError
        If ``n_timepoints`` is not a built-in integer.
    ValueError
        If ``n_timepoints`` is less than one, baseline rates are invalid,
        or a temporal process returns an unexpected shape.
    """
    if type(n_timepoints) is not int:
        raise TypeError(
            "n_timepoints must be a positive integer, "
            f"got {type(n_timepoints).__name__}."
        )
    if n_timepoints < 1:
        raise ValueError(
            f"n_timepoints must be a positive integer, got {n_timepoints}."
        )

    baseline_rates = self._validate_baseline_rates(self._sample_baseline_rates())
    logit_transform = SigmoidTransform().inv

    result: dict[str, ArrayLike] = {}
    for signal in self.signals:
        baseline_rate = baseline_rates[signal]
        numpyro.deterministic(
            f"{self.name}_baseline_{signal}",
            baseline_rate,
        )

        process = self.temporal_processes.get(signal)
        if process is None:
            rate = baseline_rate
        else:
            deviation = jnp.asarray(
                process.sample(
                    n_timepoints=n_timepoints,
                    initial_value=0.0,
                    n_processes=1,
                    name_prefix=f"{self.name}_{signal}",
                    first_day_dow=first_day_dow,
                )
            )
            required_shape = (n_timepoints, 1)
            if deviation.shape != required_shape:
                raise ValueError(
                    f"Ascertainment model {self.name!r}, signal {signal!r}, "
                    f"received temporal-process shape {deviation.shape}; "
                    f"required shape is {required_shape}."
                )
            deviation = jnp.squeeze(deviation, axis=-1)
            rate = logit_transform.inv(logit_transform(baseline_rate) + deviation)

        numpyro.deterministic(f"{self.name}_{signal}", rate)
        result[signal] = rate

    return result

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 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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,
    )

IndependentAscertainment 

IndependentAscertainment(
    name: str,
    rate_rvs: Mapping[str, RandomVariable],
    temporal_processes: Mapping[str, TemporalProcess] | None = None,
)

Bases: AscertainmentModel

Ascertainment rates with independently sampled scalar baselines.

Use this model for one signal or for multiple signals whose baseline rates do not share a joint prior. Add temporal processes only for signals whose rates should vary over time.

Initialize independent ascertainment rates.

Parameters:

Name Type Description Default
name str

Name of the ascertainment model.

 required
rate_rvs Mapping[str, RandomVariable]

Non-empty mapping from signal names to random variables for their scalar baseline rates. Mapping order determines signal order.

 required
temporal_processes Mapping[str, TemporalProcess] | None

Optional temporal processes keyed by signal name. Signals without a process retain their scalar baseline rate.

 None

Raises:

Type Description
TypeError

If rate_rvs is not a mapping or a value is not a RandomVariable.
ValueError

If rate_rvs is empty or contains an invalid signal name.
Source code in pyrenew/ascertainment/independent.py 
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    name: str,
    rate_rvs: Mapping[str, RandomVariable],
    temporal_processes: Mapping[str, TemporalProcess] | None = None,
) -> None:
    """Initialize independent ascertainment rates.

    Parameters
    ----------
    name
        Name of the ascertainment model.
    rate_rvs
        Non-empty mapping from signal names to random variables for their
        scalar baseline rates. Mapping order determines signal order.
    temporal_processes
        Optional temporal processes keyed by signal name. Signals without
        a process retain their scalar baseline rate.

    Raises
    ------
    TypeError
        If ``rate_rvs`` is not a mapping or a value is not a
        ``RandomVariable``.
    ValueError
        If ``rate_rvs`` is empty or contains an invalid signal name.
    """
    if not isinstance(rate_rvs, Mapping):
        raise TypeError(
            f"rate_rvs must be a mapping, got {type(rate_rvs).__name__}."
        )
    if len(rate_rvs) == 0:
        raise ValueError("rate_rvs must be a non-empty mapping.")

    for signal, rate_rv in rate_rvs.items():
        if not isinstance(signal, str) or len(signal) == 0:
            raise ValueError("rate_rvs keys must be non-empty strings.")
        if not isinstance(rate_rv, RandomVariable):
            raise TypeError(
                f"rate_rv for signal {signal!r} must be a RandomVariable, "
                f"got {type(rate_rv).__name__}."
            )

    self.rate_rvs = dict(rate_rvs)
    super().__init__(
        name=name,
        signals=tuple(self.rate_rvs),
        temporal_processes=temporal_processes,
    )

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,
    temporal_processes: Mapping[str, TemporalProcess] | None = None,
)

Bases: AscertainmentModel

Joint prior for baseline 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 sampled baseline is scalar. Optional temporal processes can vary any subset of the final rates 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
temporal_processes Mapping[str, TemporalProcess] | None

Optional temporal processes keyed by signal name. Signals without a process retain their scalar baseline rate.

 None
Source code in pyrenew/ascertainment/joint.py 
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
88
89
90
91
92
93
94
95
96
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,
    temporal_processes: Mapping[str, TemporalProcess] | 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.
    temporal_processes
        Optional temporal processes keyed by signal name. Signals without
        a process retain their scalar baseline rate.
    """
    super().__init__(
        name=name,
        signals=signals,
        temporal_processes=temporal_processes,
    )
    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.

RatioLinkedAscertainment 

RatioLinkedAscertainment(
    name: str,
    base_signal: str,
    linked_signal: str,
    base_rate_rv: RandomVariable,
    ratio_rv: RandomVariable,
    temporal_processes: Mapping[str, TemporalProcess] | None = None,
)

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. Optional temporal processes act on the resulting scalar baselines and do not preserve the ratio pointwise over time.

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
temporal_processes Mapping[str, TemporalProcess] | None

Optional temporal processes keyed by signal name. Signals without a process retain their scalar baseline rate.

 None
Source code in pyrenew/ascertainment/linked.py 
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
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    name: str,
    base_signal: str,
    linked_signal: str,
    base_rate_rv: RandomVariable,
    ratio_rv: RandomVariable,
    temporal_processes: Mapping[str, TemporalProcess] | None = None,
) -> 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.
    temporal_processes
        Optional temporal processes keyed by signal name. Signals without
        a process retain their scalar baseline rate.
    """
    super().__init__(
        name=name,
        signals=(base_signal, linked_signal),
        temporal_processes=temporal_processes,
    )
    self.base_signal = base_signal
    self.linked_signal = linked_signal
    self.base_rate_rv = base_rate_rv
    self.ratio_rv = ratio_rv