Autotune

The autotune subsystem provides tools for identifying unknown system parameters and discovering governing dynamics from raw data.

Why this subsystem is review-oriented

Autotune is designed as an evidence generator before controller changes, not as a direct production control channel. It turns observed traces into candidate hypotheses that humans can review against domain constraints and safety policy.

In practical usage, teams typically use it to:

• discover coupling hypotheses in previously unmapped domains,
• validate model-form assumptions before full policy activation,
• and generate bounded transfer proposals that can be replayed and compared.

The review-only boundary is intentional: it preserves explainability and prevents autonomous, opaque changes from entering live control surfaces without policy inspection.

Phase-SINDy Symbolic Discovery

The PhaseSINDy module implements Sparse Identification of Nonlinear Dynamics tailored for phase oscillator networks. It allows the orchestrator to act as an "Autonomous Physicist," reverse-engineering the differential equations of a system from observed time-series data.

Theoretical Basis

SINDy assumes that the dynamics \(\dot{\theta}\) can be represented as a sparse linear combination of terms from a library \(\Theta\):

\[ \dot{\theta} = \Theta(\theta) \Xi \]

For SPO, the library \(\Theta\) includes: 1. Constant terms: Representing natural frequencies \(\omega_i\). 2. Coupling terms: \(\sin(\theta_j - \theta_i)\) representing Kuramoto-style interactions.

The model uses Sequentially Thresholded Least Squares (STLSQ) to discover the sparsest set of coefficients that explain the data, effectively filtering out noise and revealing the underlying topology.

Use Cases

• System Identification: Discovering the coupling strength \(K_{nm}\) in a biological network where the wiring is unknown.
• Topological Verification: Verifying that a physical system actually follows the assumed Kuramoto model before engageing control logic.
• Anomaly Detection: Detecting shifts in the governing equations (e.g., a component failure that changes the interaction physics).

sindy

Sparse symbolic discovery of phase-dynamics equations from trajectories.

PhaseSINDy builds per-node trigonometric libraries, fits sparse regression coefficients, and formats discovered equations after a successful fit. Threshold and iteration counts are validated at construction, and the optional Rust path is remapped into the same Python coefficient layout. The class mutates only its own coefficients and feature-name history; it does not update live coupling state.

Classes

PhaseSINDy

PhaseSINDy(threshold: float = 0.05, max_iter: int = 10)

Symbolic Discovery of Phase Dynamics using SINDy.

Discovers the governing equations of a coupled oscillator network by performing sparse regression on a library of trigonometric interaction terms.

Create a SINDy estimator with validated sparsity controls.

Source code in src/scpn_phase_orchestrator/autotune/sindy.py

def __init__(self, threshold: float = 0.05, max_iter: int = 10):
    """Create a SINDy estimator with validated sparsity controls."""
    if _is_boolean_alias(threshold) or not isinstance(threshold, Real):
        raise ValueError("threshold must be finite and non-negative")
    parsed_threshold = float(threshold)
    if not isfinite(parsed_threshold) or parsed_threshold < 0.0:
        raise ValueError("threshold must be non-negative and finite")
    if _is_boolean_alias(max_iter) or not isinstance(max_iter, Integral):
        raise ValueError("max_iter must be an integer >= 1")
    if max_iter < 1:
        raise ValueError(f"max_iter must be >= 1, got {max_iter}")
    parsed_max_iter = int(max_iter)
    self.threshold: float = parsed_threshold
    self.max_iter: int = parsed_max_iter
    self.coefficients: list[FloatArray] = []
    self.feature_names: list[list[str]] = []

Methods:

fit

fit(phases: FloatArray, dt: float) -> list[FloatArray]

Discover equations node-by-node to handle independent coupling.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,). dt : float Integration step size.

Returns

list[FloatArray] Equations node-by-node to handle independent coupling.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/autotune/sindy.py

def fit(self, phases: FloatArray, dt: float) -> list[FloatArray]:
    """Discover equations node-by-node to handle independent coupling.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.
    dt : float
        Integration step size.

    Returns
    -------
    list[FloatArray]
        Equations node-by-node to handle independent coupling.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if _is_boolean_alias(dt) or not isinstance(dt, Real):
        raise ValueError("dt must be a finite and positive scalar")
    parsed_dt = float(dt)
    if not isfinite(parsed_dt) or parsed_dt <= 0.0:
        raise ValueError("dt must be a finite and positive scalar")

    if _contains_boolean_alias(phases):
        raise ValueError("phases must not contain boolean values")
    raw_phases = np.asarray(phases)
    if raw_phases.dtype == np.bool_:
        raise ValueError("phases must not contain boolean values")
    if np.iscomplexobj(raw_phases):
        raise ValueError("phases must be a finite 2D numeric array")

    try:
        phases_array = np.asarray(raw_phases, dtype=np.float64)
    except (TypeError, ValueError) as exc:
        raise ValueError("phases must be a finite 2D numeric array") from exc

    if phases_array.ndim != 2:
        raise ValueError("phases must be a 2D array [T, N]")

    if not np.isfinite(phases_array).all():
        raise ValueError("phases must be finite and numeric")

    T, N = phases_array.shape

    if T < 2 or N < 1:
        self.coefficients = []
        self.feature_names = []
        raise ValueError(
            "phases must contain at least two time samples and one oscillator"
        )
    if T - 1 < N:
        self.coefficients = []
        self.feature_names = []
        raise ValueError(
            "phases must provide at least one derivative sample per feature"
        )

    if _HAS_RUST:
        if _rust_sindy_fit is None:  # pragma: no cover - set when _HAS_RUST
            raise RuntimeError("Rust SINDy backend unavailable")
        p_flat = np.ascontiguousarray(phases_array, dtype=np.float64).ravel()
        result_flat = _rust_sindy_fit(
            p_flat,
            N,
            T,
            parsed_dt,
            self.threshold,
            self.max_iter,
        )
        try:
            result_flat = np.asarray(result_flat, dtype=np.float64).ravel()
        except (TypeError, ValueError) as exc:
            raise ValueError(
                "Rust SINDy returned non-numeric coefficients"
            ) from exc
        expected = N * N
        if result_flat.size != expected:
            raise ValueError(
                "Rust SINDy returned wrong number of coefficients: "
                f"{result_flat.size} != {expected}"
            )
        if not np.all(np.isfinite(result_flat)):
            raise ValueError("Rust SINDy returned non-finite coefficients")
        result = result_flat.reshape(N, N)
        # Remap: Rust stores [ω at diagonal, K_ij off-diagonal]
        # Python expects [ω, K_j1, K_j2, ...] (constant first, then j≠i)
        self.coefficients = []
        self.feature_names = []
        for i in range(N):
            xi = [result[i, i]]  # constant (ω)
            names = ["1"]
            for j in range(N):
                if j != i:
                    xi.append(result[i, j])
                    names.append(f"sin(theta_{j} - theta_{i})")
            self.coefficients.append(np.array(xi, dtype=np.float64))
            self.feature_names.append(names)
        return self.coefficients

    unwrapped = np.unwrap(phases_array, axis=0)
    theta_dot = np.diff(unwrapped, axis=0) / parsed_dt
    X = phases_array[:-1, :]

    self.coefficients = []
    self.feature_names = []

    for i in range(N):
        # 1. Build library for node i: [1, sin(theta_j - theta_i) for all j != i]
        library = [np.ones((T - 1, 1))]
        f_names = ["1"]

        for j in range(N):
            if i == j:
                continue
            diff = X[:, j] - X[:, i]
            library.append(np.sin(diff)[:, np.newaxis])
            f_names.append(f"sin(theta_{j} - theta_{i})")

        Theta = np.hstack(library)

        # 2. STLSQ for this node
        xi = _coerce_lstsq_coefficients(
            lstsq(Theta, theta_dot[:, i])[0],
            Theta.shape[1],
        )

        for _ in range(self.max_iter):
            small_indices = np.abs(xi) < self.threshold
            xi[small_indices] = 0
            big_indices = ~small_indices
            if np.any(big_indices):
                xi[big_indices] = _coerce_lstsq_coefficients(
                    lstsq(Theta[:, big_indices], theta_dot[:, i])[0],
                    int(np.count_nonzero(big_indices)),
                )

        self.coefficients.append(xi)
        self.feature_names.append(f_names)

    return self.coefficients

get_equations

get_equations() -> list[str]

Format fitted sparse coefficients as per-node phase equations.

Returns

list[str] Format fitted sparse coefficients as per-node phase equations.

Raises

RuntimeError If the operation fails.

Source code in src/scpn_phase_orchestrator/autotune/sindy.py

def get_equations(self) -> list[str]:
    """Format fitted sparse coefficients as per-node phase equations.

    Returns
    -------
    list[str]
        Format fitted sparse coefficients as per-node phase equations.

    Raises
    ------
    RuntimeError
        If the operation fails.
    """
    if not self.coefficients:
        raise RuntimeError("PhaseSINDy.get_equations() called before fit()")
    equations = []
    for i, xi in enumerate(self.coefficients):
        terms = []
        for j, val in enumerate(xi):
            if abs(val) > 1e-6:
                terms.append(f"{val:.4f} * {self.feature_names[i][j]}")
        equations.append(
            f"d(theta_{i})/dt = " + (" + ".join(terms) if terms else "0")
        )
    return equations

Frequency Identification

Identifies natural frequencies \(\omega_i\) from phase time-series.

The dedicated frequency-identification reference page owns the full mkdocstrings inventory for scpn_phase_orchestrator.autotune.freq_id. This aggregate page links to that surface instead of declaring a second primary mkdocstrings target for the same dataclasses.

See Frequency Identification.

Coupling Estimation

Estimates the coupling matrix \(K_{nm}\) assuming a fixed interaction model.

coupling_est

Least-squares coupling estimators for observed phase trajectories.

The primary estimator fits pairwise sinusoidal Kuramoto coupling from phase-history derivatives and natural frequencies, returning a dense matrix with zero diagonal. The harmonics variant expands the regression library with higher Fourier sine and cosine terms. Both routines are offline inference helpers: they estimate parameters from caller-provided arrays and perform no runtime actuation or binding updates.

Functions:

estimate_coupling

estimate_coupling(
    phases: FloatArray, omegas: FloatArray, dt: float
) -> FloatArray

Estimate K_ij coupling matrix from observed phase trajectories.

Least-squares fit of dθ_i/dt - ω_i = Σ_j K_ij sin(θ_j - θ_i). Constructs the regression matrix from pairwise sin(Δθ) and solves for K_ij via pseudoinverse.

Parameters

phases : FloatArray (n_oscillators, n_timesteps) phase trajectories. omegas : FloatArray (n_oscillators,) natural frequencies. dt : float timestep between samples.

Returns

FloatArray (n_oscillators, n_oscillators) estimated coupling matrix K_ij.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/autotune/coupling_est.py

def estimate_coupling(
    phases: FloatArray,
    omegas: FloatArray,
    dt: float,
) -> FloatArray:
    """Estimate K_ij coupling matrix from observed phase trajectories.

    Least-squares fit of dθ_i/dt - ω_i = Σ_j K_ij sin(θ_j - θ_i).
    Constructs the regression matrix from pairwise sin(Δθ) and solves
    for K_ij via pseudoinverse.

    Parameters
    ----------
    phases : FloatArray
        (n_oscillators, n_timesteps) phase trajectories.
    omegas : FloatArray
        (n_oscillators,) natural frequencies.
    dt : float
        timestep between samples.

    Returns
    -------
    FloatArray
        (n_oscillators, n_oscillators) estimated coupling matrix K_ij.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    phases, omegas, dt = _validate_inputs(phases, omegas, dt)
    n, T = phases.shape
    if T < 3:
        raise ValueError(f"Need >= 3 timesteps, got {T}")

    # Phase derivative: (dθ/dt)_i ≈ (θ_{t+1} - θ_{t-1}) / (2*dt)
    dphase = np.diff(np.unwrap(phases, axis=1), axis=1) / dt
    # Use interior points for derivative
    phases_mid = phases[:, :-1]
    T_eff = dphase.shape[1]

    knm = np.zeros((n, n), dtype=np.float64)

    for i in range(n):
        # Target: dθ_i/dt - ω_i at each timestep
        target = dphase[i, :] - omegas[i]

        # Regressor: sin(θ_j - θ_i) for each j, at each timestep
        regressors = np.sin(
            phases_mid[:, :T_eff] - phases_mid[i : i + 1, :T_eff]
        )  # (n, T_eff)

        # Least squares: target = K_i · regressors
        # K_i = target @ regressors^T @ (regressors @ regressors^T)^{-1}
        with contextlib.suppress(np.linalg.LinAlgError):
            coeffs = np.linalg.lstsq(regressors.T, target, rcond=None)[0]
            if np.all(np.isfinite(coeffs)):
                knm[i, :] = coeffs

    np.fill_diagonal(knm, 0.0)
    return knm

