Skip to content

MESQUAL Congestion Rent Calculator

CongestionRentCalculator dataclass

Calculates congestion rents for electricity transmission lines.

Congestion rent represents the economic value captured by transmission assets due to price differences between nodes. It's calculated as the product of power flow, price spread, and time granularity.

The calculator handles bidirectional flows (up/down) and accounts for transmission losses by using separate sent and received quantities. This provides accurate congestion rent calculations that reflect actual market conditions and physical constraints.

Mathematical formulation: - Congestion rent (up) = granularity × (received_up × price_to - sent_up × price_from) - Congestion rent (down) = granularity × (received_down × price_from - sent_down × price_to) - Total congestion rent = congestion_rent_up + congestion_rent_down

Attributes:

Name Type Description
sent_up Series

Power sent in up direction (MW or MWh)
received_up Series

Power received in up direction after losses (MW or MWh)

sent_down Series

Power sent in down direction (MW or MWh)
received_down Series

Power received in down direction after losses (MW or MWh)
price_node_from Series

Price at sending node (€/MWh)
price_node_to Series

Price at receiving node (€/MWh)
granularity_hrs Series | float

Time granularity in hours (auto-detected if None)

Example:

>>> import pandas as pd
>>> # Time series data
>>> index = pd.date_range('2024-01-01', periods=3, freq='h')
>>> # Flow and price data
>>> calc = CongestionRentCalculator(
...     sent_up=pd.Series([100, 150, 200], index=index),
...     received_up=pd.Series([95, 142, 190], index=index),  # 5% losses
...     sent_down=pd.Series([50, 75, 100], index=index),
...     received_down=pd.Series([48, 71, 95], index=index),  # 4% losses
...     price_node_from=pd.Series([45, 50, 55], index=index),
...     price_node_to=pd.Series([65, 70, 75], index=index)
... )
>>> total_rent = calc.calculate()
>>> print(total_rent)  # Congestion rent in €
Source code in submodules/mesqual/mesqual/energy_data_handling/variable_utils/congestion_rent.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@dataclass
class CongestionRentCalculator:
    """Calculates congestion rents for electricity transmission lines.

    Congestion rent represents the economic value captured by transmission
    assets due to price differences between nodes. It's calculated as the
    product of power flow, price spread, and time granularity.

    The calculator handles bidirectional flows (up/down) and accounts for
    transmission losses by using separate sent and received quantities.
    This provides accurate congestion rent calculations that reflect actual
    market conditions and physical constraints.

    Mathematical formulation:
    - Congestion rent (up) = granularity × (received_up × price_to - sent_up × price_from)
    - Congestion rent (down) = granularity × (received_down × price_from - sent_down × price_to)
    - Total congestion rent = congestion_rent_up + congestion_rent_down

    Attributes:
        sent_up: Power sent in up direction (MW or MWh)
        received_up: Power received in up direction after losses (MW or MWh)  
        sent_down: Power sent in down direction (MW or MWh)
        received_down: Power received in down direction after losses (MW or MWh)
        price_node_from: Price at sending node (€/MWh)
        price_node_to: Price at receiving node (€/MWh)
        granularity_hrs: Time granularity in hours (auto-detected if None)

    Example:

        >>> import pandas as pd
        >>> # Time series data
        >>> index = pd.date_range('2024-01-01', periods=3, freq='h')
        >>> # Flow and price data
        >>> calc = CongestionRentCalculator(
        ...     sent_up=pd.Series([100, 150, 200], index=index),
        ...     received_up=pd.Series([95, 142, 190], index=index),  # 5% losses
        ...     sent_down=pd.Series([50, 75, 100], index=index),
        ...     received_down=pd.Series([48, 71, 95], index=index),  # 4% losses
        ...     price_node_from=pd.Series([45, 50, 55], index=index),
        ...     price_node_to=pd.Series([65, 70, 75], index=index)
        ... )
        >>> total_rent = calc.calculate()
        >>> print(total_rent)  # Congestion rent in €
    """
    sent_up: pd.Series
    received_up: pd.Series
    sent_down: pd.Series
    received_down: pd.Series
    price_node_from: pd.Series
    price_node_to: pd.Series
    granularity_hrs: pd.Series | float = None

    def __post_init__(self):
        """Initialize granularity and validate input consistency."""
        if self.granularity_hrs is None:
            if isinstance(self.sent_up.index, pd.DatetimeIndex):
                if len(self.sent_up.index) > 0:
                    from mesqual.energy_data_handling.granularity_analyzer import TimeSeriesGranularityAnalyzer
                    analyzer = TimeSeriesGranularityAnalyzer(strict_mode=False)
                    self.granularity_hrs = analyzer.get_granularity_as_series_of_hours(self.sent_up.index)
                else:
                    self.granularity_hrs = 0
            else:
                logger.warning(f'Granularity for CongestionRentCalculator is defaulting back to 1 hrs.')
                self.granularity_hrs = 1
        if isinstance(self.granularity_hrs, (float, int)):
            self.granularity_hrs = pd.Series(self.granularity_hrs, index=self.sent_up.index)
        self.__check_indices()

    def __check_indices(self):
        """Validate that all input Series have matching indices.

        Raises:
            ValueError: If any Series has mismatched index with sent_up
        """
        ref_index = self.sent_up.index
        to_check = [
            self.received_up,
            self.sent_down,
            self.received_down,
            self.price_node_from,
            self.price_node_to
        ]
        for v in to_check:
            if not ref_index.equals(v.index):
                raise ValueError(f'All indices of provided series must be equal.')

    @property
    def congestion_rent_up(self) -> pd.Series:
        """Calculate congestion rent for up direction flows.

        Returns:
            Series with congestion rents in up direction (€)
        """
        return self.granularity_hrs * (
                self.received_up * self.price_node_to -
                self.sent_up * self.price_node_from
        )

    @property
    def congestion_rent_down(self) -> pd.Series:
        """Calculate congestion rent for down direction flows.

        Returns:
            Series with congestion rents in down direction (€)
        """
        return self.granularity_hrs * (
                self.received_down * self.price_node_from -
                self.sent_down * self.price_node_to
        )

    @property
    def congestion_rent_total(self) -> pd.Series:
        """Calculate total congestion rent (sum of up and down directions).

        Returns:
            Series with total congestion rents (€)
        """
        return self.congestion_rent_up + self.congestion_rent_down

    def calculate(self) -> pd.Series:
        """Calculate total congestion rent (convenience method).

        Returns:
            Series with total congestion rents in € (same as congestion_rent_total property)
        """
        return self.congestion_rent_total

    @classmethod
    def from_net_flow_without_losses(
            cls,
            net_flow: pd.Series,
            price_node_from: pd.Series,
            price_node_to: pd.Series,
            granularity_hrs: float = None
    ) -> pd.Series:
        """Calculate congestion rent from net flow data assuming no losses.

        Convenience method for cases where transmission losses are negligible
        or not available. Splits net flow into unidirectional components and
        assumes sent equals received for each direction.

        Args:
            net_flow: Net power flow (positive = up direction, negative = down direction) in MW or MWh
            price_node_from: Price at sending node (€/MWh)
            price_node_to: Price at receiving node (€/MWh)
            granularity_hrs: Time granularity in hours (auto-detected if None)

        Returns:
            Series with total congestion rents in €

        Example:

            >>> # Net flow with price data
            >>> net_flow = pd.Series([100, -50, 75], index=time_index)  
            >>> rent = CongestionRentCalculator.from_net_flow_without_losses(
            ...     net_flow, price_from, price_to
            ... )
        """
        sent_up = net_flow.clip(lower=0)
        sent_down = (-net_flow).clip(lower=0)
        return cls(
            sent_up=sent_up,
            received_up=sent_up,
            sent_down=sent_down,
            received_down=sent_down,
            price_node_from=price_node_from,
            price_node_to=price_node_to,
            granularity_hrs=granularity_hrs
        ).calculate()

    @classmethod
    def from_up_and_down_flow_without_losses(
            cls,
            flow_up: pd.Series,
            flow_down: pd.Series,
            price_node_from: pd.Series,
            price_node_to: pd.Series,
            granularity_hrs: float = None
    ) -> pd.Series:
        """Calculate congestion rent from bidirectional flow data assuming no losses.

        Convenience method for cases where flows are already separated into up/down
        directions and transmission losses are negligible. Assumes sent equals
        received for each direction.

        Args:
            flow_up: Power flow in up direction (MW or MWh, non-negative)
            flow_down: Power flow in down direction (MW or MWh, non-negative)  
            price_node_from: Price at sending node (€/MWh)
            price_node_to: Price at receiving node (€/MWh)
            granularity_hrs: Time granularity in hours (auto-detected if None)

        Returns:
            Series with total congestion rents in €

        Example:

            >>> # Separate up/down flows
            >>> flow_up = pd.Series([100, 0, 75], index=time_index)
            >>> flow_down = pd.Series([0, 50, 0], index=time_index)
            >>> rent = CongestionRentCalculator.from_up_and_down_flow_without_losses(
            ...     flow_up, flow_down, price_from, price_to
            ... )
        """
        return cls(
            sent_up=flow_up,
            received_up=flow_up,
            sent_down=flow_down,
            received_down=flow_down,
            price_node_from=price_node_from,
            price_node_to=price_node_to,
            granularity_hrs=granularity_hrs
        ).calculate()

