"""

BusinessReport — plain-English stakeholder narrative from any diff-diff result.

Wraps any of the 16 fitted result types and produces:

- ``summary()``: a short paragraph block suitable for an email or Slack message.

- ``full_report()``: a multi-section markdown report with headline, assumptions,

pre-trends, main result, robustness, sample, and an optional academic appendix.

- ``to_dict()``: a stable AI-legible structured schema (single source of truth —

prose is rendered from this dict, not templated alongside it).

Design principles:

- Plain English, not academic jargon. The library ships this in addition to, not

in place of, the estimator's existing ``results.summary()`` academic output.

- No estimator fitting and no variance re-derivation. Every effect, SE, p-value,

CI, and sensitivity bound is either read from ``results`` or produced by an

existing diff-diff utility. The report layer does compose a few cross-period

summaries from per-period inputs already on the result (joint-Wald / Bonferroni

pre-trends p-value, MDV-to-ATT ratio, heterogeneity dispersion over

post-treatment effects); see ``docs/methodology/REPORTING.md`` for the full

enumeration.

- Optional business context via keyword args (``outcome_label``, ``outcome_unit``,

``business_question``, ``treatment_label``). Without them, BusinessReport uses

generic fallbacks — the zero-config path works.

- Diagnostic integration is implicit by default: ``BusinessReport(results)``

auto-constructs a ``DiagnosticReport`` so the summary can mention pre-trends,

robustness, and design-effect findings. Pass ``auto_diagnostics=False`` or an

explicit ``diagnostics=`` object to override.

Methodology deviations (no traffic-light gates, pre-trends verdict thresholds,

power-aware phrasing, unit-translation policy, schema stability) are documented

in ``docs/methodology/REPORTING.md``. The ``to_dict()`` schema is marked

experimental in v3.2.

"""

from __future__ import annotations

import re

from dataclasses import dataclass

from typing import Any, Dict, FrozenSet, List, Optional, Union

import numpy as np

from diff_diff._reporting_helpers import describe_target_parameter

from diff_diff.diagnostic_report import DiagnosticReport, DiagnosticReportResults

BUSINESS_REPORT_SCHEMA_VERSION = "2.0"

__all__ = [

"BusinessReport",

"BusinessContext",

"BUSINESS_REPORT_SCHEMA_VERSION",

]

# Recognized ``outcome_unit`` values mapped to a coarse "kind" used by the

# formatter. Unrecognized strings are accepted and rendered verbatim without

# arithmetic translation (``unit_kind = "unknown"``).

_UNIT_KINDS: Dict[str, str] = {

"$": "currency",

"usd": "currency",

"%": "percent",

"pp": "percentage_points",

"percentage_points": "percentage_points",

"percent": "percent",

"log_points": "log_points",

"log": "log_points",

"count": "count",

"users": "count",

}

@dataclass(frozen=True)

class BusinessContext:

"""Frozen bundle of business-framing metadata used when rendering prose.

Populated from ``BusinessReport`` constructor kwargs. Falls back to

neutral labels when fields are not supplied.

"""

outcome_label: str

outcome_unit: Optional[str]

outcome_direction: Optional[str]

business_question: Optional[str]

treatment_label: str

alpha: float

class BusinessReport:

"""Produce a stakeholder-ready narrative from any diff-diff results object.

Parameters

----------

results : Any

A fitted diff-diff results object. Any of the 16 result types is

accepted. ``BaconDecompositionResults`` is not a valid input — Bacon

is a diagnostic, not an estimator; use ``DiagnosticReport`` for that.

outcome_label : str, optional

Stakeholder-friendly outcome name (e.g. ``"Revenue per user"``).

outcome_unit : str, optional

Unit label: ``"$"`` / ``"%"`` / ``"pp"`` / ``"log_points"`` / ``"count"``

(recognized for formatting) or any free-form string (used verbatim

without arithmetic translation).

outcome_direction : str, optional

``"higher_is_better"`` or ``"lower_is_better"``. Drives whether the

effect is described as "lift" / "drag" rather than just "increase" /

"decrease".

business_question : str, optional

Question the analysis answers (prepended to the summary).

treatment_label : str, optional

Stakeholder-friendly treatment name (e.g. ``"the campaign"``).

alpha : float, optional

Significance level. Defaults to ``results.alpha`` when not supplied.

Single knob: drives both CI level and significance phrasing.

honest_did_results : HonestDiDResults or SensitivityResults, optional

Pre-computed sensitivity result. When supplied, this is forwarded to

the internal ``DiagnosticReport`` so sensitivity is not re-computed.

auto_diagnostics : bool, default True

When ``True`` and ``diagnostics`` is ``None``, auto-construct a

``DiagnosticReport``. Set ``False`` to skip diagnostics entirely.

diagnostics : DiagnosticReport or DiagnosticReportResults, optional

Explicit diagnostics object. Takes precedence over ``auto_diagnostics``.

include_appendix : bool, default True

Whether ``full_report()`` appends the estimator's academic

``results.summary()`` output under a "Technical Appendix" section.

data, outcome, treatment, unit, time, first_treat : optional

Raw panel + column names forwarded to the auto-constructed

``DiagnosticReport`` so data-dependent checks (2x2 PT on simple

DiD, Bacon-from-scratch, EfficientDiD Hausman pretest) can run.

survey_design : SurveyDesign, optional

The ``SurveyDesign`` object used to fit a survey-weighted

estimator. Forwarded to the auto-constructed ``DiagnosticReport``

for fit-faithful Goodman-Bacon replay. When the fit carries

``survey_metadata`` but ``survey_design`` is not supplied, Bacon

is skipped with an explicit reason rather than replaying an

unweighted decomposition for a design that does not match the

estimate. The simple 2x2 parallel-trends helper

(``utils.check_parallel_trends``) has no survey-aware variant;

on a survey-backed ``DiDResults`` it is skipped unconditionally

regardless of ``survey_design``. Supply

``precomputed={'parallel_trends': ...}`` with a survey-aware

pretest to opt in. See ``docs/methodology/REPORTING.md``.

precomputed : dict, optional

Pre-computed diagnostic objects forwarded to the auto-

constructed ``DiagnosticReport`` (same keys as

``DiagnosticReport(precomputed=...)``): ``"parallel_trends"``,

``"sensitivity"``, ``"pretrends_power"``, ``"bacon"``. DR

validates keys and rejects estimator-incompatible entries

(e.g., HonestDiD bounds or generic PT on SDiD / TROP).

``honest_did_results`` remains a shorthand for ``sensitivity``;

an explicit ``precomputed['sensitivity']`` wins on conflict.

"""