estimate_coupling_harmonics

estimate_coupling_harmonics(
    phases: FloatArray,
    omegas: FloatArray,
    dt: float,
    n_harmonics: int = 2,
) -> dict[str, FloatArray]

Estimate coupling with higher Fourier harmonics.

Fits: dθ_i/dt - ω_i = Σ_j Σ_k [a_jk sin(k·Δθ) + b_jk cos(k·Δθ)] for k = 1..n_harmonics.

Real biological oscillators have non-sinusoidal coupling (Stankovski 2017, Rev. Mod. Phys.).

Returns dict with keys 'sin_1', 'cos_1', 'sin_2', 'cos_2', ... each an (n, n) matrix of coefficients.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,). omegas : FloatArray Natural frequencies in rad/s, shape (N,). dt : float Integration step size. n_harmonics : int Number of harmonics to fit.

Returns

dict[str, FloatArray] Coupling with higher Fourier harmonics.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/autotune/coupling_est.py

def estimate_coupling_harmonics(
    phases: FloatArray,
    omegas: FloatArray,
    dt: float,
    n_harmonics: int = 2,
) -> dict[str, FloatArray]:
    """Estimate coupling with higher Fourier harmonics.

    Fits: dθ_i/dt - ω_i = Σ_j Σ_k [a_jk sin(k·Δθ) + b_jk cos(k·Δθ)]
    for k = 1..n_harmonics.

    Real biological oscillators have non-sinusoidal coupling
    (Stankovski 2017, Rev. Mod. Phys.).

    Returns dict with keys 'sin_1', 'cos_1', 'sin_2', 'cos_2', ...
    each an (n, n) matrix of coefficients.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.
    dt : float
        Integration step size.
    n_harmonics : int
        Number of harmonics to fit.

    Returns
    -------
    dict[str, FloatArray]
        Coupling with higher Fourier harmonics.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    phases, omegas, dt = _validate_inputs(phases, omegas, dt)
    n_harmonics = _validate_n_harmonics(n_harmonics)
    n, T = phases.shape
    if T < 3:
        raise ValueError(f"Need >= 3 timesteps, got {T}")

    dphase = np.diff(np.unwrap(phases, axis=1), axis=1) / dt
    phases_mid = phases[:, :-1]
    T_eff = dphase.shape[1]

    result: dict[str, FloatArray] = {}
    for k in range(1, n_harmonics + 1):
        result[f"sin_{k}"] = np.zeros((n, n), dtype=np.float64)
        result[f"cos_{k}"] = np.zeros((n, n), dtype=np.float64)

    for i in range(n):
        target = dphase[i, :] - omegas[i]
        diff = phases_mid[:, :T_eff] - phases_mid[i : i + 1, :T_eff]

        # Build regressor matrix: [sin(Δθ), cos(Δθ), sin(2Δθ), cos(2Δθ), ...]
        blocks = []
        for k in range(1, n_harmonics + 1):
            blocks.append(np.sin(k * diff))
            blocks.append(np.cos(k * diff))
        regressors = np.vstack(blocks)  # (2*n_harmonics*n, T_eff)

        try:
            coeffs = np.linalg.lstsq(regressors.T, target, rcond=None)[0]
        except np.linalg.LinAlgError:
            continue
        if not np.all(np.isfinite(coeffs)):
            continue

        # Unpack coefficients
        idx = 0
        for k in range(1, n_harmonics + 1):
            result[f"sin_{k}"][i, :] = coeffs[idx : idx + n]
            idx += n
            result[f"cos_{k}"][i, :] = coeffs[idx : idx + n]
            idx += n

    for key in result:
        np.fill_diagonal(result[key], 0.0)

    return result

End-to-End Pipeline

The pipeline module composes phase extraction, frequency identification, SINDy-style discovery, and coupling estimation into reviewable auto-binding candidate records.

pipeline

Offline auto-tune pipeline from raw channels to inferred coupling settings.

identify_binding_spec extracts per-channel phases and dominant frequencies, estimates a non-negative coupling matrix, initializes zero phase lags, and asks the universal prior for a critical-coupling estimate. The result is an AutoTuneResult for review or downstream proposal generation; the function does not write a binding file, change runtime configuration, or activate the inferred parameters.

Classes

AutoTuneResult dataclass

AutoTuneResult(
    omegas: list[float],
    knm: FloatArray,
    alpha: FloatArray,
    n_layers: int,
    dominant_freqs: list[float],
    K_c_estimate: float,
)

Output of the auto-tune pipeline: inferred frequencies and coupling.

Functions:

identify_binding_spec

identify_binding_spec(
    time_series: FloatArray,
    fs: float,
    n_layers: int | None = None,
) -> AutoTuneResult

Full auto-tune pipeline: raw multichannel data → coupling parameters.

1. Phase extraction (Hilbert) per channel → ω_i
2. Coupling estimation (least squares) → K_ij
3. K_c estimate from universal prior

Parameters

time_series : FloatArray (n_channels, n_samples) raw data. fs : float sampling frequency in Hz. n_layers : int | None override number of layers (default: n_channels).

Returns

AutoTuneResult The result.

Raises

ValueError If the inputs are invalid or inconsistent. TypeError If an argument has the wrong type.

Source code in src/scpn_phase_orchestrator/autotune/pipeline.py

def identify_binding_spec(
    time_series: FloatArray,
    fs: float,
    n_layers: int | None = None,
) -> AutoTuneResult:
    """Full auto-tune pipeline: raw multichannel data → coupling parameters.

    1. Phase extraction (Hilbert) per channel → ω_i
    2. Coupling estimation (least squares) → K_ij
    3. K_c estimate from universal prior

    Parameters
    ----------
    time_series : FloatArray
        (n_channels, n_samples) raw data.
    fs : float
        sampling frequency in Hz.
    n_layers : int | None
        override number of layers (default: n_channels).

    Returns
    -------
    AutoTuneResult
        The result.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    TypeError
        If an argument has the wrong type.
    """
    sample_rate = _positive_real(fs, "fs")
    data = np.atleast_2d(_real_time_series(time_series))
    n_ch, n_t = data.shape
    if n_t < 4:
        raise ValueError(f"time_series needs >= 4 samples, got {n_t}")
    if not np.all(np.isfinite(data)):
        raise ValueError("time_series must contain only finite values")
    if not np.any(np.abs(data - data.mean(axis=1, keepdims=True)) > 0.0):
        raise ValueError("time_series must contain non-zero temporal dynamics")

    if n_layers is None:
        n_layers = n_ch
    elif isinstance(n_layers, bool) or not isinstance(n_layers, Integral):
        raise TypeError(f"n_layers must be an integer, got {n_layers!r}")
    elif n_layers <= 0:
        raise ValueError("n_layers must be a positive integer")
    else:
        n_layers = int(n_layers)

    # Step 1: extract phases and frequencies per channel
    channel_phases = []
    dominant_freqs = []
    for ch in range(n_ch):
        pr = extract_phases(data[ch], sample_rate)
        channel_phases.append(pr.phases)
        dominant_freqs.append(pr.dominant_freq)

    omegas = [2 * np.pi * f for f in dominant_freqs]

    # Step 2: estimate coupling from phase trajectories
    phase_matrix = np.array(channel_phases)
    dt = 1.0 / sample_rate
    knm = estimate_coupling(phase_matrix, np.array(omegas), dt)

    # Ensure non-negative coupling
    knm = np.maximum(knm, 0.0)
    np.fill_diagonal(knm, 0.0)

    alpha = np.zeros((n_ch, n_ch), dtype=np.float64)

    # Step 3: K_c from universal prior
    prior = UniversalPrior()
    kc_result = prior.estimate_Kc(np.array(omegas), n_ch)

    return AutoTuneResult(
        omegas=omegas,
        knm=knm,
        alpha=alpha,
        n_layers=n_layers,
        dominant_freqs=dominant_freqs,
        K_c_estimate=kc_result.K_c_estimate,
    )

Reviewable Binding Proposals

The binding-proposal module converts time-series CSV, event-log JSON, and graph JSON payloads into StudioProjectState records containing reviewable binding_spec.yaml text, confidence factors, provenance, and binding-validator diagnostics.

binding_proposal

Review-only binding proposal builders for CSV, event-log, and graph inputs.

The module converts imported source text into StudioProjectState proposals with provenance, confidence factors, validator output, and inferred channel assignments. CSV samples are checked for numeric finite channels, event logs for event structure, and graph payloads for node/edge integrity before YAML is generated. The functions never activate bindings or mutate runtime state; they prepare operator-review artifacts.

Classes

Functions:

propose_binding_from_time_series_csv

propose_binding_from_time_series_csv(
    csv_text: str,
    *,
    sample_rate_hz: float | None,
    project_name: str,
    sindy_options: SindyOptions | None = None,
) -> StudioProjectState

Propose a review-only binding for a tabular time-series replay.

Parameters

csv_text : str Raw CSV text. sample_rate_hz : float | None Sampling rate in Hz. project_name : str Name of the project. sindy_options : SindyOptions | None Operator options for phase-SINDy discovery — the sparsity threshold and the confidence policy. Defaults to the conservative shared options.

Returns

StudioProjectState A review-only binding for a tabular time-series replay.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/autotune/binding_proposal.py

def propose_binding_from_time_series_csv(
    csv_text: str,
    *,
    sample_rate_hz: float | None,
    project_name: str,
    sindy_options: SindyOptions | None = None,
) -> StudioProjectState:
    """Propose a review-only binding for a tabular time-series replay.

    Parameters
    ----------
    csv_text : str
        Raw CSV text.
    sample_rate_hz : float | None
        Sampling rate in Hz.
    project_name : str
        Name of the project.
    sindy_options : SindyOptions | None
        Operator options for phase-SINDy discovery — the sparsity threshold and
        the confidence policy. Defaults to the conservative shared options.

    Returns
    -------
    StudioProjectState
        A review-only binding for a tabular time-series replay.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    options = sindy_options or DEFAULT_SINDY_OPTIONS
    payload = csv_text.encode("utf-8")
    reader = csv.DictReader(io.StringIO(csv_text))
    if not reader.fieldnames:
        raise ValueError("CSV header is required")

    rows = list(reader)
    time_columns = {"time", "timestamp", "t"}
    channels = tuple(
        field
        for field in reader.fieldnames
        if field.strip().lower() not in time_columns
    )
    if not channels:
        raise ValueError("CSV must contain at least one signal channel")
    if not rows:
        raise ValueError("CSV must contain at least one sample")
    signal_table = _numeric_signal_table(rows, channels)

    inferred_channels = _inferred_channels(len(channels), prefer_event=False)
    resolved_sample_rate_hz, sample_rate_inference = _resolve_sample_rate_hz(
        sample_rate_hz,
        rows=rows,
        fieldnames=reader.fieldnames,
    )
    sample_period_s = 1.0 / resolved_sample_rate_hz
    discovery = discover_time_series_structure(
        signal_table,
        columns=channels,
        sample_period_s=sample_period_s,
        config=options.to_discovery_config(),
    )
    discovered_dynamics = discovery.discovered_dynamics(
        policy=options.confidence_policy,
    )
    family_specs = _families_for_time_series(
        channels=channels,
        inferred_channels=inferred_channels,
        signal_table=signal_table,
        sample_rate_hz=resolved_sample_rate_hz,
        sample_period_s=sample_period_s,
    )
    extractor_parameter_proposals = _extractor_parameter_proposals(family_specs)
    initial_coupling_proposal = _initial_coupling_proposal(
        discovery=discovery.to_audit_record(),
        inferred_channels=inferred_channels,
    )
    yaml_text = _binding_yaml(
        project_name=project_name,
        sample_period_s=sample_period_s,
        family_specs=family_specs,
        initial_coupling_proposal=initial_coupling_proposal,
    )
    validation_errors = _validation_errors(yaml_text)
    confidence: dict[str, float] = {
        "phase_quality": _bounded_confidence(min(1.0, len(rows) / 3.0)),
        "channel_coverage": _bounded_confidence(min(1.0, len(channels) / 2.0)),
        "validator_acceptance": 1.0 if not validation_errors else 0.0,
    }
    confidence.update(
        {
            name: _bounded_confidence(value)
            for name, value in discovery.confidence_evidence.items()
        }
    )
    source_columns: list[JsonValue] = []
    source_columns.extend(channels)
    provenance: dict[str, JsonValue] = {
        "input_family": "time_series",
        "sample_rate_hz": float(resolved_sample_rate_hz),
        "sample_rate_inference": sample_rate_inference,
        "source_columns": source_columns,
        "extractor_parameter_proposals": extractor_parameter_proposals,
        "initial_coupling_proposal": initial_coupling_proposal,
        "discovery_evidence": discovery.to_audit_record(),
        "discovered_dynamics": discovered_dynamics.to_audit_record(),
        "sindy_options": {
            "phase_sindy_threshold": options.phase_sindy_threshold,
            "confidence_policy": {
                "min_r_squared": options.confidence_policy.min_r_squared,
                "min_samples_per_parameter": (
                    options.confidence_policy.min_samples_per_parameter
                ),
            },
        },
        "validator": "load_binding_spec+validate_binding_spec",
    }
    source = ImportedSourceSummary.from_payload(
        source_kind="time_series_csv",
        payload=payload,
        channel_count=len(channels),
        sample_count=len(rows),
    )
    return _project_state(
        project_name=project_name,
        source=source,
        binding=BindingProposal(
            yaml_text=yaml_text,
            validation_errors=validation_errors,
            inferred_channels=inferred_channels,
            confidence_factors=confidence,
            provenance=provenance,
        ),
    )

