"""Capability-aware post-fit diagnostics, validation, and replay metadata.
Assessment lives on a fitted result because its questions depend on the artifacts the
method actually produced. The facades in this module do not infer support from a result
class and hope for the best: every public operation has a declaration for every public
scalar result family, including deliberate refusals.
"""
from __future__ import annotations
import importlib
import inspect
from collections.abc import Callable, Container, Iterator, Mapping, Sequence
from dataclasses import dataclass, field, replace
from enum import Enum, StrEnum
from functools import cached_property
from types import MappingProxyType
from typing import Any, ClassVar, Literal
import numpy as np
from ._assessment_cache import (
_RETAIN_PACKED,
_cache_key,
_cached,
_frame_payload,
_normalize,
_pack_cached,
_unpack_cached,
)
from ._declarations import declarations_pass
from ._inference_status import is_labelled, precedent_status, status_record, supplies_inference
from ._typing import CumulativeGBounds
from .data.weighting import REPORTED_DRAW, format_score_load
from .exceptions import (
NO_ALTERNATIVE,
CapabilityError,
DataError,
Refusal,
RefusalReason,
inference_refusal,
)
from .inference.influence import spread_name
from .targets.population_intervention import (
NATURAL_COURSE_SUPPORT_REFUSAL,
is_natural_course_fit,
natural_course_names,
reads_natural_course_mean,
)
from .utils.frames import emit_frame
from .utils.memos import without_memos
from .utils.text import format_draw, format_table
from .validation.drtmle import IDENTITY_TOLERANCE
from .validation.longitudinal import (
LONGITUDINAL_CENSORING_NOT_FITTED,
LONGITUDINAL_CONSTANT_TARGET,
LONGITUDINAL_CONSTANT_TARGET_IN_FOLD,
LONGITUDINAL_HELD_DECISION,
LONGITUDINAL_KNOWN_MECHANISM,
LONGITUDINAL_NO_CENSORING,
LONGITUDINAL_NO_CENSORING_IN_FOLD,
LONGITUDINAL_NO_OUTCOME_REGRESSION,
LONGITUDINAL_SIMULATED_MODELS,
LongitudinalDiagnostics,
LongitudinalNuisanceDiagnostics,
LongitudinalNuisanceOmission,
LongitudinalNuisanceRow,
LongitudinalScoreDiagnostics,
LongitudinalScoreRow,
LongitudinalStageRow,
_longitudinal_nuisances,
_longitudinal_scores,
_longitudinal_stagewise,
)
from .validation.score import DEFAULT_TOLERANCE
__all__ = [
"ASSESSMENT_CAPABILITIES",
"LONGITUDINAL_CENSORING_NOT_FITTED",
"LONGITUDINAL_CONSTANT_TARGET",
"LONGITUDINAL_CONSTANT_TARGET_IN_FOLD",
"LONGITUDINAL_HELD_DECISION",
"LONGITUDINAL_KNOWN_MECHANISM",
"LONGITUDINAL_NO_CENSORING",
"LONGITUDINAL_NO_CENSORING_IN_FOLD",
"LONGITUDINAL_NO_OUTCOME_REGRESSION",
"LONGITUDINAL_SIMULATED_MODELS",
"SENSITIVITY_ROUTES",
"VALIDATION_OPERATIONS",
"AssessmentCapability",
"AssessmentItem",
"AssessmentReport",
"AssessmentStatus",
"DiagnosticReport",
"DiagnosticsFacade",
"LongitudinalDiagnostics",
"LongitudinalNuisanceDiagnostics",
"LongitudinalNuisanceOmission",
"LongitudinalNuisanceRow",
"LongitudinalScoreDiagnostics",
"LongitudinalScoreRow",
"LongitudinalStageRow",
"Replayability",
"SensitivityFacade",
"SensitivityRoute",
"ValidationReport",
"assessment_capabilities",
"replayability",
"validate_result",
]
#: The operations the validation battery owns, in the order it runs them.
#:
#: :func:`validate_result` runs each one argument-free, and
#: :meth:`AssessmentReport._presented` shows the validation row rather than the
#: diagnostics row of the same name. A caller argument for one of these names would
#: therefore be answered on the diagnostics surface and then discarded, so
#: :func:`assess_result` refuses it instead. That refusal needs a name here *and* a
#: parameter the caller could fill: ``support`` and ``nuisance_models`` take none, so an
#: argument for either is a ``TypeError`` from the signature rather than a composition
#: this module has to rule out.
VALIDATION_OPERATIONS: tuple[str, ...] = ("score_equations", "support", "nuisance_models")
[docs]
class AssessmentStatus(StrEnum): # numpydoc ignore=PR01,PR02
"""Status returned by a diagnostic or validation operation.
``DEFERRED`` means that the operation can run after the caller supplies a
required choice or cost opt-in. ``NOT_APPLICABLE`` means that the operation
does not apply to the fitted estimand. ``UNAVAILABLE`` means that the operation
applies, but this fit cannot run it. Four causes reach that status: a missing
method, a missing derivation, a missing replay artifact, and a requested variant
that has no support. An operation that raises a capability refusal after the
combined report invoked it is unavailable for the same reason.
Reach a status by name, as ``AssessmentStatus.PASSED``.
The synthetic constructor that :class:`enum.StrEnum` gives every member is not a
caller argument, and its parameters *differ by interpreter*: 3.11 exposes ``value``,
``names``, ``module``, ``qualname``, ``type``, ``start`` and ``boundary``, while 3.12
collapses several into ``*values``. Documenting either one satisfies numpydoc on that
version and fails it on the other, which is how a ``Parameters`` block written for 3.12
left `PR01` and `PR02` firing on the 3.11 build. Ignoring both is the version-independent
answer, and ``docs/development/contributing.md`` tells a contributor to build the
documentation on whichever interpreter they have.
"""
PASSED = "passed"
FAILED = "failed"
WARNING = "warning"
COMPLETED = "completed"
DEFERRED = "deferred"
NOT_APPLICABLE = "not_applicable"
UNAVAILABLE = "unavailable"
# How a combined report presents one status. The three sets below partition
# ``AssessmentStatus``, and ``tests/unit/test_assessment_contract.py`` asserts that
# partition, so a new member lands in exactly one of them rather than in none of them.
# The three groupings were spelled inline at three call sites with nothing tying them
# together, and adding ``DEFERRED`` reached two of the three by hand.
#
# Presentation policy belongs beside the enum rather than on it: which bucket a status
# falls into is a decision ``AssessmentReport`` makes, not a property of the status. A
# member on a ``StrEnum``, beside a class attribute of the same shape, also reads as one
# more status at every call site.
#: A row the reader has to act on. Both statuses come from a check that ran and reached a
#: verdict, which is why ``WARNING`` is here and no omission is.
_ATTENTION: frozenset[AssessmentStatus] = frozenset(
{AssessmentStatus.FAILED, AssessmentStatus.WARNING}
)
#: A row that produced no verdict, whether the caller, the fitted estimand, or the fit's
#: artifacts are the reason. These are omissions the report keeps visible.
_OMISSIONS: frozenset[AssessmentStatus] = frozenset(
{
AssessmentStatus.DEFERRED,
AssessmentStatus.NOT_APPLICABLE,
AssessmentStatus.UNAVAILABLE,
}
)
#: A row that ran and asks nothing of the reader. ``COMPLETED`` is a descriptive analysis
#: with no pass criterion, and ``PASSED`` is a check whose condition holds.
_SETTLED: frozenset[AssessmentStatus] = frozenset(
{AssessmentStatus.PASSED, AssessmentStatus.COMPLETED}
)
#: Compact check rows follow returned descriptive results. A failed check comes first,
#: then a warning and a passed check. ``test_every_status_is_presented_by_exactly_one_grouping``
#: in ``tests/unit/test_assessment_contract.py`` pins two separate claims about this tuple
#: and ``_SUMMARY_OMISSION_ORDER``: every status other than ``COMPLETED`` belongs to exactly
#: one of them, and each tuple equals the literal sequence written here. Membership alone
#: would leave the reading order free, and the order is what a reader sees first.
_SUMMARY_CHECK_ORDER: tuple[AssessmentStatus, ...] = (
AssessmentStatus.FAILED,
AssessmentStatus.WARNING,
AssessmentStatus.PASSED,
)
#: Compact omission rows follow checks. Caller-deferred work comes first because the
#: caller can resolve it. A fit limitation follows, then a question that does not apply.
#: The test named above pins this membership and this literal order too.
_SUMMARY_OMISSION_ORDER: tuple[AssessmentStatus, ...] = (
AssessmentStatus.DEFERRED,
AssessmentStatus.UNAVAILABLE,
AssessmentStatus.NOT_APPLICABLE,
)
#: The columns both compact inventories carry. A check row and an omission row differ in
#: which statuses they can hold, not in shape.
_INVENTORY_HEADERS: tuple[str, ...] = ("status", "count", "operations")
#: How the compact summary closes. Both sentences name the surface that carries the detail
#: the summary drops, so the wording is a contract rather than decoration:
#: ``docs/technical-reference/validation-methods.md`` describes these lines, and
#: ``tests/unit/test_post_fit_assessment_battery.py`` re-types them to assert the summary
#: ends in exactly this text. That test spells the literal out on purpose, so importing
#: this name there would make the assertion agree with itself.
_SUMMARY_FOOTER: str = (
"Full ledger: call to_frame(). Next steps: call next_steps().\n"
"Retained payloads: call report(...)."
)
#: What stops :attr:`ValidationReport.passed`. This one cuts across the three above rather
#: than refining one of them: a ``WARNING`` is actionable and still passes, a
#: ``NOT_APPLICABLE`` is an omission and still passes, and a ``DEFERRED`` required check is
#: an unanswered question rather than evidence that validation succeeded.
_BLOCKING: frozenset[AssessmentStatus] = frozenset(
{
AssessmentStatus.FAILED,
AssessmentStatus.DEFERRED,
AssessmentStatus.UNAVAILABLE,
}
)
[docs]
@dataclass(frozen=True)
class AssessmentCapability:
"""Describe whether and how a result supports one assessment operation.
Parameters
----------
operation : str
Public operation name on a diagnostics or sensitivity facade.
result_family : str
Result family for which the declaration applies.
methods : tuple of str
Estimation methods covered by the declaration.
available : bool
Whether the operation can run on the result family.
status : AssessmentStatus
Status to report when the operation cannot run. ``DEFERRED`` means that the caller
lifts the refusal by passing every name in ``requires_arguments``.
required_artifacts : tuple of str
Fitted artifacts that the operation reads.
execution : {"summarize", "retarget", "refit"}
Most expensive work the operation performs.
deterministic_from_saved : bool
Whether saved artifacts determine the result without a refit.
interpretation : str
Statistical question that the operation answers.
cost : {"cheap", "moderate", "expensive"}
Relative cost category used by combined reports.
refusal : Refusal or None
The record of why the operation is not available: its reason code, the failed
condition and the alternative. :attr:`reason` is its text, and a raise of the
operation carries the same record.
requires_arguments : tuple of str
Arguments the caller must pass. An entry either has no default, or has a
default that the operation refuses on this result family.
accepts_random_state : bool
Whether a combined run can forward its top-level seed.
requires_replay : str or None
Name of the :class:`Replayability` field the operation needs. ``None`` means the
operation reads stored artifacts only.
"""
operation: str
result_family: str
methods: tuple[str, ...]
available: bool
status: AssessmentStatus
required_artifacts: tuple[str, ...]
execution: Literal["summarize", "retarget", "refit"]
deterministic_from_saved: bool
interpretation: str
cost: Literal["cheap", "moderate", "expensive"]
refusal: Refusal | None = None
#: Arguments the caller must supply, so a combined report cannot run this operation.
#: An entry covers two cases: the operation has no default for it, or the operation
#: refuses its own default on this fit. ``simulated_confounding`` is the second case,
#: because a continuous fit refuses the bare ``estimand="ate"`` default. Declared
#: here rather than special-cased by name in ``run_all``, which knows nothing about
#: any particular operation.
requires_arguments: tuple[str, ...] = ()
accepts_random_state: bool = False
#: The :class:`Replayability` field this operation needs from the stored result.
#: One declaration per row, applied once in :meth:`_CapabilityFacade._capability_map`,
#: because two facades each patching their own rows left ``refute`` claiming
#: ``available=True`` on a result with no estimator while ``truncation_curve``,
#: ``benchmark`` and ``simulated_confounding`` beside it reported the truth.
requires_replay: str | None = None
@property
def reason(self) -> str | None:
"""The text of :attr:`refusal`, or ``None`` when the row declares none."""
return None if self.refusal is None else self.refusal.message()
def _capability(
operation: str,
family: str,
*,
artifacts: Sequence[str],
interpretation: str,
execution: Literal["summarize", "retarget", "refit"] = "summarize",
deterministic: bool = True,
cost: Literal["cheap", "moderate", "expensive"] = "cheap",
available: bool = True,
status: AssessmentStatus = AssessmentStatus.PASSED,
refusal: Refusal | None = None,
requires_arguments: Sequence[str] = (),
methods: Sequence[str] | None = None,
accepts_random_state: bool = False,
requires_replay: str | None = None,
) -> AssessmentCapability:
return AssessmentCapability(
operation=operation,
result_family=family,
methods=tuple(
methods
if methods is not None
else (
("tmle", "collaborative_tmle", "drtmle")
if family == "point"
else ("tmle", "sdr", "ipsi", "gcomp", "ipw", "parametric_gformula")
)
),
available=available,
status=status,
required_artifacts=tuple(artifacts),
execution=execution,
deterministic_from_saved=deterministic,
interpretation=interpretation,
cost=cost,
refusal=refusal,
requires_arguments=tuple(requires_arguments),
accepts_random_state=accepts_random_state,
requires_replay=requires_replay,
)
# This is intentionally data rather than branches hidden inside the facade. Contract tests
# check both directions: every operation covers every family, and no declaration names a
# family or operation that the public result surface no longer exposes.
ASSESSMENT_CAPABILITIES: tuple[AssessmentCapability, ...] = (
_capability(
"support",
"point",
artifacts=("fitted analytic mechanism", "targeting weights"),
interpretation="arm, regimen, shift, or incremental support for the identified functional",
),
_capability(
"nuisance_models",
"point",
artifacts=(
"out-of-fold nuisance predictions",
"method selection state",
"repeat-specific point estimates",
"reported standard errors",
),
interpretation=(
"held-out nuisance fit, collaborative selection, and descriptive split spread"
),
),
_capability(
"score_equations",
"point",
artifacts=("targeting state", "influence curves"),
interpretation="the score equations the selected method actually solved",
),
_capability(
"corrections",
"point",
artifacts=("doubly-robust correction state",),
interpretation="the correction identities solved by guarded doubly-robust targeting",
methods=("drtmle",),
),
_capability(
"truncation_curve",
"point",
artifacts=("fitted nuisance predictions", "targeting state"),
execution="retarget",
cost="moderate",
interpretation="estimate stability across declared mechanism bounds",
requires_replay="retarget_cached_nuisances",
),
_capability(
"refute",
"point",
artifacts=("fitted estimator configuration", "analysis data"),
execution="refit",
deterministic=False,
cost="expensive",
accepts_random_state=True,
interpretation="behavior under placebo, noise, and subsampling perturbations",
requires_replay="refit_nuisances",
),
_capability(
"support",
"longitudinal",
artifacts=("cumulative mechanism products", "regimen histories"),
interpretation="history-specific support and cumulative leverage at every node",
),
_capability(
"nuisance_models",
"longitudinal",
artifacts=(
"observed-law treatment predictions",
"observed-law censoring predictions",
"node pseudo-outcomes",
"initial node predictions",
"nuisance learner diagnostics",
),
interpretation=(
"weighted treatment, censoring, outcome, and pseudo-outcome fit by node and "
"fitted recursion"
),
),
_capability(
"score_equations",
"longitudinal",
artifacts=("stagewise targeting state",),
interpretation="one targeting score and convergence record per regimen and node",
),
_capability(
"corrections",
"longitudinal",
artifacts=(),
available=False,
status=AssessmentStatus.NOT_APPLICABLE,
refusal=Refusal(
RefusalReason.INVALID_DECLARATION,
"longitudinal targeting does not use the point-treatment correction system",
),
interpretation="the correction identities solved by guarded doubly-robust targeting",
),
_capability(
"truncation_curve",
"longitudinal",
artifacts=(
"raw sequential mechanism predictions, out of fold when cross-fitted",
"unfitted recursive learner templates",
"resolved regimen plans",
),
execution="refit",
cost="expensive",
requires_arguments=("bounds",),
requires_replay="refit_nuisances",
interpretation="estimate stability across declared mechanism bounds",
),
_capability(
"refute",
"longitudinal",
artifacts=("fitted sequential estimator configuration",),
execution="refit",
deterministic=False,
cost="expensive",
available=False,
status=AssessmentStatus.UNAVAILABLE,
refusal=Refusal(
RefusalReason.IMPLEMENTATION_GAP,
"no evidence-backed longitudinal perturbation/refit adapter is implemented.",
alternative=NO_ALTERNATIVE,
roadmap="X67",
),
interpretation="behavior under longitudinal data perturbations",
),
)
@dataclass(frozen=True)
class SensitivityRoute:
"""Where one sensitivity operation is implemented, and how its estimand is supplied.
``needs_estimand`` is a property of the target's *signature*, not of the operation's
name: four omitted-variable analyses and :func:`tipping_gamma` all take ``estimand``
as their second positional argument, so the facade can fill that position in for a fit
that reports no bare ``"ate"``. ``benchmark`` and ``missingness`` take ``covariates``
and ``gamma`` there, and ``evalue`` selects for itself from a ``None`` sentinel --
injecting a name into any of those positions would silently pass it as something else.
It does not say whether the operation takes an estimand at all. ``benchmark`` takes
one as a keyword-only argument with the same ambiguous ``"ate"`` default, and the
facade fills that in by keyword. :func:`_defaults_to_ambiguous_estimand` answers that
second question from the signature, and it is the one the deferral reads.
"""
module: str
function: str
needs_estimand: bool = False
#: Data for the same reason ``ASSESSMENT_CAPABILITIES`` is, and paired with it by a contract
#: test in both directions: every declared sensitivity capability has a route, and every
#: route is declared. The alternative -- a ``module``/``function`` pair passed by each
#: method plus an ``operation in {...}`` set inside the dispatcher -- is a second registry
#: that no test can see, and it is how ``tipping_gamma`` came to be the one operation of its
#: signature shape that did not get a default estimand.
SENSITIVITY_ROUTES: dict[str, SensitivityRoute] = {
"omitted_confounding": SensitivityRoute(
"omitted_variable", "omitted_variable_bounds", needs_estimand=True
),
"robustness_value": SensitivityRoute(
"omitted_variable", "robustness_value", needs_estimand=True
),
"elements": SensitivityRoute("omitted_variable", "sensitivity_elements", needs_estimand=True),
"benchmark": SensitivityRoute("omitted_variable", "benchmark"),
"contour": SensitivityRoute("omitted_variable", "contour_data", needs_estimand=True),
"evalue": SensitivityRoute("evalue", "evalue"),
"missingness": SensitivityRoute("missingness", "missingness_tilt"),
"tipping_gamma": SensitivityRoute("missingness", "tipping_gamma", needs_estimand=True),
"simulated_confounding": SensitivityRoute(
"simulated_confounding", "simulated_confounding", needs_estimand=True
),
}
#: Why an SDR fit has no score-equation report.
SDR_SCORE_EQUATIONS_REFUSAL = Refusal(
RefusalReason.API_REDIRECTION,
"the SDR estimator fits no fluctuation, so no node solves a score equation.",
alternative="diagnostics.support() reports each node's transform range and bound share.",
)
#: Why a sequential baseline fit has no score-equation report.
BASELINE_SCORE_EQUATIONS_REFUSAL = Refusal(
RefusalReason.API_REDIRECTION,
"the sequential g-computation, IPW and parametric g-formula estimators fit no fluctuation, "
"so no node solves a score equation.",
alternative="diagnostics.support() reports each node's weights and bound share.",
)
#: Why an IPSI fit has no score-equation report.
IPSI_SCORE_EQUATIONS_REFUSAL = Refusal(
RefusalReason.API_REDIRECTION,
"the IPSI estimator fits no fluctuation, so no node solves a score equation.",
alternative=(
"diagnostics.support() reports each node's weights, bound share and mechanism term."
),
)
def _family(result: Any) -> str:
family = getattr(result, "assessment_family", None)
if not isinstance(family, str) or not family:
raise TypeError(
"assessment requires the fitted artifact to declare a non-empty assessment_family"
)
return family
def _method(result: Any) -> str:
declared = getattr(result, "fitted_method", None)
if isinstance(declared, str) and declared:
return declared
raise TypeError("assessment requires the fitted artifact to declare its fitted method")
def assessment_capabilities(result: Any) -> tuple[AssessmentCapability, ...]:
"""All operation declarations for the result's family."""
family = _family(result)
rows = tuple(item for item in ASSESSMENT_CAPABILITIES if item.result_family == family)
if family == "point" and is_natural_course_fit(result):
rows = tuple(
replace(
row,
available=False,
status=AssessmentStatus.UNAVAILABLE,
refusal=NATURAL_COURSE_SUPPORT_REFUSAL,
)
if row.operation == "support"
else row
for row in rows
)
estimator = getattr(result.config, "estimator", "tmle") if family == "longitudinal" else None
if estimator in ("sdr", "ipsi", "gcomp", "ipw", "parametric_gformula"):
# An SDR, IPSI or baseline fit has no fluctuation, so no node solves a score
# equation. The support report carries its weights and bound shares instead.
refusal = {
"sdr": SDR_SCORE_EQUATIONS_REFUSAL,
"ipsi": IPSI_SCORE_EQUATIONS_REFUSAL,
"gcomp": BASELINE_SCORE_EQUATIONS_REFUSAL,
"ipw": BASELINE_SCORE_EQUATIONS_REFUSAL,
"parametric_gformula": BASELINE_SCORE_EQUATIONS_REFUSAL,
}[estimator]
rows = tuple(
replace(
row,
available=False,
status=AssessmentStatus.NOT_APPLICABLE,
refusal=refusal,
)
if row.operation == "score_equations"
else row
for row in rows
)
if estimator in ("gcomp", "parametric_gformula"):
# A g-computation or g-formula estimate reads no treatment mechanism, so no bound
# moves it.
from .longitudinal.estimator import GCOMP_TRUNCATION_REFUSAL, GFORMULA_TRUNCATION_REFUSAL
truncation = (
GCOMP_TRUNCATION_REFUSAL if estimator == "gcomp" else GFORMULA_TRUNCATION_REFUSAL
)
rows = tuple(
replace(
row,
available=False,
status=AssessmentStatus.UNAVAILABLE,
refusal=truncation,
)
if row.operation == "truncation_curve"
else row
for row in rows
)
if family == "point" and result.config.parameter_axis == "learned_rule":
# A refit relearns the rule inside each training fold, so no refutation compares
# like with like, whatever the estimand. Declared for the whole fit, so the row
# does not defer on an argument that no value lifts. ``refute`` raises the same
# sentence from its own rule table.
from .interventions.learned import LEARNED_RULE_REFIT_REFUSAL
rows = tuple(
replace(
row,
available=False,
status=AssessmentStatus.UNAVAILABLE,
refusal=LEARNED_RULE_REFIT_REFUSAL,
)
if row.operation == "refute"
else row
for row in rows
)
if (
family == "point"
and result.data.treatment_kind == "continuous"
and result.config.parameter_axis == "msm"
):
# The arm positivity report cannot interpret a continuous density. No
# dose-grid support diagnostic is implemented for this projection axis.
rows = tuple(
replace(
row,
available=False,
status=AssessmentStatus.UNAVAILABLE,
refusal=Refusal(
RefusalReason.IMPLEMENTATION_GAP,
"a continuous MSM has no implemented dose-grid support diagnostic.",
alternative=NO_ALTERNATIVE,
roadmap="X65",
),
)
if row.operation == "support"
else row
for row in rows
)
rows = tuple(
replace(
row,
available=False,
status=AssessmentStatus.NOT_APPLICABLE,
refusal=Refusal(
RefusalReason.INVALID_DECLARATION,
"the fitted DR-TMLE guard subtracts no correction term"
if _method(result) == "drtmle"
else "the fitted method does not use the correction system",
),
)
if row.operation == "corrections"
and row.available
and not getattr(result, "solved_corrections", False)
else row
for row in rows
)
from .sensitivity.positivity import refits_reduced_regressions
if refits_reduced_regressions(result):
# A guarded DR-TMLE truncation curve retargets cached primary nuisances and
# refits reduced regressions inside that step. It does not call the outer
# estimator.refit(), whose configuration preflight controls refit_nuisances.
# The ordinary reduction uses its fitted bounds; missing-outcome reductions
# receive the swept bounds because those define two regression targets.
# `truncation_refusal` asks the reduction settings that step reads.
#
# Declared here rather than as a second registry row because a family may declare
# each operation once, and the cost is a fact about the fitted result rather than
# about the family. `reason` stays unset: it is documented, and contract-tested,
# as the explanation an *unavailable* row owes its caller, and this row runs.
rows = tuple(
replace(
row,
execution="refit",
cost="expensive",
requires_replay="retarget_cached_nuisances",
)
if row.operation == "truncation_curve" and row.available
else row
for row in rows
)
return rows
class _AbsentReport(Enum):
TOKEN = "absent report"
#: What an interpreter reads when nobody supplied arguments for its operation.
_NO_ARGUMENTS: Mapping[str, Any] = MappingProxyType({})
@dataclass(frozen=True)
class AssessmentItem:
"""One immutable result in a combined diagnostic or validation report.
Parameters
----------
name : str
Operation name.
status : AssessmentStatus
Outcome or omission status.
detail : str
Interpreted findings or the omission reason.
next_steps : tuple of str
Suggested follow-up actions.
_report : Any
Retained payload, including a legitimate None, or the private absence sentinel.
Excluded from equality. Dataframes use immutable cached storage.
arguments : mapping of str to Any
The invocation this row describes, and empty when there is none. A row whose
operation ran, and a row whose operation refused after it was invoked, carry the
effective arguments, which include the signature defaults and the resolved seed. A
row that its request refuses before any invocation carries its request with the
signature defaults bound. A deferred row carries the request as supplied, plus the
combined run's seed when the operation accepts one. It binds no signature default,
because the default is what such a row refuses: an ambiguous ``estimand="ate"``
recorded as effective would name an argument the operation declines. A row this fit
refuses outright carries nothing, because no request was considered and none can
make it run. Excluded from equality because argument values can contain arrays.
refusal : Refusal or None
The record of the refusal behind an omission: its reason code, the failed condition
and the alternative. ``None`` for an operation that ran, and for a deferral that
no refusal causes, such as a cost the caller has not paid for.
"""
name: str
status: AssessmentStatus
detail: str
next_steps: tuple[str, ...] = ()
_report: Any = field(default=_AbsentReport.TOKEN, compare=False, repr=False)
arguments: Mapping[str, Any] = field(default_factory=dict, compare=False, repr=False)
refusal: Refusal | None = None
@property
def report(self) -> Any:
"""Return the retained payload, or None when this operation did not run."""
return None if self._report is _AbsentReport.TOKEN else _unpack_cached(self._report)
def _item_columns(items: Sequence[AssessmentItem], **extra: Any) -> dict[str, Any]:
"""The frame payload every assessment surface emits, after any leading columns."""
return {
**extra,
"check": [item.name for item in items],
"status": [item.status.value for item in items],
"detail": [item.detail for item in items],
"next_steps": ["; ".join(item.next_steps) for item in items],
}
def _item_rows(items: Sequence[AssessmentItem], *, next_steps: bool = True) -> list[list[str]]:
"""The printable rows of a report table, with or without the next-step column."""
return [
[item.name, item.status.value, item.detail]
+ (["; ".join(item.next_steps)] if next_steps else [])
for item in items
]
def _retained(item: AssessmentItem, what: str) -> Any:
"""The payload an operation kept, or a ``KeyError`` naming what did not run."""
if item._report is _AbsentReport.TOKEN:
raise KeyError(f"{what} {item.name!r} did not run")
return item.report
def _distinct_steps(items: Sequence[AssessmentItem]) -> tuple[str, ...]:
"""Every suggested next step across ``items``, in its first-seen order."""
return tuple(dict.fromkeys(step for item in items for step in item.next_steps))
def _grouped_by_status(
presented: Sequence[tuple[str, AssessmentItem]],
) -> dict[AssessmentStatus, list[tuple[str, AssessmentItem]]]:
"""Presented rows by status, each group in presentation order.
One pass reads as one statement about the rows. The alternative filters ``presented``
once per status and once more for the returned results, which reads as several
independent selections that happen to agree. A report holds about sixteen rows, so
this is a readability choice and not a speed one.
"""
grouped: dict[AssessmentStatus, list[tuple[str, AssessmentItem]]] = {}
for surface, item in presented:
grouped.setdefault(item.status, []).append((surface, item))
return grouped
def _inventory_rows(
grouped: Mapping[AssessmentStatus, Sequence[tuple[str, AssessmentItem]]],
statuses: Sequence[AssessmentStatus],
) -> list[list[str]]:
"""Compact rows for ``statuses``: a status and its count once, then its operations.
A status with no row contributes nothing. The second and later rows of a status leave
the status and count cells empty, so the count reads as one claim about the group.
"""
rows: list[list[str]] = []
for status in statuses:
group = grouped.get(status, ())
for index, (surface, item) in enumerate(group):
rows.append(
[
status.value if index == 0 else "",
str(len(group)) if index == 0 else "",
f"{surface}.{item.name}",
]
)
return rows
def _summary_section(title: str, headers: Sequence[str], rows: Sequence[Sequence[str]]) -> str:
"""One titled and underlined section, or ``none`` when it holds no row."""
body = format_table(headers, rows) if rows else "none"
return f"{title}\n{'-' * len(title)}\n{body}"
[docs]
@dataclass(frozen=True)
class DiagnosticReport:
"""Collect statuses from a combined diagnostic or sensitivity run.
Parameters
----------
items : tuple of AssessmentItem
Results for the requested operations.
include_refits : bool
Whether the run allowed operations that refit nuisance models.
include_retargets : bool
Whether the run allowed moderate retargets, beyond the default cheap retargets.
backend : str or None
Dataframe backend used by :meth:`to_frame` when ``data`` is omitted.
See Also
--------
ValidationReport : The battery that reads stored artifacts only.
cleverly.assessment.DiagnosticsFacade : What produces this report.
cleverly.AssessmentCapability : The declaration behind one item.
Notes
-----
A skipped or refused item remains in the report. Caller-deferred work has its own
status. A capability known to be unsupported is an omission. A refusal raised during
an aggregate run remains unavailable, and later operations still run.
Examples
--------
>>> from cleverly import AssessmentStatus, DiagnosticReport
>>> from cleverly.assessment import AssessmentItem
>>> item = AssessmentItem("support", AssessmentStatus.PASSED, "no material warning")
>>> report = DiagnosticReport(items=(item,))
>>> report["support"].status == AssessmentStatus.PASSED
True
"""
items: tuple[AssessmentItem, ...]
#: The two cost classes a caller can opt into, declared separately because they are
#: disjoint: ``refute`` and ``benchmark`` refit nuisances without retargeting, and
#: ``truncation_curve``, ``missingness`` and ``tipping_gamma`` retarget cached
#: nuisances without refitting any. Folding them into one flag made whichever class
#: it did not name a silent rider on the other.
include_refits: bool = False
include_retargets: bool = False
backend: str | None = None
def __getitem__(self, name: str) -> AssessmentItem:
"""Return the report item named ``name``.
Parameters
----------
name
Operation name to retrieve.
Returns
-------
AssessmentItem
Matching report item.
Raises
------
KeyError
If the report does not contain ``name``.
"""
for item in self.items:
if item.name == name:
return item
raise KeyError(f"no diagnostic named {name!r}; have {[item.name for item in self.items]}")
[docs]
def to_frame(self, data: Any = None) -> Any:
"""Return report items as a dataframe.
Parameters
----------
data : Any
Optional dataframe whose backend selects the output type.
Returns
-------
Any
A pandas or Polars dataframe with one row per operation.
"""
return emit_frame(_item_columns(self.items), data, backend=self.backend)
[docs]
def report(self, name: str) -> Any:
"""Return the retained object for an operation that ran.
Parameters
----------
name : str
Operation name.
Returns
-------
Any
The operation's retained report or dataframe.
Raises
------
KeyError
If the operation did not run.
"""
return _retained(self[name], "diagnostic")
[docs]
def reports(self) -> dict[str, Any]:
"""Return retained objects for operations that ran.
Returns
-------
dict of str to Any
Operation names mapped to their retained reports.
"""
return {
item.name: item.report for item in self.items if item._report is not _AbsentReport.TOKEN
}
[docs]
def next_steps(self) -> tuple[str, ...]:
"""Return de-duplicated next steps in report order.
Returns
-------
tuple of str
Suggested follow-up actions in their first-seen order.
"""
return _distinct_steps(self.items)
[docs]
def summary(self) -> str:
"""Return a printable table of operation statuses.
Returns
-------
str
A printable table, one line per requested operation.
"""
return format_table(["diagnostic", "status", "detail", "next step"], _item_rows(self.items))
[docs]
@dataclass(frozen=True)
class ValidationReport:
"""Collect results from the inexpensive validation battery.
Parameters
----------
items : tuple of AssessmentItem
Validation checks and their statuses.
backend : str or None
Dataframe backend used by :meth:`to_frame` when ``data`` is omitted.
See Also
--------
DiagnosticReport : The combined run, which may retarget or refit.
cleverly.estimators.TMLEResult : Carries the artifacts this battery reads.
cleverly.validation.score_check : One of the checks the battery runs.
Notes
-----
The default battery reads stored artifacts. It does not refit nuisance
models.
Examples
--------
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from cleverly import ATE, CausalStudy, PointTreatment
>>> from cleverly.datasets import make_linear_ate
>>> frame, _ = make_linear_ate(n=200, seed=0)
>>> study = CausalStudy(
... frame,
... design=PointTreatment(
... outcome="Y", treatment="A", adjustment=("W1", "W2", "W3", "W4")
... ),
... )
>>> result = study.identify(ATE()).estimate(
... outcome_learner=LinearRegression(),
... treatment_learner=LogisticRegression(max_iter=1000),
... cross_fit=False,
... random_state=0,
... )
>>> report = result.validate()
>>> len(report.items) > 0
True
"""
items: tuple[AssessmentItem, ...]
backend: str | None = None
@property
def passed(self) -> bool:
"""Return whether every required check passed."""
return all(item.status not in _BLOCKING for item in self.items)
def __bool__(self) -> bool:
return self.passed
def __getitem__(self, name: str) -> AssessmentItem:
"""Return the validation item named ``name``.
Parameters
----------
name
Check name to retrieve.
Returns
-------
AssessmentItem
Matching validation item.
Raises
------
KeyError
If the report does not contain ``name``.
"""
for item in self.items:
if item.name == name:
return item
raise KeyError(f"no validation check named {name!r}")
[docs]
def to_frame(self, data: Any = None) -> Any:
"""Return validation items as a dataframe.
Parameters
----------
data : Any
Optional dataframe whose backend selects the output type.
Returns
-------
Any
A pandas or Polars dataframe with one row per check.
"""
return emit_frame(_item_columns(self.items), data, backend=self.backend)
[docs]
def summary(self) -> str:
"""Return a printable summary.
Returns
-------
str
A printable table, one line per validation check.
"""
heading = "Validation: PASS" if self.passed else "Validation: ATTENTION REQUIRED"
return "\n".join(
[
heading,
"-" * len(heading),
format_table(
["check", "status", "detail"],
_item_rows(self.items, next_steps=False),
),
]
)
[docs]
@dataclass(frozen=True)
class AssessmentReport:
"""Collect validation, diagnostics, and sensitivity in one post-fit battery.
Parameters
----------
validation : ValidationReport
Checks that read stored fitted artifacts.
diagnostics : DiagnosticReport
Method and support diagnostics.
sensitivity : DiagnosticReport
Sensitivity analyses and explicit omissions.
"""
validation: ValidationReport
diagnostics: DiagnosticReport
sensitivity: DiagnosticReport
@property
def backend(self) -> str | None:
"""Return the backend inherited from the validation surface."""
return self.validation.backend
def _presented(self) -> tuple[tuple[str, AssessmentItem], ...]:
owned = {item.name for item in self.validation.items}
return (
*(("validation", item) for item in self.validation.items),
*(("diagnostics", item) for item in self.diagnostics.items if item.name not in owned),
*(("sensitivity", item) for item in self.sensitivity.items),
)
@property
def attention(self) -> tuple[AssessmentItem, ...]:
"""Return rows with an explicit failure or warning."""
return tuple(item for _, item in self._presented() if item.status in _ATTENTION)
@property
def omissions(self) -> tuple[AssessmentItem, ...]:
"""Return rows that were deferred, not applicable, or unavailable."""
return tuple(item for _, item in self._presented() if item.status in _OMISSIONS)
[docs]
def next_steps(self) -> tuple[str, ...]:
"""Return de-duplicated next steps in presentation order.
Returns
-------
tuple of str
Suggested follow-up actions in their first-seen order.
"""
return _distinct_steps([item for _, item in self._presented()])
[docs]
def report(self, name: str, *, surface: str | None = None) -> Any:
"""Return one retained report by operation and optional surface.
Parameters
----------
name : str
Operation name.
surface : {"validation", "diagnostics", "sensitivity"} or None
Surface to select when names overlap.
Returns
-------
Any
Retained operation report.
"""
if surface is not None:
if surface not in {"validation", "diagnostics", "sensitivity"}:
raise KeyError(f"unknown assessment surface {surface!r}")
candidates = tuple((surface, item) for item in getattr(self, surface).items)
else:
candidates = self._presented()
matches = [(owner, item) for owner, item in candidates if item.name == name]
if not matches:
raise KeyError(f"no presented assessment report named {name!r}")
if len(matches) > 1:
owners = [owner for owner, _ in matches]
raise KeyError(
f"assessment report {name!r} is ambiguous across {owners}; pass surface="
)
return _retained(matches[0][1], "assessment operation")
[docs]
def to_frame(self, data: Any = None) -> Any:
"""Return one row per presented surface and operation.
Parameters
----------
data : Any
Optional dataframe whose backend selects the output type.
Returns
-------
Any
A pandas or Polars dataframe with a ``surface`` column.
"""
rows = self._presented()
return emit_frame(
_item_columns([item for _, item in rows], surface=[surface for surface, _ in rows]),
data,
backend=self.validation.backend,
)
[docs]
def summary(self) -> str:
"""Return completed analyses, then compact check and omission inventories.
Returns
-------
str
Printable completed-analysis rows followed by every other status.
"""
grouped = _grouped_by_status(self._presented())
result_rows = [
[surface, item.name, item.detail]
for surface, item in grouped.get(AssessmentStatus.COMPLETED, ())
]
check_rows = _inventory_rows(grouped, _SUMMARY_CHECK_ORDER)
omission_rows = _inventory_rows(grouped, _SUMMARY_OMISSION_ORDER)
return "\n\n".join(
[
_summary_section(
"Returned results", ("surface", "operation", "result"), result_rows
),
_summary_section("Checks", _INVENTORY_HEADERS, check_rows),
_summary_section("Not run", _INVENTORY_HEADERS, omission_rows),
_SUMMARY_FOOTER,
]
)
[docs]
@dataclass(frozen=True)
class Replayability:
"""Describe which post-fit actions a saved result can reproduce.
Parameters
----------
summarize_existing_artifacts : bool
Whether stored diagnostics can be summarized.
retarget_cached_nuisances : bool
Whether targeting can run again from cached primary nuisances. A guarded DR-TMLE
curve can refit reduced regressions inside that step.
evaluate_stored_representer : bool
Whether the stored representer can evaluate another parameter.
refit_nuisances : bool
Whether the result retains enough configuration for nuisance refits.
evaluate_new_data : bool
Whether the fitted result can score new observations.
unreconstructible : tuple of str
Missing components that prevent reconstruction.
"""
summarize_existing_artifacts: bool
retarget_cached_nuisances: bool
evaluate_stored_representer: bool
refit_nuisances: bool
evaluate_new_data: bool
unreconstructible: tuple[str, ...] = ()
def replayability(result: Any) -> Replayability:
"""Derive replay capabilities from stored artifacts and the normalized method."""
if _family(result) == "longitudinal":
from .longitudinal.estimator import (
LONGITUDINAL_REPLAY_MSM_DECLARATION,
LONGITUDINAL_REPLAY_REGIMEN_DECLARATION,
)
from .longitudinal.msm import refuse_evaluated_msm_functions
from .longitudinal.regimen import refuse_regimen_rules
missing = list(result.replay_recipe.omissions)
# These are the same guards the replay runs. A modified result may retain every
# learner template and evaluated array yet lack the declarations needed to use
# them as a new fitted result.
try:
refuse_regimen_rules(result.config.regimens)
except (CapabilityError, DataError):
missing.append(LONGITUDINAL_REPLAY_REGIMEN_DECLARATION)
try:
refuse_evaluated_msm_functions(result.msm)
except (CapabilityError, DataError):
missing.append(LONGITUDINAL_REPLAY_MSM_DECLARATION)
return Replayability(True, False, False, not missing, False, tuple(missing))
estimator = getattr(result, "estimator", None)
if estimator is None:
return Replayability(True, False, False, False, False, ("estimator configuration",))
refuse_functions = getattr(estimator, "_refuse_undeclared_functions", None)
if callable(refuse_functions) and not declarations_pass(refuse_functions):
return Replayability(True, False, False, False, False, (POINT_REPLAY_DECLARATION,))
# The checks a refit runs before any learner, read without fitting. A result whose
# estimator was modified to a configuration that refit() refuses keeps its cached
# nuisances, so it still retargets, and only the refit slot reads false.
refit_refusal = getattr(estimator, "_refit_configuration_refusal", None)
data = getattr(result, "data", None)
if callable(refit_refusal) and data is not None and refit_refusal(data) is not None:
return Replayability(True, True, False, False, False, (POINT_REPLAY_REFIT_CONFIGURATION,))
return Replayability(True, True, False, True, False)
#: What each :class:`Replayability` slot lets an operation do, in the words a refusal uses.
_REPLAY_WORK: dict[str, str] = {
"summarize_existing_artifacts": "summarizing the stored artifacts",
"retarget_cached_nuisances": "retargeting the cached nuisances",
"evaluate_stored_representer": "evaluating the stored representer",
"refit_nuisances": "refitting the nuisance models",
"evaluate_new_data": "scoring new observations",
}
def _require_argument_mapping(arguments: Any) -> None:
"""Refuse a per-operation argument block that is not a mapping.
Owned here rather than restated per caller: ``assess_result`` has to split the block
across two facades before either can validate it, so it reaches the same wrong type
first and must refuse it in the same words.
"""
if arguments is not None and not isinstance(arguments, Mapping):
raise TypeError("arguments must be a mapping from operation names to mappings")
def _accepts_arguments(facade: _CapabilityFacade, operation: str) -> bool:
"""Whether ``operation`` declares a parameter a caller could fill.
Read from the routed callable rather than from a list of names, so an operation that
gains or loses a parameter needs no second edit here. The first parameter is the
result or the facade itself, which no caller supplies.
"""
function, _ = facade._routed_callable(operation)
return len(inspect.signature(function).parameters) > 1
def _tilt_methods(item: AssessmentCapability, method: str) -> AssessmentCapability:
"""List ``ipsi`` on an IPSI fit only, and only ``ipsi`` there.
An odds-tilt plan refits only by IPSI, so an IPSI fit names IPSI alone. A fit by
another method keeps its own family, which does not include IPSI.
"""
if "ipsi" not in item.methods or item.result_family != "longitudinal":
return item
admitted = tuple(name for name in item.methods if (name == "ipsi") == (method == "ipsi"))
return replace(item, methods=admitted)
def _method_gated(item: AssessmentCapability, method: str) -> AssessmentCapability:
"""Refuse a row the fitted method does not declare."""
if (
not item.available
or method in item.methods
or (method == "unknown" and item.execution == "summarize")
):
return item
text = (
# An artifact records a method or it does not. Reading "the fitted method
# 'unknown' does not support this operation" sent a reader to look up a method
# named ``unknown``, when the cause is that the artifact names no method at all.
"this artifact records no fitted method, so its support for this operation "
"cannot be established"
if method == "unknown"
else f"the fitted method {method!r} does not support this operation"
)
return replace(
item,
available=False,
status=AssessmentStatus.NOT_APPLICABLE,
refusal=Refusal(RefusalReason.INVALID_DECLARATION, text),
)
#: What a code means when nothing is missing from the artifact. Most omissions are a
#: component the stored result does not carry, and the default sentence below says so. An
#: unseeded or object-seeded learner is the exception: the template *is* retained, and what
#: it cannot do is come back from a clone with the same draws. Saying "no longer carries"
#: there sends a reader to look for a field that nothing dropped.
_REPLAY_ARTIFACT_MISSING = "this stored result no longer carries the components it needs"
POINT_REPLAY_DECLARATION = "point_replay_function_declaration"
POINT_REPLAY_REFIT_CONFIGURATION = "point_replay_refit_configuration"
def _replay_omission_causes() -> dict[str, str]:
"""Return the per-code explanations that are not a missing component.
Returns
-------
dict of str to str
Stable omission code to the sentence a refusal uses for it.
"""
from .longitudinal.estimator import (
LONGITUDINAL_REPLAY_MSM_DECLARATION,
LONGITUDINAL_REPLAY_RANDOM_STATE_NON_INTEGER,
LONGITUDINAL_REPLAY_RANDOM_STATE_UNSEEDED,
LONGITUDINAL_REPLAY_REGIMEN_DECLARATION,
)
return {
POINT_REPLAY_DECLARATION: (
"a stored regime or MSM function lacks an accepted known-function declaration; "
"refit with the original functions declared known to restore recomputation"
),
POINT_REPLAY_REFIT_CONFIGURATION: (
"this result's estimator holds a configuration that refit() refuses before a "
"learner, such as a stratified fold policy or an undeclared scale for a "
"cross-fitted continuous outcome; the cached nuisances still retarget, and a new "
"fit under a supported configuration restores refits"
),
LONGITUDINAL_REPLAY_RANDOM_STATE_UNSEEDED: (
"a retained learner template declares no random_state, so a clone of it cannot "
"repeat this fit's draws"
),
LONGITUDINAL_REPLAY_RANDOM_STATE_NON_INTEGER: (
"a retained learner template carries a random generator object rather than an "
"integer seed, so a clone of it cannot repeat this fit's draws"
),
LONGITUDINAL_REPLAY_REGIMEN_DECLARATION: (
"a stored regimen rule lacks an accepted known-function declaration; refit the "
"original regimen with rule_kind='known' to restore replay"
),
LONGITUDINAL_REPLAY_MSM_DECLARATION: (
"a stored MSM projection lacks proof that its design and weights were declared "
"known; refit the original MSM with known functions to restore replay"
),
}
def _replay_gated(item: AssessmentCapability, replay: Replayability) -> AssessmentCapability:
"""Refuse a row whose declared replay slot this stored result cannot supply."""
slot = item.requires_replay
if not item.available or slot is None or getattr(replay, slot):
return item
codes = list(replay.unreconstructible)
explained = _replay_omission_causes()
# One sentence per distinct cause, in the order the codes are reported, so a result
# blocked for two reasons states both rather than the first one twice.
causes = dict.fromkeys(explained.get(code, _REPLAY_ARTIFACT_MISSING) for code in codes)
# No declared row pairs a false slot with an empty code list today, so this branch is
# hardening. Without it the sentence reads "is unavailable: . reported codes: []", which
# a reader takes for a rendering defect rather than for the refusal it is.
detail = (
f"{'; '.join(causes)}. reported codes: {codes}"
if causes
else "this stored result reports no cause for it"
)
return replace(
item,
available=False,
status=AssessmentStatus.UNAVAILABLE,
refusal=Refusal(
RefusalReason.INVALID_DECLARATION,
f"{_REPLAY_WORK[slot]} is unavailable: {detail}",
),
)
def _argument_resolved(
capability: AssessmentCapability,
argument: str,
requested: Any,
refusal: Callable[[Any], Refusal | None],
alternatives: Sequence[Any],
) -> AssessmentCapability:
"""Resolve a row whose call refuses some values of one argument and not others.
``refusal`` is the predicate the call raises from, read at one value of ``argument``,
and ``None`` stands for the omitted argument. The row follows the one rule that
``tipping_gamma(use_ci=True)`` follows.
======================================================= ===================================
request row
======================================================= ===================================
the argument is omitted, and the default runs unchanged
the argument is omitted, the default is refused, and ``deferred`` on ``argument``, with
a value in ``alternatives`` runs the call's sentence
the argument is omitted, and no value runs ``unavailable``, with the
default's sentence
the argument has a value the call refuses ``unavailable``, with the call's
sentence
======================================================= ===================================
A deferral lifts once the caller names the argument, because each request resolves
the row afresh.
Parameters
----------
capability : AssessmentCapability
The row after every gate no argument can lift.
argument : str
The name of the argument the rule resolves.
requested : Any
The value the request supplies, or ``None`` when it omits the argument.
refusal : callable
The call's predicate at one value of ``argument``: the exact refusal record, or
``None`` when that value runs.
alternatives : sequence
The values that could lift a deferral of the omitted argument.
Returns
-------
AssessmentCapability
The row this request reads.
"""
declared = refusal(requested)
if declared is None:
return capability
if requested is None and any(refusal(value) is None for value in alternatives):
return replace(
capability,
available=False,
status=AssessmentStatus.DEFERRED,
refusal=declared,
requires_arguments=(*capability.requires_arguments, argument),
)
return replace(
capability, available=False, status=AssessmentStatus.UNAVAILABLE, refusal=declared
)
#: The estimand default that a fit reporting no bare ``"ate"`` leaves ambiguous.
_AMBIGUOUS_ESTIMAND = "ate"
def _defaults_to_ambiguous_estimand(function: Callable[..., Any]) -> bool:
"""Whether calling ``function`` without an estimand asks for the bare ``"ate"``.
Read from the signature rather than from a list of operation names, so the next
operation that ships this default is gated the day it is written. It is a different
question from :attr:`SensitivityRoute.needs_estimand`, which says *where* the estimand
goes: ``benchmark`` takes ``covariates`` positionally and ``estimand="ate"`` by
keyword, so the positional flag is ``False`` while the default is as ambiguous as any
other. Gating on that flag left ``benchmark`` reporting ``unavailable`` beside six
rows that reported ``deferred`` for the identical missing choice.
Parameters
----------
function : callable
The routed implementation an operation dispatches to.
Returns
-------
bool
True when the target declares an ``estimand`` parameter defaulting to ``"ate"``.
"""
parameter = inspect.signature(function).parameters.get("estimand")
return parameter is not None and parameter.default == _AMBIGUOUS_ESTIMAND
def _tipping_natural_course_refusal(
result: Any, operation: str, candidates: tuple[str, ...]
) -> Refusal | None:
"""Why a bare ``tipping_gamma()`` needs an estimand on a fit that reads the natural course.
One spelling for the capability row and the call. ``None`` unless the operation is
``tipping_gamma``, the fit reports ``ey_obs``, ``par`` or ``paf`` with missing outcomes,
and exactly one arm parameter is left, which the facade would otherwise substitute.
"""
if operation != "tipping_gamma" or len(candidates) != 1:
return None
if not reads_natural_course_mean(result):
return None
natural = natural_course_names(result.estimates)
return Refusal(
RefusalReason.IMPLEMENTATION_GAP,
"tipping_gamma answers for one estimand, and the implemented missingness tilt is "
f"arm-specific: {natural} read the natural-course mean, which needs a "
"natural-course sensitivity parameter that is not implemented.",
alternative=f"Choose an explicit estimand from {list(candidates)}.",
roadmap="X65",
)
def _default_estimand_candidates(result: Any, eligible: Container[str]) -> tuple[str, ...]:
"""Reported parameters an operation defaulting to ``"ate"`` is left to choose between.
Eight operations across the two facades declare ``estimand="ate"`` and answer for one
parameter: the five omitted-variable analyses, ``tipping_gamma``,
``simulated_confounding`` and ``refute``. A fit that reports a bare ``"ate"`` settles
the choice, and this returns nothing. A fit that reports none has to be asked which
one, and the answer separates three cases a combined report has to tell apart:
============ =============================================================
length what the caller is owed
============ =============================================================
``0`` nothing the operation applies to, so no argument makes it run
``1`` one facade fills this name in, another defers on it
``2`` upward the ambiguity only the caller can settle, which is a deferral
============ =============================================================
Length one is therefore not one answer. ``_CapabilityFacade._substitutes_estimand``
says which facade this fit is asking, because a facade that supplies nothing leaves
the operation to run on its ambiguous default and refuse.
The length-zero and the length-two answers are the two the report used to give one
status. A multi-arm fit reports ``ate[high vs low]`` and ``ate[medium vs low]``, and
the bound refuses the bare default with "estimand 'ate' was not requested in this
fit", which is also what an incremental fit hears when the bound applies to nothing it
reports. The first is a choice; the second is a missing derivation.
Parameters
----------
result : Any
Fitted result whose reported parameters settle the choice.
eligible : container of str
Names the operation can answer for. Reported parameters outside it are not
choices, so they are not counted towards the ambiguity.
Returns
-------
tuple of str
The eligible reported names, in the order the fit reports them.
"""
if "ate" in result.estimates:
return ()
return tuple(name for name in result.estimates if name in eligible)
class _CapabilityFacade:
"""Lookup, refusal, and combined-report machinery shared by both public facades.
The two facades answer different questions from different declarations, but the way
they *route* a question is one algorithm: find the operation's row, refuse by that row
when it is unavailable, and in a combined report skip what the caller has not paid for.
Written twice, it drifted five ways -- the availability and cost checks in opposite
orders, two spellings of the cost gate, one refusal that re-derived a reason the record
already carried, different caught-exception sets, and one side that discarded what its
operations returned and so could only ever report ``passed``. Subclasses supply
:attr:`_declared` and the two labels below; everything else is settled here.
A subclass declares what its operations *are*. It does not decide whether this stored
result can run them: the method gate and the replay gate live here and run over every
row. Both were once written per facade, and ``refute`` reached the shipped surface
claiming ``available=True`` on a result with no estimator because the diagnostics side
patched one row by name and forgot the other.
"""
#: What an operation of this kind is called in a refusal: ``diagnostic 'refute' is ...``.
_kind: str
#: The attribute a caller reaches it through, for the ``next_steps`` of a skipped row.
_attribute: str
#: Whether this facade supplies the sole eligible estimand on the caller's behalf.
#: It settles what one candidate means. A facade that substitutes runs the row under
#: that name and defers only at two, and a facade that substitutes nothing has to
#: defer at one: nothing else fills the gap, so the operation would be invoked on its
#: ambiguous default and refuse. ``refute`` did exactly that and reported
#: ``unavailable`` on a fit where naming its one reported alias runs it.
_substitutes_estimand: bool = False
def __init__(self, result: Any) -> None:
self._result = result
def __getstate__(self) -> dict[str, Any]:
"""Persist the facade without the verdicts it memoized.
A result carries its facades in ``__dict__``, and ``save`` pickles the whole result,
so every :func:`~functools.cached_property` on a facade the caller has touched is
written into the artifact. Those memos are derived state: ``_declared``,
``_capability_map`` and ``_evalue_selections`` all restate what this version of the
package concludes from the stored artifacts, so derived state is not persisted.
Dropping every memo rather than the three by name, because the next one added is
stale in an artifact the moment it is written, and recomputing costs one pass over
the declarations.
Returns
-------
dict of str to Any
The instance state, without the entries a ``cached_property`` owns.
"""
return without_memos(type(self), self.__dict__)
@property
def _declared(self) -> tuple[AssessmentCapability, ...]:
raise NotImplementedError # pragma: no cover - subclasses declare their own
@property
def capabilities(self) -> tuple[AssessmentCapability, ...]:
"""Return declared operations, gated by the fitted method and by replayability."""
return tuple(self._capability_map.values())
@cached_property
def _gated_map(self) -> dict[str, AssessmentCapability]:
"""Every declared row after the method gate and the replay gate, before any request.
The two gates that no argument can lift. :meth:`_request_gated` resolves a request
on top of this map, so a row that an argument can lift is never stored lifted or
refused: the replay gate still sees it, and each request resolves it afresh.
"""
method = _method(self._result)
replay = replayability(self._result)
return {
item.operation: _replay_gated(
_method_gated(_tilt_methods(item, method), method), replay
)
for item in self._declared
}
@cached_property
def _capability_map(self) -> dict[str, AssessmentCapability]:
"""Every row resolved for the bare request, which supplies no argument.
The estimand gate and the request gate both apply, as they do to any request.
"""
return {operation: self._resolved(row, {}) for operation, row in self._gated_map.items()}
def _lookup(
self, rows: Mapping[str, AssessmentCapability], operation: str
) -> AssessmentCapability:
"""One row of ``rows``, refused by name when the facade declares none."""
try:
return rows[operation]
except KeyError:
raise KeyError(f"unknown {self._kind} {operation!r}") from None
def _gated(self, operation: str) -> AssessmentCapability:
"""One row of :attr:`_gated_map`, refused by name when the facade declares none."""
return self._lookup(self._gated_map, operation)
#: The request gate of each operation whose call refuses some argument values and not
#: others, as the name of the method that resolves it. The method is looked up by name
#: at each request, so a test that replaces it on the class replaces the gate.
_request_gates: ClassVar[Mapping[str, str]] = {}
def _request_gated(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Resolve a row for the arguments of one request.
A row whose call refuses some argument values and not others is resolved here, from
the predicate that the call raises from, by the method that :attr:`_request_gates`
names for the operation. The bare request, a combined report's request and a
direct call each pass their own arguments, so each reads the answer the call gives.
A row with no gate is returned unchanged. So is a row that a gate no argument
lifts has already refused.
Parameters
----------
capability : AssessmentCapability
The row after the method gate and the replay gate.
arguments : mapping of str to Any
Arguments the caller supplied for this operation.
Returns
-------
AssessmentCapability
The row this request reads.
"""
gate = self._request_gates.get(capability.operation)
if gate is None or not capability.available:
return capability
resolved: AssessmentCapability = getattr(self, gate)(capability, arguments)
return resolved
def _resolved(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""The estimand gate and then the request gate, applied to one gated row.
The estimand gate runs first, so a request that names no estimand defers on the
estimand before any other argument is read.
"""
return self._request_gated(self._estimand_gated(capability, arguments), arguments)
def capability(self, operation: str) -> AssessmentCapability:
"""Return the row of one operation, resolved for a request that supplies no argument.
The row passes the method gate, the replay gate, the estimand gate and the request
gate. So an operation whose ``estimand`` default this fit leaves ambiguous reads
``deferred`` on ``estimand``, and a call that names no estimand refuses. A direct
call that names one resolves its own request and does not read this row.
Parameters
----------
operation : str
A declared operation of this facade.
Returns
-------
AssessmentCapability
The row of the bare request.
Raises
------
KeyError
If this facade declares no such operation.
"""
return self._lookup(self._capability_map, operation)
def _argument_default(self, operation: str, argument: str) -> Any:
"""The value a call of ``operation`` that omits ``argument`` runs with.
Read from the routed signature, after the substitution this facade makes, so a
request gate resolves an omitted argument exactly as the call does.
"""
bound = self._bind_arguments(operation, {}, partial=True)
bound.apply_defaults()
return bound.arguments[argument]
def _require(
self, operation: str, arguments: Mapping[str, Any] | None = None
) -> AssessmentCapability:
"""Refuse a direct call by the row its request resolves to.
A call whose row an argument can lift or refuse passes its own arguments, so a
request the bare row defers still runs once the caller names the argument.
``arguments=None`` stands for a call that supplies none, which is what an
operation whose row no argument changes needs. The estimand gate is not applied:
a direct call resolves its own estimand default and raises its own refusal, which
names the estimands the fit reports. So a call that names an estimand is never
refused by the deferral that :meth:`capability` reports for the bare request.
"""
item = self._request_gated(self._gated(operation), arguments or {})
if not item.available:
# ``reason`` first, and ``interpretation`` only as a fallback: the record knows
# why *this* operation is refused, and re-deriving one from the result family
# gave an E-value on a longitudinal fit a rationale about pseudo-outcome
# recursion, which is a true sentence about a different operation.
# The raise carries the row's own record, so the code and the alternative the
# row declares are the ones the caller sees.
declared = item.refusal
if declared is None:
declared = Refusal(RefusalReason.INVALID_DECLARATION, item.interpretation)
raise CapabilityError.from_refusal(
replace(
declared,
condition=f"{self._kind} {operation!r} is {item.status.value}: "
f"{declared.condition}",
)
)
return item
def _estimand_candidates(self, operation: str) -> tuple[str, ...]:
"""Reported parameters this operation is left to choose between, or ``()``.
The one place either facade decides whether a fit leaves an operation's estimand
ambiguous. :meth:`_estimand_gated` reads it to defer the row, and
:meth:`SensitivityFacade._with_default_parameter` reads it to substitute the sole
eligible name. Written twice, the two disagreed: the substitution declined to
guess between two contrasts and the row still advertised itself as runnable, so a
combined report invoked the operation and published the refusal as ``unavailable``
with a next step that named no argument.
An operation that takes no ``estimand``, or whose default this fit settles,
returns ``()`` and is never gated. :func:`_default_estimand_candidates` states
what each length means.
Parameters
----------
operation : str
Declared operation on this facade.
Returns
-------
tuple of str
Eligible reported names, empty for an operation that faces no choice.
"""
return ()
def _estimand_gated(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Defer a row whose ``estimand`` default this fit leaves ambiguous.
The deferral is resolved per request, exactly as the E-value's is. A caller who
names a supported estimand gets the row back untouched and the operation runs, so
the row cannot stay deferred once the argument that lifts it is supplied. That
matters because :meth:`_skipped` defers unconditionally on the status.
Only the ambiguity moves. An operation the fit refuses outright keeps its own
``unavailable`` answer, because no argument makes a missing derivation, a missing
replay artifact or an unsupported contrast run.
Parameters
----------
capability : AssessmentCapability
The declared row, already gated by method and by replayability.
arguments : mapping of str to Any
Arguments the caller supplied for this operation.
Returns
-------
AssessmentCapability
The row, deferred on the estimand when this request leaves it ambiguous.
"""
if not capability.available:
# An availability refusal outranks a choice: the caller cannot name their way
# past a derivation this family does not have.
return capability
if "estimand" in capability.requires_arguments:
# ``simulated_confounding`` already declares the argument, so the
# missing-argument gate reports it and names ``grid`` beside it.
return capability
if arguments.get("estimand") is not None:
# Read the value, not the key. ``estimand=None`` is the documented public
# default, so that spelling has to defer exactly as the bare request does.
return capability
candidates = self._estimand_candidates(capability.operation)
natural = _tipping_natural_course_refusal(self._result, capability.operation, candidates)
if natural is not None:
# The facade does not substitute the one arm mean on a fit that also reports a
# natural-course target, so the row declares the argument the call needs.
return replace(
capability,
available=False,
status=AssessmentStatus.DEFERRED,
refusal=natural,
requires_arguments=(*capability.requires_arguments, "estimand"),
)
if len(candidates) < (2 if self._substitutes_estimand else 1):
# One candidate is a deferral or a substitution, and which one it is belongs
# to the facade rather than to the count. See ``_substitutes_estimand``.
return capability
return replace(
capability,
available=False,
status=AssessmentStatus.DEFERRED,
refusal=Refusal(
RefusalReason.INVALID_DECLARATION,
f"{capability.operation} answers for one estimand; choose an explicit "
f"estimand from {list(candidates)}",
),
requires_arguments=(*capability.requires_arguments, "estimand"),
)
def _capability_for_arguments(
self, operation: str, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Resolve request-specific availability and cost before aggregate execution."""
return self._resolved(self._gated(operation), arguments)
def _skipped(
self,
capability: AssessmentCapability,
arguments: Mapping[str, Any],
*,
include_refits: bool,
include_retargets: bool,
) -> AssessmentItem | None:
"""The row a combined report owes an operation, or ``None`` to run it.
A deferred row first, then availability, then the missing argument, and the cost
last. Every gate above the cost gate refuses for a reason no flag can pay off, and a
report that named the cost first sent the caller to ``include_refits=True`` and
then, on the very next call, to the argument it never mentioned. A refusal has to
name the first thing that is wrong.
Each gate decides the sentence alone, and this method attaches the request once on
the way out. Two of the three branches used to copy ``arguments`` themselves and
the third dropped them, so a combined report published two omissions that named the
caller's request and one that did not.
The availability branch tells two refusals apart by the gated row, which no argument
changes. A row refused fit-wide equals its gated row and describes no invocation,
so it carries no arguments to replay. A row that this request refuses, such as
``truncation_curve(mechanism=False)`` on an incremental fit, differs from its gated
row. It carries the invocation the request describes, with the signature defaults
bound, as an item whose call refused after it was invoked does. Its sentence names
the refused value, so it needs no next step of its own.
Parameters
----------
capability : AssessmentCapability
The row resolved for this request, not the bare declaration.
arguments : mapping of str to Any
Arguments the caller supplied for this operation.
include_refits : bool
Whether the caller paid for operations that refit nuisance models.
include_retargets : bool
Whether the caller paid for moderate retargets.
Returns
-------
AssessmentItem or None
The omission to report, or ``None`` when the operation may run.
"""
item: AssessmentItem | None
missing = tuple(name for name in capability.requires_arguments if name not in arguments)
if capability.status is AssessmentStatus.DEFERRED and capability.requires_arguments:
# A deferred row is refused by this request, not by the fit: the operation runs
# once the caller names the argument the row declares. Routed by status rather
# than by key membership, because the E-value defers on the *value* -- an
# explicit ``estimand=None`` is the documented public default and stays
# ambiguous. Testing membership alone let that spelling past both gates, into
# the invocation, and out through the generic refusal handler as ``unavailable``
# with no argument to act on, while the bare call reported ``deferred``.
item = _missing_argument_item(
capability,
self._attribute,
capability.requires_arguments,
reason=capability.reason,
refusal=capability.refusal,
)
elif not capability.available:
item = _item_from_capability(capability)
if capability == self._gated(capability.operation):
# Refused fit-wide, before any request was considered: the operation did
# not run, and no argument makes it run. Stamping the caller's request onto
# it, seed included, would report "no longitudinal benchmarking derivation
# is registered" beside a ``random_state`` that nothing ever drew from.
return item
return replace(
item, arguments=self._invocation_arguments(capability.operation, arguments)
)
elif missing:
item = _missing_argument_item(capability, self._attribute, missing)
else:
item = _cost_refusal(capability, self._attribute, include_refits, include_retargets)
if item is None:
return item
return replace(item, arguments=dict(arguments))
def _invocation_arguments(self, operation: str, arguments: Mapping[str, Any]) -> dict[str, Any]:
"""The invocation a request describes, with the signature defaults bound.
A partial binding, because a request that a row refuses need not name every
required argument: the bare ``benchmark`` request of a fit with one covariate names
none.
"""
return _bound_arguments(self._bind_arguments(operation, arguments, partial=True))
def _run_all(
self,
*,
include_refits: bool,
include_retargets: bool,
supplied: Mapping[str, Mapping[str, Any]],
random_state: int | None,
) -> DiagnosticReport:
"""Execute the combined report from arguments :meth:`_validated_arguments` cleared.
Validation is the caller's job because binding every supplied operation's signature
is not free, and the battery reaches this method through ``assess_result``, which
has already had to validate both facades to fail before either one runs. Doing it
again here bound each signature a second and a third time per ``assess()`` call.
"""
def compute() -> DiagnosticReport:
items = []
for declared in self.capabilities:
operation_arguments = dict(supplied.get(declared.operation, {}))
capability = self._capability_for_arguments(declared.operation, operation_arguments)
if capability.accepts_random_state and random_state is not None:
# Before the gates, not after them. The seed is part of the request for
# every row that accepts one, and injecting it inside the invocation
# branch left a deferred ``refute`` row reporting ``arguments == {}``
# after ``run_all(random_state=7)``. That row is not deterministic, so a
# caller replaying it from ``item.arguments`` reproduced a different
# draw from the one the report would have taken.
operation_arguments["random_state"] = random_state
skipped = self._skipped(
capability,
operation_arguments,
include_refits=include_refits,
include_retargets=include_retargets,
)
if skipped is not None:
items.append(skipped)
continue
try:
token = _RETAIN_PACKED.set(True)
try:
report = getattr(self, capability.operation)(**operation_arguments)
finally:
_RETAIN_PACKED.reset(token)
except CapabilityError as error:
# A capability refusal is an expected result of asking a broad battery to
# inspect one fitted object. Keep it visible as an omission and continue to
# later rows. Catch only the refusal type: ``KeyError`` and ``TypeError``
# are structural defects, and turning either into a scientific-sounding
# report row would hide an implementation error.
items.append(
AssessmentItem(
capability.operation,
AssessmentStatus.UNAVAILABLE,
f"the operation declined this request: {error}",
(
f"call result.{self._attribute}.{capability.operation}() "
f"directly for the refusal in full",
),
arguments=self._effective_arguments(
capability.operation, operation_arguments, None
),
refusal=error.refusal,
)
)
else:
effective = self._effective_arguments(
capability.operation, operation_arguments, report
)
interpreted = INTERPRETERS[capability.operation](
report, self._result, operation_arguments
)
items.append(
replace(
interpreted,
_report=_pack_cached(report, self._result.data.backend),
arguments=effective,
)
)
return DiagnosticReport(
tuple(items),
include_refits=include_refits,
include_retargets=include_retargets,
backend=self._result.data.backend,
)
return _cached(
self._result,
f"{self._attribute}.run_all",
(),
{
"include_refits": include_refits,
"include_retargets": include_retargets,
"arguments": supplied,
"random_state": random_state,
},
compute,
)
def _validated_arguments(
self,
arguments: Mapping[str, Mapping[str, Any]] | None,
random_state: int | None,
) -> dict[str, dict[str, Any]]:
_require_argument_mapping(arguments)
if arguments is None:
return {}
declared = {row.operation for row in self.capabilities}
unknown = sorted(set(arguments) - declared)
if unknown:
raise KeyError(f"unknown {self._kind} operation(s): {unknown}")
validated: dict[str, dict[str, Any]] = {}
for operation, values in arguments.items():
if not isinstance(values, Mapping):
raise TypeError(f"arguments[{operation!r}] must be a mapping")
_, kwargs = _single_pass((), values)
capability = self.capability(operation)
if random_state is not None and "random_state" in kwargs:
raise ValueError(
f"random_state was supplied both to run_all and arguments[{operation!r}]"
)
if "random_state" in kwargs and not capability.accepts_random_state:
raise TypeError(f"{operation!r} does not accept random_state")
bound = self._bind_arguments(
operation, kwargs, partial=True, resolve=capability.available
)
if operation == "refute":
_validate_refute_arguments(bound)
elif operation == "benchmark":
_validate_benchmark_arguments(self._result, bound)
validated[operation] = kwargs
return validated
def _routed_callable(self, operation: str) -> tuple[Callable[..., Any], bool]:
if self._attribute == "sensitivity":
route = SENSITIVITY_ROUTES[operation]
module = importlib.import_module(f".sensitivity.{route.module}", __package__)
return getattr(module, route.function), True
if operation == "refute":
from .validation.refute import refute
return refute, True
return getattr(type(self), operation), False
def _with_default_parameter(
self, operation: str, args: tuple[Any, ...], kwargs: dict[str, Any]
) -> tuple[tuple[Any, ...], dict[str, Any]]:
"""Resolve a sensitivity parameter in the subclass that owns those routes."""
raise NotImplementedError # pragma: no cover - diagnostics never need this route
def _bind_arguments(
self,
operation: str,
kwargs: Mapping[str, Any],
*,
partial: bool,
args: tuple[Any, ...] = (),
resolve: bool = True,
) -> inspect.BoundArguments:
function, result_first = self._routed_callable(operation)
if resolve and self._attribute == "sensitivity":
# Every routed operation, not the positional ones alone: the substitution
# decides for itself whether this target takes an estimand and where it goes.
args, kwargs = self._with_default_parameter(operation, args, dict(kwargs))
first = self._result if result_first else self
signature = inspect.signature(function)
binder = signature.bind_partial if partial else signature.bind
return binder(first, *args, **kwargs)
def _effective_arguments(
self,
operation: str,
kwargs: Mapping[str, Any],
report: Any,
*,
args: tuple[Any, ...] = (),
) -> dict[str, Any]:
bound = self._bind_arguments(operation, kwargs, partial=False, args=args)
effective = _bound_arguments(bound, report)
if operation == "evalue" and effective.get("estimand") is None:
# Explicit OR requests choose the approximation; keep None replayable there.
source = getattr(report, "source_estimand", None)
if source is not None:
from .sensitivity.evalue import _DERIVED_RR
selection = self._evalue_selection(None)
if (
selection.branch != _DERIVED_RR
or _arm_source_target(self._result, source) != "or"
):
effective["estimand"] = source
return effective
def _evalue_selection(self, estimand: str | None) -> Any:
raise NotImplementedError
def _invoke(self, operation: str, args: tuple[Any, ...], kwargs: Mapping[str, Any]) -> Any:
function, _ = self._routed_callable(operation)
bound = self._bind_arguments(operation, kwargs, partial=False, args=args)
effective = _bound_arguments(bound)
def compute() -> Any:
return function(*bound.args, **bound.kwargs)
if operation == "evalue":
from .sensitivity.evalue import _evalue_from_selection
selection = self._evalue_selection(effective["estimand"])
def compute() -> Any:
return _evalue_from_selection(self._result, selection)
report = _cached(self._result, f"{self._attribute}.{operation}", (), effective, compute)
resolved = _bound_arguments(bound, report)
if _normalize(resolved) != _normalize(effective):
self._result.assessment_cache[
_cache_key(f"{self._attribute}.{operation}", (), resolved)
] = _pack_cached(report, self._result.data.backend)
return report
def _arm_source_target(result: Any, source: str) -> str:
from .sensitivity._parameters import arm_parameter_keys
return str(arm_parameter_keys(result)[source].estimand)
def _bound_arguments(bound: inspect.BoundArguments, report: Any = None) -> dict[str, Any]:
"""Canonicalize one public invocation for execution, caching, and replay."""
bound.apply_defaults()
effective = dict(bound.arguments)
effective.pop(next(iter(bound.signature.parameters)), None)
for name, parameter in bound.signature.parameters.items():
if parameter.kind is inspect.Parameter.VAR_KEYWORD:
effective.update(effective.pop(name, {}))
if effective.get("random_state") is None:
resolved = getattr(report, "random_state", getattr(report, "root_seed", None))
if resolved is not None:
effective["random_state"] = resolved
# ``omitted_variable_bounds`` resolves an omitted strength to 0.03 itself, because a
# ``benchmark=`` must be told apart from an explicit strength. Record what it ran, except
# beside a benchmark, where a replay passing both strengths would conflict with it.
if effective.get("benchmark") is None:
for name in ("cf_y", "cf_d"):
if name in effective and effective[name] is None:
resolved = getattr(report, name, None)
if resolved is not None:
effective[name] = resolved
return effective
def _single_pass(
args: tuple[Any, ...], kwargs: Mapping[str, Any]
) -> tuple[tuple[Any, ...], dict[str, Any]]:
"""Read each one-shot iterator argument once, into a tuple.
A facade checks an argument before it calls the operation, a combined report records
it for replay, and the assessment cache keys the answer by it. A generator is empty
after its first read, so the call then ran on no names and the report recorded an
exhausted iterator. Its ``repr`` names a memory address that CPython reuses, so a
cache key built from it could return the answer of another generator. Every other
value passes through unchanged.
"""
def once(value: Any) -> Any:
return tuple(value) if isinstance(value, Iterator) else value
return tuple(once(value) for value in args), {
name: once(value) for name, value in kwargs.items()
}
def _validate_refute_arguments(bound: inspect.BoundArguments) -> None:
"""Check a bound refutation request before a facade starts any operation."""
from .validation.refute import _validated_tests
bound.apply_defaults()
_validated_tests(
bound.arguments["tests"],
bound.arguments["n_replicates"],
bound.arguments["negative_control_outcome"],
)
def _validate_benchmark_arguments(result: Any, bound: inspect.BoundArguments) -> None:
"""Check benchmark names before a combined report can skip or refit an operation."""
if _family(result) == "point" and "covariates" in bound.arguments:
from .sensitivity.omitted_variable import _benchmark_names
_benchmark_names(result, bound.arguments["covariates"])
def _missing_argument_item(
capability: AssessmentCapability,
attribute: str,
missing: tuple[str, ...],
*,
reason: str | None = None,
refusal: Refusal | None = None,
) -> AssessmentItem:
"""The skip a combined report owes an operation whose argument it cannot choose.
A combined report cannot run an operation until every required argument is present.
Choosing a value here -- which covariates to benchmark against -- would be a scientific
choice made silently on the caller's behalf.
A row may declare more than one argument, so the sentence has to agree in number.
``", ".join`` alone rendered "an explicit grid, estimand argument", which reads as one
argument named "grid, estimand".
``reason`` is the sentence *this* deferral prints, and the gate passes it in rather than
this builder reading ``capability.reason``, which means something else: the explanation
a refused row owes its caller. The two coincide on the one row that carries both, and
the printed sentence must not depend on that coincidence. ``None`` keeps the generic
sentence, which is what a row with a declared but unsupplied argument -- ``benchmark``
without ``covariates`` -- has to say.
"""
needed = missing[0] if len(missing) == 1 else f"{', '.join(missing[:-1])} and {missing[-1]}"
phrase = (
f"an explicit {needed} argument" if len(missing) == 1 else f"explicit {needed} arguments"
)
return AssessmentItem(
capability.operation,
AssessmentStatus.DEFERRED,
reason or f"needs {phrase}, which a combined report has no basis to choose",
(f"call result.{attribute}.{capability.operation}() directly with {needed}",),
refusal=refusal,
)
def _cost_refusal(
capability: AssessmentCapability,
attribute: str,
include_refits: bool,
include_retargets: bool,
) -> AssessmentItem | None:
"""The skip a combined report owes an operation the caller has not paid for.
Two flags rather than one because the two costs are disjoint: refutation and
benchmarking refit nuisances without retargeting, and the missingness tilt retargets
cached nuisances without refitting any. One flag made whichever class it did not name
run silently under the other's permission.
The class is read off the row rather than off the operation name, because the same
operation is in different classes on different fits. A point-treatment
``truncation_curve`` retargets the cached nuisances. A guarded DR-TMLE fit resolves it
into the ``refit`` class, its targeting alternation refitting the reduced regressions at
every bound, and a longitudinal fit resolves it there too, repeating the complete
backward recursion per bound. See :func:`assessment_capabilities`.
"""
allowed = {
"summarize": True,
"refit": include_refits,
"retarget": include_retargets or capability.cost == "cheap",
}
if allowed[capability.execution]:
return None
work = "refits nuisance models" if capability.execution == "refit" else "retargets the fit"
flag = "include_refits" if capability.execution == "refit" else "include_retargets"
return AssessmentItem(
capability.operation,
AssessmentStatus.DEFERRED,
f"not run by default because it {work}; pass {flag}=True",
(f"call result.{attribute}.{capability.operation}() directly, or pass {flag}=True",),
)
[docs]
class DiagnosticsFacade(_CapabilityFacade):
"""Access diagnostics supported by a fitted causal result.
Parameters
----------
result : TMLEResult or LongitudinalResult
Fitted point-treatment or longitudinal result.
See Also
--------
SensitivityFacade : The analyses that ask what would overturn the estimate.
cleverly.AssessmentCapability : One row of :attr:`capabilities`.
cleverly.DiagnosticReport : What :meth:`run_all` returns.
Notes
-----
Access this facade through ``result.diagnostics``. Inspect
:attr:`capabilities` before optional or potentially expensive operations.
Examples
--------
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from cleverly import ATE, CausalStudy, PointTreatment
>>> from cleverly.datasets import make_linear_ate
>>> frame, _ = make_linear_ate(n=200, seed=0)
>>> study = CausalStudy(
... frame,
... design=PointTreatment(
... outcome="Y", treatment="A", adjustment=("W1", "W2", "W3", "W4")
... ),
... )
>>> result = study.identify(ATE()).estimate(
... outcome_learner=LinearRegression(),
... treatment_learner=LogisticRegression(max_iter=1000),
... cross_fit=False,
... random_state=0,
... )
>>> support = result.diagnostics.support()
>>> support.n
200
"""
_kind = "diagnostic"
_attribute = "diagnostics"
@cached_property
def _declared(self) -> tuple[AssessmentCapability, ...]:
return assessment_capabilities(self._result)
#: ``truncation_curve`` resolves ``mechanism`` and ``refute`` resolves ``tests``.
_request_gates: ClassVar[Mapping[str, str]] = {
"truncation_curve": "_truncation_gated",
"refute": "_refute_gated",
}
def _truncation_gated(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Resolve the point-treatment ``truncation_curve`` row from the call's predicate.
:func:`~cleverly.sensitivity.positivity.truncation_refusal` is what the call raises
from, and :func:`_argument_resolved` applies the request rule to ``mechanism``.
An incremental fit with missing outcomes defers on ``mechanism``, because the
default axis is refused and ``mechanism=True`` runs. An incremental fit with
complete outcomes reads ``unavailable``, because neither axis runs.
A longitudinal row is returned as it is. The longitudinal curve has its own two
refusals.
"""
if capability.result_family != "point":
return capability
from .sensitivity.positivity import truncation_refusal
return _argument_resolved(
capability,
"mechanism",
arguments.get("mechanism"),
lambda axis: truncation_refusal(self._result, axis),
(True, False),
)
def _refute_gated(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Resolve the ``refute`` row from the call's predicate.
:func:`~cleverly.validation.refute.refute_refusal` is what the call raises from, and
:func:`_argument_resolved` applies the request rule to ``tests``. An omitted
``tests`` means the default tests. A fit given ``split_plan=`` defers on
``tests``, because the default tests include ``subset`` and ``placebo`` still
runs. So does the natural-course mean, which refuses ``placebo``.
The row resolves for the estimand the call would refit, which is the signature
default when the request names none. The estimand gate runs before this one and
defers a request that leaves the estimand ambiguous, so only a direct call reads
the refusal of an unreported default here, in the call's own sentence. The other
arguments that decide a refusal, such as ``n_replicates`` and the generated-outcome
and measurement-error declarations, are read from the request as it stands.
"""
from .validation.refute import DEFAULT_TESTS, refute_refusal
estimand = arguments.get("estimand")
if estimand is None:
estimand = self._argument_default(capability.operation, "estimand")
read = inspect.signature(refute_refusal).parameters
declared = {
name: value
for name, value in arguments.items()
if name in read and name not in ("estimand", "tests")
}
def request_refusal(tests: Sequence[str] | None) -> Refusal | None:
chosen = DEFAULT_TESTS if tests is None else tests
return refute_refusal(self._result, estimand=estimand, tests=chosen, **declared)
return _argument_resolved(
capability,
"tests",
arguments.get("tests"),
request_refusal,
tuple((name,) for name in DEFAULT_TESTS),
)
def _estimand_candidates(self, operation: str) -> tuple[str, ...]:
"""The reported aliases ``refute`` may be asked to choose between.
``refute`` refits under one alias and refuses every name the fit did not report,
so its eligible set is the reported set itself. Which diagnostics face the choice
is read from the routed signature, exactly as the sensitivity side reads it:
``refute`` is the one diagnostic today whose ``estimand`` defaults to the bare
``"ate"``, and a second one is gated the day it is written rather than the day
somebody remembers this method. :data:`SENSITIVITY_ROUTES` answers a different
question, which is where in :mod:`cleverly.sensitivity` an analysis is
implemented, and this facade routes to :func:`cleverly.validation.refute`.
No substitution follows from a single candidate here, which is why
``_substitutes_estimand`` stays false and one candidate defers the row. The
sensitivity facade may fill in a sole eligible parameter because its analyses
answer for whichever contrast they are given and it has always done so;
``refute`` refits, and refitting under a name the caller never wrote is a
scientific choice made on their behalf.
Parameters
----------
operation : str
Declared diagnostic operation.
Returns
-------
tuple of str
Reported aliases for ``refute``, and ``()`` for every other diagnostic.
"""
function, _ = self._routed_callable(operation)
if not _defaults_to_ambiguous_estimand(function):
return ()
return _default_estimand_candidates(self._result, self._result.estimates)
[docs]
def support(self) -> Any:
"""Return the support diagnostic for the fitted intervention.
Returns
-------
PositivityReport or LongitudinalDiagnostics or SupportReport or dict
Support diagnostic specialized for arms, regimens, shifts, or incremental
interventions.
"""
self._require("support")
if _family(self._result) == "longitudinal":
return _cached(
self._result,
"diagnostics.support",
(),
{},
lambda: _longitudinal_stagewise(self._result),
)
def compute() -> Any:
nuisance = self._result.nuisance
from .interventions import (
check_incremental_support,
check_policy_support,
check_support,
)
from .interventions.policy import declared_support_bounds
from .sensitivity.positivity import positivity_report
def score_arguments(group: str) -> dict[str, Any]:
"""The fitted score artifact for one group, as ``check_*`` keywords.
A group with no fluctuation at all reports no equations, which is a
different state from a fluctuation that retains no score weights, and the
two get different reasons in the report.
"""
fluctuation = self._result.fluctuations.get(group)
return {
"absolute_score_weights": getattr(fluctuation, "absolute_score_weights", None),
"equations": () if fluctuation is None else tuple(fluctuation.names),
"n_repeats": self._result.n_repeats,
}
if nuisance.regimes is not None:
return check_support(
nuisance.regimes,
self._result.data.treatment,
nuisance.propensity.values,
backend=self._result.data.backend,
**score_arguments("regime"),
)
# ``shifts`` alone, not ``shifts and density``: a shift fit without a fitted
# density is a broken shift fit, and the density-ratio report says so by name.
# Adding the second condition sent it to the arm-level report instead, which
# answers a different question or refuses for an unrelated reason.
if nuisance.policies is not None:
bound = self._result.config.missingness_bound
level = self._result.intermediate_value
mechanisms = [
values
for values in (
nuisance.bounded_missingness(bound),
None if level is None else nuisance.intermediate_density(level, bound),
)
if values is not None
]
return check_policy_support(
nuisance.policies,
nuisance.density,
self._result.data.treatment,
propensity=(nuisance.propensity.values if self._result.data.has_arms else None),
mechanisms=mechanisms,
folds=nuisance.folds,
declared_support=declared_support_bounds(self._result.data),
**score_arguments("mtp"),
)
if nuisance.incremental is not None:
return check_incremental_support(
nuisance.incremental,
self._result.data.treatment,
levels=self._result.data.treatment_levels,
**score_arguments("ipsi"),
)
return positivity_report(self._result)
return _cached(self._result, "diagnostics.support", (), {}, compute)
[docs]
def nuisance_models(self) -> Any:
"""Return method-aware held-out nuisance and split diagnostics.
Returns
-------
NuisanceDiagnostics or LongitudinalNuisanceDiagnostics
Held-out diagnostics for point or sequential nuisance models. Point-treatment
reports also retain C-TMLE selection state and pair repeated-split spread with
each reported standard error when those artifacts exist.
"""
self._require("nuisance_models")
def compute() -> Any:
if _family(self._result) == "longitudinal":
return _longitudinal_nuisances(self._result)
from .validation.nuisance import nuisance_diagnostics
return nuisance_diagnostics(self._result)
return _cached(self._result, "diagnostics.nuisance_models", (), {}, compute)
[docs]
def influence(self) -> Any:
"""How few rows carry each estimate's influence curve.
A direct call only. The operation is not part of :meth:`run_all` or the assessment
battery, so adding it moves no combined report. The fit already warns with
:class:`~cleverly.exceptions.SparseInfluenceWarning` when a Wald fit needs it.
With clusters every measure reads the cluster sums of the curve.
Returns
-------
InfluenceReport
One row per estimate (:mod:`cleverly.inference.sparse`): the Kish count, the
skewness, the raw kurtosis, the relative SD of the variance estimate, the
Edgeworth-predicted Wald coverage loss, and, with an attached bootstrap, the BCa
bias correction and acceleration.
Examples
--------
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from cleverly.datasets import make_longitudinal
>>> from cleverly.longitudinal import LTMLE
>>> frame, _ = make_longitudinal(n=200, seed=0)
>>> result = LTMLE(
... {"always": 1, "never": 0},
... outcome_learner=LinearRegression(),
... treatment_learner=LogisticRegression(max_iter=1000),
... censoring_learner=LogisticRegression(max_iter=1000),
... ).fit(
... frame,
... outcome="Y",
... treatment=["A1", "A2"],
... baseline=["W1", "W2"],
... time_varying=[[], ["L2"]],
... censoring=["C1", "C2"],
... )
>>> report = result.diagnostics.influence()
>>> report.sparse
('ey_regimen[always]', 'ey_regimen[never]', 'ate_regimen[never vs always]')
"""
def compute() -> Any:
from .inference.sparse import influence_report
alpha = float(getattr(self._result.config, "alpha_sig", 0.05))
return influence_report(
self._result.estimates,
alpha=alpha,
backend=self._result.data.backend,
cluster=getattr(self._result.data, "cluster", None),
)
return _cached(self._result, "diagnostics.influence", (), {}, compute)
[docs]
def score_equations(self, *, tolerance: float = DEFAULT_TOLERANCE) -> Any:
"""Whether targeting solved the score equations this fit relies on.
Parameters
----------
tolerance : float
Relative tolerance used to evaluate the fitted score equations.
Returns
-------
ScoreCheck or LongitudinalScoreDiagnostics
Score-equation checks for the fitted result family.
Notes
-----
``tolerance`` gates both families but on the scale each one's score lives on, and
the two are not interchangeable. A point-treatment fit compares the score in the
outcome's own units against ``tolerance * se / sqrt(n)``
(:data:`~cleverly.validation.score.DEFAULT_TOLERANCE` says why that shape). A
longitudinal fit bounds each node's *relative* score -- the largest component as a
fraction of its maximum possible magnitude -- which is the quantity the sequential
targeting loop gates on, and it can only tighten a node's verdict beyond the fit's
own convergence flag.
**A cross-fitted longitudinal fit gets one row per node**, as a single-fold fit
does, or one per node and term on a working model. Its one pooled fluctuation per
node solves the node's score over every follower, so the ``"solver"`` row gates
that score with ``tolerance``.
"""
self._require("score_equations")
def compute() -> Any:
if _family(self._result) == "longitudinal":
return _longitudinal_scores(self._result, tolerance=tolerance)
from .validation.score import score_check
return score_check(self._result, tolerance=tolerance)
return _cached(
self._result,
"diagnostics.score_equations",
(),
{"tolerance": tolerance},
compute,
)
[docs]
def corrections(
self,
*,
tolerance: float = DEFAULT_TOLERANCE,
identity_tolerance: float = IDENTITY_TOLERANCE,
) -> Any:
"""Return correction-equation checks for a DRTMLE fit.
Parameters
----------
tolerance : float
Tolerance for the reduced correction score.
identity_tolerance : float
Tolerance for the correction identity.
Returns
-------
CorrectionCheck
Correction diagnostics for the fitted result.
Raises
------
CapabilityError
If the fitted method does not use the correction system.
"""
self._require("corrections")
from .validation.drtmle import correction_check
return _cached(
self._result,
"diagnostics.corrections",
(),
{"tolerance": tolerance, "identity_tolerance": identity_tolerance},
lambda: correction_check(
self._result,
tolerance=tolerance,
identity_tolerance=identity_tolerance,
),
)
[docs]
def truncation_curve(
self,
bounds: Sequence[float] | Sequence[CumulativeGBounds] | None = None,
*,
estimands: Sequence[str] | None = None,
mechanism: bool | None = None,
) -> Any:
"""Evaluate fitted estimates across a sequence of mechanism bounds.
Point-treatment and longitudinal results intentionally have different
contracts. A point-treatment curve retargets the stored nuisance fits. A
longitudinal curve repeats the complete backward recursion while reusing the
fitted treatment/censoring mechanism, folds, prepared data, scaler, and frozen
resolved regimen plans.
A longitudinal fit refuses ``mechanism=True``. It has one cumulative
treatment-and-censoring bound, so there is no separate observation mechanism for a
second axis to vary. It also requires an explicit ``bounds`` grid, because it has no
per-parameter fitted pair to build a default grid around.
Parameters
----------
bounds : sequence of float or cumulative bound pairs, or None
For a point-treatment result, lower bounds to evaluate; treatment-mechanism
scalars use symmetric ``(bound, 1 - bound)`` bounds, while observation- and
intermediate-mechanism scalars use ``(bound, 1)``. Point-treatment defaults,
including each requested parameter's exact fitted pair, are used when omitted.
A longitudinal result requires an explicit, non-empty sequence. Each
longitudinal scalar ``b`` is cumulative ``(b, 1)`` shorthand and each
``(lower, upper)`` pair is preserved exactly.
estimands : sequence of str or None
Reported estimands to include. All compatible estimands are the default.
mechanism : bool or None
Vary the observation mechanism rather than treatment. ``None`` resolves to
the only axis the fit has: the observation mechanism for a missing-outcome
``NaturalCourseMean`` fit, which estimates no treatment law, and treatment
for every other fit. Passing ``False`` explicitly on a natural-course fit
names an axis that fit does not have, and is refused rather than redirected.
An incremental fit refuses the treatment axis, because its estimand contains
the propensity. Its row defers on ``mechanism`` when the fit also estimated an
observation mechanism, and reads ``unavailable`` when it did not.
Returns
-------
dataframe
Point-treatment output contains retargeted estimates and uncertainty at each
requested bound. Longitudinal output is descriptive only: replayed point
estimates, the exact fitted reference and movement from it, and score-cell
truncation counts; it contains no standard errors or confidence intervals.
Raises
------
CapabilityError
If the requested curve changes the estimand, names an axis the fit does not
have, or cannot be replayed. The sentence is the one
``cleverly.sensitivity.positivity.truncation_refusal`` returns. A
longitudinal result also refuses an omitted ``bounds`` grid and
``mechanism=True``, which are the two requests its contract above has no answer
to. On a declared known mechanism, either family refuses an explicit bound pair
that the fit itself refuses, because that pair would truncate a declared value.
The point-treatment default grid omits such a pair.
"""
# The row this request resolves to, not the bare row. It carries every
# point-treatment refusal of the axis, read from the predicate the module call
# raises from, and it is checked before the cache, so an unanswerable request is
# refused whether or not the cache already holds a curve.
self._require("truncation_curve", {"mechanism": mechanism})
# The cache key holds these arguments, so a one-shot iterator is read once here.
(bounds, estimands), _ = _single_pass((bounds, estimands), {})
longitudinal = _family(self._result) == "longitudinal"
if longitudinal and bounds is None:
raise CapabilityError(
"a longitudinal truncation curve requires an explicit bounds grid",
reason=RefusalReason.INVALID_DECLARATION,
)
if longitudinal and mechanism:
raise CapabilityError(
"a longitudinal fit has one cumulative treatment-and-censoring bound; "
"mechanism=True is a point-treatment option",
reason=RefusalReason.INVALID_DECLARATION,
)
if not longitudinal:
from .sensitivity.positivity import truncation_axis
# Resolved for the cache key, so the omitted argument and the axis it resolves
# to share one entry. An explicit value is kept, and the row above has already
# refused one this fit cannot sweep.
mechanism = truncation_axis(self._result, mechanism)
# The longitudinal call takes no ``None``, and the refusal above is eager, so the
# empty fallback here is unreachable on that path.
grid: Sequence[CumulativeGBounds] = () if bounds is None else bounds
def compute() -> Any:
if longitudinal:
from .longitudinal.estimator import longitudinal_truncation_curve
return longitudinal_truncation_curve(self._result, grid, estimands=estimands)
from .sensitivity.positivity import truncation_curve
return truncation_curve(self._result, bounds, estimands=estimands, mechanism=mechanism)
# One call with the family branch inside ``compute``, as ``nuisance_models`` and
# ``score_equations`` are: two calls were two key shapes for one operation, and the
# longitudinal one omitted ``mechanism`` from a key whose caller can pass it.
return _cached(
self._result,
"diagnostics.truncation_curve",
(bounds,),
{"estimands": estimands, "mechanism": mechanism},
compute,
)
[docs]
def refute(self, **kwargs: Any) -> Any:
"""Refit the analysis under the requested data perturbations.
Parameters
----------
**kwargs
Options forwarded to :func:`cleverly.validation.refute`.
Returns
-------
RefutationResult
Refutation results for each requested perturbation.
Raises
------
TypeError
If an argument is not one :func:`cleverly.validation.refute` takes.
ValueError
If a test name is unknown, ``n_replicates`` is invalid, or the requested
negative-control test has no outcome. These checks precede every refusal.
CapabilityError
If the fitted estimator cannot be reconstructed, or the row this request
resolves to is refused. A fit given ``split_plan=`` refuses ``subset``, and
the natural-course mean refuses ``placebo``, so each row defers on ``tests``.
A call that omits ``tests`` raises the deferral with the refused test's
sentence, and a call that names ``tests`` without that test runs.
"""
# The malformed arguments first, through the check the function runs first, so the
# facade and the function report a misspelled test before any refusal.
_, kwargs = _single_pass((), kwargs)
bound = self._bind_arguments("refute", kwargs, partial=False)
_validate_refute_arguments(bound)
# The row this request resolves to, not the bare row, so a request that names
# ``tests`` runs where the bare row defers on it. Its replay gate refuses a result
# whose estimator cannot refit.
self._require("refute", kwargs)
return self._invoke("refute", (), kwargs)
[docs]
def run_all(
self,
*,
include_refits: bool = False,
include_retargets: bool = False,
arguments: Mapping[str, Mapping[str, Any]] | None = None,
random_state: int | None = None,
) -> DiagnosticReport:
"""Run available diagnostics that need no new arguments.
Parameters
----------
include_refits : bool
Include operations that refit nuisance models.
include_retargets : bool
Include moderate retargets; cheap E-value retargets run by default.
arguments : mapping or None
Per-operation keyword arguments.
random_state : int or None
Common seed for stochastic refit operations.
Returns
-------
DiagnosticReport
One item for every diagnostic operation included in the combined report.
See Also
--------
cleverly.DiagnosticReport : The combined operation statuses.
DiagnosticsFacade.capabilities : Declare availability and cost before execution.
Examples
--------
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from cleverly import ATE, CausalStudy, PointTreatment
>>> from cleverly.datasets import make_linear_ate
>>> frame, _ = make_linear_ate(n=80, seed=1)
>>> study = CausalStudy(
... frame,
... design=PointTreatment(
... outcome="Y", treatment="A", adjustment=("W1", "W2", "W3", "W4")
... ),
... )
>>> result = study.identify(ATE()).estimate(
... outcome_learner=LinearRegression(),
... treatment_learner=LogisticRegression(max_iter=1000),
... cross_fit=False,
... random_state=0,
... )
>>> report = result.diagnostics.run_all()
>>> report.include_refits, report.include_retargets
(False, False)
"""
return self._run_all(
include_refits=include_refits,
include_retargets=include_retargets,
supplied=self._validated_arguments(arguments, random_state),
random_state=random_state,
)
def _item_from_capability(capability: AssessmentCapability) -> AssessmentItem:
return AssessmentItem(
capability.operation,
capability.status,
capability.reason or capability.interpretation,
refusal=capability.refusal,
)
def _score_item(
report: Any, result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
ratios = []
for row in getattr(report, "rows", ()):
if hasattr(row, "ratio"):
ratios.append(float(row.ratio))
else:
ratios.append(float(row.relative_score) / float(report.tolerance))
passed = bool(getattr(report, "passed", False))
conditioning = _reduction_conditioning_warning(result) if passed else None
status = (
AssessmentStatus.WARNING
if conditioning
else AssessmentStatus.PASSED
if passed
else AssessmentStatus.FAILED
)
worst = max((value for value in ratios if np.isfinite(value)), default=float("nan"))
detail = f"{len(ratios)} score row(s); worst abs(score) / threshold = {worst:.3g}"
if conditioning:
detail += f"; {conditioning}"
steps = (
("inspect result.repeats[*].fluctuations['mean'].reduction.ill_conditioned",)
if conditioning
else ()
if passed
else ("inspect result.diagnostics.score_equations()",)
)
return AssessmentItem("score_equations", status, detail, steps)
def _intervention_rows(report: Any) -> tuple[tuple[str, Any], ...]:
"""Every per-intervention record of a support report, whichever shape it arrived in.
A shift or incremental report *is* the mapping of its rows; a regime report keeps them
under ``regimes``. Both spellings reach the same records, and two readers that
enumerate them separately can come to disagree about which rows exist.
"""
interventions = report.items() if isinstance(report, Mapping) else ()
regimes = getattr(report, "regimes", {}).items()
return (*interventions, *regimes)
def _support_metrics(report: Any) -> tuple[float | None, float | None]:
truncated: list[float] = []
ess: list[float] = []
if hasattr(report, "truncated"):
truncated.append(float(report.truncated.get("fraction", 0.0)))
for values in getattr(report, "effective_sample_size", {}).values():
if "ratio" in values:
ess.append(float(values["ratio"]))
for values in getattr(report, "mechanisms", {}).values():
if "ess_ratio" in values:
ess.append(float(values["ess_ratio"]))
# Group rows carry two facts that must not be conflated. Their clipping fraction is
# the action of an estimand-specific propensity bound and belongs in the maximum.
# Their targeted ratio is concentration of absolute score load, not arm/mechanism
# effective sample size, so it is deliberately excluded from the pooled minimum.
for values in getattr(report, "group_score_load", {}).values():
if "clipped_fraction" in values:
clipped = float(values["clipped_fraction"])
if np.isfinite(clipped):
truncated.append(clipped)
if isinstance(report, LongitudinalDiagnostics):
truncated.extend(float(row.share_truncated) for row in report.rows)
ess.extend(float(row.effective_n / row.n_followed) for row in report.rows if row.n_followed)
if isinstance(report, Mapping):
# Only the shift and incremental rows carry a cap. A regime row has no
# `capped_fraction`, and defaulting one to zero for it would turn "this report
# truncates nothing" into a reported maximum of 0.0%.
truncated.extend(
float(capped)
for row in report.values()
if (capped := getattr(row, "capped_fraction", None)) is not None
)
ess.extend(float(getattr(row, "ess_ratio", np.nan)) for _, row in _intervention_rows(report))
clean_ess = [value for value in ess if np.isfinite(value)]
return (max(truncated) if truncated else None, min(clean_ess) if clean_ess else None)
def _support_facts(truncated: float | None, ess: float | None) -> list[str]:
"""Present the two support metrics, on one scale, for every row that reports them."""
facts = []
if truncated is not None:
facts.append(f"maximum truncated fraction {truncated:.1%}")
if ess is not None:
facts.append(f"minimum effective-sample-size ratio {ess:.1%}")
return facts
def _group_load_fact(report: Any) -> str | None:
"""The most concentrated score equation, separate from mechanism ESS.
A target-group row has no draw count of its own, so this reads the report's. Reading it
from the row instead printed "draw 01 of 01" for every repeated fit, which contradicted
the draw count :meth:`~cleverly.sensitivity.PositivityReport.summary` printed from the
same report. Copying the count onto each row would have made the two agree by making
the same number storable in two places, which is what let them disagree.
"""
group_draw = (REPORTED_DRAW, int(getattr(report, "n_repeats", 1)))
rows: list[tuple[str, str, Any, tuple[int, int] | None]] = [
("group", group, values, group_draw)
for group, values in getattr(report, "group_score_load", {}).items()
if np.isfinite(float(values.get("targeted_ratio", np.nan)))
]
# An intervention row records its own draw, so it is read from the row.
rows.extend(
("intervention", str(name), values, None)
for name, item in _intervention_rows(report)
if (values := getattr(item, "score_load", None)) is not None
and np.isfinite(float(values.get("targeted_ratio", np.nan)))
)
if not rows:
return None
kind, name, values, draw = min(rows, key=lambda item: float(item[2]["targeted_ratio"]))
rendered = format_score_load(values, style="detail", draw=draw)
return f"{kind} load: {name}:{values['equation']} {rendered}; not estimator ESS"
def _support_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
warning = _support_warning(report)
facts = _support_facts(*_support_metrics(report))
group_fact = _group_load_fact(report)
if group_fact is not None:
facts.append(group_fact)
if warning:
facts.append(warning)
detail = "; ".join(facts) if facts else "stored support report completed"
return AssessmentItem(
"support",
# `COMPLETED` rather than `PASSED`, because support has no pass criterion once the
# effective-sample-size ratio is reported rather than graded. What this row can
# still assert is a breach: truncation above the fitted bound's tolerance, or an
# intervention with estimated zero support. Absent one, the honest statement is
# that the check ran and produced numbers, and `PASSED` would read as a positivity
# clearance that no threshold here is entitled to give. `_nuisance_item` returns
# `COMPLETED` for the same reason.
AssessmentStatus.WARNING if warning else AssessmentStatus.COMPLETED,
detail,
() if warning is None else ("inspect result.diagnostics.support()",),
)
def _experimental_note(result: Any) -> str:
"""The ``; experimental: reason`` clause of a result with experimental estimates, or ``""``."""
estimates = getattr(result, "estimates", None)
if not estimates:
return ""
reasons: list[str] = []
for estimate in estimates.values():
support = getattr(estimate, "support", None)
if (
support is not None
and support.label == "experimental"
and support.reason not in reasons
):
reasons.append(support.reason)
return "".join(f"; experimental: {reason}" for reason in reasons)
def _nuisance_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
findings = tuple(getattr(report, "findings", ()))
if isinstance(report, LongitudinalNuisanceDiagnostics):
finite = [row.loss for row in report.rows if np.isfinite(row.loss)]
if not finite:
return AssessmentItem(
"nuisance_models",
AssessmentStatus.WARNING,
"no finite longitudinal nuisance loss is available",
("inspect result.diagnostics.nuisance_models()",),
)
detail = f"{len(finite)} longitudinal nuisance loss value(s) are available"
if report.omissions:
detail += f"; {len(report.omissions)} role omission(s) are recorded"
# The note a point-treatment report carries for the same status. The longitudinal
# report publishes no spread and holds no status, so the note is read off the
# result here. An interpreter called without a result adds none.
status = "influence_curve" if _result is None else _result.inference_status
if is_labelled(status):
detail += f"; {status_record(status).assessment_note}"
elif _result is not None:
# A parameter status sits beside an inferential fit status, so it is read per
# estimate: the note names how many parameters carry it.
flagged: dict[str, int] = {}
for estimate in _result.estimates.values():
if is_labelled(estimate.inference):
flagged[estimate.inference] = flagged.get(estimate.inference, 0) + 1
for flagged_status, count in flagged.items():
detail += f"; {count} parameter(s): {status_record(flagged_status).assessment_note}"
else:
facts = [
"; ".join(findings)
if findings
else f"{len(getattr(report, 'models', ()))} nuisance model report(s) are available"
]
n_repeats = int(getattr(report, "n_repeats", 1))
selection = getattr(report, "selection", None)
if selection is not None:
# The artifact says what it is. Discriminating a selector from an
# outcome-adaptive fit here as well is what let this line and the one in
# `NuisanceDiagnostics.summary` drift into two spellings of one fact.
reported = int(getattr(report, "reported_repeat", REPORTED_DRAW))
draw = f" on {format_draw(reported, n_repeats)}" if n_repeats > 1 else ""
facts.append(f"{selection.describe()}{draw}")
elif getattr(report, "selection_omission", None) is not None:
facts.append(f"C-TMLE selection unavailable: {report.selection_omission}")
# The report's own note, which is keyed on the declared inference status and not
# on the selection artifact: the outcome-adaptive path has one of those too and
# has a status of its own.
note = getattr(report, "inference_note", None)
if note is not None:
facts.append(note)
spread = tuple(getattr(report, "repeat_spread", ()))
if spread:
# The name the report's own table heads the column with.
ratio = spread_name("sd/se", getattr(report, "inference", "influence_curve"))
finite_rows: list[Any] = [
row for row in spread if np.isfinite(row.ratio_to_standard_error)
]
if finite_rows:
largest_row = max(finite_rows, key=lambda row: row.ratio_to_standard_error)
facts.append(
f"split spread for {len(spread)} parameter(s) across "
f"{n_repeats} draws; largest {ratio} "
f"{largest_row.ratio_to_standard_error:.3g} for {largest_row.estimand}"
)
else:
facts.append(
f"split spread for {len(spread)} parameter(s) across "
f"{n_repeats} draws; {ratio} unavailable"
)
spread_omission = getattr(report, "repeat_spread_omission", None)
if n_repeats > 1 and spread_omission is not None:
# Guarded on the draw count for the reason `NuisanceDiagnostics.summary` is:
# the one-draw reason is the ordinary state and states nothing a reader of an
# ordinary fit needs.
facts.append(f"split spread unavailable: {spread_omission}")
detail = "; ".join(facts)
detail += _experimental_note(_result)
return AssessmentItem(
"nuisance_models",
AssessmentStatus.WARNING if findings else AssessmentStatus.COMPLETED,
detail,
("inspect result.diagnostics.nuisance_models()",) if findings else (),
)
def _correction_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
identity = [abs(float(row.residual)) for row in report.rows if np.isfinite(row.residual)]
magnitude = [abs(float(row.reported)) for row in report.rows]
detail = (
f"contract={report.contract}; maximum identity residual "
f"{max(identity, default=float('nan')):.3g}; maximum reported correction magnitude "
f"{max(magnitude, default=float('nan')):.3g}"
)
return AssessmentItem(
"corrections",
AssessmentStatus.PASSED if report.passed else AssessmentStatus.FAILED,
detail,
() if report.passed else ("inspect result.diagnostics.corrections()",),
)
def _finite(value: Any) -> bool:
"""True when a cell carries a number a range or a baseline can use."""
return value is not None and bool(np.isfinite(value))
def _range(values: Sequence[Any]) -> tuple[float, float] | None:
finite = [float(value) for value in values if _finite(value)]
return (min(finite), max(finite)) if finite else None
def _format_range(values: tuple[float, float] | None) -> str:
return "no finite values" if values is None else f"[{values[0]:.4g}, {values[1]:.4g}]"
def _alias_rows(aliases: Sequence[Any]) -> dict[str, list[int]]:
"""Row positions of each parameter, keyed by alias in first-seen order."""
groups: dict[str, list[int]] = {}
for index, value in enumerate(aliases):
groups.setdefault(str(value), []).append(index)
return groups
def _format_alias_ranges(ranges: Mapping[str, tuple[float, float] | None]) -> str:
"""One comma-separated ``alias [low, high]`` list, in the order the mapping holds."""
return ", ".join(f"{alias} {_format_range(value)}" for alias, value in ranges.items())
def _evaluated_bound_detail(payload: Mapping[str, Sequence[Any]]) -> str:
"""Describe the one-dimensional point grid or two-dimensional cumulative grid."""
if "lower_bound" not in payload or "upper_bound" not in payload:
bounds = _range(payload.get("bound", payload.get("g_bound", ())))
return f"evaluated lower bounds {_format_range(bounds)}"
pairs: list[tuple[float, float]] = []
for lower, upper in zip(payload["lower_bound"], payload["upper_bound"], strict=True):
pair = (float(lower), float(upper))
if pair not in pairs:
pairs.append(pair)
rendered = ", ".join(f"({lower:.4g}, {upper:.4g})" for lower, upper in pairs)
return f"evaluated cumulative bound pairs [{rendered}]"
def _truncation_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
payload = _frame_payload(report)
evaluated = _evaluated_bound_detail(payload)
psi = payload.get("psi", payload.get("estimate", ()))
aliases = payload.get("estimand")
if not aliases:
detail = f"{evaluated}; estimate range {_format_range(_range(psi))}"
else:
groups = _alias_rows(aliases)
deltas = payload.get("delta_from_fitted")
markers = payload.get("is_fitted_bound")
# One signed range for each parameter, against that parameter's own fitted
# estimate. The maximum absolute movement is the larger endpoint's magnitude, so
# reporting the signed range keeps the direction and loses nothing. The fitted
# estimate and the fitted pair stay columns of the retained frame, and
# ``result.psi(name)`` returns the first: this row presents the movement on one
# scale rather than reprinting the curve.
scale = f"{len(groups)} parameter(s) over {evaluated}"
if deltas is None:
movement = _format_alias_ranges(
{alias: _range([psi[index] for index in rows]) for alias, rows in groups.items()}
)
detail = f"{scale}; estimate range: {movement}"
else:
movement = _format_alias_ranges(
{alias: _range([deltas[index] for index in rows]) for alias, rows in groups.items()}
)
detail = f"{scale}; signed movement from the fitted estimate: {movement}"
if markers is not None:
# An explicit ``bounds=`` grid keeps its cardinality, so it can omit a
# parameter's fitted pair. A count says how many curves start away from the
# estimate the fit reported. A missing column says nothing either way, so the
# clause is written only when the markers are there to read. The deltas are a
# separate column, so the markers decide this clause on their own.
omitted = sum(
not any(bool(markers[index]) for index in rows) for rows in groups.values()
)
if omitted:
detail += f"; fitted pair not evaluated for {omitted} of {len(groups)}"
return AssessmentItem(
"truncation_curve",
AssessmentStatus.COMPLETED,
detail,
)
def _refute_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
failed = [test.name for test in report.tests if not test.passed]
return AssessmentItem(
"refute",
AssessmentStatus.FAILED if failed else AssessmentStatus.PASSED,
"all refutation tests passed"
if not failed
else f"failed tests {failed}; inspect their retained draws",
()
if not failed
else tuple(f"inspect result.diagnostics.refute().draws_frame({name!r})" for name in failed),
)
def _omitted_item(
report: Any, _result: Any, arguments: Mapping[str, Any] | None = None
) -> AssessmentItem:
"""Interpret one omitted-confounding bound, and say which strengths produced it.
The bound is a statement about an assumed confounder, so a reader who did not choose
``cf_y`` and ``cf_d`` has to be told that the library did. The prefix is written here
rather than by the combined report because it is a fact about *this* operation, and a
name-matching branch in generic routing machinery is what the declaration tables exist
to remove.
``arguments=None`` means the caller's choices are unknown rather than absent, so the
row claims no provenance. ``run_all`` always passes the mapping it cleared, so only a
direct call reaches that case.
"""
spans = report.lower <= report.null_hypothesis <= report.upper
defaults = [] if arguments is None else [n for n in ("cf_y", "cf_d") if n not in arguments]
provenance = ""
benchmark = None if arguments is None else arguments.get("benchmark")
if benchmark is not None:
provenance = f"at the strengths of the benchmark against {list(benchmark.covariates)}; "
elif defaults:
provenance = (
"at the default strengths; "
if len(defaults) == 2
else f"at the default {defaults[0]} strength; "
)
return AssessmentItem(
"omitted_confounding",
AssessmentStatus.WARNING if spans else AssessmentStatus.COMPLETED,
f"{provenance}"
f"cf_y={report.cf_y:.3g}, cf_d={report.cf_d:.3g}, rho={report.rho:.3g}; "
f"bias-adjusted interval [{report.lower:.4g}, {report.upper:.4g}]",
() if not spans else ("inspect the retained omitted-confounding bounds",),
)
def _robustness_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
# ``robustness_value`` keys its limit value through ``spread_name`` and names a
# non-inferential status in the report itself, so a saved report reads the same name
# and prints the same noun with no fit at hand.
# Imported here because the sensitivity module imports this one.
from .sensitivity.omitted_variable import _UNREACHABLE_RV, limit_refusal_reason
status = report.get("inference", "influence_curve")
point = (
f"point robustness value unavailable: {_UNREACHABLE_RV}"
if report["rv"] is None
else f"point robustness value {report['rv']:.4g}"
)
if "limit_refusal" in report:
return AssessmentItem(
"robustness_value",
AssessmentStatus.COMPLETED,
f"{point}; no confidence-limit value: {report['limit_refusal']}",
)
if "nu2_estimator" in report:
# A report names its estimator only when no confidence-limit value exists under any
# name, so the row gives the bound's own reason rather than reading an absent key.
return AssessmentItem(
"robustness_value",
AssessmentStatus.COMPLETED,
f"{point}; no confidence-limit value under "
f"{limit_refusal_reason(report['nu2_estimator'])}",
)
limit_value = report[spread_name("rva", status)]
limit_detail = (
f"unavailable: {_UNREACHABLE_RV}" if limit_value is None else f"{limit_value:.4g}"
)
return AssessmentItem(
"robustness_value",
AssessmentStatus.COMPLETED,
f"{point}; {spread_name('confidence-limit value', status)} {limit_detail}",
)
def _elements_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
return AssessmentItem(
"elements",
AssessmentStatus.COMPLETED,
f"sigma2={report.sigma2:.4g}, nu2={report.nu2:.4g}, max_bias={report.max_bias:.4g}",
)
def _contour_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
payload = _frame_payload(report)
return AssessmentItem(
"contour",
AssessmentStatus.COMPLETED,
f"grid {len(set(payload['cf_d']))} x {len(set(payload['cf_y']))}; "
f"cf_d range {_format_range(_range(payload['cf_d']))}; "
f"cf_y range {_format_range(_range(payload['cf_y']))}; "
f"value range {_format_range(_range(payload['value']))}; inspect the retained frame",
)
def _benchmark_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
return AssessmentItem(
"benchmark",
AssessmentStatus.COMPLETED,
f"covariates={report.covariates}; cf_y={report.cf_y:.3g}, cf_d={report.cf_d:.3g}, "
f"rho={report.rho:.3g}, delta_psi={report.delta_psi:.4g}",
)
def _conditioning_population(report: Any) -> tuple[float | None, float | None, bool]:
"""The conditioning arm's smallest and unperturbed share, and whether it collapsed.
A surface whose target population is ``"perturbed_treatment_group"`` rebuilds its ATT
or ATC population in every cell, so a cell can move because its population changed and
not because a confounding path opened. A one-line report that gives only the movement
hides that third channel.
**The collapse rule.** The anchor cell perturbs nothing, so its
``target_population_fraction`` is the unperturbed share of the conditioning arm. A
surface has collapsed when its smallest cell keeps **less than half** of that anchor
share. The rule is relative to the fit's own anchor rather than an absolute cut, so a
study whose treated arm is a tenth of the sample is not warned about for that alone;
the constant is the half, which names "most of the group is gone" and is not tuned to
any dataset. A surface that averages over its baseline population never collapses,
because every cell reports a fraction of one.
**Every cell, not every successful cell.** The cell that collapses hardest is the one
whose conditioning arm emptied, and that cell fails: its refit raises
:exc:`~cleverly.exceptions.DataError` because the arm keeps no positive-weight row.
``simulated_confounding`` records ``target_population_fraction`` from the perturbed
treatment *before* it refits, so a failed cell still carries the fraction that explains
the failure. A minimum over ``successful_cells`` therefore drops exactly the collapse
this rule exists to name, and where only the anchor survives it reports the anchor
against itself. The minimum below runs over every cell that recorded a fraction. A
cell that failed before the surface built its treatment records ``None`` and is
skipped, because it measured no population.
Parameters
----------
report : Any
A :class:`~cleverly.sensitivity.SimulatedConfoundingResult`.
Returns
-------
tuple of (float or None, float or None, bool)
The smallest recorded cell fraction, the anchor fraction, and whether the
conditioning population collapsed.
"""
fractions = [
cell.target_population_fraction
for cell in report.cells
if cell.target_population_fraction is not None
]
from .sensitivity.simulated_confounding import _is_anchor
anchor = next(
(
cell.target_population_fraction
for cell in report.cells
if _is_anchor(cell.treatment_strength, cell.outcome_strength)
),
None,
)
minimum = min(fractions, default=None)
collapsed = (
report.population == "perturbed_treatment_group"
and minimum is not None
and anchor is not None
and minimum < 0.5 * anchor
)
return minimum, anchor, collapsed
def _simulated_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
movements = [
abs(float(cell.displacement))
for cell in report.successful_cells
if cell.displacement is not None
]
corner = report.cells[-1].induced_treatment_association if report.cells else None
minimum, anchor, collapsed = _conditioning_population(report)
detail = (
f"maximum successful displacement {max(movements, default=float('nan')):.4g}; "
f"movement scale {report.movement_scale}; "
f"failed cells {len(report.failures)}; corner association {corner}; "
+ "; ".join(report.population_lines())
+ f"; minimum target population fraction {_format_fraction(minimum)}"
+ f" against anchor {_format_fraction(anchor)}"
)
advice = []
if report.failures:
advice.append("inspect the retained cell failures")
if collapsed:
advice.append(
"read target_population_fraction beside the movement; the conditioning group "
"keeps under half its unperturbed share, so part of the movement is a change "
"of population"
)
return AssessmentItem(
"simulated_confounding",
AssessmentStatus.WARNING if report.failures or collapsed else AssessmentStatus.COMPLETED,
detail,
tuple(advice),
)
def _format_fraction(value: float | None) -> str:
"""Render one population fraction, or name its absence.
Parameters
----------
value : float or None
A conditioning-arm share, or ``None`` when no cell recorded one.
Returns
-------
str
The share to four significant figures, or ``"n/a"``.
"""
return "n/a" if value is None else f"{value:.4g}"
def _evalue_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
detail = (
f"point={report.point:.4g}, limit={report.limit:.4g}, source scale={report.scale}; "
+ ("approximate conversion" if report.approximate else "exact risk-ratio branch")
)
if report.limit == 1.0:
detail += "; the interval already includes the null"
truncated = getattr(report, "truncated_bound", None)
if truncated is not None:
# The reported lower bound is the boundary of the parameter space, not a converted
# confidence limit. A row that shows only ``point`` and ``limit`` would hide that.
detail += f"; the lower risk-ratio bound is truncated at 0 from {float(truncated):.4g}"
return AssessmentItem("evalue", AssessmentStatus.COMPLETED, detail)
def _missingness_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
payload = _frame_payload(report)
psi = payload["psi"]
facts = [f"gamma range {_format_range(_range(payload['gamma']))}"]
aliases = payload.get("estimand")
if not aliases:
facts.append(f"estimate range {_format_range(_range(psi))}")
else:
# The frame holds one row for each ``(gamma, estimand)`` pair, so a pooled minimum
# and maximum can belong to two different parameters and describe neither. The
# ``gamma == 0`` row is the estimate the fit itself reported, which makes it the
# baseline the tilted rows move away from.
groups = _alias_rows(aliases)
markers = payload.get("is_mar")
# ``missingness_tilt`` drops an untiltable estimand from the default sweep, so the
# count is what tells a reader that the curve covers fewer parameters than the fit.
facts.append(f"{len(groups)} parameter(s)")
if markers is None:
# A missing column says nothing about which row the fit itself reported, so
# the row states the levels it can read and claims no baseline. Inferring an
# absent baseline here would deny a retained ``gamma == 0`` row.
levels = {
alias: _range([psi[index] for index in rows]) for alias, rows in groups.items()
}
facts.append(f"estimate range: {_format_alias_ranges(levels)}")
else:
movement: dict[str, tuple[float, float] | None] = {}
unavailable: dict[str, tuple[float, float] | None] = {}
for alias, rows in groups.items():
# A marked row whose estimate is not finite is no more usable as a
# baseline than an absent one, and every delta taken from it is not
# finite. Both cases report the level instead.
baseline = next(
(index for index in rows if bool(markers[index]) and _finite(psi[index])),
None,
)
if baseline is None:
# The default grid contains zero, but an explicit ``gamma=`` grid need
# not. The row then reports the level it has and names what is
# missing, rather than treating one tilted estimate as the estimate the
# fit made.
unavailable[alias] = _range([psi[index] for index in rows])
else:
mar = float(psi[baseline])
movement[alias] = _range([psi[index] - mar for index in rows])
if movement:
ranges = _format_alias_ranges(movement)
facts.append(f"signed movement from the MAR estimate: {ranges}")
if unavailable:
ranges = _format_alias_ranges(unavailable)
facts.append(f"no MAR estimate retained; estimate range: {ranges}")
return AssessmentItem("missingness", AssessmentStatus.COMPLETED, "; ".join(facts))
def _tipping_item(
report: Any, _result: Any, _arguments: Mapping[str, Any] = _NO_ARGUMENTS
) -> AssessmentItem:
detail = (
"no tipping value occurred in the searched interval"
if report is None
else f"tipping gamma {float(report):.4g}"
)
return AssessmentItem("tipping_gamma", AssessmentStatus.COMPLETED, detail)
#: How each operation's own report becomes one report row.
#:
#: The third argument is what the caller supplied for that operation. An interpreter that
#: has to say how its answer was parameterized reads it here, so that the combined report
#: keeps no branch on an operation name.
#:
#: Internal routing rather than a caller surface: the values take report objects this module
#: builds, so it is deliberately absent from ``__all__``.
INTERPRETERS: dict[str, Callable[[Any, Any, Mapping[str, Any]], AssessmentItem]] = {
"score_equations": _score_item,
"support": _support_item,
"nuisance_models": _nuisance_item,
"corrections": _correction_item,
"truncation_curve": _truncation_item,
"refute": _refute_item,
"omitted_confounding": _omitted_item,
"robustness_value": _robustness_item,
"elements": _elements_item,
"contour": _contour_item,
"benchmark": _benchmark_item,
"simulated_confounding": _simulated_item,
"evalue": _evalue_item,
"missingness": _missingness_item,
"tipping_gamma": _tipping_item,
}
#: What the combined row says when the point support report is not in its ``adequate``
#: tier. The tier itself is decided by
#: :attr:`~cleverly.sensitivity.positivity.PositivityReport.severity`; these are only its
#: words on this surface, and the numbers behind them are already in the detail that
#: ``_support_facts`` builds.
#:
#: Both tiers describe truncation, because truncation is what that report grades. The
#: effective-sample-size ratio reaches this row as a *fact* through ``_support_facts`` and
#: never as a status, which is why an unbreached row is ``COMPLETED`` rather than ``PASSED``.
_POSITIVITY_SEVERITY_REASON: dict[Literal["serious", "strain"], str] = {
"serious": "the support report reports that truncation is carrying the estimate",
"strain": "the support report reports material truncation",
}
def _support_warning(report: Any) -> str | None:
from .sensitivity.positivity import PositivityReport
if isinstance(report, PositivityReport):
# Delegated rather than re-derived. This branch used to threshold the truncated
# fraction and the *fitted mechanism* effective sample sizes, and `mechanisms` is
# empty unless the fit declared `delta=` or `intermediate=`. So on an ordinary ATE
# fit it applied one check, and reached a status the retained report contradicted.
severity = report.severity
return None if severity == "adequate" else _POSITIVITY_SEVERITY_REASON[severity]
# Every branch below grades truncation and estimated zero support, and none grades an
# effective-sample-size ratio. A Kish ratio has no derived cutoff, so a status keyed
# on one would present a house convention as a finding and let a reader take `passed`
# for a positivity clearance. `_support_facts` reports the minimum ratio on every row
# instead, which is the number an analyst has to judge.
if isinstance(report, LongitudinalDiagnostics) and any(
row.share_truncated > 0.05 for row in report.rows
):
return "cumulative mechanism truncation exceeds 5% at one or more nodes"
if isinstance(report, Mapping):
# A shift or IPSI fit reports one support record per declared intervention rather
# than a single object, so none of the attribute probes above sees it. These
# dataclasses carry the same quantities the other branches grade; read them by
# field so the two classes need no separate branches -- ``IncrementalSupport`` has
# an ``ess_ratio`` but no ``unsupported`` or ``capped_fraction``, and a missing
# field must not read as a breach. The ``ess_ratio`` is reported, never graded.
for name, item in report.items():
unsupported = getattr(item, "unsupported", None)
if unsupported is not None and int(unsupported) > 0:
return f"intervention {name!r} has units with estimated zero support"
capped = getattr(item, "capped_fraction", None)
if capped is not None and float(capped) > 0.05:
return f"intervention {name!r} had more than 5% of its weights capped"
return None
regimes = getattr(report, "regimes", None)
if regimes and any(item.unsupported for item in regimes.values()):
return "one or more declared regimes has estimated zero support"
return None
def validate_result(result: Any, diagnostics: DiagnosticsFacade | None = None) -> ValidationReport:
"""Run only cheap, cache-only checks appropriate to this fitted method."""
def compute() -> ValidationReport:
facade = result.diagnostics if diagnostics is None else diagnostics
items = []
for name in VALIDATION_OPERATIONS:
capability = facade.capability(name)
if not capability.available:
items.append(_item_from_capability(capability))
continue
report = getattr(facade, name)()
items.append(
replace(
INTERPRETERS[name](report, result, _NO_ARGUMENTS),
_report=_pack_cached(report, result.data.backend),
)
)
return ValidationReport(tuple(items), result.data.backend)
return _cached(result, "validate", (), {}, compute)
def assess_result(
result: Any,
*,
include_refits: bool = False,
include_retargets: bool = False,
arguments: Mapping[str, Mapping[str, Any]] | None = None,
random_state: int | None = None,
) -> AssessmentReport:
"""Compose a battery from one pair of operation facades."""
_require_argument_mapping(arguments)
supplied = {} if arguments is None else dict(arguments)
diagnostics, sensitivity = result.diagnostics, result.sensitivity
diagnostic_names = set(diagnostics._capability_map)
sensitivity_names = set(sensitivity._capability_map)
unknown = sorted(set(supplied) - diagnostic_names - sensitivity_names)
if unknown:
raise KeyError(f"unknown assessment operation(s): {unknown}")
# Only an operation that would really hide an answer is refused. An empty mapping
# applies nothing, so nothing can be hidden, and an operation with no parameters can
# take no argument at all. For those two, ``_validated_arguments`` below reports the
# precise ``TypeError`` from binding the caller's own keywords, which names the
# signature rather than substituting a vaguer refusal.
owned = [
name
for name in VALIDATION_OPERATIONS
if supplied.get(name) and _accepts_arguments(diagnostics, name)
]
if owned:
# The battery presents the validation row for these names and hides the
# diagnostics row of the same name. Running the caller's arguments would answer
# the caller's question on the hidden row and then show the argument-free answer,
# so a check that failed at the requested tolerance would never reach
# ``attention``. Refuse the composition and name the call that does answer it.
calls = ", ".join(f"result.diagnostics.{name}(...)" for name in owned)
raise CapabilityError(
f"the validation battery owns {owned} and runs each one argument-free, so "
"assess() cannot apply arguments to them.",
reason=RefusalReason.API_REDIRECTION,
alternative=(
f"Call {calls} for one answer, or "
"result.diagnostics.run_all(arguments=...) for the diagnostics surface."
),
)
# Validate both surfaces before either one runs, and carry the cleared arguments
# forward. A block that is wrong for the sensitivity facade must not first make the
# diagnostics facade refit anything.
diagnostic_arguments = diagnostics._validated_arguments(
{k: v for k, v in supplied.items() if k in diagnostic_names}, random_state
)
sensitivity_arguments = sensitivity._validated_arguments(
{k: v for k, v in supplied.items() if k in sensitivity_names}, random_state
)
return AssessmentReport(
validation=validate_result(result, diagnostics),
diagnostics=diagnostics._run_all(
include_refits=include_refits,
include_retargets=include_retargets,
supplied=diagnostic_arguments,
random_state=random_state,
),
sensitivity=sensitivity._run_all(
include_refits=include_refits,
include_retargets=include_retargets,
supplied=sensitivity_arguments,
random_state=random_state,
),
)
def _reduction_conditioning_warning(result: Any) -> str | None:
"""Summarize successful equation-(10) solves that needed an ill-conditioned step."""
ill_conditioned = 0
rounds = 0
affected = 0
repeats = tuple(getattr(result, "repeats", ()))
for repeat in repeats:
fluctuation = getattr(repeat, "fluctuations", {}).get("mean")
reduction = getattr(fluctuation, "reduction", None)
count = int(getattr(reduction, "ill_conditioned", 0))
rounds += int(getattr(reduction, "rounds", 0))
if count <= 0:
continue
ill_conditioned += count
affected += 1
if ill_conditioned == 0:
return None
fraction = ill_conditioned / rounds if rounds else float("nan")
draws = f" across {affected} of {len(repeats)} repeat(s)" if len(repeats) > 1 else ""
return (
f"all stored score checks converged, but equation (10) reported numerical difficulty in "
f"{ill_conditioned} of {rounds} refitting round(s) ({fraction:.1%}){draws}; "
"one or more inner solves were ill-conditioned or stopped at their numerical "
"tolerance even though the returned score equations passed"
)
[docs]
class SensitivityFacade(_CapabilityFacade):
"""Access sensitivity analyses supported by a fitted causal result.
Parameters
----------
result : TMLEResult or LongitudinalResult
Fitted point-treatment or longitudinal result.
See Also
--------
DiagnosticsFacade : The checks that ask whether the fit itself is sound.
cleverly.AssessmentCapability : One row of :attr:`capabilities`.
cleverly.sensitivity.evalue.evalue : The same analysis as a free function.
Notes
-----
Access this facade through ``result.sensitivity``. Methods use the fitted
estimate and nuisance artifacts as their first input. Their remaining
arguments match the corresponding functions in :mod:`cleverly.sensitivity`.
Inspect :attr:`capabilities` to distinguish an unsupported analysis from one
that needs additional arguments or expensive work.
Examples
--------
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from cleverly import ATE, CausalStudy, PointTreatment
>>> from cleverly.datasets import make_linear_ate
>>> frame, _ = make_linear_ate(n=200, seed=0)
>>> study = CausalStudy(
... frame,
... design=PointTreatment(
... outcome="Y", treatment="A", adjustment=("W1", "W2", "W3", "W4")
... ),
... )
>>> result = study.identify(ATE()).estimate(
... outcome_learner=LinearRegression(),
... treatment_learner=LogisticRegression(max_iter=1000),
... cross_fit=False,
... random_state=0,
... )
>>> capability = result.sensitivity.capability("omitted_confounding")
>>> capability.available, capability.cost
(True, 'cheap')
"""
_kind = "sensitivity"
_attribute = "sensitivity"
#: This facade fills in the sole eligible estimand, through
#: :meth:`_with_default_parameter`, so one candidate settles the choice and only two
#: defer the row.
_substitutes_estimand = True
def _tilt_rule(self) -> tuple[str, Refusal] | None:
"""The first fit-wide tilt rule that refuses this result, with its reason.
One seam over :func:`~cleverly.sensitivity.missingness._fit_wide_tilt_rule`, which
``missingness_tilt`` and ``tipping_gamma`` raise from. Both tilt rows read it, so
a test that replaces it proves that the rows follow the predicate rather than a
flag of their own.
"""
from .sensitivity.missingness import _fit_wide_tilt_rule
return _fit_wide_tilt_rule(self._result)
@cached_property
def _declared(self) -> tuple[AssessmentCapability, ...]:
family = _family(self._result)
longitudinal = family == "longitudinal"
# The two tilt rows read the table both tilt entry points raise from. Its
# ``missing_outcome`` rule reads the data flag the bound's response rule reads, so
# every fit offered the tilt is a fit the bound refuses for its response mechanism.
tilt_rule = self._tilt_rule()
# ``simulated_confounding`` refuses the bare ``ate`` default on a continuous fit.
# A binary arm, fixed-regime, incremental, or MSM fit can use the facade's sole-
# parameter substitution. Several eligible aliases require an explicit choice.
# ``LongitudinalData`` declares neither treatment-family flag, so the family guards
# both reads. Neither one needs a ``getattr`` default: a default would answer for a
# point result whose data lack the flag, and no such result exists, because both are
# properties of ``CausalData``.
# A vector treatment with a continuous component has no arms either, so it takes
# the continuous branch: every eligible parameter is a named policy parameter.
continuous = not longitudinal and not self._result.data.has_arms
from .sensitivity._simulated_confounding_request import (
_eligible_binary_parameter_names,
_fit_wide_refusal,
)
from .sensitivity.omitted_variable import fit_wide_bound_refusal
simulated_refusal = _fit_wide_refusal(self._result)
if longitudinal or continuous:
needs_estimand = False
elif not self._result.data.is_binary_treatment:
# A multi-arm fit reports no bare ``ate`` that this surface could assess, so the
# truthful argument list names an estimand. The row is unavailable, and its
# declared arguments still describe the call a caller would have to write.
needs_estimand = True
else:
binary_parameters = _eligible_binary_parameter_names(self._result)
needs_estimand = "ate" not in binary_parameters and len(binary_parameters) > 1
# The four omitted-variable rows carry the refusal the bound itself would raise,
# declared rather than executed. ``run_all`` publishes its own wrapper sentence
# around a raised refusal, and it pays ``contour``'s moderate cost before hearing
# one. The first rule of the table returns the longitudinal literal, so that
# reason stays byte-identical to the one this facade published before.
bound_refusal = fit_wide_bound_refusal(self._result)
available = bound_refusal is None
status = AssessmentStatus.PASSED if available else AssessmentStatus.UNAVAILABLE
declared_refusal = bound_refusal
# A benchmark refits, so its longitudinal stop names the derivation it would have
# needed rather than the bound's. Every other fit-wide boundary is the bound's
# own: a benchmark compares two sets of the same elements, so a fit with no nu^2
# has no calibration either. Whether a *point* fit is replayable is settled by
# ``requires_replay`` below.
benchmark_refusal_row = (
Refusal(
RefusalReason.IMPLEMENTATION_GAP,
"no longitudinal benchmarking derivation is registered",
alternative=NO_ALTERNATIVE,
# F16 part (c) holds the omitted-variable bound on a longitudinal fit,
# which a benchmark calibrates.
roadmap="F16",
)
if longitudinal
else declared_refusal
)
def standard(
operation: str,
*,
artifacts: Sequence[str],
interpretation: str,
cost: Literal["cheap", "moderate", "expensive"] = "cheap",
) -> AssessmentCapability:
return _capability(
operation,
family,
artifacts=artifacts,
interpretation=interpretation,
cost=cost,
available=available,
status=status,
refusal=declared_refusal,
)
def tilt(operation: str, *, interpretation: str) -> AssessmentCapability:
"""The two MNAR analyses, which share every field but their interpretation.
Three cases, not two. A fit the tilt table admits can run the tilt. A fit
with no missing outcome is answering a question about a functional with no
observation mechanism in it, which is ``not_applicable``. Every other rule
names a derivation the package does not have, which is ``unavailable``. The
reason is the sentence the call raises, so the row cannot advertise a fit the
call then refuses.
"""
if tilt_rule is None:
status = AssessmentStatus.PASSED
skipped = natural_course_names(self._result.estimates)
if skipped and reads_natural_course_mean(self._result):
interpretation = (
f"{interpretation}; the default sweep skips {skipped}, which read "
"the natural-course mean and have no tilt"
)
elif tilt_rule[0] == "missing_outcome":
status = AssessmentStatus.NOT_APPLICABLE
else:
status = AssessmentStatus.UNAVAILABLE
return _capability(
operation,
family,
artifacts=("observation mechanism", "published tilt identification"),
execution="retarget",
cost="moderate",
interpretation=interpretation,
available=tilt_rule is None,
status=status,
refusal=None if tilt_rule is None else tilt_rule[1],
)
return (
standard(
"omitted_confounding",
artifacts=("fitted representer", "outcome residuals"),
interpretation="omitted-confounder bias for the fitted orthogonal score",
),
standard(
"robustness_value",
artifacts=("fitted representer", "outcome residuals"),
interpretation="confounding strength needed to move the estimate to its null",
),
standard(
"elements",
artifacts=("fitted representer", "outcome residuals"),
interpretation="raw components of the omitted-confounder bias bound",
),
_capability(
"benchmark",
artifacts=("fitted estimator configuration",),
execution="refit",
deterministic=False,
cost="expensive",
interpretation="calibration against named observed covariates",
available=available,
status=status,
refusal=benchmark_refusal_row,
requires_arguments=("covariates",),
accepts_random_state=True,
requires_replay="refit_nuisances",
family=family,
),
_capability(
"simulated_confounding",
artifacts=("fitted estimator configuration", "analysis data"),
execution="refit",
deterministic=False,
cost="expensive",
interpretation="estimate movement under a simulated common cause",
available=simulated_refusal is None,
status=(
AssessmentStatus.PASSED
if simulated_refusal is None
else AssessmentStatus.UNAVAILABLE
),
refusal=simulated_refusal,
requires_arguments=("grid", "estimand")
if continuous or needs_estimand
else ("grid",),
accepts_random_state=True,
requires_replay="refit_nuisances",
family=family,
),
standard(
"contour",
artifacts=("fitted representer", "outcome residuals"),
interpretation="bias bounds over a grid of confounding strengths",
cost="moderate",
),
self._evalue_row(),
tilt(
"missingness",
interpretation="departure from missing-at-random identification",
),
tilt(
"tipping_gamma",
interpretation="missingness departure at which the conclusion reaches its null",
),
)
@cached_property
def _evalue_selections(self) -> dict[str | None, Any]:
return {}
def _evalue_selection(self, estimand: str | None) -> Any:
from .sensitivity.evalue import _EValueRefusal, _select_evalue
if estimand not in self._evalue_selections:
try:
selected: Any = _select_evalue(self._result, estimand)
except _EValueRefusal as error:
selected = (error.status, error.refusal)
self._evalue_selections[estimand] = selected
selected = self._evalue_selections[estimand]
if isinstance(selected, tuple):
status, cached = selected
raise _EValueRefusal(
status,
cached.condition,
reason=cached.reason,
alternative=cached.alternative,
roadmap=cached.roadmap,
)
return selected
def _evalue_row(self, estimand: str | None = None) -> AssessmentCapability:
from .sensitivity.evalue import _DERIVED_RR, _EValueRefusal
status: AssessmentStatus
refusal: Refusal | None
execution: Literal["summarize", "retarget"]
requires_arguments: tuple[str, ...]
try:
selected = self._evalue_selection(estimand)
except _EValueRefusal as error:
# A refused row is refused, whichever status it carries. Reporting
# ``available=True`` beside a deferral published ``available: True | status:
# passed | reason: an E-value needs one contrast`` on a multi-arm fit, and the
# bare call the row invited then raised. ``_dispatch`` skips ``_require`` for
# an explicit ``estimand``, so a caller who supplies one still runs.
available = False
status = error.status
refusal = error.refusal
execution = "summarize"
# The argument that lifts the deferral. Only a deferral has one: an
# unsupported contrast is not a choice the caller can make.
requires_arguments = ("estimand",) if status is AssessmentStatus.DEFERRED else ()
else:
available, status, refusal = True, AssessmentStatus.PASSED, None
execution = "retarget" if selected.branch == _DERIVED_RR else "summarize"
requires_arguments = ()
return _capability(
"evalue",
_family(self._result),
artifacts=("structured arm contrast", "ratio or derivation artifacts"),
interpretation="minimum risk-ratio association needed to explain away an effect",
available=available,
status=status,
refusal=refusal,
execution=execution,
cost="cheap",
requires_arguments=requires_arguments,
)
def _capability_for_arguments(
self, operation: str, arguments: Mapping[str, Any]
) -> AssessmentCapability:
if operation == "evalue":
# The E-value selects for itself from a ``None`` sentinel rather than through
# ``SENSITIVITY_ROUTES``, so its row is rebuilt for the requested estimand.
return self._evalue_row(arguments.get("estimand"))
return super()._capability_for_arguments(operation, arguments)
#: ``benchmark`` resolves ``covariates``, ``simulated_confounding`` resolves
#: ``estimand``, and ``tipping_gamma`` resolves ``use_ci``.
_request_gates: ClassVar[Mapping[str, str]] = {
"benchmark": "_benchmark_gated",
"simulated_confounding": "_simulated_confounding_gated",
"tipping_gamma": "_tipping_interval_gated",
}
def _benchmark_gated(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Resolve the ``benchmark`` row from the call's predicate.
:func:`~cleverly.sensitivity.omitted_variable.benchmark_refusal` is what the call
raises from, and :func:`_argument_resolved` applies the request rule to
``covariates``. A fit with one covariate reads ``unavailable``, because the refit
cannot drop the only covariate and no other value exists. On a fit with two or
more, the bare row keeps its declared ``covariates`` argument, and a request that
names every covariate reads ``unavailable``.
"""
from .sensitivity.omitted_variable import _benchmark_covariates, benchmark_refusal
return _argument_resolved(
capability,
"covariates",
arguments.get("covariates"),
lambda covariates: benchmark_refusal(self._result, covariates),
tuple((name,) for name in _benchmark_covariates(self._result.data)),
)
def _simulated_confounding_gated(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Resolve the ``simulated_confounding`` row from the call's predicate.
:func:`~cleverly.sensitivity._simulated_confounding_request.simulated_confounding_refusal`
is what the call raises from, and :func:`_argument_resolved` applies the request
rule to ``estimand``. The request's ``benchmark_covariates`` stay fixed, so a
request that names a categorical or a constant covariate reads ``unavailable``
with the call's sentence. So does a request for a natural-course mean or a
zero-delta policy mean. An omitted estimand is the one the call reports: the sole
eligible parameter this facade fills in, or the signature default ``"ate"``. The
bare row of a natural-course fit therefore reads ``unavailable``, because no
reported parameter runs.
"""
from .sensitivity._simulated_confounding_request import simulated_confounding_refusal
covariates = arguments.get("benchmark_covariates", ())
def declared(estimand: str | None) -> Refusal | None:
if estimand is None:
estimand = self._argument_default(capability.operation, "estimand")
return simulated_confounding_refusal(self._result, estimand, covariates)
return _argument_resolved(
capability,
"estimand",
arguments.get("estimand"),
declared,
tuple(self._result.estimates),
)
def _tipping_interval_gated(
self, capability: AssessmentCapability, arguments: Mapping[str, Any]
) -> AssessmentCapability:
"""Resolve the ``tipping_gamma`` row for a ``use_ci=True`` request.
The interval search reads a confidence limit, and a fit that supplies no inference
has none, so :func:`~cleverly.sensitivity.tipping_gamma` refuses it. The row says
so before the call, in the sentence the call raises. The default point search
stays available, which is why a request without ``use_ci`` reads the row unchanged.
The answer reads every estimate of the fit, so a fit with one estimate that supplies
no inference, such as a constant-node parameter, makes the row unavailable.
"""
if not arguments.get("use_ci"):
return capability
from .sensitivity.missingness import _TIPPING_INTERVAL_OPERATION
status = precedent_status(
estimate.inference for estimate in self._result.estimates.values()
)
if supplies_inference(status):
return capability
return replace(
capability,
available=False,
status=AssessmentStatus.UNAVAILABLE,
refusal=inference_refusal(_TIPPING_INTERVAL_OPERATION, status),
)
def _estimand_candidates(self, operation: str) -> tuple[str, ...]:
"""The reported parameters a routed sensitivity analysis may be asked to choose.
The gate is the target's signature: a route whose ``estimand`` defaults to the
bare ``"ate"`` faces the choice, and ``evalue``, which defaults to ``None``, and
``missingness``, which takes no estimand, do not. It is not
``needs_estimand``, which answers where the estimand goes rather than whether
there is one. ``benchmark`` takes ``covariates`` positionally and the same
ambiguous default by keyword, and reading the positional flag left its row alone.
``simulated_confounding`` answers for ratio and population attributable contrasts
too, so it consults its own eligible set first and falls back to the linear set
when that set names no single parameter. The rest are functionals of one linear
contrast, and their eligible set is the arm-indexed one.
Parameters
----------
operation : str
Declared sensitivity operation.
Returns
-------
tuple of str
Eligible reported names, empty when the route settles its own estimand.
"""
function, _ = self._routed_callable(operation)
if not _defaults_to_ambiguous_estimand(function):
return ()
if operation == "simulated_confounding":
from .sensitivity._simulated_confounding_request import (
_eligible_binary_parameter_names,
)
binary = _default_estimand_candidates(
self._result, _eligible_binary_parameter_names(self._result)
)
if len(binary) == 1:
return binary
from .sensitivity._parameters import arm_parameters
return _default_estimand_candidates(self._result, arm_parameters(self._result))
[docs]
def omitted_confounding(self, *args: Any, **kwargs: Any) -> Any:
"""Bound omitted-confounder bias for a reported estimand.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.omitted_variable_bounds`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
SensitivityBounds
Bias bounds for each requested estimand.
See Also
--------
cleverly.sensitivity.omitted_variable_bounds : The same bound as a free function.
Examples
--------
>>> from sklearn.linear_model import LinearRegression, LogisticRegression
>>> from cleverly import ATE, CausalStudy, PointTreatment
>>> from cleverly.datasets import make_linear_ate
>>> frame, _ = make_linear_ate(n=80, seed=1)
>>> study = CausalStudy(
... frame,
... design=PointTreatment(
... outcome="Y", treatment="A", adjustment=("W1", "W2", "W3", "W4")
... ),
... )
>>> result = study.identify(ATE()).estimate(
... outcome_learner=LinearRegression(),
... treatment_learner=LogisticRegression(max_iter=1000),
... cross_fit=False,
... random_state=0,
... )
>>> bounds = result.sensitivity.omitted_confounding(cf_y=0.05, cf_d=0.05)
>>> bounds.lower < result["ate"].psi < bounds.upper
True
"""
return self._dispatch("omitted_confounding", args, kwargs)
[docs]
def robustness_value(self, *args: Any, **kwargs: Any) -> Any:
"""Return the confounding strength needed to cross the null.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.robustness_value`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
dict of str to float
Null-crossing strengths for the estimate and confidence limit, plus the
maximum-bias scale.
See Also
--------
cleverly.sensitivity.robustness_value : The same value as a free function.
"""
return self._dispatch("robustness_value", args, kwargs)
[docs]
def elements(self, *args: Any, **kwargs: Any) -> Any:
"""Return the variance elements used by omitted-variable bounds.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.sensitivity_elements`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
SensitivityElements
The variance elements the omitted-variable bounds are built from.
See Also
--------
cleverly.sensitivity.sensitivity_elements : The same elements as a free function.
"""
return self._dispatch("elements", args, kwargs)
[docs]
def benchmark(self, *args: Any, **kwargs: Any) -> Any:
"""Benchmark confounding strength against observed covariates.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.benchmark`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
BenchmarkResult
Confounding strength expressed in units of the named observed covariates.
See Also
--------
cleverly.sensitivity.benchmark : The same benchmark as a free function.
Notes
-----
A name the fit does not adjust for raises ``DataError`` before any refusal, as the
free function raises it.
"""
# The malformed argument first, through the check the function runs first, so the
# facade and the function report an unknown name before any refusal. A
# longitudinal fit has no covariate list to check a name against, and its row
# refuses every request.
args, kwargs = _single_pass(args, kwargs)
if _family(self._result) != "longitudinal":
bound = self._bind_arguments("benchmark", kwargs, partial=True, args=args)
_validate_benchmark_arguments(self._result, bound)
return self._dispatch("benchmark", args, kwargs)
[docs]
def contour(self, *args: Any, **kwargs: Any) -> Any:
"""Evaluate bias bounds over a confounding-strength grid.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.contour_data`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
dataframe
Bias bounds over the requested grid of confounding strengths.
See Also
--------
cleverly.sensitivity.contour_data : The same grid as a free function.
"""
return self._dispatch("contour", args, kwargs)
[docs]
def simulated_confounding(self, *args: Any, **kwargs: Any) -> Any:
"""Refit across a simulated common-cause strength grid.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.simulated_confounding`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
SimulatedConfoundingResult
Qualitative estimate movements on the scale of the surface, and retained cell
failures.
See Also
--------
cleverly.sensitivity.simulated_confounding : The same surface as a free function.
"""
return self._dispatch("simulated_confounding", args, kwargs)
[docs]
def evalue(self, *args: Any, **kwargs: Any) -> Any:
"""Return an E-value on the risk-ratio scale.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.evalue`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
EValue
The minimum risk-ratio association that would explain the effect away.
See Also
--------
cleverly.sensitivity.evalue : The same E-value as a free function.
"""
return self._dispatch("evalue", args, kwargs)
[docs]
def missingness(self, *args: Any, **kwargs: Any) -> Any:
"""Vary unobserved-outcome odds under a missingness tilt.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.missingness_tilt`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
dataframe
One row per tilt value and estimand.
See Also
--------
cleverly.sensitivity.missingness_tilt : The same tilt as a free function.
"""
return self._dispatch("missingness", args, kwargs)
[docs]
def tipping_gamma(self, *args: Any, **kwargs: Any) -> Any:
"""Find the missingness tilt at which the estimate crosses its null.
Parameters
----------
*args, **kwargs
Forwarded to :func:`cleverly.sensitivity.tipping_gamma`, without its
first argument, which this facade supplies from the fitted result.
Returns
-------
float or None
The tilt at which the requested estimand reaches its null.
See Also
--------
cleverly.sensitivity.tipping_gamma : The same search as a free function.
"""
return self._dispatch("tipping_gamma", args, kwargs)
def _dispatch(self, operation: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any:
"""Refuse by the declared capability, then call the declared implementation."""
args, kwargs = _single_pass(args, kwargs)
explicit_evalue = operation == "evalue" and (bool(args) or "estimand" in kwargs)
if not explicit_evalue:
self._require(operation)
return self._invoke(operation, args, kwargs)
def _with_default_parameter(
self, operation: str, args: tuple[Any, ...], kwargs: dict[str, Any]
) -> tuple[tuple[Any, ...], dict[str, Any]]:
"""Supply the estimand only when the fit leaves no choice about which one.
These analyses default to ``"ate"``, which a multi-arm fit never reports under that
bare name -- it reports ``"ate[medium vs low]"``. Filling the gap is worth doing
when exactly one reported parameter is one the analysis applies to, and is a
scientific choice made on the caller's behalf as soon as there are two: picking the
first would answer about ``ey1`` on an ``ey1``/``ey0`` fit, silently returning a
statement about a counterfactual mean to someone who asked about an effect.
Which parameters are eligible is :meth:`_estimand_candidates`, and a combined
report reads the same method to decide that the row is deferred rather than
unavailable. Both answers have to come from one predicate: while they did not,
this method declined to guess between two contrasts and the row beside it still
said the analysis was runnable.
A direct call still hears the analysis refuse for itself when the choice stays
ambiguous, and that refusal names every parameter it could have answered for.
:func:`~cleverly.sensitivity.omitted_variable.resolve_parameter` and
:func:`~cleverly.sensitivity.missingness.missingness_tilt` write those lists.
:func:`~cleverly.sensitivity.simulated_confounding.simulated_confounding` refuses on
its own ``"ate"`` default instead. Its binary selection message lists only the
arm, fixed-regime, and identity-MSM aliases the stored estimator can replay.
Every alias that estimator's own
boundary refuses is dropped, not natural-course means alone. Its continuous
selection message omits a zero-delta policy mean, which is the natural course and
which the surface cannot assess.
Parameters
----------
operation : str
Routed sensitivity operation, which selects the eligible parameter set.
args : tuple
Positional arguments the caller gave the facade method.
kwargs : dict
Keyword arguments the caller gave the facade method.
Returns
-------
tuple
The positional and keyword arguments, one of them naming the estimand when
this fit leaves exactly one eligible parameter.
"""
positional = SENSITIVITY_ROUTES[operation].needs_estimand
# Where the name goes is ``needs_estimand``; whether there is one to supply is the
# signature, which :meth:`_estimand_candidates` reads. ``benchmark`` takes its
# estimand by keyword, so a positional substitution would pass a parameter name
# as ``covariates``, and a positional argument the caller wrote there is not an
# estimand and does not settle the choice.
if "estimand" in kwargs or (positional and args):
return args, kwargs
candidates = self._estimand_candidates(operation)
if len(candidates) != 1:
return args, kwargs
natural = _tipping_natural_course_refusal(self._result, operation, candidates)
if natural is not None:
# The one arm candidate is not what a fit that reads the natural-course mean
# was asked about, so the facade names the gap instead of substituting it.
raise CapabilityError.from_refusal(natural)
if positional:
return (candidates[0], *args), kwargs
return args, {**kwargs, "estimand": candidates[0]}
[docs]
def run_all(
self,
*,
include_refits: bool = False,
include_retargets: bool = False,
arguments: Mapping[str, Mapping[str, Any]] | None = None,
random_state: int | None = None,
) -> DiagnosticReport:
"""Run available sensitivity analyses that need no new arguments.
Parameters
----------
include_refits : bool
Include operations that refit nuisance models.
include_retargets : bool
Include moderate retargets; cheap E-value retargets run by default.
arguments : mapping or None
Per-operation keyword arguments.
random_state : int or None
Common seed for stochastic refit operations.
Returns
-------
DiagnosticReport
One item for every declared sensitivity operation.
"""
return self._run_all(
include_refits=include_refits,
include_retargets=include_retargets,
supplied=self._validated_arguments(arguments, random_state),
random_state=random_state,
)