API reference

The public modules of SmartCer, from the entry point inwards.

Environment

Gymnasium integration: SmartCerEnv and its registration as SmartCer-v1.

Usage:

>>> import gymnasium as gym
>>> import smartcer  # registers SmartCer-v1
>>> env = gym.make('SmartCer-v1', scenario='path/to/scenario')

class smartcer.gym.SmartCerEnv(scenario, start=None, end=None, episode_length=None, random_start=False, observe_current_step=True, observation='meco', lookback=0, normalizer=None, normalizer_window=None, reward='welfare', reward_per_node=False, penalty_coefficient=10.0, forecast=None, forecast_horizon=0, oracle=None, seed=None)

Bases: Env

Renewable Energy Community with one controlled battery per storage node.

Episodes cover episode_length steps of the scenario period, starting at its first step or, with random_start, at a step drawn with self.np_random. Every step of the episode is controlled; the episode ends on its last step t with truncated=True (terminated is always False). The terminal observation is the observation of step t + 1 with the final battery states, as at the start of an episode beginning at t + 1, so that the value of the next state can be bootstrapped; only without forecasts, on the last step of the data, it is the observation of step t with the final battery states. With a forecast horizon H the episodes use the steps 0, …, n_steps - H - 2 with observe_current_step=True and 0, …, n_steps - H - 1 with False, so that the forecasts of every step of an episode and of the step after it lie in the data.

Actions are clipped to [-1, 1] without penalty, then each node clips its action to the dynamic bounds and is penalised as in section 7. Non-finite actions raise ValueError.

Variables:

• community – The scenario’s community (nodes, batteries, incentive, timestamps).

• scenario_path – Scenario folder, or None when the environment was built from a Community.

• storage_node_ids – Identifiers of the battery nodes, in action order.

• obs_layout – Feature name → slice of the observation vector.

• normalizer – Min-max normaliser of the observed generation and load (save it with env.normalizer.save(path) to reuse it in an evaluation environment).

• observe_current_step – Information timing.

• reward_type – Reward type.

• reward_per_node – Whether the per-node variant of the reward is used.

• penalty_coefficient – σ of the reward.

• episode_length – Steps per episode.

• random_start – Whether episodes start at a random step.

• forecast – Forecast provider of the observation, or None.

• forecast_horizon – Forecast horizon H in steps (0 without forecasts).

• oracle – MILP oracle of info[‘privileged’] and of the ReplayController, or None.

• action_space – Box(-1, 1, (len(storage_node_ids),), float32).

• observation_space – Flat Box, see smartcer.gym.observation.schema.

Parameters:

• scenario (str | Path | Community)

• start (Any)

• end (Any)

• episode_length (int | None)

• random_start (bool)

• observe_current_step (bool)

• observation (str)

• lookback (int)

• normalizer (str | Path | MinMaxNormalizer | None)

• normalizer_window (tuple[Any, Any] | None)

• reward (str)

• reward_per_node (bool)

• penalty_coefficient (float)

• forecast (str | ForecastProvider | None)

• forecast_horizon (int)

• oracle (str | Path | OracleProvider | None)

• seed (int | None)

__init__(scenario, start=None, end=None, episode_length=None, random_start=False, observe_current_step=True, observation='meco', lookback=0, normalizer=None, normalizer_window=None, reward='welfare', reward_per_node=False, penalty_coefficient=10.0, forecast=None, forecast_horizon=0, oracle=None, seed=None)

Create the environment.

Parameters:

• scenario (str | Path | Community) – Scenario folder (loaded with smartcer.data.load_scenario) or a Community.

• start (Any) – First timestamp of the period (with a scenario folder only).

• end (Any) – Last timestamp of the period (with a scenario folder only).

• episode_length (int | None) – Steps per episode; None for the whole period.

• random_start (bool) – Draw the first step of each episode uniformly among the feasible ones.

• observe_current_step (bool) – True: the observation of step t holds g, l and bounds of step t; False: g and l of step t - 1 and a bound without the charge cap.

• observation (str) – Observation preset, ‘meco’ ([M23] Tab. II) or ‘extended’.

• lookback (int) – Past steps of generation and load per battery node (‘extended’ only).

• normalizer (str | Path | MinMaxNormalizer | None) – Normaliser of generation and load: a MinMaxNormalizer or the path of its JSON file; None fits one on normalizer_window.

• normalizer_window (tuple[Any, Any] | None) – (first, last) timestamp of the training window the normaliser is fitted on, both inclusive, within the period; None for the whole period.

• reward (str) – Reward type, ‘welfare’ (R_t = W_t - σ K_t), ‘profit’ or ‘energy’.

• reward_per_node (bool) – Per-node variant (not for ‘welfare’): r_u = m_u - σ k_u per battery node, reported in info[‘reward’][‘nodes’], and R_t = Σ_u r_u.

• penalty_coefficient (float) – σ, ≥ 0 (default 10, as in [M23]).

• forecast (str | ForecastProvider | None) – Forecasts of the observation’s forecast section: a name of smartcer.forecasts.FORECAST_PROVIDERS (‘perfect’, ‘mmfe_perfect’, ‘mmfe_standard’, ‘mmfe_noisy’), built on the community, or a provider; None for no forecasts.

• forecast_horizon (int) – Forecast horizon H in steps, ≥ 1 with forecasts.

• oracle (str | Path | OracleProvider | None) – MILP oracle: an oracle folder (smartcer.oracle.read_oracle) or an OracleProvider of the same community; None for no oracle.

• seed (int | None) – Seed used by the first reset() called without a seed.

Raises:

ValueError – On invalid arguments.

Return type:

None

classmethod from_v062_config(cer_path, cer_config, **overrides)

Create the environment from a 0.6.2 scenario path and run configuration.

See smartcer.gym.run_config.map_v062_run_config for the key mapping (DeprecationWarning).

Parameters:

• cer_path (str | Path) – Scenario folder.

• cer_config (dict[str, Any]) – The 0.6.2 run configuration (simulation, reward, …).

• **overrides (Any) – Arguments that replace the mapped ones.

Returns:

The environment.

Return type:

SmartCerEnv

property current_step: int

Step the next action applies to (the last step of the episode once it ends).

run_config()

The run configuration, JSON-serialisable, for the output files of a run.

Returns:

SmartCer version, scenario name and folder, period, time step, episode settings, seed, observation settings and normaliser window, reward settings, forecast provider (name or class), its settings (forecast_config, from the provider’s config() if it has one, else None) and horizon, oracle folder.

Return type:

dict[str, Any]

reset(*, seed=None, options=None)

Start an episode: restore the batteries, reset the forecast provider with self.np_random and choose the first step.

Parameters:

• seed (int | None) – Seed of self.np_random and of the action space; if None, the constructor’s seed is used by the first reset.

• options (dict[str, Any] | None) – Not used.

Returns:

First observation and info (t, timestamp, privileged, episode_start, episode_end, obs_clipped: values of the observation clipped to [0, 1] because they lie outside the normaliser’s training range).

Return type:

tuple[np.ndarray, dict[str, Any]]

step(action)

Apply one action per battery for the current step.

Parameters:

action (ndarray) – Array of shape (len(storage_node_ids),).

Returns:

Observation of the next step (of the terminal observation at the end of the episode, see the class docstring), reward R_t, terminated (always False), truncated (True on the last step of the episode) and info (see _step_info, plus obs_clipped: values of the returned observation clipped to [0, 1] because they lie outside the normaliser’s training range).

Return type:

tuple[np.ndarray, float, bool, bool, dict[str, Any]]

Raises:

• RuntimeError – If called before reset() or after the end of the episode.

• ValueError – If the action has the wrong shape or is not finite.

smartcer.gym.register_envs()

Register SmartCer-v1 with Gymnasium (idempotent).

Called on import smartcer and through the gymnasium.envs entry point (used by Gymnasium versions before 1.0, which load entry points on import).

Return type:

None

Rewards

Rewards of SmartCerEnv: main term minus σ times the constraint penalty.

