cid2_utils

CID 2 utils are used to interact with the CID 2 API to retrieve live intraday market data.

CID2Api

Source code in physical_operations_utils/cid2_utils.py

class CID2Api:
    def __init__(self):
        self._verify_ssl = True
        if os.environ.get("ENVIRONMENT") == "prod":
            self.cid_api_base_url = CID_BASE_URL_PROD
        else:
            self.cid_api_base_url = CID_BASE_URL_NONPROD
            self._verify_ssl = (
                False  # Temporary fix for expired SSL certificate in nonprod
            )
        self.cid_api_username = get_secret("CustomerIntraday-PhysApiUsername")
        self.cid_api_secret = "CustomerIntraday-PhysApiPassword"
        self.cid_api_password = get_secret(self.cid_api_secret)
        logger = logging.getLogger(name="cid_2_api")
        logger.setLevel(logging.INFO)
        self.logger = logger
        self._auth_token: Optional[str] = None
        self._last_authenticated: Optional[datetime] = None

    def get_cid2_health(self) -> Dict[str, str]:
        """
        Load last updated and last inserted timestamps for different id feed data. Can be used to make a health check on the intraday feed.

        Returns:
            Dict[str, str]:
                A dictionary containing last updated and last inserted timestamps for different id feed data.

        Example:
            ```python
            from physical_operations_utils.cid2_utils import CID2API

            api = CID2Api()

            print(api.get_cid2_health())
            ```
        """
        auth_token = self._authenticate()
        headers = self._build_headers(auth_token=auth_token)
        res = requests.get(
            url=f"{self.cid_api_base_url}/api/v1/Admin/intraday/health",
            headers=headers,
            verify=self._verify_ssl,
        )
        if res.status_code != 200:
            msg = f"Error when fetching public statistics: {res.status_code} {res.text}"
            self.logger.error(msg)
            raise Exception(msg)

        body = res.json()
        return body

    def load_id_capacities_per_area(
        self,
        start_time_lb_utc: datetime,
        stop_time_lb_utc: datetime,
    ) -> pd.DataFrame:
        """
        Load ID capacities per area within a specified UTC time window.

        This function retrieves capacity records from the NordpoolIntradayFeed database for all price areas
        that have capacities between the provided lower and upper time bounds.
        It validates that the provided datetime values are in UTC and that the start time precedes
        the stop time before constructing and executing a SQL query.

        Args:
            start_time_lb_utc: The lower bound of the delivery start time in UTC.
            stop_time_lb_utc: The upper bound of the delivery start time in UTC.

        Returns:
            pandas.DataFrame:
                A DataFrame containing the following columns:
                    start_time_lb_utc: Delivery start times as UTC datetime objects.
                    stop_time_lb_utc: Delivery end times as UTC datetime objects.
                    price_area_from: Price area code corresponding to the delivery's origin.
                    price_area_to: Price area code corresponding to the delivery's destination.
                    in_capacity: Inbound capacity.
                    out_capacity: Outbound capacity.

        Raises:
            ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
            ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.
            sqlalchemy.exc.SQLAlchemyError: If there is an issue executing the database query.

        Example:
            ```python
            from physical_operations_utils.cid2_utils import CID2API
            from zoneinfo import ZoneInfo

            api = CID2Api()
            start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
            stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

            print(api.load_id_capacities_per_area(start_time_lb_utc, stop_time_lb_utc))
            ```
        """
        validate_datetime_in_utc(start_time_lb_utc)
        validate_datetime_in_utc(stop_time_lb_utc)
        validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)

        capacities_list = self._get_id_capacities_by_time(
            start_time_lb_utc=start_time_lb_utc,
            stop_time_rb_uct=stop_time_lb_utc + timedelta(hours=1),
        )
        df = pd.DataFrame(capacities_list)
        df["start_time_lb_utc"] = pd.to_datetime(df["deliveryStart"], utc=True)
        df["stop_time_lb_utc"] = pd.to_datetime(df["deliveryEnd"], utc=True)
        df["price_area_from"] = df["deliveryAreaFrom"].astype(str)
        df["price_area_to"] = df["deliveryAreaTo"].astype(str)
        df["in_capacity"] = df["inCapacity"].astype("int64")
        df["out_capacity"] = df["outCapacity"].astype("int64")
        df["resolution_seconds"] = (
            (df["stop_time_lb_utc"] - df["start_time_lb_utc"])
            .dt.total_seconds()
            .astype("int64")
        )
        return df.loc[
            (df["start_time_lb_utc"] >= start_time_lb_utc)
            & (df["stop_time_lb_utc"] <= stop_time_lb_utc),
            [
                "start_time_lb_utc",
                "stop_time_lb_utc",
                "price_area_from",
                "price_area_to",
                "in_capacity",
                "out_capacity",
                "resolution_seconds",
            ],
        ]

    def get_net_id_private_trades_per_time_interval_and_strategies_as_df(
        self,
        start_time_lb_utc: datetime,
        stop_time_lb_utc: datetime,
        resolution_seconds: int,
        include_strategies: set = {},
        exclude_strategies: set = {},
        use_net_buy_or_sell: str = "net",
        spread_hourly_trades_quarters: bool = True,
    ) -> pd.DataFrame:
        """
        Loads and aggregates net intraday traded volumes per resolution. If 15 minute resolution is requested, the function adds volumes of hourly
        products to the corresponding 15 minute time intervals. Target resolutions are 3600 and 900 seconds. Strategies can be filtered based on the
        text label of each trade.

        Parameters:
            start_time_lb_utc: The start time (UTC) for retrieving net traded volumes.
            stop_time_lb_utc: The end time (UTC) for retrieving net traded volumes.
            resolution_seconds: The desired resolution in seconds (e.g., 3600 for hourly, 900 for 15-minute).
            include_strategies: Set of strings to filter trades on. Include every trade where text label contains any given string.
            exclude_strategies: Set of strings to filter trades on. Exclude every trade where text label contains any given string.
            use_net_buy_or_sell: String to specify if net traded volume, only buy or only sell volumes should be included. Must be 'net' (default), 'buy' or 'sell'.
            spread_hourly_trades_quarters: Boolean flag to specify if hourly traded volumes should be included in quarterly resolution.

        Returns:
            pd.DataFrame: A DataFrame containing aggregated net traded volumes per resolution with columns
                `start_time_lb_utc`: The start time of the interval.
                `stop_time_lb_utc`: The stop time of the interval.
                `variable_id`: The price area.
                `id_net_traded`: The net traded volume in kW.
                `product`: The associated Nord Pool product.

        Raises:
            ValueError: If `resolution_seconds` is not supported.
            ValueError: If times are not provided as UTC or if the start time is not before the stop time.
            ValueError: If use_net_buy_or_sell is not 'net', 'buy' or 'sell'

        Example:
            ```python
            from physical_operations_utils.cid2_utils import CID2API
            from zoneinfo import ZoneInfo

            api = CID2Api()
            start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
            stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))
            resolution_seconds = 900

            print(api.get_net_id_private_trades_per_time_interval_and_strategies_as_df(start_time_lb_utc, stop_time_lb_utc, resolution_seconds))
            ```
        """
        if resolution_seconds not in [3600, 900]:
            raise ValueError(
                f"Resolution of {resolution_seconds} seconds is not supported. Supported resolutions are 3600 and 900 seconds."
            )
        if use_net_buy_or_sell not in ["net", "buy", "sell"]:
            raise ValueError("use_net_buy_or_sell must  be 'net', 'buy' or 'sell'")
        validate_datetime_in_utc(start_time_lb_utc)
        validate_datetime_in_utc(stop_time_lb_utc)
        validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)

        private_trades_dict = self._get_id_private_trades_per_time_interval(
            start_time_lb_utc=start_time_lb_utc,
            stop_time_rb_utc=stop_time_lb_utc + timedelta(hours=1),
        )
        if len(include_strategies) == 0 and len(exclude_strategies) == 0:
            filtered_trades = private_trades_dict.copy()
        else:
            if len(include_strategies) > 0:
                filtered_trades_temp = [
                    trade
                    for trade in private_trades_dict
                    if any(
                        [
                            strategy in trade.get("text", "")
                            for strategy in include_strategies
                        ]
                    )
                ]
            else:
                filtered_trades_temp = private_trades_dict.copy()
            filtered_trades = [
                trade
                for trade in filtered_trades_temp
                if not any(
                    [
                        strategy in trade.get("text", "")
                        for strategy in exclude_strategies
                    ]
                )
            ]

        if use_net_buy_or_sell == "net":
            filtered_trades_side = filtered_trades.copy()
        else:
            filtered_trades_side = [
                trade
                for trade in filtered_trades
                if trade.get("orderSide", "").lower() == use_net_buy_or_sell
            ]

        trades_df = pd.DataFrame(filtered_trades_side)

        if trades_df.empty:
            id_net_traded_final = (
                generate_empty_df_with_start_stop_time_lb_utc_variable_id_and_variable_value(
                    start_time_lb_utc=start_time_lb_utc,
                    stop_time_lb_utc=stop_time_lb_utc + timedelta(hours=1),
                    resolution_seconds=resolution_seconds,
                    variable_ids=BALANCE_PRICE_AREAS,
                    variable_value_data_type="int",
                )
                .drop(columns=["resolution_seconds"])
                .rename(columns={"variable_value": "id_net_traded"})
            )
            id_net_traded_final["product"] = id_net_traded_final.apply(
                lambda row: nordpool_product_namer(
                    row["start_time_lb_utc"], resolution_seconds
                ),
                axis=1,
            )
            return id_net_traded_final

        trades_df = trades_df[
            ["deliveryStart", "deliveryEnd", "orderSide", "quantity", "deliveryAreaId"]
        ].rename(
            columns={
                "deliveryStart": "start_time_lb_utc",
                "deliveryEnd": "stop_time_lb_utc",
            }
        )
        trades_df["start_time_lb_utc"] = pd.to_datetime(
            trades_df["start_time_lb_utc"], utc=True
        )
        trades_df["stop_time_lb_utc"] = pd.to_datetime(
            trades_df["stop_time_lb_utc"], utc=True
        )

        trades_df["signed_quantity"] = trades_df.apply(
            lambda row: (
                row["quantity"]
                if str(row["orderSide"]).lower() == "buy"
                else -row["quantity"]
            ),
            axis=1,
        )

        trades_df["resolution_seconds"] = (
            (trades_df["stop_time_lb_utc"] - trades_df["start_time_lb_utc"])
            .dt.total_seconds()
            .astype("int64")
        )

        # Aggregate
        net_trades_df = (
            trades_df.groupby(
                [
                    "start_time_lb_utc",
                    "stop_time_lb_utc",
                    "resolution_seconds",
                    "deliveryAreaId",
                ],
                as_index=False,
            )["signed_quantity"]
            .sum()
            .astype("int64")
        )

        net_trades_df["start_time_lb_utc"] = pd.to_datetime(
            net_trades_df["start_time_lb_utc"], utc=True
        )
        net_trades_df["stop_time_lb_utc"] = pd.to_datetime(
            net_trades_df["stop_time_lb_utc"], utc=True
        )
        net_trades_df["deliveryAreaId"] = net_trades_df["deliveryAreaId"].map(
            DELIVERY_AREA_ID_MAP
        )
        net_trades_df = net_trades_df.rename(
            columns={"signed_quantity": "id_net_traded", "deliveryAreaId": "price_area"}
        )

        id_net_traded_final = generate_empty_df_with_start_stop_time_lb_utc_variable_id_and_variable_value(
            start_time_lb_utc=start_time_lb_utc,
            stop_time_lb_utc=stop_time_lb_utc + timedelta(hours=1),
            resolution_seconds=resolution_seconds,
            variable_ids=BALANCE_PRICE_AREAS,
            variable_value_data_type="int",
        )
        id_net_traded_final = id_net_traded_final.rename(
            columns={"variable_value": "id_net_traded", "variable_id": "price_area"}
        )
        id_net_traded_hourly_df = (
            net_trades_df[net_trades_df["resolution_seconds"] == 3600]
            .copy(deep=True)
            .sort_values(by="start_time_lb_utc")
        )
        for _, row in id_net_traded_hourly_df.iterrows():
            if resolution_seconds == 3600 or (
                resolution_seconds == 900 and spread_hourly_trades_quarters
            ):
                for i in range(3600 // resolution_seconds):
                    resampled_start_time_lb_utc = row[
                        "start_time_lb_utc"
                    ] + pd.Timedelta(seconds=i * resolution_seconds)
                    id_net_traded_final.loc[
                        (
                            id_net_traded_final["start_time_lb_utc"]
                            == resampled_start_time_lb_utc
                        )
                        & (
                            id_net_traded_final["stop_time_lb_utc"]
                            == resampled_start_time_lb_utc
                            + pd.Timedelta(seconds=resolution_seconds)
                        )
                        & (id_net_traded_final["price_area"] == row["price_area"]),
                        "id_net_traded",
                    ] += row["id_net_traded"]

        if resolution_seconds == 900:  # Use hourly values also for every quarter
            id_net_traded_15min_df = net_trades_df[
                net_trades_df["resolution_seconds"] == 900
            ].copy(deep=True)
            for _, row in id_net_traded_15min_df.iterrows():
                id_net_traded_final.loc[
                    (
                        id_net_traded_final["start_time_lb_utc"]
                        == row["start_time_lb_utc"]
                    )
                    & (
                        id_net_traded_final["stop_time_lb_utc"]
                        == row["stop_time_lb_utc"]
                    )
                    & (id_net_traded_final["price_area"] == row["price_area"]),
                    "id_net_traded",
                ] += row["id_net_traded"]

        id_net_traded_final = id_net_traded_final.drop(columns=["resolution_seconds"])
        id_net_traded_final = id_net_traded_final.rename(
            columns={"price_area": "variable_id"}
        )
        id_net_traded_final["product"] = id_net_traded_final.apply(
            lambda row: nordpool_product_namer(
                row["start_time_lb_utc"], resolution_seconds
            ),
            axis=1,
        )
        return id_net_traded_final

    def load_best_bid_and_best_offer(
        self,
        start_time_lb_utc: datetime,
        stop_time_lb_utc: datetime,
        price_areas: List[str] = [],
        resolutions_seconds: List[int] = [],
    ) -> pd.DataFrame:
        """
        Computes the best bid and best offer on the intraday market based on active public orders within a specified time range.

        Args:
            start_time_lb_utc (datetime): The lower bound of the delivery start time in UTC.
            stop_time_lb_utc (datetime): The upper bound of the delivery start time in UTC.
            price_areas (List[str]): List of price areas to be included. If left empty, all areas present in CID will be returned.
            resolutions_seconds (List[int]): List of resolutions to be included. If left empty, all areas present in CID will be returned.

        Returns:
            pd.DataFrame: A DataFrame containing aggregated order book data with columns
                `start_time_lb_utc` (datetime64[ns, UTC]): Contract delivery start time.
                `stop_time_lb_utc` (datetime64[ns, UTC]): Contract delivery end time.
                `price_area` (str): Area code for pricing.
                `contract_id` (str): Unique contract identifier.
                `product` (str): Contract product name.
                `product_type` (str): Type of product.
                `resolution_seconds` (Int64): Resolution of the contract in seconds.
                `best_bid_ct` (Int64): Highest bid price.
                `best_offer_ct` (Int64): Lowest offer price.

        Raises:
            ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
            ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.

        Example:
            ```python
            from physical_operations_utils.cid2_utils import CID2API
            from zoneinfo import ZoneInfo

            api = CID2Api()
            start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
            stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

            print(api.load_best_bid_and_best_offer(start_time_lb_utc, stop_time_lb_utc))
            ```
        """
        validate_datetime_in_utc(start_time_lb_utc)
        validate_datetime_in_utc(stop_time_lb_utc)
        validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)
        expected_api_response_keys = [
            "deliveryStart",
            "deliveryEnd",
            "deliveryArea",
            "contractId",
            "product",
            "productType",
            "bestBid",
            "bestOffer",
            "resolutionSeconds",
        ]
        api_response = self._get_best_bid_best_offer_per_time_interval(
            start_time_lb_utc=start_time_lb_utc,
            stop_time_rb_utc=stop_time_lb_utc,
            price_areas=price_areas,
            resolutions_seconds=resolutions_seconds,
        )
        best_bid_best_offer_and_offered_qty_df = pd.DataFrame(api_response)
        if best_bid_best_offer_and_offered_qty_df.empty:
            best_bid_best_offer_and_offered_qty_df = pd.DataFrame(
                columns=expected_api_response_keys
            )
        best_bid_best_offer_and_offered_qty_df = (
            best_bid_best_offer_and_offered_qty_df.rename(
                columns={
                    "deliveryArea": "price_area",
                    "contractId": "contract_id",
                    "productType": "product_type",
                }
            )
        )
        best_bid_best_offer_and_offered_qty_df.loc[:, "start_time_lb_utc"] = (
            pd.to_datetime(best_bid_best_offer_and_offered_qty_df["deliveryStart"])
        )
        best_bid_best_offer_and_offered_qty_df.loc[:, "stop_time_lb_utc"] = (
            pd.to_datetime(best_bid_best_offer_and_offered_qty_df["deliveryEnd"])
        )
        best_bid_best_offer_and_offered_qty_df.loc[:, "best_bid_ct"] = (
            best_bid_best_offer_and_offered_qty_df.loc[:, "bestBid"].astype("Int64")
        )
        best_bid_best_offer_and_offered_qty_df.loc[:, "best_offer_ct"] = (
            best_bid_best_offer_and_offered_qty_df.loc[:, "bestOffer"].astype("Int64")
        )
        best_bid_best_offer_and_offered_qty_df.loc[:, "resolution_seconds"] = (
            best_bid_best_offer_and_offered_qty_df.loc[:, "resolutionSeconds"].astype(
                "Int64"
            )
        )
        col_order = [
            "start_time_lb_utc",
            "stop_time_lb_utc",
            "price_area",
            "contract_id",
            "product",
            "product_type",
            "resolution_seconds",
            "best_offer_ct",
            "best_bid_ct",
        ]
        return best_bid_best_offer_and_offered_qty_df.loc[:, col_order]

    def load_id_orderbook_with_contract_information(
        self,
        start_time_lb_utc: datetime,
        stop_time_lb_utc: datetime,
    ) -> pd.DataFrame:
        """
        Retrieves intraday orders with associated contract information from the CustomerIntraday database.

        Args:
            start_time_lb_utc: The lower bound of the delivery start time in UTC.
            stop_time_lb_utc: The upper bound of the delivery start time in UTC.

        Returns:
            pd.DataFrame: A DataFrame containing order book data with columns:
                `start_time_lb_utc` (datetime64[ns, UTC]): Contract delivery start time.
                `stop_time_lb_utc` (datetime64[ns, UTC]): Contract delivery end time.
                `price_area` (str): Price area.
                `price` (int64): Order price in cent.
                `qty` (int64): Order quantity in kW.
                `direction` (str): Order direction (i.e., BUY/SELL).
                `contract_id` (str): Unique contract identifier.
                `product` (str): Contract product name.
                `product_type` (str): Type of product.
                `resolution_seconds` (int64): Resolution of the contract in seconds.

        Raises:
            ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
            ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.

        Example:
            ```python
            from physical_operations_utils.cid2_utils import CID2API
            from zoneinfo import ZoneInfo

            api = CID2Api()
            start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
            stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

            print(api.load_id_orderbook_with_contract_information(start_time_lb_utc, stop_time_lb_utc))
            ```
        """
        validate_datetime_in_utc(start_time_lb_utc)
        validate_datetime_in_utc(stop_time_lb_utc)
        validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)

        expected_api_response_keys = [
            "deliveryStart",
            "deliveryEnd",
            "deliveryAreaId",
            "price",
            "qty",
            "orderSide",
            "contractId",
            "contractName",
            "productType",
            "resolutionSeconds",
        ]

        orderbook = self._get_id_public_orders_per_time_interval(
            start_time_lb_utc=start_time_lb_utc, stop_time_rb_utc=stop_time_lb_utc
        )

        orderbook_df = pd.DataFrame(orderbook)

        if orderbook_df.empty:
            orderbook_df = pd.DataFrame(columns=expected_api_response_keys)

        orderbook_df["start_time_lb_utc"] = pd.to_datetime(
            orderbook_df["deliveryStart"], utc=True
        )
        orderbook_df["stop_time_lb_utc"] = pd.to_datetime(
            orderbook_df["deliveryEnd"], utc=True
        )
        orderbook_df["price_area"] = (
            orderbook_df["deliveryAreaId"].map(DELIVERY_AREA_ID_MAP).astype(str)
        )
        orderbook_df["price"] = orderbook_df["price"].astype("int64")
        orderbook_df["qty"] = orderbook_df["qty"].astype("int64")
        orderbook_df["direction"] = orderbook_df["orderSide"].str.upper().astype(str)
        orderbook_df["contract_id"] = orderbook_df["contractId"].astype(str)
        orderbook_df["product"] = orderbook_df["contractName"].astype(str)
        orderbook_df["product_type"] = orderbook_df["productType"].astype(str)
        orderbook_df["resolution_seconds"] = orderbook_df["resolutionSeconds"].astype(
            "int64"
        )
        return orderbook_df.loc[
            :,
            [
                "start_time_lb_utc",
                "stop_time_lb_utc",
                "price_area",
                "price",
                "qty",
                "direction",
                "contract_id",
                "product",
                "product_type",
                "resolution_seconds",
            ],
        ]

    def load_id_public_statistics_with_contract_information(
        self, start_time_lb_utc: datetime, stop_time_lb_utc: datetime
    ) -> pd.DataFrame:
        """
        Retrieves intraday public statistics with associated contract information from the CID2 intraday feed. Returns only information for
        products with resolution 900s, 1800s and 3600s.

        Args:
            start_time_lb_utc: The lower bound of the delivery start time in UTC.
            stop_time_lb_utc: The upper bound of the delivery start time in UTC.

        Returns:
            pd.DataFrame: A DataFrame containing public statistics data with columns:
                `start_time_lb_utc` (datetime64[ns, UTC]): Contract delivery start time.
                `stop_time_lb_utc` (datetime64[ns, UTC]): Contract delivery end time.
                `price_area` (str): Price area.
                `last_price` (int64): Last traded price in ct.
                `last_quantity` (int64): Last traded quantity in kW.
                `last_trade_time` (datetime64[ns, UTC]): Time of last trade. If no trades were made yet it is set to datetime(1, 1, 1, tzinfo=ZoneInfo("UTC")).
                `highest_price` (int64): Highest trade price in ct.
                `vwap` (int64): Volume-weighted average price in ct.
                `turnover` (int64): Turnover in kW.
                `spot_price_eur_ct` (int64): Day-ahead market price in euro cents.
                `tendency` (str): Market tendency (e.g., UP, DOWN, EQUAL).
                `contract_id` (str): Unique contract identifier.
                `product` (str): Contract product name.
                `product_type` (str): Type of product.
                `resolution_seconds` (int64): Resolution of the contract in seconds.

        Raises:
            ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
            ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.

        Example:
            ```python
            from physical_operations_utils.cid2_utils import CID2API
            from zoneinfo import ZoneInfo

            api = CID2Api()
            start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
            stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

            print(api.load_id_public_statistics_with_contract_information(start_time_lb_utc, stop_time_lb_utc))
            ```
        """
        validate_datetime_in_utc(start_time_lb_utc)
        validate_datetime_in_utc(stop_time_lb_utc)
        validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)
        columns = [
            "start_time_lb_utc",
            "stop_time_lb_utc",
            "price_area",
            "last_price",
            "last_quantity",
            "last_trade_time",
            "highest_price",
            "vwap",
            "turnover",
            "spot_price_eur_ct",
            "tendency",
            "contract_id",
            "product",
            "product_type",
            "resolution_seconds",
        ]
        response = self._get_id_public_statistics_per_time_interval(
            start_time_lb_utc=start_time_lb_utc,
            stop_time_rb_utc=stop_time_lb_utc + timedelta(hours=1),
        )
        df = pd.DataFrame(response)
        if df.empty:
            return pd.DataFrame(columns=columns)
        df["start_time_lb_utc"] = pd.to_datetime(df["deliveryStart"], utc=True)
        df["stop_time_lb_utc"] = pd.to_datetime(df["deliveryEnd"], utc=True)
        df["price_area"] = df["deliveryArea"].astype(str)
        df["last_price"] = df["lastPrice"].astype("float64").round(0)
        df["last_quantity"] = df["lastQuantity"].astype("float64").round(0)
        df["last_trade_time"] = pd.to_datetime(
            df["lastTradeTime"], utc=True, errors="coerce"
        )
        df["highest_price"] = df["highestPrice"].astype("float64").round(0)
        df["vwap"] = df["vwap"].astype("float64").round(0)
        df["turnover"] = df["turnover"].astype("float64").round(0)
        df["spot_price_eur_ct"] = df["dayAheadPrice"].astype("float64").round(0)
        df["tendency"] = df["tendency"].astype(str)
        df["contract_id"] = df["contractId"].astype(str)
        df["resolution_seconds"] = (
            (df["stop_time_lb_utc"] - df["start_time_lb_utc"])
            .dt.total_seconds()
            .round(0)
            .astype("int64")
        )
        df = filter_dataframe_by_resolution_seconds(
            df=df, keep_resolutions=[900, 1800, 3600]
        )
        if not df.empty:
            df["product"] = df.apply(
                lambda row: nordpool_product_namer(
                    start_time_lb_utc=row["start_time_lb_utc"],
                    resolution_seconds=row["resolution_seconds"],
                ),
                axis=1,
            )
            df["product_type"] = df.apply(
                lambda row: NORDPOOL_PRODUCT_TYPE_MAP.get(
                    row["resolution_seconds"], "NA"
                ),
                axis=1,
            )
        else:
            df["product"] = "NA"
            df["product_type"] = "NA"
        return df.loc[:, columns]

    def _authenticate(self) -> str:
        now_utc = get_utc_now_custom_precision()
        if (
            self._auth_token
            and self._last_authenticated
            and (now_utc - self._last_authenticated).total_seconds() < 3600
        ):
            return self._auth_token
        self._last_authenticated = now_utc
        auth_res = requests.post(
            url=f"{self.cid_api_base_url}/api/v1/customer/Account/Authenticate/",
            json={"username": self.cid_api_username, "password": self.cid_api_password},
            headers={"Content-Type": "application/json"},
            verify=self._verify_ssl,
        )
        if auth_res.status_code != 200:
            msg = f"Failed to authenticate with CID API: {auth_res.status_code} {auth_res.text}"
            self.logger.error(msg)
            self._last_authenticated = None
            self._auth_token = None
            raise Exception(msg)
        auth_res_body = auth_res.json()
        self._auth_token = auth_res_body["authToken"]
        return self._auth_token

    def _build_headers(self, auth_token: str) -> dict:
        return {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {auth_token}",
        }

    def _get_best_bid_best_offer_per_time_interval(
        self,
        start_time_lb_utc: datetime,
        stop_time_rb_utc: datetime,
        price_areas: List[str] = [],
        resolutions_seconds: List[int] = [],
    ):
        auth_token = self._authenticate()
        headers = self._build_headers(auth_token=auth_token)
        res = requests.get(
            url=f"{self.cid_api_base_url}/api/v1/intraday/Contract/price",
            headers=headers,
            json={
                "fromDate": start_time_lb_utc.isoformat(),
                "toDate": stop_time_rb_utc.isoformat(),
                "deliveryAreas": price_areas,
                "resolutions": resolutions_seconds,
            },
            verify=self._verify_ssl,
        )

        if res.status_code != 200:
            msg = f"Error when fetching best bid and best offer: {res.status_code} {res.text}"
            self.logger.error(msg)
            raise Exception(msg)

        body = res.json()
        return body

    def _get_id_private_trades_per_time_interval(
        self, start_time_lb_utc: datetime, stop_time_rb_utc: datetime
    ) -> List[dict]:
        auth_token = self._authenticate()
        headers = self._build_headers(auth_token=auth_token)
        res = requests.get(
            url=f"{self.cid_api_base_url}/api/v1/intraday/private-trade/",
            headers=headers,
            params={
                "start": start_time_lb_utc.isoformat(),
                "end": stop_time_rb_utc.isoformat(),
            },
            verify=self._verify_ssl,
        )

        if res.status_code != 200:
            msg = f"Error when fetching private trades: {res.status_code} {res.text}"
            self.logger.error(msg)
            raise Exception(msg)

        body = res.json()
        return body

    def _get_id_public_orders_per_time_interval(
        self, start_time_lb_utc: datetime, stop_time_rb_utc: datetime
    ) -> List[dict]:
        auth_token = self._authenticate()
        headers = self._build_headers(auth_token=auth_token)
        res = requests.get(
            url=f"{self.cid_api_base_url}/api/v1/intraday/Order/",
            headers=headers,
            params={
                "start": start_time_lb_utc.isoformat(),
                "end": stop_time_rb_utc.isoformat(),
            },
            verify=self._verify_ssl,
        )
        if res.status_code != 200:
            msg = f"Error when fetching public orders: {res.status_code} {res.text}"
            self.logger.error(msg)
            raise Exception(msg)

        body = res.json()
        return body

    def _get_id_capacities_by_time(
        self, start_time_lb_utc: datetime, stop_time_rb_uct: datetime
    ) -> List[dict]:
        auth_token = self._authenticate()
        headers = self._build_headers(auth_token=auth_token)
        include_areas = [
            "EE",
            "FI",
            "LT",
            "NO2",
            "NO3",
            "NO4",
            "NO5",
            "SE1",
            "SE2",
            "SE3",
            "SE4",
        ]
        body = {
            "fromAreas": include_areas,
            "toAreas": include_areas,
            "fromDate": start_time_lb_utc.isoformat(),
            "toDate": stop_time_rb_uct.isoformat(),
        }
        res = requests.post(
            url=f"{self.cid_api_base_url}/api/v1/intraday/Capacity/",
            headers=headers,
            json=body,
            verify=self._verify_ssl,
        )
        if res.status_code != 200:
            msg = f"Error when fetching id capacities: {res.status_code} {res.text}"
            self.logger.error(msg)
            raise Exception(msg)
        body = res.json()
        return body

    def _get_id_public_statistics_per_time_interval(
        self, start_time_lb_utc: datetime, stop_time_rb_utc: datetime
    ) -> List[dict]:
        auth_token = self._authenticate()
        headers = self._build_headers(auth_token=auth_token)
        res = requests.get(
            url=f"{self.cid_api_base_url}/api/v1/intraday/public-statistic",
            headers=headers,
            params={
                "start": start_time_lb_utc.isoformat(),
                "end": stop_time_rb_utc.isoformat(),
            },
            verify=self._verify_ssl,
        )
        if res.status_code != 200:
            msg = f"Error when fetching public statistics: {res.status_code} {res.text}"
            self.logger.error(msg)
            raise Exception(msg)

        body = res.json()
        return body

