ldtc.lmeas¶

The "L" measurement subsystem. Computes loop and exchange predictive influence (𝓛_loop, 𝓛_ex), evaluates the loop- dominance margin M (dB), decides NC1 / SC1, and manages the deterministic (C, Ex) partition.

Module	Headline symbols	Use it for
`estimators`	`estimate_L`, `LResult`	Per-window estimation of `𝓛_loop`, `𝓛_ex` with bootstrapped CIs. Three methods: linear (Granger-like), sklearn MI, Kraskov k-NN MI.
`metrics`	`m_db`, `sc1_evaluate`, `SC1Stats`	Convert `𝓛` to `M (dB)`; evaluate SC1 from baseline / trough / recovery.
`partition`	`Partition`, `PartitionManager`, `greedy_suggest_C`	Maintain `(C, Ex)` with hysteresis; freeze during `Ω`; greedy regrowth proposals.
`diagnostics`	`stationarity_checks`, `var_nt_ratio`	Per-window ADF / KPSS and VAR `N / T` ratio diagnostics surfaced into the audit.

Loop-measurement primitives.

lmeas is the heart of LDTC's measurement pipeline. It turns a window of multivariate telemetry into the scalar indicators that NC1 and SC1 are defined in terms of:

estimators holds predictive-dependence estimators for loop influence L_loop and exchange influence L_ex, plus their bootstrap CIs.
metrics computes the loop-dominance metric M = 10 · log10(L_loop / L_ex) (dB) and the SC1 pass / fail evaluator.
diagnostics runs stationarity and VAR-health checks consumed by smell-tests.
partition manages the (C, Ex) partition with hysteresis and a greedy regrowth suggestor.

The CLI verification runs feed each window through these modules to compute the NC1 / SC1 indicators that the attest subpackage then signs and emits as artifacts.

estimators¶

estimators ¶

Estimators for loop and exchange influence (𝓛).

Lightweight predictive-dependence estimators used to compute loop influence L_loop and exchange influence L_ex over a (C, Ex) partition. Includes linear (Granger-like) and mutual-information methods, with optional transfer entropy and directed information proxies. Confidence intervals are computed via circular block bootstrap per window.

The shape convention used throughout this module is X of shape (T, N) with time on the first axis and signals on the second. C and Ex are disjoint index lists into the columns of X. The headline entrypoint is estimate_L, which returns an LResult with point estimates and 95% CIs for both partitions.

See Also

paper/main.tex: Criterion; Methods: Measurement and Attestation.

Classes:

Name	Description
`LResult`	Loop and exchange influence with their 95% CI bounds.

Functions:

Name	Description
`estimate_L`	Estimate loop and exchange influence.

LResult `dataclass` ¶

LResult(L_loop: float, L_ex: float, ci_loop: Tuple[float, float], ci_ex: Tuple[float, float])

Loop and exchange influence with their 95% CI bounds.

Returned by estimate_L. The CIs come from a circular block bootstrap over the time axis; for the linear estimator a marginal samples-per-parameter ratio (see var_nt_ratio) further inflates the bounds to advertise uncertainty downstream.

Attributes:

Name	Type	Description
`L_loop`	`float`	Point estimate of loop influence over the `C` partition.
`L_ex`	`float`	Point estimate of exchange influence into `C` from `Ex`.
`ci_loop`	`Tuple[float, float]`	Tuple of `(lo, hi)` CI bounds for `L_loop`.
`ci_ex`	`Tuple[float, float]`	Tuple of `(lo, hi)` CI bounds for `L_ex`.

estimate_L ¶

estimate_L(X: ndarray, C: Sequence[int], Ex: Sequence[int], method: str = 'linear', p: int = 3, lag_mi: int = 1, n_boot: int = 64, mi_k: int = 5) -> LResult

Estimate loop and exchange influence.

Computes L_loop over partition C and L_ex from Ex -> C using the selected predictive-dependence metric, then attaches a percentile bootstrap CI for both. The linear estimator uses an AR + lagged-source partial-R² scheme; the MI variants use either scikit-learn or a Kraskov KSG-I estimator.

Parameters:

Name	Type	Description	Default
`X`	`ndarray`	Time-by-signal matrix of shape `(T, N)`.	required
`C`	`Sequence[int]`	Indices of the loop partition.	required
`Ex`	`Sequence[int]`	Indices of the exchange partition. Must be disjoint from `C`.	required
`method`	`str`	One of `"linear"`, `"mi"`, `"mi_kraskov"`, `"transfer_entropy"`, or `"directed_information"`. The TE and DI methods fall back to a KSG MI proxy when no plug-in backend is installed.	`'linear'`
`p`	`int`	VAR order for the linear estimator.	`3`
`lag_mi`	`int`	Lag between sources and targets for MI, TE, and DI methods.	`1`
`n_boot`	`int`	Number of bootstrap draws for CI estimation.	`64`
`mi_k`	`int`	k-NN parameter for Kraskov MI.	`5`

Returns:

Type	Description
`LResult`	An `LResult` with point estimates and
`LResult`	`(lo, hi)` CI bounds for each metric.

Raises:

Type	Description
`ValueError`	If `method` is not one of the supported options.

Example

import numpy as np
from ldtc.lmeas.estimators import estimate_L

rng = np.random.default_rng(0)
X = rng.standard_normal((512, 4))

res = estimate_L(X, C=[0, 1], Ex=[2, 3], method="linear", p=3)
print(f"L_loop={res.L_loop:.3f}  L_ex={res.L_ex:.3f}")

metrics¶

metrics ¶

Metrics and thresholds.

Helper metrics for loop-dominance M (dB) and SC1 evaluation used by the verification harness and figures.

M = 10 · log10(L_loop / L_ex) is LDTC's headline scalar: it summarizes how strongly closed-loop influence dominates exchange influence, in decibels. SC1 then asks whether M is preserved (or recovers within τ_max seconds) under a perturbation Ω.

See Also

paper/main.tex: Criterion; SC1; Methods: Threshold Calibration.

Classes:

Name	Description
`SC1Stats`	Summary statistics used for SC1 evaluation.

Functions:

Name	Description
`m_db`	Compute loop-dominance in decibels.
`sc1_evaluate`	Evaluate SC1 pass/fail and return stats.

SC1Stats `dataclass` ¶

SC1Stats(delta: float, tau_rec: float, M_post: float)

Summary statistics used for SC1 evaluation.

Attributes:

Name	Type	Description
`delta`	`float`	Fractional drop in `L_loop` during the perturbation window (`Ω`), normalized by the pre-`Ω` baseline.
`tau_rec`	`float`	Estimated recovery time in seconds.
`M_post`	`float`	Decibel margin `M` measured after the recovery gate.

m_db ¶

m_db(L_loop: float, L_ex: float, eps: float = 1e-12) -> float

Compute loop-dominance in decibels.

Returns M = 10 · log10(L_loop / L_ex) with small positive floors on both numerator and denominator to avoid division-by-zero or log10(0).

Parameters:

Name	Type	Description	Default
`L_loop`	`float`	Loop influence value (typically from `estimate_L`).	required
`L_ex`	`float`	Exchange influence value (same source).	required
`eps`	`float`	Numerical floor applied to both numerator and denominator.	`1e-12`

Returns:

Type	Description
`float`	Decibel ratio of loop to exchange influence.

sc1_evaluate ¶

sc1_evaluate(L_loop_baseline: float, L_loop_trough: float, L_loop_recovered: float, M_post: float, epsilon: float, tau_rec_measured: float, Mmin: float, tau_max: float) -> Tuple[bool, SC1Stats]

Evaluate SC1 pass/fail and return stats.

Checks that the fractional loop drop delta and recovery time tau_rec are within preset limits and that the post-recovery margin M_post exceeds Mmin.

Parameters:

Name	Type	Description	Default
`L_loop_baseline`	`float`	Baseline loop influence before `Ω`.	required
`L_loop_trough`	`float`	Minimum loop influence measured during `Ω`.	required
`L_loop_recovered`	`float`	Loop influence after recovery. Currently unused in the decision; included for symmetry and future use.	required
`M_post`	`float`	Post-recovery decibel margin `M`.	required
`epsilon`	`float`	Maximum allowed fractional drop.	required
`tau_rec_measured`	`float`	Measured recovery time, in seconds.	required
`Mmin`	`float`	Minimum acceptable decibel margin after recovery.	required
`tau_max`	`float`	Maximum allowed recovery time, in seconds.	required

Returns:

Type	Description
`bool`	A tuple `(passed, stats)`. `passed` is the SC1 decision; `stats` is
`SC1Stats`	an `SC1Stats` with `delta`,
`Tuple[bool, SC1Stats]`	`tau_rec`, and `M_post`.

partition¶

partition ¶

Partition management and greedy regrowth.

A deterministic (C, Ex) partition representation, a manager that adds hysteresis around partition flips, and a greedy suggestor that proposes new C membership to increase loop influence under a sparsity penalty.