congestion_rent_up property 

congestion_rent_up: Series

Calculate congestion rent for up direction flows.

Returns:

Type Description
Series

Series with congestion rents in up direction (€)

congestion_rent_down property 

congestion_rent_down: Series

Calculate congestion rent for down direction flows.

Returns:

Type Description
Series

Series with congestion rents in down direction (€)

congestion_rent_total property 

congestion_rent_total: Series

Calculate total congestion rent (sum of up and down directions).

Returns:

Type Description
Series

Series with total congestion rents (€)

__post_init__ 

__post_init__()

Initialize granularity and validate input consistency.

Source code in submodules/mesqual/mesqual/energy_data_handling/variable_utils/congestion_rent.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __post_init__(self):
    """Initialize granularity and validate input consistency."""
    if self.granularity_hrs is None:
        if isinstance(self.sent_up.index, pd.DatetimeIndex):
            if len(self.sent_up.index) > 0:
                from mesqual.energy_data_handling.granularity_analyzer import TimeSeriesGranularityAnalyzer
                analyzer = TimeSeriesGranularityAnalyzer(strict_mode=False)
                self.granularity_hrs = analyzer.get_granularity_as_series_of_hours(self.sent_up.index)
            else:
                self.granularity_hrs = 0
        else:
            logger.warning(f'Granularity for CongestionRentCalculator is defaulting back to 1 hrs.')
            self.granularity_hrs = 1
    if isinstance(self.granularity_hrs, (float, int)):
        self.granularity_hrs = pd.Series(self.granularity_hrs, index=self.sent_up.index)
    self.__check_indices()