class smartcer.gym.rewards.EnergyReward(storage_node_ids, penalty_coefficient=10.0, per_node=False)

Bases: RewardCalculator

0.6.2 energy reward: minus the squared energy imported by each node.

Parameters:

• storage_node_ids (Sequence[str])

• penalty_coefficient (float)

• per_node (bool)

node_main(node)

-(i^gri_{u,t})².

Parameters:

node (NodeStep) – Node step record.

Returns:

m_{u,t}.

Return type:

float

collective_main(step)

Σ_u -(i^gri_{u,t})² over all nodes.

Parameters:

step (CommunityStep) – Community step record.

Returns:

M_t.

Return type:

float

class smartcer.gym.rewards.NodeReward(main, penalty, total)

Bases: object

Reward of one battery node (per-node variant).

Variables:

• main (float) – Main term m_{u,t}.

• penalty (float) – Weighted penalty σ k_{u,t}.

• total (float) – r_{u,t} = main - penalty.

Parameters:

• main (float)

• penalty (float)

• total (float)

class smartcer.gym.rewards.ProfitReward(storage_node_ids, penalty_coefficient=10.0, per_node=False)

Bases: RewardCalculator

Net profit of each node; the collective variant adds the incentive (and equals the welfare).

Parameters:

• storage_node_ids (Sequence[str])

• penalty_coefficient (float)

• per_node (bool)

node_main(node)

Net profit of a node: revenue - cost - storage cost.

Parameters:

node (NodeStep) – Node step record.

Returns:

m_{u,t}.

Return type:

float

collective_main(step)

Σ_u m_{u,t} over all nodes plus the incentive I_t.

Parameters:

step (CommunityStep) – Community step record.

Returns:

M_t.

Return type:

float

class smartcer.gym.rewards.RewardBreakdown(main, penalty, total, nodes=None)

Bases: object

Reward of one step.

Variables:

• main (float) – Main term: M_t (collective) or Σ_u m_{u,t} (per node).

• penalty (float) – Weighted penalty: σ K_t (collective) or Σ_u σ k_{u,t} (per node).

• total (float) – The scalar reward R_t = main - penalty.

• nodes (dict[str, smartcer.gym.rewards.base.NodeReward] | None) – Battery node id → NodeReward (per-node variant), None for the collective one.

Parameters:

• main (float)

• penalty (float)

• total (float)

• nodes (dict[str, NodeReward] | None)

class smartcer.gym.rewards.RewardCalculator(storage_node_ids, penalty_coefficient=10.0, per_node=False)

Bases: ABC

Reward as a main term minus σ times the constraint penalty.

Subclasses implement collective_main and, if they support the per-node variant, node_main.

Variables:

• name (ClassVar[str]) – Reward type name (class attribute).

• supports_per_node (ClassVar[bool]) – Whether the per-node variant exists (class attribute).

• penalty_coefficient – σ.

• per_node – Whether the per-node variant is used.

• storage_node_ids – Identifiers of the battery nodes, in action order.

Parameters:

• storage_node_ids (Sequence[str])

• penalty_coefficient (float)

• per_node (bool)

abstractmethod collective_main(step)

Main term M_t of the collective variant.

Parameters:

step (CommunityStep) – Community step record.

Returns:

M_t.

Return type:

float

node_main(node)

Main term m_{u,t} of the per-node variant.

Parameters:

node (NodeStep) – Node step record.

Returns:

m_{u,t}.

Return type:

float

Raises:

NotImplementedError – For reward types without a per-node variant.

compute(step)

Reward of a step: main term minus σ times the constraint penalty.

Parameters:

step (CommunityStep) – Community step record.

Returns:

Main term, weighted penalty, total and, per node, their values.

Return type:

RewardBreakdown

class smartcer.gym.rewards.RewardConfig(name, description, calculator)

Bases: object

Entry of the reward registry.

Variables:

• name (str) – Reward type name.

• description (str) – Objective of the reward.

• calculator (type[smartcer.gym.rewards.base.RewardCalculator]) – Calculator class.

Parameters:

• name (str)

• description (str)

• calculator (type[RewardCalculator])

class smartcer.gym.rewards.WelfareReward(storage_node_ids, penalty_coefficient=10.0, per_node=False)

Bases: RewardCalculator

R_t = W_t - σ K_t ([M23] eq. 8).

Parameters:

• storage_node_ids (Sequence[str])

• penalty_coefficient (float)

• per_node (bool)

collective_main(step)

Social welfare W_t (section 10).

Parameters:

step (CommunityStep) – Community step record.

Returns:

W_t.

Return type:

float

smartcer.gym.rewards.make_reward(reward_type, storage_node_ids, penalty_coefficient=10.0, per_node=False)

Create the calculator of a registered reward type.

Parameters:

• reward_type (str) – Name in REWARD_REGISTRY.

• storage_node_ids (Sequence[str]) – Identifiers of the battery nodes.

• penalty_coefficient (float) – σ, ≥ 0.

• per_node (bool) – Use the per-node variant.

Returns:

The calculator.

Return type:

RewardCalculator

Raises:

ValueError – For an unknown type, an invalid σ or an unsupported per-node variant.

Observation

Observation of SmartCerEnv: flat vector with a named layout.

class smartcer.gym.observation.MinMaxNormalizer(bounds, window=None, scenario=None)

Bases: object

Per-feature (min, max) bounds; values are scaled to [0, 1] and clipped.

Variables:

• bounds – Series name → (min, max) over the training window.

• window – First and last timestamp of the training window (ISO strings), if known.

• scenario – Name of the scenario it was fitted on, if known.

Parameters:

• bounds (Mapping[str, tuple[float, float]])

• window (tuple[str, str] | None)

• scenario (str | None)

classmethod fit(series, window=None, scenario=None)

Fit the bounds on the values of the training window.

Parameters:

• series (Mapping[str, ArrayLike]) – Series name → values of the training window only.

• window (tuple[str, str] | None) – First and last timestamp of the training window, stored for reference.

• scenario (str | None) – Scenario name, stored for reference.

Returns:

The fitted normaliser.

Return type:

MinMaxNormalizer

Raises:

ValueError – If a series is empty.

require(names)

Check that the normaliser has bounds for the given series.

Parameters:

names (Iterable[str]) – Series names.

Raises:

ValueError – Naming the missing series.

Return type:

None

transform(name, values)

Scale values with the bounds of a series and clip them to [0, 1].

A constant series (min == max) maps to 0; values outside the training range, e.g. in a test window, are clipped to 0 or 1.

Parameters:

• name (str) – Series name.

• values (ArrayLike) – Values.

Returns:

Scaled values in [0, 1].

Return type:

np.ndarray

Raises:

KeyError – If the series has no bounds.

outside_range(name, values)

Values outside the training range of a series, which transform clips.

Parameters:

• name (str) – Series name.

• values (ArrayLike) – Values.

Returns:

Boolean mask, True where a value is below the minimum or above the maximum.

Return type:

np.ndarray

Raises:

KeyError – If the series has no bounds.

to_dict()

JSON-serialisable representation.

Returns:

format, type, scenario, window and bounds.

Return type:

dict[str, Any]

classmethod from_dict(data)

Normaliser from the output of to_dict.

Parameters:

data (Mapping[str, Any]) – Dictionary with format, type and bounds (window, scenario optional).

Returns:

The normaliser.

Return type:

MinMaxNormalizer

Raises:

ValueError – If the format or the type is not supported.

save(path)

Write the normaliser to a JSON file.

Parameters:

path (str | Path) – File path.

Returns:

The path written.

Return type:

Path

classmethod load(path)

Read a normaliser written by save.

Parameters:

path (str | Path) – File path.

Returns:

The normaliser.

Return type:

MinMaxNormalizer

class smartcer.gym.observation.ObservationBuilder(community, preset='meco', observe_current_step=True, lookback=0, normalizer=None, forecast=None, horizon=0)

Bases: object

Builds the flat observation of a community for a step.

