Skip to content

nordpool_api

This is a wrapper to communicate with the Nordpool API.

NordpoolApi

Client for the Nord Pool Market Data REST API.

Uses OAuth (password grant) to obtain a bearer token and exposes helpers for day-ahead prices, price indices, balance market data, and volumes.

Docs: https://developers.nordpoolgroup.com/reference/marketdata-introduction

Example

from datetime import datetime
from nordpool_utils.NordpoolApi import NordpoolApi

api = NordpoolApi()
start = datetime(2023, 3, 27)
end = datetime(2023, 3, 29)
areas = ["SE3", "SE4"]
df_spot = api.get_up_regulating_volumes(start, end, areas)
Source code in physical_operations_utils/nordpool_utils/NordpoolApi.py
 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
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
class NordpoolApi:
    """
    Client for the Nord Pool **Market Data REST API**.

    Uses OAuth (password grant) to obtain a bearer token and exposes helpers for
    day-ahead prices, price indices, balance market data, and volumes.

    **Docs:** https://developers.nordpoolgroup.com/reference/marketdata-introduction

    **Example**
    ```python
    from datetime import datetime
    from nordpool_utils.NordpoolApi import NordpoolApi

    api = NordpoolApi()
    start = datetime(2023, 3, 27)
    end = datetime(2023, 3, 29)
    areas = ["SE3", "SE4"]
    df_spot = api.get_up_regulating_volumes(start, end, areas)
    ```
    """

    def __init__(
        self, username: str = None, password: str = None, subscription_key: str = None
    ):
        """
        Initialize the API client and obtain an auth token.

        **Args**
        - `username`: Nord Pool account username. If `None`, read from keys file.
        - `password`: Nord Pool account password or Azure Key Vault secret value.
          If `None`, retrieved via `get_secret`.
        - `subscription_key`: API subscription key. If `None`, read from keys file.

        **Raises**
        - `urllib.error.URLError`: If token request fails.
        - `KeyError`: If required keys are missing from configuration.

        **Example**
        ```python
        from nordpool_utils.NordpoolApi import NordpoolApi
        api = NordpoolApi()  # pulls creds from your keys config/Azure Key Vault
        ```
        """
        keys = get_keys_yaml_file().get("nordpool")

        if username is None:
            username = keys["username"]
        if password is None:
            secret = keys["secret"]
            password = get_secret(secret)
        if subscription_key is None:
            subscription_key = keys["subscription_key"]

        self.base_url = "https://sts.nordpoolgroup.com/connect/token"
        self.headers = {
            "Authorization": "Basic Y2xpZW50X21hcmtldGRhdGFfYXBpOmNsaWVudF9tYXJrZXRkYXRhX2FwaQ==",
            "Content-Type": "application/x-www-form-urlencoded",
        }

        body = {
            "grant_type": "password",
            "scope": "marketdata_api",
            "username": username,
            "password": password,
        }

        body_encoded = urllib.parse.urlencode(body).encode()
        req = urllib.request.Request(
            self.base_url, data=body_encoded, headers=self.headers
        )
        response = urllib.request.urlopen(req)
        json_response = json.load(response)

        self.auth_token = json_response["access_token"]
        self.api_url = "marketdata-api.nordpoolgroup.com"
        self.subscription_key = subscription_key
        self.headers2 = {
            # Request headers
            "Accept-Encoding": "gzip",
            "Ocp-Apim-Subscription-Key": self.subscription_key,
            "Authorization": "Bearer " + self.auth_token,
        }

        self.headers2_no_gzip = {
            # Request headers
            "Accept-Encoding": "",
            "Ocp-Apim-Subscription-Key": self.subscription_key,
            "Authorization": "Bearer " + self.auth_token,
        }

        self.nordic_areas = [
            "DK1",
            "DK2",
            "SE1",
            "SE2",
            "SE3",
            "SE4",
            "NO1",
            "NO2",
            "NO3",
            "NO4",
            "NO5",
            "FI",
        ]

        self.baltic_areas = ["EE", "LT", "LV"]

    def get_regulation_market_data_single_day(
        self, date_utc: datetime, currency: str, areas: List[str]
    ) -> pd.DataFrame:
        """
        Fetch **manual frequency restoration reserves** (regulation market) for one UTC day.

        Returns a DataFrame that may include both 15-minute and hourly rows and may
        contain nulls if some intervals are not published.

        **Args**
        - `date_utc`: Date in UTC (tz-aware); CET date is derived from this.
        - `currency`: One of `{"EUR","SEK","NOK","DKK"}`.
        - `areas`: List of Nordic delivery areas (e.g., `["SE3","NO2"]`).

        **Raises**
        - `ValueError`: If currency/areas/timezone are invalid.
        - `requests.HTTPError`: If the HTTP response is not successful.

        **Returns**
        - `pd.DataFrame`: API fields plus `status`, `currency`, `deliveryArea`
          with `deliveryStart`/`deliveryEnd` parsed to UTC.

        **Example**
        ```python
        from datetime import datetime, timezone
        api = NordpoolApi()
        df = api.get_regulation_market_data_single_day(
            date_utc=datetime(2025, 2, 20, tzinfo=timezone.utc),
            currency="EUR",
            areas=["SE3","SE4"],
        )
        ```
        """
        SUPPORTED_CURRENCIES = [
            "EUR",
            "SEK",
            "NOK",
            "DKK",
        ]
        if currency not in SUPPORTED_CURRENCIES:
            raise ValueError(
                "Currency not supported. Supported currencies are: "
                + ", ".join(SUPPORTED_CURRENCIES)
            )
        if not set(areas).issubset(self.nordic_areas):
            raise ValueError(
                "One or more areas are not supported. Supported areas are: "
                + ", ".join(self.nordic_areas)
            )
        if not str(date_utc.tzinfo) == "UTC":
            raise ValueError("date_cet must be in UTC timezone.")
        date_cet = date_utc.astimezone(pytz.timezone("Europe/Stockholm"))
        params = {
            "currency": currency,
            "date": date_cet.strftime("%Y-%m-%d"),
            "areas": ",".join(areas),
        }
        endpoint = "/api/v2/BalanceMarket/ManualFrequencyRestorationReserves/ByAreas"
        res = requests.get(
            url=f"https://data-api.nordpoolgroup.com{endpoint}",
            params=params,
            headers=self.headers2,
        )
        if res.status_code != 200:
            logging.error(f"HTTP error {res.status_code}")
            logging.error(res.text)
            res.raise_for_status()
        if res.text == "":
            raise ValueError(f"No data found with status {res.status_code}.")
        data = res.json()
        df_final = pd.DataFrame()
        for area_data in data:
            area = area_data["deliveryArea"]
            status = area_data["status"]
            currency = area_data["currency"]
            df_area = pd.DataFrame(area_data["manualFrequencyRestorationReserves"])
            if df_area.empty or not {"deliveryStart", "deliveryEnd"}.issubset(
                set(df_area.columns)
            ):
                logging.warning(f"No data found for {area} on {date_cet}.")
                continue
            df_area["status"] = status
            df_area["currency"] = currency
            df_area["deliveryArea"] = area
            df_area["deliveryStart"] = pd.to_datetime(
                df_area["deliveryStart"], utc=True
            )
            df_area["deliveryEnd"] = pd.to_datetime(df_area["deliveryEnd"], utc=True)
            df_final = pd.concat([df_final, df_area])
        return df_final

    def get_hourly_spot_prices_single_day(
        self, date_europe_stockholm: datetime, currency: str, areas: List[str]
    ) -> pd.DataFrame:
        """
        Get **day-ahead hourly spot prices** for a CET/CEST date and one or more areas.

        Filters out rows with status other than *Preliminary* or *Final*. Timestamps are
        returned as UTC.

        **Args**
        - `date_europe_stockholm`: Date in the `Europe/Stockholm` timezone (tz-aware).
        - `currency`: One of `{"EUR","GBP","SEK","NOK","DKK","PLN"}`.
        - `areas`: List of delivery areas (Nordic & Baltic supported).

        **Returns**
        - `pd.DataFrame`: Columns
          `start_time_lb_utc`, `end_time_lb_utc`, `price_area`, `price`, `currency`, `status`.

        **Raises**
        - `ValueError`: On invalid currency/areas/timezone.
        - `requests.HTTPError`: If the HTTP response is not successful.

        **Example**
        ```python
        from zoneinfo import ZoneInfo
        api = NordpoolApi()
        df = api.get_hourly_spot_prices_single_day(
            date_europe_stockholm=datetime(2025, 2, 20, tzinfo=ZoneInfo("Europe/Stockholm")),
            currency="EUR",
            areas=["SE1","SE2","SE3","SE4"],
        )
        ```
        """
        SUPPORTED_CURRENCIES = ["EUR", "GBP", "SEK", "NOK", "DKK", "PLN"]
        if currency not in SUPPORTED_CURRENCIES:
            raise ValueError(
                "Currency not supported. Supported currencies are: "
                + ", ".join(SUPPORTED_CURRENCIES)
            )
        if (
            date_europe_stockholm.tzinfo is None
            or str(date_europe_stockholm.tzinfo) != "Europe/Stockholm"
        ):
            raise ValueError(
                "date_europe_stockholm must be in Europe/Stockholm timezone."
            )
        if not set(areas).issubset(self.nordic_areas + self.baltic_areas):
            raise ValueError("One or more areas are not supported.")
        params = {
            "market": "DayAhead",
            "areas": ",".join(areas),
            "currency": currency,
            "date": date_europe_stockholm.strftime("%Y-%m-%d"),
        }
        endpoint = "/api/v2/Auction/Prices/ByAreas"
        res = requests.get(
            url=f"https://data-api.nordpoolgroup.com{endpoint}",
            params=params,
            headers=self.headers2,
        )
        res.raise_for_status()
        if res.status_code == 200 and res.text != "":
            df_final = pd.DataFrame()
            data = res.json()
            for area_data in data:
                price_status = area_data["status"]
                if price_status not in ["Preliminary", "Final"]:
                    continue
                df_area = pd.DataFrame(area_data["prices"])
                df_area["status"] = price_status
                df_area["price_area"] = area_data["deliveryArea"]
                df_area["currency"] = area_data["currency"]
                df_area["start_time_lb_utc"] = pd.to_datetime(
                    df_area["deliveryStart"], utc=True
                )
                df_area["end_time_lb_utc"] = pd.to_datetime(
                    df_area["deliveryEnd"], utc=True
                )
                df_final = pd.concat([df_final, df_area])
            if not df_final.empty:
                df_final = df_final[
                    [
                        "start_time_lb_utc",
                        "end_time_lb_utc",
                        "price_area",
                        "price",
                        "currency",
                        "status",
                    ]
                ]
            return df_final
        else:
            raise ValueError("No data found.")

    def get_market_price_indices_range(
        self,
        start_time_lb_utc: datetime,
        end_time_lb_utc: datetime,
        currency: str,
        index_names: list[str],
        resolution_seconds: int = 3600,  # 3600, 1800, 900
        market: str = "DayAhead",
    ) -> pd.DataFrame:
        """
        Get **market price indices** (e.g., DayAhead, IntraDay) for a UTC time window
        and chosen resolution.

        Calls the Nord Pool `/Auction/PriceIndices/ByIndexNames` endpoint for each CET/CEST
        date spanned by the UTC window. Returns the API response in base (list-of-objects)
        form as a DataFrame. Nested fields such as `prices` and `deliveryAreas` remain
        unflattened.

        **Args**
        - `start_time_lb_utc`: Inclusive UTC lower bound (tz-aware).
        - `end_time_lb_utc`: Inclusive UTC upper bound (tz-aware).
        - `currency`: One of `{"EUR","GBP","SEK","NOK","DKK","PLN","RON"}`.
        - `index_names`: Index names, e.g. `["SE1","SE2","SE3","SE4","SYSTEM"]`.
        - `resolution_seconds`: Resolution in seconds (`3600`, `1800`, `900`).
        - `market`: Market identifier, e.g. `"DayAhead"`, `"IntraDay"`.

        **Returns**
        - `pd.DataFrame`: One row per index object from the API; nested fields preserved.

        **Raises**
        - `ValueError`: On invalid arguments.
        - `requests.HTTPError`: If the HTTP response is not successful.

        **Example**
        ```python
        from datetime import datetime, timezone
        from nordpool_utils.NordpoolApi import NordpoolApi

        api = NordpoolApi()
        df_idx = api.get_market_price_indices_range(
            start_time_lb_utc=datetime(2025, 2, 20, tzinfo=timezone.utc),
            end_time_lb_utc=datetime(2025, 2, 22, tzinfo=timezone.utc),
            currency="EUR",
            index_names=["SE1", "SE2", "SYSTEM"],
            resolution_seconds=3600,  # hourly
            market="DayAhead",
        )
        print(df_idx.head())
        ```
        """
        SUPPORTED_CURRENCIES = {"EUR", "GBP", "SEK", "NOK", "DKK", "PLN", "RON"}
        SUPPORTED_RES = {3600: "PT1H", 1800: "PT30M", 900: "PT15M"}

        # --- Validation ---
        if currency not in SUPPORTED_CURRENCIES:
            raise ValueError(
                f"Currency not supported. Supported: {', '.join(sorted(SUPPORTED_CURRENCIES))}"
            )
        if resolution_seconds not in SUPPORTED_RES:
            raise ValueError(
                f"resolution_seconds must be one of: {list(SUPPORTED_RES.keys())}"
            )
        if start_time_lb_utc.tzinfo is None or end_time_lb_utc.tzinfo is None:
            raise ValueError(
                "start_time_lb_utc and end_time_lb_utc must be tz-aware (UTC)."
            )
        if end_time_lb_utc < start_time_lb_utc:
            raise ValueError("end_time_lb_utc must be >= start_time_lb_utc")
        if not index_names:
            return pd.DataFrame()

        resolution_iso = SUPPORTED_RES[resolution_seconds]

        endpoint = "/api/v2/Auction/PriceIndices/ByIndexNames"
        base_url = f"https://data-api.nordpoolgroup.com{endpoint}"
        cet = ZoneInfo("Europe/Stockholm")

        cur_date_cet = start_time_lb_utc.astimezone(cet).date()
        end_date_cet = end_time_lb_utc.astimezone(cet).date()

        all_rows = []
        while cur_date_cet <= end_date_cet:
            params = {
                "market": market,
                "indexNames": ",".join(index_names),
                "currency": currency,
                "date": cur_date_cet.strftime("%Y-%m-%d"),
                "resolution": resolution_iso,
            }

            res = requests.get(url=base_url, params=params, headers=self.headers2)
            res.raise_for_status()

            if res.status_code == 200 and res.text:
                payload = res.json()
                if isinstance(payload, list):
                    all_rows.extend(payload)

            cur_date_cet += timedelta(days=1)

        if not all_rows:
            return pd.DataFrame()

        return pd.DataFrame(all_rows)