__check_indices 

__check_indices()

Validate that all input Series have matching indices.

Raises:

Type Description
ValueError

If any Series has mismatched index with sent_up
Source code in submodules/mesqual/mesqual/energy_data_handling/variable_utils/congestion_rent.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __check_indices(self):
    """Validate that all input Series have matching indices.

    Raises:
        ValueError: If any Series has mismatched index with sent_up
    """
    ref_index = self.sent_up.index
    to_check = [
        self.received_up,
        self.sent_down,
        self.received_down,
        self.price_node_from,
        self.price_node_to
    ]
    for v in to_check:
        if not ref_index.equals(v.index):
            raise ValueError(f'All indices of provided series must be equal.')

calculate 

calculate() -> Series

Calculate total congestion rent (convenience method).

Returns:

Type Description
Series

Series with total congestion rents in € (same as congestion_rent_total property)
Source code in submodules/mesqual/mesqual/energy_data_handling/variable_utils/congestion_rent.py 
129
130
131
132
133
134
135
def calculate(self) -> pd.Series:
    """Calculate total congestion rent (convenience method).

    Returns:
        Series with total congestion rents in € (same as congestion_rent_total property)
    """
    return self.congestion_rent_total

from_net_flow_without_losses classmethod 

from_net_flow_without_losses(net_flow: Series, price_node_from: Series, price_node_to: Series, granularity_hrs: float = None) -> Series

Calculate congestion rent from net flow data assuming no losses.

Convenience method for cases where transmission losses are negligible or not available. Splits net flow into unidirectional components and assumes sent equals received for each direction.

Parameters:

Name Type Description Default
net_flow Series

Net power flow (positive = up direction, negative = down direction) in MW or MWh

 required
price_node_from Series

Price at sending node (€/MWh)

 required
price_node_to Series

Price at receiving node (€/MWh)

 required
granularity_hrs float

Time granularity in hours (auto-detected if None)

 None

Returns:

Type Description
Series

Series with total congestion rents in €

Example:

>>> # Net flow with price data
>>> net_flow = pd.Series([100, -50, 75], index=time_index)  
>>> rent = CongestionRentCalculator.from_net_flow_without_losses(
...     net_flow, price_from, price_to
... )
Source code in submodules/mesqual/mesqual/energy_data_handling/variable_utils/congestion_rent.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@classmethod
def from_net_flow_without_losses(
        cls,
        net_flow: pd.Series,
        price_node_from: pd.Series,
        price_node_to: pd.Series,
        granularity_hrs: float = None
) -> pd.Series:
    """Calculate congestion rent from net flow data assuming no losses.

    Convenience method for cases where transmission losses are negligible
    or not available. Splits net flow into unidirectional components and
    assumes sent equals received for each direction.

    Args:
        net_flow: Net power flow (positive = up direction, negative = down direction) in MW or MWh
        price_node_from: Price at sending node (€/MWh)
        price_node_to: Price at receiving node (€/MWh)
        granularity_hrs: Time granularity in hours (auto-detected if None)

    Returns:
        Series with total congestion rents in €

    Example:

        >>> # Net flow with price data
        >>> net_flow = pd.Series([100, -50, 75], index=time_index)  
        >>> rent = CongestionRentCalculator.from_net_flow_without_losses(
        ...     net_flow, price_from, price_to
        ... )
    """
    sent_up = net_flow.clip(lower=0)
    sent_down = (-net_flow).clip(lower=0)
    return cls(
        sent_up=sent_up,
        received_up=sent_up,
        sent_down=sent_down,
        received_down=sent_down,
        price_node_from=price_node_from,
        price_node_to=price_node_to,
        granularity_hrs=granularity_hrs
    ).calculate()

from_up_and_down_flow_without_losses classmethod 

from_up_and_down_flow_without_losses(flow_up: Series, flow_down: Series, price_node_from: Series, price_node_to: Series, granularity_hrs: float = None) -> Series

Calculate congestion rent from bidirectional flow data assuming no losses.

Convenience method for cases where flows are already separated into up/down directions and transmission losses are negligible. Assumes sent equals received for each direction.

Parameters:

Name Type Description Default
flow_up Series

Power flow in up direction (MW or MWh, non-negative)

 required
flow_down Series

Power flow in down direction (MW or MWh, non-negative)

required
price_node_from Series

Price at sending node (€/MWh)

 required
price_node_to Series

Price at receiving node (€/MWh)

 required
granularity_hrs float

Time granularity in hours (auto-detected if None)

 None

Returns:

Type Description
Series

Series with total congestion rents in €

Example:

>>> # Separate up/down flows
>>> flow_up = pd.Series([100, 0, 75], index=time_index)
>>> flow_down = pd.Series([0, 50, 0], index=time_index)
>>> rent = CongestionRentCalculator.from_up_and_down_flow_without_losses(
...     flow_up, flow_down, price_from, price_to
... )
Source code in submodules/mesqual/mesqual/energy_data_handling/variable_utils/congestion_rent.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@classmethod
def from_up_and_down_flow_without_losses(
        cls,
        flow_up: pd.Series,
        flow_down: pd.Series,
        price_node_from: pd.Series,
        price_node_to: pd.Series,
        granularity_hrs: float = None
) -> pd.Series:
    """Calculate congestion rent from bidirectional flow data assuming no losses.

    Convenience method for cases where flows are already separated into up/down
    directions and transmission losses are negligible. Assumes sent equals
    received for each direction.

    Args:
        flow_up: Power flow in up direction (MW or MWh, non-negative)
        flow_down: Power flow in down direction (MW or MWh, non-negative)  
        price_node_from: Price at sending node (€/MWh)
        price_node_to: Price at receiving node (€/MWh)
        granularity_hrs: Time granularity in hours (auto-detected if None)

    Returns:
        Series with total congestion rents in €

    Example:

        >>> # Separate up/down flows
        >>> flow_up = pd.Series([100, 0, 75], index=time_index)
        >>> flow_down = pd.Series([0, 50, 0], index=time_index)
        >>> rent = CongestionRentCalculator.from_up_and_down_flow_without_losses(
        ...     flow_up, flow_down, price_from, price_to
        ... )
    """
    return cls(
        sent_up=flow_up,
        received_up=flow_up,
        sent_down=flow_down,
        received_down=flow_down,
        price_node_from=price_node_from,
        price_node_to=price_node_to,
        granularity_hrs=granularity_hrs
    ).calculate()