Variables:

• preset – Observation preset.

• observe_current_step – Information timing (docs/FORMULATION.md section 12).

• lookback – Past steps of g and l per battery node (extended only).

• normalizer – Min-max normaliser of generation and load.

• layout – Feature name → slice of the observation vector.

• sections – Section name → feature names, in order (forecast is empty without a forecast provider).

• space – The observation space, Box(low, high, (dim,), float32).

• global_prices – True if the prices are one global pair.

• clipped – Per step, the number of scaled values of the observation that lie outside the normaliser’s training range and are clipped to [0, 1] (the zeros before the first row of the data are not counted), without the forecasts.

• forecast – Forecast provider, or None.

• horizon – Forecast horizon H in steps (0 without a provider).

• forecast_keys – Provider keys of the forecasts in the observation, in order (‘community/<quantity>’, then ‘<node_id>/<quantity>’).

• last_clipped – Clipped values of the last built observation, forecasts included.

Parameters:

• community (Community)

• preset (str)

• observe_current_step (bool)

• lookback (int)

• normalizer (MinMaxNormalizer | None)

• forecast (ForecastProvider | None)

• horizon (int)

property dim: int

Length of the observation vector.

forecast_start(t)

First forecast step of the observation of step t.

Parameters:

t (int) – Step index.

Returns:

t + 1 with observe_current_step=True, t with False.

Return type:

int

build(t)

Observation for choosing the action of step t, from the current battery states and, with a forecast provider, its forecasts.

Parameters:

t (int) – Step index.

Returns:

Float32 vector inside space.

Return type:

np.ndarray

Raises:

ValueError – If the forecast provider returns invalid forecasts.

smartcer.gym.observation.fit_normalizer(community, window=None)

Fit the observation normaliser on the steps of a training window.

Only the values of the window are read, so data outside it (e.g. a test period) do not affect the bounds.

Parameters:

• community (Community) – The community.

• window (tuple[Any, Any] | None) – (first, last) timestamp, both inclusive; None for the whole period.

Returns:

Bounds of scaled_series(community) over the window.

Return type:

MinMaxNormalizer

smartcer.gym.observation.scaled_series(community)

The series that the observation min-max scales, over the whole community period.

Parameters:

community (Community) – The community.

Returns:

<node_id>/generation and <node_id>/load of every battery node, and community/generation, community/load (totals of the nodes without a battery that have generation or load).

Return type:

dict[str, np.ndarray]

Controllers

Baseline controllers of SmartCerEnv, the baseline runs and the replay check of the oracle.

class smartcer.gym.controllers.Controller(env)

Bases: ABC

Chooses the actions of a SmartCerEnv, one per battery.

Variables:

• name (str) – Controller name, used for the output folders of a run; a subclass may set it per instance (e.g. from a mode).

• env – The unwrapped environment the controller acts on.

Parameters:

env (Env)

reset()

Prepare a new episode; called after env.reset(). Stateless controllers do nothing.

Return type:

None

abstractmethod act(observation, info)

Actions for the step env.current_step.

Parameters:

• observation (ndarray) – Latest observation of the environment.

• info (dict[str, Any]) – Latest info of the environment.

Returns:

One action per battery in [-1, 1], float32 or float64, in storage_node_ids order.

Return type:

np.ndarray

class smartcer.gym.controllers.NaiveController(env)

Bases: Controller

Zero action for every battery: the community without battery control.

With zero actions the batteries only lose their self-discharge, and the clipping penalty is 0.

Parameters:

env (Env)

act(observation, info)

Zero actions.

Parameters:

• observation (ndarray) – Not used.

• info (dict[str, Any]) – Not used.

Returns:

Zeros, one per battery.

Return type:

np.ndarray

class smartcer.gym.controllers.RBCController(env, mode='community')

Bases: Controller

Charges with surplus and discharges to cover deficit, within the dynamic bounds.

Every action lies in [-A^lower_{u,t}, A^upper_{u,t}] (FORMULATION section 6): the energies the rule asks for are limited by the power limits, the charge cap, the efficiencies, S^min, S^max and the self-discharge, as for the environment’s clipping, so the clipping penalty is always 0 (the actions are float64 for this reason). Below, E^cha_u = A^upper_{u,t} P^cha_u Δt and E^dis_u = A^lower_{u,t} P^dis_u Δt are the energies (node side) the battery can take and deliver in step t.

• node, for every battery node u with d = g_{u,t} - l_{u,t}: d ≥ 0 charges min(d, E^cha_u), d < 0 discharges min(-d, E^dis_u); a producer with g_{u,t} = 0 discharges E^dis_u (the 0.6.2 rule; with d = g it would never discharge). Each node only looks at itself.

• community, with the balance B_t = Σ_u g_{u,t} - Σ_u l_{u,t} over all nodes: B_t > 0 charges min(B_t, Σ E^cha) split pro rata to E^cha_u; B_t < 0 discharges min(-B_t, Σ E^dis) split pro rata to E^dis_u. This is the class A reference baseline.

The controller reads g and l of step t from the scenario, as in [M23], whatever observe_current_step is.

Variables:

mode – node or community.

Parameters:

• env (Env)

• mode (Literal['node', 'community'])

act(observation, info)

Actions of section 13 for the step env.current_step.

Parameters:

• observation (ndarray) – Not used.

• info (dict[str, Any]) – Not used.

Returns:

One action per battery in [-A^lower, A^upper], float64: the environment applies it exactly, while a float32 action equal to a bound may exceed it by about 1e-8 and be penalised.

Return type:

np.ndarray

class smartcer.gym.controllers.ReplayCheck(steps, penalty_max, penalty_steps, soc_error_max_kwh, simultaneous_rows, clipped_rows, clipped_max_kwh, welfare, objective, relative_gap, rtol, fit_score, passed)

Bases: object

Result of replaying the oracle actions over the whole period (check_replay).

Variables:

• steps (int) – Steps replayed.

• penalty_max (float) – Largest K_t.

• penalty_steps (int) – Steps with K_t above PENALTY_TOLERANCE (1e-9).

• soc_error_max_kwh (float) – Largest |s_t - s*_t| between the replayed and the oracle state of charge, over steps and batteries (kWh).

• simultaneous_rows (int) – Oracle rows of the period (step and battery) with both p^cha > 0 and p^dis > 0, which the replay cannot reproduce (OracleProvider.simultaneous_rows).

• clipped_rows (int) – Oracle actions (step and battery) beyond a dynamic bound by at most REPLAY_TOLERANCE_KWH (1e-6 kWh), moved onto the bound without penalty (ReplayController).

• clipped_max_kwh (float) – Largest of those violations (kWh).

• welfare (float) – Cumulative welfare of the replay Σ_t W_t (€).

• objective (float | None) – Cumulative MILP objective of the days of the period (€), or None without an objective file.

• relative_gap (float | None) – |welfare - objective| / |objective|, or None without an objective.

• rtol (float) – Relative tolerance of the welfare check.

• fit_score (float | None) – Fit Score of the replay’s daily welfare against the daily objective, or None (no objective, fewer than two days, or a constant objective).

• passed (bool) – True if no step is penalised and, with an objective, relative_gap ≤ rtol.

Parameters:

• steps (int)

• penalty_max (float)

• penalty_steps (int)

• soc_error_max_kwh (float)

• simultaneous_rows (int)

• clipped_rows (int)

• clipped_max_kwh (float)

• welfare (float)

• objective (float | None)

• relative_gap (float | None)

• rtol (float)

• fit_score (float | None)

• passed (bool)

class smartcer.gym.controllers.ReplayController(env)

Bases: Controller

Plays the optimal actions a* of the environment’s oracle (env.oracle).

An oracle action beyond a dynamic bound of the environment by at most 1e-6 kWh (REPLAY_TOLERANCE_KWH, the energy (a* - A^upper) P^cha ΔT or (-a* - A^lower) P^dis ΔT on the node side) is moved onto the bound, so that the environment does not penalise it, and counted. Larger violations are played unchanged: the environment clips and penalises them, which is what the replay check (check_replay) detects. The actions are float64.