def __init__(

self,

results: Any,

*,

outcome_label: Optional[str] = None,

outcome_unit: Optional[str] = None,

outcome_direction: Optional[str] = None,

business_question: Optional[str] = None,

treatment_label: Optional[str] = None,

alpha: Optional[float] = None,

honest_did_results: Optional[Any] = None,

auto_diagnostics: bool = True,

diagnostics: Optional[Union[DiagnosticReport, DiagnosticReportResults]] = None,

include_appendix: bool = True,

data: Optional[Any] = None,

outcome: Optional[str] = None,

treatment: Optional[str] = None,

unit: Optional[str] = None,

time: Optional[str] = None,

first_treat: Optional[str] = None,

survey_design: Optional[Any] = None,

precomputed: Optional[Dict[str, Any]] = None,

):

if type(results).__name__ == "BaconDecompositionResults":

raise TypeError(

"BaconDecompositionResults is a diagnostic, not an estimator; "

"wrap the underlying estimator with BusinessReport and pass the "

"Bacon object to DiagnosticReport(precomputed={'bacon': ...})."

)

if diagnostics is not None and not isinstance(

diagnostics, (DiagnosticReport, DiagnosticReportResults)

):

raise TypeError(

"diagnostics= must be a DiagnosticReport or "

"DiagnosticReportResults instance; "

f"got {type(diagnostics).__name__}."

)

# Estimator-aware validation for ``honest_did_results``. SDiD /

# TROP route robustness to ``estimator_native_diagnostics``

# (SDiD: ``in_time_placebo``, ``sensitivity_to_zeta_omega``;

# TROP: factor-model fit metrics) and do not accept HonestDiD

# bounds because they are methodology-incompatible with the

# documented native-routing contract in REPORTING.md. Reject

# the passthrough here so it doesn't silently forward to the

# auto-constructed ``DiagnosticReport`` (which now also

# rejects it at construction time — round-21 P1 CI review on

# PR #318).

if honest_did_results is not None and type(results).__name__ in {

"SyntheticDiDResults",

"TROPResults",

}:

raise ValueError(

f"{type(results).__name__} routes robustness to "

"``estimator_native_diagnostics`` — ``honest_did_results`` "

"is not accepted on this estimator because HonestDiD "

"bounds are methodology-incompatible with the native "

"routing documented in REPORTING.md. Use the result "

"object's native diagnostics "

"(SDiD: ``in_time_placebo()``, ``sensitivity_to_zeta_omega()``, "

"``pre_treatment_fit``; TROP: ``effective_rank``, "

"``loocv_score``) — BusinessReport surfaces these "

"automatically under ``estimator_native_diagnostics``."

)

# Round-44 P1 CI review on PR #318: mirror the SDiD/TROP

# rejection pattern for ``CallawaySantAnna`` fits with

# ``base_period != "universal"``. HonestDiD Rambachan-Roth

# bounds are not valid for interpretation on the consecutive-

# comparison pre-period surface produced by ``varying`` base,

# so narrating precomputed sensitivity (whether passed as

# ``honest_did_results`` or ``precomputed['sensitivity']``)

# alongside a displayed varying-base fit mixes provenance the

# bounds don't support. DR enforces the same guard at

# construction; BR duplicates the check so the error fires

# before the auto-DR is built, matching the existing

# SDiD/TROP UX. REGISTRY.md §CallawaySantAnna line 410,

# §HonestDiD line 2458.

_cs_with_varying_base = type(results).__name__ == "CallawaySantAnnaResults" and (

getattr(results, "base_period", "universal") != "universal"

)

