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"]

    # Mutations

    def update_curve_bids_package(
        self,
        *,
        address: str,
        data_group: str,
        delivery_date: str,
        period_length: str,
        currency: str,
        bid_type: str = "PERIOD_MATRIX_DA",
        bids: List[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """
        Submit (create/update) curve bid packages via the SDA GraphQL mutation.

        Args:
            address: SDA bid address.
            data_group: SDA data group.
            delivery_date: Delivery date (ISO-8601 date, e.g. ``"2026-05-21"``).
            period_length: Period length enum name (e.g. ``"PT15M"``, ``"PT60M"``).
            currency: Currency enum name (e.g. ``"EUR"``).
            bid_type: Bid type enum name (default ``"PERIOD_MATRIX_DA"``).
            bids: List of bid dicts, each with ``price`` (float),
                ``deliveryStartTimeUtc`` (str), and ``periodVolumes`` (list of float).

        Returns:
            Dict[str, Any]: Raw mutation response containing ``value`` and ``errors``.

        Raises:
            VolueSdaApiError: If the API returns GraphQL errors.

        Example:
            ```python
                api = SdaApi()

                result = api.update_curve_bids_package(
                    address="DA_MLK_TEST_MATRIX",
                    data_group="MLK_DA_MATRIX",
                    delivery_date="2026-05-21",
                    period_length="PT15M",
                    currency="EUR",
                    bid_type="PERIOD_MATRIX_DA",
                    bids=[{
                        "price": 2.0,
                        "deliveryStartTimeUtc": "2026-05-20T23:00:00.000Z",
                        "periodVolumes": [-13.20, -13.20, ...],
                    }],
                )
            ```
        """
        input_package = {
            "address": address,
            "dataGroup": data_group,
            "deliveryDate": delivery_date,
            "periodLength": _GqlEnum(period_length),
            "currency": _GqlEnum(currency),
            "type": _GqlEnum(bid_type),
            "bids": bids,
        }

        q = f"""
        mutation {{
          trading {{
            updateCurveBidsPackage(input: {_gql([input_package])}) {{
              value {{
                address
                dataGroup
                success
              }}
              errors {{
                code: __typename
                ... on Error {{ message }}
                ... on RestRequestError {{ message }}
              }}
            }}
          }}
        }}
        """
        return self._trading(q)["updateCurveBidsPackage"]

__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"]

update_curve_bids_package(*, address, data_group, delivery_date, period_length, currency, bid_type='PERIOD_MATRIX_DA', bids)

Submit (create/update) curve bid packages via the SDA GraphQL mutation.

Parameters:

Name	Type	Description	Default
address	str	SDA bid address.	required
data_group	str	SDA data group.	required
delivery_date	str	Delivery date (ISO-8601 date, e.g. "2026-05-21").	required
period_length	str	Period length enum name (e.g. "PT15M", "PT60M").	required
currency	str	Currency enum name (e.g. "EUR").	required
bid_type	str	Bid type enum name (default "PERIOD_MATRIX_DA").	'PERIOD_MATRIX_DA'
bids	List[Dict[str, Any]]	List of bid dicts, each with price (float), deliveryStartTimeUtc (str), and periodVolumes (list of float).	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Raw mutation response containing value and errors.

Raises:

Type	Description
VolueSdaApiError	If the API returns GraphQL errors.

Example

    api = SdaApi()

    result = api.update_curve_bids_package(
        address="DA_MLK_TEST_MATRIX",
        data_group="MLK_DA_MATRIX",
        delivery_date="2026-05-21",
        period_length="PT15M",
        currency="EUR",
        bid_type="PERIOD_MATRIX_DA",
        bids=[{
            "price": 2.0,
            "deliveryStartTimeUtc": "2026-05-20T23:00:00.000Z",
            "periodVolumes": [-13.20, -13.20, ...],
        }],
    )

Source code in physical_operations_utils/ems_utils/SdaApi.py

def update_curve_bids_package(
    self,
    *,
    address: str,
    data_group: str,
    delivery_date: str,
    period_length: str,
    currency: str,
    bid_type: str = "PERIOD_MATRIX_DA",
    bids: List[Dict[str, Any]],
) -> Dict[str, Any]:
    """
    Submit (create/update) curve bid packages via the SDA GraphQL mutation.

    Args:
        address: SDA bid address.
        data_group: SDA data group.
        delivery_date: Delivery date (ISO-8601 date, e.g. ``"2026-05-21"``).
        period_length: Period length enum name (e.g. ``"PT15M"``, ``"PT60M"``).
        currency: Currency enum name (e.g. ``"EUR"``).
        bid_type: Bid type enum name (default ``"PERIOD_MATRIX_DA"``).
        bids: List of bid dicts, each with ``price`` (float),
            ``deliveryStartTimeUtc`` (str), and ``periodVolumes`` (list of float).

    Returns:
        Dict[str, Any]: Raw mutation response containing ``value`` and ``errors``.

    Raises:
        VolueSdaApiError: If the API returns GraphQL errors.

    Example:
        ```python
            api = SdaApi()

            result = api.update_curve_bids_package(
                address="DA_MLK_TEST_MATRIX",
                data_group="MLK_DA_MATRIX",
                delivery_date="2026-05-21",
                period_length="PT15M",
                currency="EUR",
                bid_type="PERIOD_MATRIX_DA",
                bids=[{
                    "price": 2.0,
                    "deliveryStartTimeUtc": "2026-05-20T23:00:00.000Z",
                    "periodVolumes": [-13.20, -13.20, ...],
                }],
            )
        ```
    """
    input_package = {
        "address": address,
        "dataGroup": data_group,
        "deliveryDate": delivery_date,
        "periodLength": _GqlEnum(period_length),
        "currency": _GqlEnum(currency),
        "type": _GqlEnum(bid_type),
        "bids": bids,
    }

    q = f"""
    mutation {{
      trading {{
        updateCurveBidsPackage(input: {_gql([input_package])}) {{
          value {{
            address
            dataGroup
            success
          }}
          errors {{
            code: __typename
            ... on Error {{ message }}
            ... on RestRequestError {{ message }}
          }}
        }}
      }}
    }}
    """
    return self._trading(q)["updateCurveBidsPackage"]

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."""

format_curve_bids_for_sda(df, *, address, data_group, delivery_date, bid_type='PERIOD_MATRIX_DA')

Transform database bid rows into kwargs for :meth:SdaApi.update_curve_bids_package.

The DataFrame is expected to have the following columns (matching the database schema):

• start_time_lb_utc: Period start time (datetime-like).
• price: Bid price level.
• volume_mw: Volume for this period/price combination.
• currency: Currency code (e.g. "EUR").
• resolution_seconds: Period length in seconds (e.g. 900 for 15 min).

Rows are grouped by price, sorted by start_time_lb_utc, and the volumes are collected into periodVolumes lists.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame with bid rows from the database.	required
address	str	SDA bid address for the mutation.	required
data_group	str	SDA data group for the mutation.	required
delivery_date	str	Delivery date string (ISO-8601 date, e.g. "2026-05-21").	required
bid_type	str	Bid type enum name (default "PERIOD_MATRIX_DA").	'PERIOD_MATRIX_DA'

Returns:

Type	Description
Dict[str, Any]	Dict ready to unpack into
Dict[str, Any]	meth:SdaApi.update_curve_bids_package as **kwargs.

Raises:

Type	Description
ValueError	If the DataFrame is empty or has unsupported resolution.

Example

    import pandas as pd
    from physical_operations_utils.ems_utils.SdaApi import (
        SdaApi, format_curve_bids_for_sda,
    )

    df = pd.read_sql(query, connection)

    kwargs = format_curve_bids_for_sda(
        df,
        address="DA_MLK_TEST_MATRIX",
        data_group="MLK_DA_MATRIX",
        delivery_date="2026-05-21",
    )

    api = SdaApi()
    result = api.update_curve_bids_package(**kwargs)

Source code in physical_operations_utils/ems_utils/SdaApi.py

def format_curve_bids_for_sda(
    df: pd.DataFrame,
    *,
    address: str,
    data_group: str,
    delivery_date: str,
    bid_type: str = "PERIOD_MATRIX_DA",
) -> Dict[str, Any]:
    """
    Transform database bid rows into kwargs for
    :meth:`SdaApi.update_curve_bids_package`.

    The DataFrame is expected to have the following columns
    (matching the database schema):

    - ``start_time_lb_utc``: Period start time (datetime-like).
    - ``price``: Bid price level.
    - ``volume_mw``: Volume for this period/price combination.
    - ``currency``: Currency code (e.g. ``"EUR"``).
    - ``resolution_seconds``: Period length in seconds (e.g. 900 for 15 min).

    Rows are grouped by ``price``, sorted by ``start_time_lb_utc``, and the
    volumes are collected into ``periodVolumes`` lists.

    Args:
        df: DataFrame with bid rows from the database.
        address: SDA bid address for the mutation.
        data_group: SDA data group for the mutation.
        delivery_date: Delivery date string (ISO-8601 date, e.g.
            ``"2026-05-21"``).
        bid_type: Bid type enum name (default ``"PERIOD_MATRIX_DA"``).

    Returns:
        Dict ready to unpack into
        :meth:`SdaApi.update_curve_bids_package` as ``**kwargs``.

    Raises:
        ValueError: If the DataFrame is empty or has unsupported resolution.

    Example:
        ```python
            import pandas as pd
            from physical_operations_utils.ems_utils.SdaApi import (
                SdaApi, format_curve_bids_for_sda,
            )

            df = pd.read_sql(query, connection)

            kwargs = format_curve_bids_for_sda(
                df,
                address="DA_MLK_TEST_MATRIX",
                data_group="MLK_DA_MATRIX",
                delivery_date="2026-05-21",
            )

            api = SdaApi()
            result = api.update_curve_bids_package(**kwargs)
        ```
    """
    if df.empty:
        raise ValueError("DataFrame is empty, cannot format bids.")

    resolution_seconds = int(df["resolution_seconds"].iloc[0])
    period_length_enum = PERIOD_LENGTH_MAP.get(resolution_seconds)
    if period_length_enum is None:
        raise ValueError(
            f"Unsupported resolution_seconds={resolution_seconds}. "
            f"Supported: {list(PERIOD_LENGTH_MAP.keys())}"
        )

    currency = str(df["currency"].iloc[0])

    delivery_start_utc = pd.Timestamp(df["start_time_lb_utc"].min()).strftime(
        "%Y-%m-%dT%H:%M:%S.000Z"
    )

    bids = []
    for price, group in df.groupby("price", sort=True):
        group_sorted = group.sort_values("start_time_lb_utc")
        period_volumes = group_sorted["volume_mw"].tolist()
        bids.append(
            {
                "price": float(price),
                "deliveryStartTimeUtc": delivery_start_utc,
                "periodVolumes": period_volumes,
            }
        )

    return {
        "address": address,
        "data_group": data_group,
        "delivery_date": delivery_date,
        "period_length": period_length_enum.value,
        "currency": currency,
        "bid_type": bid_type,
        "bids": bids,
    }