feat(smart): compare_estimators method_hints + sensitivity Sprint-B dims + g_computation error compaction

Three smart-layer enhancements closing the remaining Sprint-B
ergonomic gaps, plus five fixes from a fresh-eyes code review.

1. sp.compare_estimators(method_hints={...})
   New kwarg forwards per-method kwargs to each estimator when the
   shared (data, y, treatment, covariates, id, time, instrument)
   signature is too narrow. Enables six Sprint-B methods in batch
   comparison:
     - proximal            — needs {proxy_z, proxy_w}
     - msm                 — needs {time_varying} + id + time
     - principal_strat     — needs {strata}
     - front_door          — needs {mediator}
     - mediate             — needs {mediator}
     - mediate_interventional — needs {mediator, tv_confounders?}
   _resolve_hints() merges shared + hint per ROADMAP §6:
   per-method hints take precedence, UserWarning fires on conflict.
   List comparison is set-based (order-insensitive) so a
   reordered-but-set-equal list doesn't trigger spurious warnings.

2. sp.sensitivity_dashboard — Sprint-B-aware dimensions
   - 'first_stage_f' (proximal): surfaces the IV-weakness check from
     model_info as its own dashboard row with F ≥ 10 Stock-Yogo pass
     logic.
   - 'trim_sweep' (msm): max-stabilized-weight positivity watchdog
     against the same 50 threshold as the diagnostic battery.
   - 'monotonicity' (principal_strat): monotonicity-violation
     fraction, 5% tolerance.
   When dimensions= is left at default None, the list now
   auto-expands with the matching Sprint-B token so callers don't
   need to know the token. Baseline extraction handles
   PrincipalStratResult (no .estimate / .params) by pulling the
   complier row from .effects by label, not by .iloc[0] — so an
   upstream reorder of the effects DataFrame can't silently make
   the dashboard report an always-taker effect. Method label
   prefers model_info['estimator'] for PSR.

3. sp.g_computation error message compaction
   Non-binary D previously dumped the full sorted list of unique
   values (~20 KB for n=400 continuous D) which exploded to a
   multi-KB UserWarning when compare_estimators wrapped it. The
   message now truncates to 5 values + "(N more)" count, filters
   NaNs out of the sort (NaN sort order is undefined), and notes
   NaN presence separately.

Five code-review findings applied:
- BLOCKER: hint-wins test now verifies the hint VALUE (via a second
  run with a different hint set that produces a different estimate),
  not just that a warning fired.
- MAJOR: list equality in _resolve_hints is set-based
  (reorder-tolerant).
- MAJOR: complier row extraction uses stratum name filter, not
  .iloc[0] positional indexing, in both sensitivity.py and
  compare.py.
- MINOR: sorted(uniq) in g_computation filters NaN pre-sort and
  annotates separately.
- MINOR: sensitivity_dashboard default dimensions auto-expand with
  Sprint-B tokens so callers without the explicit token still see
  the method-specific rows.

Regression: 200/200 Sprint-B tests pass (12 new this round).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: brycewang-stanford

src/statspai/inference/g_computation.py

