Skip to content

fingrid_api

This is a wrapper to communicate with the Fingrid API.

FingridAPI

A client for interacting with the Fingrid API.

Example 
from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

api_key = 'your_api_key_here'  # Replace 'your_api_key_here' with your actual API key, it will be fetched from Key Vault if None
fingrid_api = FingridAPI(api_key)

datasets = {
    "Wind power generation forecast": 1,
    "Solar power generation forecast": 2
} # Dataset name: ID mapping
start_time_utc = '2025-12-01 05:00:00.000+00:00'
end_time_utc = '2025-12-01 20:00:00.000+00:00'

df = fingrid_api.fetch_data(datasets, start_time_utc, end_time_utc)

if df is not None:
    print(df)
Source code in physical_operations_utils/fingrid_utils/FingridApi.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
class FingridAPI:
    """
    A client for interacting with the Fingrid API.

    Example:
        ```python
        from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

        api_key = 'your_api_key_here'  # Replace 'your_api_key_here' with your actual API key, it will be fetched from Key Vault if None
        fingrid_api = FingridAPI(api_key)

        datasets = {
            "Wind power generation forecast": 1,
            "Solar power generation forecast": 2
        } # Dataset name: ID mapping
        start_time_utc = '2025-12-01 05:00:00.000+00:00'
        end_time_utc = '2025-12-01 20:00:00.000+00:00'

        df = fingrid_api.fetch_data(datasets, start_time_utc, end_time_utc)

        if df is not None:
            print(df)
        ```
    """

    def __init__(self, api_key: str = None):
        """Initialize a Fingrid API client for a specific asset.

        Args:
            api_key: Optional, used for testing API key, gets secret from the key vault if empty.

        Example:
            ```python
            from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

            fingrid_api = FingridAPI()
            ```
        """
        self.api_key = api_key
        self.base_url = "https://data.fingrid.fi/api/data"
        self.locale = "en"
        if api_key is None:
            self.api_key = get_secret(get_keys_yaml_file().get("fingrid").get("secret"))

    def fetch_data(
        self,
        datasets: dict,
        start_time_utc: str,
        end_time_utc: str,
        format: str = "json",
        one_row_per_time_period: bool = True,
        page_size=20000,
    ) -> pd.DataFrame:
        """Returns data from Fingrid API for the specified datasets and time range.

        Args:
            datasets (dict): A dictionary where keys are dataset names and values are dataset IDs.
            start_time_utc (str): The start time for the data retrieval in ISO 8601 format.
            end_time_utc (str): The end time for the data retrieval in ISO 8601 format.
            format (str, optional): The format of the returned data. Defaults to "json".
            one_row_per_time_period (bool, optional): Whether to return one row per time period. Defaults to True.
            page_size (int, optional): The number of records per page. Defaults to 20000.

        Returns:
            pandas.DataFrame: A DataFrame containing the time intervals and corresponding values for the requested datasets.

        Example:
            ```python
            from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

            fingrid_api = FingridAPI()
            df = fingrid_api.fetch_data(datasets= {"Wind power generation forecast": 1}, start_time_utc='2025-12-01 05:00:00.000+00:00', end_time_utc='2025-12-01 20:00:00.000+00:00')
            ```
        """

        dataset_ids = ",".join(str(id) for id in datasets.values())
        params = {
            "datasets": dataset_ids,
            "startTime": start_time_utc,
            "endTime": end_time_utc,
            "format": format,
            "oneRowPerTimePeriod": one_row_per_time_period,
            "locale": self.locale,
            "pageSize": page_size,
        }

        headers = {"x-api-key": self.api_key}

        try:
            response = requests.get(self.base_url, headers=headers, params=params)
            response.raise_for_status()

            data = response.json()["data"]
            df = pd.json_normalize(data)

            return df

        except requests.RequestException as e:
            print(f"HTTP request error: {e}")

        except ValueError as e:
            print(f"Data parsing error: {e}")
            return None

    def reformat(
        self, df: pd.DataFrame, variable_id: str, variable_unit: str
    ) -> pd.DataFrame:
        """Reformats the Fingrid API DataFrame to the standardized CommonTradingData format.

        Args:
            df (pandas.DataFrame): The input DataFrame from the Fingrid API.
            variable_id (str): The variable ID to assign to the reformatted data, commonly the corresponding market region (e.g., "FI", "NO2").
            variable_unit (str): The unit of the variable (e.g., "cent", "kw").

        Returns:
            pandas.DataFrame: The reformatted DataFrame with standardized columns.

        Example:
            ```python
            from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI
            import pandas as pd

            fingrid_api = FingridAPI()

            df = fingrid_api.fetch_data(datasets= {"Wind power generation forecast": 1}, start_time_utc='2025-12-01 05:00:00.000+00:00', end_time_utc='2025-12-01 20:00:00.000+00:00')
            reformatted_df = fingrid_api.reformat(df, variable_id="FI", variable_unit="kw")
            ```
        """

        out = pd.DataFrame()

        out["start_time_lb_utc"] = pd.to_datetime(df["startTime"], utc=True)
        out["stop_time_lb_utc"] = pd.to_datetime(df["endTime"], utc=True)
        out["db_updated_utc"] = get_utc_now_custom_precision(set_microsecond_0=True)

        value_cols = [c for c in df.columns if c not in ("startTime", "endTime")]
        if not value_cols:
            raise ValueError(
                "No value column found (expected at least one non-time column)."
            )
        value_col = value_cols[0]

        out["variable_id"] = variable_id

        if variable_unit == "cent":
            out["variable_value"] = (
                pd.to_numeric(df[value_col] * 100, errors="coerce")
                .round(0)
                .astype("Int64")
            )
        elif variable_unit == "kw" or variable_unit == "kwh":
            out["variable_value"] = (
                pd.to_numeric(df[value_col] * 1000, errors="coerce")
                .round(0)
                .astype("Int64")
            )
        else:
            out["variable_value"] = pd.to_numeric(df[value_col], errors="coerce")

        out["variable_unit"] = variable_unit

        out["resolution_seconds"] = (
            (out["stop_time_lb_utc"] - out["start_time_lb_utc"])
            .dt.total_seconds()
            .astype(int)
        )

        return out

