API reference¶
Everything below is importable from the top-level package, from gridfm_trustlayer import .... The Python API guide shows how the pieces fit together.
Calibrators¶
gridfm_trustlayer.NaiveIntervalCalibrator
dataclass
¶
NaiveIntervalCalibrator(alpha: float, entity_ids: ndarray, entity_name: str, channel_names: tuple[str, ...], lower_residual_quantiles: dict[str, ndarray], upper_residual_quantiles: dict[str, ndarray], strat: str | None = None, stratification_values: ndarray | None = None, stratified_lower_residual_quantiles: dict[str, ndarray] = dict(), stratified_upper_residual_quantiles: dict[str, ndarray] = dict(), system_load_n_bins: int | None = None, system_load_bin_edges: ndarray | None = None, pred_n_bins: int | None = None, pred_bin_edges: dict[str, ndarray] = dict(), calibration_fingerprint: str | None = None, ground_truth_reference: dict[str, Any] | None = None, sparse_strata: str = DEFAULT_SPARSE_STRATA, _pooled: dict[str, ndarray] = dict())
Stratified split-conformal intervals (cp_type: element).
Each variable is calibrated independently per entity, and per stratum when
strat is set. Each tail uses the split-conformal order statistic at level
alpha / 2, so the interval contains the true value with probability at
least 1 - alpha (within each stratum) when calibration and test scenarios
are exchangeable, for any number of calibration samples. An element with
too few samples in a stratum for that guarantee gets its quantile over all
strata there (sparse_strata="pooled", the default), or infinite bounds
(sparse_strata="infinite").
Create it with fit; the class is named "naive" for historical reasons.
fit
classmethod
¶
fit(predictions: PredictionSet | ChannelSet, ground_truth: GroundTruthSet | ChannelSet, *, alpha: float, channel_names: Iterable[str] = ('Vm', 'Va'), strat: str | None = None, system_load_n_bins: int | None = None, system_load_bin_edges: ndarray | None = None, pred_n_bins: int | None = None, pred_bin_edges: dict[str, ndarray] | None = None, sparse_strata: str = DEFAULT_SPARSE_STRATA) -> NaiveIntervalCalibrator
Fit per-entity (and per-stratum) residual quantiles on calibration data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predictions
|
PredictionSet | ChannelSet
|
Calibration predictions. |
required |
ground_truth
|
GroundTruthSet | ChannelSet
|
True values for the same scenarios and entities. |
required |
alpha
|
float
|
Two-sided miscoverage level; |
required |
channel_names
|
Iterable[str]
|
Variables to calibrate. |
('Vm', 'Va')
|
strat
|
str | None
|
Optional stratification: |
None
|
system_load_n_bins
|
int | None
|
Number of load bins, for |
None
|
system_load_bin_edges
|
ndarray | None
|
Stored load-bin edges, for |
None
|
pred_n_bins
|
int | None
|
Number of prediction bins, for |
None
|
pred_bin_edges
|
dict[str, ndarray] | None
|
Bin edges per variable, for |
None
|
sparse_strata
|
str
|
|
DEFAULT_SPARSE_STRATA
|
Returns:
| Type | Description |
|---|---|
NaiveIntervalCalibrator
|
The fitted calibrator. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the scenarios or entities of the two inputs are not
aligned, |
predict_intervals ¶
predict_intervals(predictions: PredictionSet | ChannelSet) -> dict[str, tuple[np.ndarray, np.ndarray]]
Intervals for new predictions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predictions
|
PredictionSet | ChannelSet
|
Predictions with the same entities as at fit time. With scenario stratification, they must carry the stratum column. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, tuple[ndarray, ndarray]]
|
|
dict[str, tuple[ndarray, ndarray]]
|
Where an element's stratum had too few samples or none, the bounds |
dict[str, tuple[ndarray, ndarray]]
|
use its quantile over all strata (marked in |
dict[str, tuple[ndarray, ndarray]]
|
|
dict[str, tuple[ndarray, ndarray]]
|
strata together are too few. |
gridfm_trustlayer.KernelIntervalCalibrator
dataclass
¶
KernelIntervalCalibrator(alpha: float, entity_ids: ndarray, entity_name: str, channel_names: tuple[str, ...], strat: str | None, stratification_values: ndarray, k_nn: int, n_eff_min: float, bandwidth: str, n_eff_target: float, distance: str, embedding_pooling: str, embedding_mean: ndarray, embedding_scale: ndarray, calibration_embeddings: ndarray, calibration_strata: ndarray, residuals: dict[str, ndarray], valid_masks: dict[str, ndarray], fallback_lower_quantiles: dict[str, ndarray] = dict(), fallback_upper_quantiles: dict[str, ndarray] = dict(), system_load_n_bins: int | None = None, system_load_bin_edges: ndarray | None = None, pred_n_bins: int | None = None, pred_bin_edges: dict[str, ndarray] = dict(), calibration_fingerprint: str | None = None, ground_truth_reference: dict[str, Any] | None = None, tuning: dict[str, Any] | None = None, sparse_strata: str = DEFAULT_SPARSE_STRATA, n_jobs: int | None = None, _banks: dict[str, list[_StratumBank]] = dict(), _diagnostics: dict[str, ndarray] = dict(), _pooled: dict[str, ndarray] = dict(), _pooled_quantiles: dict[str, tuple[ndarray, ndarray]] = dict())
Kernel-weighted conformal intervals (cp_type: kernel).
For each test scenario, calibration residuals are weighted by a Gaussian kernel on the distance between scenario embeddings, within the scenario's stratum, and the bound is a weighted quantile with a point mass for the unseen test residual. The margin therefore adapts to each scenario. Coverage is checked empirically, not guaranteed in finite samples.
Create it with fit; the artifact keeps all calibration residuals and
standardized embeddings.
diagnostics
property
¶
Per-scenario diagnostics of the last predict_intervals call.
{"n_eff": ..., "kernel_bandwidth_sq": ..., "kernel_fallback": ...}, one
value per test scenario: the effective sample size of its weights, the
squared bandwidth used, and whether it fell back to the stratum-wide
quantile for at least one entity.
fit
classmethod
¶
fit(predictions: PredictionSet | ChannelSet, ground_truth: GroundTruthSet | ChannelSet, *, alpha: float, channel_names: Iterable[str] = ('Vm', 'Va'), strat: str | None = None, embeddings: ndarray, embedding_pooling: str = 'mean', k_nn: int = DEFAULT_K_NN, n_eff_min: float = DEFAULT_N_EFF_MIN, bandwidth: str = DEFAULT_BANDWIDTH, n_eff_target: float | None = None, distance: str = DEFAULT_DISTANCE, system_load_n_bins: int | None = None, system_load_bin_edges: ndarray | None = None, sparse_strata: str = DEFAULT_SPARSE_STRATA) -> KernelIntervalCalibrator
Store the calibration residuals and embeddings that predict_intervals weights.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predictions
|
PredictionSet | ChannelSet
|
Calibration predictions. |
required |
ground_truth
|
GroundTruthSet | ChannelSet
|
True values for the same scenarios and entities. |
required |
alpha
|
float
|
Two-sided miscoverage level; |
required |
channel_names
|
Iterable[str]
|
Variables to calibrate. |
('Vm', 'Va')
|
strat
|
str | None
|
Optional scenario stratification ( |
None
|
embeddings
|
ndarray
|
|
required |
embedding_pooling
|
str
|
Name of the pooling that produced |
'mean'
|
k_nn
|
int
|
Neighbour rank that sets the bandwidth, for |
DEFAULT_K_NN
|
n_eff_min
|
float
|
Effective sample size below which a scenario falls back to the stratum-wide quantile. |
DEFAULT_N_EFF_MIN
|
bandwidth
|
str
|
|
DEFAULT_BANDWIDTH
|
n_eff_target
|
float | None
|
Target effective sample size for |
None
|
distance
|
str
|
|
DEFAULT_DISTANCE
|
system_load_n_bins
|
int | None
|
Number of load bins, for |
None
|
system_load_bin_edges
|
ndarray | None
|
Stored load-bin edges, for |
None
|
sparse_strata
|
str
|
|
DEFAULT_SPARSE_STRATA
|
Returns:
| Type | Description |
|---|---|
KernelIntervalCalibrator
|
The fitted calibrator. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If inputs are misaligned, an option is invalid, or
|
predict_intervals ¶
predict_intervals(predictions: PredictionSet | ChannelSet) -> dict[str, tuple[np.ndarray, np.ndarray]]
Scenario-specific intervals for new predictions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predictions
|
PredictionSet | ChannelSet
|
Predictions with the same entities as at fit time, with
|
required |
Returns:
| Type | Description |
|---|---|
dict[str, tuple[ndarray, ndarray]]
|
|
dict[str, tuple[ndarray, ndarray]]
|
Per-scenario diagnostics are available afterwards in |
Data containers¶
gridfm_trustlayer.ChannelSet
dataclass
¶
ChannelSet(scenario_ids: ndarray, entity_ids: ndarray, entity_name: str, channels: dict[str, ndarray], valid_masks: dict[str, ndarray] = dict(), entity_columns: dict[str, ndarray] = dict(), scenario_columns: dict[str, ndarray] = dict(), scenario_embeddings: ndarray | None = None)
Variables of one dataset as aligned (scenario x entity) arrays.
The input of both calibrators: predictions and true values are two
ChannelSets with the same scenario_ids and entity_ids, in the same order.
Attributes:
| Name | Type | Description |
|---|---|---|
scenario_ids |
ndarray
|
Scenario identifiers, shape |
entity_ids |
ndarray
|
Bus or branch identifiers, shape |
entity_name |
str
|
|
channels |
dict[str, ndarray]
|
One |
valid_masks |
dict[str, ndarray]
|
Optional boolean array per variable; |
entity_columns |
dict[str, ndarray]
|
Optional per-entity metadata, e.g. |
scenario_columns |
dict[str, ndarray]
|
Per-scenario values used for stratification, e.g.
|
scenario_embeddings |
ndarray | None
|
|
gridfm_trustlayer.PredictionSet
dataclass
¶
PredictionSet(task_name: str, channel_names: tuple[str, ...], scenario_ids: ndarray, bus_ids: ndarray, Vm: ndarray, Va: ndarray, channels: dict[str, ndarray], bus_type_masks: dict[str, ndarray], generator_predictions: GeneratorPredictionSet | None = None, scenario_embeddings: ndarray | None = None)
Predictions of one dataset as (scenario x bus) arrays, from load_predictions.
Attributes:
| Name | Type | Description |
|---|---|---|
task_name |
str
|
|
channel_names |
tuple[str, ...]
|
Predicted variables found in the file. |
scenario_ids |
ndarray
|
Scenario identifiers, sorted, shape |
bus_ids |
ndarray
|
Bus identifiers, shape |
Vm |
ndarray
|
Predicted voltage magnitudes, |
Va |
ndarray
|
Predicted voltage angles in radians, |
channels |
dict[str, ndarray]
|
Every predicted variable, including |
bus_type_masks |
dict[str, ndarray]
|
Boolean |
generator_predictions |
GeneratorPredictionSet | None
|
OPF generator predictions, when loaded. |
scenario_embeddings |
ndarray | None
|
Pooled scenario embeddings, when attached. |
gridfm_trustlayer.GroundTruthSet
dataclass
¶
GroundTruthSet(scenario_ids: ndarray, bus_ids: ndarray, branch_ids: ndarray, n_outages: ndarray, outage_type: ndarray, system_load: ndarray, removed_mask: ndarray, Vm: ndarray, Va: ndarray, channels: dict[str, ndarray], admittances: dict[str, ndarray], f_bus: ndarray, t_bus: ndarray, rate_a: ndarray, base_kv: ndarray, sn_mva: float, bus_type_masks: dict[str, ndarray])
True AC solution and network data of one dataset, from load_ground_truth.
Attributes:
| Name | Type | Description |
|---|---|---|
scenario_ids |
ndarray
|
Scenario identifiers, shape |
bus_ids |
ndarray
|
Bus identifiers, shape |
branch_ids |
ndarray
|
Branch identifiers, shape |
n_outages |
ndarray
|
Out-of-service branches plus generators, per scenario. |
outage_type |
ndarray
|
Outage regime per scenario: 0 |
system_load |
ndarray
|
Total active load per scenario. |
removed_mask |
ndarray
|
|
Vm |
ndarray
|
True voltage magnitudes, |
Va |
ndarray
|
True voltage angles in radians, |
channels |
dict[str, ndarray]
|
Every true bus variable, including |
admittances |
dict[str, ndarray]
|
Branch admittance coefficients, |
f_bus |
ndarray
|
From-bus of each branch. |
t_bus |
ndarray
|
To-bus of each branch. |
rate_a |
ndarray
|
Rating of each branch. |
base_kv |
ndarray
|
Base voltage of each bus. |
sn_mva |
float
|
System base power. |
bus_type_masks |
dict[str, ndarray]
|
Boolean |
Loading GridFM files¶
gridfm_trustlayer.load_predictions ¶
load_predictions(parquet_path: str, apply_task_correction: bool = True, task_name: str | None = None, generator_parquet_path: str | None = None, *, apply_pf_mask: bool | None = None, scenario_ids: ndarray | None = None) -> PredictionSet
Read a predictions parquet (one row per scenario and bus) into arrays.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parquet_path
|
str
|
Predictions file, e.g. graphkit's |
required |
apply_task_correction
|
bool
|
Apply the task's conventions; for |
True
|
task_name
|
str | None
|
|
None
|
generator_parquet_path
|
str | None
|
OPF generator predictions file, optional. |
None
|
apply_pf_mask
|
bool | None
|
Deprecated alias of |
None
|
scenario_ids
|
ndarray | None
|
Only these scenarios; all when not given. Reading a block of scenarios at a time keeps memory bounded on large datasets. |
None
|
Returns:
| Type | Description |
|---|---|
PredictionSet
|
The predictions, with scenarios sorted by id. |
gridfm_trustlayer.load_ground_truth ¶
load_ground_truth(pf_node_parquet: str, pf_edge_parquet: str, bus_params_csv: str | None = None, edge_params_csv: str | None = None, *, scenario_ids: Iterable[int] | None = None, channel_names: Iterable[str] | None = None, sn_mva: float = 100.0, reference: GroundTruthReference | None = None, solved: bool = True) -> GroundTruthSet
Read datakit's raw bus and branch parquets into arrays.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
pf_node_parquet
|
str
|
Bus ground truth, datakit |
required |
pf_edge_parquet
|
str
|
Branch ground truth, datakit |
required |
bus_params_csv
|
str | None
|
Unused; kept for compatibility. |
None
|
edge_params_csv
|
str | None
|
Unused; kept for compatibility. |
None
|
scenario_ids
|
Iterable[int] | None
|
Scenarios to load, e.g. |
None
|
channel_names
|
Iterable[str] | None
|
Bus variables to load besides |
None
|
sn_mva
|
float
|
System base power. |
100.0
|
reference
|
GroundTruthReference | None
|
Dataset-wide outage reference and angle unit, from
|
None
|
solved
|
bool
|
Whether to read the AC solution ( |
True
|
Returns:
| Type | Description |
|---|---|
GroundTruthSet
|
The ground truth. |
gridfm_trustlayer.load_scenario_embeddings ¶
load_scenario_embeddings(parquet_path: str, *, pooling: str = DEFAULT_POOLING, scenario_ids: ndarray | None = None) -> ScenarioEmbeddingSet
Read an embeddings parquet and pool it to one vector per scenario.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
parquet_path
|
str
|
Embeddings file with |
required |
pooling
|
str
|
|
DEFAULT_POOLING
|
scenario_ids
|
ndarray | None
|
Scenarios to return, in this order; all when not given. |
None
|
Returns:
| Type | Description |
|---|---|
ScenarioEmbeddingSet
|
A |
ScenarioEmbeddingSet
|
|
gridfm_trustlayer.compute_line_loading ¶
Line loading implied by bus voltages: max(|S_from|, |S_to|) / rate_a.
The AC branch flows are computed by datakit's branch-flow routine, so
gridfm_datakit must be installed (the graphkit extra).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
Vm
|
ndarray
|
Voltage magnitudes, |
required |
Va_rad
|
ndarray
|
Voltage angles in radians, same shape. |
required |
gt
|
GroundTruthSet
|
Ground truth of the same scenarios, for the branch parameters. |
required |
Returns:
| Type | Description |
|---|---|
ndarray
|
|
ndarray
|
out-of-service or unrated branches. |
Artifacts¶
gridfm_trustlayer.save_interval_artifact ¶
Write a fitted calibrator to a .trustlayer file.
JSON for NaiveIntervalCalibrator, a compressed NumPy archive for
KernelIntervalCalibrator; the format the CLI writes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
calibrator
|
IntervalCalibrator
|
The fitted calibrator. |
required |
output_path
|
str
|
Destination file; parent directories are created. |
required |
gridfm_trustlayer.load_interval_artifact ¶
Read a calibrator from a .trustlayer file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
artifact_path
|
str
|
A file written by |
required |
Returns:
| Type | Description |
|---|---|
IntervalCalibrator
|
The calibrator, of the class it was saved from. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the file holds several calibrators (a mixed CLI run). |
gridfm_trustlayer.load_interval_artifacts ¶
Read every calibrator of a .trustlayer file.
A CLI run with both bus voltages and line loading stores one calibrator per family in the same file; a single-family file holds one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
artifact_path
|
str
|
A file written by |
required |
Returns:
| Type | Description |
|---|---|
dict[str, IntervalCalibrator]
|
The calibrators by family: |
dict[str, IntervalCalibrator]
|
( |
Screening¶
gridfm_trustlayer.classify_status ¶
classify_status(lower: ndarray, upper: ndarray, *, upper_limit: float | None = None, lower_limit: float | None = None) -> np.ndarray
Status of each value: within_limits, uncertain or violation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
lower
|
ndarray
|
Lower bounds, any shape (e.g. scenarios x elements). |
required |
upper
|
ndarray
|
Upper bounds, the same shape. |
required |
upper_limit
|
float | None
|
Upper operating limit, e.g. |
None
|
lower_limit
|
float | None
|
Lower operating limit, e.g. |
None
|
Returns:
| Type | Description |
|---|---|
ndarray
|
An array of status names of the bounds' shape. Pass the point |
ndarray
|
prediction as both |
Raises:
| Type | Description |
|---|---|
ValueError
|
If neither limit is given. |
gridfm_trustlayer.scenario_status ¶
Status of each operating scenario: the worst status of its values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
statuses
|
ndarray
|
Value statuses from |
required |
axis
|
int
|
The axis of the values within a scenario. |
-1
|
valid
|
ndarray | None
|
Optional mask of the same shape; |
None
|
Returns:
| Type | Description |
|---|---|
ndarray
|
|
ndarray
|
any valid value is |
Kernel bandwidth tuning¶
gridfm_trustlayer.conformal.tuning.tune_n_eff_target ¶
tune_n_eff_target(predictions: ChannelSet, ground_truth: ChannelSet, *, alpha: float, channel_names: Sequence[str], strat: str | None = None, embedding_pooling: str = 'mean', distance: str = 'euclidean', n_eff_min: float = 5.0, options: TuningOptions | None = None, groups: ndarray | None = None, system_load_n_bins: int | None = None, system_load_bin_edges: ndarray | None = None, sparse_strata: str = DEFAULT_SPARSE_STRATA, n_jobs: int | None = 1) -> BandwidthTuningResult
Choose n_eff_target (and optionally distance) by K-fold cross-validation.
predictions and ground_truth are the calibration ChannelSets, with
scenario embeddings attached to predictions. distance is used unless
options.distances lists the distances to compare. groups (one value per
scenario, e.g. the datakit load_scenario_idx) keeps related scenarios in
the same fold; without it, near-duplicate scenarios on both sides of a fold
make localization look better than it is. sparse_strata is passed to every
fold's calibrator, so the candidates are compared as they will be applied.
The fold x candidate fits are independent and run in n_jobs processes
(see gridfm_trustlayer.parallel; None for the default of 4); the result
is the same as with n_jobs=1. A script that passes n_jobs other than 1
must start from an if __name__ == "__main__": block, as processes are
spawned.
gridfm_trustlayer.conformal.tuning.TuningOptions
dataclass
¶
TuningOptions(folds: int = DEFAULT_TUNING_FOLDS, grid: tuple[float, ...] = DEFAULT_TUNING_GRID, distances: tuple[str, ...] | None = None, search: str = 'grid', budget: int = DEFAULT_TUNING_BUDGET, n_eff_range: tuple[float, float] | None = None, group_by: str = 'auto', seed: int = 0, coverage_tolerance: float | str = 'auto')
Options of the cross-validated bandwidth choice (the config's tuning: block).
Attributes:
| Name | Type | Description |
|---|---|---|
folds |
int
|
Number of cross-validation folds within the calibration set. |
grid |
tuple[float, ...]
|
Candidate |
distances |
tuple[str, ...] | None
|
Distances to compare; |
search |
str
|
|
budget |
int
|
Number of settings tried by |
n_eff_range |
tuple[float, float] | None
|
|
group_by |
str
|
|
seed |
int
|
Seed of the fold assignment and the random search. |
coverage_tolerance |
float | str
|
Allowed shortfall below |
Config builder¶
gridfm_trustlayer.interactive.config_builder ¶
Return an interactive form that builds and saves a trustlayer config.
Call it as the last line of a Jupyter cell so the form is displayed.
gridfm_trustlayer.interactive.build_config ¶
build_config(*, calibration: Mapping[str, str | None], test: Mapping[str, str | None], task_name: str = 'PowerFlow', cp_type: str = 'element', alpha: float = 0.1, variables: Sequence[str] = ('Vm', 'Va'), strat: str | None = None, system_load_n_bins: int | None = None, pred_n_bins: int | None = None, sparse_strata: str = 'pooled', embedding_pooling: str = DEFAULT_POOLING, distance: str = DEFAULT_DISTANCE, bandwidth: str = DEFAULT_BANDWIDTH, k_nn: int = DEFAULT_K_NN, n_eff_target: float | None = None, n_eff_min: float = DEFAULT_N_EFF_MIN, tune_n_eff_target: bool = False, tuning: Mapping[str, Any] | None = None, screen_bounds: Mapping[str, Mapping[str, float]] | None = None, apply: Mapping[str, str | None] | None = None, output_dir: str = 'trustlayer_outputs') -> dict[str, Any]
Assemble a config dict, keeping only the keys that apply to the choices.
Options that do not apply are dropped rather than written, e.g. kernel knobs
for cp_type="element", k_nn under bandwidth="n_eff", embeddings for
an element run, or bounds for variables that are not calibrated.
tune_n_eff_target=True (kernel with bandwidth="n_eff" only) writes
n_eff_target: auto and a tuning: block, so compute chooses the target
by cross-validation on the calibration set; tuning overrides its defaults
(folds, grid, distances, search, budget, n_eff_range,
group_by, seed, coverage_tolerance).
screen_bounds=None (or bounds only for unselected variables) gives a config
without a screening step: compute and calibrate produce the prediction
intervals, and no screen_bounds, screen_dir or screen command is set.
sparse_strata="infinite" (with strat) gives elements in a stratum too
small for a finite bound infinite bounds instead of their quantile over all
calibration scenarios (the default, "pooled").
apply (optional) describes new scenarios without a known solution, for the
apply command: their predictions, and the bus and branch files describing
them when the layer is stratified by outages or load or calibrates
line_loading.
gridfm_trustlayer.interactive.validate_config ¶
Raise ValueError if compute, calibrate or screen would reject it.
gridfm_trustlayer.interactive.save_config ¶
Validate and write config to path; returns the written path.