Variables:

• clipped_rows – Actions (step and battery) of the episode moved onto a bound.

• clipped_max_kwh – Largest violation moved onto a bound in the episode (kWh).

Parameters:

env (Env)

reset()

Start a new episode: the counts of clipped actions restart from 0.

Return type:

None

act(observation, info)

Oracle actions of the step env.current_step, with violations of the dynamic bounds up to REPLAY_TOLERANCE_KWH moved onto the bounds.

Parameters:

• observation (ndarray) – Latest observation (not used).

• info (dict[str, Any]) – Latest info (not used).

Returns:

a* of the step, float64, in storage_node_ids order.

Return type:

np.ndarray

smartcer.gym.controllers.check_output_dir(output_dir, env)

Refuse output directories inside the scenario folder or a cer_data folder (S18).

Parameters:

• output_dir (str | Path) – Output directory.

• env (SmartCerEnv) – Environment whose scenario folder is protected.

Returns:

The resolved output directory.

Return type:

Path

Raises:

ValueError – If the directory lies inside the scenario folder or a cer_data folder.

smartcer.gym.controllers.check_replay(env, output_dir, rtol=0.001, seed=None)

Play the oracle actions a* over the whole period and compare the result with the MILP.

The period must consist of whole days (first step at midnight, last step ending at midnight), so that every day of the replay is compared with a whole MILP day, and it must start at the first step of the oracle file: the replay starts from soc_initial, which is the oracle’s state only there. Oracle actions beyond a dynamic bound by at most 1e-6 kWh (solver round-off) are moved onto the bound and counted (ReplayController); larger violations are clipped and penalised by the environment. Checks that no step is penalised (K_t ≤ 1e-9) and, when the oracle has a daily objective, that the cumulative welfare lies within rtol of the cumulative objective. Writes <output_dir>/oracle_daily.csv (per day: date, welfare of the replay and, with an objective, objective and difference = welfare - objective) and <output_dir>/oracle_replay.json (the ReplayCheck fields and the run configuration).

Parameters:

• env (SmartCerEnv) – Environment created with oracle, whose episode covers the whole period (no episode_length and no forecasts).

• output_dir (str | Path) – Run output directory, outside the scenario folder and any cer_data folder.

• rtol (float) – Relative tolerance on the cumulative welfare, ≥ 0 (default 1e-3).

• seed (int | None) – Seed of the reset.

Returns:

The comparison.

Return type:

ReplayCheck

Raises:

ValueError – If the environment has no oracle, its episode does not cover the whole period, the period is not made of whole days or does not start at the first step of the oracle file, rtol is negative or the output directory is not allowed.

smartcer.gym.controllers.compute_baselines(env, output_dir, controllers=('naive', 'rbc_community'), seed=None)

Run one episode per baseline controller and write <output_dir>/<controller>/: the community table, one table per node, the KPIs and the run configuration (ResultsTracker.save).

Parameters:

• env (SmartCerEnv) – The environment of the run (its period and episode settings are used).

• output_dir (str | Path) – Run output directory, outside the scenario folder and any cer_data folder.

• controllers (Iterable[str]) – Names of CONTROLLERS to run. Defaults to the naive controller and the community RBC, the class A reference baseline.

• seed (int | None) – Seed of every episode’s reset.

Returns:

Controller name → KPIs.

Return type:

dict[str, dict[str, Any]]

Raises:

ValueError – If a controller is unknown or the output directory is not allowed.

smartcer.gym.controllers.run_episode(env, controller, seed=None)

Run one episode of an environment with a controller and record it.

Parameters:

• env (Env) – A SmartCerEnv (possibly wrapped, but not with TrackResults).

• controller (Controller) – Controller bound to the same environment.

• seed (int | None) – Seed of env.reset.

Returns:

The tracker holding the completed episode.

Return type:

ResultsTracker

Raises:

ValueError – If the controller is bound to another environment.

Tracking and metrics

Results tracking: community and node tables of a run, CSV export, Stable-Baselines3 callback.

class smartcer.gym.tracking.ResultsCallback(*args, **kwargs)

Bases: BaseCallback

Logs the KPIs of each completed episode of the first environment and saves its tables.

The training environment must be wrapped with TrackResults and vectorised in the same process (e.g. DummyVecEnv, the default of make_vec_env).

Variables:

• output_dir – Folder of the saved episodes (episode_<n>/, see ResultsTracker.save).

• save_freq – Save every save_freq completed episodes; 0 never saves.

• run_config – Run configuration written with every saved episode, if given.

Parameters:

• output_dir (str | Path)

• save_freq (int)

• run_config (dict[str, Any] | None)

• verbose (int)

class smartcer.gym.tracking.ResultsTracker(community, oracle=None)

Bases: object

Collects the step info of a community environment into tables.

Variables:

• node_ids – Node identifiers, in community order.

• capacities – Battery node id → capacity S^max (kWh).

• soc_min – Battery node id → minimum state of charge S^min (kWh).

• oracle – MILP oracle whose state of charge the battery tables are compared with, or None.

• episode – Number of the current episode (incremented at every reset).

• episodes_completed – Number of episodes that reached their last step.

Parameters:

• community (Community)

• oracle (OracleProvider | None)

on_reset()

Start a new episode; the rows of an unfinished episode are dropped.

Return type:

None

record(info, done)

Record one step.

Parameters:

• info (dict[str, Any]) – Info of step().

• done (bool) – True on the last step of the episode, which completes it.

Return type:

None

community_table()

Community table of the last completed episode.

Returns:

One row per step: datetime, episode, t, reward, reward_main, reward_penalty (σ-weighted), welfare and its components (€, kWh), penalty (K_t), obs_clipped.

Return type:

pd.DataFrame

node_tables()

Node tables of the last completed episode.

Returns:

Node id → one row per step with datetime, episode, t, the NodeStep fields, with per-node rewards reward, reward_main, reward_penalty and, for a battery node with an oracle, oracle_soc_kwh (the oracle state of charge at the end of step t, the instant of soc_kwh) and soc_gap_kwh (soc_kwh - oracle_soc_kwh).

Return type:

dict[str, pd.DataFrame]

kpis()

KPIs of the last completed episode (smartcer.gym.metrics.compute_kpis).

Returns:

The KPIs.

Return type:

dict[str, Any]

save(output_dir, run_config=None)

Write the last completed episode: community.csv, nodes/<node_id>.csv, kpis.json and, if given, run_config.json.

Parameters:

• output_dir (str | Path) – Output folder, created if missing.

• run_config (dict[str, Any] | None) – Run configuration (e.g. SmartCerEnv.run_config()).

Returns:

The output folder.

Return type:

Path

class smartcer.gym.tracking.TrackResults(env)

Bases: Wrapper

Wrapper that records every step of a SmartCerEnv in a ResultsTracker.

Variables:

tracker – The tracker (reachable through env.get_wrapper_attr(‘tracker’)).

Parameters:

env (Env)

reset(*, seed=None, options=None)

Reset the environment and start a new episode in the tracker.

Parameters:

• seed (int | None) – Seed passed to the environment.

• options (dict[str, Any] | None) – Options passed to the environment.

Returns:

Observation and info of the environment.

Return type:

tuple[Any, dict[str, Any]]

step(action)

Step the environment and record the step.

Parameters:

action (Any) – Action passed to the environment.

Returns:

The environment’s step result.

Return type:

tuple[Any, SupportsFloat, bool, bool, dict[str, Any]]

smartcer.gym.tracking.write_json(path, data)

Write a dictionary as indented JSON (numpy scalars converted).

Parameters:

• path (Path) – File path.

• data (dict[str, Any]) – JSON-serialisable data, numpy scalars allowed.

Return type:

None

Metrics of a run: KPIs from the tracker tables and the Fit Score (formulas in KPI.md).