The split between C (closed loop) and Ex (exchange) is what gives M its meaning: it answers "which signals are part of the controller and which are exogenous I/O?" The greedy suggestor lets the harness periodically reconsider that split based on the data, while the manager prevents thrashing.

See Also

paper/main.tex: Criterion; Methods: Partitioning algorithm.

Classes:

Name	Description
`Partition`	`(C, Ex)` partition state with freeze flag and flip counter.
`PartitionManager`	Deterministic `(C, Ex)` partition with simple hysteresis.

Functions:

Name	Description
`greedy_suggest_C`	Greedy regrowth of `C` using `ΔL_loop` gain with sparsity penalty.

Partition `dataclass` ¶

Partition(C: List[int], Ex: List[int], frozen: bool = False, flips: int = 0)

(C, Ex) partition state with freeze flag and flip counter.

Attributes:

Name	Type	Description
`C`	`List[int]`	Indices belonging to the loop (closed) set.
`Ex`	`List[int]`	Indices belonging to the exchange set.
`frozen`	`bool`	If `True`, updates are suppressed (e.g., during `Ω` windows where partition flips would confound SC1 evaluation).
`flips`	`int`	Number of accepted partition flips since creation.

PartitionManager ¶

PartitionManager(N_signals: int, seed_C: Sequence[int])

Deterministic (C, Ex) partition with simple hysteresis.

Provides a minimal manager that can be frozen and that only accepts a new partition when a suggested C set yields a sufficient decibel gain ΔM over the current partition for a required number of consecutive windows. The hysteresis prevents the harness from chattering between similarly-scoring partitions.

Parameters:

Name	Type	Description	Default
`N_signals`	`int`	Total number of signals `N`.	required
`seed_C`	`Sequence[int]`	Initial indices for the `C` set; the remainder form `Ex`.	required

Initialize the manager with a seed partition.

Parameters:

Name	Type	Description	Default
`N_signals`	`int`	Total number of signals `N`.	required
`seed_C`	`Sequence[int]`	Initial indices for the `C` set; the remainder form `Ex`.	required

Methods:

Name	Description
`get`	Return the current partition state.
`freeze`	Enable or disable freeze to suppress updates.
`update_current_M`	Record the latest measured `M` for the current partition.
`maybe_regrow`	Consider adopting `suggested_C` using hysteresis on the `ΔM` gain.

get ¶

get() -> Partition

Return the current partition state.

Returns:

Type	Description
`Partition`	The current `Partition` dataclass.

freeze ¶

freeze(on: bool) -> None

Enable or disable freeze to suppress updates.

Parameters:

Name	Type	Description	Default
`on`	`bool`	`True` to freeze, `False` to unfreeze.	required

update_current_M ¶

update_current_M(M_db: float) -> None

Record the latest measured M for the current partition.

Parameters:

Name	Type	Description	Default
`M_db`	`float`	Decibel loop-dominance value `M = 10 · log10(L_loop / L_ex)`.	required

maybe_regrow ¶

maybe_regrow(suggested_C: Sequence[int], delta_M_db: float, delta_M_min_db: float = 0.5, consecutive_required: int = 3) -> None

Consider adopting suggested_C using hysteresis on the ΔM gain.

Updates are ignored when the manager is frozen. A suggestion is only accepted when the same suggested_C persists for consecutive_required calls and the gain exceeds delta_M_min_db each time.

Parameters:

Name	Type	Description	Default
`suggested_C`	`Sequence[int]`	Candidate list of indices for `C`.	required
`delta_M_db`	`float`	Decibel gain relative to the current partition's `M`.	required
`delta_M_min_db`	`float`	Minimum required `ΔM` to count toward acceptance.	`0.5`
`consecutive_required`	`int`	Number of consecutive qualifying windows required before the flip is committed.	`3`

greedy_suggest_C ¶

greedy_suggest_C(X: Any, C: List[int] | Sequence[int], Ex: List[int] | Sequence[int], *, estimator: Callable[..., Any], method: str = 'linear', p: int = 3, lag_mi: int = 1, n_boot_candidates: int = 8, mi_k: int = 5, lam: float = 0.0, theta: float = 0.0, kappa: int | None = None) -> Tuple[List[int], float, Dict[str, Any]]

Greedy regrowth of C using ΔL_loop gain with sparsity penalty.

Starting from the current (C, Ex), iteratively add the candidate from Ex that maximizes the penalized gain in L_loop until the marginal gain falls below theta or the cap kappa is reached. Candidates are evaluated in lexicographic index order so ties break deterministically.

