Feature/v3/feature/381 feature equalize sizes andor flow rates between scenarios

CHANGELOG.md

@@ -51,6 +51,8 @@ This release introduces new model dimensions (periods and scenarios) for multi-p
 
 - **Period dimension**: Enables multi-period investment modeling with distinct decisions in each period for transformation pathway optimization
 - **Scenario dimension**: Supports stochastic modeling with weighted scenarios for robust decision-making under uncertainty (demand, prices, weather)
+    - Control variable independence across scenarios via `scenario_independent_sizes` and `scenario_independent_flow_rates` parameters
+    - By default, investment sizes are shared across scenarios while flow rates vary per scenario
 
 **Redesigned effect sharing system:**
 

docs/user-guide/mathematical-notation/dimensions.md

@@ -132,9 +132,9 @@ $$
 
 ### Shared Periodic Decisions: The Exception
 
-**Within a period, periodic (investment) decisions are shared across all scenarios:**
+**Investment decisions (sizes) can be shared across all scenarios:**
 
-If a period has multiple scenarios, periodic variables (e.g., component size) are **scenario-independent** but **temporal variables are scenario-specific**.
+By default, sizes (e.g., Storage capacity, Thermal power, ...) are **scenario-independent** but **flow_rates are scenario-specific**.
 
 **Example - Flow with investment:**
 
@@ -153,6 +153,28 @@ $$
 
 This reflects real-world investment under uncertainty: you build capacity once (periodic/investment decision), but it operates under different conditions (temporal/operational decisions per scenario).
 
+**Mathematical Flexibility:**
+
+Variables can be either scenario-independent or scenario-specific:
+
+| Variable Type | Scenario-Independent | Scenario-Specific |
+|---------------|---------------------|-------------------|
+| **Sizes** (e.g., $\text{P}$) | $\text{P}(y)$ - Single value per period | $\text{P}(y, s)$ - Different per scenario |
+| **Flow rates** (e.g., $p(\text{t}_i)$) | $p(\text{t}_i, y)$ - Same across scenarios | $p(\text{t}_i, y, s)$ - Different per scenario |
+
+**Use Cases:**
+
+*Investment problems (with InvestParameters):*
+- **Sizes shared** (default): Investment under uncertainty - build capacity that performs well across all scenarios
+- **Sizes vary**: Scenario-specific capacity planning where different investments can be made for each future
+- **Selected sizes shared**: Mix of shared critical infrastructure and scenario-specific optional/flexible capacity
+
+*Dispatch problems (fixed sizes, no investments):*
+- **Flow rates shared**: Robust dispatch - find a single operational strategy that works across all forecast scenarios (e.g., day-ahead unit commitment under demand/weather uncertainty)
+- **Flow rates vary** (default): Scenario-adaptive dispatch - optimize operations for each scenario's specific conditions (demand, weather, prices)
+
+For implementation details on controlling scenario independence, see the [`FlowSystem`][flixopt.flow_system.FlowSystem] API reference.
+
 ---
 
 ## Dimensional Impact on Objective Function

flixopt/flow_system.py

@@ -60,6 +60,14 @@ class FlowSystem(Interface):
             If you use an array, take care that its long enough to cover all previous values!
         weights: The weights of each period and scenario. If None, all scenarios have the same weight (normalized to 1).
             Its recommended to normalize the weights to sum up to 1.
+        scenario_independent_sizes: Controls whether investment sizes are equalized across scenarios.
+            - True: All sizes are shared/equalized across scenarios
+            - False: All sizes are optimized separately per scenario
+            - list[str]: Only specified components (by label_full) are equalized across scenarios
+        scenario_independent_flow_rates: Controls whether flow rates are equalized across scenarios.
+            - True: All flow rates are shared/equalized across scenarios
+            - False: All flow rates are optimized separately per scenario
+            - list[str]: Only specified flows (by label_full) are equalized across scenarios
 
     Notes:
         - Creates an empty registry for components and buses, an empty EffectCollection, and a placeholder for a SystemModel.
@@ -75,6 +83,8 @@ def __init__(
         hours_of_last_timestep: float | None = None,
         hours_of_previous_timesteps: int | float | np.ndarray | None = None,
         weights: PeriodicDataUser | None = None,
+        scenario_independent_sizes: bool | list[str] = True,
+        scenario_independent_flow_rates: bool | list[str] = False,
     ):
         self.timesteps = self._validate_timesteps(timesteps)
         self.timesteps_extra = self._create_timesteps_with_extra(self.timesteps, hours_of_last_timestep)
@@ -104,6 +114,10 @@ def __init__(
 
         self._network_app = None
 
+        # Use properties to validate and store scenario dimension settings
+        self.scenario_independent_sizes = scenario_independent_sizes
+        self.scenario_independent_flow_rates = scenario_independent_flow_rates
+
     @staticmethod
     def _validate_timesteps(timesteps: pd.DatetimeIndex) -> pd.DatetimeIndex:
         """Validate timesteps format and rename if needed."""
@@ -268,6 +282,8 @@ def from_dataset(cls, ds: xr.Dataset) -> FlowSystem:
             else None,
             hours_of_last_timestep=reference_structure.get('hours_of_last_timestep'),
             hours_of_previous_timesteps=reference_structure.get('hours_of_previous_timesteps'),
+            scenario_independent_sizes=reference_structure.get('scenario_independent_sizes', True),
+            scenario_independent_flow_rates=reference_structure.get('scenario_independent_flow_rates', False),
         )
 
         # Restore components