smartcer.gym.metrics.compute_kpis(community, nodes, capacities, soc_min=None)

KPIs of one run.

Parameters:

• community (DataFrame) – Community table (one row per step) with the columns of COMMUNITY_COLUMNS.

• nodes (Mapping[str, DataFrame]) – Node id → node table; only the battery nodes of capacities are read, with the columns of NODE_COLUMNS.

• capacities (Mapping[str, float]) – Battery node id → capacity S^max (kWh).

• soc_min (Mapping[str, float] | None) – Battery node id → minimum state of charge S^min (kWh), for the cycles on the usable capacity S^max - S^min; nodes not listed (or None) have S^min = 0.

Returns:

steps; sums of reward, welfare, revenue, cost, storage_cost, incentive (€) and shared_energy_kwh, export_kwh, import_kwh (kWh); sharing_rate = shared / exported and coverage_rate = shared / imported (fractions); throughput_kwh, equivalent_full_cycles (on S^max) and efc_usable (on S^max - S^min) of all batteries; penalty_mean, penalty_max, penalty_steps of K_t; batteries: per battery node, charge_kwh, discharge_kwh, throughput_kwh, equivalent_full_cycles, efc_usable, the penalty statistics of k_u and, when its table has soc_gap_kwh (runs with an oracle), soc_gap_mean_kwh and soc_gap_max_kwh: mean and maximum of the absolute gap between the battery’s and the oracle’s state of charge.

Return type:

dict[str, Any]

Raises:

ValueError – If a column or a battery node table is missing, a capacity is not positive or a minimum state of charge is outside [0, S^max).

smartcer.gym.metrics.daily_welfare(community)

Welfare per calendar day of a community table.

Parameters:

community (DataFrame) – Community table with the columns datetime and welfare.

Returns:

Sum of W_t per day (€), indexed by date.

Return type:

pd.Series

Raises:

ValueError – If a column is missing.

smartcer.gym.metrics.fit_score(reference, candidate)

Fit Score F = (1 - Σ_d (T2_d - T1_d)² / Σ_d (T1_d - mean(T1))²) · 100.

100 when the candidate equals the reference; 0 when it is as far from the reference as the reference’s mean; negative when it is farther.

Parameters:

• reference (ArrayLike) – Daily series T1 (e.g. the welfare of the MILP oracle).

• candidate (ArrayLike) – Daily series T2 of the same days (e.g. the welfare of a controller).

Returns:

F (percentage).

Return type:

float

Raises:

ValueError – If the series are empty, of different lengths or not finite, or the reference is constant (F undefined).

Forecasts

Forecast providers for the observation of SmartCerEnv (forecast and forecast_horizon).

class smartcer.forecasts.ForecastProvider(*args, **kwargs)

Bases: Protocol

Source of forecasts of generation and load, keyed ‘<node_id>/<quantity>’ for a node and ‘community/<quantity>’ for the totals of the nodes without a battery.

reset(rng)

Start an episode.

Parameters:

rng (Generator) – The environment’s random generator (env.np_random).

Return type:

None

get(t, horizon)

Forecasts of the steps t, …, t + horizon - 1.

The environment asks for the steps after the last observed one (docs/FORMULATION.md section 12): t + 1, …, t + H with observe_current_step=True, t, …, t + H - 1 with False; the k-th value (k = 0, 1, …) is therefore the forecast of lead time k + 1 steps.

Parameters:

• t (int) – First forecast step.

• horizon (int) – Number of steps H, ≥ 1.

Returns:

Key → H values (kWh per step). The environment needs at least the keys of forecast_series: generation and load of every battery node and the community totals.

Return type:

dict[str, np.ndarray]

class smartcer.forecasts.MMFEForecast(community, parameters='standard')

Bases: object

Forecast provider with MMFE errors (module docstring) on the series of forecast_series: generation and load of every battery node and the community totals.

Variables:

• parameters – The error parameters.

• preset – Name of the preset in MMFE_PRESETS, or None for custom parameters.

• keys – Forecast keys, in order.

• n_steps – Number of steps of the community period.

Parameters:

• community (Community)

• parameters (MMFEParameters | str)

config()

The provider settings, JSON-serialisable, for the run configuration.

Returns:

preset (or None) and the parameters.

Return type:

dict[str, Any]

reset(rng)

Start an episode: new innovations, drawn from a generator spawned from rng.

Parameters:

rng (Generator) – The environment’s random generator.

Return type:

None

get(t, horizon)

Forecasts of the steps t, …, t + horizon - 1, with lead times 1, …, horizon.

The innovations of a target step are drawn the first time it is forecast in the episode, so later issues for it are consistent, as long as the issues move forward in time (a request before the steps drawn so far, or after a gap, starts a new chain of draws).

Parameters:

• t (int) – First forecast step.

• horizon (int) – Number of steps H, ≥ 1, at most the horizon of the first request of the episode.

Returns:

keys → H forecast values (kWh per step).

Return type:

dict[str, np.ndarray]

Raises:

• RuntimeError – If called before reset.

• ValueError – If the horizon is not positive or exceeds the episode’s, or the steps leave the period.

class smartcer.forecasts.MMFEParameters(sigma_generation, sigma_load, rho=0.0, phi=0.0, bias_generation=0.0, bias_load=0.0, sigma_overrides=<factory>)

Bases: object

Parameters of the MMFE forecast errors.

Variables:

• sigma_generation (float) – σ of the generation innovations, as a fraction of the reference profile.

• sigma_load (float) – σ of the load innovations, as a fraction of the load.

• rho (float) – Share ρ of the innovation variance due to the factor common to all series of a quantity, in [0, 1].

• phi (float) – AR(1) coefficient φ of the common factor over the target steps, in (-1, 1).

• bias_generation (float) – Mean b of a generation innovation, as a fraction of the reference profile; the mean error at lead time h is h b.

• bias_load (float) – Mean b of a load innovation, as a fraction of the load.

• sigma_overrides (collections.abc.Mapping[str, float]) – Forecast key (e.g. ‘community/load’, ‘<node_id>/generation’) → σ of that series, replacing the σ of its quantity. An aggregated load (e.g. many consumers summed in community/load) should have a smaller relative σ than a single member: the independent parts of the members’ errors partly cancel.

Parameters:

• sigma_generation (float)

• sigma_load (float)

• rho (float)

• phi (float)

• bias_generation (float)

• bias_load (float)

• sigma_overrides (Mapping[str, float])

class smartcer.forecasts.PerfectForecast(community)

Bases: object

Forecast provider that returns the true generation and load of every node and the community totals of the nodes without a battery (community_totals).

Generation is zero for a consumer and load is zero for a producer, as in the node data.

Variables:

n_steps – Number of steps of the community period.

Parameters:

community (Community)

reset(rng)

Start an episode; a perfect forecast is deterministic and ignores the generator.

Parameters:

rng (Generator) – The environment’s random generator.

Return type:

None

get(t, horizon)

True values of the steps t, …, t + horizon - 1.

Parameters:

• t (int) – First forecast step.

• horizon (int) – Number of steps H, ≥ 1.

Returns:

‘<node_id>/generation’ and ‘<node_id>/load’ of every node and the community totals → copies of the H values (kWh per step).

Return type:

dict[str, np.ndarray]

Raises:

ValueError – If the horizon is not positive or the steps leave the period.

smartcer.forecasts.check_steps(t, horizon, n_steps)

Check that the forecast steps t, …, t + horizon - 1 lie in the period.

Parameters:

• t (int) – First forecast step.

• horizon (int) – Number of steps.

• n_steps (int) – Number of steps of the period.

Raises:

ValueError – If the horizon is not positive or the steps leave the period.

Return type:

None

smartcer.forecasts.community_totals(community)

Total generation and load of the nodes without a battery (e.g. aggregated consumers).

Parameters:

community (Community) – The community.

Returns:

community/generation (producers and prosumers) and community/load (consumers and prosumers), in kWh per step, each only if such nodes exist.

Return type:

dict[str, np.ndarray]