get_cid2_health()

Load last updated and last inserted timestamps for different id feed data. Can be used to make a health check on the intraday feed.

Returns:

Type	Description
Dict[str, str]	Dict[str, str]: A dictionary containing last updated and last inserted timestamps for different id feed data.

Example

from physical_operations_utils.cid2_utils import CID2API

api = CID2Api()

print(api.get_cid2_health())

Source code in physical_operations_utils/cid2_utils.py

def get_cid2_health(self) -> Dict[str, str]:
    """
    Load last updated and last inserted timestamps for different id feed data. Can be used to make a health check on the intraday feed.

    Returns:
        Dict[str, str]:
            A dictionary containing last updated and last inserted timestamps for different id feed data.

    Example:
        ```python
        from physical_operations_utils.cid2_utils import CID2API

        api = CID2Api()

        print(api.get_cid2_health())
        ```
    """
    auth_token = self._authenticate()
    headers = self._build_headers(auth_token=auth_token)
    res = requests.get(
        url=f"{self.cid_api_base_url}/api/v1/Admin/intraday/health",
        headers=headers,
        verify=self._verify_ssl,
    )
    if res.status_code != 200:
        msg = f"Error when fetching public statistics: {res.status_code} {res.text}"
        self.logger.error(msg)
        raise Exception(msg)

    body = res.json()
    return body

