| """Method v1.1: response-free firing-rate-delta error controller. |
| |
| The W1 gate remains an eligibility detector. Once eligible, both direction |
| and dose come directly from the signed one-step firing-rate-delta error: |
| |
| signed_alpha = clip(alpha_gain * (delta_target - delta_current), |
| -alpha_max, alpha_max) |
| |
| No intervention-response calibration artifact is required. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
|
|
| import numpy as np |
|
|
| from analysis.native_derivative_gate import NativeDerivativeGate, NativeDerivativeGateConfig |
|
|
|
|
| @dataclass(frozen=True) |
| class NativeDerivativeErrorConfig: |
| gate: NativeDerivativeGateConfig |
| success_delta_quantiles: tuple[tuple[float, ...], ...] |
| alpha_gain: float = 10.0 |
| alpha_max: float = 10.0 |
| target_mode: str = "median" |
| error_deadband: float = 0.0 |
|
|
| def __post_init__(self) -> None: |
| if len(self.success_delta_quantiles) != len(self.gate.success_quantiles): |
| raise ValueError("success delta references must match gate windows") |
| if self.alpha_gain <= 0 or self.alpha_max <= 0: |
| raise ValueError("alpha_gain and alpha_max must be positive") |
| if self.target_mode not in {"median", "band"}: |
| raise ValueError("target_mode must be median or band") |
| if self.error_deadband < 0: |
| raise ValueError("error_deadband must be non-negative") |
|
|
|
|
| class NativeDerivativeErrorController: |
| """Proportional signed-alpha controller used by Method v1.1.""" |
|
|
| def __init__(self, config: NativeDerivativeErrorConfig): |
| self.config = config |
| self.gate = NativeDerivativeGate(config.gate) |
|
|
| def observe_decision(self, firing_rate: float) -> dict: |
| obs = self.gate.observe_decision(firing_rate) |
| window = int(obs["window_index"]) |
| current = float(obs["one_step_delta"]) |
| reference = np.asarray(self.config.success_delta_quantiles[window], dtype=np.float64) |
| reference = reference[np.isfinite(reference)] |
| invalid_control = not np.isfinite(current) |
| missing_target = reference.size == 0 |
| target = lower = upper = error = float("nan") |
| if not missing_target: |
| target = float(np.median(reference)) |
| lower = float(np.quantile(reference, 0.25)) |
| upper = float(np.quantile(reference, 0.75)) |
| if not invalid_control and not missing_target: |
| if self.config.target_mode == "band": |
| error = lower - current if current < lower else upper - current if current > upper else 0.0 |
| else: |
| error = target - current |
| if abs(error) <= self.config.error_deadband: |
| error = 0.0 |
|
|
| raw_alpha = 0.0 |
| signed_alpha = 0.0 |
| saturated = False |
| eligible = bool(obs["gate_on_next_decision"]) |
| if eligible and np.isfinite(error) and error != 0.0: |
| raw_alpha = self.config.alpha_gain * error |
| signed_alpha = float(np.clip(raw_alpha, -self.config.alpha_max, self.config.alpha_max)) |
| saturated = abs(raw_alpha) > self.config.alpha_max + 1e-12 |
| direction = "amplify" if signed_alpha > 0 else "suppress" if signed_alpha < 0 else "none" |
| return { |
| **obs, |
| "controller_method": "methods_v1_1_error_proportional", |
| "control_signal": "delta", |
| "delta_current": current, |
| "delta_target": target, |
| "delta_success_lower": lower, |
| "delta_success_upper": upper, |
| "delta_error": error, |
| "alpha_gain": float(self.config.alpha_gain), |
| "raw_signed_alpha": raw_alpha, |
| "command_direction_next_decision": direction, |
| "requested_delta_correction": abs(error) if np.isfinite(error) else 0.0, |
| "predicted_delta_correction": float("nan"), |
| "signed_alpha_next_decision": signed_alpha, |
| "controller_saturated": saturated, |
| "invalid_control_signal": invalid_control, |
| "missing_target_support": missing_target, |
| "missing_response_support": False, |
| } |
|
|
|
|
| def load_native_derivative_error_configs( |
| payload: dict, |
| *, |
| task: int, |
| layers: list[str], |
| alpha_max_override: float | None = None, |
| alpha_gain_override: float | None = None, |
| ) -> dict[str, NativeDerivativeErrorConfig]: |
| if payload.get("model") != "pi05": |
| raise ValueError("Method v1.1 requires calibration.model == 'pi05'") |
| defaults = payload.get("v1_1_controller_defaults", {}) |
| gate_defaults = payload.get("gate_defaults", {}) |
| out: dict[str, NativeDerivativeErrorConfig] = {} |
| for layer in layers: |
| matches = [ |
| row for row in payload.get("selected_features", []) |
| if row.get("layer") == layer and int(row.get("task", -1)) == int(task) |
| ] |
| if not matches: |
| raise KeyError(f"No Method v1.1 feature for task={task}, layer={layer}") |
| row = matches[0] |
| gate = NativeDerivativeGateConfig( |
| feature_id=int(row["feature_id"]), |
| success_quantiles=tuple(tuple(float(v) for v in q) for q in row["success_quantiles"]), |
| failure_quantiles=tuple(tuple(float(v) for v in q) for q in row["failure_quantiles"]), |
| window_size_env_steps=int(payload["window_size_env_steps"]), |
| replan_steps=int(payload.get("replan_steps", 5)), |
| smooth_width_decisions=int(payload["smooth_width_decisions"]), |
| derivative_width_decisions=int(payload["derivative_width_decisions"]), |
| recent_window_decisions=int(row.get("recent_window_decisions", gate_defaults.get("recent_window_decisions", 3))), |
| mode="native_derivative_dynamic", |
| theta_on=float(row.get("theta_on", gate_defaults.get("theta_on", 0.2))), |
| theta_off=float(row.get("theta_off", gate_defaults.get("theta_off", 0.05))), |
| min_on_decisions=int(row.get("min_on_decisions", gate_defaults.get("min_on_decisions", 2))), |
| min_off_decisions=int(gate_defaults.get("min_off_decisions", 2)), |
| ema_beta=float(gate_defaults.get("ema_beta", 0.5)), |
| ) |
| out[layer] = NativeDerivativeErrorConfig( |
| gate=gate, |
| success_delta_quantiles=tuple( |
| tuple(float(v) for v in q) for q in row["success_delta_quantiles"] |
| ), |
| alpha_gain=float(alpha_gain_override if alpha_gain_override is not None else defaults.get("alpha_gain", 10.0)), |
| alpha_max=float(alpha_max_override if alpha_max_override is not None else defaults.get("alpha_max", 10.0)), |
| target_mode=str(defaults.get("target_mode", "median")), |
| error_deadband=float(defaults.get("error_deadband", 0.0)), |
| ) |
| return out |
|
|