smartcer.forecasts.forecast_series(community)

The series the observation forecasts, over the whole community period.

Parameters:

community (Community) – The community.

Returns:

<node_id>/generation and <node_id>/load of every battery node, in storage_node_ids order, then the totals of community_totals (kWh per step).

Return type:

dict[str, np.ndarray]

MILP oracle

MILP oracle for the critic of SmartCerEnv (oracle) and the ReplayController.

class smartcer.oracle.OracleData(p_cha_kw, p_dis_kw, soc_kwh, daily_objective, source)

Bases: object

MILP outputs, independent of the file format.

The three tables have the same sorted timestamps (rows) and node ids (columns).

Variables:

• p_cha_kw (pandas.DataFrame) – Charge power of each step, node side (kW), ≥ 0.

• p_dis_kw (pandas.DataFrame) – Discharge power of each step, node side (kW), ≥ 0.

• soc_kwh (pandas.DataFrame) – State of charge at the end of each step (kWh).

• daily_objective (pandas.Series | None) – Objective per calendar day (€, welfare), indexed by datetime.date, or None.

• source (str) – Where the data come from (the oracle folder).

Parameters:

• p_cha_kw (DataFrame)

• p_dis_kw (DataFrame)

• soc_kwh (DataFrame)

• daily_objective (Series | None)

• source (str)

class smartcer.oracle.OracleProvider(data, community, soc_initial_tolerance_kwh=0.001)

Bases: object

The oracle trajectory of a community: optimal actions and states of charge per step.

Variables:

• source – Where the data come from (the oracle folder).

• first_timestamp – First step of the oracle file, whose state before the step is S^ini.

• node_ids – Battery nodes, in action order (community.storage_node_ids).

• timestamps – Steps of the community period.

• actions – Optimal actions a* = p^cha / P^cha - p^dis / P^dis, shape (n_steps, N), in [-1, 1].

• soc_kwh – Oracle state of charge at the end of each step (kWh), shape (n_steps, N).

• soc_start_kwh – Oracle state of charge at the start of each step (kWh), shape (n_steps, N): the scenario’s S^ini at the first step of the oracle file, else the state at the end of the previous step.

• daily_objective – MILP objective per calendar day of the period (€), indexed by datetime.date, or None.

• simultaneous_rows – Rows of the period (step and node) with both p^cha > 0 and p^dis > 0.

Parameters:

• data (OracleData)

• community (Community)

• soc_initial_tolerance_kwh (float)

classmethod from_folder(folder, community, soc_initial_tolerance_kwh=0.001)

Read an oracle folder and align it with a community.

Parameters:

• folder (str | Path) – Oracle folder (read_oracle).

• community (Community) – The community the oracle was computed for.

• soc_initial_tolerance_kwh (float) – Largest accepted difference between the inferred initial state of charge and the scenario’s (kWh), ≥ 0.

Returns:

The provider.

Return type:

OracleProvider

action(t)

Optimal actions of step t.

Parameters:

t (int) – Step index.

Returns:

a*_t, one value per battery in action order (float64 copy).

Return type:

np.ndarray

soc(t)

Oracle state of charge at the start of step t (before a*_t), comparable with the state of charge an agent observes for step t.

Parameters:

t (int) – Step index.

Returns:

One value per battery in action order (kWh, float64 copy).

Return type:

np.ndarray

exception smartcer.oracle.OracleWarning

Bases: UserWarning

The oracle data are readable but the replay cannot reproduce them exactly (e.g. simultaneous charge and discharge).

smartcer.oracle.read_oracle(folder)

Read the MILP outputs of an oracle folder (the adapter of the oracle file format).

This is the only function that knows the file format (module docstring); when the format of the oracle files changes, only this function changes.

Parameters:

folder (str | Path) – Oracle folder with trajectory.csv and optionally objective.csv.

Returns:

Powers and states of charge per step and node, daily objective.

Return type:

OracleData

Raises:

• FileNotFoundError – If trajectory.csv is missing.

• ValueError – On a missing column, a value that is not a timestamp or a finite number, a power below -1e-9 kW, a duplicated (datetime, node_id) pair, a step that lacks a node, or a duplicated day in objective.csv; the message names the file and the first offending line.

Scenario data

Scenario configuration and loading (docs/CONFIG.md).

class smartcer.data.ScenarioConfig(name, path, format_version, delta_t, units, incentive, nodes)

Bases: object

Parsed scenario configuration.

Variables:

• name (str) – Scenario name (the name key, or the folder name).

• path (pathlib.Path) – Scenario folder.

• format_version (str) – ‘1.0’ or ‘0.6.2’.

• delta_t (float) – Duration of a step (hours).

• units (str) – Units of the CSV energies, ‘kWh’ (per step) or ‘kW’ (average power).

• incentive (smartcer.data.schema.IncentiveConfig) – Incentive model configuration.

• nodes (tuple[smartcer.data.schema.NodeConfig, ...]) – Node configurations in scenario order.

Parameters:

• name (str)

• path (Path)

• format_version (str)

• delta_t (float)

• units (str)

• incentive (IncentiveConfig)

• nodes (tuple[NodeConfig, ...])

exception smartcer.data.TimestampWarning

Bases: UserWarning

Irregular timestamps in a 0.6.2 scenario, whose rows are aligned by position as in 0.6.2.

smartcer.data.load_scenario(path, start=None, end=None)

Load a scenario in the 1.0 or the 0.6.2 format.

Parameters:

• path (str | Path) – Scenario folder (with config.json).

• start (str | Timestamp | datetime64 | None) – First timestamp of the period, inclusive; None for the first row.

• end (str | Timestamp | datetime64 | None) – Last timestamp of the period, inclusive; None for the last row.

Returns:

Nodes, incentive model and timestamps of the period.

Return type:

Community

Raises:

• FileNotFoundError – If config.json or a CSV file does not exist.

• ValueError – If the configuration or the data are invalid (see the module docstring).

smartcer.data.parse_config(raw, scenario_dir)

Parse the content of config.json in the 1.0 or the 0.6.2 format.

The 1.0 format has format_version: “1.0”; a configuration without it and with Producers, Consumers or Prosumers is read as 0.6.2, with a DeprecationWarning that also lists the keys that are not used.

Parameters:

• raw (Any) – Decoded JSON content.

• scenario_dir (Path) – Scenario folder (relative CSV paths are resolved against it).

Returns:

The parsed configuration.

Return type:

ScenarioConfig

Raises:

ValueError – If the configuration is invalid; the message names the key and the node.

smartcer.data.read_config(path)

Read and parse <path>/config.json.

Parameters:

path (str | Path) – Scenario folder.

Returns:

The parsed configuration.

Return type:

ScenarioConfig

Raises:

• FileNotFoundError – If config.json does not exist.

• ValueError – If the file is not valid JSON or the configuration is invalid.

Core: physics and economics

Physics and economics of SmartCer 1.0 (docs/FORMULATION.md), independent of Gymnasium.

Modules in this package must not import gymnasium or smartcer.gym.

class smartcer.core.Battery(capacity, power_charge_max, power_discharge_max, efficiency_charge, efficiency_discharge, soc_min=0.0, soc_initial=None, alpha=0.0, usage_cost=0.0, charge_limit='production')

Bases: StorageSystem

Battery with node-side power limits, efficiencies, minimum state of charge and self-discharge.

The state of charge s is kept in kWh; soc_min and soc_initial are given as fractions of the capacity. Each step: self-discharge s̃ = max(S_min, (1 - α)^Δt · s), bounds from s̃, clipping of the action, update s = s̃ + η_cha · p_cha · Δt - p_dis · Δt / η_dis.

Variables:

• capacity – Capacity S^max (kWh).

• power_charge_max – Maximum charge power P^cha, node side (kW).

• power_discharge_max – Maximum discharge power P^dis, node side (kW).

• efficiency_charge – Charge efficiency η^cha, in (0, 1].

• efficiency_discharge – Discharge efficiency η^dis, in (0, 1].