get_net_id_private_trades_per_time_interval_and_strategies_as_df(start_time_lb_utc, stop_time_lb_utc, resolution_seconds, include_strategies={}, exclude_strategies={}, use_net_buy_or_sell='net', spread_hourly_trades_quarters=True)

Loads and aggregates net intraday traded volumes per resolution. If 15 minute resolution is requested, the function adds volumes of hourly products to the corresponding 15 minute time intervals. Target resolutions are 3600 and 900 seconds. Strategies can be filtered based on the text label of each trade.

Parameters:

Name	Type	Description	Default
start_time_lb_utc	datetime	The start time (UTC) for retrieving net traded volumes.	required
stop_time_lb_utc	datetime	The end time (UTC) for retrieving net traded volumes.	required
resolution_seconds	int	The desired resolution in seconds (e.g., 3600 for hourly, 900 for 15-minute).	required
include_strategies	set	Set of strings to filter trades on. Include every trade where text label contains any given string.	{}
exclude_strategies	set	Set of strings to filter trades on. Exclude every trade where text label contains any given string.	{}
use_net_buy_or_sell	str	String to specify if net traded volume, only buy or only sell volumes should be included. Must be 'net' (default), 'buy' or 'sell'.	'net'
spread_hourly_trades_quarters	bool	Boolean flag to specify if hourly traded volumes should be included in quarterly resolution.	True