__init__(username=None, password=None, subscription_key=None)

Initialize the API client and obtain an auth token.

Args - username: Nord Pool account username. If None, read from keys file. - password: Nord Pool account password or Azure Key Vault secret value. If None, retrieved via get_secret. - subscription_key: API subscription key. If None, read from keys file.

Raises - urllib.error.URLError: If token request fails. - KeyError: If required keys are missing from configuration.

Example

from nordpool_utils.NordpoolApi import NordpoolApi
api = NordpoolApi()  # pulls creds from your keys config/Azure Key Vault
Source code in physical_operations_utils/nordpool_utils/NordpoolApi.py
 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
def __init__(
    self, username: str = None, password: str = None, subscription_key: str = None
):
    """
    Initialize the API client and obtain an auth token.

    **Args**
    - `username`: Nord Pool account username. If `None`, read from keys file.
    - `password`: Nord Pool account password or Azure Key Vault secret value.
      If `None`, retrieved via `get_secret`.
    - `subscription_key`: API subscription key. If `None`, read from keys file.

    **Raises**
    - `urllib.error.URLError`: If token request fails.
    - `KeyError`: If required keys are missing from configuration.

    **Example**
    ```python
    from nordpool_utils.NordpoolApi import NordpoolApi
    api = NordpoolApi()  # pulls creds from your keys config/Azure Key Vault
    ```
    """
    keys = get_keys_yaml_file().get("nordpool")

    if username is None:
        username = keys["username"]
    if password is None:
        secret = keys["secret"]
        password = get_secret(secret)
    if subscription_key is None:
        subscription_key = keys["subscription_key"]

    self.base_url = "https://sts.nordpoolgroup.com/connect/token"
    self.headers = {
        "Authorization": "Basic Y2xpZW50X21hcmtldGRhdGFfYXBpOmNsaWVudF9tYXJrZXRkYXRhX2FwaQ==",
        "Content-Type": "application/x-www-form-urlencoded",
    }

    body = {
        "grant_type": "password",
        "scope": "marketdata_api",
        "username": username,
        "password": password,
    }

    body_encoded = urllib.parse.urlencode(body).encode()
    req = urllib.request.Request(
        self.base_url, data=body_encoded, headers=self.headers
    )
    response = urllib.request.urlopen(req)
    json_response = json.load(response)

    self.auth_token = json_response["access_token"]
    self.api_url = "marketdata-api.nordpoolgroup.com"
    self.subscription_key = subscription_key
    self.headers2 = {
        # Request headers
        "Accept-Encoding": "gzip",
        "Ocp-Apim-Subscription-Key": self.subscription_key,
        "Authorization": "Bearer " + self.auth_token,
    }

    self.headers2_no_gzip = {
        # Request headers
        "Accept-Encoding": "",
        "Ocp-Apim-Subscription-Key": self.subscription_key,
        "Authorization": "Bearer " + self.auth_token,
    }

    self.nordic_areas = [
        "DK1",
        "DK2",
        "SE1",
        "SE2",
        "SE3",
        "SE4",
        "NO1",
        "NO2",
        "NO3",
        "NO4",
        "NO5",
        "FI",
    ]

    self.baltic_areas = ["EE", "LT", "LV"]