propose_binding_from_event_log

propose_binding_from_event_log(
    json_text: str, *, project_name: str
) -> StudioProjectState

Propose a review-only binding for a JSON event log.

Parameters

json_text : str Raw JSON text. project_name : str Name of the project.

Returns

StudioProjectState A review-only binding for a JSON event log.

Source code in src/scpn_phase_orchestrator/autotune/binding_proposal.py

def propose_binding_from_event_log(
    json_text: str,
    *,
    project_name: str,
) -> StudioProjectState:
    """Propose a review-only binding for a JSON event log.

    Parameters
    ----------
    json_text : str
        Raw JSON text.
    project_name : str
        Name of the project.

    Returns
    -------
    StudioProjectState
        A review-only binding for a JSON event log.
    """
    payload = json_text.encode("utf-8")
    events = _event_sequence(json.loads(json_text))
    source_names = sorted(
        {
            str(event.get("source"))
            for event in events
            if isinstance(event.get("source"), str) and event.get("source")
        }
    )
    times = [
        float(event["time"])
        for event in events
        if isinstance(event.get("time"), int | float)
        and not isinstance(event.get("time"), bool)
    ]
    span_s = max(times) - min(times) if len(times) >= 2 else 0.0
    yaml_text = _binding_yaml(
        project_name=project_name,
        sample_period_s=1.0,
        family_specs=(("event_log", "I", "event"),),
    )
    validation_errors = _validation_errors(yaml_text)
    event_density = _bounded_confidence(min(1.0, len(events) / 10.0))
    source_name_values: list[JsonValue] = []
    source_name_values.extend(source_names)
    provenance: dict[str, JsonValue] = {
        "input_family": "event_log",
        "event_count": len(events),
        "source_names": source_name_values,
        "time_span_s": span_s,
        "validator": "load_binding_spec+validate_binding_spec",
    }
    source = _EventLogSourceSummary.from_payload(
        source_kind="event_log_json",
        payload=payload,
        channel_count=max(1, len(source_names)),
        sample_count=len(events),
    )
    return _project_state(
        project_name=project_name,
        source=source,
        binding=BindingProposal(
            yaml_text=yaml_text,
            validation_errors=validation_errors,
            inferred_channels=("I",),
            confidence_factors={
                "event_density": event_density,
                "source_diversity": _bounded_confidence(
                    min(1.0, len(source_names) / 3.0)
                ),
                "validator_acceptance": 1.0 if not validation_errors else 0.0,
            },
            provenance=provenance,
        ),
    )