• soc_min – Minimum state of charge as a fraction of capacity, in [0, 1).

• soc_initial – Initial state of charge as a fraction of capacity, in [soc_min, 1].

• alpha – Self-discharge per hour α (scenario key storage.alpha), in [0, 1): fraction of the state of charge lost in one hour; over a step of Δt hours the battery keeps (1 - α)^Δt of its charge, i.e. α_step = 1 - (1 - α)^Δt.

• usage_cost – Unit cost of storage usage π^sto (€/kWh of cell throughput), ≥ 0.

• charge_limit – Charge rule used by the node to compute the charge cap: ‘production’ (c = g) or ‘surplus’ (c = max(0, g - l)); the battery only stores it.

• soc_min_kwh – Minimum state of charge S^min (kWh).

• soc_initial_kwh – Initial state of charge S^ini (kWh).

• soc_kwh – Current state of charge s (kWh).

Parameters:

• capacity (float)

• power_charge_max (float)

• power_discharge_max (float)

• efficiency_charge (float)

• efficiency_discharge (float)

• soc_min (float)

• soc_initial (float | None)

• alpha (float)

• usage_cost (float)

• charge_limit (str)

Example

>>> battery = Battery(capacity=10.0, power_charge_max=5.0, power_discharge_max=5.0,
...                   efficiency_charge=0.95, efficiency_discharge=0.95, soc_initial=0.5)
>>> result = battery.step(action=1.0, delta_t=1.0, charge_cap_kwh=None)
>>> round(result.soc_kwh, 2), round(result.charge_kwh, 2)
(9.75, 5.0)

classmethod from_config(config)

Create a battery from a scenario storage section, in the 1.0 or the 0.6.2 format.

1.0 keys: see CONFIG_KEYS (docs/FORMULATION.md section 3). 0.6.2 keys are mapped with a DeprecationWarning: rated_power → both power limits, efficiency → both efficiencies, alfa → alpha; beta (degradation) is read and ignored; soc_min is a fraction in both formats.

Parameters:

config (dict[str, Any]) – Storage section of a node.

Returns:

The battery.

Return type:

Battery

Raises:

ValueError – On unknown keys, on a 0.6.2 key combined with the 1.0 key it maps to, on a missing required key, or on invalid values.

get_bounds(delta_t, charge_cap_kwh)

Dynamic action bounds A^upper, A^lower for the coming step (docs/FORMULATION.md section 6).

Computed from the current state of charge after self-discharge; the state is not changed.

Parameters:

• delta_t (float) – Duration of the step (hours), > 0.

• charge_cap_kwh (float | None) – Maximum energy the node can supply for charging in the step (kWh), ≥ 0; None means no cap.

Returns:

(a_upper, a_lower), both in [0, 1]; feasible actions lie in [-a_lower, a_upper].

Return type:

tuple[float, float]

Raises:

ValueError – If delta_t or charge_cap_kwh is invalid.

step(action, delta_t, charge_cap_kwh)

Clip the action to its bounds, apply it for one step and update the state of charge.

Parameters:

• action (float) – Requested normalised action in [-1, 1]; positive charges, negative discharges.

• delta_t (float) – Duration of the step (hours), > 0.

• charge_cap_kwh (float | None) – Maximum energy the node can supply for charging in the step (kWh), ≥ 0; None means no cap.

Returns:

Requested and applied action, bounds, powers, energies, cost, penalty and the new state of charge.

Return type:

BatteryStep

Raises:

• ValueError – If the action is not finite or outside [-1, 1], or if delta_t or charge_cap_kwh is invalid.

• RuntimeError – If the new state of charge leaves [S_min, S_max] by more than the rounding tolerance (an implementation error).

reset()

Restore the initial state of charge S^ini.

Return type:

None

class smartcer.core.BatteryStep(action_requested, action_applied, a_upper, a_lower, p_cha_kw, p_dis_kw, charge_kwh, discharge_kwh, throughput_kwh, storage_cost, penalty, soc_kwh)

Bases: object

Record of one battery step (docs/FORMULATION.md sections 4-7 and 10).

Variables:

• action_requested (float) – Action a requested by the agent, in [-1, 1].

• action_applied (float) – Action after clipping to [-a_lower, a_upper].

• a_upper (float) – Upper action bound A^upper of the step, in [0, 1].

• a_lower (float) – Lower action bound A^lower of the step, in [0, 1].

• p_cha_kw (float) – Charge power p^cha, node side (kW).

• p_dis_kw (float) – Discharge power p^dis, node side (kW).

• charge_kwh (float) – Energy supplied by the node to the battery, p^cha · Δt (kWh).

• discharge_kwh (float) – Energy delivered by the battery to the node, p^dis · Δt (kWh).

• throughput_kwh (float) – Energy entering or leaving the cell, η^cha · p^cha · Δt + p^dis · Δt / η^dis (kWh).

• storage_cost (float) – Storage usage cost π^sto · throughput (€).

• penalty (float) – Constraint penalty k of the action (section 7), ≥ 0.

• soc_kwh (float) – State of charge at the end of the step (kWh).

Parameters:

• action_requested (float)

• action_applied (float)

• a_upper (float)

• a_lower (float)

• p_cha_kw (float)

• p_dis_kw (float)

• charge_kwh (float)

• discharge_kwh (float)

• throughput_kwh (float)

• storage_cost (float)

• penalty (float)

• soc_kwh (float)

class smartcer.core.Community(nodes, incentive, timestamps, delta_t, name='')

Bases: object

Nodes, incentive model and time axis of a scenario.

Variables:

• name – Scenario name.

• nodes – Nodes in a fixed order.

• node_ids – Node identifiers in the same order.

• storage_node_ids – Identifiers of the nodes with a battery, in node order; this is the order of the actions.

• incentive – Incentive model.

• timestamps – Start time of each step, numpy datetime64 array.

• delta_t – Duration of a step (hours).

Parameters:

• nodes (Sequence[Node])

• incentive (IncentiveModel)

• timestamps (ndarray)

• delta_t (float)

• name (str)

property n_steps: int

Number of steps of the scenario.

node(node_id)

Node by identifier.

Parameters:

node_id (str) – Node identifier.

Returns:

The node.

Return type:

Node

Raises:

KeyError – If the node does not exist.

reset()

Restore every battery’s initial state of charge.

Return type:

None

step(t, actions=None)

Apply one action per battery in step t and compute the community results.

Parameters:

• t (int) – Step index.

• actions (ArrayLike | None) – One action in [-1, 1] per battery, in storage_node_ids order; None leaves every battery idle.

Returns:

Node records, welfare breakdown and mean penalty.

Return type:

CommunityStep

Raises:

ValueError – If the number of actions differs from the number of batteries.

class smartcer.core.CommunityStep(t, nodes, welfare, penalty)

Bases: object

Record of one community step.

Variables:

• t (int) – Step index.

• nodes (tuple[smartcer.core.nodes.node.NodeStep, ...]) – One record per node, in the community’s node order.

• welfare (smartcer.core.community.welfare.WelfareBreakdown) – Welfare breakdown of the step.

• penalty (float) – Mean constraint penalty over the nodes with a battery, K_t = Σ_{u ∈ U_b} k_u / |U_b| (0 without batteries).

Parameters:

• t (int)

• nodes (tuple[NodeStep, ...])

• welfare (WelfareBreakdown)

• penalty (float)

class smartcer.core.Node(node_id, node_type, delta_t, generation_kwh, load_kwh, sell_price, buy_price, battery=None, photovoltaic=None, category=None)

Bases: object

Producer, consumer or prosumer with an optional battery.

Variables:

• node_id – Node identifier.

• node_type – ‘producer’, ‘consumer’ or ‘prosumer’.

• delta_t – Duration of a step (hours).

• generation_kwh – Generation per step g (kWh), zero for a consumer.

• load_kwh – Load per step l (kWh), zero for a producer.

• sell_price – Sell price π^egr per step (€/kWh); zero for a consumer without the column.