get_hourly_spot_prices_single_day(date_europe_stockholm, currency, areas)

Get day-ahead hourly spot prices for a CET/CEST date and one or more areas.

Filters out rows with status other than Preliminary or Final. Timestamps are returned as UTC.

Args - date_europe_stockholm: Date in the Europe/Stockholm timezone (tz-aware). - currency: One of {"EUR","GBP","SEK","NOK","DKK","PLN"}. - areas: List of delivery areas (Nordic & Baltic supported).

Returns - pd.DataFrame: Columns start_time_lb_utc, end_time_lb_utc, price_area, price, currency, status.

Raises - ValueError: On invalid currency/areas/timezone. - requests.HTTPError: If the HTTP response is not successful.

Example

from zoneinfo import ZoneInfo
api = NordpoolApi()
df = api.get_hourly_spot_prices_single_day(
    date_europe_stockholm=datetime(2025, 2, 20, tzinfo=ZoneInfo("Europe/Stockholm")),
    currency="EUR",
    areas=["SE1","SE2","SE3","SE4"],
)
Source code in physical_operations_utils/nordpool_utils/NordpoolApi.py
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def get_hourly_spot_prices_single_day(
    self, date_europe_stockholm: datetime, currency: str, areas: List[str]
) -> pd.DataFrame:
    """
    Get **day-ahead hourly spot prices** for a CET/CEST date and one or more areas.

    Filters out rows with status other than *Preliminary* or *Final*. Timestamps are
    returned as UTC.

    **Args**
    - `date_europe_stockholm`: Date in the `Europe/Stockholm` timezone (tz-aware).
    - `currency`: One of `{"EUR","GBP","SEK","NOK","DKK","PLN"}`.
    - `areas`: List of delivery areas (Nordic & Baltic supported).

    **Returns**
    - `pd.DataFrame`: Columns
      `start_time_lb_utc`, `end_time_lb_utc`, `price_area`, `price`, `currency`, `status`.

    **Raises**
    - `ValueError`: On invalid currency/areas/timezone.
    - `requests.HTTPError`: If the HTTP response is not successful.

    **Example**
    ```python
    from zoneinfo import ZoneInfo
    api = NordpoolApi()
    df = api.get_hourly_spot_prices_single_day(
        date_europe_stockholm=datetime(2025, 2, 20, tzinfo=ZoneInfo("Europe/Stockholm")),
        currency="EUR",
        areas=["SE1","SE2","SE3","SE4"],
    )
    ```
    """
    SUPPORTED_CURRENCIES = ["EUR", "GBP", "SEK", "NOK", "DKK", "PLN"]
    if currency not in SUPPORTED_CURRENCIES:
        raise ValueError(
            "Currency not supported. Supported currencies are: "
            + ", ".join(SUPPORTED_CURRENCIES)
        )
    if (
        date_europe_stockholm.tzinfo is None
        or str(date_europe_stockholm.tzinfo) != "Europe/Stockholm"
    ):
        raise ValueError(
            "date_europe_stockholm must be in Europe/Stockholm timezone."
        )
    if not set(areas).issubset(self.nordic_areas + self.baltic_areas):
        raise ValueError("One or more areas are not supported.")
    params = {
        "market": "DayAhead",
        "areas": ",".join(areas),
        "currency": currency,
        "date": date_europe_stockholm.strftime("%Y-%m-%d"),
    }
    endpoint = "/api/v2/Auction/Prices/ByAreas"
    res = requests.get(
        url=f"https://data-api.nordpoolgroup.com{endpoint}",
        params=params,
        headers=self.headers2,
    )
    res.raise_for_status()
    if res.status_code == 200 and res.text != "":
        df_final = pd.DataFrame()
        data = res.json()
        for area_data in data:
            price_status = area_data["status"]
            if price_status not in ["Preliminary", "Final"]:
                continue
            df_area = pd.DataFrame(area_data["prices"])
            df_area["status"] = price_status
            df_area["price_area"] = area_data["deliveryArea"]
            df_area["currency"] = area_data["currency"]
            df_area["start_time_lb_utc"] = pd.to_datetime(
                df_area["deliveryStart"], utc=True
            )
            df_area["end_time_lb_utc"] = pd.to_datetime(
                df_area["deliveryEnd"], utc=True
            )
            df_final = pd.concat([df_final, df_area])
        if not df_final.empty:
            df_final = df_final[
                [
                    "start_time_lb_utc",
                    "end_time_lb_utc",
                    "price_area",
                    "price",
                    "currency",
                    "status",
                ]
            ]
        return df_final
    else:
        raise ValueError("No data found.")