@@ -762,6 +778,77 @@ def coords(self) -> dict[FlowSystemDimensions, pd.Index]:
     def used_in_calculation(self) -> bool:
         return self._used_in_calculation
 
+    def _validate_scenario_parameter(self, value: bool | list[str], param_name: str, element_type: str) -> None:
+        """
+        Validate scenario parameter value.
+
+        Args:
+            value: The value to validate
+            param_name: Name of the parameter (for error messages)
+            element_type: Type of elements expected in list (e.g., 'component label_full', 'flow label_full')
+
+        Raises:
+            TypeError: If value is not bool or list[str]
+            ValueError: If list contains non-string elements
+        """
+        if isinstance(value, bool):
+            return  # Valid
+        elif isinstance(value, list):
+            if not all(isinstance(item, str) for item in value):
+                raise ValueError(f'{param_name} list must contain only strings ({element_type} values)')
+        else:
+            raise TypeError(f'{param_name} must be bool or list[str], got {type(value).__name__}')
+
+    @property
+    def scenario_independent_sizes(self) -> bool | list[str]:
+        """
+        Controls whether investment sizes are equalized across scenarios.
+
+        Returns:
+            bool or list[str]: Configuration for scenario-independent sizing
+        """
+        return self._scenario_independent_sizes
+
+    @scenario_independent_sizes.setter
+    def scenario_independent_sizes(self, value: bool | list[str]) -> None:
+        """
+        Set whether investment sizes should be equalized across scenarios.
+
+        Args:
+            value: True (all equalized), False (all vary), or list of component label_full strings to equalize
+
+        Raises:
+            TypeError: If value is not bool or list[str]
+            ValueError: If list contains non-string elements
+        """
+        self._validate_scenario_parameter(value, 'scenario_independent_sizes', 'Element.label_full')
+        self._scenario_independent_sizes = value
+
+    @property
+    def scenario_independent_flow_rates(self) -> bool | list[str]:
+        """
+        Controls whether flow rates are equalized across scenarios.
+
+        Returns:
+            bool or list[str]: Configuration for scenario-independent flow rates
+        """
+        return self._scenario_independent_flow_rates
+
+    @scenario_independent_flow_rates.setter
+    def scenario_independent_flow_rates(self, value: bool | list[str]) -> None:
+        """
+        Set whether flow rates should be equalized across scenarios.
+
+        Args:
+            value: True (all equalized), False (all vary), or list of flow label_full strings to equalize
+
+        Raises:
+            TypeError: If value is not bool or list[str]
+            ValueError: If list contains non-string elements
+        """
+        self._validate_scenario_parameter(value, 'scenario_independent_flow_rates', 'Flow.label_full')
+        self._scenario_independent_flow_rates = value
+
     def sel(
         self,
         time: str | slice | list[str] | pd.Timestamp | pd.DatetimeIndex | None = None,

flixopt/structure.py

@@ -103,6 +103,55 @@ def do_modeling(self):
         for bus in self.flow_system.buses.values():
             bus.create_model(self)
 
+        # Add scenario equality constraints after all elements are modeled
+        self._add_scenario_equality_constraints()
+
+    def _add_scenario_equality_for_parameter_type(
+        self,
+        parameter_type: Literal['flow_rate', 'size'],
+        config: bool | list[str],
+    ):
+        """Add scenario equality constraints for a specific parameter type.
+
+        Args:
+            parameter_type: The type of parameter ('flow_rate' or 'size')
+            config: Configuration value (True = equalize all, False = equalize none, list = equalize these)
+        """
+        if config is False:
+            return  # All vary per scenario, no constraints needed
+
+        suffix = f'|{parameter_type}'
+        if config is True:
+            # All should be scenario-independent
+            vars_to_constrain = [var for var in self.variables if var.endswith(suffix)]
+        else:
+            # Only those in the list should be scenario-independent
+            all_vars = [var for var in self.variables if var.endswith(suffix)]
+            to_equalize = {f'{element}{suffix}' for element in config}
+            vars_to_constrain = [var for var in all_vars if var in to_equalize]
+
+        # Validate that all specified variables exist
+        missing_vars = [v for v in vars_to_constrain if v not in self.variables]
+        if missing_vars:
+            param_name = 'scenario_independent_sizes' if parameter_type == 'size' else 'scenario_independent_flow_rates'
+            raise ValueError(f'{param_name} contains invalid labels: {missing_vars}')
+
+        logger.debug(f'Adding scenario equality constraints for {len(vars_to_constrain)} {parameter_type} variables')
+        for var in vars_to_constrain:
+            self.add_constraints(
+                self.variables[var].isel(scenario=0) == self.variables[var].isel(scenario=slice(1, None)),
+                name=f'{var}|scenario_independent',
+            )
+
+    def _add_scenario_equality_constraints(self):
+        """Add equality constraints to equalize variables across scenarios based on FlowSystem configuration."""
+        # Only proceed if we have scenarios
+        if self.flow_system.scenarios is None or len(self.flow_system.scenarios) <= 1:
+            return
+
+        self._add_scenario_equality_for_parameter_type('flow_rate', self.flow_system.scenario_independent_flow_rates)
+        self._add_scenario_equality_for_parameter_type('size', self.flow_system.scenario_independent_sizes)
+
     @property
     def solution(self):
         solution = super().solution