added capacity tariff feature

[mk2lehe]

Summary by Sourcery

Introduce configurable capacity tariff support to the optimization workflow, including runtime configuration, optimization penalties, constraints, and reporting of tariff-related results.

New Features:

• Add optional capacity tariff configuration to optimization, supporting scalar and time-series power thresholds with configurable penalties.
• Expose capacity tariff settings (enable flag, threshold, penalty) via runtime parameters, with validation and sensible defaults.
• Include capacity tariff excess power, thresholds, and penalty costs in the optimization outputs and incorporate them into the profit objective.

Enhancements:

• Add logging and basic validation for capacity tariff configuration, including handling of mismatched list lengths and irregular time indices.

[sourcery-ai]

Reviewer's Guide

Implements a configurable capacity tariff feature in the optimization workflow, including runtime parameter handling, objective/constraints integration, and reporting of tariff-related results.

Sequence diagram for capacity tariff-aware optimization run

sequenceDiagram
    actor User
    participant ClientAPI
    participant RuntimeParamsProcessor
    participant Optimization
    participant LPModel

    User ->> ClientAPI: Submit optimization request with runtimeparams
    ClientAPI ->> RuntimeParamsProcessor: treat_runtimeparams(runtimeparams, params, forecast_dates, logger)
    RuntimeParamsProcessor ->> RuntimeParamsProcessor: Derive set_capacity_tariff
    RuntimeParamsProcessor ->> RuntimeParamsProcessor: Derive capacity_tariff_threshold (scalar or list)
    RuntimeParamsProcessor ->> RuntimeParamsProcessor: Derive capacity_tariff_penalty
    RuntimeParamsProcessor ->> ClientAPI: Updated params.optim_conf

    ClientAPI ->> Optimization: __init__(optim_conf, plant_conf, logger, lp_solver, lp_solver_path, num_threads)
    Optimization ->> Optimization: Read set_capacity_tariff
    Optimization ->> Optimization: Initialize capacity_threshold, capacity_threshold_is_list, capacity_penalty

    ClientAPI ->> Optimization: perform_optimization(data_opt, def_total_timestep, def_total_hours)
    Optimization ->> Optimization: Prepare capacity_threshold_array if list
    Optimization ->> Optimization: Validate threshold length vs data_opt index
    Optimization ->> Optimization: Build objective (base terms)
    Optimization ->> Optimization: Create P_excess[i] variables
    Optimization ->> Optimization: Add capacity_penalty_term to objective
    Optimization ->> LPModel: setObjective(objective)
    Optimization ->> LPModel: Add capacity tariff constraints
    Optimization ->> LPModel: solve()
    LPModel -->> Optimization: Solution with P_excess[i].varValue

    Optimization ->> Optimization: Build excess_values and penalty_values
    Optimization ->> Optimization: Add capacity_threshold, P_excess, capacity_penalty_cost to opt_tp
    Optimization ->> Optimization: Compute cost_fun_profit including penalty_per_timestep

    Optimization -->> ClientAPI: Optimization results DataFrame opt_tp
    ClientAPI -->> User: Return results with capacity tariff metrics

Loading

Class diagram for Optimization and runtime capacity tariff handling

classDiagram
    class Optimization {
        <<class>>
        +logger
        +lp_solver
        +lp_solver_path
        +num_threads
        +optim_conf
        +plant_conf
        +set_capacity_tariff : bool
        +capacity_threshold : float or list~float~
        +capacity_threshold_is_list : bool
        +capacity_penalty : float
        +__init__(optim_conf, plant_conf, logger, lp_solver, lp_solver_path, num_threads)
        +perform_optimization(data_opt, def_total_timestep, def_total_hours)
    }

    class CapacityTariffRuntimeConfig {
        <<class>>
        +set_capacity_tariff : bool
        +capacity_tariff_threshold : float or list~float~
        +capacity_tariff_penalty : float
    }

    class RuntimeParamsProcessor {
        <<utility>>
        +treat_runtimeparams(runtimeparams, params, forecast_dates, logger)
    }

    class OptimizationInternals {
        <<class>>
        +capacity_threshold_array : ndarray
        +P_excess : dict~int, LpVariable~
        +active_periods : dict~float, list~int~~
        +objective
        +constraints : dict
        +opt_tp : DataFrame
    }

    class LPModel {
        <<external>>
        +setObjective(objective)
        +solve()
    }

    RuntimeParamsProcessor --> CapacityTariffRuntimeConfig : populates
    CapacityTariffRuntimeConfig --> Optimization : provided via optim_conf
    Optimization --> OptimizationInternals : uses
    OptimizationInternals --> LPModel : configures objective and constraints
    Optimization --> LPModel : owns
    OptimizationInternals --> Optimization : results stored in opt_tp and cost_fun_profit