get_market_price_indices_range(start_time_lb_utc, end_time_lb_utc, currency, index_names, resolution_seconds=3600, market='DayAhead')

Get market price indices (e.g., DayAhead, IntraDay) for a UTC time window and chosen resolution.

Calls the Nord Pool /Auction/PriceIndices/ByIndexNames endpoint for each CET/CEST date spanned by the UTC window. Returns the API response in base (list-of-objects) form as a DataFrame. Nested fields such as prices and deliveryAreas remain unflattened.

Args - start_time_lb_utc: Inclusive UTC lower bound (tz-aware). - end_time_lb_utc: Inclusive UTC upper bound (tz-aware). - currency: One of {"EUR","GBP","SEK","NOK","DKK","PLN","RON"}. - index_names: Index names, e.g. ["SE1","SE2","SE3","SE4","SYSTEM"]. - resolution_seconds: Resolution in seconds (3600, 1800, 900). - market: Market identifier, e.g. "DayAhead", "IntraDay".

Returns - pd.DataFrame: One row per index object from the API; nested fields preserved.

Raises - ValueError: On invalid arguments. - requests.HTTPError: If the HTTP response is not successful.

Example

from datetime import datetime, timezone
from nordpool_utils.NordpoolApi import NordpoolApi

api = NordpoolApi()
df_idx = api.get_market_price_indices_range(
    start_time_lb_utc=datetime(2025, 2, 20, tzinfo=timezone.utc),
    end_time_lb_utc=datetime(2025, 2, 22, tzinfo=timezone.utc),
    currency="EUR",
    index_names=["SE1", "SE2", "SYSTEM"],
    resolution_seconds=3600,  # hourly
    market="DayAhead",
)
print(df_idx.head())
Source code in physical_operations_utils/nordpool_utils/NordpoolApi.py
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
def get_market_price_indices_range(
    self,
    start_time_lb_utc: datetime,
    end_time_lb_utc: datetime,
    currency: str,
    index_names: list[str],
    resolution_seconds: int = 3600,  # 3600, 1800, 900
    market: str = "DayAhead",
) -> pd.DataFrame:
    """
    Get **market price indices** (e.g., DayAhead, IntraDay) for a UTC time window
    and chosen resolution.

    Calls the Nord Pool `/Auction/PriceIndices/ByIndexNames` endpoint for each CET/CEST
    date spanned by the UTC window. Returns the API response in base (list-of-objects)
    form as a DataFrame. Nested fields such as `prices` and `deliveryAreas` remain
    unflattened.

    **Args**
    - `start_time_lb_utc`: Inclusive UTC lower bound (tz-aware).
    - `end_time_lb_utc`: Inclusive UTC upper bound (tz-aware).
    - `currency`: One of `{"EUR","GBP","SEK","NOK","DKK","PLN","RON"}`.
    - `index_names`: Index names, e.g. `["SE1","SE2","SE3","SE4","SYSTEM"]`.
    - `resolution_seconds`: Resolution in seconds (`3600`, `1800`, `900`).
    - `market`: Market identifier, e.g. `"DayAhead"`, `"IntraDay"`.

    **Returns**
    - `pd.DataFrame`: One row per index object from the API; nested fields preserved.

    **Raises**
    - `ValueError`: On invalid arguments.
    - `requests.HTTPError`: If the HTTP response is not successful.

    **Example**
    ```python
    from datetime import datetime, timezone
    from nordpool_utils.NordpoolApi import NordpoolApi

    api = NordpoolApi()
    df_idx = api.get_market_price_indices_range(
        start_time_lb_utc=datetime(2025, 2, 20, tzinfo=timezone.utc),
        end_time_lb_utc=datetime(2025, 2, 22, tzinfo=timezone.utc),
        currency="EUR",
        index_names=["SE1", "SE2", "SYSTEM"],
        resolution_seconds=3600,  # hourly
        market="DayAhead",
    )
    print(df_idx.head())
    ```
    """
    SUPPORTED_CURRENCIES = {"EUR", "GBP", "SEK", "NOK", "DKK", "PLN", "RON"}
    SUPPORTED_RES = {3600: "PT1H", 1800: "PT30M", 900: "PT15M"}

    # --- Validation ---
    if currency not in SUPPORTED_CURRENCIES:
        raise ValueError(
            f"Currency not supported. Supported: {', '.join(sorted(SUPPORTED_CURRENCIES))}"
        )
    if resolution_seconds not in SUPPORTED_RES:
        raise ValueError(
            f"resolution_seconds must be one of: {list(SUPPORTED_RES.keys())}"
        )
    if start_time_lb_utc.tzinfo is None or end_time_lb_utc.tzinfo is None:
        raise ValueError(
            "start_time_lb_utc and end_time_lb_utc must be tz-aware (UTC)."
        )
    if end_time_lb_utc < start_time_lb_utc:
        raise ValueError("end_time_lb_utc must be >= start_time_lb_utc")
    if not index_names:
        return pd.DataFrame()

    resolution_iso = SUPPORTED_RES[resolution_seconds]

    endpoint = "/api/v2/Auction/PriceIndices/ByIndexNames"
    base_url = f"https://data-api.nordpoolgroup.com{endpoint}"
    cet = ZoneInfo("Europe/Stockholm")

    cur_date_cet = start_time_lb_utc.astimezone(cet).date()
    end_date_cet = end_time_lb_utc.astimezone(cet).date()

    all_rows = []
    while cur_date_cet <= end_date_cet:
        params = {
            "market": market,
            "indexNames": ",".join(index_names),
            "currency": currency,
            "date": cur_date_cet.strftime("%Y-%m-%d"),
            "resolution": resolution_iso,
        }

        res = requests.get(url=base_url, params=params, headers=self.headers2)
        res.raise_for_status()

        if res.status_code == 200 and res.text:
            payload = res.json()
            if isinstance(payload, list):
                all_rows.extend(payload)

        cur_date_cet += timedelta(days=1)

    if not all_rows:
        return pd.DataFrame()

    return pd.DataFrame(all_rows)