if _cs_with_varying_base:

_rejected_inputs: List[str] = []

if honest_did_results is not None:

_rejected_inputs.append("honest_did_results")

if precomputed is not None and "sensitivity" in precomputed:

_rejected_inputs.append("precomputed['sensitivity']")

if _rejected_inputs:

_base_period = getattr(results, "base_period", "universal")

raise ValueError(

f"CallawaySantAnnaResults with "

f"``base_period={_base_period!r}`` cannot be "

"summarized alongside a precomputed HonestDiD "

"sensitivity object. The Rambachan-Roth bounds are "

"not valid for interpretation on the consecutive-"

"comparison pre-period surface this base yields "

"(REGISTRY.md §CallawaySantAnna / §HonestDiD). "

"Rejected inputs: " + ", ".join(_rejected_inputs) + ". "

"Re-fit the main estimator with "

"``CallawaySantAnna(base_period='universal')`` "

"before passing precomputed sensitivity, or drop "

"the sensitivity passthrough to let BR skip the "

"section with a methodology-critical reason."

)

self._results = results

self._honest_did_results = honest_did_results

self._auto_diagnostics = auto_diagnostics

self._diagnostics_arg = diagnostics

self._include_appendix = include_appendix

# Raw-data passthrough so the auto-constructed DR can run

# data-dependent checks (2x2 PT on simple DiD, Bacon-from-

# scratch on staggered estimators, EfficientDiD Hausman

# pretest). Without these, the auto path silently skips those

# checks (round-12 CI review on PR #318).

self._dr_data = data

self._dr_outcome = outcome

self._dr_treatment = treatment

self._dr_unit = unit

self._dr_time = time

self._dr_first_treat = first_treat

# Round-40 P1 CI review on PR #318: survey-backed fits need

# the ``SurveyDesign`` threaded through to the auto-constructed

# DR so Bacon decomposition is fit-faithful and the 2x2 PT

# skip path triggers for DiDResults with ``survey_metadata``.

# Without this passthrough, the auto path silently replays an

# unweighted decomposition / PT verdict for a weighted fit.

self._dr_survey_design = survey_design

# Round-43 P2 CI review on PR #318: BR docs and docstrings

# advertised a ``precomputed={'parallel_trends': ...}`` opt-in

# for survey-aware 2x2 PT and other escape hatches, but BR did

# not actually accept a ``precomputed=`` kwarg — the auto path

# only synthesized ``{"sensitivity": honest_did_results}``, so

# callers following the BR docs hit a ``TypeError`` on

# ``__init__``. Accept the passthrough here and forward every

# key to the auto-constructed DR (which owns validation against

# its implemented-key set and estimator-aware rejection rules).

# ``honest_did_results`` still feeds into ``sensitivity`` as a

# convenience; an explicit ``precomputed['sensitivity']`` wins

# on conflict.

self._dr_precomputed: Dict[str, Any] = dict(precomputed or {})

# Round-43 P2 CI review on PR #318: mirror DR's eager key

# validation so users get the "unsupported key" error at BR

# construction rather than lazily when the DR is built inside

# ``to_dict()``. Kept in sync with ``DiagnosticReport``'s

# ``_supported_precomputed`` set; the cheapest way to avoid

# drift would be to import the set, but DR currently scopes it

# locally to ``__init__`` so mirror the literal here with a

# pointer comment.

_br_supported_precomputed = {

"parallel_trends",

"sensitivity",

"pretrends_power",

"bacon",

}

_br_unsupported = set(self._dr_precomputed) - _br_supported_precomputed

if _br_unsupported:

raise ValueError(

"precomputed= contains keys that are not implemented: "

f"{sorted(_br_unsupported)}. Supported keys: "

f"{sorted(_br_supported_precomputed)}. ``design_effect``, "

"``heterogeneity``, and ``epv`` are read directly from the "

"fitted result and do not accept precomputed overrides."

)

resolved_alpha = alpha if alpha is not None else getattr(results, "alpha", 0.05)

self._context = BusinessContext(

outcome_label=outcome_label or "the outcome",

outcome_unit=outcome_unit,

outcome_direction=outcome_direction,

business_question=business_question,

treatment_label=treatment_label or "the treatment",

alpha=float(resolved_alpha),

)

self._cached_schema: Optional[Dict[str, Any]] = None

# -- Public API ---------------------------------------------------------

def to_dict(self) -> Dict[str, Any]:

"""Return the AI-legible structured schema (single source of truth)."""

if self._cached_schema is None:

self._cached_schema = self._build_schema()

return self._cached_schema

def to_json(self, *, indent: int = 2) -> str:

"""Return ``to_dict()`` serialized as JSON."""

import json

return json.dumps(self.to_dict(), indent=indent)

def summary(self) -> str:

"""Return a short plain-English paragraph block (6-10 sentences)."""

return _render_summary(self.to_dict())

def full_report(self) -> str:

"""Return a structured multi-section markdown report."""

base = _render_full_report(self.to_dict())

if self._include_appendix:

try:

appendix = self._results.summary()

except Exception: # noqa: BLE001

appendix = None

if appendix:

base = base + "\n\n## Technical Appendix\n\n```\n" + str(appendix) + "\n```\n"

return base

def export_markdown(self) -> str:

"""Alias for ``full_report()`` (discoverability)."""

return self.full_report()

def headline(self) -> str:

"""Return just the headline sentence."""

return _render_headline_sentence(self.to_dict())

def caveats(self) -> List[Dict[str, str]]:

"""Return the list of structured caveats (severity + topic + message)."""

return list(self.to_dict().get("caveats", []))

def __repr__(self) -> str:

estimator = type(self._results).__name__

headline = self.to_dict().get("headline") or {}

val = headline.get("effect")

if isinstance(val, (int, float)) and np.isfinite(val):

return f"BusinessReport(results={estimator}, effect={val:.3g})"

return f"BusinessReport(results={estimator})"

def __str__(self) -> str:

return self.summary()

# -- Implementation detail ---------------------------------------------

def _resolve_diagnostics(self) -> Optional[DiagnosticReportResults]:

"""Return the DiagnosticReportResults to embed, or ``None`` if skipped."""

if self._diagnostics_arg is not None:

if isinstance(self._diagnostics_arg, DiagnosticReportResults):

return self._diagnostics_arg

if isinstance(self._diagnostics_arg, DiagnosticReport):

return self._diagnostics_arg.run_all()

raise TypeError("diagnostics= must be a DiagnosticReport or DiagnosticReportResults")

if not self._auto_diagnostics:

return None

# Round-43 P2 CI review on PR #318: forward the user's

# ``precomputed`` dict through to DR. ``honest_did_results``

# stays a convenience shortcut for ``sensitivity`` only; an

# explicit ``precomputed['sensitivity']`` from the caller

# wins. DR handles key validation (rejects unsupported keys

# and estimator-incompatible sensitivities / parallel_trends

# entries) so BR just merges and forwards.

precomputed: Dict[str, Any] = dict(self._dr_precomputed)

if self._honest_did_results is not None:

precomputed.setdefault("sensitivity", self._honest_did_results)

dr = DiagnosticReport(

self._results,

alpha=self._context.alpha,

precomputed=precomputed or None,

outcome_label=self._context.outcome_label,

treatment_label=self._context.treatment_label,

data=self._dr_data,

outcome=self._dr_outcome,

treatment=self._dr_treatment,

unit=self._dr_unit,

time=self._dr_time,

first_treat=self._dr_first_treat,

survey_design=self._dr_survey_design,

)

return dr.run_all()

def _build_schema(self) -> Dict[str, Any]:

"""Assemble the structured schema.

Pulls validation content (PT, sensitivity, Bacon, DEFF, EPV, ...) from

the internal ``DiagnosticReport``; extracts the stakeholder-facing

headline and sample metadata from the fitted result itself.

"""

estimator_name = type(self._results).__name__

diagnostics_results = self._resolve_diagnostics()

dr_schema: Optional[Dict[str, Any]] = (

diagnostics_results.schema if diagnostics_results is not None else None

)

# PR #347 R4 P1: compute target_parameter BEFORE extracting

# the headline so the no-scalar-by-design case

# (``aggregation == "no_scalar_headline"``, e.g., dCDH

# ``trends_linear=True`` with ``L_max >= 2``) can route the

# headline through a dedicated branch that names the intentional

# NaN rather than an estimation-failure path.

target_parameter = describe_target_parameter(self._results)

if target_parameter.get("aggregation") == "no_scalar_headline":

# PR #347 R12 P1: the no-scalar ``reason`` must distinguish

# the populated-surface case (per-horizon table exists) from

# the empty-surface subcase (``linear_trends_effects=None``

# — no horizons survived estimation). Telling a user with

# an empty surface to "see linear_trends_effects" is

# dead-end guidance.

_surface_empty = getattr(self._results, "linear_trends_effects", None) is None

# PR #347 R14 P1: the empty-surface reason must use the

# covariate-adjusted label when covariates are active.

_has_controls = getattr(self._results, "covariate_residuals", None) is not None

_empty_surface_label = "DID^{X,fd}_l" if _has_controls else "DID^{fd}_l"

if _surface_empty:

no_scalar_reason = (

"The fitted estimator intentionally does not produce a "

"scalar overall ATT on this configuration "

"(``trends_linear=True`` with ``L_max >= 2``), and on "

f"this fit no cumulated level effects ``{_empty_surface_label}`` "

"survived estimation — the per-horizon surface is "

"empty. Re-fit with a larger ``L_max`` or with "

"``trends_linear=False`` if you need a reportable "

"estimand."

)

else:

no_scalar_reason = (

"The fitted estimator intentionally does not produce a "

"scalar overall ATT on this configuration "

"(``trends_linear=True`` with ``L_max >= 2``). Per-horizon "

"cumulated level effects are on "

"``results.linear_trends_effects[l]``."

)