Loading

Flow diagram for capacity tariff configuration and usage

flowchart TD
    A_runtimeparams[Start: runtimeparams input] --> B_check_set[Check set_capacity_tariff in runtimeparams]
    B_check_set -->|missing| B1_set_false[set_capacity_tariff = False]
    B_check_set -->|present| B2_parse_flag[Parse set_capacity_tariff as bool]

    B1_set_false --> C_threshold_default
    B2_parse_flag --> C_threshold_source

    C_threshold_source{capacity_tariff_threshold provided?} -->|no| C_threshold_default[capacity_tariff_threshold = 5000]
    C_threshold_source -->|yes| C_threshold_raw[Read capacity_tariff_threshold]

    C_threshold_raw --> C_is_str{Is string?}
    C_is_str -->|yes| C_parse_str[ast.literal_eval to list or float]
    C_is_str -->|no| C_use_value[Use raw value]

    C_parse_str --> C_after_parse
    C_use_value --> C_after_parse

    C_after_parse --> C_is_list{Is list?}
    C_is_list -->|yes| C_validate_len[Validate list length vs forecast_dates]
    C_is_list -->|no| C_scalar[Use scalar threshold]

    C_validate_len -->|len >= forecast| C_list_ok[Accept time-series threshold]
    C_validate_len -->|len < forecast| C_fallback_scalar[Fallback to first nonzero or 5000]

    C_list_ok --> D_threshold_to_optim_conf[Store capacity_tariff_threshold in optim_conf]
    C_scalar --> D_threshold_to_optim_conf
    C_fallback_scalar --> D_threshold_to_optim_conf
    C_threshold_default --> D_threshold_to_optim_conf

    A_runtimeparams --> E_penalty_source[Read capacity_tariff_penalty]
    E_penalty_source --> E_try_parse[Parse to float]
    E_try_parse -->|ok| E_penalty_ok[Use parsed penalty]
    E_try_parse -->|error| E_penalty_default[capacity_tariff_penalty = 10.0]

    E_penalty_ok --> F_penalty_to_optim_conf[Store capacity_tariff_penalty in optim_conf]
    E_penalty_default --> F_penalty_to_optim_conf

    B1_set_false --> G_flag_to_optim_conf[Store set_capacity_tariff in optim_conf]
    B2_parse_flag --> G_flag_to_optim_conf

    D_threshold_to_optim_conf --> H_optim_init[Optimization.__init__ reads optim_conf]
    F_penalty_to_optim_conf --> H_optim_init
    G_flag_to_optim_conf --> H_optim_init

    H_optim_init --> H1_check_flag{set_capacity_tariff?}
    H1_check_flag -->|no| Z_end[Optimization without capacity tariff]
    H1_check_flag -->|yes| H2_init_capacity[Init capacity_threshold, capacity_threshold_is_list, capacity_penalty]

    H2_init_capacity --> I_perform_opt[perform_optimization starts]

    I_perform_opt --> I1_prepare_array{capacity_threshold_is_list?}
    I1_prepare_array -->|yes| I2_build_array[Build capacity_threshold_array from list]
    I1_prepare_array -->|no| I4_skip_array[Use scalar threshold only]

    I2_build_array --> I3_validate_len[Check length vs data_opt index]
    I3_validate_len -->|mismatch| I3_disable[Log error and disable set_capacity_tariff]
    I3_validate_len -->|ok| I3_stats[Log active_count and unique thresholds]

    I3_stats --> J_objective_start[Build objective function]
    I4_skip_array --> J_objective_start
    I3_disable --> Z_end

    J_objective_start --> J1_build_P_excess[Create P_excess variables per timestep]
    J1_build_P_excess --> J2_add_penalty[Add capacity_penalty_term to objective]

    J2_add_penalty --> K_constraints_start[Build constraints]
    K_constraints_start --> K1_excess_constraints[Add capacity tariff excess constraints]

    K1_excess_constraints --> L_solve[Solve LP model]
    L_solve --> M_collect_results[Collect optimization results into opt_tp]

    M_collect_results --> M1_extract_excess[Extract P_excess.varValue per timestep]
    M1_extract_excess --> M2_compute_penalty[Compute capacity_penalty_cost per timestep]
    M2_compute_penalty --> M3_add_columns[Add capacity_threshold, P_excess, capacity_penalty_cost columns]

    M3_add_columns --> N_costfun_profit{costfun == profit?}
    N_costfun_profit -->|yes| N1_add_penalty_profit[Add penalty_per_timestep to cost_fun_profit]
    N_costfun_profit -->|no| N2_other_costfun[Other costfun paths]

    N1_add_penalty_profit --> Z_end
    N2_other_costfun --> Z_end