get_regulation_market_data_single_day(date_utc, currency, areas)

Fetch manual frequency restoration reserves (regulation market) for one UTC day.

Returns a DataFrame that may include both 15-minute and hourly rows and may contain nulls if some intervals are not published.

Args - date_utc: Date in UTC (tz-aware); CET date is derived from this. - currency: One of {"EUR","SEK","NOK","DKK"}. - areas: List of Nordic delivery areas (e.g., ["SE3","NO2"]).

Raises - ValueError: If currency/areas/timezone are invalid. - requests.HTTPError: If the HTTP response is not successful.

Returns - pd.DataFrame: API fields plus status, currency, deliveryArea with deliveryStart/deliveryEnd parsed to UTC.

Example

from datetime import datetime, timezone
api = NordpoolApi()
df = api.get_regulation_market_data_single_day(
    date_utc=datetime(2025, 2, 20, tzinfo=timezone.utc),
    currency="EUR",
    areas=["SE3","SE4"],
)
Source code in physical_operations_utils/nordpool_utils/NordpoolApi.py
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
def get_regulation_market_data_single_day(
    self, date_utc: datetime, currency: str, areas: List[str]
) -> pd.DataFrame:
    """
    Fetch **manual frequency restoration reserves** (regulation market) for one UTC day.

    Returns a DataFrame that may include both 15-minute and hourly rows and may
    contain nulls if some intervals are not published.

    **Args**
    - `date_utc`: Date in UTC (tz-aware); CET date is derived from this.
    - `currency`: One of `{"EUR","SEK","NOK","DKK"}`.
    - `areas`: List of Nordic delivery areas (e.g., `["SE3","NO2"]`).

    **Raises**
    - `ValueError`: If currency/areas/timezone are invalid.
    - `requests.HTTPError`: If the HTTP response is not successful.

    **Returns**
    - `pd.DataFrame`: API fields plus `status`, `currency`, `deliveryArea`
      with `deliveryStart`/`deliveryEnd` parsed to UTC.

    **Example**
    ```python
    from datetime import datetime, timezone
    api = NordpoolApi()
    df = api.get_regulation_market_data_single_day(
        date_utc=datetime(2025, 2, 20, tzinfo=timezone.utc),
        currency="EUR",
        areas=["SE3","SE4"],
    )
    ```
    """
    SUPPORTED_CURRENCIES = [
        "EUR",
        "SEK",
        "NOK",
        "DKK",
    ]
    if currency not in SUPPORTED_CURRENCIES:
        raise ValueError(
            "Currency not supported. Supported currencies are: "
            + ", ".join(SUPPORTED_CURRENCIES)
        )
    if not set(areas).issubset(self.nordic_areas):
        raise ValueError(
            "One or more areas are not supported. Supported areas are: "
            + ", ".join(self.nordic_areas)
        )
    if not str(date_utc.tzinfo) == "UTC":
        raise ValueError("date_cet must be in UTC timezone.")
    date_cet = date_utc.astimezone(pytz.timezone("Europe/Stockholm"))
    params = {
        "currency": currency,
        "date": date_cet.strftime("%Y-%m-%d"),
        "areas": ",".join(areas),
    }
    endpoint = "/api/v2/BalanceMarket/ManualFrequencyRestorationReserves/ByAreas"
    res = requests.get(
        url=f"https://data-api.nordpoolgroup.com{endpoint}",
        params=params,
        headers=self.headers2,
    )
    if res.status_code != 200:
        logging.error(f"HTTP error {res.status_code}")
        logging.error(res.text)
        res.raise_for_status()
    if res.text == "":
        raise ValueError(f"No data found with status {res.status_code}.")
    data = res.json()
    df_final = pd.DataFrame()
    for area_data in data:
        area = area_data["deliveryArea"]
        status = area_data["status"]
        currency = area_data["currency"]
        df_area = pd.DataFrame(area_data["manualFrequencyRestorationReserves"])
        if df_area.empty or not {"deliveryStart", "deliveryEnd"}.issubset(
            set(df_area.columns)
        ):
            logging.warning(f"No data found for {area} on {date_cet}.")
            continue
        df_area["status"] = status
        df_area["currency"] = currency
        df_area["deliveryArea"] = area
        df_area["deliveryStart"] = pd.to_datetime(
            df_area["deliveryStart"], utc=True
        )
        df_area["deliveryEnd"] = pd.to_datetime(df_area["deliveryEnd"], utc=True)
        df_final = pd.concat([df_final, df_area])
    return df_final