File size: 6,809 Bytes
a529eec | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | """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
|