Add the pull-elexon-imbalance command

Author: mw2b

src/skypro/commands/pull_elexon_imbalance/README.md

@@ -0,0 +1,4 @@
+# pull-elexon-imbalance
+
+This is a script to pulls a months worth of half-hourly imbalance price and volume data from Elexon and saves it to disk.
+The data is saved in monthly CSV files in the directory defined by the `MARKET_DATA_DIR` variable in the environment configuration.

src/skypro/commands/pull_elexon_imbalance/__init__.py

@@ -0,0 +1,104 @@
+import functools
+import json
+import logging
+
+import requests
+from datetime import date, datetime, timedelta
+from io import StringIO
+from typing import Callable
+
+import pandas as pd
+
+from skypro.common.cli_utils.cli_utils import get_user_ack_of_warning_or_exit, read_yaml_file
+from skypro.common.data.utility import prepare_data_dir
+
+from skypro.common.timeutils.month_str import get_first_and_last_date
+from skypro.commands.pull_elexon_imbalance.utils import daterange, with_retries
+
+ELEXON_API_MAX_RETRIES = 5
+ELEXON_API_RETRY_DELAY = timedelta(seconds=1)
+
+
+def pull_elexon_imbalance(month_str: str, env_file_path: str):
+    """
+    Pulls a months worth of half-hourly imbalance price and volume data from Elexon and saves it to disk.
+    The data is saved in monthly CSV files in the directory defined by the MARKET_DATA_DIR in the environment configuration.
+    """
+
+    start_date, end_date = get_first_and_last_date(month_str)
+
+    today = datetime.now().date()
+    if end_date > today:
+        end_date = today
+        get_user_ack_of_warning_or_exit(f"The month has not ended yet, so data will be incomplete after {end_date}")
+
+    env_config = read_yaml_file(env_file_path)
+    data_dir = env_config["vars"]["MARKET_DATA_DIR"]
+
+    df = _fetch_multiple_days(
+        start=start_date,
+        end=end_date,
+        fetch_func=_fetch_day,
+    )
+
+    df["price"] = df["price"] / 10  # £/MW to p/kW
+
+    prices_file_path = prepare_data_dir(data_dir, "elexon", "imbalance_price", start_date)
+    volume_file_path = prepare_data_dir(data_dir, "elexon", "imbalance_volume",  start_date)
+
+    logging.info(f"Saving pricing data to '{prices_file_path}'")
+    df[["spUTCTime", "spClockTime", "price"]].to_csv(prices_file_path, index=False)
+    logging.info(f"Saving volume data to '{volume_file_path}'")
+    df[["spUTCTime", "spClockTime", "volume"]].to_csv(volume_file_path, index=False)
+
+
+def _fetch_multiple_days(start: date, end: date, fetch_func: Callable) -> pd.DataFrame:
+    """
+    The elexon API pulls for a single day, so this function calls the elexon API repeatedly for each day and stacks up
+    the results.
+    """
+    df = pd.DataFrame()
+    for day in daterange(start, end):
+
+        logging.info(f"Fetching imbalance data for '{str(day)}'...")
+        day_df = with_retries(  # The Elexon API can be busy/unreliable at times so use retries to get past temporary failures
+            functools.partial(fetch_func, day),
+            ELEXON_API_MAX_RETRIES,
+            ELEXON_API_RETRY_DELAY
+        )
+
+        df = pd.concat([df, day_df])
+
+    # The values come through in the opposite order to what you'd expect
+    df = df.sort_values(by=["spUTCTime"], ignore_index=True)
+
+    return df
+
+
+def _fetch_day(day: date) -> pd.DataFrame:
+    """
+    Pulls a single days worth of imbalance data from Elexon.
+    """
+
+    day_str = day.isoformat()
+    # Send a GET request to the API
+    response = requests.get(
+        url=f"https://data.elexon.co.uk/bmrs/api/v1/balancing/settlement/system-prices/{day_str}",
+        params={"format": "json"}
+    )
+    response.raise_for_status()
+
+    json_data = json.load(StringIO(response.text))
+    day_df = pd.DataFrame.from_dict(json_data["data"])
+
+    day_df["startTime"] = pd.to_datetime(day_df["startTime"], utc=True)
+    day_df = day_df[["startTime", "systemSellPrice", "netImbalanceVolume"]]
+    day_df = day_df.rename(columns={
+        "startTime": "spUTCTime",
+        "systemSellPrice": "price",
+        "netImbalanceVolume": "volume"
+    })
+
+    day_df.insert(1, "spClockTime", day_df["spUTCTime"].dt.tz_convert("Europe/London"))
+
+    return day_df

src/skypro/commands/pull_elexon_imbalance/main.py