• buy_price – Buy price π^igr per step (€/kWh); zero for a producer.

• battery – The node’s battery, or None.

• photovoltaic – TIP parameters of the node’s plant (tip_base, tip_threshold, fi_zone, pnrr, …), or None.

• category – Free-text category from the scenario (0.6.2 Type, e.g. ‘factory’), or None.

Parameters:

• node_id (str)

• node_type (str)

• delta_t (float)

• generation_kwh (ndarray)

• load_kwh (ndarray)

• sell_price (ndarray)

• buy_price (ndarray)

• battery (Battery | None)

• photovoltaic (dict[str, Any] | None)

• category (str | None)

property n_steps: int

Number of steps of the node’s data.

charge_cap(t)

Charge cap c of step t (docs/FORMULATION.md section 6).

production: c = g; surplus: c = max(0, g - l).

Parameters:

t (int) – Step index.

Returns:

c (kWh); 0 for a node without a battery.

Return type:

float

get_bounds(t)

Dynamic action bounds of step t for the current state of charge.

Parameters:

t (int) – Step index.

Returns:

(a_upper, a_lower) in [0, 1]; (0, 0) without a battery.

Return type:

tuple[float, float]

step(t, action=None)

Apply the battery action of step t and compute the node’s flows and economics.

Parameters:

• t (int) – Step index.

• action (float | None) – Battery action in [-1, 1]; None (or 0) leaves the battery idle. Must be None or 0 for a node without a battery.

Returns:

Flows, economics and battery record of the step.

Return type:

NodeStep

Raises:

• IndexError – If t is outside [0, n_steps).

• ValueError – If a node without a battery receives a non-zero action, or the battery rejects the action.

reset()

Restore the battery’s initial state of charge (the data are not reloaded).

Return type:

None

class smartcer.core.NodeStep(node_id, t, generation_kwh, load_kwh, p_cha_kw, p_dis_kw, charge_kwh, discharge_kwh, net_kwh, export_kwh, import_kwh, revenue, cost, throughput_kwh, storage_cost, penalty, soc_kwh, action_requested, action_applied, a_upper, a_lower)

Bases: object

Record of one node step (docs/FORMULATION.md sections 4-8 and 10).

For a node without a battery the battery fields are zero and soc_kwh is None.

Variables:

• node_id (str) – Node identifier.

• t (int) – Step index.

• generation_kwh (float) – Generation g (kWh).

• load_kwh (float) – Load l (kWh).

• p_cha_kw (float) – Charge power p^cha, node side (kW).

• p_dis_kw (float) – Discharge power p^dis, node side (kW).

• charge_kwh (float) – Energy supplied by the node to its battery, p^cha · Δt (kWh).

• discharge_kwh (float) – Energy delivered by the battery to the node, p^dis · Δt (kWh).

• net_kwh (float) – Net energy exchanged with the grid n = g - l + p^dis Δt - p^cha Δt (kWh).

• export_kwh (float) – Energy fed into the grid e^gri = max(0, n) (kWh).

• import_kwh (float) – Energy withdrawn from the grid i^gri = max(0, -n) (kWh).

• revenue (float) – π^egr · e^gri (€).

• cost (float) – π^igr · i^gri (€).

• throughput_kwh (float) – Energy entering or leaving the cell (kWh).

• storage_cost (float) – π^sto · throughput (€).

• penalty (float) – Constraint penalty k of the battery action (section 7).

• soc_kwh (float | None) – State of charge at the end of the step (kWh), None without a battery.

• action_requested (float) – Requested battery action.

• action_applied (float) – Battery action after clipping.

• a_upper (float) – Upper action bound A^upper of the step.

• a_lower (float) – Lower action bound A^lower of the step.

Parameters:

• node_id (str)

• t (int)

• generation_kwh (float)

• load_kwh (float)

• p_cha_kw (float)

• p_dis_kw (float)

• charge_kwh (float)

• discharge_kwh (float)

• net_kwh (float)

• export_kwh (float)

• import_kwh (float)

• revenue (float)

• cost (float)

• throughput_kwh (float)

• storage_cost (float)

• penalty (float)

• soc_kwh (float | None)

• action_requested (float)

• action_applied (float)

• a_upper (float)

• a_lower (float)

class smartcer.core.StorageSystem

Bases: ABC

Abstract class of an energy storage system controlled by a normalised action.

Implementations keep their own state (e.g. the state of charge) between steps; the caller provides the time step and, for each step, the cap on the energy that may be charged.

abstractmethod get_bounds(delta_t, charge_cap_kwh)

Dynamic action bounds for the coming step, given the current state.

Parameters:

• delta_t (float) – Duration of the step (hours).

• charge_cap_kwh (float | None) – Maximum energy the node can supply for charging in the step (kWh); None means no cap.

Returns:

(a_upper, a_lower), both in [0, 1]; feasible actions lie in [-a_lower, a_upper].

Return type:

tuple[float, float]

abstractmethod step(action, delta_t, charge_cap_kwh)

Apply one action for one step and update the state.

Parameters:

• action (float) – Requested normalised action in [-1, 1].

• delta_t (float) – Duration of the step (hours).

• charge_cap_kwh (float | None) – Maximum energy the node can supply for charging in the step (kWh); None means no cap.

Returns:

Implementation-specific record of the step.

Return type:

Any

abstractmethod reset()

Restore the initial state.

Return type:

None

class smartcer.core.TipIncentive(coefficients)

Bases: IncentiveModel

tip model (the 0.6.2 per-plant model) with per-step, per-node coefficients.

Variables:

coefficients – TIP coefficient per step and node (€/kWh), shape (n_steps, n_nodes), 0 for nodes without a plant.

Parameters:

coefficients (ArrayLike)

potential(t, export_kwh)

Potential TIP of step t, Σ_u c_u e^gri_u (the 0.6.2 tip_potential).

Parameters:

• t (int) – Step index.

• export_kwh (ndarray) – Energy fed into the grid by each node (kWh).

Returns:

Potential TIP (€).

Return type:

float

compute(t, shared_kwh, export_kwh)

Incentive of step t, E^sh / Σ e^gri · Σ c e^gri.

Parameters:

• t (int) – Step index.

• shared_kwh (float) – Shared energy of the step (kWh).

• export_kwh (ndarray) – Energy fed into the grid by each node (kWh).

Returns:

Incentive (€).

Return type:

float

class smartcer.core.UniformIncentive(unit_value, n_steps)

Bases: IncentiveModel

uniform model with a constant or per-step unit value π^inc_t.

Variables:

unit_value – π^inc per step (€/kWh), shape (n_steps,).

Parameters:

• unit_value (float | ArrayLike)

• n_steps (int)

compute(t, shared_kwh, export_kwh)

Incentive of step t, π^inc_t · E^sh_t.

Parameters:

• t (int) – Step index.

• shared_kwh (float) – Shared energy of the step (kWh).

• export_kwh (ndarray) – Energy fed into the grid by each node (kWh); not used.

Returns:

Incentive (€).

Return type:

float

class smartcer.core.WelfareBreakdown(revenue, cost, storage_cost, incentive, shared_energy_kwh, export_kwh, import_kwh, total)

Bases: object

Welfare of one step and its components.

Variables:

• revenue (float) – Σ_u π^egr_u e^gri_u (€).

• cost (float) – Σ_u π^igr_u i^gri_u (€).

• storage_cost (float) – Σ_u π^sto_u · cell throughput (€).

• incentive (float) – Incentive on shared energy I (€).

• shared_energy_kwh (float) – Shared energy E^sh (kWh).

• export_kwh (float) – Σ_u e^gri_u (kWh).

• import_kwh (float) – Σ_u i^gri_u (kWh).

• total (float) – Welfare W = revenue - cost - storage_cost + incentive (€).

Parameters:

• revenue (float)

• cost (float)

• storage_cost (float)

• incentive (float)

• shared_energy_kwh (float)

• export_kwh (float)

• import_kwh (float)

• total (float)