Skip to content

Read from ENTSO-E API

PythonEntsoeSource

Bases: SourceInterface

The Python ENTSO-E Source is used to read day-ahead prices from ENTSO-E.

Example

from rtdip_sdk.pipelines.sources import PythonEntsoeSource

entsoe_source = PythonEntsoeSource(
    api_key={API_KEY},
    start='20230101',
    end='20231001',
    country_code='NL'
)

entsoe_source.read_batch()

Parameters:

Name Type Description Default
api_key str

API token for ENTSO-E, to request access see documentation here

 required
start str

Start time in the format YYYYMMDD

 required
end str

End time in the format YYYYMMDD

 required
country_code str

Country code to query from. A full list of country codes can be found here

 required
resolution optional str

Frequency of values; '60T' for hourly values, '30T' for half-hourly values or '15T' for quarterly values

 '60T'
Source code in src/sdk/python/rtdip_sdk/pipelines/sources/python/entsoe.py 
 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
class PythonEntsoeSource(SourceInterface):
    """
    The Python ENTSO-E Source is used to read day-ahead prices from ENTSO-E.

    Example
    --------
    ```python
    from rtdip_sdk.pipelines.sources import PythonEntsoeSource

    entsoe_source = PythonEntsoeSource(
        api_key={API_KEY},
        start='20230101',
        end='20231001',
        country_code='NL'
    )

    entsoe_source.read_batch()
    ```

    Args:
        api_key (str): API token for ENTSO-E, to request access see documentation [here](https://transparency.entsoe.eu/content/static_content/Static%20content/web%20api/Guide.html#_authentication_and_authorisation)
        start (str): Start time in the format YYYYMMDD
        end (str): End time in the format YYYYMMDD
        country_code (str): Country code to query from. A full list of country codes can be found [here](https://github.com/EnergieID/entsoe-py/blob/master/entsoe/mappings.py#L48)
        resolution (optional str): Frequency of values; '60T' for hourly values, '30T' for half-hourly values or '15T' for quarterly values
    """

    api_key: str
    start: str
    end: str
    country_code: str
    resolution: str

    def __init__(
        self, api_key, start, end, country_code, resolution: str = "60T"
    ) -> None:
        self.key = api_key
        self.start = pd.Timestamp(start, tz="UTC")
        self.end = pd.Timestamp(end, tz="UTC")
        self.country = country_code
        self.resolution = resolution

    @staticmethod
    def system_type():
        """
        Attributes:
            SystemType (Environment): Requires Python
        """
        return SystemType.PYTHON

    @staticmethod
    def libraries():
        libraries = Libraries()
        return libraries

    @staticmethod
    def settings() -> dict:
        return {}

    def pre_read_validation(self):
        return True

    def post_read_validation(self):
        return True

    def read_batch(self) -> DataFrame:
        """
        Reads batch from ENTSO-E API.
        """
        client = EntsoePandasClient(api_key=self.key)
        df = client.query_day_ahead_prices(self.country, start=self.start, end=self.end)
        df = pd.DataFrame(df, columns=["Price"])
        df["Name"] = "APX"
        return df

    def read_stream(self):
        """
        Raises:
            NotImplementedError: ENTSO-E connector does not support the stream operation.
        """
        raise NotImplementedError(
            "ENTSO-E connector does not support the stream operation."
        )

system_type() staticmethod

Attributes:

Name Type Description
SystemType Environment

Requires Python
Source code in src/sdk/python/rtdip_sdk/pipelines/sources/python/entsoe.py 
65
66
67
68
69
70
71
@staticmethod
def system_type():
    """
    Attributes:
        SystemType (Environment): Requires Python
    """
    return SystemType.PYTHON

read_batch()

Reads batch from ENTSO-E API.

Source code in src/sdk/python/rtdip_sdk/pipelines/sources/python/entsoe.py 
88
89
90
91
92
93
94
95
96
def read_batch(self) -> DataFrame:
    """
    Reads batch from ENTSO-E API.
    """
    client = EntsoePandasClient(api_key=self.key)
    df = client.query_day_ahead_prices(self.country, start=self.start, end=self.end)
    df = pd.DataFrame(df, columns=["Price"])
    df["Name"] = "APX"
    return df

read_stream()

Raises:

Type Description
NotImplementedError

ENTSO-E connector does not support the stream operation.
Source code in src/sdk/python/rtdip_sdk/pipelines/sources/python/entsoe.py 
 98
 99
100
101
102
103
104
105
def read_stream(self):
    """
    Raises:
        NotImplementedError: ENTSO-E connector does not support the stream operation.
    """
    raise NotImplementedError(
        "ENTSO-E connector does not support the stream operation."
    )