@@ -0,0 +1,33 @@
+import logging
+from datetime import date, timedelta
+from time import sleep
+from typing import Callable, Any
+
+
+def with_retries(func: Callable, max_attempts: int, delay: timedelta) -> Any:
+    """
+    This will call the given `func`, and retry up to `max_attempt` times if an exception is encountered.
+    """
+    attempts = 0
+    while True:
+        try:
+            attempts += 1
+            result = func()
+            return result
+        except Exception as e:
+            logging.warning(f"Call to function failed: {e}")
+            if attempts < max_attempts:
+                logging.info(f"Retrying {attempts}/{max_attempts}")
+                sleep(delay.total_seconds())
+                continue
+            else:
+                logging.error("Max attempts reached.")
+                raise e
+
+
+def daterange(start: date, end: date):
+    """
+    Yields every date between start and end.
+    """
+    for n in range(int((end - start).days + 1)):
+        yield start + timedelta(days=n)

src/skypro/commands/pull_elexon_imbalance/utils.py

@@ -23,7 +23,7 @@
 
 from skypro.common.cli_utils.cli_utils import read_yaml_file
 from skypro.common.data.utility import prepare_data_dir
-from skypro.commands.report.time import get_month_timerange
+from skypro.common.timeutils.month_str import get_month_timerange
 from skypro.commands.report.config.config import parse_config, Config
 from skypro.commands.report.microgrid_flow_calcs import calc_flows
 from skypro.commands.report.plots import plot_load_and_solar

src/skypro/commands/report/main.py

@@ -1,6 +1,6 @@
 import os
-from datetime import datetime, timedelta
-from typing import Callable, Optional, List
+from datetime import datetime, timedelta, date
+from typing import Callable, Optional, List, Union
 
 import pandas as pd
 
@@ -103,7 +103,7 @@ def _notice_level_for_duration(duration: timedelta) -> NoticeLevel:
         return NoticeLevel.INFO
 
 
-def prepare_data_dir(data_dir: str, data_source: str, sub_dir: str, date_tag: datetime) -> str:
+def prepare_data_dir(data_dir: str, data_source: str, sub_dir: str, date_tag: Union[datetime | date]) -> str:
     """
     Creates a directory for saving data into and returns the file name to use
     - `data_dir` is the base directory

src/skypro/common/data/utility.py

@@ -7,7 +7,7 @@
 
 def get_first_and_last_date(month_str: str) -> Tuple[date, date]:
     """
-    Returns the first and last date of the given month.
+    Returns the first and last date of the given month string (formatted like '2025-06').
     """
     try:
         start_date = datetime.strptime(month_str, '%Y-%m').date()
@@ -23,7 +23,7 @@ def get_first_and_last_date(month_str: str) -> Tuple[date, date]:
 def get_month_timerange(month_str: str, timezone_str: str) -> Tuple[datetime, datetime]:
     """
     Returns the first instant of the given month, and the first instant of the next month. So the months range
-    is start <= t < end.
+    is start <= t < end. The month is specified as a string in a format like '2025-06'.
     Note, this function doesn't use dateutils.relativedelta because it does not seem to honour DST timezones!
     """
     try:
@@ -38,4 +38,4 @@ def get_month_timerange(month_str: str, timezone_str: str) -> Tuple[datetime, da
     start = tz.localize(start)
     end = tz.localize(end)
 
-    return start, end
+    return start, end

src/skypro/commands/report/time.py src/skypro/common/timeutils/month_str.pysrc/skypro/commands/report/time.py renamed to src/skypro/common/timeutils/month_str.py

@@ -2,6 +2,7 @@
 import logging
 import importlib.metadata
 
+from skypro.commands.pull_elexon_imbalance.main import pull_elexon_imbalance
 from skypro.commands.report.main import report_cli
 from skypro.commands.simulator.main import simulate
 
@@ -20,7 +21,8 @@ def main():
     # Create a dictionary of commands, mapping to their python function
     commands = {
         "simulate": simulate,
-        "report": report_cli
+        "report": report_cli,
+        "pull-elexon-imbalance": pull_elexon_imbalance,
     }
 
     parser = argparse.ArgumentParser()
@@ -100,6 +102,15 @@ def main():
     )
     add_env_file_arg(parser_report)
 
+    parser_pull_elexon_imbalance = subparsers.add_parser('pull-elexon-imbalance')
+    parser_pull_elexon_imbalance.add_argument(
+        '-m', '--month',
+        dest='month_str',
+        required=True,
+        help='The month to pull data for, e.g. 2024-04'
+    )
+    add_env_file_arg(parser_pull_elexon_imbalance)
+
     kwargs = vars(parser.parse_args())
 
     command = kwargs.pop('subparser')