Returns:

Type	Description
DataFrame	pd.DataFrame: A DataFrame containing aggregated net traded volumes per resolution with columns start_time_lb_utc: The start time of the interval. stop_time_lb_utc: The stop time of the interval. variable_id: The price area. id_net_traded: The net traded volume in kW. product: The associated Nord Pool product.

Raises:

Type	Description
ValueError	If resolution_seconds is not supported.
ValueError	If times are not provided as UTC or if the start time is not before the stop time.
ValueError	If use_net_buy_or_sell is not 'net', 'buy' or 'sell'

Example

from physical_operations_utils.cid2_utils import CID2API
from zoneinfo import ZoneInfo

api = CID2Api()
start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))
resolution_seconds = 900

print(api.get_net_id_private_trades_per_time_interval_and_strategies_as_df(start_time_lb_utc, stop_time_lb_utc, resolution_seconds))

Source code in physical_operations_utils/cid2_utils.py

def get_net_id_private_trades_per_time_interval_and_strategies_as_df(
    self,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
    resolution_seconds: int,
    include_strategies: set = {},
    exclude_strategies: set = {},
    use_net_buy_or_sell: str = "net",
    spread_hourly_trades_quarters: bool = True,
) -> pd.DataFrame:
    """
    Loads and aggregates net intraday traded volumes per resolution. If 15 minute resolution is requested, the function adds volumes of hourly
    products to the corresponding 15 minute time intervals. Target resolutions are 3600 and 900 seconds. Strategies can be filtered based on the
    text label of each trade.

    Parameters:
        start_time_lb_utc: The start time (UTC) for retrieving net traded volumes.
        stop_time_lb_utc: The end time (UTC) for retrieving net traded volumes.
        resolution_seconds: The desired resolution in seconds (e.g., 3600 for hourly, 900 for 15-minute).
        include_strategies: Set of strings to filter trades on. Include every trade where text label contains any given string.
        exclude_strategies: Set of strings to filter trades on. Exclude every trade where text label contains any given string.
        use_net_buy_or_sell: String to specify if net traded volume, only buy or only sell volumes should be included. Must be 'net' (default), 'buy' or 'sell'.
        spread_hourly_trades_quarters: Boolean flag to specify if hourly traded volumes should be included in quarterly resolution.

    Returns:
        pd.DataFrame: A DataFrame containing aggregated net traded volumes per resolution with columns
            `start_time_lb_utc`: The start time of the interval.
            `stop_time_lb_utc`: The stop time of the interval.
            `variable_id`: The price area.
            `id_net_traded`: The net traded volume in kW.
            `product`: The associated Nord Pool product.

    Raises:
        ValueError: If `resolution_seconds` is not supported.
        ValueError: If times are not provided as UTC or if the start time is not before the stop time.
        ValueError: If use_net_buy_or_sell is not 'net', 'buy' or 'sell'

    Example:
        ```python
        from physical_operations_utils.cid2_utils import CID2API
        from zoneinfo import ZoneInfo

        api = CID2Api()
        start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
        stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))
        resolution_seconds = 900

        print(api.get_net_id_private_trades_per_time_interval_and_strategies_as_df(start_time_lb_utc, stop_time_lb_utc, resolution_seconds))
        ```
    """
    if resolution_seconds not in [3600, 900]:
        raise ValueError(
            f"Resolution of {resolution_seconds} seconds is not supported. Supported resolutions are 3600 and 900 seconds."
        )
    if use_net_buy_or_sell not in ["net", "buy", "sell"]:
        raise ValueError("use_net_buy_or_sell must  be 'net', 'buy' or 'sell'")
    validate_datetime_in_utc(start_time_lb_utc)
    validate_datetime_in_utc(stop_time_lb_utc)
    validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)

    private_trades_dict = self._get_id_private_trades_per_time_interval(
        start_time_lb_utc=start_time_lb_utc,
        stop_time_rb_utc=stop_time_lb_utc + timedelta(hours=1),
    )
    if len(include_strategies) == 0 and len(exclude_strategies) == 0:
        filtered_trades = private_trades_dict.copy()
    else:
        if len(include_strategies) > 0:
            filtered_trades_temp = [
                trade
                for trade in private_trades_dict
                if any(
                    [
                        strategy in trade.get("text", "")
                        for strategy in include_strategies
                    ]
                )
            ]
        else:
            filtered_trades_temp = private_trades_dict.copy()
        filtered_trades = [
            trade
            for trade in filtered_trades_temp
            if not any(
                [
                    strategy in trade.get("text", "")
                    for strategy in exclude_strategies
                ]
            )
        ]

    if use_net_buy_or_sell == "net":
        filtered_trades_side = filtered_trades.copy()
    else:
        filtered_trades_side = [
            trade
            for trade in filtered_trades
            if trade.get("orderSide", "").lower() == use_net_buy_or_sell
        ]

    trades_df = pd.DataFrame(filtered_trades_side)

    if trades_df.empty:
        id_net_traded_final = (
            generate_empty_df_with_start_stop_time_lb_utc_variable_id_and_variable_value(
                start_time_lb_utc=start_time_lb_utc,
                stop_time_lb_utc=stop_time_lb_utc + timedelta(hours=1),
                resolution_seconds=resolution_seconds,
                variable_ids=BALANCE_PRICE_AREAS,
                variable_value_data_type="int",
            )
            .drop(columns=["resolution_seconds"])
            .rename(columns={"variable_value": "id_net_traded"})
        )
        id_net_traded_final["product"] = id_net_traded_final.apply(
            lambda row: nordpool_product_namer(
                row["start_time_lb_utc"], resolution_seconds
            ),
            axis=1,
        )
        return id_net_traded_final

    trades_df = trades_df[
        ["deliveryStart", "deliveryEnd", "orderSide", "quantity", "deliveryAreaId"]
    ].rename(
        columns={
            "deliveryStart": "start_time_lb_utc",
            "deliveryEnd": "stop_time_lb_utc",
        }
    )
    trades_df["start_time_lb_utc"] = pd.to_datetime(
        trades_df["start_time_lb_utc"], utc=True
    )
    trades_df["stop_time_lb_utc"] = pd.to_datetime(
        trades_df["stop_time_lb_utc"], utc=True
    )

    trades_df["signed_quantity"] = trades_df.apply(
        lambda row: (
            row["quantity"]
            if str(row["orderSide"]).lower() == "buy"
            else -row["quantity"]
        ),
        axis=1,
    )

    trades_df["resolution_seconds"] = (
        (trades_df["stop_time_lb_utc"] - trades_df["start_time_lb_utc"])
        .dt.total_seconds()
        .astype("int64")
    )

    # Aggregate
    net_trades_df = (
        trades_df.groupby(
            [
                "start_time_lb_utc",
                "stop_time_lb_utc",
                "resolution_seconds",
                "deliveryAreaId",
            ],
            as_index=False,
        )["signed_quantity"]
        .sum()
        .astype("int64")
    )

    net_trades_df["start_time_lb_utc"] = pd.to_datetime(
        net_trades_df["start_time_lb_utc"], utc=True
    )
    net_trades_df["stop_time_lb_utc"] = pd.to_datetime(
        net_trades_df["stop_time_lb_utc"], utc=True
    )
    net_trades_df["deliveryAreaId"] = net_trades_df["deliveryAreaId"].map(
        DELIVERY_AREA_ID_MAP
    )
    net_trades_df = net_trades_df.rename(
        columns={"signed_quantity": "id_net_traded", "deliveryAreaId": "price_area"}
    )

    id_net_traded_final = generate_empty_df_with_start_stop_time_lb_utc_variable_id_and_variable_value(
        start_time_lb_utc=start_time_lb_utc,
        stop_time_lb_utc=stop_time_lb_utc + timedelta(hours=1),
        resolution_seconds=resolution_seconds,
        variable_ids=BALANCE_PRICE_AREAS,
        variable_value_data_type="int",
    )
    id_net_traded_final = id_net_traded_final.rename(
        columns={"variable_value": "id_net_traded", "variable_id": "price_area"}
    )
    id_net_traded_hourly_df = (
        net_trades_df[net_trades_df["resolution_seconds"] == 3600]
        .copy(deep=True)
        .sort_values(by="start_time_lb_utc")
    )
    for _, row in id_net_traded_hourly_df.iterrows():
        if resolution_seconds == 3600 or (
            resolution_seconds == 900 and spread_hourly_trades_quarters
        ):
            for i in range(3600 // resolution_seconds):
                resampled_start_time_lb_utc = row[
                    "start_time_lb_utc"
                ] + pd.Timedelta(seconds=i * resolution_seconds)
                id_net_traded_final.loc[
                    (
                        id_net_traded_final["start_time_lb_utc"]
                        == resampled_start_time_lb_utc
                    )
                    & (
                        id_net_traded_final["stop_time_lb_utc"]
                        == resampled_start_time_lb_utc
                        + pd.Timedelta(seconds=resolution_seconds)
                    )
                    & (id_net_traded_final["price_area"] == row["price_area"]),
                    "id_net_traded",
                ] += row["id_net_traded"]

    if resolution_seconds == 900:  # Use hourly values also for every quarter
        id_net_traded_15min_df = net_trades_df[
            net_trades_df["resolution_seconds"] == 900
        ].copy(deep=True)
        for _, row in id_net_traded_15min_df.iterrows():
            id_net_traded_final.loc[
                (
                    id_net_traded_final["start_time_lb_utc"]
                    == row["start_time_lb_utc"]
                )
                & (
                    id_net_traded_final["stop_time_lb_utc"]
                    == row["stop_time_lb_utc"]
                )
                & (id_net_traded_final["price_area"] == row["price_area"]),
                "id_net_traded",
            ] += row["id_net_traded"]

    id_net_traded_final = id_net_traded_final.drop(columns=["resolution_seconds"])
    id_net_traded_final = id_net_traded_final.rename(
        columns={"price_area": "variable_id"}
    )
    id_net_traded_final["product"] = id_net_traded_final.apply(
        lambda row: nordpool_product_namer(
            row["start_time_lb_utc"], resolution_seconds
        ),
        axis=1,
    )
    return id_net_traded_final

load_best_bid_and_best_offer(start_time_lb_utc, stop_time_lb_utc, price_areas=[], resolutions_seconds=[])

Computes the best bid and best offer on the intraday market based on active public orders within a specified time range.

Parameters:

Name	Type	Description	Default
start_time_lb_utc	datetime	The lower bound of the delivery start time in UTC.	required
stop_time_lb_utc	datetime	The upper bound of the delivery start time in UTC.	required
price_areas	List[str]	List of price areas to be included. If left empty, all areas present in CID will be returned.	[]
resolutions_seconds	List[int]	List of resolutions to be included. If left empty, all areas present in CID will be returned.	[]

Returns:

Type

Description

DataFrame

pd.DataFrame: A DataFrame containing aggregated order book data with columns start_time_lb_utc (datetime64[ns, UTC]): Contract delivery start time. stop_time_lb_utc (datetime64[ns, UTC]): Contract delivery end time. price_area (str): Area code for pricing. contract_id (str): Unique contract identifier. product (str): Contract product name. product_type (str): Type of product. resolution_seconds (Int64): Resolution of the contract in seconds. best_bid_ct (Int64): Highest bid price. best_offer_ct (Int64): Lowest offer price.

Raises:

Type	Description
ValueError	If start_time_lb_utc or stop_time_lb_utc is not in UTC format.
ValueError	If start_time_lb_utc is later than stop_time_lb_utc.

Example

from physical_operations_utils.cid2_utils import CID2API
from zoneinfo import ZoneInfo

api = CID2Api()
start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

print(api.load_best_bid_and_best_offer(start_time_lb_utc, stop_time_lb_utc))

Source code in physical_operations_utils/cid2_utils.py

def load_best_bid_and_best_offer(
    self,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
    price_areas: List[str] = [],
    resolutions_seconds: List[int] = [],
) -> pd.DataFrame:
    """
    Computes the best bid and best offer on the intraday market based on active public orders within a specified time range.

    Args:
        start_time_lb_utc (datetime): The lower bound of the delivery start time in UTC.
        stop_time_lb_utc (datetime): The upper bound of the delivery start time in UTC.
        price_areas (List[str]): List of price areas to be included. If left empty, all areas present in CID will be returned.
        resolutions_seconds (List[int]): List of resolutions to be included. If left empty, all areas present in CID will be returned.

    Returns:
        pd.DataFrame: A DataFrame containing aggregated order book data with columns
            `start_time_lb_utc` (datetime64[ns, UTC]): Contract delivery start time.
            `stop_time_lb_utc` (datetime64[ns, UTC]): Contract delivery end time.
            `price_area` (str): Area code for pricing.
            `contract_id` (str): Unique contract identifier.
            `product` (str): Contract product name.
            `product_type` (str): Type of product.
            `resolution_seconds` (Int64): Resolution of the contract in seconds.
            `best_bid_ct` (Int64): Highest bid price.
            `best_offer_ct` (Int64): Lowest offer price.

    Raises:
        ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
        ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.

    Example:
        ```python
        from physical_operations_utils.cid2_utils import CID2API
        from zoneinfo import ZoneInfo

        api = CID2Api()
        start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
        stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

        print(api.load_best_bid_and_best_offer(start_time_lb_utc, stop_time_lb_utc))
        ```
    """
    validate_datetime_in_utc(start_time_lb_utc)
    validate_datetime_in_utc(stop_time_lb_utc)
    validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)
    expected_api_response_keys = [
        "deliveryStart",
        "deliveryEnd",
        "deliveryArea",
        "contractId",
        "product",
        "productType",
        "bestBid",
        "bestOffer",
        "resolutionSeconds",
    ]
    api_response = self._get_best_bid_best_offer_per_time_interval(
        start_time_lb_utc=start_time_lb_utc,
        stop_time_rb_utc=stop_time_lb_utc,
        price_areas=price_areas,
        resolutions_seconds=resolutions_seconds,
    )
    best_bid_best_offer_and_offered_qty_df = pd.DataFrame(api_response)
    if best_bid_best_offer_and_offered_qty_df.empty:
        best_bid_best_offer_and_offered_qty_df = pd.DataFrame(
            columns=expected_api_response_keys
        )
    best_bid_best_offer_and_offered_qty_df = (
        best_bid_best_offer_and_offered_qty_df.rename(
            columns={
                "deliveryArea": "price_area",
                "contractId": "contract_id",
                "productType": "product_type",
            }
        )
    )
    best_bid_best_offer_and_offered_qty_df.loc[:, "start_time_lb_utc"] = (
        pd.to_datetime(best_bid_best_offer_and_offered_qty_df["deliveryStart"])
    )
    best_bid_best_offer_and_offered_qty_df.loc[:, "stop_time_lb_utc"] = (
        pd.to_datetime(best_bid_best_offer_and_offered_qty_df["deliveryEnd"])
    )
    best_bid_best_offer_and_offered_qty_df.loc[:, "best_bid_ct"] = (
        best_bid_best_offer_and_offered_qty_df.loc[:, "bestBid"].astype("Int64")
    )
    best_bid_best_offer_and_offered_qty_df.loc[:, "best_offer_ct"] = (
        best_bid_best_offer_and_offered_qty_df.loc[:, "bestOffer"].astype("Int64")
    )
    best_bid_best_offer_and_offered_qty_df.loc[:, "resolution_seconds"] = (
        best_bid_best_offer_and_offered_qty_df.loc[:, "resolutionSeconds"].astype(
            "Int64"
        )
    )
    col_order = [
        "start_time_lb_utc",
        "stop_time_lb_utc",
        "price_area",
        "contract_id",
        "product",
        "product_type",
        "resolution_seconds",
        "best_offer_ct",
        "best_bid_ct",
    ]
    return best_bid_best_offer_and_offered_qty_df.loc[:, col_order]