@@ -129,9 +129,25 @@ def g_computation(
     if estimand in ('ATE', 'ATT'):
         uniq = set(np.unique(D))
         if not uniq.issubset({0, 1}):
+            # Truncate the value dump: a continuous treatment can have
+            # hundreds of unique values and the full list turns the
+            # error message into a ~20 KB wall of floats — this
+            # explodes warning-message size when a caller (e.g. the
+            # compare_estimators batch loop) wraps the ValueError in
+            # a UserWarning. Filter NaNs before sorting (a lone NaN
+            # would sit in a non-deterministic position in Python's
+            # sort and omit real values from the preview).
+            finite = [v for v in uniq
+                      if not (isinstance(v, float) and np.isnan(v))]
+            has_nan = len(finite) < len(uniq)
+            vals = sorted(finite)
+            preview = ', '.join(f'{v:.4g}' for v in vals[:5])
+            tail = f', ... ({len(vals) - 5} more)' if len(vals) > 5 else ''
+            nan_note = ' (plus NaN values present)' if has_nan else ''
             raise ValueError(
                 f"estimand='{estimand}' requires binary treatment (0/1); "
-                f"found values {sorted(uniq)}. For multi-valued or "
+                f"treatment has {len(vals)} unique values "
+                f"[{preview}{tail}]{nan_note}. For multi-valued or "
                 f"continuous D, use estimand='dose_response'."
             )
         grid = np.array([0.0, 1.0])

src/statspai/smart/compare.py

@@ -115,6 +115,7 @@ def compare_estimators(
     time: str = None,
     instrument: str = None,
     alpha: float = 0.05,
+    method_hints: Optional[Dict[str, Dict[str, Any]]] = None,
 ) -> ComparisonResult:
     """
     Run multiple estimators on the same data and compare.
@@ -131,16 +132,31 @@ def compare_estimators(
         Treatment variable (binary).
     methods : list of str, optional
         Estimators to compare. Default auto-selects based on data.
-        Options: 'ols', 'matching', 'ipw', 'aipw', 'dml',
-        'g_computation', 'causal_forest', 'did', 'panel_fe'.
-
-        Not supported here (require extra arguments beyond the shared
-        signature): ``'proximal'`` (needs proxy_z / proxy_w),
-        ``'msm'`` (needs time_varying / id / time),
-        ``'principal_strat'`` (needs a post-treatment strata variable),
-        and the mediation family (needs a mediator). Call those
-        estimators directly; see docs/ROADMAP.md for the planned
-        hint-aware multi-method comparison surface.
+        Classical options: ``'ols'``, ``'matching'``, ``'ipw'``,
+        ``'aipw'``, ``'dml'``, ``'g_computation'``,
+        ``'causal_forest'``, ``'did'``, ``'panel_fe'``.
+
+        Hint-driven Sprint-B options (require ``method_hints``):
+        ``'proximal'``, ``'msm'``, ``'principal_strat'``, ``'mediate'``,
+        ``'mediate_interventional'``, ``'front_door'``. Each needs
+        method-specific kwargs the shared signature does not expose
+        (proxy_z/proxy_w, time_varying, strata, mediator, etc.) — pass
+        them through ``method_hints``.
+    method_hints : dict, optional
+        Per-method keyword overrides, merged with the shared kwargs
+        when dispatching each estimator. Structure::
+
+            {'proximal': {'proxy_z': ['z'], 'proxy_w': ['w']},
+             'msm':      {'time_varying': ['L_lag']},
+             'principal_strat': {'strata': 's'}}
+
+        **Collision rule** (docs/ROADMAP.md §6): per-method hints
+        take precedence over the shared kwargs for the method they
+        name. If the top-level ``covariates=['age']`` disagrees with
+        ``method_hints={'proximal': {'covariates': ['age', 'educ']}}``,
+        proximal uses the hint and every other method uses the
+        shared arg. A ``UserWarning`` fires on conflict so the
+        override is visible in the log.
     covariates : list of str, optional
     id : str, optional
         Panel unit ID.
@@ -181,6 +197,44 @@ def compare_estimators(
     n = len(df)
     z_crit = stats.norm.ppf(1 - alpha / 2)
 
+    # Resolve per-method hints with the ROADMAP §6 collision rule:
+    # hint-supplied args win over shared args for the method they name,
+    # with a UserWarning on explicit conflict.
+    method_hints = method_hints or {}
+
+    def _values_differ(a, b):
+        """
+        Compare two hint values, with list equality being set-based so
+        that pure reordering (e.g. ``['x1','x2']`` vs ``['x2','x1']``)
+        does not trigger a spurious conflict warning. Tuple / set
+        callers get the same treatment; everything else falls back
+        to standard ``!=``.
+        """
+        if isinstance(a, (list, tuple, set)) and isinstance(b, (list, tuple, set)):
+            try:
+                return set(a) != set(b)
+            except TypeError:
+                # Unhashable elements (e.g. nested lists) — fall back
+                # to strict element-wise comparison, still
+                # order-sensitive, but better than crashing.
+                return list(a) != list(b)
+        return a != b
+
+    def _resolve_hints(method_name, shared_kwargs):
+        """Merge shared kwargs with per-method hints; hint wins on conflict."""
+        hint = method_hints.get(method_name, {}) or {}
+        merged = dict(shared_kwargs)
+        for k, v in hint.items():
+            if k in merged and _values_differ(merged[k], v):
+                warnings.warn(
+                    f"compare_estimators: method_hints['{method_name}']"
+                    f"['{k}'] overrides the shared {k}={shared_kwargs[k]!r} "
+                    f"with {v!r}.",
+                    UserWarning, stacklevel=2,
+                )
+            merged[k] = v
+        return merged
+
     results = {}
     rows = []
 
@@ -279,6 +333,137 @@ def compare_estimators(
                 results['DID'] = r
                 name = 'Difference-in-Differences'
 
+            # ----- Sprint-B hint-driven branches -------------------- #
+            # Each requires per-method kwargs supplied via
+            # method_hints[method]; we raise a clean ValueError if the
+            # mandatory hint is missing so the outer try/except turns
+            # it into a compact UserWarning rather than silent skip.
+
+            elif method == 'proximal':
+                hint = method_hints.get('proximal', {}) or {}
+                proxy_z = hint.get('proxy_z')
+                proxy_w = hint.get('proxy_w')
+                if not proxy_z or not proxy_w:
+                    raise ValueError(
+                        "method='proximal' requires "
+                        "method_hints['proximal']={'proxy_z': [...], "
+                        "'proxy_w': [...]}."
+                    )
+                merged = _resolve_hints('proximal', {
+                    'covariates': [c for c in covariates[:10]
+                                   if c not in proxy_z + proxy_w],
+                })
+                r = sp.proximal(df, y=y, treat=treatment,
+                                proxy_z=list(proxy_z),
+                                proxy_w=list(proxy_w),
+                                covariates=merged['covariates'])
+                est = r.estimate
+                se = r.se
+                results['Proximal'] = r
+                name = 'Proximal Causal Inference (bridge 2SLS)'
+
+            elif method == 'msm':
+                hint = method_hints.get('msm', {}) or {}
+                tv = hint.get('time_varying')
+                if not tv or not id or not time:
+                    raise ValueError(
+                        "method='msm' requires id + time + "
+                        "method_hints['msm']={'time_varying': [...]}."
+                    )
+                merged = _resolve_hints('msm', {
+                    'baseline': [c for c in covariates[:5] if c not in tv],
+                })
+                r = sp.msm(df, y=y, treat=treatment, id=id, time=time,
+                           time_varying=list(tv),
+                           baseline=merged.get('baseline'))
+                est = r.estimate
+                se = r.se
+                results['MSM'] = r
+                name = 'Marginal Structural Model (IPTW)'
+
+            elif method == 'principal_strat':
+                hint = method_hints.get('principal_strat', {}) or {}
+                strata = hint.get('strata')
+                if not strata:
+                    raise ValueError(
+                        "method='principal_strat' requires "
+                        "method_hints['principal_strat']={'strata': <col>}."
+                    )
+                merged = _resolve_hints('principal_strat', {
+                    'covariates': covariates[:10],
+                    'method': 'principal_score' if covariates else 'monotonicity',
+                })
+                r = sp.principal_strat(
+                    df, y=y, treat=treatment, strata=strata,
+                    covariates=merged['covariates'] if merged['method'] == 'principal_score' else None,
+                    method=merged['method'],
+                )
+                # PrincipalStratResult is not a CausalResult; pull
+                # the complier PCE (= LATE) by stratum NAME rather
+                # than .iloc[0] so an upstream reordering of the
+                # effects table can't silently make us report the
+                # always-taker effect instead.
+                effects = getattr(r, 'effects', None)
+                if effects is not None and len(effects):
+                    _complier = effects['stratum'].astype(str).str.lower().str.contains('complier')
+                    _row = effects[_complier].iloc[0] if _complier.any() else effects.iloc[0]
+                    est = float(_row['estimate'])
+                    se = float(_row['se']) if 'se' in effects.columns else 0.0
+                else:
+                    est, se = np.nan, np.nan
+                results['Principal Strat'] = r
+                name = 'Principal Stratification (complier PCE)'
+
+            elif method == 'front_door':
+                hint = method_hints.get('front_door', {}) or {}
+                mediator = hint.get('mediator')
+                if not mediator:
+                    raise ValueError(
+                        "method='front_door' requires "
+                        "method_hints['front_door']={'mediator': <col>}."
+                    )
+                merged = _resolve_hints('front_door', {
+                    'covariates': [c for c in covariates[:10] if c != mediator],
+                })
+                r = sp.front_door(df, y=y, treat=treatment,
+                                  mediator=mediator,
+                                  covariates=merged['covariates'])
+                est = r.estimate
+                se = r.se
+                results['Front-door'] = r
+                name = 'Front-door adjustment'
+
+            elif method in ('mediate', 'mediate_interventional'):
+                hint = method_hints.get(method, {}) or {}
+                mediator = hint.get('mediator')
+                if not mediator:
+                    raise ValueError(
+                        f"method='{method}' requires "
+                        f"method_hints['{method}']={{'mediator': <col>}}."
+                    )
+                merged = _resolve_hints(method, {
+                    'covariates': [c for c in covariates[:10] if c != mediator],
+                })
+                if method == 'mediate':
+                    r = sp.mediate(df, y=y, treat=treatment,
+                                   mediator=mediator,
+                                   covariates=merged['covariates'] or None)
+                    label = 'Causal Mediation (ACME)'
+                    key = 'Mediation (natural)'
+                else:
+                    tv_c = hint.get('tv_confounders')
+                    r = sp.mediate_interventional(
+                        df, y=y, treat=treatment, mediator=mediator,
+                        covariates=merged['covariates'] or None,
+                        tv_confounders=list(tv_c) if tv_c else None,
+                    )
+                    label = 'Interventional Mediation (IIE)'
+                    key = 'Mediation (interventional)'
+                est = r.estimate
+                se = r.se
+                results[key] = r
+                name = label
+
             else:
                 continue