Loading

File-Level Changes

Change	Details	Files
Add capacity tariff configuration to the optimizer and support scalar or time-series thresholds with penalties.	Read capacity tariff activation flag, threshold, and penalty from optim_conf with default values. Support both scalar and list/ndarray thresholds, tracking whether a time-series is used and logging configuration details. Validate that time-series thresholds match the optimization horizon length, disabling the feature if mismatched and logging diagnostics. Prepare a NumPy array of thresholds and log active timesteps and unique threshold levels when in list mode.	src/emhass/optimization.py
Integrate capacity tariff penalties into the objective and constraints, and expose the results in the optimization output.	Create per-timestep excess power variables P_excess when the capacity tariff is enabled, with separate handling for scalar vs time-series thresholds. Augment the objective with a per-timestep penalty term proportional to excess power, updating logging to reflect which mode is used. Add capacity tariff constraints linking P_excess, P_grid_pos, and the threshold(s) so that P_excess captures grid import above the threshold. Populate output DataFrame with capacity tariff-related columns (threshold, P_excess, capacity_penalty_cost), log aggregate penalty/excess, and add a temporary diagnostic on irregular time indices. Include the per-timestep capacity tariff penalty values when computing cost_fun_profit so profit reflects capacity charges.	src/emhass/optimization.py
Handle capacity tariff parameters in runtime configuration parsing.	Parse set_capacity_tariff flag from runtime parameters into optim_conf, defaulting to False when not provided. Parse capacity_tariff_threshold from runtime parameters, supporting scalar or list input (including string-encoded), validating list length vs forecast horizon, and falling back to scalar mode with sane defaults when invalid. Parse capacity_tariff_penalty from runtime parameters into optim_conf with numeric coercion and logging-based fallback to a default value on invalid input.	src/emhass/utils.py

Possibly linked issues

• #[Feature request] Capacity tariff: The PR adds configurable capacity tariff thresholds and penalties, integrating them into optimization and cost/profit calculations as requested.

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sonarqubecloud]

Quality Gate passed

Issues
2 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[davidusb-geek]

Please include the needed unittest to thoroughly test this new feature.
You need to add a new method to test_optimization.py

[davidusb-geek]

Also please document how to use this 👍

[scruysberghs]

It would also be relevant to keep track what we mean with "capacity tariff" and how it is calculated towards your monthly electricity bill. There are many totally different implementations of this concept all over the world.

imho:

• if it is an increased price for distribution/grid usage added to each kwh drawn from the grid during certain moments of the day, these can already be added to the load_cost_forecast list as a runtime paramter. That would keep deferrable loads out the expensive periods and control your battery to prevent drawing any power for the grid during those moments if that is the outcome of the optimisation algorithm.
• if capacity tariff is a fixed fee you have to pay for the highest monthly 15min electricity usage (kW) like it is here in Belgium you can now already use the "max power from grid" constraint and calculate it's optimal value at the beginning of each month.