load_id_capacities_per_area(start_time_lb_utc, stop_time_lb_utc)

Load ID capacities per area within a specified UTC time window.

This function retrieves capacity records from the NordpoolIntradayFeed database for all price areas that have capacities between the provided lower and upper time bounds. It validates that the provided datetime values are in UTC and that the start time precedes the stop time before constructing and executing a SQL query.

Parameters:

Name	Type	Description	Default
start_time_lb_utc	datetime	The lower bound of the delivery start time in UTC.	required
stop_time_lb_utc	datetime	The upper bound of the delivery start time in UTC.	required

Returns:

Type

Description

DataFrame

pandas.DataFrame: A DataFrame containing the following columns: start_time_lb_utc: Delivery start times as UTC datetime objects. stop_time_lb_utc: Delivery end times as UTC datetime objects. price_area_from: Price area code corresponding to the delivery's origin. price_area_to: Price area code corresponding to the delivery's destination. in_capacity: Inbound capacity. out_capacity: Outbound capacity.

Raises:

Type	Description
ValueError	If start_time_lb_utc or stop_time_lb_utc is not in UTC format.
ValueError	If start_time_lb_utc is later than stop_time_lb_utc.
SQLAlchemyError	If there is an issue executing the database query.

Example

from physical_operations_utils.cid2_utils import CID2API
from zoneinfo import ZoneInfo