headline = {

"status": "no_scalar_by_design",

"effect": None,

"se": None,

"ci_lower": None,

"ci_upper": None,

"alpha_was_honored": True,

"alpha_override_caveat": None,

"ci_level": int(round((1.0 - self._context.alpha) * 100)),

"p_value": None,

"is_significant": False,

"near_significance_threshold": False,

"unit": self._context.outcome_unit,

"unit_kind": _UNIT_KINDS.get(

self._context.outcome_unit.lower() if self._context.outcome_unit else "",

"unknown",

),

"sign": "none",

"breakdown_M": None,

"reason": no_scalar_reason,

}

else:

headline = self._extract_headline(dr_schema)

sample = self._extract_sample()

heterogeneity = _lift_heterogeneity(dr_schema)

pre_trends = _lift_pre_trends(dr_schema)

sensitivity = _lift_sensitivity(dr_schema)

robustness = _lift_robustness(dr_schema)

assumption = _apply_anticipation_to_assumption(

_describe_assumption(estimator_name, self._results),

self._results,

)

next_steps = (dr_schema or {}).get("next_steps", [])

caveats = _build_caveats(self._results, headline, sample, dr_schema)

references = _references_for(estimator_name)

if diagnostics_results is None:

diagnostics_block: Dict[str, Any] = {

"status": "skipped",

"reason": "auto_diagnostics=False",

}

else:

diagnostics_block = {

"status": "ran",

"schema": dr_schema,

"overall_interpretation": (

dr_schema.get("overall_interpretation", "") if dr_schema is not None else ""

),

}

return {

"schema_version": BUSINESS_REPORT_SCHEMA_VERSION,

"estimator": {

"class_name": estimator_name,

"display_name": estimator_name,

},

"context": {

"outcome_label": self._context.outcome_label,

"outcome_unit": self._context.outcome_unit,

"outcome_direction": self._context.outcome_direction,

"business_question": self._context.business_question,

"treatment_label": self._context.treatment_label,

"alpha": self._context.alpha,

},

"headline": headline,

"target_parameter": target_parameter,

"assumption": assumption,

"pre_trends": pre_trends,

"sensitivity": sensitivity,

"sample": sample,

"heterogeneity": heterogeneity,

"robustness": robustness,

"diagnostics": diagnostics_block,

"next_steps": next_steps,

"caveats": caveats,

"references": references,

}

def _extract_headline(self, dr_schema: Optional[Dict[str, Any]]) -> Dict[str, Any]:

"""Extract the headline effect + CI + p-value from the result."""

r = self._results

# Delegate the attribute-alias lookup to the shared helper in the

# diagnostic_report module so BR and DR agree on which fields a

# result class exposes for its headline (including

# ``ContinuousDiDResults`` which uses ``overall_att_se`` /

# ``overall_att_p_value`` / ``overall_att_conf_int``).

from diff_diff.diagnostic_report import _extract_scalar_headline

extracted = _extract_scalar_headline(r, fallback_alpha=self._context.alpha)

att: Optional[float] = None

se: Optional[float] = None

p: Optional[float] = None

ci: Optional[List[float]] = None

alpha = self._context.alpha

result_alpha: Optional[float] = None

if extracted is not None:

_name, att, se, p, ci, result_alpha = extracted

# On any alpha mismatch, preserve the fitted CI at its native

# level. A faithful CI cannot be recomputed from point estimate

# and SE alone without reproducing the fit's inference contract

# (finite-df t-quantile, percentile bootstrap, wild cluster

# bootstrap, survey replicate quantile, rank-deficient

# undefined-df, etc.), and the 16 result classes do not expose

# a uniform descriptor for that. Two separate alpha values:

# ``display_alpha`` drives ``ci_level`` so the displayed CI

# label matches the preserved bounds; the caller's requested

# alpha drives the significance phrasing (``is_significant`` /

# ``near_threshold``). A caveat records the override.

display_alpha = alpha

phrasing_alpha = alpha

alpha_was_honored = True

alpha_override_caveat: Optional[str] = None

if (

result_alpha is not None

and not np.isclose(alpha, result_alpha)

and att is not None

and se is not None

):

inference_method = getattr(r, "inference_method", "analytical")

if inference_method == "wild_bootstrap":

inference_label = "wild cluster bootstrap"

elif (

inference_method == "bootstrap" or getattr(r, "bootstrap_results", None) is not None

):

inference_label = "bootstrap"

elif getattr(r, "bootstrap_distribution", None) is not None:

inference_label = "bootstrap"

elif getattr(r, "variance_method", None) in {"bootstrap", "jackknife", "placebo"}:

variance_method = getattr(r, "variance_method", None)

inference_label = f"{variance_method} variance"

else:

df_survey = getattr(

r,

"df_survey",

getattr(getattr(r, "survey_metadata", None), "df_survey", None),

)

if isinstance(df_survey, (int, float)) and df_survey > 0:

inference_label = "finite-df survey"

elif isinstance(df_survey, (int, float)) and df_survey == 0:

# Rank-deficient replicate design: the fit deliberately

# left inference undefined. Preserve (NaN bounds remain NaN).

inference_label = "undefined-df (replicate-weight)"

else:

# Ordinary analytical fit with a finite but unexposed

# ``df`` (``DifferenceInDifferences`` / ``MultiPeriodDiD``

