sda_api

This is a wrapper to communicate with the EMS SDA API.

SdaApi

Client for the Volue SDA‑EMS GraphQL API.

The client provides high‑level methods for:

• Discovering available bid addresses
• Fetching block bid packages
• Fetching curve bid packages

Authentication is handled via OAuth2 client‑credentials. Credentials are loaded from the environment key configuration and resolved via Azure Key Vault.

Attributes:

Name	Type	Description
env		Current execution environment (e.g. "nonprod", "prod").
base_url		GraphQL endpoint URL selected based on environment.
client_id		OAuth2 client ID.
client_secret		OAuth2 client secret.
session		Persistent HTTP session.

Example

        from physical_operations_utils.ems_utils.SdaApi import SdaApi

        api = SdaApi()

Source code in physical_operations_utils/ems_utils/SdaApi.py

class SdaApi:
    """
    Client for the Volue SDA‑EMS GraphQL API.

    The client provides high‑level methods for:

    - Discovering available bid addresses
    - Fetching block bid packages
    - Fetching curve bid packages

    Authentication is handled via OAuth2 client‑credentials.
    Credentials are loaded from the environment key configuration
    and resolved via Azure Key Vault.

    Attributes:
        env: Current execution environment (e.g. ``"nonprod"``, ``"prod"``).
        base_url: GraphQL endpoint URL selected based on environment.
        client_id: OAuth2 client ID.
        client_secret: OAuth2 client secret.
        session: Persistent HTTP session.

    Example:
        ```python
                from physical_operations_utils.ems_utils.SdaApi import SdaApi

                api = SdaApi()
        ```
    """

    def __init__(self, verify_ssl: bool = True, timeout_s: float = 30.0):
        """
        Initialize an SDA‑EMS API client.

        During initialization the client:

        - Resolves the current environment
        - Loads OAuth2 credentials from Key Vault
        - Selects the correct GraphQL endpoint
        - Creates a persistent HTTP session

        Args:
            verify_ssl: Whether SSL certificates are verified.
            timeout_s: HTTP request timeout in seconds.
        """
        self.verify_ssl = verify_ssl
        self.timeout_s = timeout_s
        self.env = setup_environment()

        sda_cfg = get_keys_yaml_file().get("volue_sda_api", {})
        self.client_id = get_secret(sda_cfg["client_id"])
        self.client_secret = get_secret(sda_cfg["client_secret"])

        self.base_url = (
            BASE_URL_TEST if self.env.lower() == "nonprod" else BASE_URL_PROD
        )

        self.session = requests.Session()
        self.session.headers.update(
            {"Accept": "application/json", "Content-Type": "application/json"}
        )

        self._access_token: Optional[str] = None
        self._access_token_expires_at: float = 0.0

    def _get_access_token(self) -> str:
        now = time.time()
        if self._access_token and now < (self._access_token_expires_at - 60):
            return self._access_token

        r = self.session.post(
            TOKEN_URL,
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret,
                "scope": SCOPE,
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=self.timeout_s,
            verify=self.verify_ssl,
        )

        try:
            r.raise_for_status()
        except requests.HTTPError:
            logger.warning(
                "Volue SDA token request failed | url=%s | status=%s | response=%s",
                r.url,
                r.status_code,
                r.text[:500],
            )
            raise

        payload = r.json()
        token = payload.get("access_token")
        expires_in = float(payload.get("expires_in", 3600))

        if not token:
            raise VolueSdaApiError(f"Token response missing access_token: {payload}")

        self._access_token = token
        self._access_token_expires_at = now + expires_in
        return token

    def _post_graphql(self, query: str) -> Dict[str, Any]:
        token = self._get_access_token()
        r = self.session.post(
            self.base_url,
            json={"query": query, "variables": {}},
            headers={"Authorization": f"Bearer {token}"},
            timeout=self.timeout_s,
            verify=self.verify_ssl,
        )

        # retry once on 401
        if r.status_code == 401:
            self._access_token = None
            token = self._get_access_token()
            r = self.session.post(
                self.base_url,
                json={"query": query, "variables": {}},
                headers={"Authorization": f"Bearer {token}"},
                timeout=self.timeout_s,
                verify=self.verify_ssl,
            )

        try:
            r.raise_for_status()
        except requests.HTTPError:
            logger.warning(
                "Volue SDA GraphQL HTTP failed | url=%s | status=%s | response=%s",
                r.url,
                r.status_code,
                r.text[:800],
            )
            raise

        payload = r.json()

        if payload.get("errors"):
            logger.warning(
                "Volue SDA GraphQL errors | errors=%s", str(payload["errors"])[:2000]
            )
            raise VolueSdaApiError(f"GraphQL errors: {payload['errors']}")

        data = payload.get("data")
        if data is None:
            raise VolueSdaApiError(f"GraphQL response missing data: {payload}")

        return data

    def _trading(self, query: str) -> Dict[str, Any]:
        data = self._post_graphql(query)
        trading = data.get("trading")
        if trading is None:
            raise VolueSdaApiError(f"Missing 'trading' in response: {data}")
        return trading

    # Queries

    def bids_addresses(self, *, data_group: str) -> List[str]:
        """
        List available SDA bid addresses for a given data group.

        This method returns all external matrix or bid addresses that
        exist within the specified SDA data group.

        Args:
            data_group: SDA data group identifier
                (e.g. ``"DA_MATRICES_OUT"``).

        Returns:
            List[str]: List of available SDA bid addresses.

        Example:
        ```python
                api = SdaApi()

                addresses = api.bids_addresses(
                    data_group="DA_MATRICES_OUT"
                )
        ```
        """
        q = f"""
        query {{
          trading {{
            bidsAddresses(dataGroup: {_gql(data_group)})
          }}
        }}
        """
        return self._trading(q)["bidsAddresses"]

    def block_bids_package(
        self,
        *,
        address: str,
        data_group: str,
        delivery_date_utc: Optional[str] = None,
        from_delivery_date_utc: Optional[str] = None,
        to_delivery_date_utc: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Fetch block bid packages for a given SDA address.

        Exactly one of the following must be provided:
        - ``delivery_date_utc``
        - ``from_delivery_date_utc`` and ``to_delivery_date_utc``

        Args:
            address: SDA bid address.
            data_group: SDA data group.
            delivery_date_utc: Single delivery date (UTC, ISO‑8601).
            from_delivery_date_utc: Start of delivery date range (UTC).
            to_delivery_date_utc: End of delivery date range (UTC).

        Returns:
            Dict[str, Any]: Raw block bid package response.

        Raises:
            ValueError: If no valid delivery date filter is provided.
            VolueSdaApiError: If the API returns an error.

        Example:
        ```python
                api = SdaApi()

                result = api.block_bids_package(
                    address="BORLÄNGE_CHP_MATRIX",
                    data_group="DA_MATRICES_OUT",
                    delivery_date_utc="2025-10-01T22:00Z",
                )

                print(result["success"])
                print(result["bids"])
        ```
        """
        criteria: Dict[str, Any] = {"address": address, "dataGroup": data_group}

        if delivery_date_utc:
            criteria["getBidByDeliveryDate"] = {"deliveryDate": delivery_date_utc}
        elif from_delivery_date_utc and to_delivery_date_utc:
            criteria["getBidsByRange"] = {
                "fromDeliveryDate": from_delivery_date_utc,
                "toDeliveryDate": to_delivery_date_utc,
            }
        else:
            raise ValueError(
                "Provide delivery_date_utc OR (from_delivery_date_utc and to_delivery_date_utc)"
            )

        q = f"""
        query {{
          trading {{
            blockBidsPackage(criteria: {_gql(criteria)}) {{
              success
              errors {{ errorCode message }}
              bids {{
                dataGroup
                address
                deliveryDate
                periodLength
                type
                currency
                bids {{
                  reference
                  linkedTo
                  deliveryStartTimeUtc
                  periodCount
                  price
                  volume
                }}
              }}
            }}
          }}
        }}
        """
        return self._trading(q)["blockBidsPackage"]

    def curve_bids_package(
        self,
        *,
        address: str,
        data_group: str,
        delivery_date_utc: Optional[str] = None,
        from_delivery_date_utc: Optional[str] = None,
        to_delivery_date_utc: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Fetch curve bid packages for a given SDA address.

        Exactly one of the following must be provided:
        - ``delivery_date_utc``
        - ``from_delivery_date_utc`` and ``to_delivery_date_utc``

        Args:
            address: SDA bid address.
            data_group: SDA data group.
            delivery_date_utc: Single delivery date (UTC, ISO‑8601).
            from_delivery_date_utc: Start of delivery date range (UTC).
            to_delivery_date_utc: End of delivery date range (UTC).

        Returns:
            Dict[str, Any]: Raw curve bid package response.

        Raises:
            ValueError: If no valid delivery date filter is provided.
            VolueSdaApiError: If the API returns an error.

        Notes:
            The SDA backend may return ``Unexpected Execution Error``
            when no curve bids exist for the given criteria. This should
            be treated as “no data”.

        Example:
            ```python
                api = SdaApi()

                try:
                    result = api.curve_bids_package(
                        address="BORLÄNGE_CHP_MATRIX",
                        data_group="DA_MATRICES_OUT",
                        from_delivery_date_utc="2025-10-01T22:00Z",
                        to_delivery_date_utc="2025-10-02T22:00Z",
                    )
                except VolueSdaApiError:
                    result = {"bids": []}

                for bid in result.get("bids", []):
                    print(bid["deliveryDate"], bid["type"])
            ```
        """
        criteria: Dict[str, Any] = {"address": address, "dataGroup": data_group}

        if delivery_date_utc:
            criteria["getBidByDeliveryDate"] = {"deliveryDate": delivery_date_utc}
        elif from_delivery_date_utc and to_delivery_date_utc:
            criteria["getBidsByRange"] = {
                "fromDeliveryDate": from_delivery_date_utc,
                "toDeliveryDate": to_delivery_date_utc,
            }
        else:
            raise ValueError(
                "Provide delivery_date_utc OR (from_delivery_date_utc and to_delivery_date_utc)"
            )

        q = f"""
        query {{
          trading {{
            curveBidsPackage(criteria: {_gql(criteria)}) {{
              success
              errors {{ errorCode message }}
              bids {{
                address
                deliveryDate
                dataGroup
                type
                periodLength
                currency
                bids {{
                  price
                  deliveryStartTimeUtc
                  periodVolumes
                }}
              }}
            }}
          }}
        }}
        """
        return self._trading(q)["curveBidsPackage"]

__init__(verify_ssl=True, timeout_s=30.0)

Initialize an SDA‑EMS API client.

During initialization the client:

• Resolves the current environment
• Loads OAuth2 credentials from Key Vault
• Selects the correct GraphQL endpoint
• Creates a persistent HTTP session

Parameters:

Name	Type	Description	Default
verify_ssl	bool	Whether SSL certificates are verified.	True
timeout_s	float	HTTP request timeout in seconds.	30.0

Source code in physical_operations_utils/ems_utils/SdaApi.py

def __init__(self, verify_ssl: bool = True, timeout_s: float = 30.0):
    """
    Initialize an SDA‑EMS API client.

    During initialization the client:

    - Resolves the current environment
    - Loads OAuth2 credentials from Key Vault
    - Selects the correct GraphQL endpoint
    - Creates a persistent HTTP session

    Args:
        verify_ssl: Whether SSL certificates are verified.
        timeout_s: HTTP request timeout in seconds.
    """
    self.verify_ssl = verify_ssl
    self.timeout_s = timeout_s
    self.env = setup_environment()

    sda_cfg = get_keys_yaml_file().get("volue_sda_api", {})
    self.client_id = get_secret(sda_cfg["client_id"])
    self.client_secret = get_secret(sda_cfg["client_secret"])

    self.base_url = (
        BASE_URL_TEST if self.env.lower() == "nonprod" else BASE_URL_PROD
    )

    self.session = requests.Session()
    self.session.headers.update(
        {"Accept": "application/json", "Content-Type": "application/json"}
    )

    self._access_token: Optional[str] = None
    self._access_token_expires_at: float = 0.0

bids_addresses(*, data_group)

List available SDA bid addresses for a given data group.

This method returns all external matrix or bid addresses that exist within the specified SDA data group.

Parameters:

Name	Type	Description	Default
data_group	str	SDA data group identifier (e.g. "DA_MATRICES_OUT").	required

Returns:

Type	Description
List[str]	List[str]: List of available SDA bid addresses.

Example:

        api = SdaApi()

        addresses = api.bids_addresses(
            data_group="DA_MATRICES_OUT"
        )

Source code in physical_operations_utils/ems_utils/SdaApi.py

def bids_addresses(self, *, data_group: str) -> List[str]:
    """
    List available SDA bid addresses for a given data group.

    This method returns all external matrix or bid addresses that
    exist within the specified SDA data group.

    Args:
        data_group: SDA data group identifier
            (e.g. ``"DA_MATRICES_OUT"``).

    Returns:
        List[str]: List of available SDA bid addresses.

    Example:
    ```python
            api = SdaApi()

            addresses = api.bids_addresses(
                data_group="DA_MATRICES_OUT"
            )
    ```
    """
    q = f"""
    query {{
      trading {{
        bidsAddresses(dataGroup: {_gql(data_group)})
      }}
    }}
    """
    return self._trading(q)["bidsAddresses"]

block_bids_package(*, address, data_group, delivery_date_utc=None, from_delivery_date_utc=None, to_delivery_date_utc=None)

Fetch block bid packages for a given SDA address.

Exactly one of the following must be provided: - delivery_date_utc - from_delivery_date_utc and to_delivery_date_utc

Parameters:

Name	Type	Description	Default
address	str	SDA bid address.	required
data_group	str	SDA data group.	required
delivery_date_utc	Optional[str]	Single delivery date (UTC, ISO‑8601).	None
from_delivery_date_utc	Optional[str]	Start of delivery date range (UTC).	None
to_delivery_date_utc	Optional[str]	End of delivery date range (UTC).	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Raw block bid package response.

Raises:

Type	Description
ValueError	If no valid delivery date filter is provided.
VolueSdaApiError	If the API returns an error.

Example:

        api = SdaApi()

        result = api.block_bids_package(
            address="BORLÄNGE_CHP_MATRIX",
            data_group="DA_MATRICES_OUT",
            delivery_date_utc="2025-10-01T22:00Z",
        )

        print(result["success"])
        print(result["bids"])

Source code in physical_operations_utils/ems_utils/SdaApi.py

def block_bids_package(
    self,
    *,
    address: str,
    data_group: str,
    delivery_date_utc: Optional[str] = None,
    from_delivery_date_utc: Optional[str] = None,
    to_delivery_date_utc: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Fetch block bid packages for a given SDA address.

    Exactly one of the following must be provided:
    - ``delivery_date_utc``
    - ``from_delivery_date_utc`` and ``to_delivery_date_utc``

    Args:
        address: SDA bid address.
        data_group: SDA data group.
        delivery_date_utc: Single delivery date (UTC, ISO‑8601).
        from_delivery_date_utc: Start of delivery date range (UTC).
        to_delivery_date_utc: End of delivery date range (UTC).

    Returns:
        Dict[str, Any]: Raw block bid package response.

    Raises:
        ValueError: If no valid delivery date filter is provided.
        VolueSdaApiError: If the API returns an error.

    Example:
    ```python
            api = SdaApi()

            result = api.block_bids_package(
                address="BORLÄNGE_CHP_MATRIX",
                data_group="DA_MATRICES_OUT",
                delivery_date_utc="2025-10-01T22:00Z",
            )

            print(result["success"])
            print(result["bids"])
    ```
    """
    criteria: Dict[str, Any] = {"address": address, "dataGroup": data_group}

    if delivery_date_utc:
        criteria["getBidByDeliveryDate"] = {"deliveryDate": delivery_date_utc}
    elif from_delivery_date_utc and to_delivery_date_utc:
        criteria["getBidsByRange"] = {
            "fromDeliveryDate": from_delivery_date_utc,
            "toDeliveryDate": to_delivery_date_utc,
        }
    else:
        raise ValueError(
            "Provide delivery_date_utc OR (from_delivery_date_utc and to_delivery_date_utc)"
        )

    q = f"""
    query {{
      trading {{
        blockBidsPackage(criteria: {_gql(criteria)}) {{
          success
          errors {{ errorCode message }}
          bids {{
            dataGroup
            address
            deliveryDate
            periodLength
            type
            currency
            bids {{
              reference
              linkedTo
              deliveryStartTimeUtc
              periodCount
              price
              volume
            }}
          }}
        }}
      }}
    }}
    """
    return self._trading(q)["blockBidsPackage"]

curve_bids_package(*, address, data_group, delivery_date_utc=None, from_delivery_date_utc=None, to_delivery_date_utc=None)

Fetch curve bid packages for a given SDA address.

Exactly one of the following must be provided: - delivery_date_utc - from_delivery_date_utc and to_delivery_date_utc

Parameters:

Name	Type	Description	Default
address	str	SDA bid address.	required
data_group	str	SDA data group.	required
delivery_date_utc	Optional[str]	Single delivery date (UTC, ISO‑8601).	None
from_delivery_date_utc	Optional[str]	Start of delivery date range (UTC).	None
to_delivery_date_utc	Optional[str]	End of delivery date range (UTC).	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Raw curve bid package response.

Raises:

Type	Description
ValueError	If no valid delivery date filter is provided.
VolueSdaApiError	If the API returns an error.

Notes

The SDA backend may return Unexpected Execution Error when no curve bids exist for the given criteria. This should be treated as “no data”.

Example

    api = SdaApi()

    try:
        result = api.curve_bids_package(
            address="BORLÄNGE_CHP_MATRIX",
            data_group="DA_MATRICES_OUT",
            from_delivery_date_utc="2025-10-01T22:00Z",
            to_delivery_date_utc="2025-10-02T22:00Z",
        )
    except VolueSdaApiError:
        result = {"bids": []}

    for bid in result.get("bids", []):
        print(bid["deliveryDate"], bid["type"])

Source code in physical_operations_utils/ems_utils/SdaApi.py

def curve_bids_package(
    self,
    *,
    address: str,
    data_group: str,
    delivery_date_utc: Optional[str] = None,
    from_delivery_date_utc: Optional[str] = None,
    to_delivery_date_utc: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Fetch curve bid packages for a given SDA address.

    Exactly one of the following must be provided:
    - ``delivery_date_utc``
    - ``from_delivery_date_utc`` and ``to_delivery_date_utc``

    Args:
        address: SDA bid address.
        data_group: SDA data group.
        delivery_date_utc: Single delivery date (UTC, ISO‑8601).
        from_delivery_date_utc: Start of delivery date range (UTC).
        to_delivery_date_utc: End of delivery date range (UTC).

    Returns:
        Dict[str, Any]: Raw curve bid package response.

    Raises:
        ValueError: If no valid delivery date filter is provided.
        VolueSdaApiError: If the API returns an error.

    Notes:
        The SDA backend may return ``Unexpected Execution Error``
        when no curve bids exist for the given criteria. This should
        be treated as “no data”.

    Example:
        ```python
            api = SdaApi()

            try:
                result = api.curve_bids_package(
                    address="BORLÄNGE_CHP_MATRIX",
                    data_group="DA_MATRICES_OUT",
                    from_delivery_date_utc="2025-10-01T22:00Z",
                    to_delivery_date_utc="2025-10-02T22:00Z",
                )
            except VolueSdaApiError:
                result = {"bids": []}

            for bid in result.get("bids", []):
                print(bid["deliveryDate"], bid["type"])
        ```
    """
    criteria: Dict[str, Any] = {"address": address, "dataGroup": data_group}

    if delivery_date_utc:
        criteria["getBidByDeliveryDate"] = {"deliveryDate": delivery_date_utc}
    elif from_delivery_date_utc and to_delivery_date_utc:
        criteria["getBidsByRange"] = {
            "fromDeliveryDate": from_delivery_date_utc,
            "toDeliveryDate": to_delivery_date_utc,
        }
    else:
        raise ValueError(
            "Provide delivery_date_utc OR (from_delivery_date_utc and to_delivery_date_utc)"
        )

    q = f"""
    query {{
      trading {{
        curveBidsPackage(criteria: {_gql(criteria)}) {{
          success
          errors {{ errorCode message }}
          bids {{
            address
            deliveryDate
            dataGroup
            type
            periodLength
            currency
            bids {{
              price
              deliveryStartTimeUtc
              periodVolumes
            }}
          }}
        }}
      }}
    }}
    """
    return self._trading(q)["curveBidsPackage"]

VolueSdaApiError

Bases: RuntimeError

Raised for HTTP/GraphQL failures when calling the Volue SDA-EMS API.

Source code in physical_operations_utils/ems_utils/SdaApi.py

class VolueSdaApiError(RuntimeError):
    """Raised for HTTP/GraphQL failures when calling the Volue SDA-EMS API."""