api = CID2Api()
start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

print(api.load_id_capacities_per_area(start_time_lb_utc, stop_time_lb_utc))

Source code in physical_operations_utils/cid2_utils.py

def load_id_capacities_per_area(
    self,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
) -> pd.DataFrame:
    """
    Load ID capacities per area within a specified UTC time window.

    This function retrieves capacity records from the NordpoolIntradayFeed database for all price areas
    that have capacities between the provided lower and upper time bounds.
    It validates that the provided datetime values are in UTC and that the start time precedes
    the stop time before constructing and executing a SQL query.

    Args:
        start_time_lb_utc: The lower bound of the delivery start time in UTC.
        stop_time_lb_utc: The upper bound of the delivery start time in UTC.

    Returns:
        pandas.DataFrame:
            A DataFrame containing the following columns:
                start_time_lb_utc: Delivery start times as UTC datetime objects.
                stop_time_lb_utc: Delivery end times as UTC datetime objects.
                price_area_from: Price area code corresponding to the delivery's origin.
                price_area_to: Price area code corresponding to the delivery's destination.
                in_capacity: Inbound capacity.
                out_capacity: Outbound capacity.

    Raises:
        ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
        ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.
        sqlalchemy.exc.SQLAlchemyError: If there is an issue executing the database query.

    Example:
        ```python
        from physical_operations_utils.cid2_utils import CID2API
        from zoneinfo import ZoneInfo

        api = CID2Api()
        start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
        stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

        print(api.load_id_capacities_per_area(start_time_lb_utc, stop_time_lb_utc))
        ```
    """
    validate_datetime_in_utc(start_time_lb_utc)
    validate_datetime_in_utc(stop_time_lb_utc)
    validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)

    capacities_list = self._get_id_capacities_by_time(
        start_time_lb_utc=start_time_lb_utc,
        stop_time_rb_uct=stop_time_lb_utc + timedelta(hours=1),
    )
    df = pd.DataFrame(capacities_list)
    df["start_time_lb_utc"] = pd.to_datetime(df["deliveryStart"], utc=True)
    df["stop_time_lb_utc"] = pd.to_datetime(df["deliveryEnd"], utc=True)
    df["price_area_from"] = df["deliveryAreaFrom"].astype(str)
    df["price_area_to"] = df["deliveryAreaTo"].astype(str)
    df["in_capacity"] = df["inCapacity"].astype("int64")
    df["out_capacity"] = df["outCapacity"].astype("int64")
    df["resolution_seconds"] = (
        (df["stop_time_lb_utc"] - df["start_time_lb_utc"])
        .dt.total_seconds()
        .astype("int64")
    )
    return df.loc[
        (df["start_time_lb_utc"] >= start_time_lb_utc)
        & (df["stop_time_lb_utc"] <= stop_time_lb_utc),
        [
            "start_time_lb_utc",
            "stop_time_lb_utc",
            "price_area_from",
            "price_area_to",
            "in_capacity",
            "out_capacity",
            "resolution_seconds",
        ],
    ]