# / most staggered estimators / TROP). We cannot

# reproduce the t-quantile without the fit's ``df``.

inference_label = "analytical (native degrees of freedom)"

display_alpha = float(result_alpha)

alpha_was_honored = False

alpha_override_caveat = (

f"Requested alpha ({phrasing_alpha:.2f}) was not honored "

f"for the confidence interval because this fit uses "

f"{inference_label} inference; the displayed CI remains "

f"at the fit's native level "

f"({int(round((1.0 - result_alpha) * 100))}%). The "

f"significance phrasing still uses the requested alpha."

)

unit = self._context.outcome_unit

unit_kind = _UNIT_KINDS.get(unit.lower() if unit else "", "unknown")

sign = (

"positive"

if (att is not None and att > 0)

else (

"negative"

if (att is not None and att < 0)

else ("null" if att == 0 else "undefined")

)

)

if att is None or not np.isfinite(att):

sign = "undefined"

ci_level = int(round((1.0 - display_alpha) * 100))

is_significant = (

p is not None and np.isfinite(p) and p < phrasing_alpha if p is not None else False

)

near_threshold = (

p is not None

and np.isfinite(p)

and (phrasing_alpha - 0.01) < p < (phrasing_alpha + 0.001)

)

# Use DR-computed breakdown_M if available for quick reference.

breakdown_M: Optional[float] = None

if dr_schema:

sens_section = dr_schema.get("sensitivity") or {}

if sens_section.get("status") == "ran":

breakdown_M = sens_section.get("breakdown_M")

return {

"effect": att,

"se": se,

"ci_lower": ci[0] if ci else None,

"ci_upper": ci[1] if ci else None,

"alpha_was_honored": alpha_was_honored,

"alpha_override_caveat": alpha_override_caveat,

"ci_level": ci_level,

"p_value": p,

"is_significant": is_significant,

"near_significance_threshold": near_threshold,

"unit": unit,

"unit_kind": unit_kind,

"sign": sign,

"breakdown_M": breakdown_M,

}

def _extract_sample(self) -> Dict[str, Any]:

"""Extract sample metadata from the fitted result."""

r = self._results

survey = self._extract_survey_block()

n_treated = _safe_int(getattr(r, "n_treated", getattr(r, "n_treated_units", None)))

n_control_units = _safe_int(getattr(r, "n_control", getattr(r, "n_control_units", None)))

# Control-group semantics. For estimators that expose a

# ``control_group`` kwarg (CS, EfficientDiD, ContinuousDiD,

# StaggeredTripleDiff, ...), the meaning of ``n_control_units``

# depends on it. When the mode is "not-yet-treated" (dynamic

# comparison set), the fixed tally stored on the result is only

# the fully-untreated subset — the actual comparison set varies

# by (g, t) cell. Label the exposed count accordingly so prose

# surfaces the dynamic context instead of misreporting

# "0 control" (round-13 / round-17 / round-18 CI review).

#

# Canonicalize both ``"not_yet_treated"`` (CS / EfficientDiD /

# ContinuousDiD / Wooldridge) and ``"notyettreated"``

# (StaggeredTripleDiff) as the same dynamic mode.

#

# Per-estimator fixed-subset field:

# * CS / SA / Imputation / TwoStage / EfficientDiD /

# dCDH / ContinuousDiD — ``n_control_units`` is the

# never-treated tally; surface as ``n_never_treated``.

# * StaggeredTripleDiff — ``n_control_units`` is a composite

# total; the fixed subset is ``n_never_enabled`` (stored

# separately on the result).

# * Wooldridge — ``n_control_units`` is total eligible

# comparisons (never-treated + future-treated) and does not

# map to a never-treated count. Keep on the fixed-count

# path even in dynamic mode.

# * Stacked — ``n_control_units`` is "distinct control units

# across the trimmed set" (stacked_did_results.py L59-62).

# Under ``clean_control="not_yet_treated"``, the trimmed

# set uses the rule ``A_s > a + kappa_post`` which admits

# future-treated controls; it is NOT a never-treated tally

# and cannot be relabeled as ``n_never_treated``. Keep

# Stacked on the fixed-count path (round-21 P1 CI review

# on PR #318 flagged the earlier relabeling as a

# semantic-contract violation).

control_group = _control_group_choice(r)

name = type(r).__name__

n_never_treated: Optional[int] = None

n_never_enabled: Optional[int] = None

n_control: Optional[int] = n_control_units

_never_treated_count_contract = name in {

"CallawaySantAnnaResults",

"SunAbrahamResults",

"ImputationDiDResults",

"TwoStageDiDResults",

"EfficientDiDResults",

"ChaisemartinDHaultfoeuilleResults",

"ContinuousDiDResults",

}

_canonical_control = (

control_group.replace("_", "").lower() if isinstance(control_group, str) else None

)

# Stacked has two dynamic (sub-experiment-specific) modes:

# ``not_yet_treated`` (A_s > a + kappa_post) and ``strict``

# (A_s > a + kappa_post + kappa_pre). Only ``never_treated``

# (A_s = infinity) is a fixed never-treated pool. Round-22 P1