Parameters:

Name	Type	Description	Default
`X`	`Any`	Telemetry matrix `(T, N)` consumed by `estimator`.	required
`C`	`List[int] \| Sequence[int]`	Current loop-set indices.	required
`Ex`	`List[int] \| Sequence[int]`	Current exchange-set indices.	required
`estimator`	`Callable[..., Any]`	Callable compatible with `estimate_L`.	required
`method`	`str`	Estimation method forwarded to `estimator`.	`'linear'`
`p`	`int`	VAR order for the linear estimator.	`3`
`lag_mi`	`int`	Lag for MI-based estimators.	`1`
`n_boot_candidates`	`int`	Number of bootstrap draws used during candidate evaluation.	`8`
`mi_k`	`int`	k-NN parameter for Kraskov MI.	`5`
`lam`	`float`	Sparsity penalty per added node.	`0.0`
`theta`	`float`	Minimum penalized gain required to accept a candidate.	`0.0`
`kappa`	`int \| None`	Optional cap on `\|C\|`.	`None`

Returns:

Type	Description
`List[int]`	A tuple `(suggested_C, delta_M_db, details)` where `details`
`float`	contains provenance about added indices and intermediate gains.

diagnostics¶

diagnostics ¶

Diagnostic helpers for measurement stability.

Wrappers for ADF / KPSS stationarity tests, a stationarity summary dataclass, and a VAR N/T ratio heuristic. These are used to annotate audit records and to guard the loop-influence estimators against ill-posed regimes (too few samples per parameter, non-stationary series, and so on).

See Also

paper/main.tex: Methods: Measurement; Smell-tests and invalidation.

Classes:

Name	Description
`StationaritySummary`	Summary of per-series stationarity flags.

Functions:

Name	Description
`stationarity_checks`	Run ADF and KPSS per column and summarize.
`var_nt_ratio`	Rule-of-thumb samples-per-parameter ratio for VAR(p).

StationaritySummary `dataclass` ¶

StationaritySummary(adf_nonstationary_frac: float, kpss_nonstationary_frac: float, per_series: List[Tuple[bool, bool]])

Summary of per-series stationarity flags.

Attributes:

Name	Type	Description
`adf_nonstationary_frac`	`float`	Fraction flagged non-stationary by ADF (fails to reject the unit-root null at 5%).
`kpss_nonstationary_frac`	`float`	Fraction flagged non-stationary by KPSS (rejects stationarity at 5%).
`per_series`	`List[Tuple[bool, bool]]`	List of tuples `(adf_nonstat, kpss_nonstat)` per column.

stationarity_checks ¶

stationarity_checks(X: ndarray) -> StationaritySummary

Run ADF and KPSS per column and summarize.

Parameters:

Name	Type	Description	Default
`X`	`ndarray`	Array of shape `(T, N)` with time along axis 0.	required

Returns:

Type	Description
`StationaritySummary`	A `StationaritySummary`
`StationaritySummary`	with per-series flags and overall fractions.

Raises:

Type	Description
`ValueError`	If `X` is not a 2D array.

var_nt_ratio ¶

var_nt_ratio(T: int, N: int, p: int) -> float

Rule-of-thumb samples-per-parameter ratio for VAR(p).

Computes (T - p) / (N * p) where T is the number of time samples, N is the number of signals, and p is the VAR lag order. Lower values indicate a more marginal regression setting; the linear estimator in estimate_L inflates its CIs when this ratio drops below ~1.5.

Parameters:

Name	Type	Description	Default
`T`	`int`	Number of time samples.	required
`N`	`int`	Number of signals (columns).	required
`p`	`int`	VAR lag order.	required

Returns:

Type	Description
`float`	Samples-per-parameter ratio. Returns `inf` if `N` or `p` is
`float`	non-positive (ill-defined regression).

ldtc.lmeas¶

estimators¶

estimators ¶

LResult dataclass ¶

estimate_L ¶

metrics¶

metrics ¶

SC1Stats dataclass ¶

m_db ¶

sc1_evaluate ¶

partition¶

partition ¶

Partition dataclass ¶

PartitionManager ¶

get ¶

freeze ¶

update_current_M ¶

maybe_regrow ¶

greedy_suggest_C ¶

diagnostics¶

diagnostics ¶

StationaritySummary dataclass ¶

stationarity_checks ¶

var_nt_ratio ¶

LResult `dataclass` ¶

SC1Stats `dataclass` ¶

Partition `dataclass` ¶

StationaritySummary `dataclass` ¶