load_id_orderbook_with_contract_information(start_time_lb_utc, stop_time_lb_utc)

Retrieves intraday orders with associated contract information from the CustomerIntraday database.

Parameters:

Name	Type	Description	Default
start_time_lb_utc	datetime	The lower bound of the delivery start time in UTC.	required
stop_time_lb_utc	datetime	The upper bound of the delivery start time in UTC.	required

Returns:

Type

Description

DataFrame

pd.DataFrame: A DataFrame containing order book data with columns: start_time_lb_utc (datetime64[ns, UTC]): Contract delivery start time. stop_time_lb_utc (datetime64[ns, UTC]): Contract delivery end time. price_area (str): Price area. price (int64): Order price in cent. qty (int64): Order quantity in kW. direction (str): Order direction (i.e., BUY/SELL). contract_id (str): Unique contract identifier. product (str): Contract product name. product_type (str): Type of product. resolution_seconds (int64): Resolution of the contract in seconds.

Raises:

Type	Description
ValueError	If start_time_lb_utc or stop_time_lb_utc is not in UTC format.
ValueError	If start_time_lb_utc is later than stop_time_lb_utc.

Example

from physical_operations_utils.cid2_utils import CID2API
from zoneinfo import ZoneInfo

api = CID2Api()
start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

print(api.load_id_orderbook_with_contract_information(start_time_lb_utc, stop_time_lb_utc))

Source code in physical_operations_utils/cid2_utils.py

def load_id_orderbook_with_contract_information(
    self,
    start_time_lb_utc: datetime,
    stop_time_lb_utc: datetime,
) -> pd.DataFrame:
    """
    Retrieves intraday orders with associated contract information from the CustomerIntraday database.

    Args:
        start_time_lb_utc: The lower bound of the delivery start time in UTC.
        stop_time_lb_utc: The upper bound of the delivery start time in UTC.

    Returns:
        pd.DataFrame: A DataFrame containing order book data with columns:
            `start_time_lb_utc` (datetime64[ns, UTC]): Contract delivery start time.
            `stop_time_lb_utc` (datetime64[ns, UTC]): Contract delivery end time.
            `price_area` (str): Price area.
            `price` (int64): Order price in cent.
            `qty` (int64): Order quantity in kW.
            `direction` (str): Order direction (i.e., BUY/SELL).
            `contract_id` (str): Unique contract identifier.
            `product` (str): Contract product name.
            `product_type` (str): Type of product.
            `resolution_seconds` (int64): Resolution of the contract in seconds.

    Raises:
        ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
        ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.

    Example:
        ```python
        from physical_operations_utils.cid2_utils import CID2API
        from zoneinfo import ZoneInfo

        api = CID2Api()
        start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
        stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

        print(api.load_id_orderbook_with_contract_information(start_time_lb_utc, stop_time_lb_utc))
        ```
    """
    validate_datetime_in_utc(start_time_lb_utc)
    validate_datetime_in_utc(stop_time_lb_utc)
    validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)

    expected_api_response_keys = [
        "deliveryStart",
        "deliveryEnd",
        "deliveryAreaId",
        "price",
        "qty",
        "orderSide",
        "contractId",
        "contractName",
        "productType",
        "resolutionSeconds",
    ]

    orderbook = self._get_id_public_orders_per_time_interval(
        start_time_lb_utc=start_time_lb_utc, stop_time_rb_utc=stop_time_lb_utc
    )

    orderbook_df = pd.DataFrame(orderbook)

    if orderbook_df.empty:
        orderbook_df = pd.DataFrame(columns=expected_api_response_keys)

    orderbook_df["start_time_lb_utc"] = pd.to_datetime(
        orderbook_df["deliveryStart"], utc=True
    )
    orderbook_df["stop_time_lb_utc"] = pd.to_datetime(
        orderbook_df["deliveryEnd"], utc=True
    )
    orderbook_df["price_area"] = (
        orderbook_df["deliveryAreaId"].map(DELIVERY_AREA_ID_MAP).astype(str)
    )
    orderbook_df["price"] = orderbook_df["price"].astype("int64")
    orderbook_df["qty"] = orderbook_df["qty"].astype("int64")
    orderbook_df["direction"] = orderbook_df["orderSide"].str.upper().astype(str)
    orderbook_df["contract_id"] = orderbook_df["contractId"].astype(str)
    orderbook_df["product"] = orderbook_df["contractName"].astype(str)
    orderbook_df["product_type"] = orderbook_df["productType"].astype(str)
    orderbook_df["resolution_seconds"] = orderbook_df["resolutionSeconds"].astype(
        "int64"
    )
    return orderbook_df.loc[
        :,
        [
            "start_time_lb_utc",
            "stop_time_lb_utc",
            "price_area",
            "price",
            "qty",
            "direction",
            "contract_id",
            "product",
            "product_type",
            "resolution_seconds",
        ],
    ]