# CI review on PR #318 flagged that ``strict`` was being

# misrendered as a fixed control design.

is_stacked_dynamic = name == "StackedDiDResults" and _canonical_control in {

"notyettreated",

"strict",

}

is_dynamic_control = _canonical_control == "notyettreated" or is_stacked_dynamic

# StaggeredTripleDiff comparison-group contract:

# ``n_control_units`` is a composite total that also includes

# the eligibility-denied / larger-cohort cells. Regardless of

# the ``control_group`` mode the valid fixed comparison is the

# never-enabled cohort (``staggered_triple_diff.py:384``,

# REGISTRY.md §StaggeredTripleDifference line 1730). Round-37

# P1 CI review on PR #318: under ``control_group="never_treated"``

# (i.e., ``_canonical_control == "nevertreated"``) the composite

# total was being narrated as "control". Surface

# ``n_never_enabled`` instead on both the ``nevertreated`` and

# the dynamic ``notyettreated`` modes.

if name == "StaggeredTripleDiffResults" and _canonical_control == "nevertreated":

n_never_enabled = _safe_int(getattr(r, "n_never_enabled", None))

n_control = None

if is_dynamic_control:

if name == "StaggeredTripleDiffResults":

n_never_enabled = _safe_int(getattr(r, "n_never_enabled", None))

n_control = None

elif name == "StackedDiDResults":

# ``n_control_units`` is "distinct control units across

# the trimmed set" (stacked_did_results.py L59-62) which

# includes future-treated controls by construction under

# both dynamic modes. Do NOT relabel as

# ``n_never_treated``; instead surface the count under

# ``n_distinct_controls_trimmed`` (sub-experiment-

# specific context) and clear ``n_control`` so the

# report does not narrate a fixed control pool.

n_control = None

elif _never_treated_count_contract:

n_never_treated = n_control_units

n_control = None

# Panel-vs-RCS count semantics. CallawaySantAnnaResults stores

# treated/control counts as OBSERVATIONS (not units) when the

# fit used ``panel=False`` — ``staggered_results.py L183-L184``

# renders those counts as "obs:" rather than "units:". BR

# previously labeled them as "units" / "present in the panel",

# which misstates the sample composition for repeated cross-

# section fits. Carry the flag into the schema so rendering can

# branch. Round-28 P2 CI review on PR #318.

count_unit = "observations" if getattr(r, "panel", True) is False else "units"

sample_block: Dict[str, Any] = {

"n_obs": _safe_int(getattr(r, "n_obs", None)),

"n_treated": n_treated,

"n_control": n_control,

"n_never_treated": n_never_treated,

"control_group": control_group if isinstance(control_group, str) else None,

"dynamic_control": is_dynamic_control,

"n_periods": _safe_int(getattr(r, "n_periods", None)),

"pre_periods": _safe_list_len(getattr(r, "pre_periods", None)),

"post_periods": _safe_list_len(getattr(r, "post_periods", None)),

"count_unit": count_unit,

"survey": survey,

}

if n_never_enabled is not None:

sample_block["n_never_enabled"] = n_never_enabled

# Stacked-specific: surface the distinct-control-units tally on a

# dedicated key so agents see the sub-experiment-specific

# comparison count without misreading it as a never-treated

# subset (round-21 / round-22 CI review).

if name == "StackedDiDResults":

sample_block["n_distinct_controls_trimmed"] = n_control_units

return sample_block

def _extract_survey_block(self) -> Optional[Dict[str, Any]]:

sm = getattr(self._results, "survey_metadata", None)

if sm is None:

return None

deff = _safe_float(getattr(sm, "design_effect", None))

return {

"weight_type": getattr(sm, "weight_type", None),

"effective_n": _safe_float(getattr(sm, "effective_n", None)),

"design_effect": deff,

# Round-43 P2 CI review on PR #318: the ``is_trivial``

# upper bound matches DR's ``_check_design_effect`` and

# REPORTING.md's ``trivial`` band definition

# ``0.95 <= deff < 1.05`` (half-open). The prior closed

# interval ``<= 1.05`` produced ``is_trivial=True`` at

# exactly ``deff == 1.05`` while the DR schema emitted

# ``band_label="slightly_reduces"`` for the same value,

# suppressing BR's non-trivial prose at that boundary.

"is_trivial": deff is not None and 0.95 <= deff < 1.05,

"n_strata": _safe_int(getattr(sm, "n_strata", None)),

"n_psu": _safe_int(getattr(sm, "n_psu", None)),

"df_survey": _safe_int(getattr(sm, "df_survey", None)),

"replicate_method": getattr(sm, "replicate_method", None),

}

# ---------------------------------------------------------------------------

# Schema helpers (module-private)

# ---------------------------------------------------------------------------

def _safe_float(val: Any) -> Optional[float]:

if val is None:

return None

try:

return float(val)

except (TypeError, ValueError):

return None

def _safe_int(val: Any) -> Optional[int]:

if val is None:

return None

try:

return int(val)

except (TypeError, ValueError):

return None

def _safe_ci(ci: Any) -> Optional[List[float]]:

if ci is None:

return None

try:

lo, hi = ci

except (TypeError, ValueError):

return None

lo_f = _safe_float(lo)

hi_f = _safe_float(hi)

if lo_f is None or hi_f is None:

return None

return [lo_f, hi_f]

def _safe_list_len(val: Any) -> Optional[int]:

if val is None:

return None

try:

return int(len(val))

except TypeError:

return None

def _lift_pre_trends(dr: Optional[Dict[str, Any]]) -> Dict[str, Any]:

"""Pull pre-trends + power into a single BR-facing block."""

if dr is None:

return {"status": "skipped", "reason": "auto_diagnostics=False"}

pt = dr.get("parallel_trends") or {}

pp = dr.get("pretrends_power") or {}

if pt.get("status") != "ran":

return {

"status": pt.get("status", "not_run"),

"reason": pt.get("reason"),

}

return {

"status": "computed",

"method": pt.get("method"),

"joint_p_value": pt.get("joint_p_value"),

"verdict": pt.get("verdict"),

"n_pre_periods": pt.get("n_pre_periods"),

# Preserve DR's inconclusive-PT provenance on the BR schema so

# downstream consumers (and BR's own summary renderer) see the

# undefined-row count and DR's detailed reason without having

# to re-consult the DR schema (round-39 P3 CI review on PR

# #318). These fields are populated only when

# ``verdict == "inconclusive"`` per ``_pt_event_study``'s

# inconclusive branch (``diagnostic_report.py:999``).

"n_dropped_undefined": pt.get("n_dropped_undefined"),

"reason": pt.get("reason"),

# Carry the denominator df through when the survey F-reference

# branch was used so BR consumers can flag the finite-sample

# correction without re-consulting the DR schema (round-28 P3

# CI review on PR #318).

"df_denom": pt.get("df_denom"),

"power_status": pp.get("status"),

# Dedicated reason field so schema consumers see the fallback

# explanation when ``compute_pretrends_power`` cannot run

# (``status in {"skipped", "error", "not_applicable"}``).

# REPORTING.md lines 118-125 promise this provenance; round-29

# P3 CI review on PR #318 flagged that only the enum status was

# being exposed and the reason was dropped at the lift boundary.

# ``power_status`` stays the machine-readable enum; ``power_reason``

# carries the plain-English explanation.

"power_reason": pp.get("reason"),

"power_tier": pp.get("tier"),

"mdv": pp.get("mdv"),

"mdv_share_of_att": pp.get("mdv_share_of_att"),

# Carry the covariance-source annotation through so BR can hedge the

# power-tier phrasing when compute_pretrends_power silently used a

# diagonal fallback despite event_study_vcov being available.

"power_covariance_source": pp.get("covariance_source"),

}

def _lift_sensitivity(dr: Optional[Dict[str, Any]]) -> Dict[str, Any]:

if dr is None:

return {"status": "skipped", "reason": "auto_diagnostics=False"}

sens = dr.get("sensitivity") or {}

if sens.get("status") != "ran":

# Preserve ``method`` through to the BR schema so downstream

# consumers can distinguish a native-routed skip

# (``method="estimator_native"`` for SDiD / TROP, where

# robustness is covered by the native battery) from a

# methodology-blocked skip (e.g., CS with

# ``base_period='varying'``). Without it, agents reading the BR

# schema alone cannot tell these cases apart and would have to

# re-consult the DR schema to disambiguate.

return {

"status": sens.get("status", "not_run"),

"reason": sens.get("reason"),

"method": sens.get("method"),

}

return {

"status": "computed",

"method": sens.get("method"),

"breakdown_M": sens.get("breakdown_M"),

"conclusion": sens.get("conclusion"),

"grid": sens.get("grid"),

}

def _lift_heterogeneity(dr: Optional[Dict[str, Any]]) -> Dict[str, Any]:

"""Return the heterogeneity section of the BR schema.

Round-31 P2 CI review on PR #318: the lift previously returned

``None`` on any non-``ran`` path, which broke the schema contract

that every top-level BR key resolves to a dict with a ``status``

field. Downstream consumers had to special-case this one section.

Now returns a dict-shaped ``{"status": ..., "reason": ...}`` block

mirroring DR's own status enum so ``schema["heterogeneity"]

["status"]`` is always readable.

"""

if dr is None:

return {"status": "skipped", "reason": "auto_diagnostics=False"}

het = dr.get("heterogeneity") or {}

status = het.get("status")

if status != "ran":

return {

"status": status or "not_run",

"reason": het.get("reason"),

}

return {

"status": "ran",

"source": het.get("source"),

"n_effects": het.get("n_effects"),

"min": het.get("min"),

"max": het.get("max"),

"cv": het.get("cv"),

"sign_consistent": het.get("sign_consistent"),

}

def _lift_robustness(dr: Optional[Dict[str, Any]]) -> Dict[str, Any]:

if dr is None:

return {"status": "skipped", "reason": "auto_diagnostics=False"}

bacon = dr.get("bacon") or {}

native = dr.get("estimator_native_diagnostics") or {}

return {

"bacon": {

"status": bacon.get("status"),