fix(pipeline): reject estimator keywords instead of discarding them

src/specmod/core/collection.py

@@ -135,6 +135,32 @@ def log_bin(
     return BinnedSpectrum(freq=centres[keep], amp=amps[keep])
 
 
+def _resolution_floor(spectrum: Spectrum) -> float:
+    """The lowest frequency ``spectrum`` can actually resolve.
+
+    Read from ``meta`` when it is there, and only derived from the axis when it
+    is not. That order is the whole point. Deriving it works exactly once: the
+    noise is interpolated onto the signal's axis before binning, so from then
+    on ``noise.freq.min()`` is the *signal's* lowest frequency and the noise's
+    own is gone. :func:`specmod.pipeline.spectrum_from_trace` records it on the
+    spectrum for that reason, and until now nothing read it.
+
+    The consequence was that a converted pair had a lower floor than the pair
+    it came from — the shorter noise window's limit silently replaced by the
+    longer signal window's. `to_motion` therefore let the band open into the
+    region below the noise's resolution, where :func:`interpolate_onto` is
+    repeating an edge value rather than reporting a measurement, and the
+    signal-to-noise ratio has an invented denominator.
+
+    Falls back to the axis for a spectrum built by hand rather than by the
+    pipeline, which is the only case where the axis is still the truth.
+    """
+    recorded = spectrum.meta.get("resolution_floor")
+    if recorded is not None:
+        return float(recorded)
+    return float(spectrum.freq.min()) if spectrum.freq.size else 0.0
+
+
 def parseval_scale(n_signal: int, n_noise: int) -> float:
     """Factor putting a noise spectrum on the signal's energy footing.
 
@@ -242,10 +268,7 @@ def compare(
         because afterwards the noise carries the signal's axis and its own
         lowest resolvable frequency is unrecoverable.
         """
-        floor = max(
-            float(signal.freq.min()) if signal.freq.size else 0.0,
-            float(noise.freq.min()) if noise.freq.size else 0.0,
-        )
+        floor = max(_resolution_floor(signal), _resolution_floor(noise))
 
         noise_amp = np.asarray(noise.amp, dtype=np.float64)
         if scale_parseval:
@@ -264,20 +287,32 @@ def compare(
         )
 
         if rotate_noise:
-            # Derived on the binned axis and applied to both, rather than
-            # computed twice: the unbinned factor is the binned one
-            # interpolated up, which is what the legacy code does and what
-            # keeps the two representations of "the noise" consistent.
+            # The factor is derived on the binned axis — that is where the
+            # method is defined — and applied to the *unbinned* noise, which
+            # then becomes the single source the binned noise is derived from.
+            #
+            # The order matters and used to be the other way round: the lift
+            # multiplied the bins directly and, separately, the unbinned array
+            # by the factor interpolated up. Those two operations do not agree.
+            # A bin holds the geometric mean of `log10(amp)`, so binning the
+            # lifted noise gives `mean(log a) + mean(log f)` while lifting the
+            # bin gives `mean(log a) + log f(centre)` — equal only where the
+            # factor is flat across the bin.
+            #
+            # The result was that a stored pair's `binned_noise` was not the
+            # binning of its own `noise`, by up to 18.8% on the PNR windows.
+            # Every pair was born inconsistent; a domain change re-bins, so
+            # `to_motion` silently *repaired* it and looked like the culprit.
             model = _resolve_noise_model(noise_model, rotation_space)
             factor = model.factor(
                 binned_noise.freq, binned_noise.amp, binned_signal.amp
             )
-            binned_noise = BinnedSpectrum(
-                freq=binned_noise.freq, amp=binned_noise.amp * factor
-            )
             noise_amp = noise_amp * interpolate_onto(
                 signal.freq, binned_noise.freq, factor
             )
+            binned_noise = log_bin(
+                signal.freq, noise_amp, f_min=f_min, f_max=f_max, n_bins=n_bins
+            )
 
         snr = binned_signal.amp / binned_noise.amp
         band = find_bandwidth(binned_signal.freq, snr, threshold, method=bandwidth)

src/specmod/pipeline.py

@@ -100,7 +100,37 @@ def estimate_spectrum(
         for key, value in dataclasses.asdict(transform).items()
         if key in accepted
     }
-    settings.update({k: v for k, v in kwargs.items() if k in accepted})
+
+    # Filtering the *configuration* to what this estimator accepts is right:
+    # `[transform]` holds every estimator's settings at once, and a CWT
+    # parameter is not an error when the FFT is selected. Filtering the
+    # caller's own keywords the same way is not. It silently discarded
+    # anything the estimator did not recognise, so a typo, or an argument
+    # meant for a different stage, simply did nothing.
+    #
+    # This is not hypothetical: `spectrum_set_from_streams(rotate_noise=False)`
+    # looks exactly like it works. `rotate_noise` belongs to `compare`, not to
+    # an estimator, so it was dropped here and the run silently kept the
+    # configured value — which cost three wrong measurements before the
+    # recorded settings gave it away.
+    unknown = sorted(set(kwargs) - accepted)
+    if unknown:
+        compare_only = sorted(set(unknown) & set(_compare_settings()))
+        hint = ""
+        if compare_only:
+            verb = "configures" if len(compare_only) == 1 else "configure"
+            this = "it" if len(compare_only) == 1 else "them"
+            hint = (
+                f" {', '.join(compare_only)} {verb} the signal-to-noise "
+                f"comparison, not the transform — pass {this} as "
+                f"compare={{{', '.join(f'{k!r}: ...' for k in compare_only)}}}."
+            )
+        raise TypeError(
+            f"{name} does not accept {', '.join(unknown)}. It takes "
+            f"{', '.join(sorted(accepted - {'self'}))}.{hint}"
+        )
+
+    settings.update(kwargs)
     result: Spectrum = cls(**settings).estimate(data, delta, motion=motion)
     return result