load_id_public_statistics_with_contract_information(start_time_lb_utc, stop_time_lb_utc)

Retrieves intraday public statistics with associated contract information from the CID2 intraday feed. Returns only information for products with resolution 900s, 1800s and 3600s.

Parameters:

Name	Type	Description	Default
start_time_lb_utc	datetime	The lower bound of the delivery start time in UTC.	required
stop_time_lb_utc	datetime	The upper bound of the delivery start time in UTC.	required

Returns:

Type

Description

DataFrame

pd.DataFrame: A DataFrame containing public statistics data with columns: start_time_lb_utc (datetime64[ns, UTC]): Contract delivery start time. stop_time_lb_utc (datetime64[ns, UTC]): Contract delivery end time. price_area (str): Price area. last_price (int64): Last traded price in ct. last_quantity (int64): Last traded quantity in kW. last_trade_time (datetime64[ns, UTC]): Time of last trade. If no trades were made yet it is set to datetime(1, 1, 1, tzinfo=ZoneInfo("UTC")). highest_price (int64): Highest trade price in ct. vwap (int64): Volume-weighted average price in ct. turnover (int64): Turnover in kW. spot_price_eur_ct (int64): Day-ahead market price in euro cents. tendency (str): Market tendency (e.g., UP, DOWN, EQUAL). contract_id (str): Unique contract identifier. product (str): Contract product name. product_type (str): Type of product. resolution_seconds (int64): Resolution of the contract in seconds.

Raises:

Type	Description
ValueError	If start_time_lb_utc or stop_time_lb_utc is not in UTC format.
ValueError	If start_time_lb_utc is later than stop_time_lb_utc.

Example

from physical_operations_utils.cid2_utils import CID2API
from zoneinfo import ZoneInfo

api = CID2Api()
start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

print(api.load_id_public_statistics_with_contract_information(start_time_lb_utc, stop_time_lb_utc))

Source code in physical_operations_utils/cid2_utils.py

def load_id_public_statistics_with_contract_information(
    self, start_time_lb_utc: datetime, stop_time_lb_utc: datetime
) -> pd.DataFrame:
    """
    Retrieves intraday public statistics with associated contract information from the CID2 intraday feed. Returns only information for
    products with resolution 900s, 1800s and 3600s.

    Args:
        start_time_lb_utc: The lower bound of the delivery start time in UTC.
        stop_time_lb_utc: The upper bound of the delivery start time in UTC.

    Returns:
        pd.DataFrame: A DataFrame containing public statistics data with columns:
            `start_time_lb_utc` (datetime64[ns, UTC]): Contract delivery start time.
            `stop_time_lb_utc` (datetime64[ns, UTC]): Contract delivery end time.
            `price_area` (str): Price area.
            `last_price` (int64): Last traded price in ct.
            `last_quantity` (int64): Last traded quantity in kW.
            `last_trade_time` (datetime64[ns, UTC]): Time of last trade. If no trades were made yet it is set to datetime(1, 1, 1, tzinfo=ZoneInfo("UTC")).
            `highest_price` (int64): Highest trade price in ct.
            `vwap` (int64): Volume-weighted average price in ct.
            `turnover` (int64): Turnover in kW.
            `spot_price_eur_ct` (int64): Day-ahead market price in euro cents.
            `tendency` (str): Market tendency (e.g., UP, DOWN, EQUAL).
            `contract_id` (str): Unique contract identifier.
            `product` (str): Contract product name.
            `product_type` (str): Type of product.
            `resolution_seconds` (int64): Resolution of the contract in seconds.

    Raises:
        ValueError: If `start_time_lb_utc` or `stop_time_lb_utc` is not in UTC format.
        ValueError: If `start_time_lb_utc` is later than `stop_time_lb_utc`.

    Example:
        ```python
        from physical_operations_utils.cid2_utils import CID2API
        from zoneinfo import ZoneInfo

        api = CID2Api()
        start_time_lb_utc = datetime(2025, 8, 7, 15, tzinfo=ZoneInfo("UTC"))
        stop_time_lb_utc = datetime(2025, 8, 7, 18, tzinfo=ZoneInfo("UTC"))

        print(api.load_id_public_statistics_with_contract_information(start_time_lb_utc, stop_time_lb_utc))
        ```
    """
    validate_datetime_in_utc(start_time_lb_utc)
    validate_datetime_in_utc(stop_time_lb_utc)
    validate_start_time_before_end_time(start_time_lb_utc, stop_time_lb_utc)
    columns = [
        "start_time_lb_utc",
        "stop_time_lb_utc",
        "price_area",
        "last_price",
        "last_quantity",
        "last_trade_time",
        "highest_price",
        "vwap",
        "turnover",
        "spot_price_eur_ct",
        "tendency",
        "contract_id",
        "product",
        "product_type",
        "resolution_seconds",
    ]
    response = self._get_id_public_statistics_per_time_interval(
        start_time_lb_utc=start_time_lb_utc,
        stop_time_rb_utc=stop_time_lb_utc + timedelta(hours=1),
    )
    df = pd.DataFrame(response)
    if df.empty:
        return pd.DataFrame(columns=columns)
    df["start_time_lb_utc"] = pd.to_datetime(df["deliveryStart"], utc=True)
    df["stop_time_lb_utc"] = pd.to_datetime(df["deliveryEnd"], utc=True)
    df["price_area"] = df["deliveryArea"].astype(str)
    df["last_price"] = df["lastPrice"].astype("float64").round(0)
    df["last_quantity"] = df["lastQuantity"].astype("float64").round(0)
    df["last_trade_time"] = pd.to_datetime(
        df["lastTradeTime"], utc=True, errors="coerce"
    )
    df["highest_price"] = df["highestPrice"].astype("float64").round(0)
    df["vwap"] = df["vwap"].astype("float64").round(0)
    df["turnover"] = df["turnover"].astype("float64").round(0)
    df["spot_price_eur_ct"] = df["dayAheadPrice"].astype("float64").round(0)
    df["tendency"] = df["tendency"].astype(str)
    df["contract_id"] = df["contractId"].astype(str)
    df["resolution_seconds"] = (
        (df["stop_time_lb_utc"] - df["start_time_lb_utc"])
        .dt.total_seconds()
        .round(0)
        .astype("int64")
    )
    df = filter_dataframe_by_resolution_seconds(
        df=df, keep_resolutions=[900, 1800, 3600]
    )
    if not df.empty:
        df["product"] = df.apply(
            lambda row: nordpool_product_namer(
                start_time_lb_utc=row["start_time_lb_utc"],
                resolution_seconds=row["resolution_seconds"],
            ),
            axis=1,
        )
        df["product_type"] = df.apply(
            lambda row: NORDPOOL_PRODUCT_TYPE_MAP.get(
                row["resolution_seconds"], "NA"
            ),
            axis=1,
        )
    else:
        df["product"] = "NA"
        df["product_type"] = "NA"
    return df.loc[:, columns]