propose_binding_from_graph

propose_binding_from_graph(
    json_text: str, *, project_name: str
) -> StudioProjectState

Propose a review-only binding for a graph JSON payload.

Parameters

json_text : str Raw JSON text. project_name : str Name of the project.

Returns

StudioProjectState A review-only binding for a graph JSON payload.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/autotune/binding_proposal.py

def propose_binding_from_graph(
    json_text: str,
    *,
    project_name: str,
) -> StudioProjectState:
    """Propose a review-only binding for a graph JSON payload.

    Parameters
    ----------
    json_text : str
        Raw JSON text.
    project_name : str
        Name of the project.

    Returns
    -------
    StudioProjectState
        A review-only binding for a graph JSON payload.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    payload = json_text.encode("utf-8")
    graph = _mapping(json.loads(json_text), "graph")
    nodes = _sequence(graph.get("nodes"), "graph.nodes")
    edges = _sequence(graph.get("edges", ()), "graph.edges")
    node_ids = {_node_id(node) for node in nodes}
    if not node_ids:
        raise ValueError("graph must contain at least one node")
    for edge in edges:
        edge_map = _mapping(edge, "graph.edges[]")
        edge_source = str(edge_map.get("source", ""))
        target = str(edge_map.get("target", ""))
        missing = sorted({edge_source, target} - node_ids)
        if missing:
            raise ValueError(f"unknown graph node in edge: {missing[0]}")

    yaml_text = _binding_yaml(
        project_name=project_name,
        sample_period_s=1.0,
        family_specs=(("graph_topology", "S", "graph"),),
    )
    validation_errors = _validation_errors(yaml_text)
    source = ImportedSourceSummary.from_payload(
        source_kind="graph_json",
        payload=payload,
        channel_count=1,
        sample_count=len(nodes),
    )
    return _project_state(
        project_name=project_name,
        source=source,
        binding=BindingProposal(
            yaml_text=yaml_text,
            validation_errors=validation_errors,
            inferred_channels=("S",),
            confidence_factors={
                "topology_integrity": 1.0,
                "edge_density": _bounded_confidence(
                    min(1.0, len(edges) / len(node_ids))
                ),
                "validator_acceptance": 1.0 if not validation_errors else 0.0,
            },
            provenance={
                "input_family": "graph",
                "node_count": len(node_ids),
                "edge_count": len(edges),
                "validator": "load_binding_spec+validate_binding_spec",
            },
        ),
    )

Time-Series Discovery Evidence

The discovery module extracts deterministic review evidence from raw time-series tables: sparse derivative regressions, phase-aware Kuramoto SINDy fits for phase-like columns, residual-scored SINDy library selection, correlation graph edges, lagged directed graph inference, connected-component clusters, and regular time-column sample-rate inference. Non-phase data carries an explicit phase-SINDy skipped status. The reports are JSON-ready provenance for binding review and do not promote actuation.

discovery

Deterministic evidence extraction for review-only auto-binding proposals.

Classes

TimeSeriesDiscoveryConfig dataclass

TimeSeriesDiscoveryConfig(
    correlation_threshold: float = 0.75,
    sindy_threshold: float = 0.05,
    phase_sindy_threshold: float = 0.05,
    learned_graph_threshold: float = 0.2,
)

Configuration for deterministic review evidence extraction.

Methods:

__post_init__

__post_init__() -> None

Validate and canonicalise scalar discovery thresholds.

Source code in src/scpn_phase_orchestrator/autotune/discovery.py

def __post_init__(self) -> None:
    """Validate and canonicalise scalar discovery thresholds."""
    correlation_threshold = _finite_real_scalar(
        self.correlation_threshold,
        "correlation_threshold",
    )
    sindy_threshold = _finite_real_scalar(self.sindy_threshold, "sindy_threshold")
    phase_sindy_threshold = _finite_real_scalar(
        self.phase_sindy_threshold,
        "phase_sindy_threshold",
    )
    learned_graph_threshold = _finite_real_scalar(
        self.learned_graph_threshold,
        "learned_graph_threshold",
    )
    if not 0.0 <= correlation_threshold <= 1.0:
        raise ValueError("correlation_threshold must be in [0, 1]")
    if sindy_threshold < 0.0:
        raise ValueError("sindy_threshold must be finite and non-negative")
    if phase_sindy_threshold < 0.0:
        raise ValueError("phase_sindy_threshold must be finite and non-negative")
    if learned_graph_threshold < 0.0:
        raise ValueError("learned_graph_threshold must be finite and non-negative")
    object.__setattr__(self, "correlation_threshold", correlation_threshold)
    object.__setattr__(self, "sindy_threshold", sindy_threshold)
    object.__setattr__(self, "phase_sindy_threshold", phase_sindy_threshold)
    object.__setattr__(self, "learned_graph_threshold", learned_graph_threshold)

TimeSeriesDiscoveryReport dataclass

TimeSeriesDiscoveryReport(
    sample_period_s: float,
    sample_count: int,
    columns: tuple[str, ...],
    sindy: Mapping[str, JsonValue],
    phase_sindy: Mapping[str, JsonValue],
    sindy_model_selection: Mapping[str, JsonValue],
    learned_graph: Mapping[str, JsonValue],
    correlation_graph: Mapping[str, JsonValue],
    clustering: Mapping[str, JsonValue],
)

JSON-ready discovery report for an imported time-series table.

Attributes

sindy_sparsity property

sindy_sparsity: float

Sparse-regression support fraction reported by the SINDy evidence.

Returns

float Sparse-regression support fraction reported by the SINDy evidence.

correlation_graph_density property

correlation_graph_density: float

Density of the thresholded correlation graph evidence.

Returns

float Density of the thresholded correlation graph evidence.

cluster_coverage property

cluster_coverage: float

Fraction of channels covered by the largest discovered cluster.

Returns

float Fraction of channels covered by the largest discovered cluster.

confidence_evidence property

confidence_evidence: dict[str, float]

Confidence factors derived from fitted discovery evidence blocks.

Returns

dict[str, float] Confidence factors derived from fitted discovery evidence blocks.

phase_sindy_confidence property

phase_sindy_confidence: SindyConfidence

Honest tier and discovery posture for the phase-SINDy fit.

Returns

SindyConfidence The confidence verdict; a self-fit is capped at the partial tier and can never be externally_validated.

Methods:

discovered_dynamics

discovered_dynamics(
    *,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> DiscoveredDynamics

Return the operator-facing discovered-dynamics record.

Parameters

policy : SindyConfidencePolicy, optional Thresholds separating a credible discovery from weak evidence.

Returns

DiscoveredDynamics The recovered equations and coupling edges paired with the honest confidence verdict and a provenance hash.

Source code in src/scpn_phase_orchestrator/autotune/discovery.py

def discovered_dynamics(
    self,
    *,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> DiscoveredDynamics:
    """Return the operator-facing discovered-dynamics record.

    Parameters
    ----------
    policy : SindyConfidencePolicy, optional
        Thresholds separating a credible discovery from weak evidence.

    Returns
    -------
    DiscoveredDynamics
        The recovered equations and coupling edges paired with the honest
        confidence verdict and a provenance hash.
    """
    return discovered_dynamics_from_block(self.phase_sindy, policy=policy)

to_audit_record

to_audit_record() -> dict[str, JsonValue]

Return the complete JSON-safe discovery evidence record.

Returns

dict[str, JsonValue] The complete JSON-safe discovery evidence record.

Source code in src/scpn_phase_orchestrator/autotune/discovery.py

def to_audit_record(self) -> dict[str, JsonValue]:
    """Return the complete JSON-safe discovery evidence record.

    Returns
    -------
    dict[str, JsonValue]
        The complete JSON-safe discovery evidence record.
    """
    return {
        "sample_period_s": self.sample_period_s,
        "sample_count": self.sample_count,
        "columns": list(self.columns),
        "sindy": dict(self.sindy),
        "phase_sindy": dict(self.phase_sindy),
        "phase_sindy_confidence": self.phase_sindy_confidence.to_audit_record(),
        "sindy_model_selection": dict(self.sindy_model_selection),
        "learned_graph": dict(self.learned_graph),
        "correlation_graph": dict(self.correlation_graph),
        "clustering": dict(self.clustering),
    }

Functions:

infer_sample_rate_from_time_column

infer_sample_rate_from_time_column(
    rows: Sequence[Mapping[str, str]],
    fieldnames: Sequence[str],
) -> tuple[float, str]

Infer a sampling rate from a regular finite time column.

Parameters

rows : Sequence[Mapping[str, str]] Data rows. fieldnames : Sequence[str] CSV field names.

Returns

tuple[float, str] A sampling rate from a regular finite time column.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/autotune/discovery.py

def infer_sample_rate_from_time_column(
    rows: Sequence[Mapping[str, str]],
    fieldnames: Sequence[str],
) -> tuple[float, str]:
    """Infer a sampling rate from a regular finite time column.

    Parameters
    ----------
    rows : Sequence[Mapping[str, str]]
        Data rows.
    fieldnames : Sequence[str]
        CSV field names.

    Returns
    -------
    tuple[float, str]
        A sampling rate from a regular finite time column.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if any(not isinstance(field, str) for field in fieldnames):
        raise ValueError("fieldnames must be strings")
    time_column = next(
        (field for field in fieldnames if field.strip().lower() in _TIME_COLUMNS),
        None,
    )
    if time_column is None:
        raise ValueError("sample_rate_hz is required when CSV has no time column")
    if len(rows) < 2:
        raise ValueError("sample_rate_hz requires at least two timed samples")
    times: list[float] = []
    for row_index, row in enumerate(rows):
        try:
            value = _finite_time_sample(
                row[time_column],
                f"time column {time_column!r} sample",
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise ValueError(
                f"time column {time_column!r} has non-numeric sample at row {row_index}"
            ) from exc
        times.append(value)
    deltas = np.diff(np.asarray(times, dtype=np.float64))
    if np.any(deltas <= 0.0):
        raise ValueError("time column must be strictly increasing")
    sample_period_s = float(np.median(deltas))
    if not np.allclose(deltas, sample_period_s, rtol=1e-6, atol=1e-12):
        raise ValueError("time column must use a regular sampling interval")
    if sample_period_s <= 0.0 or not isfinite(sample_period_s):
        raise ValueError("sample_rate_hz could not be inferred from time column")
    return 1.0 / sample_period_s, "time_column"

discover_time_series_structure

discover_time_series_structure(
    samples: FloatArray,
    *,
    columns: Sequence[str],
    sample_period_s: float,
    config: TimeSeriesDiscoveryConfig | None = None,
) -> TimeSeriesDiscoveryReport

Extract sparse-derivative, graph, and cluster evidence from a table.

Parameters

samples : FloatArray Sample array. columns : Sequence[str] Column names. sample_period_s : float Sample period in seconds. config : TimeSeriesDiscoveryConfig | None The configuration object.

Returns

TimeSeriesDiscoveryReport Sparse-derivative, graph, and cluster evidence from a table.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/autotune/discovery.py

def discover_time_series_structure(
    samples: FloatArray,
    *,
    columns: Sequence[str],
    sample_period_s: float,
    config: TimeSeriesDiscoveryConfig | None = None,
) -> TimeSeriesDiscoveryReport:
    """Extract sparse-derivative, graph, and cluster evidence from a table.

    Parameters
    ----------
    samples : FloatArray
        Sample array.
    columns : Sequence[str]
        Column names.
    sample_period_s : float
        Sample period in seconds.
    config : TimeSeriesDiscoveryConfig | None
        The configuration object.

    Returns
    -------
    TimeSeriesDiscoveryReport
        Sparse-derivative, graph, and cluster evidence from a table.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    cfg = config or TimeSeriesDiscoveryConfig()
    sample_period_s = _finite_real_scalar(sample_period_s, "sample_period_s")
    table = _real_table(samples)
    if table.ndim != 2:
        raise ValueError("samples must be a 2-D table")
    if table.shape[0] < 2:
        raise ValueError("samples must contain at least two rows")
    if table.shape[1] != len(columns):
        raise ValueError("column count must match samples width")
    if table.shape[1] < 1:
        raise ValueError("samples must contain at least one signal column")
    if sample_period_s <= 0.0:
        raise ValueError("sample_period_s must be positive")
    column_names = tuple(_normalised_column_name(column) for column in columns)
    correlation_graph = _correlation_graph(
        table,
        column_names,
        threshold=cfg.correlation_threshold,
    )
    clustering = _correlation_clusters(
        column_names,
        edges=correlation_graph["edges"],
    )
    sindy = _sparse_derivative_library(
        table,
        column_names,
        sample_period_s=sample_period_s,
        threshold=cfg.sindy_threshold,
    )
    phase_sindy = _phase_sindy_library(
        table,
        column_names,
        sample_period_s=sample_period_s,
        threshold=cfg.phase_sindy_threshold,
    )
    sindy_model_selection = _sindy_model_selection(
        sindy=sindy,
        phase_sindy=phase_sindy,
    )
    learned_graph = _lagged_learned_graph(
        table,
        column_names,
        threshold=cfg.learned_graph_threshold,
    )
    return TimeSeriesDiscoveryReport(
        sample_period_s=sample_period_s,
        sample_count=int(table.shape[0]),
        columns=column_names,
        sindy=sindy,
        phase_sindy=phase_sindy,
        sindy_model_selection=sindy_model_selection,
        learned_graph=learned_graph,
        correlation_graph=correlation_graph,
        clustering=clustering,
    )

Phase-SINDy Discovery Confidence

The confidence module classifies a phase-SINDy fit into an honest validation tier and a discovery posture. A fit on the operator's own data is self-consistency, not independent validation, so the classifier cannot award the externally_validated tier: its ceiling is partial and its default is scaffold. The posture is discovered only for a well-determined fit that explains the derivative variance, and otherwise insufficient_evidence or refused, each with human-readable reasons.

sindy_confidence

Honest confidence classification for phase-SINDy discovery.

A phase-SINDy fit recovers a Kuramoto-style coupling structure from the operator's own time series. Fitting a model to the same data it was learnt from is self-consistency, not independent validation, so this classifier is built so that it cannot award the externally_validated tier — that tier is reserved for clearing an independent-reference test on real data, which a self-fit can never do. The ceiling here is partial; the honest default is scaffold.

The classifier is pure: it reads the numeric summary of a fit (its status, R², sample and node counts, and term counts) and returns a tier plus a discovery posture with human-readable reasons. It performs no I/O, no fitting, and no mutation, so it is trivially testable and deterministic.

Postures

discovered A fit was performed, explains the data well (R² at or above the policy threshold), is well determined (enough derivative samples per parameter), and selected at least one active term. Tier partial. insufficient_evidence A fit was performed but the evidence is too weak to stand behind the recovered structure (poor R², under-determined, or no active terms). Tier scaffold. refused No fit was performed at all (the discovery step skipped this library). Tier scaffold.

Classes

SindyConfidencePolicy dataclass

SindyConfidencePolicy(
    min_r_squared: float = 0.9,
    min_samples_per_parameter: float = 5.0,
)

Thresholds that separate a credible discovery from weak evidence.

Parameters

min_r_squared : float Smallest coefficient of determination a fit must reach before its recovered structure may be called discovered. The default of 0.9 demands the model explain the large majority of the derivative variance. min_samples_per_parameter : float Smallest ratio of regressed derivative samples to per-node parameters a fit must reach before it is considered well determined. The default of 5.0 keeps the per-node regression comfortably over-determined.

SindyConfidence dataclass

SindyConfidence(
    tier: str,
    posture: str,
    r_squared: float | None,
    samples_per_parameter: float | None,
    reasons: tuple[str, ...] = tuple(),
)

The honest confidence verdict for a single phase-SINDy fit.

Parameters

tier : str Validation tier, drawn from the canonical vocabulary. Never externally_validated — a self-fit cannot earn it. posture : str Discovery posture: discovered, insufficient_evidence or refused. r_squared : float or None The scale-free fit quality the verdict was based on, or None when no fit was performed. samples_per_parameter : float or None Regressed derivative samples per per-node parameter, or None when no fit was performed or the parameter count was unknown. reasons : tuple of str Human-readable justifications for the verdict, in evaluation order.

Methods:

to_audit_record

to_audit_record() -> dict[str, Any]

Return the JSON-safe confidence record.

Returns

dict A JSON-serialisable mapping of the verdict fields.

Source code in src/scpn_phase_orchestrator/autotune/sindy_confidence.py

def to_audit_record(self) -> dict[str, Any]:
    """Return the JSON-safe confidence record.

    Returns
    -------
    dict
        A JSON-serialisable mapping of the verdict fields.
    """
    return {
        "tier": self.tier,
        "posture": self.posture,
        "r_squared": self.r_squared,
        "samples_per_parameter": self.samples_per_parameter,
        "reasons": list(self.reasons),
    }

Functions:

classify_phase_sindy_confidence

classify_phase_sindy_confidence(
    *,
    status: str,
    r_squared: float | None,
    sample_count: int,
    node_count: int,
    active_terms: int,
    total_terms: int,
    sparsity: float,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> SindyConfidence

Classify a phase-SINDy fit into an honest tier and discovery posture.

Parameters

status : str The fit status from the discovery block. Any value other than "fitted" is a skip and yields the refused posture. r_squared : float or None The scale-free coefficient of determination of the fit, or None when no fit was performed. sample_count : int Number of derivative samples actually regressed. node_count : int Number of oscillator nodes; equal to the per-node parameter count of the Kuramoto sine-difference library. active_terms : int Number of coefficients selected above the sparsity threshold. total_terms : int Total number of coefficients in the library. sparsity : float Support-sparsity fraction of the fit; carried through for the record but not itself a gate. policy : SindyConfidencePolicy, optional Thresholds separating a credible discovery from weak evidence.

Returns

SindyConfidence The tier, posture, the quantities the verdict rested on, and the ordered reasons.

Source code in src/scpn_phase_orchestrator/autotune/sindy_confidence.py

def classify_phase_sindy_confidence(
    *,
    status: str,
    r_squared: float | None,
    sample_count: int,
    node_count: int,
    active_terms: int,
    total_terms: int,
    sparsity: float,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> SindyConfidence:
    """Classify a phase-SINDy fit into an honest tier and discovery posture.

    Parameters
    ----------
    status : str
        The fit status from the discovery block. Any value other than
        ``"fitted"`` is a skip and yields the ``refused`` posture.
    r_squared : float or None
        The scale-free coefficient of determination of the fit, or ``None``
        when no fit was performed.
    sample_count : int
        Number of derivative samples actually regressed.
    node_count : int
        Number of oscillator nodes; equal to the per-node parameter count of
        the Kuramoto sine-difference library.
    active_terms : int
        Number of coefficients selected above the sparsity threshold.
    total_terms : int
        Total number of coefficients in the library.
    sparsity : float
        Support-sparsity fraction of the fit; carried through for the record
        but not itself a gate.
    policy : SindyConfidencePolicy, optional
        Thresholds separating a credible discovery from weak evidence.

    Returns
    -------
    SindyConfidence
        The tier, posture, the quantities the verdict rested on, and the
        ordered reasons.
    """
    if status != FITTED_STATUS:
        return SindyConfidence(
            tier=VALIDATION_TIER_SCAFFOLD,
            posture=POSTURE_REFUSED,
            r_squared=None,
            samples_per_parameter=None,
            reasons=(f"no phase-SINDy fit was performed (status={status!r})",),
        )

    samples_per_parameter = (
        None if node_count <= 0 else float(sample_count) / float(node_count)
    )

    reasons: list[str] = []
    if active_terms <= 0:
        reasons.append("the fit selected no active terms above the sparsity threshold")
    if r_squared is None:
        reasons.append("the fit reported no R² to judge explanatory power")
    elif r_squared < policy.min_r_squared:
        reasons.append(
            f"R² {r_squared:.4f} is below the discovery threshold "
            f"{policy.min_r_squared:.4f}"
        )
    if samples_per_parameter is None:
        reasons.append("the per-node parameter count is unknown")
    elif samples_per_parameter < policy.min_samples_per_parameter:
        reasons.append(
            f"under-determined: {samples_per_parameter:.2f} derivative samples "
            f"per parameter is below the required "
            f"{policy.min_samples_per_parameter:.2f}"
        )

    if reasons:
        return SindyConfidence(
            tier=VALIDATION_TIER_SCAFFOLD,
            posture=POSTURE_INSUFFICIENT_EVIDENCE,
            r_squared=r_squared,
            samples_per_parameter=samples_per_parameter,
            reasons=tuple(reasons),
        )

    # An empty reasons list means every gate above passed, which in turn means
    # r_squared and samples_per_parameter are both non-None floats; the checks
    # accumulate reasons rather than narrow, so cast to record that here.
    resolved_r_squared = cast(float, r_squared)
    resolved_samples_per_parameter = cast(float, samples_per_parameter)
    return SindyConfidence(
        tier=VALIDATION_TIER_PARTIAL,
        posture=POSTURE_DISCOVERED,
        r_squared=resolved_r_squared,
        samples_per_parameter=resolved_samples_per_parameter,
        reasons=(
            "self-consistent recovery: the fit explains the derivative variance "
            f"(R² {resolved_r_squared:.4f}) and is over-determined "
            f"({resolved_samples_per_parameter:.2f} samples per parameter); tier "
            "is capped at 'partial' because a self-fit is not external validation",
        ),
    )

classify_phase_sindy_block

classify_phase_sindy_block(
    block: Mapping[str, Any],
    *,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> SindyConfidence

Classify a phase-SINDy evidence block mapping.

A thin, pure adapter over :func:classify_phase_sindy_confidence that reads the fields emitted by the discovery report's phase_sindy block.

Parameters

block : Mapping A phase_sindy evidence block carrying at least status; fitted blocks additionally carry r_squared, sample_count, node_count, active_terms, total_terms and sparsity. policy : SindyConfidencePolicy, optional Thresholds separating a credible discovery from weak evidence.

Returns

SindyConfidence The honest confidence verdict for the block.

Source code in src/scpn_phase_orchestrator/autotune/sindy_confidence.py

def classify_phase_sindy_block(
    block: Mapping[str, Any],
    *,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> SindyConfidence:
    """Classify a phase-SINDy evidence block mapping.

    A thin, pure adapter over :func:`classify_phase_sindy_confidence` that
    reads the fields emitted by the discovery report's ``phase_sindy`` block.

    Parameters
    ----------
    block : Mapping
        A ``phase_sindy`` evidence block carrying at least ``status``; fitted
        blocks additionally carry ``r_squared``, ``sample_count``,
        ``node_count``, ``active_terms``, ``total_terms`` and ``sparsity``.
    policy : SindyConfidencePolicy, optional
        Thresholds separating a credible discovery from weak evidence.

    Returns
    -------
    SindyConfidence
        The honest confidence verdict for the block.
    """
    status = str(block.get("status", ""))
    raw_r_squared = block.get("r_squared")
    r_squared = None if raw_r_squared is None else float(raw_r_squared)
    return classify_phase_sindy_confidence(
        status=status,
        r_squared=r_squared,
        sample_count=int(block.get("sample_count", 0)),
        node_count=int(block.get("node_count", 0)),
        active_terms=int(block.get("active_terms", 0)),
        total_terms=int(block.get("total_terms", 0)),
        sparsity=float(block.get("sparsity", 1.0)),
        policy=policy,
    )

Operator SINDy Options

The options module bundles the two knobs an operator turns when running phase-SINDy discovery through a binding proposal or the CLI: the sparsity threshold that decides which coupling coefficients survive, and the confidence policy that decides how strong a fit must be before it is called discovered.

sindy_options

Operator-facing options for the phase-SINDy discovery honesty surface.

A single bundle carries the two knobs an operator turns when running phase-SINDy discovery through a binding proposal or the CLI: the sparsity threshold that decides which coupling coefficients survive, and the confidence policy that decides how strong a fit must be before its recovered structure is called discovered. Keeping them together means the binding proposal and the CLI configure discovery the same way without duplicating the mapping.

Classes

SindyOptions dataclass

SindyOptions(
    phase_sindy_threshold: float = 0.05,
    confidence_policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
)

Operator configuration for phase-SINDy discovery and its confidence.

Parameters

phase_sindy_threshold : float Sparsity threshold below which a coupling coefficient is dropped from the phase-SINDy fit. Must be finite and non-negative; defaults to the discovery default of 0.05. confidence_policy : SindyConfidencePolicy Thresholds separating a credible discovery from weak evidence. Defaults to the conservative shared policy.

Methods:

__post_init__

__post_init__() -> None

Validate the threshold is finite and non-negative.

Source code in src/scpn_phase_orchestrator/autotune/sindy_options.py

def __post_init__(self) -> None:
    """Validate the threshold is finite and non-negative."""
    threshold = float(self.phase_sindy_threshold)
    if not isfinite(threshold) or threshold < 0.0:
        raise ValueError("phase_sindy_threshold must be finite and non-negative")
    object.__setattr__(self, "phase_sindy_threshold", threshold)

to_discovery_config

to_discovery_config() -> TimeSeriesDiscoveryConfig

Return the discovery config carrying the phase-SINDy threshold.

Only the phase-SINDy threshold is overridden; the other discovery thresholds keep their defaults.

Returns

TimeSeriesDiscoveryConfig A config with phase_sindy_threshold set from these options.

Source code in src/scpn_phase_orchestrator/autotune/sindy_options.py

def to_discovery_config(self) -> TimeSeriesDiscoveryConfig:
    """Return the discovery config carrying the phase-SINDy threshold.

    Only the phase-SINDy threshold is overridden; the other discovery
    thresholds keep their defaults.

    Returns
    -------
    TimeSeriesDiscoveryConfig
        A config with ``phase_sindy_threshold`` set from these options.
    """
    return TimeSeriesDiscoveryConfig(
        phase_sindy_threshold=self.phase_sindy_threshold
    )

Discovered-Dynamics Record

The discovered-dynamics module presents the recovered equations and per-node coupling edges paired — inseparably — with the confidence verdict, so a skipped or weak fit still produces a record but is never mistaken for a validated model. Every record carries a canonical-JSON SHA-256 content hash for a tamper-evident provenance trail.

discovered_dynamics

Operator-facing record of dynamics discovered by phase-SINDy.

Where :mod:scpn_phase_orchestrator.autotune.discovery emits the raw evidence blocks and :mod:scpn_phase_orchestrator.autotune.sindy_confidence judges how far to trust a fit, this module presents the result the way an operator reads it: the recovered equations, the per-node coupling edges, and — inseparably — the honest confidence verdict that says how much weight the structure carries.

The recovered equations are never shown without their posture. A skipped or weak fit still produces a record, but its confidence marks it refused or insufficient_evidence so the equations cannot be mistaken for a validated model. Every record carries a canonical-JSON SHA-256 content hash for a tamper-evident provenance trail.

Classes

DiscoveredDynamics dataclass

DiscoveredDynamics(
    library: str,
    status: str,
    equations: tuple[str, ...],
    coupling_edges: tuple[Mapping[str, Any], ...],
    confidence: SindyConfidence,
)

A discovered phase-dynamics model paired with its honest confidence.

Parameters

library : str The feature library the fit used, e.g. the Kuramoto sine-difference library. status : str The fit status from the discovery block ("fitted" or a skip reason). equations : tuple of str Human-readable recovered equations, one per node; empty when no fit was performed. coupling_edges : tuple of Mapping Per-node coupling edges (source, target, coefficient, abs_coefficient); empty when no fit was performed. confidence : SindyConfidence The honest tier and discovery posture for the fit.

Attributes

content_hash property

content_hash: str

Canonical-JSON SHA-256 digest of the record content.

Returns

str Lowercase hexadecimal SHA-256 digest of the canonical payload.

Methods:

to_audit_record

to_audit_record() -> dict[str, Any]

Return the complete JSON-safe record, including the content hash.

Returns

dict The canonical payload with the content_hash provenance field appended.

Source code in src/scpn_phase_orchestrator/autotune/discovered_dynamics.py

def to_audit_record(self) -> dict[str, Any]:
    """Return the complete JSON-safe record, including the content hash.

    Returns
    -------
    dict
        The canonical payload with the ``content_hash`` provenance field
        appended.
    """
    record = self._canonical_payload()
    record["content_hash"] = self.content_hash
    return record

Functions:

discovered_dynamics_from_block

discovered_dynamics_from_block(
    block: Mapping[str, Any],
    *,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> DiscoveredDynamics

Build an operator-facing record from a phase-SINDy evidence block.

Parameters

block : Mapping A phase_sindy evidence block as emitted by the discovery report. policy : SindyConfidencePolicy, optional Thresholds separating a credible discovery from weak evidence.

Returns

DiscoveredDynamics The recovered equations and coupling edges paired with the honest confidence verdict.

Source code in src/scpn_phase_orchestrator/autotune/discovered_dynamics.py

def discovered_dynamics_from_block(
    block: Mapping[str, Any],
    *,
    policy: SindyConfidencePolicy = DEFAULT_SINDY_CONFIDENCE_POLICY,
) -> DiscoveredDynamics:
    """Build an operator-facing record from a phase-SINDy evidence block.

    Parameters
    ----------
    block : Mapping
        A ``phase_sindy`` evidence block as emitted by the discovery report.
    policy : SindyConfidencePolicy, optional
        Thresholds separating a credible discovery from weak evidence.

    Returns
    -------
    DiscoveredDynamics
        The recovered equations and coupling edges paired with the honest
        confidence verdict.
    """
    confidence = classify_phase_sindy_block(block, policy=policy)
    equations = tuple(str(equation) for equation in block.get("equations", ()))
    coupling_edges = tuple(dict(edge) for edge in block.get("coupling_edges", ()))
    return DiscoveredDynamics(
        library=str(block.get("library", "")),
        status=str(block.get("status", "")),
        equations=equations,
        coupling_edges=coupling_edges,
        confidence=confidence,
    )

Replay-Only Learners

The learner module exposes PPO-like, SAC-like, and hybrid-physics proposal generators behind the existing replay gates. These helpers emit audit records and keep actuation_permitted false.

learners

Learner-shaped replay-only autotune proposal generators.

Classes

LearnerPolicyProposal dataclass

LearnerPolicyProposal(
    learner_kind: str,
    policy_search: ReplayPolicySearchResult,
    actuation_permitted: bool = False,
    learner_parameters: AuditMapping = dict(),
    physics_prior: AuditMapping = dict(),
)

Replay-trained learner proposal record for audit review only.

Methods:

__post_init__

__post_init__() -> None

Validate the replay-only learner proposal envelope.

Source code in src/scpn_phase_orchestrator/autotune/learners.py

def __post_init__(self) -> None:
    """Validate the replay-only learner proposal envelope."""
    if not isinstance(self.learner_kind, str) or not self.learner_kind.strip():
        raise ValueError("learner_kind must be a non-empty string")
    object.__setattr__(self, "learner_kind", self.learner_kind.strip())
    if not isinstance(self.policy_search, ReplayPolicySearchResult):
        raise TypeError("policy_search must be a ReplayPolicySearchResult")
    if self.actuation_permitted is True:
        raise ValueError(
            "learner proposals are replay-only; actuation_permitted must be False"
        )
    if self.actuation_permitted is not False:
        raise ValueError("actuation_permitted must be exactly False")
    if not isinstance(self.learner_parameters, Mapping):
        raise TypeError("learner_parameters must be a mapping")
    if not isinstance(self.physics_prior, Mapping):
        raise TypeError("physics_prior must be a mapping")

to_audit_record

to_audit_record() -> dict[str, object]

Return an audit-serialisable learner proposal record.

Returns

dict[str, object] An audit-serialisable learner proposal record.

Source code in src/scpn_phase_orchestrator/autotune/learners.py

def to_audit_record(self) -> dict[str, object]:
    """Return an audit-serialisable learner proposal record.

    Returns
    -------
    dict[str, object]
        An audit-serialisable learner proposal record.
    """
    return _json_safe_record(
        {
            "learner_kind": self.learner_kind,
            "actuation_permitted": self.actuation_permitted,
            "learner_parameters": dict(self.learner_parameters),
            "physics_prior": dict(self.physics_prior),
            "policy_search": self.policy_search.to_audit_record(),
        }
    )

Functions:

generate_ppo_like_proposal

generate_ppo_like_proposal(
    seed: KnobPolicyCandidate,
    evaluator: ReplayPolicyEvaluator,
    *,
    seed_value: int | None = None,
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> LearnerPolicyProposal

Generate a deterministic PPO-shaped proposal from replay evaluations.

Parameters

seed : KnobPolicyCandidate Seed for the deterministic RNG. evaluator : ReplayPolicyEvaluator The objective evaluator. seed_value : int | None Seed value for the deterministic RNG. reward_config : RewardConfig | None The reward configuration. proposal_config : PolicyProposalConfig | None The proposal configuration.

Returns

LearnerPolicyProposal A deterministic PPO-shaped proposal from replay evaluations.

Source code in src/scpn_phase_orchestrator/autotune/learners.py

def generate_ppo_like_proposal(
    seed: KnobPolicyCandidate,
    evaluator: ReplayPolicyEvaluator,
    *,
    seed_value: int | None = None,
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> LearnerPolicyProposal:
    """Generate a deterministic PPO-shaped proposal from replay evaluations.

    Parameters
    ----------
    seed : KnobPolicyCandidate
        Seed for the deterministic RNG.
    evaluator : ReplayPolicyEvaluator
        The objective evaluator.
    seed_value : int | None
        Seed value for the deterministic RNG.
    reward_config : RewardConfig | None
        The reward configuration.
    proposal_config : PolicyProposalConfig | None
        The proposal configuration.

    Returns
    -------
    LearnerPolicyProposal
        A deterministic PPO-shaped proposal from replay evaluations.
    """
    seed_value = _validate_seed_value(seed_value)
    clip_range = _uniform(seed_value, low=0.08, high=0.18)
    search_config = OfflinePolicySearchConfig(
        K_step=clip_range,
        alpha_step=clip_range * 0.5,
        zeta_step=clip_range * 0.5,
        Psi_step=clip_range * 0.25,
        channel_weight_step=clip_range * 0.25,
        cross_channel_gain_step=clip_range * 0.25,
        max_abs_knob=2.0,
    )
    return LearnerPolicyProposal(
        learner_kind="ppo_like_replay",
        policy_search=_safe_replay_search(
            seed,
            evaluator,
            search_config,
            reward_config,
            proposal_config,
        ),
        learner_parameters={
            "clip_range": clip_range,
            "seed_value": seed_value,
        },
    )

generate_sac_like_proposal

generate_sac_like_proposal(
    seed: KnobPolicyCandidate,
    evaluator: ReplayPolicyEvaluator,
    *,
    seed_value: int | None = None,
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> LearnerPolicyProposal

Generate a deterministic SAC-shaped proposal from replay evaluations.

Parameters

seed : KnobPolicyCandidate Seed for the deterministic RNG. evaluator : ReplayPolicyEvaluator The objective evaluator. seed_value : int | None Seed value for the deterministic RNG. reward_config : RewardConfig | None The reward configuration. proposal_config : PolicyProposalConfig | None The proposal configuration.

Returns

LearnerPolicyProposal A deterministic SAC-shaped proposal from replay evaluations.

Source code in src/scpn_phase_orchestrator/autotune/learners.py

def generate_sac_like_proposal(
    seed: KnobPolicyCandidate,
    evaluator: ReplayPolicyEvaluator,
    *,
    seed_value: int | None = None,
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> LearnerPolicyProposal:
    """Generate a deterministic SAC-shaped proposal from replay evaluations.

    Parameters
    ----------
    seed : KnobPolicyCandidate
        Seed for the deterministic RNG.
    evaluator : ReplayPolicyEvaluator
        The objective evaluator.
    seed_value : int | None
        Seed value for the deterministic RNG.
    reward_config : RewardConfig | None
        The reward configuration.
    proposal_config : PolicyProposalConfig | None
        The proposal configuration.

    Returns
    -------
    LearnerPolicyProposal
        A deterministic SAC-shaped proposal from replay evaluations.
    """
    seed_value = _validate_seed_value(seed_value)
    entropy_temperature = _uniform(seed_value, low=0.03, high=0.12)
    search_config = OfflinePolicySearchConfig(
        K_step=0.04 + entropy_temperature,
        alpha_step=0.04 + entropy_temperature,
        zeta_step=0.02 + entropy_temperature * 0.5,
        Psi_step=0.02 + entropy_temperature * 0.5,
        channel_weight_step=entropy_temperature * 0.5,
        cross_channel_gain_step=entropy_temperature * 0.5,
        max_abs_knob=2.0,
    )
    return LearnerPolicyProposal(
        learner_kind="sac_like_replay",
        policy_search=_safe_replay_search(
            seed,
            evaluator,
            search_config,
            reward_config,
            proposal_config,
        ),
        learner_parameters={
            "entropy_temperature": entropy_temperature,
            "seed_value": seed_value,
        },
    )

generate_hybrid_physics_proposal

generate_hybrid_physics_proposal(
    seed: KnobPolicyCandidate,
    evaluator: ReplayPolicyEvaluator,
    *,
    critical_coupling_estimate: float,
    seed_value: int | None = None,
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> LearnerPolicyProposal

Generate a replay proposal shaped by a critical-coupling prior.

Parameters

seed : KnobPolicyCandidate Seed for the deterministic RNG. evaluator : ReplayPolicyEvaluator The objective evaluator. critical_coupling_estimate : float Estimated critical coupling K_c. seed_value : int | None Seed value for the deterministic RNG. reward_config : RewardConfig | None The reward configuration. proposal_config : PolicyProposalConfig | None The proposal configuration.

Returns

LearnerPolicyProposal A replay proposal shaped by a critical-coupling prior.

Source code in src/scpn_phase_orchestrator/autotune/learners.py

def generate_hybrid_physics_proposal(
    seed: KnobPolicyCandidate,
    evaluator: ReplayPolicyEvaluator,
    *,
    critical_coupling_estimate: float,
    seed_value: int | None = None,
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> LearnerPolicyProposal:
    """Generate a replay proposal shaped by a critical-coupling prior.

    Parameters
    ----------
    seed : KnobPolicyCandidate
        Seed for the deterministic RNG.
    evaluator : ReplayPolicyEvaluator
        The objective evaluator.
    critical_coupling_estimate : float
        Estimated critical coupling ``K_c``.
    seed_value : int | None
        Seed value for the deterministic RNG.
    reward_config : RewardConfig | None
        The reward configuration.
    proposal_config : PolicyProposalConfig | None
        The proposal configuration.

    Returns
    -------
    LearnerPolicyProposal
        A replay proposal shaped by a critical-coupling prior.
    """
    seed_value = _validate_seed_value(seed_value)
    critical_coupling_estimate = _positive_real(
        critical_coupling_estimate,
        "critical_coupling_estimate",
    )

    current_k = _mean_seed_k(seed)
    prior_gap = critical_coupling_estimate - current_k
    prior_step = min(0.5, max(0.01, abs(prior_gap) * 0.25))
    jitter = _uniform(seed_value, low=0.0, high=0.02)
    search_config = OfflinePolicySearchConfig(
        K_step=prior_step + jitter,
        alpha_step=0.03,
        zeta_step=0.03,
        Psi_step=0.02,
        channel_weight_step=0.02,
        cross_channel_gain_step=0.02,
        max_abs_knob=max(critical_coupling_estimate * 2.0, 2.0),
    )
    return LearnerPolicyProposal(
        learner_kind="hybrid_physics_replay",
        policy_search=_safe_replay_search(
            seed,
            evaluator,
            search_config,
            reward_config,
            proposal_config,
        ),
        learner_parameters={
            "prior_gap": prior_gap,
            "prior_step": prior_step,
            "seed_value": seed_value,
        },
        physics_prior={
            "critical_coupling_estimate": float(critical_coupling_estimate),
        },
    )

Operator use model

Autotune in this system is intended as a discovery and review surface first. Its outputs should be understood as candidate proposals with evidence, not as immediate production actions.

That separation is reflected by the actuation_permitted=false audit flag and the existing replay-only flow: operators can inspect candidate dynamics, compare against domain constraints, and explicitly promote a policy only through normal supervision gates.

In practical terms, autotune is most valuable in three moments: - preflight analysis on unknown domains, - topological recovery after a major drift event, - and proposal generation for domain-specific handoff when new systems are onboarded.

The same evidence record model used here is what allows these candidate policies to be replayed and compared across time windows and boundary profiles.