__init__(api_key=None)

Initialize a Fingrid API client for a specific asset.

Parameters:

Name Type Description Default
api_key str

Optional, used for testing API key, gets secret from the key vault if empty.

 None
Example 
from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

fingrid_api = FingridAPI()
Source code in physical_operations_utils/fingrid_utils/FingridApi.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(self, api_key: str = None):
    """Initialize a Fingrid API client for a specific asset.

    Args:
        api_key: Optional, used for testing API key, gets secret from the key vault if empty.

    Example:
        ```python
        from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

        fingrid_api = FingridAPI()
        ```
    """
    self.api_key = api_key
    self.base_url = "https://data.fingrid.fi/api/data"
    self.locale = "en"
    if api_key is None:
        self.api_key = get_secret(get_keys_yaml_file().get("fingrid").get("secret"))

fetch_data(datasets, start_time_utc, end_time_utc, format='json', one_row_per_time_period=True, page_size=20000)

Returns data from Fingrid API for the specified datasets and time range.

Parameters:

Name Type Description Default
datasets dict

A dictionary where keys are dataset names and values are dataset IDs.

 required
start_time_utc str

The start time for the data retrieval in ISO 8601 format.

 required
end_time_utc str

The end time for the data retrieval in ISO 8601 format.

 required
format str

The format of the returned data. Defaults to "json".

 'json'
one_row_per_time_period bool

Whether to return one row per time period. Defaults to True.

 True
page_size int

The number of records per page. Defaults to 20000.

 20000

Returns:

Type Description
DataFrame

pandas.DataFrame: A DataFrame containing the time intervals and corresponding values for the requested datasets.
Example 
from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

fingrid_api = FingridAPI()
df = fingrid_api.fetch_data(datasets= {"Wind power generation forecast": 1}, start_time_utc='2025-12-01 05:00:00.000+00:00', end_time_utc='2025-12-01 20:00:00.000+00:00')
Source code in physical_operations_utils/fingrid_utils/FingridApi.py 
 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
def fetch_data(
    self,
    datasets: dict,
    start_time_utc: str,
    end_time_utc: str,
    format: str = "json",
    one_row_per_time_period: bool = True,
    page_size=20000,
) -> pd.DataFrame:
    """Returns data from Fingrid API for the specified datasets and time range.

    Args:
        datasets (dict): A dictionary where keys are dataset names and values are dataset IDs.
        start_time_utc (str): The start time for the data retrieval in ISO 8601 format.
        end_time_utc (str): The end time for the data retrieval in ISO 8601 format.
        format (str, optional): The format of the returned data. Defaults to "json".
        one_row_per_time_period (bool, optional): Whether to return one row per time period. Defaults to True.
        page_size (int, optional): The number of records per page. Defaults to 20000.

    Returns:
        pandas.DataFrame: A DataFrame containing the time intervals and corresponding values for the requested datasets.

    Example:
        ```python
        from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI

        fingrid_api = FingridAPI()
        df = fingrid_api.fetch_data(datasets= {"Wind power generation forecast": 1}, start_time_utc='2025-12-01 05:00:00.000+00:00', end_time_utc='2025-12-01 20:00:00.000+00:00')
        ```
    """

    dataset_ids = ",".join(str(id) for id in datasets.values())
    params = {
        "datasets": dataset_ids,
        "startTime": start_time_utc,
        "endTime": end_time_utc,
        "format": format,
        "oneRowPerTimePeriod": one_row_per_time_period,
        "locale": self.locale,
        "pageSize": page_size,
    }

    headers = {"x-api-key": self.api_key}

    try:
        response = requests.get(self.base_url, headers=headers, params=params)
        response.raise_for_status()

        data = response.json()["data"]
        df = pd.json_normalize(data)

        return df

    except requests.RequestException as e:
        print(f"HTTP request error: {e}")

    except ValueError as e:
        print(f"Data parsing error: {e}")
        return None

reformat(df, variable_id, variable_unit)

Reformats the Fingrid API DataFrame to the standardized CommonTradingData format.

Parameters:

Name Type Description Default
df DataFrame

The input DataFrame from the Fingrid API.

 required
variable_id str

The variable ID to assign to the reformatted data, commonly the corresponding market region (e.g., "FI", "NO2").

 required
variable_unit str

The unit of the variable (e.g., "cent", "kw").

 required

Returns:

Type Description
DataFrame

pandas.DataFrame: The reformatted DataFrame with standardized columns.
Example 
from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI
import pandas as pd

fingrid_api = FingridAPI()

df = fingrid_api.fetch_data(datasets= {"Wind power generation forecast": 1}, start_time_utc='2025-12-01 05:00:00.000+00:00', end_time_utc='2025-12-01 20:00:00.000+00:00')
reformatted_df = fingrid_api.reformat(df, variable_id="FI", variable_unit="kw")
Source code in physical_operations_utils/fingrid_utils/FingridApi.py 
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
def reformat(
    self, df: pd.DataFrame, variable_id: str, variable_unit: str
) -> pd.DataFrame:
    """Reformats the Fingrid API DataFrame to the standardized CommonTradingData format.

    Args:
        df (pandas.DataFrame): The input DataFrame from the Fingrid API.
        variable_id (str): The variable ID to assign to the reformatted data, commonly the corresponding market region (e.g., "FI", "NO2").
        variable_unit (str): The unit of the variable (e.g., "cent", "kw").

    Returns:
        pandas.DataFrame: The reformatted DataFrame with standardized columns.

    Example:
        ```python
        from physical_operations_utils.fingrid_utils.FingridApi import FingridAPI
        import pandas as pd

        fingrid_api = FingridAPI()

        df = fingrid_api.fetch_data(datasets= {"Wind power generation forecast": 1}, start_time_utc='2025-12-01 05:00:00.000+00:00', end_time_utc='2025-12-01 20:00:00.000+00:00')
        reformatted_df = fingrid_api.reformat(df, variable_id="FI", variable_unit="kw")
        ```
    """

    out = pd.DataFrame()

    out["start_time_lb_utc"] = pd.to_datetime(df["startTime"], utc=True)
    out["stop_time_lb_utc"] = pd.to_datetime(df["endTime"], utc=True)
    out["db_updated_utc"] = get_utc_now_custom_precision(set_microsecond_0=True)

    value_cols = [c for c in df.columns if c not in ("startTime", "endTime")]
    if not value_cols:
        raise ValueError(
            "No value column found (expected at least one non-time column)."
        )
    value_col = value_cols[0]

    out["variable_id"] = variable_id

    if variable_unit == "cent":
        out["variable_value"] = (
            pd.to_numeric(df[value_col] * 100, errors="coerce")
            .round(0)
            .astype("Int64")
        )
    elif variable_unit == "kw" or variable_unit == "kwh":
        out["variable_value"] = (
            pd.to_numeric(df[value_col] * 1000, errors="coerce")
            .round(0)
            .astype("Int64")
        )
    else:
        out["variable_value"] = pd.to_numeric(df[value_col], errors="coerce")

    out["variable_unit"] = variable_unit

    out["resolution_seconds"] = (
        (out["stop_time_lb_utc"] - out["start_time_lb_utc"])
        .dt.total_seconds()
        .astype(int)
    )

    return out