Index

frequenz.client.electricity_trading

Electricity Trading API client for Python.

Frequenz Electricity Trading API Client

This module provides an easy-to-use Python interface to interact with the Frequenz Electricity Trading API. It allows you to create orders, manage market data, and interact with the electricity trading ecosystem.

Features

• Create and manage gridpool orders: Place new orders, update existing ones, and cancel orders when necessary.
• Stream live data: Get real-time updates on market data, including order books, trades, and market prices.
• Retrieve historical data: Access historical data on market trades.

Installation

You can install the Frequenz Electricity Trading API client via pip. Replace VERSION with the specific version you wish to install.

Choose the version you want to install

VERSION=0.2.5
pip install frequenz-client-electricity-trading==$VERSION

Initialization

First, initialize the client with the appropriate server URL and API key.

Initialize the client

import asyncio
from frequenz.client.electricity_trading import Client

# Change server address if needed
SERVICE_URL = "grpc://electricity-trading.api.frequenz.com:443?ssl=true"
with open('/path/to/api_key.txt', 'r', encoding='utf-8') as f:
    API_KEY = f.read().strip()

async def initialize():
    client = Client(
        server_url=SERVICE_URL,
        auth_key=API_KEY
    )

asyncio.run(initialize())

Example Usage

Create an Order

Here's an example of how to create a limit order to buy energy.

Create a limit order

import asyncio
from frequenz.client.electricity_trading import (
    Client,
    Currency,
    DeliveryArea,
    DeliveryPeriod,
    EnergyMarketCodeType,
    MarketSide,
    OrderType,
    Power,
    Price,
)
from datetime import datetime, timedelta
from decimal import Decimal

# Change server address if needed
SERVICE_URL = "grpc://electricity-trading.api.frequenz.com:443?ssl=true"
with open('/path/to/api_key.txt', 'r', encoding='utf-8') as f:
    API_KEY = f.read().strip()

async def create_order():
    client = Client(
        server_url=SERVICE_URL,
        auth_key=API_KEY
    )

    # Define order parameters
    gridpool_id = 1
    delivery_area = DeliveryArea(
        code="10YDE-EON------1",  # TenneT
        code_type=EnergyMarketCodeType.EUROPE_EIC
    )
    delivery_period = DeliveryPeriod(
        start=datetime.fromisoformat("2024-05-01T00:00:00+00:00"),
        duration=timedelta(minutes=15)
    )
    price = Price(amount=Decimal("50.0"), currency=Currency.EUR)
    quantity = Power(mw=Decimal("0.1"))
    order = await client.create_gridpool_order(
        gridpool_id=gridpool_id,
        delivery_area=delivery_area,
        delivery_period=delivery_period,
        order_type=OrderType.LIMIT,
        side=MarketSide.BUY,
        price=price,
        quantity=quantity,
    )

asyncio.run(create_order())

List Orders for a Gridpool

Orders for a given gridpool can be listed using various filters.

List orders for a gridpool

import asyncio
from frequenz.client.electricity_trading import ( Client, MarketSide )

# Change server address if needed
SERVICE_URL = "grpc://electricity-trading.api.frequenz.com:443?ssl=true"
with open('/path/to/api_key.txt', 'r', encoding='utf-8') as f:
    API_KEY = f.read().strip()

async def list_orders():
    client = Client(
        server_url=SERVICE_URL,
        auth_key=API_KEY
    )

    gridpool_id: int = 1

    # List all orders for a given gridpool
    orders = await client.list_gridpool_orders(
        gridpool_id=gridpool_id,
    )

    # List only the buy orders for a given gridpool
    buy_orders = await client.list_gridpool_orders(
        gridpool_id=gridpool_id,
        side=MarketSide.BUY,
    )

asyncio.run(list_orders())

Streaming Public Trades

To get real-time updates on market trades, use the following code:

Stream public trades

import asyncio
from frequenz.client.electricity_trading import Client

# Change server address if needed
SERVICE_URL = "grpc://electricity-trading.api.frequenz.com:443?ssl=true"
with open('/path/to/api_key.txt', 'r', encoding='utf-8') as f:
    API_KEY = f.read().strip()

async def stream_trades():
    client = Client(
        server_url=SERVICE_URL,
        auth_key=API_KEY
    )
    stream_public_trades = await client.stream_public_trades()
    async for public_trade in stream_public_trades:
        print(f"Received public trade: {public_trade}")

asyncio.run(stream_trades())

Classes

frequenz.client.electricity_trading.Client

Bases: BaseApiClient[ElectricityTradingServiceStub]

Electricity trading client.

Source code in frequenz/client/electricity_trading/_client.py

class Client(BaseApiClient[ElectricityTradingServiceStub]):
    """Electricity trading client."""

    _instances: dict[tuple[str, str | None], "Client"] = {}

    def __new__(
        cls, server_url: str, connect: bool = True, auth_key: str | None = None
    ) -> "Client":
        """
        Create a new instance of the client or return an existing one if it already exists.

        Args:
            server_url: The URL of the Electricity Trading service.
            connect: Whether to connect to the server immediately.
            auth_key: The API key for the authorization.

        Returns:
            The client instance.
        """
        key = (server_url, auth_key)

        # Check if an instance already exists for this key
        if key not in cls._instances:
            # If not, create a new instance and store it in the cache
            instance = super(Client, cls).__new__(cls)
            cls._instances[key] = instance

        return cls._instances[key]

    def __init__(
        self, server_url: str, connect: bool = True, auth_key: str | None = None
    ) -> None:
        """Initialize the client.

        Args:
            server_url: The URL of the Electricity Trading service.
            connect: Whether to connect to the server immediately.
            auth_key: The API key for the authorization.
        """
        if not hasattr(
            self, "_initialized"
        ):  # Prevent re-initialization of existing instances
            super().__init__(
                server_url,
                connect=connect,
                create_stub=ElectricityTradingServiceStub,
            )
            self._initialized = True

        self._gridpool_orders_streams: dict[
            tuple[int, GridpoolOrderFilter],
            GrpcStreamBroadcaster[
                electricity_trading_pb2.ReceiveGridpoolOrdersStreamResponse, OrderDetail
            ],
        ] = {}

        self._gridpool_trades_streams: dict[
            tuple[int, GridpoolTradeFilter],
            GrpcStreamBroadcaster[
                electricity_trading_pb2.ReceiveGridpoolTradesStreamResponse, Trade
            ],
        ] = {}

        self._public_trades_streams: dict[
            PublicTradeFilter,
            GrpcStreamBroadcaster[
                electricity_trading_pb2.ReceivePublicTradesStreamResponse, PublicTrade
            ],
        ] = {}

        self._metadata = (("key", auth_key),) if auth_key else ()

    @property
    def stub(self) -> electricity_trading_pb2_grpc.ElectricityTradingServiceAsyncStub:
        """
        Get the gRPC stub for the Electricity Trading service.

        Returns:
            The gRPC stub.

        Raises:
            ClientNotConnected: If the client is not connected to the server.
        """
        if self._stub is None:
            raise ClientNotConnected(server_url=self.server_url, operation="stub")
        # This type: ignore is needed because we need to cast the sync stub to
        # the async stub, but we can't use cast because the async stub doesn't
        # actually exists to the eyes of the interpreter, it only exists for the
        # type-checker, so it can only be used for type hints.
        return self._stub  # type: ignore

    def gridpool_orders_stream(
        # pylint: disable=too-many-arguments, too-many-positional-arguments
        self,
        gridpool_id: int,
        order_states: list[OrderState] | None = None,
        market_side: MarketSide | None = None,
        delivery_area: DeliveryArea | None = None,
        delivery_period: DeliveryPeriod | None = None,
        tag: str | None = None,
    ) -> GrpcStreamBroadcaster[
        electricity_trading_pb2.ReceiveGridpoolOrdersStreamResponse, OrderDetail
    ]:
        """
        Stream gridpool orders.

        Args:
            gridpool_id: ID of the gridpool to stream orders for.
            order_states: List of order states to filter for.
            market_side: Market side to filter for.
            delivery_area: Delivery area to filter for.
            delivery_period: Delivery period to filter for.
            tag: Tag to filter for.

        Returns:
            Async generator of orders.

        Raises:
            grpc.RpcError: If an error occurs while streaming the orders.
        """
        self.validate_params(delivery_period=delivery_period)

        gridpool_order_filter = GridpoolOrderFilter(
            order_states=order_states,
            side=market_side,
            delivery_area=delivery_area,
            delivery_period=delivery_period,
            tag=tag,
        )

        stream_key = (gridpool_id, gridpool_order_filter)

        if (
            stream_key not in self._gridpool_orders_streams
            or not self._gridpool_orders_streams[stream_key].is_running
        ):
            try:
                self._gridpool_orders_streams[stream_key] = GrpcStreamBroadcaster(
                    f"electricity-trading-{stream_key}",
                    lambda: self.stub.ReceiveGridpoolOrdersStream(
                        electricity_trading_pb2.ReceiveGridpoolOrdersStreamRequest(
                            gridpool_id=gridpool_id,
                            filter=gridpool_order_filter.to_pb(),
                        ),
                        metadata=self._metadata,
                    ),
                    lambda response: OrderDetail.from_pb(response.order_detail),
                )
            except grpc.RpcError as e:
                _logger.exception(
                    "Error occurred while streaming gridpool orders: %s", e
                )
                raise
        return self._gridpool_orders_streams[stream_key]

    def gridpool_trades_stream(
        # pylint: disable=too-many-arguments, too-many-positional-arguments
        self,
        gridpool_id: int,
        trade_states: list[TradeState] | None = None,
        trade_ids: list[int] | None = None,
        market_side: MarketSide | None = None,
        delivery_period: DeliveryPeriod | None = None,
        delivery_area: DeliveryArea | None = None,
    ) -> GrpcStreamBroadcaster[
        electricity_trading_pb2.ReceiveGridpoolTradesStreamResponse, Trade
    ]:
        """
        Stream gridpool trades.

        Args:
            gridpool_id: The ID of the gridpool to stream trades for.
            trade_states: List of trade states to filter for.
            trade_ids: List of trade IDs to filter for.
            market_side: The market side to filter for.
            delivery_period: The delivery period to filter for.
            delivery_area: The delivery area to filter for.

        Returns:
            The gridpool trades streamer.

        Raises:
            grpc.RpcError: If an error occurs while streaming gridpool trades.
        """
        self.validate_params(delivery_period=delivery_period)

        gridpool_trade_filter = GridpoolTradeFilter(
            trade_states=trade_states,
            trade_ids=trade_ids,
            side=market_side,
            delivery_period=delivery_period,
            delivery_area=delivery_area,
        )

        stream_key = (gridpool_id, gridpool_trade_filter)

        if (
            stream_key not in self._gridpool_trades_streams
            or not self._gridpool_trades_streams[stream_key].is_running
        ):
            try:
                self._gridpool_trades_streams[stream_key] = GrpcStreamBroadcaster(
                    f"electricity-trading-{stream_key}",
                    lambda: self.stub.ReceiveGridpoolTradesStream(
                        electricity_trading_pb2.ReceiveGridpoolTradesStreamRequest(
                            gridpool_id=gridpool_id,
                            filter=gridpool_trade_filter.to_pb(),
                        ),
                        metadata=self._metadata,
                    ),
                    lambda response: Trade.from_pb(response.trade),
                )
            except grpc.RpcError as e:
                _logger.exception(
                    "Error occurred while streaming gridpool trades: %s", e
                )
                raise
        return self._gridpool_trades_streams[stream_key]

    def public_trades_stream(
        # pylint: disable=too-many-arguments, too-many-positional-arguments
        self,
        states: list[TradeState] | None = None,
        delivery_period: DeliveryPeriod | None = None,
        buy_delivery_area: DeliveryArea | None = None,
        sell_delivery_area: DeliveryArea | None = None,
    ) -> GrpcStreamBroadcaster[
        electricity_trading_pb2.ReceivePublicTradesStreamResponse, PublicTrade
    ]:
        """
        Stream public trades.

        Args:
            states: List of order states to filter for.
            delivery_period: Delivery period to filter for.
            buy_delivery_area: Buy delivery area to filter for.
            sell_delivery_area: Sell delivery area to filter for.

        Returns:
            Async generator of orders.

        Raises:
            grpc.RpcError: If an error occurs while streaming public trades.
        """
        self.validate_params(delivery_period=delivery_period)

        public_trade_filter = PublicTradeFilter(
            states=states,
            delivery_period=delivery_period,
            buy_delivery_area=buy_delivery_area,
            sell_delivery_area=sell_delivery_area,
        )

        if (
            public_trade_filter not in self._public_trades_streams
            or not self._public_trades_streams[public_trade_filter].is_running
        ):
            try:
                self._public_trades_streams[public_trade_filter] = (
                    GrpcStreamBroadcaster(
                        f"electricity-trading-{public_trade_filter}",
                        lambda: self.stub.ReceivePublicTradesStream(
                            electricity_trading_pb2.ReceivePublicTradesStreamRequest(
                                filter=public_trade_filter.to_pb(),
                            ),
                            metadata=self._metadata,
                        ),
                        lambda response: PublicTrade.from_pb(response.public_trade),
                    )
                )
            except grpc.RpcError as e:
                _logger.exception("Error occurred while streaming public trades: %s", e)
                raise
        return self._public_trades_streams[public_trade_filter]

    def validate_params(
        # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-branches
        self,
        price: Price | None | _Sentinel = NO_VALUE,
        quantity: Power | None | _Sentinel = NO_VALUE,
        stop_price: Price | None | _Sentinel = NO_VALUE,
        peak_price_delta: Price | None | _Sentinel = NO_VALUE,
        display_quantity: Power | None | _Sentinel = NO_VALUE,
        delivery_period: DeliveryPeriod | None = None,
        valid_until: datetime | None | _Sentinel = NO_VALUE,
        execution_option: OrderExecutionOption | None | _Sentinel = NO_VALUE,
        order_type: OrderType | None = None,
    ) -> None:
        """
        Validate the parameters of an order.

        This method ensures the following:
        - Price and quantity values have the correct number of decimal places and are positive.
        - The delivery_start and valid_until values are in the future.

        Args:
            price: The price of the order.
            quantity: The quantity of the order.
            stop_price: The stop price of the order.
            peak_price_delta: The peak price delta of the order.
            display_quantity: The display quantity of the order.
            delivery_period: The delivery period of the order.
            valid_until: The valid until of the order.
            execution_option: The execution option of the order.
            order_type: The order type.

        Raises:
            ValueError: If the parameters are invalid.
            NotImplementedError: If the order type is not supported.
        """
        if not isinstance(price, _Sentinel) and price is not None:
            if price.amount < MIN_PRICE or price.amount > MAX_PRICE:
                raise ValueError(f"Price must be between {MIN_PRICE} and {MAX_PRICE}.")
            validate_decimal_places(price.amount, PRECISION_DECIMAL_PRICE, "price")
        if not isinstance(quantity, _Sentinel) and quantity is not None:
            if quantity.mw <= 0:
                raise ValueError("Quantity must be strictly positive")
            if quantity.mw < MIN_QUANTITY_MW:
                raise ValueError(f"Quantity must be at least {MIN_QUANTITY_MW} MW.")
            validate_decimal_places(quantity.mw, PRECISION_DECIMAL_QUANTITY, "quantity")
        if not isinstance(stop_price, _Sentinel) and stop_price is not None:
            raise NotImplementedError(
                "STOP_LIMIT orders are not supported yet, so stop_price cannot be set."
            )
        if not isinstance(peak_price_delta, _Sentinel) and peak_price_delta is not None:
            raise NotImplementedError(
                "ICEBERG orders are not supported yet, so peak_price_delta cannot be set."
            )
        if not isinstance(display_quantity, _Sentinel) and display_quantity is not None:
            raise NotImplementedError(
                "ICEBERG orders are not supported yet, so display_quantity cannot be set."
            )
        if delivery_period is not None:
            if delivery_period.start < datetime.now(timezone.utc):
                raise ValueError("delivery_period must be in the future")
        if not isinstance(valid_until, _Sentinel) and valid_until is not None:
            if (
                not isinstance(execution_option, _Sentinel)
                and execution_option is not None
            ):
                if execution_option in [
                    OrderExecutionOption.AON,
                    OrderExecutionOption.FOK,
                    OrderExecutionOption.IOC,
                ]:
                    raise ValueError(
                        "valid_until must be None when execution_option is set to AON, FOK, or IOC"
                    )

            if valid_until < datetime.now(timezone.utc):
                raise ValueError("valid_until must be in the future")
        if order_type is not None:
            if order_type != OrderType.LIMIT:
                raise NotImplementedError("Currently only limit orders are supported.")

    async def create_gridpool_order(
        # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-locals
        self,
        gridpool_id: int,
        delivery_area: DeliveryArea,
        delivery_period: DeliveryPeriod,
        order_type: OrderType,
        side: MarketSide,
        price: Price,
        quantity: Power,
        stop_price: Price | None = None,
        peak_price_delta: Price | None = None,
        display_quantity: Power | None = None,
        execution_option: OrderExecutionOption | None = None,
        valid_until: datetime | None = None,
        payload: dict[str, struct_pb2.Value] | None = None,
        tag: str | None = None,
        timeout: timedelta | None = None,
    ) -> OrderDetail:
        """
        Create a gridpool order.

        Args:
            gridpool_id: ID of the gridpool to create the order for.
            delivery_area: Delivery area of the order.
            delivery_period: Delivery period of the order.
            order_type: Type of the order.
            side: Side of the order.
            price: Price of the order.
            quantity: Quantity of the order.
            stop_price: Stop price of the order.
            peak_price_delta: Peak price delta of the order.
            display_quantity: Display quantity of the order.
            execution_option: Execution option of the order.
            valid_until: Valid until of the order.
            payload: Payload of the order.
            tag: Tag of the order.
            timeout: Timeout duration, defaults to None.

        Returns:
            The created order.

        Raises:
            grpc.RpcError: An error occurred while creating the order.
        """
        self.validate_params(
            price=price,
            quantity=quantity,
            stop_price=stop_price,
            peak_price_delta=peak_price_delta,
            display_quantity=display_quantity,
            delivery_period=delivery_period,
            valid_until=valid_until,
            execution_option=execution_option,
            order_type=order_type,
        )
        order = Order(
            delivery_area=delivery_area,
            delivery_period=delivery_period,
            type=order_type,
            side=side,
            price=price,
            quantity=quantity,
            stop_price=stop_price,
            peak_price_delta=peak_price_delta,
            display_quantity=display_quantity,
            execution_option=execution_option,
            valid_until=valid_until,
            payload=payload,
            tag=tag,
        )

        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.CreateGridpoolOrderResponse],
                grpc_call_with_timeout(
                    self.stub.CreateGridpoolOrder,
                    electricity_trading_pb2.CreateGridpoolOrderRequest(
                        gridpool_id=gridpool_id, order=order.to_pb()
                    ),
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )
        except grpc.RpcError as e:
            _logger.exception("Error occurred while creating gridpool order: %s", e)
            raise

        return OrderDetail.from_pb(response.order_detail)

    async def update_gridpool_order(
        # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-locals
        self,
        gridpool_id: int,
        order_id: int,
        price: Price | None | _Sentinel = NO_VALUE,
        quantity: Power | None | _Sentinel = NO_VALUE,
        stop_price: Price | None | _Sentinel = NO_VALUE,
        peak_price_delta: Price | None | _Sentinel = NO_VALUE,
        display_quantity: Power | None | _Sentinel = NO_VALUE,
        execution_option: OrderExecutionOption | None | _Sentinel = NO_VALUE,
        valid_until: datetime | None | _Sentinel = NO_VALUE,
        payload: dict[str, struct_pb2.Value] | None | _Sentinel = NO_VALUE,
        tag: str | None | _Sentinel = NO_VALUE,
        timeout: timedelta | None = None,
    ) -> OrderDetail:
        """
        Update an existing order for a given Gridpool.

        Args:
            gridpool_id: ID of the Gridpool the order belongs to.
            order_id: Order ID.
            price: The updated limit price at which the contract is to be traded.
                This is the maximum price for a BUY order or the minimum price for a SELL order.
            quantity: The updated quantity of the contract being traded, specified in MW.
            stop_price: Applicable for STOP_LIMIT orders. This is the updated stop price that
                triggers the limit order.
            peak_price_delta: Applicable for ICEBERG orders. This is the updated price difference
                between the peak price and the limit price.
            display_quantity: Applicable for ICEBERG orders. This is the updated quantity of the
                order to be displayed in the order book.
            execution_option: Updated execution options such as All or None, Fill or Kill, etc.
            valid_until: This is an updated timestamp defining the time after which the order
                should be cancelled if not filled. The timestamp is in UTC.
            payload: Updated user-defined payload individual to a specific order. This can be any
                data that the user wants to associate with the order.
            tag: Updated user-defined tag to group related orders.
            timeout: Timeout duration, defaults to None.

        Returns:
            The updated order.

        Raises:
            ValueError: If no fields to update are provided.
            grpc.RpcError: An error occurred while updating the order.
        """
        self.validate_params(
            price=price,
            quantity=quantity,
            stop_price=stop_price,
            peak_price_delta=peak_price_delta,
            display_quantity=display_quantity,
            valid_until=valid_until,
            execution_option=execution_option,
        )

        params = {
            "price": price,
            "quantity": quantity,
            "stop_price": stop_price,
            "peak_price_delta": peak_price_delta,
            "display_quantity": display_quantity,
            "execution_option": execution_option,
            "valid_until": valid_until,
            "payload": payload,
            "tag": tag,
        }

        if all(value is NO_VALUE for value in params.values()):
            raise ValueError("At least one field to update must be provided.")

        paths = [param for param, value in params.items() if value is not NO_VALUE]

        # Field mask specifying which fields should be updated
        # This is used so that we can update parameters with None values
        update_mask = field_mask_pb2.FieldMask(paths=paths)

        update_order_fields = UpdateOrder(
            price=None if price is NO_VALUE else price,  # type: ignore
            quantity=None if quantity is NO_VALUE else quantity,  # type: ignore
            stop_price=None if stop_price is NO_VALUE else stop_price,  # type: ignore
            peak_price_delta=(
                None if peak_price_delta is NO_VALUE else peak_price_delta  # type: ignore
            ),
            display_quantity=(
                None if display_quantity is NO_VALUE else display_quantity  # type: ignore
            ),
            execution_option=(
                None if execution_option is NO_VALUE else execution_option  # type: ignore
            ),
            valid_until=(
                None if valid_until is NO_VALUE else valid_until  # type: ignore
            ),
            payload=None if payload is NO_VALUE else payload,  # type: ignore
            tag=None if tag is NO_VALUE else tag,  # type: ignore
        )

        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.UpdateGridpoolOrderResponse],
                grpc_call_with_timeout(
                    self.stub.UpdateGridpoolOrder,
                    electricity_trading_pb2.UpdateGridpoolOrderRequest(
                        gridpool_id=gridpool_id,
                        order_id=order_id,
                        update_order_fields=update_order_fields.to_pb(),
                        update_mask=update_mask,
                    ),
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )
            return OrderDetail.from_pb(response.order_detail)

        except grpc.RpcError as e:
            _logger.exception("Error occurred while updating gridpool order: %s", e)
            raise

    async def cancel_gridpool_order(
        self, gridpool_id: int, order_id: int, timeout: timedelta | None = None
    ) -> OrderDetail:
        """
        Cancel a single order for a given Gridpool.

        Args:
            gridpool_id: The Gridpool to cancel the order for.
            order_id: The order to cancel.
            timeout: Timeout duration, defaults to None.

        Returns:
            The cancelled order.

        Raises:
            grpc.RpcError: If an error occurs while cancelling the gridpool order.
        """
        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.CancelGridpoolOrderResponse],
                grpc_call_with_timeout(
                    self.stub.CancelGridpoolOrder,
                    electricity_trading_pb2.CancelGridpoolOrderRequest(
                        gridpool_id=gridpool_id, order_id=order_id
                    ),
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )
            return OrderDetail.from_pb(response.order_detail)
        except grpc.RpcError as e:
            _logger.exception("Error occurred while cancelling gridpool order: %s", e)
            raise

    async def cancel_all_gridpool_orders(
        self, gridpool_id: int, timeout: timedelta | None = None
    ) -> int:
        """
        Cancel all orders for a specific Gridpool.

        Args:
            gridpool_id: The Gridpool to cancel the orders for.
            timeout: Timeout duration, defaults to None.

        Returns:
            The ID of the Gridpool for which the orders were cancelled.

        Raises:
            grpc.RpcError: If an error occurs while cancelling all gridpool orders.
        """
        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.CancelAllGridpoolOrdersResponse],
                grpc_call_with_timeout(
                    self.stub.CancelAllGridpoolOrders,
                    electricity_trading_pb2.CancelAllGridpoolOrdersRequest(
                        gridpool_id=gridpool_id
                    ),
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )

            return response.gridpool_id
        except grpc.RpcError as e:
            _logger.exception(
                "Error occurred while cancelling all gridpool orders: %s", e
            )
            raise

    async def get_gridpool_order(
        self, gridpool_id: int, order_id: int, timeout: timedelta | None = None
    ) -> OrderDetail:
        """
        Get a single order from a given gridpool.

        Args:
            gridpool_id: The Gridpool to retrieve the order for.
            order_id: The order to retrieve.
            timeout: Timeout duration, defaults to None.

        Returns:
            The order.

        Raises:
            grpc.RpcError: If an error occurs while getting the order.
        """
        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.GetGridpoolOrderResponse],
                grpc_call_with_timeout(
                    self.stub.GetGridpoolOrder,
                    electricity_trading_pb2.GetGridpoolOrderRequest(
                        gridpool_id=gridpool_id, order_id=order_id
                    ),
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )

            return OrderDetail.from_pb(response.order_detail)
        except grpc.RpcError as e:
            _logger.exception("Error occurred while getting gridpool order: %s", e)
            raise

    async def list_gridpool_orders(
        # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-locals
        self,
        gridpool_id: int,
        order_states: list[OrderState] | None = None,
        side: MarketSide | None = None,
        delivery_period: DeliveryPeriod | None = None,
        delivery_area: DeliveryArea | None = None,
        tag: str | None = None,
        page_size: int | None = None,
        timeout: timedelta | None = None,
    ) -> AsyncIterator[OrderDetail]:
        """
        List orders for a specific Gridpool with optional filters.

        Args:
            gridpool_id: The Gridpool to retrieve the orders for.
            order_states: List of order states to filter by.
            side: The side of the market to filter by.
            delivery_period: The delivery period to filter by.
            delivery_area: The delivery area to filter by.
            tag: The tag to filter by.
            page_size: The number of orders to return per page.
            timeout: Timeout duration, defaults to None.

        Yields:
            The list of orders for the given gridpool.

        Raises:
            grpc.RpcError: If an error occurs while listing the orders.
        """
        gridpool_order_filter = GridpoolOrderFilter(
            order_states=order_states,
            side=side,
            delivery_period=delivery_period,
            delivery_area=delivery_area,
            tag=tag,
        )

        request = electricity_trading_pb2.ListGridpoolOrdersRequest(
            gridpool_id=gridpool_id,
            filter=gridpool_order_filter.to_pb(),
            pagination_params=(
                Params(page_size=page_size).to_proto() if page_size else None
            ),
        )
        while True:
            try:
                response = await cast(
                    Awaitable[electricity_trading_pb2.ListGridpoolOrdersResponse],
                    grpc_call_with_timeout(
                        self.stub.ListGridpoolOrders,
                        request,
                        metadata=self._metadata,
                        timeout=timeout,
                    ),
                )

                for order_detail in response.order_details:
                    yield OrderDetail.from_pb(order_detail)

                if response.pagination_info.next_page_token:
                    request.pagination_params.CopyFrom(
                        PaginationParams(
                            page_token=response.pagination_info.next_page_token
                        )
                    )
                else:
                    break

            except grpc.RpcError as e:
                _logger.exception("Error occurred while listing gridpool orders: %s", e)
                raise

    async def list_gridpool_trades(
        # pylint: disable=too-many-arguments, too-many-positional-arguments
        self,
        gridpool_id: int,
        trade_states: list[TradeState] | None = None,
        trade_ids: list[int] | None = None,
        market_side: MarketSide | None = None,
        delivery_period: DeliveryPeriod | None = None,
        delivery_area: DeliveryArea | None = None,
        page_size: int | None = None,
        timeout: timedelta | None = None,
    ) -> AsyncIterator[Trade]:
        """
        List trades for a specific Gridpool with optional filters.

        Args:
            gridpool_id: The Gridpool to retrieve the trades for.
            trade_states: List of trade states to filter by.
            trade_ids: List of trade IDs to filter by.
            market_side: The side of the market to filter by.
            delivery_period: The delivery period to filter by.
            delivery_area: The delivery area to filter by.
            page_size: The number of trades to return per page.
            timeout: Timeout duration, defaults to None.

        Yields:
            The list of trades for the given gridpool.

        Raises:
            grpc.RpcError: If an error occurs while listing gridpool trades.
        """
        gridpool_trade_filter = GridpoolTradeFilter(
            trade_states=trade_states,
            trade_ids=trade_ids,
            side=market_side,
            delivery_period=delivery_period,
            delivery_area=delivery_area,
        )

        request = electricity_trading_pb2.ListGridpoolTradesRequest(
            gridpool_id=gridpool_id,
            filter=gridpool_trade_filter.to_pb(),
            pagination_params=(
                Params(page_size=page_size).to_proto() if page_size else None
            ),
        )

        while True:
            try:
                response = await cast(
                    Awaitable[electricity_trading_pb2.ListGridpoolTradesResponse],
                    grpc_call_with_timeout(
                        self.stub.ListGridpoolTrades,
                        request,
                        metadata=self._metadata,
                        timeout=timeout,
                    ),
                )

                for trade in response.trades:
                    yield Trade.from_pb(trade)

                if response.pagination_info.next_page_token:
                    request.pagination_params.CopyFrom(
                        PaginationParams(
                            page_token=response.pagination_info.next_page_token
                        )
                    )
                else:
                    break

            except grpc.RpcError as e:
                _logger.exception("Error occurred while listing gridpool trades: %s", e)
                raise

    async def list_public_trades(
        # pylint: disable=too-many-arguments, too-many-positional-arguments
        self,
        states: list[TradeState] | None = None,
        delivery_period: DeliveryPeriod | None = None,
        buy_delivery_area: DeliveryArea | None = None,
        sell_delivery_area: DeliveryArea | None = None,
        page_size: int | None = None,
        timeout: timedelta | None = None,
    ) -> AsyncIterator[PublicTrade]:
        """
        List all executed public orders with optional filters and pagination.

        Args:
            states: List of order states to filter by.
            delivery_period: The delivery period to filter by.
            buy_delivery_area: The buy delivery area to filter by.
            sell_delivery_area: The sell delivery area to filter by.
            page_size: The number of public trades to return per page.
            timeout: Timeout duration, defaults to None.

        Yields:
            The list of public trades for each page.

        Raises:
            grpc.RpcError: If an error occurs while listing public trades.
        """
        public_trade_filter = PublicTradeFilter(
            states=states,
            delivery_period=delivery_period,
            buy_delivery_area=buy_delivery_area,
            sell_delivery_area=sell_delivery_area,
        )

        request = electricity_trading_pb2.ListPublicTradesRequest(
            filter=public_trade_filter.to_pb(),
            pagination_params=(
                Params(page_size=page_size).to_proto() if page_size else None
            ),
        )

        while True:
            try:
                response = await cast(
                    Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
                    grpc_call_with_timeout(
                        self.stub.ListPublicTrades,
                        request,
                        metadata=self._metadata,
                        timeout=timeout,
                    ),
                )

                for public_trade in response.public_trades:
                    yield PublicTrade.from_pb(public_trade)

                if response.pagination_info.next_page_token:
                    request.pagination_params.CopyFrom(
                        PaginationParams(
                            page_token=response.pagination_info.next_page_token
                        )
                    )
                else:
                    break

            except grpc.RpcError as e:
                _logger.exception("Error occurred while listing public trades: %s", e)
                raise

Attributes

channel property

channel: Channel

The underlying gRPC channel used to communicate with the server.

Warning

This channel is provided as a last resort for advanced users. It is not recommended to use this property directly unless you know what you are doing and you don't care about being tied to a specific gRPC library.

RAISES	DESCRIPTION
ClientNotConnected	If the client is not connected to the server.

channel_defaults property

channel_defaults: ChannelOptions

The default options for the gRPC channel.

is_connected property

is_connected: bool

Whether the client is connected to the server.

server_url property

server_url: str

The URL of the server.

stub property

stub: ElectricityTradingServiceAsyncStub

Get the gRPC stub for the Electricity Trading service.

RETURNS	DESCRIPTION
ElectricityTradingServiceAsyncStub	The gRPC stub.

RAISES	DESCRIPTION
ClientNotConnected	If the client is not connected to the server.

Functions

__aenter__ async

__aenter__() -> Self

Enter a context manager.

Source code in frequenz/client/base/client.py

async def __aenter__(self) -> Self:
    """Enter a context manager."""
    self.connect()
    return self

__aexit__ async

__aexit__(
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: Any | None,
) -> bool | None

Exit a context manager.

Source code in frequenz/client/base/client.py

async def __aexit__(
    self,
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: Any | None,
) -> bool | None:
    """Exit a context manager."""
    if self._channel is None:
        return None
    result = await self._channel.__aexit__(_exc_type, _exc_val, _exc_tb)
    self._channel = None
    self._stub = None
    return result

__init__

__init__(
    server_url: str,
    connect: bool = True,
    auth_key: str | None = None,
) -> None

Initialize the client.

PARAMETER	DESCRIPTION
server_url	The URL of the Electricity Trading service. TYPE: str
connect	Whether to connect to the server immediately. TYPE: bool DEFAULT: True
auth_key	The API key for the authorization. TYPE: str | None DEFAULT: None

Source code in frequenz/client/electricity_trading/_client.py

def __init__(
    self, server_url: str, connect: bool = True, auth_key: str | None = None
) -> None:
    """Initialize the client.

    Args:
        server_url: The URL of the Electricity Trading service.
        connect: Whether to connect to the server immediately.
        auth_key: The API key for the authorization.
    """
    if not hasattr(
        self, "_initialized"
    ):  # Prevent re-initialization of existing instances
        super().__init__(
            server_url,
            connect=connect,
            create_stub=ElectricityTradingServiceStub,
        )
        self._initialized = True

    self._gridpool_orders_streams: dict[
        tuple[int, GridpoolOrderFilter],
        GrpcStreamBroadcaster[
            electricity_trading_pb2.ReceiveGridpoolOrdersStreamResponse, OrderDetail
        ],
    ] = {}

    self._gridpool_trades_streams: dict[
        tuple[int, GridpoolTradeFilter],
        GrpcStreamBroadcaster[
            electricity_trading_pb2.ReceiveGridpoolTradesStreamResponse, Trade
        ],
    ] = {}

    self._public_trades_streams: dict[
        PublicTradeFilter,
        GrpcStreamBroadcaster[
            electricity_trading_pb2.ReceivePublicTradesStreamResponse, PublicTrade
        ],
    ] = {}

    self._metadata = (("key", auth_key),) if auth_key else ()

__new__

__new__(
    server_url: str,
    connect: bool = True,
    auth_key: str | None = None,
) -> "Client"

Create a new instance of the client or return an existing one if it already exists.

PARAMETER	DESCRIPTION
server_url	The URL of the Electricity Trading service. TYPE: str
connect	Whether to connect to the server immediately. TYPE: bool DEFAULT: True
auth_key	The API key for the authorization. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
'Client'	The client instance.

Source code in frequenz/client/electricity_trading/_client.py

def __new__(
    cls, server_url: str, connect: bool = True, auth_key: str | None = None
) -> "Client":
    """
    Create a new instance of the client or return an existing one if it already exists.

    Args:
        server_url: The URL of the Electricity Trading service.
        connect: Whether to connect to the server immediately.
        auth_key: The API key for the authorization.

    Returns:
        The client instance.
    """
    key = (server_url, auth_key)

    # Check if an instance already exists for this key
    if key not in cls._instances:
        # If not, create a new instance and store it in the cache
        instance = super(Client, cls).__new__(cls)
        cls._instances[key] = instance

    return cls._instances[key]

cancel_all_gridpool_orders async

cancel_all_gridpool_orders(
    gridpool_id: int, timeout: timedelta | None = None
) -> int

Cancel all orders for a specific Gridpool.

PARAMETER	DESCRIPTION
gridpool_id	The Gridpool to cancel the orders for. TYPE: int
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

RETURNS	DESCRIPTION
int	The ID of the Gridpool for which the orders were cancelled.

RAISES	DESCRIPTION
RpcError	If an error occurs while cancelling all gridpool orders.

Source code in frequenz/client/electricity_trading/_client.py

async def cancel_all_gridpool_orders(
    self, gridpool_id: int, timeout: timedelta | None = None
) -> int:
    """
    Cancel all orders for a specific Gridpool.

    Args:
        gridpool_id: The Gridpool to cancel the orders for.
        timeout: Timeout duration, defaults to None.

    Returns:
        The ID of the Gridpool for which the orders were cancelled.

    Raises:
        grpc.RpcError: If an error occurs while cancelling all gridpool orders.
    """
    try:
        response = await cast(
            Awaitable[electricity_trading_pb2.CancelAllGridpoolOrdersResponse],
            grpc_call_with_timeout(
                self.stub.CancelAllGridpoolOrders,
                electricity_trading_pb2.CancelAllGridpoolOrdersRequest(
                    gridpool_id=gridpool_id
                ),
                metadata=self._metadata,
                timeout=timeout,
            ),
        )

        return response.gridpool_id
    except grpc.RpcError as e:
        _logger.exception(
            "Error occurred while cancelling all gridpool orders: %s", e
        )
        raise

cancel_gridpool_order async

cancel_gridpool_order(
    gridpool_id: int,
    order_id: int,
    timeout: timedelta | None = None,
) -> OrderDetail

Cancel a single order for a given Gridpool.

PARAMETER	DESCRIPTION
gridpool_id	The Gridpool to cancel the order for. TYPE: int
order_id	The order to cancel. TYPE: int
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

RETURNS	DESCRIPTION
OrderDetail	The cancelled order.

RAISES	DESCRIPTION
RpcError	If an error occurs while cancelling the gridpool order.

Source code in frequenz/client/electricity_trading/_client.py

async def cancel_gridpool_order(
    self, gridpool_id: int, order_id: int, timeout: timedelta | None = None
) -> OrderDetail:
    """
    Cancel a single order for a given Gridpool.

    Args:
        gridpool_id: The Gridpool to cancel the order for.
        order_id: The order to cancel.
        timeout: Timeout duration, defaults to None.

    Returns:
        The cancelled order.

    Raises:
        grpc.RpcError: If an error occurs while cancelling the gridpool order.
    """
    try:
        response = await cast(
            Awaitable[electricity_trading_pb2.CancelGridpoolOrderResponse],
            grpc_call_with_timeout(
                self.stub.CancelGridpoolOrder,
                electricity_trading_pb2.CancelGridpoolOrderRequest(
                    gridpool_id=gridpool_id, order_id=order_id
                ),
                metadata=self._metadata,
                timeout=timeout,
            ),
        )
        return OrderDetail.from_pb(response.order_detail)
    except grpc.RpcError as e:
        _logger.exception("Error occurred while cancelling gridpool order: %s", e)
        raise

connect

connect(server_url: str | None = None) -> None

Connect to the server, possibly using a new URL.

If the client is already connected and the URL is the same as the previous URL, this method does nothing. If you want to force a reconnection, you can call disconnect() first.

PARAMETER	DESCRIPTION
server_url	The URL of the server to connect to. If not provided, the previously used URL is used. TYPE: str | None DEFAULT: None

Source code in frequenz/client/base/client.py

def connect(self, server_url: str | None = None) -> None:
    """Connect to the server, possibly using a new URL.

    If the client is already connected and the URL is the same as the previous URL,
    this method does nothing. If you want to force a reconnection, you can call
    [disconnect()][frequenz.client.base.client.BaseApiClient.disconnect] first.

    Args:
        server_url: The URL of the server to connect to. If not provided, the
            previously used URL is used.
    """
    if server_url is not None and server_url != self._server_url:  # URL changed
        self._server_url = server_url
    elif self.is_connected:
        return
    self._channel = parse_grpc_uri(self._server_url, self._channel_defaults)
    self._stub = self._create_stub(self._channel)

create_gridpool_order async

create_gridpool_order(
    gridpool_id: int,
    delivery_area: DeliveryArea,
    delivery_period: DeliveryPeriod,
    order_type: OrderType,
    side: MarketSide,
    price: Price,
    quantity: Power,
    stop_price: Price | None = None,
    peak_price_delta: Price | None = None,
    display_quantity: Power | None = None,
    execution_option: OrderExecutionOption | None = None,
    valid_until: datetime | None = None,
    payload: dict[str, Value] | None = None,
    tag: str | None = None,
    timeout: timedelta | None = None,
) -> OrderDetail

Create a gridpool order.

PARAMETER	DESCRIPTION
gridpool_id	ID of the gridpool to create the order for. TYPE: int
delivery_area	Delivery area of the order. TYPE: DeliveryArea
delivery_period	Delivery period of the order. TYPE: DeliveryPeriod
order_type	Type of the order. TYPE: OrderType
side	Side of the order. TYPE: MarketSide
price	Price of the order. TYPE: Price
quantity	Quantity of the order. TYPE: Power
stop_price	Stop price of the order. TYPE: Price | None DEFAULT: None
peak_price_delta	Peak price delta of the order. TYPE: Price | None DEFAULT: None
display_quantity	Display quantity of the order. TYPE: Power | None DEFAULT: None
execution_option	Execution option of the order. TYPE: OrderExecutionOption | None DEFAULT: None
valid_until	Valid until of the order. TYPE: datetime | None DEFAULT: None
payload	Payload of the order. TYPE: dict[str, Value] | None DEFAULT: None
tag	Tag of the order. TYPE: str | None DEFAULT: None
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

RETURNS	DESCRIPTION
OrderDetail	The created order.

RAISES	DESCRIPTION
RpcError	An error occurred while creating the order.

Source code in frequenz/client/electricity_trading/_client.py

async def create_gridpool_order(
    # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-locals
    self,
    gridpool_id: int,
    delivery_area: DeliveryArea,
    delivery_period: DeliveryPeriod,
    order_type: OrderType,
    side: MarketSide,
    price: Price,
    quantity: Power,
    stop_price: Price | None = None,
    peak_price_delta: Price | None = None,
    display_quantity: Power | None = None,
    execution_option: OrderExecutionOption | None = None,
    valid_until: datetime | None = None,
    payload: dict[str, struct_pb2.Value] | None = None,
    tag: str | None = None,
    timeout: timedelta | None = None,
) -> OrderDetail:
    """
    Create a gridpool order.

    Args:
        gridpool_id: ID of the gridpool to create the order for.
        delivery_area: Delivery area of the order.
        delivery_period: Delivery period of the order.
        order_type: Type of the order.
        side: Side of the order.
        price: Price of the order.
        quantity: Quantity of the order.
        stop_price: Stop price of the order.
        peak_price_delta: Peak price delta of the order.
        display_quantity: Display quantity of the order.
        execution_option: Execution option of the order.
        valid_until: Valid until of the order.
        payload: Payload of the order.
        tag: Tag of the order.
        timeout: Timeout duration, defaults to None.

    Returns:
        The created order.

    Raises:
        grpc.RpcError: An error occurred while creating the order.
    """
    self.validate_params(
        price=price,
        quantity=quantity,
        stop_price=stop_price,
        peak_price_delta=peak_price_delta,
        display_quantity=display_quantity,
        delivery_period=delivery_period,
        valid_until=valid_until,
        execution_option=execution_option,
        order_type=order_type,
    )
    order = Order(
        delivery_area=delivery_area,
        delivery_period=delivery_period,
        type=order_type,
        side=side,
        price=price,
        quantity=quantity,
        stop_price=stop_price,
        peak_price_delta=peak_price_delta,
        display_quantity=display_quantity,
        execution_option=execution_option,
        valid_until=valid_until,
        payload=payload,
        tag=tag,
    )

    try:
        response = await cast(
            Awaitable[electricity_trading_pb2.CreateGridpoolOrderResponse],
            grpc_call_with_timeout(
                self.stub.CreateGridpoolOrder,
                electricity_trading_pb2.CreateGridpoolOrderRequest(
                    gridpool_id=gridpool_id, order=order.to_pb()
                ),
                metadata=self._metadata,
                timeout=timeout,
            ),
        )
    except grpc.RpcError as e:
        _logger.exception("Error occurred while creating gridpool order: %s", e)
        raise

    return OrderDetail.from_pb(response.order_detail)

disconnect async

disconnect() -> None

Disconnect from the server.

If the client is not connected, this method does nothing.

Source code in frequenz/client/base/client.py

async def disconnect(self) -> None:
    """Disconnect from the server.

    If the client is not connected, this method does nothing.
    """
    await self.__aexit__(None, None, None)

get_gridpool_order async

get_gridpool_order(
    gridpool_id: int,
    order_id: int,
    timeout: timedelta | None = None,
) -> OrderDetail

Get a single order from a given gridpool.

PARAMETER	DESCRIPTION
gridpool_id	The Gridpool to retrieve the order for. TYPE: int
order_id	The order to retrieve. TYPE: int
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

RETURNS	DESCRIPTION
OrderDetail	The order.

RAISES	DESCRIPTION
RpcError	If an error occurs while getting the order.

Source code in frequenz/client/electricity_trading/_client.py

async def get_gridpool_order(
    self, gridpool_id: int, order_id: int, timeout: timedelta | None = None
) -> OrderDetail:
    """
    Get a single order from a given gridpool.

    Args:
        gridpool_id: The Gridpool to retrieve the order for.
        order_id: The order to retrieve.
        timeout: Timeout duration, defaults to None.

    Returns:
        The order.

    Raises:
        grpc.RpcError: If an error occurs while getting the order.
    """
    try:
        response = await cast(
            Awaitable[electricity_trading_pb2.GetGridpoolOrderResponse],
            grpc_call_with_timeout(
                self.stub.GetGridpoolOrder,
                electricity_trading_pb2.GetGridpoolOrderRequest(
                    gridpool_id=gridpool_id, order_id=order_id
                ),
                metadata=self._metadata,
                timeout=timeout,
            ),
        )

        return OrderDetail.from_pb(response.order_detail)
    except grpc.RpcError as e:
        _logger.exception("Error occurred while getting gridpool order: %s", e)
        raise

gridpool_orders_stream

gridpool_orders_stream(
    gridpool_id: int,
    order_states: list[OrderState] | None = None,
    market_side: MarketSide | None = None,
    delivery_area: DeliveryArea | None = None,
    delivery_period: DeliveryPeriod | None = None,
    tag: str | None = None,
) -> GrpcStreamBroadcaster[
    ReceiveGridpoolOrdersStreamResponse, OrderDetail
]

Stream gridpool orders.

PARAMETER	DESCRIPTION
gridpool_id	ID of the gridpool to stream orders for. TYPE: int
order_states	List of order states to filter for. TYPE: list[OrderState] | None DEFAULT: None
market_side	Market side to filter for. TYPE: MarketSide | None DEFAULT: None
delivery_area	Delivery area to filter for. TYPE: DeliveryArea | None DEFAULT: None
delivery_period	Delivery period to filter for. TYPE: DeliveryPeriod | None DEFAULT: None
tag	Tag to filter for. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
GrpcStreamBroadcaster[ReceiveGridpoolOrdersStreamResponse, OrderDetail]	Async generator of orders.

RAISES	DESCRIPTION
RpcError	If an error occurs while streaming the orders.

Source code in frequenz/client/electricity_trading/_client.py

def gridpool_orders_stream(
    # pylint: disable=too-many-arguments, too-many-positional-arguments
    self,
    gridpool_id: int,
    order_states: list[OrderState] | None = None,
    market_side: MarketSide | None = None,
    delivery_area: DeliveryArea | None = None,
    delivery_period: DeliveryPeriod | None = None,
    tag: str | None = None,
) -> GrpcStreamBroadcaster[
    electricity_trading_pb2.ReceiveGridpoolOrdersStreamResponse, OrderDetail
]:
    """
    Stream gridpool orders.

    Args:
        gridpool_id: ID of the gridpool to stream orders for.
        order_states: List of order states to filter for.
        market_side: Market side to filter for.
        delivery_area: Delivery area to filter for.
        delivery_period: Delivery period to filter for.
        tag: Tag to filter for.

    Returns:
        Async generator of orders.

    Raises:
        grpc.RpcError: If an error occurs while streaming the orders.
    """
    self.validate_params(delivery_period=delivery_period)

    gridpool_order_filter = GridpoolOrderFilter(
        order_states=order_states,
        side=market_side,
        delivery_area=delivery_area,
        delivery_period=delivery_period,
        tag=tag,
    )

    stream_key = (gridpool_id, gridpool_order_filter)

    if (
        stream_key not in self._gridpool_orders_streams
        or not self._gridpool_orders_streams[stream_key].is_running
    ):
        try:
            self._gridpool_orders_streams[stream_key] = GrpcStreamBroadcaster(
                f"electricity-trading-{stream_key}",
                lambda: self.stub.ReceiveGridpoolOrdersStream(
                    electricity_trading_pb2.ReceiveGridpoolOrdersStreamRequest(
                        gridpool_id=gridpool_id,
                        filter=gridpool_order_filter.to_pb(),
                    ),
                    metadata=self._metadata,
                ),
                lambda response: OrderDetail.from_pb(response.order_detail),
            )
        except grpc.RpcError as e:
            _logger.exception(
                "Error occurred while streaming gridpool orders: %s", e
            )
            raise
    return self._gridpool_orders_streams[stream_key]

gridpool_trades_stream

gridpool_trades_stream(
    gridpool_id: int,
    trade_states: list[TradeState] | None = None,
    trade_ids: list[int] | None = None,
    market_side: MarketSide | None = None,
    delivery_period: DeliveryPeriod | None = None,
    delivery_area: DeliveryArea | None = None,
) -> GrpcStreamBroadcaster[
    ReceiveGridpoolTradesStreamResponse, Trade
]

Stream gridpool trades.

PARAMETER	DESCRIPTION
gridpool_id	The ID of the gridpool to stream trades for. TYPE: int
trade_states	List of trade states to filter for. TYPE: list[TradeState] | None DEFAULT: None
trade_ids	List of trade IDs to filter for. TYPE: list[int] | None DEFAULT: None
market_side	The market side to filter for. TYPE: MarketSide | None DEFAULT: None
delivery_period	The delivery period to filter for. TYPE: DeliveryPeriod | None DEFAULT: None
delivery_area	The delivery area to filter for. TYPE: DeliveryArea | None DEFAULT: None

RETURNS	DESCRIPTION
GrpcStreamBroadcaster[ReceiveGridpoolTradesStreamResponse, Trade]	The gridpool trades streamer.

RAISES	DESCRIPTION
RpcError	If an error occurs while streaming gridpool trades.

Source code in frequenz/client/electricity_trading/_client.py

def gridpool_trades_stream(
    # pylint: disable=too-many-arguments, too-many-positional-arguments
    self,
    gridpool_id: int,
    trade_states: list[TradeState] | None = None,
    trade_ids: list[int] | None = None,
    market_side: MarketSide | None = None,
    delivery_period: DeliveryPeriod | None = None,
    delivery_area: DeliveryArea | None = None,
) -> GrpcStreamBroadcaster[
    electricity_trading_pb2.ReceiveGridpoolTradesStreamResponse, Trade
]:
    """
    Stream gridpool trades.

    Args:
        gridpool_id: The ID of the gridpool to stream trades for.
        trade_states: List of trade states to filter for.
        trade_ids: List of trade IDs to filter for.
        market_side: The market side to filter for.
        delivery_period: The delivery period to filter for.
        delivery_area: The delivery area to filter for.

    Returns:
        The gridpool trades streamer.

    Raises:
        grpc.RpcError: If an error occurs while streaming gridpool trades.
    """
    self.validate_params(delivery_period=delivery_period)

    gridpool_trade_filter = GridpoolTradeFilter(
        trade_states=trade_states,
        trade_ids=trade_ids,
        side=market_side,
        delivery_period=delivery_period,
        delivery_area=delivery_area,
    )

    stream_key = (gridpool_id, gridpool_trade_filter)

    if (
        stream_key not in self._gridpool_trades_streams
        or not self._gridpool_trades_streams[stream_key].is_running
    ):
        try:
            self._gridpool_trades_streams[stream_key] = GrpcStreamBroadcaster(
                f"electricity-trading-{stream_key}",
                lambda: self.stub.ReceiveGridpoolTradesStream(
                    electricity_trading_pb2.ReceiveGridpoolTradesStreamRequest(
                        gridpool_id=gridpool_id,
                        filter=gridpool_trade_filter.to_pb(),
                    ),
                    metadata=self._metadata,
                ),
                lambda response: Trade.from_pb(response.trade),
            )
        except grpc.RpcError as e:
            _logger.exception(
                "Error occurred while streaming gridpool trades: %s", e
            )
            raise
    return self._gridpool_trades_streams[stream_key]

list_gridpool_orders async

list_gridpool_orders(
    gridpool_id: int,
    order_states: list[OrderState] | None = None,
    side: MarketSide | None = None,
    delivery_period: DeliveryPeriod | None = None,
    delivery_area: DeliveryArea | None = None,
    tag: str | None = None,
    page_size: int | None = None,
    timeout: timedelta | None = None,
) -> AsyncIterator[OrderDetail]

List orders for a specific Gridpool with optional filters.

PARAMETER	DESCRIPTION
gridpool_id	The Gridpool to retrieve the orders for. TYPE: int
order_states	List of order states to filter by. TYPE: list[OrderState] | None DEFAULT: None
side	The side of the market to filter by. TYPE: MarketSide | None DEFAULT: None
delivery_period	The delivery period to filter by. TYPE: DeliveryPeriod | None DEFAULT: None
delivery_area	The delivery area to filter by. TYPE: DeliveryArea | None DEFAULT: None
tag	The tag to filter by. TYPE: str | None DEFAULT: None
page_size	The number of orders to return per page. TYPE: int | None DEFAULT: None
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

YIELDS	DESCRIPTION
AsyncIterator[OrderDetail]	The list of orders for the given gridpool.

RAISES	DESCRIPTION
RpcError	If an error occurs while listing the orders.

Source code in frequenz/client/electricity_trading/_client.py

async def list_gridpool_orders(
    # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-locals
    self,
    gridpool_id: int,
    order_states: list[OrderState] | None = None,
    side: MarketSide | None = None,
    delivery_period: DeliveryPeriod | None = None,
    delivery_area: DeliveryArea | None = None,
    tag: str | None = None,
    page_size: int | None = None,
    timeout: timedelta | None = None,
) -> AsyncIterator[OrderDetail]:
    """
    List orders for a specific Gridpool with optional filters.

    Args:
        gridpool_id: The Gridpool to retrieve the orders for.
        order_states: List of order states to filter by.
        side: The side of the market to filter by.
        delivery_period: The delivery period to filter by.
        delivery_area: The delivery area to filter by.
        tag: The tag to filter by.
        page_size: The number of orders to return per page.
        timeout: Timeout duration, defaults to None.

    Yields:
        The list of orders for the given gridpool.

    Raises:
        grpc.RpcError: If an error occurs while listing the orders.
    """
    gridpool_order_filter = GridpoolOrderFilter(
        order_states=order_states,
        side=side,
        delivery_period=delivery_period,
        delivery_area=delivery_area,
        tag=tag,
    )

    request = electricity_trading_pb2.ListGridpoolOrdersRequest(
        gridpool_id=gridpool_id,
        filter=gridpool_order_filter.to_pb(),
        pagination_params=(
            Params(page_size=page_size).to_proto() if page_size else None
        ),
    )
    while True:
        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.ListGridpoolOrdersResponse],
                grpc_call_with_timeout(
                    self.stub.ListGridpoolOrders,
                    request,
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )

            for order_detail in response.order_details:
                yield OrderDetail.from_pb(order_detail)

            if response.pagination_info.next_page_token:
                request.pagination_params.CopyFrom(
                    PaginationParams(
                        page_token=response.pagination_info.next_page_token
                    )
                )
            else:
                break

        except grpc.RpcError as e:
            _logger.exception("Error occurred while listing gridpool orders: %s", e)
            raise

list_gridpool_trades async

list_gridpool_trades(
    gridpool_id: int,
    trade_states: list[TradeState] | None = None,
    trade_ids: list[int] | None = None,
    market_side: MarketSide | None = None,
    delivery_period: DeliveryPeriod | None = None,
    delivery_area: DeliveryArea | None = None,
    page_size: int | None = None,
    timeout: timedelta | None = None,
) -> AsyncIterator[Trade]

List trades for a specific Gridpool with optional filters.

PARAMETER	DESCRIPTION
gridpool_id	The Gridpool to retrieve the trades for. TYPE: int
trade_states	List of trade states to filter by. TYPE: list[TradeState] | None DEFAULT: None
trade_ids	List of trade IDs to filter by. TYPE: list[int] | None DEFAULT: None
market_side	The side of the market to filter by. TYPE: MarketSide | None DEFAULT: None
delivery_period	The delivery period to filter by. TYPE: DeliveryPeriod | None DEFAULT: None
delivery_area	The delivery area to filter by. TYPE: DeliveryArea | None DEFAULT: None
page_size	The number of trades to return per page. TYPE: int | None DEFAULT: None
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

YIELDS	DESCRIPTION
AsyncIterator[Trade]	The list of trades for the given gridpool.

RAISES	DESCRIPTION
RpcError	If an error occurs while listing gridpool trades.

Source code in frequenz/client/electricity_trading/_client.py

async def list_gridpool_trades(
    # pylint: disable=too-many-arguments, too-many-positional-arguments
    self,
    gridpool_id: int,
    trade_states: list[TradeState] | None = None,
    trade_ids: list[int] | None = None,
    market_side: MarketSide | None = None,
    delivery_period: DeliveryPeriod | None = None,
    delivery_area: DeliveryArea | None = None,
    page_size: int | None = None,
    timeout: timedelta | None = None,
) -> AsyncIterator[Trade]:
    """
    List trades for a specific Gridpool with optional filters.

    Args:
        gridpool_id: The Gridpool to retrieve the trades for.
        trade_states: List of trade states to filter by.
        trade_ids: List of trade IDs to filter by.
        market_side: The side of the market to filter by.
        delivery_period: The delivery period to filter by.
        delivery_area: The delivery area to filter by.
        page_size: The number of trades to return per page.
        timeout: Timeout duration, defaults to None.

    Yields:
        The list of trades for the given gridpool.

    Raises:
        grpc.RpcError: If an error occurs while listing gridpool trades.
    """
    gridpool_trade_filter = GridpoolTradeFilter(
        trade_states=trade_states,
        trade_ids=trade_ids,
        side=market_side,
        delivery_period=delivery_period,
        delivery_area=delivery_area,
    )

    request = electricity_trading_pb2.ListGridpoolTradesRequest(
        gridpool_id=gridpool_id,
        filter=gridpool_trade_filter.to_pb(),
        pagination_params=(
            Params(page_size=page_size).to_proto() if page_size else None
        ),
    )

    while True:
        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.ListGridpoolTradesResponse],
                grpc_call_with_timeout(
                    self.stub.ListGridpoolTrades,
                    request,
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )

            for trade in response.trades:
                yield Trade.from_pb(trade)

            if response.pagination_info.next_page_token:
                request.pagination_params.CopyFrom(
                    PaginationParams(
                        page_token=response.pagination_info.next_page_token
                    )
                )
            else:
                break

        except grpc.RpcError as e:
            _logger.exception("Error occurred while listing gridpool trades: %s", e)
            raise

list_public_trades async

list_public_trades(
    states: list[TradeState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
    page_size: int | None = None,
    timeout: timedelta | None = None,
) -> AsyncIterator[PublicTrade]

List all executed public orders with optional filters and pagination.

PARAMETER	DESCRIPTION
states	List of order states to filter by. TYPE: list[TradeState] | None DEFAULT: None
delivery_period	The delivery period to filter by. TYPE: DeliveryPeriod | None DEFAULT: None
buy_delivery_area	The buy delivery area to filter by. TYPE: DeliveryArea | None DEFAULT: None
sell_delivery_area	The sell delivery area to filter by. TYPE: DeliveryArea | None DEFAULT: None
page_size	The number of public trades to return per page. TYPE: int | None DEFAULT: None
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

YIELDS	DESCRIPTION
AsyncIterator[PublicTrade]	The list of public trades for each page.

RAISES	DESCRIPTION
RpcError	If an error occurs while listing public trades.

Source code in frequenz/client/electricity_trading/_client.py

async def list_public_trades(
    # pylint: disable=too-many-arguments, too-many-positional-arguments
    self,
    states: list[TradeState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
    page_size: int | None = None,
    timeout: timedelta | None = None,
) -> AsyncIterator[PublicTrade]:
    """
    List all executed public orders with optional filters and pagination.

    Args:
        states: List of order states to filter by.
        delivery_period: The delivery period to filter by.
        buy_delivery_area: The buy delivery area to filter by.
        sell_delivery_area: The sell delivery area to filter by.
        page_size: The number of public trades to return per page.
        timeout: Timeout duration, defaults to None.

    Yields:
        The list of public trades for each page.

    Raises:
        grpc.RpcError: If an error occurs while listing public trades.
    """
    public_trade_filter = PublicTradeFilter(
        states=states,
        delivery_period=delivery_period,
        buy_delivery_area=buy_delivery_area,
        sell_delivery_area=sell_delivery_area,
    )

    request = electricity_trading_pb2.ListPublicTradesRequest(
        filter=public_trade_filter.to_pb(),
        pagination_params=(
            Params(page_size=page_size).to_proto() if page_size else None
        ),
    )

    while True:
        try:
            response = await cast(
                Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
                grpc_call_with_timeout(
                    self.stub.ListPublicTrades,
                    request,
                    metadata=self._metadata,
                    timeout=timeout,
                ),
            )

            for public_trade in response.public_trades:
                yield PublicTrade.from_pb(public_trade)

            if response.pagination_info.next_page_token:
                request.pagination_params.CopyFrom(
                    PaginationParams(
                        page_token=response.pagination_info.next_page_token
                    )
                )
            else:
                break

        except grpc.RpcError as e:
            _logger.exception("Error occurred while listing public trades: %s", e)
            raise

public_trades_stream

public_trades_stream(
    states: list[TradeState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
) -> GrpcStreamBroadcaster[
    ReceivePublicTradesStreamResponse, PublicTrade
]

Stream public trades.

PARAMETER	DESCRIPTION
states	List of order states to filter for. TYPE: list[TradeState] | None DEFAULT: None
delivery_period	Delivery period to filter for. TYPE: DeliveryPeriod | None DEFAULT: None
buy_delivery_area	Buy delivery area to filter for. TYPE: DeliveryArea | None DEFAULT: None
sell_delivery_area	Sell delivery area to filter for. TYPE: DeliveryArea | None DEFAULT: None

RETURNS	DESCRIPTION
GrpcStreamBroadcaster[ReceivePublicTradesStreamResponse, PublicTrade]	Async generator of orders.

RAISES	DESCRIPTION
RpcError	If an error occurs while streaming public trades.

Source code in frequenz/client/electricity_trading/_client.py

def public_trades_stream(
    # pylint: disable=too-many-arguments, too-many-positional-arguments
    self,
    states: list[TradeState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
) -> GrpcStreamBroadcaster[
    electricity_trading_pb2.ReceivePublicTradesStreamResponse, PublicTrade
]:
    """
    Stream public trades.

    Args:
        states: List of order states to filter for.
        delivery_period: Delivery period to filter for.
        buy_delivery_area: Buy delivery area to filter for.
        sell_delivery_area: Sell delivery area to filter for.

    Returns:
        Async generator of orders.

    Raises:
        grpc.RpcError: If an error occurs while streaming public trades.
    """
    self.validate_params(delivery_period=delivery_period)

    public_trade_filter = PublicTradeFilter(
        states=states,
        delivery_period=delivery_period,
        buy_delivery_area=buy_delivery_area,
        sell_delivery_area=sell_delivery_area,
    )

    if (
        public_trade_filter not in self._public_trades_streams
        or not self._public_trades_streams[public_trade_filter].is_running
    ):
        try:
            self._public_trades_streams[public_trade_filter] = (
                GrpcStreamBroadcaster(
                    f"electricity-trading-{public_trade_filter}",
                    lambda: self.stub.ReceivePublicTradesStream(
                        electricity_trading_pb2.ReceivePublicTradesStreamRequest(
                            filter=public_trade_filter.to_pb(),
                        ),
                        metadata=self._metadata,
                    ),
                    lambda response: PublicTrade.from_pb(response.public_trade),
                )
            )
        except grpc.RpcError as e:
            _logger.exception("Error occurred while streaming public trades: %s", e)
            raise
    return self._public_trades_streams[public_trade_filter]

update_gridpool_order async

update_gridpool_order(
    gridpool_id: int,
    order_id: int,
    price: Price | None | _Sentinel = NO_VALUE,
    quantity: Power | None | _Sentinel = NO_VALUE,
    stop_price: Price | None | _Sentinel = NO_VALUE,
    peak_price_delta: Price | None | _Sentinel = NO_VALUE,
    display_quantity: Power | None | _Sentinel = NO_VALUE,
    execution_option: (
        OrderExecutionOption | None | _Sentinel
    ) = NO_VALUE,
    valid_until: datetime | None | _Sentinel = NO_VALUE,
    payload: dict[str, Value] | None | _Sentinel = NO_VALUE,
    tag: str | None | _Sentinel = NO_VALUE,
    timeout: timedelta | None = None,
) -> OrderDetail

Update an existing order for a given Gridpool.

PARAMETER	DESCRIPTION
gridpool_id	ID of the Gridpool the order belongs to. TYPE: int
order_id	Order ID. TYPE: int
price	The updated limit price at which the contract is to be traded. This is the maximum price for a BUY order or the minimum price for a SELL order. TYPE: Price | None | _Sentinel DEFAULT: NO_VALUE
quantity	The updated quantity of the contract being traded, specified in MW. TYPE: Power | None | _Sentinel DEFAULT: NO_VALUE
stop_price	Applicable for STOP_LIMIT orders. This is the updated stop price that triggers the limit order. TYPE: Price | None | _Sentinel DEFAULT: NO_VALUE
peak_price_delta	Applicable for ICEBERG orders. This is the updated price difference between the peak price and the limit price. TYPE: Price | None | _Sentinel DEFAULT: NO_VALUE
display_quantity	Applicable for ICEBERG orders. This is the updated quantity of the order to be displayed in the order book. TYPE: Power | None | _Sentinel DEFAULT: NO_VALUE
execution_option	Updated execution options such as All or None, Fill or Kill, etc. TYPE: OrderExecutionOption | None | _Sentinel DEFAULT: NO_VALUE
valid_until	This is an updated timestamp defining the time after which the order should be cancelled if not filled. The timestamp is in UTC. TYPE: datetime | None | _Sentinel DEFAULT: NO_VALUE
payload	Updated user-defined payload individual to a specific order. This can be any data that the user wants to associate with the order. TYPE: dict[str, Value] | None | _Sentinel DEFAULT: NO_VALUE
tag	Updated user-defined tag to group related orders. TYPE: str | None | _Sentinel DEFAULT: NO_VALUE
timeout	Timeout duration, defaults to None. TYPE: timedelta | None DEFAULT: None

RETURNS	DESCRIPTION
OrderDetail	The updated order.

RAISES	DESCRIPTION
ValueError	If no fields to update are provided.
RpcError	An error occurred while updating the order.

Source code in frequenz/client/electricity_trading/_client.py

async def update_gridpool_order(
    # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-locals
    self,
    gridpool_id: int,
    order_id: int,
    price: Price | None | _Sentinel = NO_VALUE,
    quantity: Power | None | _Sentinel = NO_VALUE,
    stop_price: Price | None | _Sentinel = NO_VALUE,
    peak_price_delta: Price | None | _Sentinel = NO_VALUE,
    display_quantity: Power | None | _Sentinel = NO_VALUE,
    execution_option: OrderExecutionOption | None | _Sentinel = NO_VALUE,
    valid_until: datetime | None | _Sentinel = NO_VALUE,
    payload: dict[str, struct_pb2.Value] | None | _Sentinel = NO_VALUE,
    tag: str | None | _Sentinel = NO_VALUE,
    timeout: timedelta | None = None,
) -> OrderDetail:
    """
    Update an existing order for a given Gridpool.

    Args:
        gridpool_id: ID of the Gridpool the order belongs to.
        order_id: Order ID.
        price: The updated limit price at which the contract is to be traded.
            This is the maximum price for a BUY order or the minimum price for a SELL order.
        quantity: The updated quantity of the contract being traded, specified in MW.
        stop_price: Applicable for STOP_LIMIT orders. This is the updated stop price that
            triggers the limit order.
        peak_price_delta: Applicable for ICEBERG orders. This is the updated price difference
            between the peak price and the limit price.
        display_quantity: Applicable for ICEBERG orders. This is the updated quantity of the
            order to be displayed in the order book.
        execution_option: Updated execution options such as All or None, Fill or Kill, etc.
        valid_until: This is an updated timestamp defining the time after which the order
            should be cancelled if not filled. The timestamp is in UTC.
        payload: Updated user-defined payload individual to a specific order. This can be any
            data that the user wants to associate with the order.
        tag: Updated user-defined tag to group related orders.
        timeout: Timeout duration, defaults to None.

    Returns:
        The updated order.

    Raises:
        ValueError: If no fields to update are provided.
        grpc.RpcError: An error occurred while updating the order.
    """
    self.validate_params(
        price=price,
        quantity=quantity,
        stop_price=stop_price,
        peak_price_delta=peak_price_delta,
        display_quantity=display_quantity,
        valid_until=valid_until,
        execution_option=execution_option,
    )

    params = {
        "price": price,
        "quantity": quantity,
        "stop_price": stop_price,
        "peak_price_delta": peak_price_delta,
        "display_quantity": display_quantity,
        "execution_option": execution_option,
        "valid_until": valid_until,
        "payload": payload,
        "tag": tag,
    }

    if all(value is NO_VALUE for value in params.values()):
        raise ValueError("At least one field to update must be provided.")

    paths = [param for param, value in params.items() if value is not NO_VALUE]

    # Field mask specifying which fields should be updated
    # This is used so that we can update parameters with None values
    update_mask = field_mask_pb2.FieldMask(paths=paths)

    update_order_fields = UpdateOrder(
        price=None if price is NO_VALUE else price,  # type: ignore
        quantity=None if quantity is NO_VALUE else quantity,  # type: ignore
        stop_price=None if stop_price is NO_VALUE else stop_price,  # type: ignore
        peak_price_delta=(
            None if peak_price_delta is NO_VALUE else peak_price_delta  # type: ignore
        ),
        display_quantity=(
            None if display_quantity is NO_VALUE else display_quantity  # type: ignore
        ),
        execution_option=(
            None if execution_option is NO_VALUE else execution_option  # type: ignore
        ),
        valid_until=(
            None if valid_until is NO_VALUE else valid_until  # type: ignore
        ),
        payload=None if payload is NO_VALUE else payload,  # type: ignore
        tag=None if tag is NO_VALUE else tag,  # type: ignore
    )

    try:
        response = await cast(
            Awaitable[electricity_trading_pb2.UpdateGridpoolOrderResponse],
            grpc_call_with_timeout(
                self.stub.UpdateGridpoolOrder,
                electricity_trading_pb2.UpdateGridpoolOrderRequest(
                    gridpool_id=gridpool_id,
                    order_id=order_id,
                    update_order_fields=update_order_fields.to_pb(),
                    update_mask=update_mask,
                ),
                metadata=self._metadata,
                timeout=timeout,
            ),
        )
        return OrderDetail.from_pb(response.order_detail)

    except grpc.RpcError as e:
        _logger.exception("Error occurred while updating gridpool order: %s", e)
        raise

validate_params

validate_params(
    price: Price | None | _Sentinel = NO_VALUE,
    quantity: Power | None | _Sentinel = NO_VALUE,
    stop_price: Price | None | _Sentinel = NO_VALUE,
    peak_price_delta: Price | None | _Sentinel = NO_VALUE,
    display_quantity: Power | None | _Sentinel = NO_VALUE,
    delivery_period: DeliveryPeriod | None = None,
    valid_until: datetime | None | _Sentinel = NO_VALUE,
    execution_option: (
        OrderExecutionOption | None | _Sentinel
    ) = NO_VALUE,
    order_type: OrderType | None = None,
) -> None

Validate the parameters of an order.

This method ensures the following: - Price and quantity values have the correct number of decimal places and are positive. - The delivery_start and valid_until values are in the future.

PARAMETER	DESCRIPTION
price	The price of the order. TYPE: Price | None | _Sentinel DEFAULT: NO_VALUE
quantity	The quantity of the order. TYPE: Power | None | _Sentinel DEFAULT: NO_VALUE
stop_price	The stop price of the order. TYPE: Price | None | _Sentinel DEFAULT: NO_VALUE
peak_price_delta	The peak price delta of the order. TYPE: Price | None | _Sentinel DEFAULT: NO_VALUE
display_quantity	The display quantity of the order. TYPE: Power | None | _Sentinel DEFAULT: NO_VALUE
delivery_period	The delivery period of the order. TYPE: DeliveryPeriod | None DEFAULT: None
valid_until	The valid until of the order. TYPE: datetime | None | _Sentinel DEFAULT: NO_VALUE
execution_option	The execution option of the order. TYPE: OrderExecutionOption | None | _Sentinel DEFAULT: NO_VALUE
order_type	The order type. TYPE: OrderType | None DEFAULT: None

RAISES	DESCRIPTION
ValueError	If the parameters are invalid.
NotImplementedError	If the order type is not supported.

Source code in frequenz/client/electricity_trading/_client.py

def validate_params(
    # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-branches
    self,
    price: Price | None | _Sentinel = NO_VALUE,
    quantity: Power | None | _Sentinel = NO_VALUE,
    stop_price: Price | None | _Sentinel = NO_VALUE,
    peak_price_delta: Price | None | _Sentinel = NO_VALUE,
    display_quantity: Power | None | _Sentinel = NO_VALUE,
    delivery_period: DeliveryPeriod | None = None,
    valid_until: datetime | None | _Sentinel = NO_VALUE,
    execution_option: OrderExecutionOption | None | _Sentinel = NO_VALUE,
    order_type: OrderType | None = None,
) -> None:
    """
    Validate the parameters of an order.

    This method ensures the following:
    - Price and quantity values have the correct number of decimal places and are positive.
    - The delivery_start and valid_until values are in the future.

    Args:
        price: The price of the order.
        quantity: The quantity of the order.
        stop_price: The stop price of the order.
        peak_price_delta: The peak price delta of the order.
        display_quantity: The display quantity of the order.
        delivery_period: The delivery period of the order.
        valid_until: The valid until of the order.
        execution_option: The execution option of the order.
        order_type: The order type.

    Raises:
        ValueError: If the parameters are invalid.
        NotImplementedError: If the order type is not supported.
    """
    if not isinstance(price, _Sentinel) and price is not None:
        if price.amount < MIN_PRICE or price.amount > MAX_PRICE:
            raise ValueError(f"Price must be between {MIN_PRICE} and {MAX_PRICE}.")
        validate_decimal_places(price.amount, PRECISION_DECIMAL_PRICE, "price")
    if not isinstance(quantity, _Sentinel) and quantity is not None:
        if quantity.mw <= 0:
            raise ValueError("Quantity must be strictly positive")
        if quantity.mw < MIN_QUANTITY_MW:
            raise ValueError(f"Quantity must be at least {MIN_QUANTITY_MW} MW.")
        validate_decimal_places(quantity.mw, PRECISION_DECIMAL_QUANTITY, "quantity")
    if not isinstance(stop_price, _Sentinel) and stop_price is not None:
        raise NotImplementedError(
            "STOP_LIMIT orders are not supported yet, so stop_price cannot be set."
        )
    if not isinstance(peak_price_delta, _Sentinel) and peak_price_delta is not None:
        raise NotImplementedError(
            "ICEBERG orders are not supported yet, so peak_price_delta cannot be set."
        )
    if not isinstance(display_quantity, _Sentinel) and display_quantity is not None:
        raise NotImplementedError(
            "ICEBERG orders are not supported yet, so display_quantity cannot be set."
        )
    if delivery_period is not None:
        if delivery_period.start < datetime.now(timezone.utc):
            raise ValueError("delivery_period must be in the future")
    if not isinstance(valid_until, _Sentinel) and valid_until is not None:
        if (
            not isinstance(execution_option, _Sentinel)
            and execution_option is not None
        ):
            if execution_option in [
                OrderExecutionOption.AON,
                OrderExecutionOption.FOK,
                OrderExecutionOption.IOC,
            ]:
                raise ValueError(
                    "valid_until must be None when execution_option is set to AON, FOK, or IOC"
                )

        if valid_until < datetime.now(timezone.utc):
            raise ValueError("valid_until must be in the future")
    if order_type is not None:
        if order_type != OrderType.LIMIT:
            raise NotImplementedError("Currently only limit orders are supported.")

frequenz.client.electricity_trading.Currency

Bases: Enum

List of supported currencies.

New currencies can be added to this enum to support additional markets.

Source code in frequenz/client/electricity_trading/_types.py

class Currency(enum.Enum):
    """
    List of supported currencies.

    New currencies can be added to this enum to support additional markets.
    """

    UNSPECIFIED = price_pb2.Price.Currency.CURRENCY_UNSPECIFIED
    """Currency is not specified."""

    USD = price_pb2.Price.Currency.CURRENCY_USD

    CAD = price_pb2.Price.Currency.CURRENCY_CAD

    EUR = price_pb2.Price.Currency.CURRENCY_EUR

    GBP = price_pb2.Price.Currency.CURRENCY_GBP

    CHF = price_pb2.Price.Currency.CURRENCY_CHF

    CNY = price_pb2.Price.Currency.CURRENCY_CNY

    JPY = price_pb2.Price.Currency.CURRENCY_JPY

    AUD = price_pb2.Price.Currency.CURRENCY_AUD

    NZD = price_pb2.Price.Currency.CURRENCY_NZD

    SGD = price_pb2.Price.Currency.CURRENCY_SGD

    @classmethod
    @from_pb
    def from_pb(cls, currency: price_pb2.Price.Currency.ValueType) -> "Currency":
        """Convert a protobuf Currency value to Currency enum.

        Args:
            currency: Currency to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == currency for e in cls):
            _logger.warning("Unknown currency %s. Returning UNSPECIFIED.", currency)
            return cls.UNSPECIFIED

        return cls(currency)

    def to_pb(self) -> price_pb2.Price.Currency.ValueType:
        """Convert a Currency object to protobuf Currency.

        Returns:
            Protobuf message corresponding to the Currency object.
        """
        return price_pb2.Price.Currency.ValueType(self.value)

Attributes

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = CURRENCY_UNSPECIFIED

Currency is not specified.

Functions

from_pb classmethod

from_pb(currency: ValueType) -> 'Currency'

Convert a protobuf Currency value to Currency enum.

PARAMETER	DESCRIPTION
currency	Currency to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'Currency'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, currency: price_pb2.Price.Currency.ValueType) -> "Currency":
    """Convert a protobuf Currency value to Currency enum.

    Args:
        currency: Currency to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == currency for e in cls):
        _logger.warning("Unknown currency %s. Returning UNSPECIFIED.", currency)
        return cls.UNSPECIFIED

    return cls(currency)

to_pb

to_pb() -> ValueType

Convert a Currency object to protobuf Currency.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the Currency object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> price_pb2.Price.Currency.ValueType:
    """Convert a Currency object to protobuf Currency.

    Returns:
        Protobuf message corresponding to the Currency object.
    """
    return price_pb2.Price.Currency.ValueType(self.value)

frequenz.client.electricity_trading.DeliveryArea dataclass

Geographical or administrative region.

These are, usually defined and maintained by a Transmission System Operator (TSO), where electricity deliveries for a contract occur.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass(frozen=True)
class DeliveryArea:
    """
    Geographical or administrative region.

    These are, usually defined and maintained by a Transmission System Operator (TSO), where
    electricity deliveries for a contract occur.
    """

    code: str
    """Code representing the unique identifier for the delivery area."""

    code_type: EnergyMarketCodeType
    """Type of code used for identifying the delivery area itself."""

    @classmethod
    @from_pb
    def from_pb(cls, delivery_area: delivery_area_pb2.DeliveryArea) -> Self:
        """Convert a protobuf DeliveryArea to DeliveryArea object.

        Args:
            delivery_area: DeliveryArea to convert.

        Returns:
            DeliveryArea object corresponding to the protobuf message.
        """
        return cls(
            code=delivery_area.code,
            code_type=EnergyMarketCodeType.from_pb(delivery_area.code_type),
        )

    def to_pb(self) -> delivery_area_pb2.DeliveryArea:
        """Convert a DeliveryArea object to protobuf DeliveryArea.

        Returns:
            Protobuf message corresponding to the DeliveryArea object.
        """
        return delivery_area_pb2.DeliveryArea(
            code=self.code, code_type=self.code_type.to_pb()
        )

Attributes

code instance-attribute

code: str

Code representing the unique identifier for the delivery area.

code_type instance-attribute

code_type: EnergyMarketCodeType

Type of code used for identifying the delivery area itself.

Functions

from_pb classmethod

from_pb(delivery_area: DeliveryArea) -> Self

Convert a protobuf DeliveryArea to DeliveryArea object.

PARAMETER	DESCRIPTION
delivery_area	DeliveryArea to convert. TYPE: DeliveryArea

RETURNS	DESCRIPTION
Self	DeliveryArea object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, delivery_area: delivery_area_pb2.DeliveryArea) -> Self:
    """Convert a protobuf DeliveryArea to DeliveryArea object.

    Args:
        delivery_area: DeliveryArea to convert.

    Returns:
        DeliveryArea object corresponding to the protobuf message.
    """
    return cls(
        code=delivery_area.code,
        code_type=EnergyMarketCodeType.from_pb(delivery_area.code_type),
    )

to_pb

to_pb() -> DeliveryArea

Convert a DeliveryArea object to protobuf DeliveryArea.

RETURNS	DESCRIPTION
DeliveryArea	Protobuf message corresponding to the DeliveryArea object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> delivery_area_pb2.DeliveryArea:
    """Convert a DeliveryArea object to protobuf DeliveryArea.

    Returns:
        Protobuf message corresponding to the DeliveryArea object.
    """
    return delivery_area_pb2.DeliveryArea(
        code=self.code, code_type=self.code_type.to_pb()
    )

frequenz.client.electricity_trading.DeliveryDuration

Bases: Enum

Specifies the time increment, in minutes, used for electricity deliveries and trading.

These durations serve as the basis for defining the delivery period in contracts, and they dictate how energy is scheduled and delivered to meet contractual obligations.

Source code in frequenz/client/electricity_trading/_types.py

class DeliveryDuration(enum.Enum):
    """
    Specifies the time increment, in minutes, used for electricity deliveries and trading.

    These durations serve as the basis for defining the delivery period in contracts, and they
    dictate how energy is scheduled and delivered to meet contractual obligations.
    """

    UNSPECIFIED = delivery_duration_pb2.DeliveryDuration.DELIVERY_DURATION_UNSPECIFIED
    """Default value, indicates that the duration is unspecified."""

    MINUTES_5 = delivery_duration_pb2.DeliveryDuration.DELIVERY_DURATION_5
    """5-minute duration."""

    MINUTES_15 = delivery_duration_pb2.DeliveryDuration.DELIVERY_DURATION_15
    """15-minute contract duration."""

    MINUTES_30 = delivery_duration_pb2.DeliveryDuration.DELIVERY_DURATION_30
    """30-minute contract duration."""

    MINUTES_60 = delivery_duration_pb2.DeliveryDuration.DELIVERY_DURATION_60
    """1-hour contract duration."""

    @classmethod
    @from_pb
    def from_pb(
        cls, delivery_duration: delivery_duration_pb2.DeliveryDuration.ValueType
    ) -> "DeliveryDuration":
        """Convert a protobuf DeliveryDuration value to DeliveryDuration enum.

        Args:
            delivery_duration: Delivery duration to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == delivery_duration for e in cls):
            _logger.warning(
                "Unknown delivery duration %s. Returning UNSPECIFIED.",
                delivery_duration,
            )
            return cls.UNSPECIFIED

        return cls(delivery_duration)

    def to_pb(self) -> delivery_duration_pb2.DeliveryDuration.ValueType:
        """Convert a DeliveryDuration object to protobuf DeliveryDuration.

        Returns:
            Protobuf message corresponding to the DeliveryDuration object.
        """
        return delivery_duration_pb2.DeliveryDuration.ValueType(self.value)

Attributes

MINUTES_15 class-attribute instance-attribute

MINUTES_15 = DELIVERY_DURATION_15

15-minute contract duration.

MINUTES_30 class-attribute instance-attribute

MINUTES_30 = DELIVERY_DURATION_30

30-minute contract duration.

MINUTES_5 class-attribute instance-attribute

MINUTES_5 = DELIVERY_DURATION_5

5-minute duration.

MINUTES_60 class-attribute instance-attribute

MINUTES_60 = DELIVERY_DURATION_60

1-hour contract duration.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = DELIVERY_DURATION_UNSPECIFIED

Default value, indicates that the duration is unspecified.

Functions

from_pb classmethod

from_pb(delivery_duration: ValueType) -> 'DeliveryDuration'

Convert a protobuf DeliveryDuration value to DeliveryDuration enum.

PARAMETER	DESCRIPTION
delivery_duration	Delivery duration to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'DeliveryDuration'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, delivery_duration: delivery_duration_pb2.DeliveryDuration.ValueType
) -> "DeliveryDuration":
    """Convert a protobuf DeliveryDuration value to DeliveryDuration enum.

    Args:
        delivery_duration: Delivery duration to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == delivery_duration for e in cls):
        _logger.warning(
            "Unknown delivery duration %s. Returning UNSPECIFIED.",
            delivery_duration,
        )
        return cls.UNSPECIFIED

    return cls(delivery_duration)

to_pb

to_pb() -> ValueType

Convert a DeliveryDuration object to protobuf DeliveryDuration.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the DeliveryDuration object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> delivery_duration_pb2.DeliveryDuration.ValueType:
    """Convert a DeliveryDuration object to protobuf DeliveryDuration.

    Returns:
        Protobuf message corresponding to the DeliveryDuration object.
    """
    return delivery_duration_pb2.DeliveryDuration.ValueType(self.value)

frequenz.client.electricity_trading.DeliveryPeriod

Time period during which the contract is delivered.

It is defined by a start timestamp and a duration.

Source code in frequenz/client/electricity_trading/_types.py

class DeliveryPeriod:
    """
    Time period during which the contract is delivered.

    It is defined by a start timestamp and a duration.
    """

    start: datetime
    """Start UTC timestamp represents the beginning of the delivery period.
        This timestamp is inclusive, meaning that the delivery period starts
        from this point in time."""

    duration: DeliveryDuration
    """The length of the delivery period."""

    def __init__(
        self,
        start: datetime,
        duration: timedelta,
    ) -> None:
        """
        Initialize the DeliveryPeriod object.

        Args:
            start: Start UTC timestamp represents the beginning of the delivery period.
            duration: The length of the delivery period.

        Raises:
            ValueError: If the start timestamp does not have a timezone.
                        or if the duration is not 5, 15, 30, or 60 minutes.
        """
        if start.tzinfo is None:
            raise ValueError("Start timestamp must have a timezone.")
        if start.tzinfo != timezone.utc:
            _logger.warning(
                "Start timestamp is not in UTC timezone. Converting to UTC."
            )
            start = start.astimezone(timezone.utc)
        self.start = start

        minutes = duration.total_seconds() / 60
        match minutes:
            case 5:
                self.duration = DeliveryDuration.MINUTES_5
            case 15:
                self.duration = DeliveryDuration.MINUTES_15
            case 30:
                self.duration = DeliveryDuration.MINUTES_30
            case 60:
                self.duration = DeliveryDuration.MINUTES_60
            case _:
                raise ValueError(
                    "Invalid duration value. Duration must be 5, 15, 30, or 60 minutes."
                )

    def __hash__(self) -> int:
        """
        Create hash of the DeliveryPeriod object.

        Returns:
            Hash of the DeliveryPeriod object.
        """
        return hash((self.start, self.duration))

    def __eq__(
        self,
        other: object,
    ) -> bool:
        """Check if two DeliveryPeriod objects are equal."""
        if not isinstance(other, DeliveryPeriod):
            return NotImplemented

        return self.start == other.start and self.duration == other.duration

    def __str__(self) -> str:
        """
        Return string representation of the DeliveryPeriod object.

        Returns:
            String representation of the DeliveryPeriod object.
        """
        duration_map = {
            DeliveryDuration.MINUTES_5: "5min",
            DeliveryDuration.MINUTES_15: "15min",
            DeliveryDuration.MINUTES_30: "30min",
            DeliveryDuration.MINUTES_60: "60min",
        }
        duration_str = duration_map.get(self.duration, "Unknown duration")
        start_str = self.start.strftime("%Y-%m-%d %H:%M:%S %Z")
        return f"DeliveryPeriod(start={start_str}, duration={duration_str})"

    def __repr__(self) -> str:
        """
        Developer-friendly representation of the DeliveryPeriod object.

        Returns:
            Developer-friendly representation of the DeliveryPeriod object.
        """
        return self.__str__()

    @classmethod
    @from_pb
    def from_pb(cls, delivery_period: delivery_duration_pb2.DeliveryPeriod) -> Self:
        """Convert a protobuf DeliveryPeriod to DeliveryPeriod object.

        Args:
            delivery_period: DeliveryPeriod to convert.

        Returns:
            DeliveryPeriod object corresponding to the protobuf message.

        Raises:
            ValueError: If the duration is not 5, 15, 30, or 60 minutes.
        """
        start = delivery_period.start.ToDatetime(tzinfo=timezone.utc)
        delivery_duration_enum = DeliveryDuration.from_pb(delivery_period.duration)

        match delivery_duration_enum:
            case DeliveryDuration.MINUTES_5:
                duration = timedelta(minutes=5)
            case DeliveryDuration.MINUTES_15:
                duration = timedelta(minutes=15)
            case DeliveryDuration.MINUTES_30:
                duration = timedelta(minutes=30)
            case DeliveryDuration.MINUTES_60:
                duration = timedelta(minutes=60)
            case _:
                raise ValueError(
                    "Invalid duration value. Duration must be 5, 15, 30, or 60 minutes."
                )
        return cls(start=start, duration=duration)

    def to_pb(self) -> delivery_duration_pb2.DeliveryPeriod:
        """Convert a DeliveryPeriod object to protobuf DeliveryPeriod.

        Returns:
            Protobuf message corresponding to the DeliveryPeriod object.
        """
        start = timestamp_pb2.Timestamp()
        start.FromDatetime(self.start)
        return delivery_duration_pb2.DeliveryPeriod(
            start=start,
            duration=self.duration.to_pb(),
        )

Attributes

duration instance-attribute

duration: DeliveryDuration = MINUTES_60

The length of the delivery period.

start instance-attribute

start: datetime = start

Start UTC timestamp represents the beginning of the delivery period. This timestamp is inclusive, meaning that the delivery period starts from this point in time.

Functions

__eq__

__eq__(other: object) -> bool

Check if two DeliveryPeriod objects are equal.

Source code in frequenz/client/electricity_trading/_types.py

def __eq__(
    self,
    other: object,
) -> bool:
    """Check if two DeliveryPeriod objects are equal."""
    if not isinstance(other, DeliveryPeriod):
        return NotImplemented

    return self.start == other.start and self.duration == other.duration

__hash__

__hash__() -> int

Create hash of the DeliveryPeriod object.

RETURNS	DESCRIPTION
int	Hash of the DeliveryPeriod object.

Source code in frequenz/client/electricity_trading/_types.py

def __hash__(self) -> int:
    """
    Create hash of the DeliveryPeriod object.

    Returns:
        Hash of the DeliveryPeriod object.
    """
    return hash((self.start, self.duration))

__init__

__init__(start: datetime, duration: timedelta) -> None

Initialize the DeliveryPeriod object.

PARAMETER	DESCRIPTION
start	Start UTC timestamp represents the beginning of the delivery period. TYPE: datetime
duration	The length of the delivery period. TYPE: timedelta

RAISES	DESCRIPTION
ValueError	If the start timestamp does not have a timezone. or if the duration is not 5, 15, 30, or 60 minutes.

Source code in frequenz/client/electricity_trading/_types.py

def __init__(
    self,
    start: datetime,
    duration: timedelta,
) -> None:
    """
    Initialize the DeliveryPeriod object.

    Args:
        start: Start UTC timestamp represents the beginning of the delivery period.
        duration: The length of the delivery period.

    Raises:
        ValueError: If the start timestamp does not have a timezone.
                    or if the duration is not 5, 15, 30, or 60 minutes.
    """
    if start.tzinfo is None:
        raise ValueError("Start timestamp must have a timezone.")
    if start.tzinfo != timezone.utc:
        _logger.warning(
            "Start timestamp is not in UTC timezone. Converting to UTC."
        )
        start = start.astimezone(timezone.utc)
    self.start = start

    minutes = duration.total_seconds() / 60
    match minutes:
        case 5:
            self.duration = DeliveryDuration.MINUTES_5
        case 15:
            self.duration = DeliveryDuration.MINUTES_15
        case 30:
            self.duration = DeliveryDuration.MINUTES_30
        case 60:
            self.duration = DeliveryDuration.MINUTES_60
        case _:
            raise ValueError(
                "Invalid duration value. Duration must be 5, 15, 30, or 60 minutes."
            )

__repr__

__repr__() -> str

Developer-friendly representation of the DeliveryPeriod object.

RETURNS	DESCRIPTION
str	Developer-friendly representation of the DeliveryPeriod object.

Source code in frequenz/client/electricity_trading/_types.py

def __repr__(self) -> str:
    """
    Developer-friendly representation of the DeliveryPeriod object.

    Returns:
        Developer-friendly representation of the DeliveryPeriod object.
    """
    return self.__str__()

__str__

__str__() -> str

Return string representation of the DeliveryPeriod object.

RETURNS	DESCRIPTION
str	String representation of the DeliveryPeriod object.

Source code in frequenz/client/electricity_trading/_types.py

def __str__(self) -> str:
    """
    Return string representation of the DeliveryPeriod object.

    Returns:
        String representation of the DeliveryPeriod object.
    """
    duration_map = {
        DeliveryDuration.MINUTES_5: "5min",
        DeliveryDuration.MINUTES_15: "15min",
        DeliveryDuration.MINUTES_30: "30min",
        DeliveryDuration.MINUTES_60: "60min",
    }
    duration_str = duration_map.get(self.duration, "Unknown duration")
    start_str = self.start.strftime("%Y-%m-%d %H:%M:%S %Z")
    return f"DeliveryPeriod(start={start_str}, duration={duration_str})"

from_pb classmethod

from_pb(delivery_period: DeliveryPeriod) -> Self

Convert a protobuf DeliveryPeriod to DeliveryPeriod object.

PARAMETER	DESCRIPTION
delivery_period	DeliveryPeriod to convert. TYPE: DeliveryPeriod

RETURNS	DESCRIPTION
Self	DeliveryPeriod object corresponding to the protobuf message.

RAISES	DESCRIPTION
ValueError	If the duration is not 5, 15, 30, or 60 minutes.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, delivery_period: delivery_duration_pb2.DeliveryPeriod) -> Self:
    """Convert a protobuf DeliveryPeriod to DeliveryPeriod object.

    Args:
        delivery_period: DeliveryPeriod to convert.

    Returns:
        DeliveryPeriod object corresponding to the protobuf message.

    Raises:
        ValueError: If the duration is not 5, 15, 30, or 60 minutes.
    """
    start = delivery_period.start.ToDatetime(tzinfo=timezone.utc)
    delivery_duration_enum = DeliveryDuration.from_pb(delivery_period.duration)

    match delivery_duration_enum:
        case DeliveryDuration.MINUTES_5:
            duration = timedelta(minutes=5)
        case DeliveryDuration.MINUTES_15:
            duration = timedelta(minutes=15)
        case DeliveryDuration.MINUTES_30:
            duration = timedelta(minutes=30)
        case DeliveryDuration.MINUTES_60:
            duration = timedelta(minutes=60)
        case _:
            raise ValueError(
                "Invalid duration value. Duration must be 5, 15, 30, or 60 minutes."
            )
    return cls(start=start, duration=duration)

to_pb

to_pb() -> DeliveryPeriod

Convert a DeliveryPeriod object to protobuf DeliveryPeriod.

RETURNS	DESCRIPTION
DeliveryPeriod	Protobuf message corresponding to the DeliveryPeriod object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> delivery_duration_pb2.DeliveryPeriod:
    """Convert a DeliveryPeriod object to protobuf DeliveryPeriod.

    Returns:
        Protobuf message corresponding to the DeliveryPeriod object.
    """
    start = timestamp_pb2.Timestamp()
    start.FromDatetime(self.start)
    return delivery_duration_pb2.DeliveryPeriod(
        start=start,
        duration=self.duration.to_pb(),
    )

frequenz.client.electricity_trading.EnergyMarketCodeType

Bases: Enum

Specifies the type of identification code used in the energy market.

This is used for uniquely identifying various entities such as delivery areas, market participants, and grid components. This enumeration aims to offer compatibility across different jurisdictional standards.

Source code in frequenz/client/electricity_trading/_types.py

class EnergyMarketCodeType(enum.Enum):
    """
    Specifies the type of identification code used in the energy market.

    This is used for uniquely identifying various entities such as delivery areas, market
    participants, and grid components. This enumeration aims to offer compatibility across
    different jurisdictional standards.
    """

    UNSPECIFIED = (
        delivery_area_pb2.EnergyMarketCodeType.ENERGY_MARKET_CODE_TYPE_UNSPECIFIED
    )
    """Unspecified type. This value is a placeholder and should not be used."""

    EUROPE_EIC = (
        delivery_area_pb2.EnergyMarketCodeType.ENERGY_MARKET_CODE_TYPE_EUROPE_EIC
    )
    """European Energy Identification Code Standard."""

    US_NERC = delivery_area_pb2.EnergyMarketCodeType.ENERGY_MARKET_CODE_TYPE_US_NERC
    """North American Electric Reliability Corporation identifiers."""

    @classmethod
    @from_pb
    def from_pb(
        cls, energy_market_code_type: delivery_area_pb2.EnergyMarketCodeType.ValueType
    ) -> "EnergyMarketCodeType":
        """Convert a protobuf EnergyMarketCodeType value to EnergyMarketCodeType enum.

        Args:
            energy_market_code_type: Energy market code type to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == energy_market_code_type for e in cls):
            _logger.warning(
                "Unknown energy market code type %s. Returning UNSPECIFIED.",
                energy_market_code_type,
            )
            return cls.UNSPECIFIED

        return cls(energy_market_code_type)

    def to_pb(self) -> delivery_area_pb2.EnergyMarketCodeType.ValueType:
        """Convert a EnergyMarketCodeType object to protobuf EnergyMarketCodeType.

        Returns:
            Protobuf message corresponding to the EnergyMarketCodeType object.
        """
        return delivery_area_pb2.EnergyMarketCodeType.ValueType(self.value)

Attributes

EUROPE_EIC class-attribute instance-attribute

EUROPE_EIC = ENERGY_MARKET_CODE_TYPE_EUROPE_EIC

European Energy Identification Code Standard.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = ENERGY_MARKET_CODE_TYPE_UNSPECIFIED

Unspecified type. This value is a placeholder and should not be used.

US_NERC class-attribute instance-attribute

US_NERC = ENERGY_MARKET_CODE_TYPE_US_NERC

North American Electric Reliability Corporation identifiers.

Functions

from_pb classmethod

from_pb(
    energy_market_code_type: ValueType,
) -> "EnergyMarketCodeType"

Convert a protobuf EnergyMarketCodeType value to EnergyMarketCodeType enum.

PARAMETER	DESCRIPTION
energy_market_code_type	Energy market code type to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'EnergyMarketCodeType'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, energy_market_code_type: delivery_area_pb2.EnergyMarketCodeType.ValueType
) -> "EnergyMarketCodeType":
    """Convert a protobuf EnergyMarketCodeType value to EnergyMarketCodeType enum.

    Args:
        energy_market_code_type: Energy market code type to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == energy_market_code_type for e in cls):
        _logger.warning(
            "Unknown energy market code type %s. Returning UNSPECIFIED.",
            energy_market_code_type,
        )
        return cls.UNSPECIFIED

    return cls(energy_market_code_type)

to_pb

to_pb() -> ValueType

Convert a EnergyMarketCodeType object to protobuf EnergyMarketCodeType.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the EnergyMarketCodeType object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> delivery_area_pb2.EnergyMarketCodeType.ValueType:
    """Convert a EnergyMarketCodeType object to protobuf EnergyMarketCodeType.

    Returns:
        Protobuf message corresponding to the EnergyMarketCodeType object.
    """
    return delivery_area_pb2.EnergyMarketCodeType.ValueType(self.value)

frequenz.client.electricity_trading.GridpoolOrderFilter dataclass

Parameters for filtering Gridpool orders.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass(frozen=True)
class GridpoolOrderFilter:
    """Parameters for filtering Gridpool orders."""

    order_states: list[OrderState] | None = None
    """List of order states to filter for."""

    side: MarketSide | None = None
    """Market side to filter for."""

    delivery_period: DeliveryPeriod | None = None
    """Delivery period to filter for."""

    delivery_area: DeliveryArea | None = None
    """Delivery area to filter for."""

    tag: str | None = None
    """Tag associated with the orders to be filtered."""

    def __eq__(self, other: object) -> bool:
        """
        Check if two GridpoolOrderFilter objects are equal.

        Args:
            other: GridpoolOrderFilter object to compare with.

        Returns:
            True if the two GridpoolOrderFilter objects are equal, False otherwise.
        """
        if not isinstance(other, GridpoolOrderFilter):
            return NotImplemented
        return (
            self.order_states == other.order_states
            and self.side == other.side
            and self.delivery_period == other.delivery_period
            and self.delivery_area == other.delivery_area
            and self.tag == other.tag
        )

    def __hash__(self) -> int:
        """
        Create hash of the GridpoolOrderFilter object.

        Returns:
            Hash of the GridpoolOrderFilter object.
        """
        return hash(
            (
                tuple(self.order_states) if self.order_states is not None else None,
                self.side,
                self.delivery_period,
                self.delivery_area,
                self.tag,
            )
        )

    @classmethod
    @from_pb
    def from_pb(
        cls, gridpool_order_filter: electricity_trading_pb2.GridpoolOrderFilter
    ) -> Self:
        """Convert a protobuf GridpoolOrderFilter to GridpoolOrderFilter object.

        Args:
            gridpool_order_filter: GridpoolOrderFilter to convert.

        Returns:
            GridpoolOrderFilter object corresponding to the protobuf message.
        """
        return cls(
            order_states=(
                [OrderState.from_pb(state) for state in gridpool_order_filter.states]
                if gridpool_order_filter.states
                else None
            ),
            side=(
                MarketSide.from_pb(gridpool_order_filter.side)
                if gridpool_order_filter.HasField("side")
                else None
            ),
            delivery_period=(
                DeliveryPeriod.from_pb(gridpool_order_filter.delivery_period)
                if gridpool_order_filter.HasField("delivery_period")
                else None
            ),
            delivery_area=(
                DeliveryArea.from_pb(gridpool_order_filter.delivery_area)
                if gridpool_order_filter.HasField("delivery_area")
                else None
            ),
            tag=(
                gridpool_order_filter.tag
                if gridpool_order_filter.HasField("tag")
                else None
            ),
        )

    def to_pb(self) -> electricity_trading_pb2.GridpoolOrderFilter:
        """Convert a GridpoolOrderFilter object to protobuf GridpoolOrderFilter.

        Returns:
            Protobuf GridpoolOrderFilter corresponding to the object.
        """
        return electricity_trading_pb2.GridpoolOrderFilter(
            states=(
                [
                    electricity_trading_pb2.OrderState.ValueType(state.value)
                    for state in self.order_states
                ]
                if self.order_states
                else None
            ),
            side=(
                electricity_trading_pb2.MarketSide.ValueType(self.side.value)
                if self.side
                else None
            ),
            delivery_period=(
                self.delivery_period.to_pb() if self.delivery_period else None
            ),
            delivery_area=self.delivery_area.to_pb() if self.delivery_area else None,
            tag=self.tag if self.tag else None,
        )

Attributes

delivery_area class-attribute instance-attribute

delivery_area: DeliveryArea | None = None

Delivery area to filter for.

delivery_period class-attribute instance-attribute

delivery_period: DeliveryPeriod | None = None

Delivery period to filter for.

order_states class-attribute instance-attribute

order_states: list[OrderState] | None = None

List of order states to filter for.

side class-attribute instance-attribute

side: MarketSide | None = None

Market side to filter for.

tag class-attribute instance-attribute

tag: str | None = None

Tag associated with the orders to be filtered.

Functions

__eq__

__eq__(other: object) -> bool

Check if two GridpoolOrderFilter objects are equal.

PARAMETER	DESCRIPTION
other	GridpoolOrderFilter object to compare with. TYPE: object

RETURNS	DESCRIPTION
bool	True if the two GridpoolOrderFilter objects are equal, False otherwise.

Source code in frequenz/client/electricity_trading/_types.py

def __eq__(self, other: object) -> bool:
    """
    Check if two GridpoolOrderFilter objects are equal.

    Args:
        other: GridpoolOrderFilter object to compare with.

    Returns:
        True if the two GridpoolOrderFilter objects are equal, False otherwise.
    """
    if not isinstance(other, GridpoolOrderFilter):
        return NotImplemented
    return (
        self.order_states == other.order_states
        and self.side == other.side
        and self.delivery_period == other.delivery_period
        and self.delivery_area == other.delivery_area
        and self.tag == other.tag
    )

__hash__

__hash__() -> int

Create hash of the GridpoolOrderFilter object.

RETURNS	DESCRIPTION
int	Hash of the GridpoolOrderFilter object.

Source code in frequenz/client/electricity_trading/_types.py

def __hash__(self) -> int:
    """
    Create hash of the GridpoolOrderFilter object.

    Returns:
        Hash of the GridpoolOrderFilter object.
    """
    return hash(
        (
            tuple(self.order_states) if self.order_states is not None else None,
            self.side,
            self.delivery_period,
            self.delivery_area,
            self.tag,
        )
    )

from_pb classmethod

from_pb(gridpool_order_filter: GridpoolOrderFilter) -> Self

Convert a protobuf GridpoolOrderFilter to GridpoolOrderFilter object.

PARAMETER	DESCRIPTION
gridpool_order_filter	GridpoolOrderFilter to convert. TYPE: GridpoolOrderFilter

RETURNS	DESCRIPTION
Self	GridpoolOrderFilter object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, gridpool_order_filter: electricity_trading_pb2.GridpoolOrderFilter
) -> Self:
    """Convert a protobuf GridpoolOrderFilter to GridpoolOrderFilter object.

    Args:
        gridpool_order_filter: GridpoolOrderFilter to convert.

    Returns:
        GridpoolOrderFilter object corresponding to the protobuf message.
    """
    return cls(
        order_states=(
            [OrderState.from_pb(state) for state in gridpool_order_filter.states]
            if gridpool_order_filter.states
            else None
        ),
        side=(
            MarketSide.from_pb(gridpool_order_filter.side)
            if gridpool_order_filter.HasField("side")
            else None
        ),
        delivery_period=(
            DeliveryPeriod.from_pb(gridpool_order_filter.delivery_period)
            if gridpool_order_filter.HasField("delivery_period")
            else None
        ),
        delivery_area=(
            DeliveryArea.from_pb(gridpool_order_filter.delivery_area)
            if gridpool_order_filter.HasField("delivery_area")
            else None
        ),
        tag=(
            gridpool_order_filter.tag
            if gridpool_order_filter.HasField("tag")
            else None
        ),
    )

to_pb

to_pb() -> GridpoolOrderFilter

Convert a GridpoolOrderFilter object to protobuf GridpoolOrderFilter.

RETURNS	DESCRIPTION
GridpoolOrderFilter	Protobuf GridpoolOrderFilter corresponding to the object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.GridpoolOrderFilter:
    """Convert a GridpoolOrderFilter object to protobuf GridpoolOrderFilter.

    Returns:
        Protobuf GridpoolOrderFilter corresponding to the object.
    """
    return electricity_trading_pb2.GridpoolOrderFilter(
        states=(
            [
                electricity_trading_pb2.OrderState.ValueType(state.value)
                for state in self.order_states
            ]
            if self.order_states
            else None
        ),
        side=(
            electricity_trading_pb2.MarketSide.ValueType(self.side.value)
            if self.side
            else None
        ),
        delivery_period=(
            self.delivery_period.to_pb() if self.delivery_period else None
        ),
        delivery_area=self.delivery_area.to_pb() if self.delivery_area else None,
        tag=self.tag if self.tag else None,
    )

frequenz.client.electricity_trading.GridpoolTradeFilter dataclass

Parameters for filtering Gridpool trades.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass(frozen=True)
class GridpoolTradeFilter:
    """Parameters for filtering Gridpool trades."""

    trade_states: list[TradeState] | None = None
    """List of trade states to filter for."""

    trade_ids: list[int] | None = None
    """List of trade ids to filter for."""

    side: MarketSide | None = None
    """Market side to filter for."""

    delivery_period: DeliveryPeriod | None = None
    """Delivery period to filter for."""

    delivery_area: DeliveryArea | None = None
    """Delivery area to filter for."""

    def __eq__(self, other: object) -> bool:
        """
        Check if two GridpoolTradeFilter objects are equal.

        Args:
            other: GridpoolTradeFilter object to compare with.

        Returns:
            True if the two GridpoolTradeFilter objects are equal, False otherwise.
        """
        if not isinstance(other, GridpoolTradeFilter):
            return NotImplemented
        return (
            self.trade_states == other.trade_states
            and self.trade_ids == other.trade_ids
            and self.side == other.side
            and self.delivery_period == other.delivery_period
            and self.delivery_area == other.delivery_area
        )

    def __hash__(self) -> int:
        """
        Create hash of the GridpoolTradeFilter object.

        Returns:
            Hash of the GridpoolTradeFilter object.
        """
        return hash(
            (
                tuple(self.trade_states) if self.trade_states is not None else None,
                tuple(self.trade_ids) if self.trade_ids is not None else None,
                self.side,
                self.delivery_period,
                self.delivery_area,
            )
        )

    @classmethod
    @from_pb
    def from_pb(
        cls, gridpool_trade_filter: electricity_trading_pb2.GridpoolTradeFilter
    ) -> "GridpoolTradeFilter":
        """Convert a protobuf GridpoolTradeFilter to GridpoolTradeFilter object.

        Args:
            gridpool_trade_filter: GridpoolTradeFilter to convert.

        Returns:
            GridpoolTradeFilter object corresponding to the protobuf message.
        """
        return cls(
            trade_states=(
                [TradeState.from_pb(state) for state in gridpool_trade_filter.states]
                if gridpool_trade_filter.states
                else None
            ),
            trade_ids=(
                list(gridpool_trade_filter.trade_ids)
                if gridpool_trade_filter.trade_ids
                else None
            ),
            side=(
                MarketSide.from_pb(gridpool_trade_filter.side)
                if gridpool_trade_filter.HasField("side")
                else None
            ),
            delivery_period=(
                DeliveryPeriod.from_pb(gridpool_trade_filter.delivery_period)
                if gridpool_trade_filter.HasField("delivery_period")
                else None
            ),
            delivery_area=(
                DeliveryArea.from_pb(gridpool_trade_filter.delivery_area)
                if gridpool_trade_filter.HasField("delivery_area")
                else None
            ),
        )

    def to_pb(self) -> electricity_trading_pb2.GridpoolTradeFilter:
        """
        Convert a GridpoolTradeFilter object to protobuf GridpoolTradeFilter.

        Returns:
            Protobuf GridpoolTradeFilter corresponding to the object.
        """
        return electricity_trading_pb2.GridpoolTradeFilter(
            states=(
                [TradeState.to_pb(state) for state in self.trade_states]
                if self.trade_states
                else None
            ),
            trade_ids=self.trade_ids if self.trade_ids else None,
            side=MarketSide.to_pb(self.side) if self.side else None,
            delivery_period=(
                self.delivery_period.to_pb() if self.delivery_period else None
            ),
            delivery_area=self.delivery_area.to_pb() if self.delivery_area else None,
        )

Attributes

delivery_area class-attribute instance-attribute

delivery_area: DeliveryArea | None = None

Delivery area to filter for.

delivery_period class-attribute instance-attribute

delivery_period: DeliveryPeriod | None = None

Delivery period to filter for.

side class-attribute instance-attribute

side: MarketSide | None = None

Market side to filter for.

trade_ids class-attribute instance-attribute

trade_ids: list[int] | None = None

List of trade ids to filter for.

trade_states class-attribute instance-attribute

trade_states: list[TradeState] | None = None

List of trade states to filter for.

Functions

__eq__

__eq__(other: object) -> bool

Check if two GridpoolTradeFilter objects are equal.

PARAMETER	DESCRIPTION
other	GridpoolTradeFilter object to compare with. TYPE: object

RETURNS	DESCRIPTION
bool	True if the two GridpoolTradeFilter objects are equal, False otherwise.

Source code in frequenz/client/electricity_trading/_types.py

def __eq__(self, other: object) -> bool:
    """
    Check if two GridpoolTradeFilter objects are equal.

    Args:
        other: GridpoolTradeFilter object to compare with.

    Returns:
        True if the two GridpoolTradeFilter objects are equal, False otherwise.
    """
    if not isinstance(other, GridpoolTradeFilter):
        return NotImplemented
    return (
        self.trade_states == other.trade_states
        and self.trade_ids == other.trade_ids
        and self.side == other.side
        and self.delivery_period == other.delivery_period
        and self.delivery_area == other.delivery_area
    )

__hash__

__hash__() -> int

Create hash of the GridpoolTradeFilter object.

RETURNS	DESCRIPTION
int	Hash of the GridpoolTradeFilter object.

Source code in frequenz/client/electricity_trading/_types.py

def __hash__(self) -> int:
    """
    Create hash of the GridpoolTradeFilter object.

    Returns:
        Hash of the GridpoolTradeFilter object.
    """
    return hash(
        (
            tuple(self.trade_states) if self.trade_states is not None else None,
            tuple(self.trade_ids) if self.trade_ids is not None else None,
            self.side,
            self.delivery_period,
            self.delivery_area,
        )
    )

from_pb classmethod

from_pb(
    gridpool_trade_filter: GridpoolTradeFilter,
) -> "GridpoolTradeFilter"

Convert a protobuf GridpoolTradeFilter to GridpoolTradeFilter object.

PARAMETER	DESCRIPTION
gridpool_trade_filter	GridpoolTradeFilter to convert. TYPE: GridpoolTradeFilter

RETURNS	DESCRIPTION
'GridpoolTradeFilter'	GridpoolTradeFilter object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, gridpool_trade_filter: electricity_trading_pb2.GridpoolTradeFilter
) -> "GridpoolTradeFilter":
    """Convert a protobuf GridpoolTradeFilter to GridpoolTradeFilter object.

    Args:
        gridpool_trade_filter: GridpoolTradeFilter to convert.

    Returns:
        GridpoolTradeFilter object corresponding to the protobuf message.
    """
    return cls(
        trade_states=(
            [TradeState.from_pb(state) for state in gridpool_trade_filter.states]
            if gridpool_trade_filter.states
            else None
        ),
        trade_ids=(
            list(gridpool_trade_filter.trade_ids)
            if gridpool_trade_filter.trade_ids
            else None
        ),
        side=(
            MarketSide.from_pb(gridpool_trade_filter.side)
            if gridpool_trade_filter.HasField("side")
            else None
        ),
        delivery_period=(
            DeliveryPeriod.from_pb(gridpool_trade_filter.delivery_period)
            if gridpool_trade_filter.HasField("delivery_period")
            else None
        ),
        delivery_area=(
            DeliveryArea.from_pb(gridpool_trade_filter.delivery_area)
            if gridpool_trade_filter.HasField("delivery_area")
            else None
        ),
    )

to_pb

to_pb() -> GridpoolTradeFilter

Convert a GridpoolTradeFilter object to protobuf GridpoolTradeFilter.

RETURNS	DESCRIPTION
GridpoolTradeFilter	Protobuf GridpoolTradeFilter corresponding to the object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.GridpoolTradeFilter:
    """
    Convert a GridpoolTradeFilter object to protobuf GridpoolTradeFilter.

    Returns:
        Protobuf GridpoolTradeFilter corresponding to the object.
    """
    return electricity_trading_pb2.GridpoolTradeFilter(
        states=(
            [TradeState.to_pb(state) for state in self.trade_states]
            if self.trade_states
            else None
        ),
        trade_ids=self.trade_ids if self.trade_ids else None,
        side=MarketSide.to_pb(self.side) if self.side else None,
        delivery_period=(
            self.delivery_period.to_pb() if self.delivery_period else None
        ),
        delivery_area=self.delivery_area.to_pb() if self.delivery_area else None,
    )

frequenz.client.electricity_trading.MarketActor

Bases: Enum

Actors responsible for an order state change.

Source code in frequenz/client/electricity_trading/_types.py

class MarketActor(enum.Enum):
    """Actors responsible for an order state change."""

    UNSPECIFIED = (
        electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.MARKET_ACTOR_UNSPECIFIED
    )
    """The actor responsible for the state change has not been specified."""

    USER = electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.MARKET_ACTOR_USER
    """The user was the actor."""

    MARKET_OPERATOR = (
        electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.MARKET_ACTOR_MARKET_OPERATOR
    )
    """The market operator was the actor."""

    SYSTEM = (
        electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.MARKET_ACTOR_SYSTEM
    )
    """The system was the actor."""

    @classmethod
    @from_pb
    def from_pb(
        cls,
        market_actor: electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.ValueType,
    ) -> "MarketActor":
        """Convert a protobuf MarketActor value to MarketActor enum.

        Args:
            market_actor: Market actor to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == market_actor for e in cls):
            _logger.warning(
                "Unknown market actor %s. Returning UNSPECIFIED.", market_actor
            )
            return cls.UNSPECIFIED

        return cls(market_actor)

    def to_pb(
        self,
    ) -> electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.ValueType:
        """Convert a MarketActor enum to protobuf MarketActor value.

        Returns:
            Protobuf message corresponding to the MarketActor enum.
        """
        return self.value

Attributes

MARKET_OPERATOR class-attribute instance-attribute

MARKET_OPERATOR = MARKET_ACTOR_MARKET_OPERATOR

The market operator was the actor.

SYSTEM class-attribute instance-attribute

SYSTEM = MARKET_ACTOR_SYSTEM

The system was the actor.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = MARKET_ACTOR_UNSPECIFIED

The actor responsible for the state change has not been specified.

USER class-attribute instance-attribute

USER = MARKET_ACTOR_USER

The user was the actor.

Functions

from_pb classmethod

from_pb(market_actor: ValueType) -> 'MarketActor'

Convert a protobuf MarketActor value to MarketActor enum.

PARAMETER	DESCRIPTION
market_actor	Market actor to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'MarketActor'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls,
    market_actor: electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.ValueType,
) -> "MarketActor":
    """Convert a protobuf MarketActor value to MarketActor enum.

    Args:
        market_actor: Market actor to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == market_actor for e in cls):
        _logger.warning(
            "Unknown market actor %s. Returning UNSPECIFIED.", market_actor
        )
        return cls.UNSPECIFIED

    return cls(market_actor)

to_pb

to_pb() -> ValueType

Convert a MarketActor enum to protobuf MarketActor value.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the MarketActor enum.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(
    self,
) -> electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.ValueType:
    """Convert a MarketActor enum to protobuf MarketActor value.

    Returns:
        Protobuf message corresponding to the MarketActor enum.
    """
    return self.value

frequenz.client.electricity_trading.MarketSide

Bases: Enum

Which side of the market the order is on, either buying or selling.

Source code in frequenz/client/electricity_trading/_types.py

class MarketSide(enum.Enum):
    """Which side of the market the order is on, either buying or selling."""

    UNSPECIFIED = electricity_trading_pb2.MarketSide.MARKET_SIDE_UNSPECIFIED
    """The side of the market has not been set."""

    BUY = electricity_trading_pb2.MarketSide.MARKET_SIDE_BUY
    """Order to purchase electricity, referred to as a 'bid' in the order book."""

    SELL = electricity_trading_pb2.MarketSide.MARKET_SIDE_SELL
    """Order to sell electricity, referred to as an 'ask' or 'offer' in the order book."""

    @classmethod
    @from_pb
    def from_pb(
        cls, market_side: electricity_trading_pb2.MarketSide.ValueType
    ) -> "MarketSide":
        """Convert a protobuf MarketSide value to MarketSide enum.

        Args:
            market_side: Market side to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == market_side for e in cls):
            _logger.warning(
                "Unknown market side %s. Returning UNSPECIFIED.", market_side
            )
            return cls.UNSPECIFIED

        return cls(market_side)

    def to_pb(self) -> electricity_trading_pb2.MarketSide.ValueType:
        """Convert a MarketSide enum to protobuf MarketSide value.

        Returns:
            Protobuf message corresponding to the MarketSide enum.
        """
        return self.value

Attributes

BUY class-attribute instance-attribute

BUY = MARKET_SIDE_BUY

Order to purchase electricity, referred to as a 'bid' in the order book.

SELL class-attribute instance-attribute

SELL = MARKET_SIDE_SELL

Order to sell electricity, referred to as an 'ask' or 'offer' in the order book.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = MARKET_SIDE_UNSPECIFIED

The side of the market has not been set.

Functions

from_pb classmethod

from_pb(market_side: ValueType) -> 'MarketSide'

Convert a protobuf MarketSide value to MarketSide enum.

PARAMETER	DESCRIPTION
market_side	Market side to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'MarketSide'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, market_side: electricity_trading_pb2.MarketSide.ValueType
) -> "MarketSide":
    """Convert a protobuf MarketSide value to MarketSide enum.

    Args:
        market_side: Market side to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == market_side for e in cls):
        _logger.warning(
            "Unknown market side %s. Returning UNSPECIFIED.", market_side
        )
        return cls.UNSPECIFIED

    return cls(market_side)

to_pb

to_pb() -> ValueType

Convert a MarketSide enum to protobuf MarketSide value.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the MarketSide enum.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.MarketSide.ValueType:
    """Convert a MarketSide enum to protobuf MarketSide value.

    Returns:
        Protobuf message corresponding to the MarketSide enum.
    """
    return self.value

frequenz.client.electricity_trading.Order dataclass

Represents an order in the electricity market.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass()
class Order:  # pylint: disable=too-many-instance-attributes
    """Represents an order in the electricity market."""

    delivery_area: DeliveryArea
    """The delivery area where the contract is to be delivered."""

    delivery_period: DeliveryPeriod
    """The delivery period for the contract."""

    type: OrderType
    """The type of order."""

    side: MarketSide
    """Indicates if the order is on the Buy or Sell side of the market."""

    price: Price
    """The limit price at which the contract is to be traded."""

    quantity: Power
    """The quantity of the contract being traded."""

    stop_price: Price | None = None
    """Applicable for STOP_LIMIT orders. The stop price that triggers the limit order."""

    peak_price_delta: Price | None = None
    """Applicable for ICEBERG orders. The price difference between the peak price and
    the limit price."""

    display_quantity: Power | None = None
    """Applicable for ICEBERG orders. The quantity of the order to be displayed in the order
    book."""

    execution_option: OrderExecutionOption | None = None
    """Order execution options such as All or None, Fill or Kill, etc."""

    valid_until: datetime | None = None
    """UTC timestamp defining the time after which the order should be cancelled if not filled."""

    payload: dict[str, struct_pb2.Value] | None = None
    """User-defined payload individual to a specific order. This can be any data that needs to be
    associated with the order."""

    tag: str | None = None
    """User-defined tag to group related orders."""

    def __post_init__(self) -> None:
        """Post initialization checks to ensure that all datetimes are UTC."""
        if self.valid_until is not None:
            if self.valid_until.tzinfo is None:
                raise ValueError("Valid until must be a UTC datetime.")
            if self.valid_until.tzinfo != timezone.utc:
                _logger.warning("Valid until is not a UTC datetime. Converting to UTC.")
                self.valid_until = self.valid_until.astimezone(timezone.utc)

    @classmethod
    @from_pb
    def from_pb(cls, order: electricity_trading_pb2.Order) -> Self:
        """Convert a protobuf Order to Order object.

        Args:
            order: Order to convert.

        Returns:
            Order object corresponding to the protobuf message.
        """
        return cls(
            delivery_area=DeliveryArea.from_pb(order.delivery_area),
            delivery_period=DeliveryPeriod.from_pb(order.delivery_period),
            type=OrderType.from_pb(order.type),
            side=MarketSide.from_pb(order.side),
            price=(
                Price.from_pb(order.price)
                if order.HasField("price")
                else Price(Decimal("NaN"), Currency.UNSPECIFIED)
            ),
            quantity=(
                Power.from_pb(order.quantity)
                if order.HasField("quantity")
                else Power(Decimal("NaN"))
            ),
            stop_price=(
                Price.from_pb(order.stop_price)
                if order.HasField("stop_price")
                else None
            ),
            peak_price_delta=(
                Price.from_pb(order.peak_price_delta)
                if order.HasField("peak_price_delta")
                else None
            ),
            display_quantity=(
                Power.from_pb(order.display_quantity)
                if order.HasField("display_quantity")
                else None
            ),
            execution_option=(
                OrderExecutionOption.from_pb(order.execution_option)
                if order.HasField("execution_option")
                else None
            ),
            valid_until=(
                order.valid_until.ToDatetime(tzinfo=timezone.utc)
                if order.HasField("valid_until")
                else None
            ),
            payload=json_format.MessageToDict(order.payload) if order.payload else None,
            tag=order.tag if order.tag else None,
        )

    def to_pb(self) -> electricity_trading_pb2.Order:
        """
        Convert an Order object to protobuf Order.

        Returns:
            Protobuf message corresponding to the Order object.
        """
        if self.valid_until:
            valid_until = timestamp_pb2.Timestamp()
            valid_until.FromDatetime(self.valid_until)
        else:
            valid_until = None
        return electricity_trading_pb2.Order(
            delivery_area=self.delivery_area.to_pb(),
            delivery_period=self.delivery_period.to_pb(),
            type=electricity_trading_pb2.OrderType.ValueType(self.type.value),
            side=electricity_trading_pb2.MarketSide.ValueType(self.side.value),
            price=self.price.to_pb(),
            quantity=self.quantity.to_pb(),
            stop_price=self.stop_price.to_pb() if self.stop_price else None,
            peak_price_delta=(
                self.peak_price_delta.to_pb() if self.peak_price_delta else None
            ),
            display_quantity=(
                self.display_quantity.to_pb() if self.display_quantity else None
            ),
            execution_option=(
                electricity_trading_pb2.OrderExecutionOption.ValueType(
                    self.execution_option.value
                )
                if self.execution_option
                else None
            ),
            valid_until=valid_until,
            payload=struct_pb2.Struct(fields=self.payload) if self.payload else None,
            tag=self.tag if self.tag else None,
        )

    def __eq__(self, other: object) -> bool:
        """
        Check if two orders are equal.

        Args:
            other: The other order to compare to.

        Returns:
            True if the orders are equal, False otherwise.
        """
        if not isinstance(other, Order):
            return NotImplemented
        return (
            self.delivery_area == other.delivery_area
            and self.delivery_period.start == other.delivery_period.start
            and self.delivery_period.duration == other.delivery_period.duration
            and self.type == other.type
            and self.side == other.side
            and self.price == other.price
            and self.quantity == other.quantity
            and self.stop_price == other.stop_price
            and self.peak_price_delta == other.peak_price_delta
            and self.display_quantity == other.display_quantity
            and (
                self.execution_option == other.execution_option
                or (
                    self.execution_option is None
                    and other.execution_option == OrderExecutionOption.UNSPECIFIED
                )
                or (
                    self.execution_option == OrderExecutionOption.UNSPECIFIED
                    and other.execution_option is None
                )
            )
            and self.valid_until == other.valid_until
            and self.payload == other.payload
            and self.tag == other.tag
        )

Attributes

delivery_area instance-attribute

delivery_area: DeliveryArea

The delivery area where the contract is to be delivered.

delivery_period instance-attribute

delivery_period: DeliveryPeriod

The delivery period for the contract.

display_quantity class-attribute instance-attribute

display_quantity: Power | None = None

Applicable for ICEBERG orders. The quantity of the order to be displayed in the order book.

execution_option class-attribute instance-attribute

execution_option: OrderExecutionOption | None = None

Order execution options such as All or None, Fill or Kill, etc.

payload class-attribute instance-attribute

payload: dict[str, Value] | None = None

User-defined payload individual to a specific order. This can be any data that needs to be associated with the order.

peak_price_delta class-attribute instance-attribute

peak_price_delta: Price | None = None

Applicable for ICEBERG orders. The price difference between the peak price and the limit price.

price instance-attribute

price: Price

The limit price at which the contract is to be traded.

quantity instance-attribute

quantity: Power

The quantity of the contract being traded.

side instance-attribute

side: MarketSide

Indicates if the order is on the Buy or Sell side of the market.

stop_price class-attribute instance-attribute

stop_price: Price | None = None

Applicable for STOP_LIMIT orders. The stop price that triggers the limit order.

tag class-attribute instance-attribute

tag: str | None = None

User-defined tag to group related orders.

type instance-attribute

type: OrderType

The type of order.

valid_until class-attribute instance-attribute

valid_until: datetime | None = None

UTC timestamp defining the time after which the order should be cancelled if not filled.

Functions

__eq__

__eq__(other: object) -> bool

Check if two orders are equal.

PARAMETER	DESCRIPTION
other	The other order to compare to. TYPE: object

RETURNS	DESCRIPTION
bool	True if the orders are equal, False otherwise.

Source code in frequenz/client/electricity_trading/_types.py

def __eq__(self, other: object) -> bool:
    """
    Check if two orders are equal.

    Args:
        other: The other order to compare to.

    Returns:
        True if the orders are equal, False otherwise.
    """
    if not isinstance(other, Order):
        return NotImplemented
    return (
        self.delivery_area == other.delivery_area
        and self.delivery_period.start == other.delivery_period.start
        and self.delivery_period.duration == other.delivery_period.duration
        and self.type == other.type
        and self.side == other.side
        and self.price == other.price
        and self.quantity == other.quantity
        and self.stop_price == other.stop_price
        and self.peak_price_delta == other.peak_price_delta
        and self.display_quantity == other.display_quantity
        and (
            self.execution_option == other.execution_option
            or (
                self.execution_option is None
                and other.execution_option == OrderExecutionOption.UNSPECIFIED
            )
            or (
                self.execution_option == OrderExecutionOption.UNSPECIFIED
                and other.execution_option is None
            )
        )
        and self.valid_until == other.valid_until
        and self.payload == other.payload
        and self.tag == other.tag
    )

__post_init__

__post_init__() -> None

Post initialization checks to ensure that all datetimes are UTC.

Source code in frequenz/client/electricity_trading/_types.py

def __post_init__(self) -> None:
    """Post initialization checks to ensure that all datetimes are UTC."""
    if self.valid_until is not None:
        if self.valid_until.tzinfo is None:
            raise ValueError("Valid until must be a UTC datetime.")
        if self.valid_until.tzinfo != timezone.utc:
            _logger.warning("Valid until is not a UTC datetime. Converting to UTC.")
            self.valid_until = self.valid_until.astimezone(timezone.utc)

from_pb classmethod

from_pb(order: Order) -> Self

Convert a protobuf Order to Order object.

PARAMETER	DESCRIPTION
order	Order to convert. TYPE: Order

RETURNS	DESCRIPTION
Self	Order object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, order: electricity_trading_pb2.Order) -> Self:
    """Convert a protobuf Order to Order object.

    Args:
        order: Order to convert.

    Returns:
        Order object corresponding to the protobuf message.
    """
    return cls(
        delivery_area=DeliveryArea.from_pb(order.delivery_area),
        delivery_period=DeliveryPeriod.from_pb(order.delivery_period),
        type=OrderType.from_pb(order.type),
        side=MarketSide.from_pb(order.side),
        price=(
            Price.from_pb(order.price)
            if order.HasField("price")
            else Price(Decimal("NaN"), Currency.UNSPECIFIED)
        ),
        quantity=(
            Power.from_pb(order.quantity)
            if order.HasField("quantity")
            else Power(Decimal("NaN"))
        ),
        stop_price=(
            Price.from_pb(order.stop_price)
            if order.HasField("stop_price")
            else None
        ),
        peak_price_delta=(
            Price.from_pb(order.peak_price_delta)
            if order.HasField("peak_price_delta")
            else None
        ),
        display_quantity=(
            Power.from_pb(order.display_quantity)
            if order.HasField("display_quantity")
            else None
        ),
        execution_option=(
            OrderExecutionOption.from_pb(order.execution_option)
            if order.HasField("execution_option")
            else None
        ),
        valid_until=(
            order.valid_until.ToDatetime(tzinfo=timezone.utc)
            if order.HasField("valid_until")
            else None
        ),
        payload=json_format.MessageToDict(order.payload) if order.payload else None,
        tag=order.tag if order.tag else None,
    )

to_pb

to_pb() -> Order

Convert an Order object to protobuf Order.

RETURNS	DESCRIPTION
Order	Protobuf message corresponding to the Order object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.Order:
    """
    Convert an Order object to protobuf Order.

    Returns:
        Protobuf message corresponding to the Order object.
    """
    if self.valid_until:
        valid_until = timestamp_pb2.Timestamp()
        valid_until.FromDatetime(self.valid_until)
    else:
        valid_until = None
    return electricity_trading_pb2.Order(
        delivery_area=self.delivery_area.to_pb(),
        delivery_period=self.delivery_period.to_pb(),
        type=electricity_trading_pb2.OrderType.ValueType(self.type.value),
        side=electricity_trading_pb2.MarketSide.ValueType(self.side.value),
        price=self.price.to_pb(),
        quantity=self.quantity.to_pb(),
        stop_price=self.stop_price.to_pb() if self.stop_price else None,
        peak_price_delta=(
            self.peak_price_delta.to_pb() if self.peak_price_delta else None
        ),
        display_quantity=(
            self.display_quantity.to_pb() if self.display_quantity else None
        ),
        execution_option=(
            electricity_trading_pb2.OrderExecutionOption.ValueType(
                self.execution_option.value
            )
            if self.execution_option
            else None
        ),
        valid_until=valid_until,
        payload=struct_pb2.Struct(fields=self.payload) if self.payload else None,
        tag=self.tag if self.tag else None,
    )

frequenz.client.electricity_trading.OrderDetail dataclass

Represents an order with full details, including its ID, state, and associated UTC timestamps.

ATTRIBUTE	DESCRIPTION
order_id	Unique identifier of the order. TYPE: int
order	The details of the order. TYPE: Order
state_detail	Details of the order's current state. TYPE: StateDetail
open_quantity	Remaining open quantity for this order. TYPE: Power
filled_quantity	Filled quantity for this order. TYPE: Power
create_time	UTC Timestamp when the order was created. TYPE: datetime
modification_time	UTC Timestamp of the last update to the order. TYPE: datetime

Source code in frequenz/client/electricity_trading/_types.py

@dataclass()
class OrderDetail:
    """
    Represents an order with full details, including its ID, state, and associated UTC timestamps.

    Attributes:
        order_id: Unique identifier of the order.
        order: The details of the order.
        state_detail: Details of the order's current state.
        open_quantity: Remaining open quantity for this order.
        filled_quantity: Filled quantity for this order.
        create_time: UTC Timestamp when the order was created.
        modification_time: UTC Timestamp of the last update to the order.
    """

    order_id: int
    order: Order
    state_detail: StateDetail
    open_quantity: Power
    filled_quantity: Power
    create_time: datetime
    modification_time: datetime

    def __post_init__(self) -> None:
        """
        Post initialization checks to ensure that all datetimes are UTC.

        Raises:
            ValueError: If create_time or modification_time do not have timezone information.

        """
        if self.create_time.tzinfo is None:
            raise ValueError("Create time must have timezone information")
        if self.create_time.tzinfo != timezone.utc:
            _logger.warning("Create time is not in UTC timezone. Converting to UTC.")
            self.create_time = self.create_time.astimezone(timezone.utc)

        if self.modification_time.tzinfo is None:
            raise ValueError("Modification time must have timezone information")
        if self.modification_time.tzinfo != timezone.utc:
            _logger.warning(
                "Modification time is not in UTC timezone. Converting to UTC."
            )
            self.modification_time = self.modification_time.astimezone(timezone.utc)

    @classmethod
    @from_pb
    def from_pb(cls, order_detail: electricity_trading_pb2.OrderDetail) -> Self:
        """Convert a protobuf OrderDetail to OrderDetail object.

        Args:
            order_detail: OrderDetail to convert.

        Returns:
            OrderDetail object corresponding to the protobuf message.

        Raises:
            ValueError: If the order price and quantity are not specified for a non-canceled order.
        """
        od = cls(
            order_id=order_detail.order_id,
            order=Order.from_pb(order_detail.order),
            state_detail=StateDetail.from_pb(order_detail.state_detail),
            open_quantity=Power.from_pb(order_detail.open_quantity),
            filled_quantity=Power.from_pb(order_detail.filled_quantity),
            create_time=order_detail.create_time.ToDatetime(tzinfo=timezone.utc),
            modification_time=order_detail.modification_time.ToDatetime(
                tzinfo=timezone.utc
            ),
        )

        # Only cancelled orders are allowed to have missing price or quantity
        missing_price_or_quantity = (
            od.order.price.amount.is_nan()
            or od.order.price.currency == Currency.UNSPECIFIED
            or od.order.quantity.mw.is_nan()
        )
        if missing_price_or_quantity and od.state_detail.state != OrderState.CANCELED:
            raise ValueError(
                f"Price and quantity must be specified for a non-canceled order (`{od}`)."
            )

        return od

    def to_pb(self) -> electricity_trading_pb2.OrderDetail:
        """Convert an OrderDetail object to protobuf OrderDetail.

        Returns:
            Protobuf message corresponding to the OrderDetail object.
        """
        create_time = timestamp_pb2.Timestamp()
        create_time.FromDatetime(self.create_time)
        modification_time = timestamp_pb2.Timestamp()
        modification_time.FromDatetime(self.modification_time)

        return electricity_trading_pb2.OrderDetail(
            order_id=self.order_id,
            order=self.order.to_pb(),
            state_detail=self.state_detail.to_pb(),
            open_quantity=self.open_quantity.to_pb(),
            filled_quantity=self.filled_quantity.to_pb(),
            create_time=create_time,
            modification_time=modification_time,
        )

Functions

__post_init__

__post_init__() -> None

Post initialization checks to ensure that all datetimes are UTC.

RAISES	DESCRIPTION
ValueError	If create_time or modification_time do not have timezone information.

Source code in frequenz/client/electricity_trading/_types.py

def __post_init__(self) -> None:
    """
    Post initialization checks to ensure that all datetimes are UTC.

    Raises:
        ValueError: If create_time or modification_time do not have timezone information.

    """
    if self.create_time.tzinfo is None:
        raise ValueError("Create time must have timezone information")
    if self.create_time.tzinfo != timezone.utc:
        _logger.warning("Create time is not in UTC timezone. Converting to UTC.")
        self.create_time = self.create_time.astimezone(timezone.utc)

    if self.modification_time.tzinfo is None:
        raise ValueError("Modification time must have timezone information")
    if self.modification_time.tzinfo != timezone.utc:
        _logger.warning(
            "Modification time is not in UTC timezone. Converting to UTC."
        )
        self.modification_time = self.modification_time.astimezone(timezone.utc)

from_pb classmethod

from_pb(order_detail: OrderDetail) -> Self

Convert a protobuf OrderDetail to OrderDetail object.

PARAMETER	DESCRIPTION
order_detail	OrderDetail to convert. TYPE: OrderDetail

RETURNS	DESCRIPTION
Self	OrderDetail object corresponding to the protobuf message.

RAISES	DESCRIPTION
ValueError	If the order price and quantity are not specified for a non-canceled order.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, order_detail: electricity_trading_pb2.OrderDetail) -> Self:
    """Convert a protobuf OrderDetail to OrderDetail object.

    Args:
        order_detail: OrderDetail to convert.

    Returns:
        OrderDetail object corresponding to the protobuf message.

    Raises:
        ValueError: If the order price and quantity are not specified for a non-canceled order.
    """
    od = cls(
        order_id=order_detail.order_id,
        order=Order.from_pb(order_detail.order),
        state_detail=StateDetail.from_pb(order_detail.state_detail),
        open_quantity=Power.from_pb(order_detail.open_quantity),
        filled_quantity=Power.from_pb(order_detail.filled_quantity),
        create_time=order_detail.create_time.ToDatetime(tzinfo=timezone.utc),
        modification_time=order_detail.modification_time.ToDatetime(
            tzinfo=timezone.utc
        ),
    )

    # Only cancelled orders are allowed to have missing price or quantity
    missing_price_or_quantity = (
        od.order.price.amount.is_nan()
        or od.order.price.currency == Currency.UNSPECIFIED
        or od.order.quantity.mw.is_nan()
    )
    if missing_price_or_quantity and od.state_detail.state != OrderState.CANCELED:
        raise ValueError(
            f"Price and quantity must be specified for a non-canceled order (`{od}`)."
        )

    return od

to_pb

to_pb() -> OrderDetail

Convert an OrderDetail object to protobuf OrderDetail.

RETURNS	DESCRIPTION
OrderDetail	Protobuf message corresponding to the OrderDetail object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.OrderDetail:
    """Convert an OrderDetail object to protobuf OrderDetail.

    Returns:
        Protobuf message corresponding to the OrderDetail object.
    """
    create_time = timestamp_pb2.Timestamp()
    create_time.FromDatetime(self.create_time)
    modification_time = timestamp_pb2.Timestamp()
    modification_time.FromDatetime(self.modification_time)

    return electricity_trading_pb2.OrderDetail(
        order_id=self.order_id,
        order=self.order.to_pb(),
        state_detail=self.state_detail.to_pb(),
        open_quantity=self.open_quantity.to_pb(),
        filled_quantity=self.filled_quantity.to_pb(),
        create_time=create_time,
        modification_time=modification_time,
    )

frequenz.client.electricity_trading.OrderExecutionOption

Bases: Enum

Specific behavior for the execution of an order.

These options provide control on how an order is handled in the market.

If no OrderExecutionOption is set, the order remains open until it's fully fulfilled, cancelled by the client, valid_until timestamp is reached, or the end of the trading session.

Source code in frequenz/client/electricity_trading/_types.py

class OrderExecutionOption(enum.Enum):
    """
    Specific behavior for the execution of an order.

    These options provide control on how an order is handled in the market.

    If no OrderExecutionOption is set, the order remains open until it's fully
    fulfilled, cancelled by the client, `valid_until` timestamp is reached, or
    the end of the trading session.
    """

    UNSPECIFIED = (
        electricity_trading_pb2.OrderExecutionOption.ORDER_EXECUTION_OPTION_UNSPECIFIED
    )
    """The order execution option has not been set."""

    AON = electricity_trading_pb2.OrderExecutionOption.ORDER_EXECUTION_OPTION_AON
    """All or None: Order must be executed in its entirety, or not executed at all."""

    FOK = electricity_trading_pb2.OrderExecutionOption.ORDER_EXECUTION_OPTION_FOK
    """Fill or Kill: Order must be executed immediately in its entirety, or not at all."""

    IOC = electricity_trading_pb2.OrderExecutionOption.ORDER_EXECUTION_OPTION_IOC
    """Immediate or Cancel: Any portion of an order that cannot be filled \
    immediately will be cancelled."""

    @classmethod
    @from_pb
    def from_pb(
        cls,
        order_execution_option: electricity_trading_pb2.OrderExecutionOption.ValueType,
    ) -> "OrderExecutionOption":
        """Convert a protobuf OrderExecutionOption value to OrderExecutionOption enum.

        Args:
            order_execution_option: order execution option to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == order_execution_option for e in OrderExecutionOption):
            _logger.warning(
                "Unknown forecast feature %s. Returning UNSPECIFIED.",
                order_execution_option,
            )
            return cls.UNSPECIFIED

        return OrderExecutionOption(order_execution_option)

    def to_pb(self) -> electricity_trading_pb2.OrderExecutionOption.ValueType:
        """Convert a OrderExecutionOption object to protobuf OrderExecutionOption.

        Returns:
            Protobuf message corresponding to the OrderExecutionOption object.
        """
        return electricity_trading_pb2.OrderExecutionOption.ValueType(self.value)

Attributes

AON class-attribute instance-attribute

AON = ORDER_EXECUTION_OPTION_AON

All or None: Order must be executed in its entirety, or not executed at all.

FOK class-attribute instance-attribute

FOK = ORDER_EXECUTION_OPTION_FOK

Fill or Kill: Order must be executed immediately in its entirety, or not at all.

IOC class-attribute instance-attribute

IOC = ORDER_EXECUTION_OPTION_IOC

Immediate or Cancel: Any portion of an order that cannot be filled immediately will be cancelled.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = ORDER_EXECUTION_OPTION_UNSPECIFIED

The order execution option has not been set.

Functions

from_pb classmethod

from_pb(
    order_execution_option: ValueType,
) -> "OrderExecutionOption"

Convert a protobuf OrderExecutionOption value to OrderExecutionOption enum.

PARAMETER	DESCRIPTION
order_execution_option	order execution option to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'OrderExecutionOption'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls,
    order_execution_option: electricity_trading_pb2.OrderExecutionOption.ValueType,
) -> "OrderExecutionOption":
    """Convert a protobuf OrderExecutionOption value to OrderExecutionOption enum.

    Args:
        order_execution_option: order execution option to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == order_execution_option for e in OrderExecutionOption):
        _logger.warning(
            "Unknown forecast feature %s. Returning UNSPECIFIED.",
            order_execution_option,
        )
        return cls.UNSPECIFIED

    return OrderExecutionOption(order_execution_option)

to_pb

to_pb() -> ValueType

Convert a OrderExecutionOption object to protobuf OrderExecutionOption.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the OrderExecutionOption object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.OrderExecutionOption.ValueType:
    """Convert a OrderExecutionOption object to protobuf OrderExecutionOption.

    Returns:
        Protobuf message corresponding to the OrderExecutionOption object.
    """
    return electricity_trading_pb2.OrderExecutionOption.ValueType(self.value)

frequenz.client.electricity_trading.OrderState

Bases: Enum

State of an order.

Source code in frequenz/client/electricity_trading/_types.py

class OrderState(enum.Enum):
    """State of an order."""

    UNSPECIFIED = electricity_trading_pb2.OrderState.ORDER_STATE_UNSPECIFIED
    """The order state is not known. Usually the default state of a newly created order object
    before any operations have been applied."""

    PENDING = electricity_trading_pb2.OrderState.ORDER_STATE_PENDING
    """The order has been sent to the marketplace but has not yet been confirmed. This can be due
    to awaiting validation or system processing."""

    ACTIVE = electricity_trading_pb2.OrderState.ORDER_STATE_ACTIVE
    """The order has been confirmed and is open in the market. It may be unfilled or partially
    filled."""

    FILLED = electricity_trading_pb2.OrderState.ORDER_STATE_FILLED
    """The order has been completely filled and there are no remaining quantities on the order."""

    CANCELED = electricity_trading_pb2.OrderState.ORDER_STATE_CANCELED
    """The order has been canceled. This can occur due to a cancellation request by the market
    participant, system, or market operator."""

    EXPIRED = electricity_trading_pb2.OrderState.ORDER_STATE_EXPIRED
    """The order has not been filled within the defined duration and has expired."""

    FAILED = electricity_trading_pb2.OrderState.ORDER_STATE_FAILED
    """The order submission failed and was unable to be placed on the order book, usually due to a
    validation error or system issue."""

    HIBERNATE = electricity_trading_pb2.OrderState.ORDER_STATE_HIBERNATE
    """The order has been entered into the system but is not currently exposed to the market. This
    could be due to certain conditions not yet being met."""

    @classmethod
    @from_pb
    def from_pb(
        cls, order_state: electricity_trading_pb2.OrderState.ValueType
    ) -> "OrderState":
        """Convert a protobuf OrderState value to OrderState enum.

        Args:
            order_state: Order state to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == order_state for e in cls):
            _logger.warning(
                "Unknown order state %s. Returning UNSPECIFIED.", order_state
            )
            return cls.UNSPECIFIED

        return cls(order_state)

    def to_pb(self) -> electricity_trading_pb2.OrderState.ValueType:
        """Convert an OrderState enum to protobuf OrderState value.

        Returns:
            Protobuf message corresponding to the OrderState enum.
        """
        return self.value

Attributes

ACTIVE class-attribute instance-attribute

ACTIVE = ORDER_STATE_ACTIVE

The order has been confirmed and is open in the market. It may be unfilled or partially filled.

CANCELED class-attribute instance-attribute

CANCELED = ORDER_STATE_CANCELED

The order has been canceled. This can occur due to a cancellation request by the market participant, system, or market operator.

EXPIRED class-attribute instance-attribute

EXPIRED = ORDER_STATE_EXPIRED

The order has not been filled within the defined duration and has expired.

FAILED class-attribute instance-attribute

FAILED = ORDER_STATE_FAILED

The order submission failed and was unable to be placed on the order book, usually due to a validation error or system issue.

FILLED class-attribute instance-attribute

FILLED = ORDER_STATE_FILLED

The order has been completely filled and there are no remaining quantities on the order.

HIBERNATE class-attribute instance-attribute

HIBERNATE = ORDER_STATE_HIBERNATE

The order has been entered into the system but is not currently exposed to the market. This could be due to certain conditions not yet being met.

PENDING class-attribute instance-attribute

PENDING = ORDER_STATE_PENDING

The order has been sent to the marketplace but has not yet been confirmed. This can be due to awaiting validation or system processing.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = ORDER_STATE_UNSPECIFIED

The order state is not known. Usually the default state of a newly created order object before any operations have been applied.

Functions

from_pb classmethod

from_pb(order_state: ValueType) -> 'OrderState'

Convert a protobuf OrderState value to OrderState enum.

PARAMETER	DESCRIPTION
order_state	Order state to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'OrderState'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, order_state: electricity_trading_pb2.OrderState.ValueType
) -> "OrderState":
    """Convert a protobuf OrderState value to OrderState enum.

    Args:
        order_state: Order state to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == order_state for e in cls):
        _logger.warning(
            "Unknown order state %s. Returning UNSPECIFIED.", order_state
        )
        return cls.UNSPECIFIED

    return cls(order_state)

to_pb

to_pb() -> ValueType

Convert an OrderState enum to protobuf OrderState value.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the OrderState enum.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.OrderState.ValueType:
    """Convert an OrderState enum to protobuf OrderState value.

    Returns:
        Protobuf message corresponding to the OrderState enum.
    """
    return self.value

frequenz.client.electricity_trading.OrderType

Bases: Enum

Type of the order (specifies how the order is to be executed in the market).

Source code in frequenz/client/electricity_trading/_types.py

class OrderType(enum.Enum):
    """Type of the order (specifies how the order is to be executed in the market)."""

    UNSPECIFIED = electricity_trading_pb2.OrderType.ORDER_TYPE_UNSPECIFIED
    """The order type has not been set."""

    LIMIT = electricity_trading_pb2.OrderType.ORDER_TYPE_LIMIT
    """Order to buy or sell at a specific price or better.
    It remains active until it is filled, cancelled, or expired."""

    STOP_LIMIT = electricity_trading_pb2.OrderType.ORDER_TYPE_STOP_LIMIT
    """An order that will be executed at a specified price,
    or better, after a given stop price has been reached."""

    ICEBERG = electricity_trading_pb2.OrderType.ORDER_TYPE_ICEBERG
    """A large order divided into smaller lots to hide the actual order quantity.
    Only the visible part of the order is shown in the order book."""

    BLOCK = electricity_trading_pb2.OrderType.ORDER_TYPE_BLOCK
    """User defined block order, generally a large quantity order filled all at once.
    (Not yet supported)."""

    BALANCE = electricity_trading_pb2.OrderType.ORDER_TYPE_BALANCE
    """Balance order aims to balance supply and demand, usually at
    a specific location or within a system.(Not yet supported)."""

    PREARRANGED = electricity_trading_pb2.OrderType.ORDER_TYPE_PREARRANGED
    """On exchange prearranged trade, a trade that has been privately
    negotiated and then submitted to the exchange. (Not yet supported)."""

    PRIVATE = electricity_trading_pb2.OrderType.ORDER_TYPE_PRIVATE
    """Private and confidential trade, not visible in the public
    order book and has no market impact. (Not yet supported)."""

    @classmethod
    @from_pb
    def from_pb(
        cls, order_type: electricity_trading_pb2.OrderType.ValueType
    ) -> "OrderType":
        """Convert a protobuf OrderType value to OrderType enum.

        Args:
            order_type: Order type to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == order_type for e in cls):
            _logger.warning("Unknown order type %s. Returning UNSPECIFIED.", order_type)
            return cls.UNSPECIFIED

        return cls(order_type)

    def to_pb(self) -> electricity_trading_pb2.OrderType.ValueType:
        """Convert an OrderType enum to protobuf OrderType value.

        Returns:
            Protobuf message corresponding to the OrderType enum.
        """
        return self.value

Attributes

BALANCE class-attribute instance-attribute

BALANCE = ORDER_TYPE_BALANCE

Balance order aims to balance supply and demand, usually at a specific location or within a system.(Not yet supported).

BLOCK class-attribute instance-attribute

BLOCK = ORDER_TYPE_BLOCK

User defined block order, generally a large quantity order filled all at once. (Not yet supported).

ICEBERG class-attribute instance-attribute

ICEBERG = ORDER_TYPE_ICEBERG

A large order divided into smaller lots to hide the actual order quantity. Only the visible part of the order is shown in the order book.

LIMIT class-attribute instance-attribute

LIMIT = ORDER_TYPE_LIMIT

Order to buy or sell at a specific price or better. It remains active until it is filled, cancelled, or expired.

PREARRANGED class-attribute instance-attribute

PREARRANGED = ORDER_TYPE_PREARRANGED

On exchange prearranged trade, a trade that has been privately negotiated and then submitted to the exchange. (Not yet supported).

PRIVATE class-attribute instance-attribute

PRIVATE = ORDER_TYPE_PRIVATE

Private and confidential trade, not visible in the public order book and has no market impact. (Not yet supported).

STOP_LIMIT class-attribute instance-attribute

STOP_LIMIT = ORDER_TYPE_STOP_LIMIT

An order that will be executed at a specified price, or better, after a given stop price has been reached.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = ORDER_TYPE_UNSPECIFIED

The order type has not been set.

Functions

from_pb classmethod

from_pb(order_type: ValueType) -> 'OrderType'

Convert a protobuf OrderType value to OrderType enum.

PARAMETER	DESCRIPTION
order_type	Order type to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'OrderType'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, order_type: electricity_trading_pb2.OrderType.ValueType
) -> "OrderType":
    """Convert a protobuf OrderType value to OrderType enum.

    Args:
        order_type: Order type to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == order_type for e in cls):
        _logger.warning("Unknown order type %s. Returning UNSPECIFIED.", order_type)
        return cls.UNSPECIFIED

    return cls(order_type)

to_pb

to_pb() -> ValueType

Convert an OrderType enum to protobuf OrderType value.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the OrderType enum.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.OrderType.ValueType:
    """Convert an OrderType enum to protobuf OrderType value.

    Returns:
        Protobuf message corresponding to the OrderType enum.
    """
    return self.value

frequenz.client.electricity_trading.Power dataclass

Represents power unit in Megawatthours (MW).

Source code in frequenz/client/electricity_trading/_types.py

@dataclass(frozen=True)
class Power:
    """Represents power unit in Megawatthours (MW)."""

    mw: Decimal

    @classmethod
    @from_pb
    def from_pb(cls, power: power_pb2.Power) -> Self:
        """Convert a protobuf Power to Power object.

        Args:
            power: Power to convert.

        Returns:
            Power object corresponding to the protobuf message.
        """
        return cls(mw=Decimal(power.mw.value))

    def to_pb(self) -> power_pb2.Power:
        """Convert a Power object to protobuf Power.

        Returns:
            Protobuf message corresponding to the Power object.
        """
        decimal_mw = decimal_pb2.Decimal()
        decimal_mw.value = str(self.mw)
        return power_pb2.Power(mw=decimal_mw)

    def __str__(self) -> str:
        """Return the string representation of the Power object.

        Returns:
            The string representation of the Power object.
        """
        return f"{self.mw} MW"

Functions

__str__

__str__() -> str

Return the string representation of the Power object.

RETURNS	DESCRIPTION
str	The string representation of the Power object.

Source code in frequenz/client/electricity_trading/_types.py

def __str__(self) -> str:
    """Return the string representation of the Power object.

    Returns:
        The string representation of the Power object.
    """
    return f"{self.mw} MW"

from_pb classmethod

from_pb(power: Power) -> Self

Convert a protobuf Power to Power object.

PARAMETER	DESCRIPTION
power	Power to convert. TYPE: Power

RETURNS	DESCRIPTION
Self	Power object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, power: power_pb2.Power) -> Self:
    """Convert a protobuf Power to Power object.

    Args:
        power: Power to convert.

    Returns:
        Power object corresponding to the protobuf message.
    """
    return cls(mw=Decimal(power.mw.value))

to_pb

to_pb() -> Power

Convert a Power object to protobuf Power.

RETURNS	DESCRIPTION
Power	Protobuf message corresponding to the Power object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> power_pb2.Power:
    """Convert a Power object to protobuf Power.

    Returns:
        Protobuf message corresponding to the Power object.
    """
    decimal_mw = decimal_pb2.Decimal()
    decimal_mw.value = str(self.mw)
    return power_pb2.Power(mw=decimal_mw)

frequenz.client.electricity_trading.Price dataclass

Price of an order.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass(frozen=True)
class Price:
    """Price of an order."""

    amount: Decimal
    """Amount of the price."""

    currency: Currency
    """Currency of the price."""

    @classmethod
    @from_pb
    def from_pb(cls, price: price_pb2.Price) -> Self:
        """Convert a protobuf Price to Price object.

        Args:
            price: Price to convert.

        Returns:
            Price object corresponding to the protobuf message.
        """
        return cls(
            amount=Decimal(price.amount.value),
            currency=Currency.from_pb(price.currency),
        )

    def to_pb(self) -> price_pb2.Price:
        """Convert a Price object to protobuf Price.

        Returns:
            Protobuf message corresponding to the Price object.
        """
        decimal_amount = decimal_pb2.Decimal()
        decimal_amount.value = str(self.amount)
        return price_pb2.Price(amount=decimal_amount, currency=self.currency.to_pb())

    def __str__(self) -> str:
        """Return string representation of the Price object.

        Returns:
            String representation of the Price object.
        """
        return f"{self.amount} {self.currency.name}"

Attributes

amount instance-attribute

amount: Decimal

Amount of the price.

currency instance-attribute

currency: Currency

Currency of the price.

Functions

__str__

__str__() -> str

Return string representation of the Price object.

RETURNS	DESCRIPTION
str	String representation of the Price object.

Source code in frequenz/client/electricity_trading/_types.py

def __str__(self) -> str:
    """Return string representation of the Price object.

    Returns:
        String representation of the Price object.
    """
    return f"{self.amount} {self.currency.name}"

from_pb classmethod

from_pb(price: Price) -> Self

Convert a protobuf Price to Price object.

PARAMETER	DESCRIPTION
price	Price to convert. TYPE: Price

RETURNS	DESCRIPTION
Self	Price object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, price: price_pb2.Price) -> Self:
    """Convert a protobuf Price to Price object.

    Args:
        price: Price to convert.

    Returns:
        Price object corresponding to the protobuf message.
    """
    return cls(
        amount=Decimal(price.amount.value),
        currency=Currency.from_pb(price.currency),
    )

to_pb

to_pb() -> Price

Convert a Price object to protobuf Price.

RETURNS	DESCRIPTION
Price	Protobuf message corresponding to the Price object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> price_pb2.Price:
    """Convert a Price object to protobuf Price.

    Returns:
        Protobuf message corresponding to the Price object.
    """
    decimal_amount = decimal_pb2.Decimal()
    decimal_amount.value = str(self.amount)
    return price_pb2.Price(amount=decimal_amount, currency=self.currency.to_pb())

frequenz.client.electricity_trading.PublicTrade dataclass

Represents a public order in the market.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass()
class PublicTrade:  # pylint: disable=too-many-instance-attributes
    """Represents a public order in the market."""

    public_trade_id: int
    """ID of the order from the public order book."""

    buy_delivery_area: DeliveryArea
    """Delivery area code of the buy side."""

    sell_delivery_area: DeliveryArea
    """Delivery area code of the sell side."""

    delivery_period: DeliveryPeriod
    """The delivery period for the contract."""

    execution_time: datetime
    """UTC Timestamp of the trades execution time."""

    price: Price
    """The limit price at which the contract is to be traded."""

    quantity: Power
    """The quantity of the contract being traded."""

    state: TradeState
    """State of the order."""

    def __post_init__(self) -> None:
        """Post initialization checks to ensure that all datetimes are UTC."""
        if self.execution_time.tzinfo is None:
            raise ValueError("Execution time must have timezone information")
        if self.execution_time.tzinfo != timezone.utc:
            _logger.warning("Execution time is not in UTC timezone. Converting to UTC.")
            self.execution_time = self.execution_time.astimezone(timezone.utc)

    @classmethod
    @from_pb
    def from_pb(cls, public_trade: electricity_trading_pb2.PublicTrade) -> Self:
        """Convert a protobuf PublicTrade to PublicTrade object.

        Args:
            public_trade: PublicTrade to convert.

        Returns:
            PublicTrade object corresponding to the protobuf message.
        """
        return cls(
            public_trade_id=public_trade.id,
            buy_delivery_area=DeliveryArea.from_pb(public_trade.buy_delivery_area),
            sell_delivery_area=DeliveryArea.from_pb(public_trade.sell_delivery_area),
            delivery_period=DeliveryPeriod.from_pb(public_trade.delivery_period),
            execution_time=public_trade.execution_time.ToDatetime(tzinfo=timezone.utc),
            price=Price.from_pb(public_trade.price),
            quantity=Power.from_pb(public_trade.quantity),
            state=TradeState.from_pb(public_trade.state),
        )

    def to_pb(self) -> electricity_trading_pb2.PublicTrade:
        """Convert a PublicTrade object to protobuf PublicTrade.

        Returns:
            Protobuf message corresponding to the PublicTrade object.
        """
        execution_time = timestamp_pb2.Timestamp()
        execution_time.FromDatetime(self.execution_time)

        return electricity_trading_pb2.PublicTrade(
            id=self.public_trade_id,
            buy_delivery_area=self.buy_delivery_area.to_pb(),
            sell_delivery_area=self.sell_delivery_area.to_pb(),
            delivery_period=self.delivery_period.to_pb(),
            execution_time=execution_time,
            price=self.price.to_pb(),
            quantity=self.quantity.to_pb(),
            state=electricity_trading_pb2.TradeState.ValueType(self.state.value),
        )

Attributes

buy_delivery_area instance-attribute

buy_delivery_area: DeliveryArea

Delivery area code of the buy side.

delivery_period instance-attribute

delivery_period: DeliveryPeriod

The delivery period for the contract.

execution_time instance-attribute

execution_time: datetime

UTC Timestamp of the trades execution time.

price instance-attribute

price: Price

The limit price at which the contract is to be traded.

public_trade_id instance-attribute

public_trade_id: int

ID of the order from the public order book.

quantity instance-attribute

quantity: Power

The quantity of the contract being traded.

sell_delivery_area instance-attribute

sell_delivery_area: DeliveryArea

Delivery area code of the sell side.

state instance-attribute

state: TradeState

State of the order.

Functions

__post_init__

__post_init__() -> None

Post initialization checks to ensure that all datetimes are UTC.

Source code in frequenz/client/electricity_trading/_types.py

def __post_init__(self) -> None:
    """Post initialization checks to ensure that all datetimes are UTC."""
    if self.execution_time.tzinfo is None:
        raise ValueError("Execution time must have timezone information")
    if self.execution_time.tzinfo != timezone.utc:
        _logger.warning("Execution time is not in UTC timezone. Converting to UTC.")
        self.execution_time = self.execution_time.astimezone(timezone.utc)

from_pb classmethod

from_pb(public_trade: PublicTrade) -> Self

Convert a protobuf PublicTrade to PublicTrade object.

PARAMETER	DESCRIPTION
public_trade	PublicTrade to convert. TYPE: PublicTrade

RETURNS	DESCRIPTION
Self	PublicTrade object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, public_trade: electricity_trading_pb2.PublicTrade) -> Self:
    """Convert a protobuf PublicTrade to PublicTrade object.

    Args:
        public_trade: PublicTrade to convert.

    Returns:
        PublicTrade object corresponding to the protobuf message.
    """
    return cls(
        public_trade_id=public_trade.id,
        buy_delivery_area=DeliveryArea.from_pb(public_trade.buy_delivery_area),
        sell_delivery_area=DeliveryArea.from_pb(public_trade.sell_delivery_area),
        delivery_period=DeliveryPeriod.from_pb(public_trade.delivery_period),
        execution_time=public_trade.execution_time.ToDatetime(tzinfo=timezone.utc),
        price=Price.from_pb(public_trade.price),
        quantity=Power.from_pb(public_trade.quantity),
        state=TradeState.from_pb(public_trade.state),
    )

to_pb

to_pb() -> PublicTrade

Convert a PublicTrade object to protobuf PublicTrade.

RETURNS	DESCRIPTION
PublicTrade	Protobuf message corresponding to the PublicTrade object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.PublicTrade:
    """Convert a PublicTrade object to protobuf PublicTrade.

    Returns:
        Protobuf message corresponding to the PublicTrade object.
    """
    execution_time = timestamp_pb2.Timestamp()
    execution_time.FromDatetime(self.execution_time)

    return electricity_trading_pb2.PublicTrade(
        id=self.public_trade_id,
        buy_delivery_area=self.buy_delivery_area.to_pb(),
        sell_delivery_area=self.sell_delivery_area.to_pb(),
        delivery_period=self.delivery_period.to_pb(),
        execution_time=execution_time,
        price=self.price.to_pb(),
        quantity=self.quantity.to_pb(),
        state=electricity_trading_pb2.TradeState.ValueType(self.state.value),
    )

frequenz.client.electricity_trading.PublicTradeFilter dataclass

Parameters for filtering the historic, publicly executed orders (trades).

Source code in frequenz/client/electricity_trading/_types.py

@dataclass(frozen=True)
class PublicTradeFilter:
    """Parameters for filtering the historic, publicly executed orders (trades)."""

    states: list[TradeState] | None = None
    """List of order states to filter for."""

    delivery_period: DeliveryPeriod | None = None
    """Delivery period to filter for."""

    buy_delivery_area: DeliveryArea | None = None
    """Delivery area to filter for on the buy side."""

    sell_delivery_area: DeliveryArea | None = None
    """Delivery area to filter for on the sell side."""

    def __eq__(self, other: object) -> bool:
        """
        Check if two PublicTradeFilter objects are equal.

        Args:
            other: PublicTradeFilter object to compare with.

        Returns:
            True if the two PublicTradeFilter objects are equal, False otherwise.
        """
        if not isinstance(other, PublicTradeFilter):
            return NotImplemented
        return (
            self.states == other.states
            and self.delivery_period == other.delivery_period
            and self.buy_delivery_area == other.buy_delivery_area
            and self.sell_delivery_area == other.sell_delivery_area
        )

    def __hash__(self) -> int:
        """
        Create hash of the PublicTradeFilter object.

        Returns:
            Hash of the PublicTradeFilter object.
        """
        return hash(
            (
                tuple(self.states) if self.states is not None else None,
                self.delivery_period,
                self.buy_delivery_area,
                self.sell_delivery_area,
            )
        )

    @classmethod
    @from_pb
    def from_pb(
        cls, public_trade_filter: electricity_trading_pb2.PublicTradeFilter
    ) -> Self:
        """Convert a protobuf PublicTradeFilter to PublicTradeFilter object.

        Args:
            public_trade_filter: PublicTradeFilter to convert.

        Returns:
            PublicTradeFilter object corresponding to the protobuf message.
        """
        return cls(
            states=(
                [TradeState.from_pb(state) for state in public_trade_filter.states]
                if public_trade_filter.states
                else None
            ),
            delivery_period=(
                DeliveryPeriod.from_pb(public_trade_filter.delivery_period)
                if public_trade_filter.HasField("delivery_period")
                else None
            ),
            buy_delivery_area=(
                DeliveryArea.from_pb(public_trade_filter.buy_delivery_area)
                if public_trade_filter.HasField("buy_delivery_area")
                else None
            ),
            sell_delivery_area=(
                DeliveryArea.from_pb(public_trade_filter.sell_delivery_area)
                if public_trade_filter.HasField("sell_delivery_area")
                else None
            ),
        )

    def to_pb(self) -> electricity_trading_pb2.PublicTradeFilter:
        """Convert a PublicTradeFilter object to protobuf PublicTradeFilter.

        Returns:
            Protobuf PublicTradeFilter corresponding to the object.
        """
        return electricity_trading_pb2.PublicTradeFilter(
            states=(
                [
                    electricity_trading_pb2.TradeState.ValueType(state.value)
                    for state in self.states
                ]
                if self.states
                else None
            ),
            delivery_period=(
                self.delivery_period.to_pb() if self.delivery_period else None
            ),
            buy_delivery_area=(
                self.buy_delivery_area.to_pb() if self.buy_delivery_area else None
            ),
            sell_delivery_area=(
                self.sell_delivery_area.to_pb() if self.sell_delivery_area else None
            ),
        )

Attributes

buy_delivery_area class-attribute instance-attribute

buy_delivery_area: DeliveryArea | None = None

Delivery area to filter for on the buy side.

delivery_period class-attribute instance-attribute

delivery_period: DeliveryPeriod | None = None

Delivery period to filter for.

sell_delivery_area class-attribute instance-attribute

sell_delivery_area: DeliveryArea | None = None

Delivery area to filter for on the sell side.

states class-attribute instance-attribute

states: list[TradeState] | None = None

List of order states to filter for.

Functions

__eq__

__eq__(other: object) -> bool

Check if two PublicTradeFilter objects are equal.

PARAMETER	DESCRIPTION
other	PublicTradeFilter object to compare with. TYPE: object

RETURNS	DESCRIPTION
bool	True if the two PublicTradeFilter objects are equal, False otherwise.

Source code in frequenz/client/electricity_trading/_types.py

def __eq__(self, other: object) -> bool:
    """
    Check if two PublicTradeFilter objects are equal.

    Args:
        other: PublicTradeFilter object to compare with.

    Returns:
        True if the two PublicTradeFilter objects are equal, False otherwise.
    """
    if not isinstance(other, PublicTradeFilter):
        return NotImplemented
    return (
        self.states == other.states
        and self.delivery_period == other.delivery_period
        and self.buy_delivery_area == other.buy_delivery_area
        and self.sell_delivery_area == other.sell_delivery_area
    )

__hash__

__hash__() -> int

Create hash of the PublicTradeFilter object.

RETURNS	DESCRIPTION
int	Hash of the PublicTradeFilter object.

Source code in frequenz/client/electricity_trading/_types.py

def __hash__(self) -> int:
    """
    Create hash of the PublicTradeFilter object.

    Returns:
        Hash of the PublicTradeFilter object.
    """
    return hash(
        (
            tuple(self.states) if self.states is not None else None,
            self.delivery_period,
            self.buy_delivery_area,
            self.sell_delivery_area,
        )
    )

from_pb classmethod

from_pb(public_trade_filter: PublicTradeFilter) -> Self

Convert a protobuf PublicTradeFilter to PublicTradeFilter object.

PARAMETER	DESCRIPTION
public_trade_filter	PublicTradeFilter to convert. TYPE: PublicTradeFilter

RETURNS	DESCRIPTION
Self	PublicTradeFilter object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, public_trade_filter: electricity_trading_pb2.PublicTradeFilter
) -> Self:
    """Convert a protobuf PublicTradeFilter to PublicTradeFilter object.

    Args:
        public_trade_filter: PublicTradeFilter to convert.

    Returns:
        PublicTradeFilter object corresponding to the protobuf message.
    """
    return cls(
        states=(
            [TradeState.from_pb(state) for state in public_trade_filter.states]
            if public_trade_filter.states
            else None
        ),
        delivery_period=(
            DeliveryPeriod.from_pb(public_trade_filter.delivery_period)
            if public_trade_filter.HasField("delivery_period")
            else None
        ),
        buy_delivery_area=(
            DeliveryArea.from_pb(public_trade_filter.buy_delivery_area)
            if public_trade_filter.HasField("buy_delivery_area")
            else None
        ),
        sell_delivery_area=(
            DeliveryArea.from_pb(public_trade_filter.sell_delivery_area)
            if public_trade_filter.HasField("sell_delivery_area")
            else None
        ),
    )

to_pb

to_pb() -> PublicTradeFilter

Convert a PublicTradeFilter object to protobuf PublicTradeFilter.

RETURNS	DESCRIPTION
PublicTradeFilter	Protobuf PublicTradeFilter corresponding to the object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.PublicTradeFilter:
    """Convert a PublicTradeFilter object to protobuf PublicTradeFilter.

    Returns:
        Protobuf PublicTradeFilter corresponding to the object.
    """
    return electricity_trading_pb2.PublicTradeFilter(
        states=(
            [
                electricity_trading_pb2.TradeState.ValueType(state.value)
                for state in self.states
            ]
            if self.states
            else None
        ),
        delivery_period=(
            self.delivery_period.to_pb() if self.delivery_period else None
        ),
        buy_delivery_area=(
            self.buy_delivery_area.to_pb() if self.buy_delivery_area else None
        ),
        sell_delivery_area=(
            self.sell_delivery_area.to_pb() if self.sell_delivery_area else None
        ),
    )

frequenz.client.electricity_trading.StateDetail dataclass

Details about the current state of the order.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass(frozen=True)
class StateDetail:
    """Details about the current state of the order."""

    state: OrderState
    """Current state of the order."""

    state_reason: StateReason
    """Reason for the current state."""

    market_actor: MarketActor
    """Actor responsible for the current state."""

    @classmethod
    @from_pb
    def from_pb(
        cls, state_detail: electricity_trading_pb2.OrderDetail.StateDetail
    ) -> Self:
        """Convert a protobuf StateDetail to StateDetail object.

        Args:
            state_detail: StateDetail to convert.

        Returns:
            StateDetail object corresponding to the protobuf message.
        """
        return cls(
            state=OrderState.from_pb(state_detail.state),
            state_reason=StateReason.from_pb(state_detail.state_reason),
            market_actor=MarketActor.from_pb(state_detail.market_actor),
        )

    def to_pb(self) -> electricity_trading_pb2.OrderDetail.StateDetail:
        """Convert a StateDetail object to protobuf StateDetail.

        Returns:
            Protobuf message corresponding to the StateDetail object.
        """
        return electricity_trading_pb2.OrderDetail.StateDetail(
            state=electricity_trading_pb2.OrderState.ValueType(self.state.value),
            state_reason=electricity_trading_pb2.OrderDetail.StateDetail.StateReason.ValueType(
                self.state_reason.value
            ),
            market_actor=electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.ValueType(
                self.market_actor.value
            ),
        )

Attributes

market_actor instance-attribute

market_actor: MarketActor

Actor responsible for the current state.

state instance-attribute

state: OrderState

Current state of the order.

state_reason instance-attribute

state_reason: StateReason

Reason for the current state.

Functions

from_pb classmethod

from_pb(state_detail: StateDetail) -> Self

Convert a protobuf StateDetail to StateDetail object.

PARAMETER	DESCRIPTION
state_detail	StateDetail to convert. TYPE: StateDetail

RETURNS	DESCRIPTION
Self	StateDetail object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, state_detail: electricity_trading_pb2.OrderDetail.StateDetail
) -> Self:
    """Convert a protobuf StateDetail to StateDetail object.

    Args:
        state_detail: StateDetail to convert.

    Returns:
        StateDetail object corresponding to the protobuf message.
    """
    return cls(
        state=OrderState.from_pb(state_detail.state),
        state_reason=StateReason.from_pb(state_detail.state_reason),
        market_actor=MarketActor.from_pb(state_detail.market_actor),
    )

to_pb

to_pb() -> StateDetail

Convert a StateDetail object to protobuf StateDetail.

RETURNS	DESCRIPTION
StateDetail	Protobuf message corresponding to the StateDetail object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.OrderDetail.StateDetail:
    """Convert a StateDetail object to protobuf StateDetail.

    Returns:
        Protobuf message corresponding to the StateDetail object.
    """
    return electricity_trading_pb2.OrderDetail.StateDetail(
        state=electricity_trading_pb2.OrderState.ValueType(self.state.value),
        state_reason=electricity_trading_pb2.OrderDetail.StateDetail.StateReason.ValueType(
            self.state_reason.value
        ),
        market_actor=electricity_trading_pb2.OrderDetail.StateDetail.MarketActor.ValueType(
            self.market_actor.value
        ),
    )

frequenz.client.electricity_trading.StateReason

Bases: Enum

Reason that led to a state change.

Source code in frequenz/client/electricity_trading/_types.py

class StateReason(enum.Enum):
    """Reason that led to a state change."""

    UNSPECIFIED = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_UNSPECIFIED
    )
    """The reason for the state change has not been specified."""

    ADD = electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_ADD
    """The order was added."""

    MODIFY = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_MODIFY
    )
    """The order was modified."""

    DELETE = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_DELETE
    )
    """The order was deleted."""

    DEACTIVATE = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_DEACTIVATE
    )
    """The order was deactivated."""

    REJECT = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_REJECT
    )
    """The order was rejected."""

    FULL_EXECUTION = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_FULL_EXECUTION
    )
    """The order was fully executed."""

    PARTIAL_EXECUTION = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_PARTIAL_EXECUTION
    )
    """The order was partially executed."""

    ICEBERG_SLICE_ADD = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_ICEBERG_SLICE_ADD
    )
    """An iceberg slice was added."""

    VALIDATION_FAIL = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_VALIDATION_FAIL
    )
    """The order failed validation."""

    UNKNOWN_STATE = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_UNKNOWN_STATE
    )
    """The state of the order is unknown."""

    QUOTE_ADD = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_QUOTE_ADD
    )
    """A quote was added."""

    QUOTE_FULL_EXECUTION = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_QUOTE_FULL_EXECUTION  # noqa: E501
    )
    """A quote was fully executed."""

    QUOTE_PARTIAL_EXECUTION = (
        electricity_trading_pb2.OrderDetail.StateDetail.StateReason.STATE_REASON_QUOTE_PARTIAL_EXECUTION  # noqa: E501
    )
    """A quote was partially executed."""

    @classmethod
    @from_pb
    def from_pb(
        cls,
        state_reason: electricity_trading_pb2.OrderDetail.StateDetail.StateReason.ValueType,
    ) -> "StateReason":
        """Convert a protobuf StateReason value to StateReason enum.

        Args:
            state_reason: State reason to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == state_reason for e in cls):
            _logger.warning(
                "Unknown state reason %s. Returning UNSPECIFIED.", state_reason
            )
            return cls.UNSPECIFIED

        return cls(state_reason)

    def to_pb(
        self,
    ) -> electricity_trading_pb2.OrderDetail.StateDetail.StateReason.ValueType:
        """Convert a StateReason enum to protobuf StateReason value.

        Returns:
            Protobuf message corresponding to the StateReason enum.
        """
        return self.value

Attributes

ADD class-attribute instance-attribute

ADD = STATE_REASON_ADD

The order was added.

DEACTIVATE class-attribute instance-attribute

DEACTIVATE = STATE_REASON_DEACTIVATE

The order was deactivated.

DELETE class-attribute instance-attribute

DELETE = STATE_REASON_DELETE

The order was deleted.

FULL_EXECUTION class-attribute instance-attribute

FULL_EXECUTION = STATE_REASON_FULL_EXECUTION

The order was fully executed.

ICEBERG_SLICE_ADD class-attribute instance-attribute

ICEBERG_SLICE_ADD = STATE_REASON_ICEBERG_SLICE_ADD

An iceberg slice was added.

MODIFY class-attribute instance-attribute

MODIFY = STATE_REASON_MODIFY

The order was modified.

PARTIAL_EXECUTION class-attribute instance-attribute

PARTIAL_EXECUTION = STATE_REASON_PARTIAL_EXECUTION

The order was partially executed.

QUOTE_ADD class-attribute instance-attribute

QUOTE_ADD = STATE_REASON_QUOTE_ADD

A quote was added.

QUOTE_FULL_EXECUTION class-attribute instance-attribute

QUOTE_FULL_EXECUTION = STATE_REASON_QUOTE_FULL_EXECUTION

A quote was fully executed.

QUOTE_PARTIAL_EXECUTION class-attribute instance-attribute

QUOTE_PARTIAL_EXECUTION = (
    STATE_REASON_QUOTE_PARTIAL_EXECUTION
)

A quote was partially executed.

REJECT class-attribute instance-attribute

REJECT = STATE_REASON_REJECT

The order was rejected.

UNKNOWN_STATE class-attribute instance-attribute

UNKNOWN_STATE = STATE_REASON_UNKNOWN_STATE

The state of the order is unknown.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = STATE_REASON_UNSPECIFIED

The reason for the state change has not been specified.

VALIDATION_FAIL class-attribute instance-attribute

VALIDATION_FAIL = STATE_REASON_VALIDATION_FAIL

The order failed validation.

Functions

from_pb classmethod

from_pb(state_reason: ValueType) -> 'StateReason'

Convert a protobuf StateReason value to StateReason enum.

PARAMETER	DESCRIPTION
state_reason	State reason to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'StateReason'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls,
    state_reason: electricity_trading_pb2.OrderDetail.StateDetail.StateReason.ValueType,
) -> "StateReason":
    """Convert a protobuf StateReason value to StateReason enum.

    Args:
        state_reason: State reason to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == state_reason for e in cls):
        _logger.warning(
            "Unknown state reason %s. Returning UNSPECIFIED.", state_reason
        )
        return cls.UNSPECIFIED

    return cls(state_reason)

to_pb

to_pb() -> ValueType

Convert a StateReason enum to protobuf StateReason value.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the StateReason enum.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(
    self,
) -> electricity_trading_pb2.OrderDetail.StateDetail.StateReason.ValueType:
    """Convert a StateReason enum to protobuf StateReason value.

    Returns:
        Protobuf message corresponding to the StateReason enum.
    """
    return self.value

frequenz.client.electricity_trading.Trade dataclass

Represents a private trade in the electricity market.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass()
class Trade:  # pylint: disable=too-many-instance-attributes
    """Represents a private trade in the electricity market."""

    id: int
    """ID of the trade."""

    order_id: int
    """ID of the corresponding order."""

    side: MarketSide
    """Indicates if the trade's order was on the Buy or Sell side of the
    market."""

    delivery_area: DeliveryArea
    """Delivery area of the trade."""

    delivery_period: DeliveryPeriod
    """The delivery period for the contract."""

    execution_time: datetime
    """UTC Timestamp of the trade's execution time."""

    price: Price
    """The price at which the trade was executed."""

    quantity: Power
    """The executed quantity of the trade."""

    state: TradeState
    """Current state of the trade."""

    def __post_init__(self) -> None:
        """Post initialization checks to ensure that all datetimes are UTC."""
        if self.execution_time.tzinfo is None:
            raise ValueError("Execution time must have timezone information")
        if self.execution_time.tzinfo != timezone.utc:
            _logger.warning("Execution timenis not in UTC timezone. Converting to UTC.")
            self.execution_time = self.execution_time.astimezone(timezone.utc)

    @classmethod
    @from_pb
    def from_pb(cls, trade: electricity_trading_pb2.Trade) -> Self:
        """Convert a protobuf Trade to Trade object.

        Args:
            trade: Trade to convert.

        Returns:
            Trade object corresponding to the protobuf message.
        """
        return cls(
            id=trade.id,
            order_id=trade.order_id,
            side=MarketSide.from_pb(trade.side),
            delivery_area=DeliveryArea.from_pb(trade.delivery_area),
            delivery_period=DeliveryPeriod.from_pb(trade.delivery_period),
            execution_time=trade.execution_time.ToDatetime(tzinfo=timezone.utc),
            price=Price.from_pb(trade.price),
            quantity=Power.from_pb(trade.quantity),
            state=TradeState.from_pb(trade.state),
        )

    def to_pb(self) -> electricity_trading_pb2.Trade:
        """Convert a Trade object to protobuf Trade.

        Returns:
            Protobuf message corresponding to the Trade object.
        """
        execution_time = timestamp_pb2.Timestamp()
        execution_time.FromDatetime(self.execution_time)

        return electricity_trading_pb2.Trade(
            id=self.id,
            order_id=self.order_id,
            side=electricity_trading_pb2.MarketSide.ValueType(self.side.value),
            delivery_area=self.delivery_area.to_pb(),
            delivery_period=self.delivery_period.to_pb(),
            execution_time=execution_time,
            price=self.price.to_pb(),
            quantity=self.quantity.to_pb(),
            state=electricity_trading_pb2.TradeState.ValueType(self.state.value),
        )

Attributes

delivery_area instance-attribute

delivery_area: DeliveryArea

Delivery area of the trade.

delivery_period instance-attribute

delivery_period: DeliveryPeriod

The delivery period for the contract.

execution_time instance-attribute

execution_time: datetime

UTC Timestamp of the trade's execution time.

id instance-attribute

id: int

ID of the trade.

order_id instance-attribute

order_id: int

ID of the corresponding order.

price instance-attribute

price: Price

The price at which the trade was executed.

quantity instance-attribute

quantity: Power

The executed quantity of the trade.

side instance-attribute

side: MarketSide

Indicates if the trade's order was on the Buy or Sell side of the market.

state instance-attribute

state: TradeState

Current state of the trade.

Functions

__post_init__

__post_init__() -> None

Post initialization checks to ensure that all datetimes are UTC.

Source code in frequenz/client/electricity_trading/_types.py

def __post_init__(self) -> None:
    """Post initialization checks to ensure that all datetimes are UTC."""
    if self.execution_time.tzinfo is None:
        raise ValueError("Execution time must have timezone information")
    if self.execution_time.tzinfo != timezone.utc:
        _logger.warning("Execution timenis not in UTC timezone. Converting to UTC.")
        self.execution_time = self.execution_time.astimezone(timezone.utc)

from_pb classmethod

from_pb(trade: Trade) -> Self

Convert a protobuf Trade to Trade object.

PARAMETER	DESCRIPTION
trade	Trade to convert. TYPE: Trade

RETURNS	DESCRIPTION
Self	Trade object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(cls, trade: electricity_trading_pb2.Trade) -> Self:
    """Convert a protobuf Trade to Trade object.

    Args:
        trade: Trade to convert.

    Returns:
        Trade object corresponding to the protobuf message.
    """
    return cls(
        id=trade.id,
        order_id=trade.order_id,
        side=MarketSide.from_pb(trade.side),
        delivery_area=DeliveryArea.from_pb(trade.delivery_area),
        delivery_period=DeliveryPeriod.from_pb(trade.delivery_period),
        execution_time=trade.execution_time.ToDatetime(tzinfo=timezone.utc),
        price=Price.from_pb(trade.price),
        quantity=Power.from_pb(trade.quantity),
        state=TradeState.from_pb(trade.state),
    )

to_pb

to_pb() -> Trade

Convert a Trade object to protobuf Trade.

RETURNS	DESCRIPTION
Trade	Protobuf message corresponding to the Trade object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.Trade:
    """Convert a Trade object to protobuf Trade.

    Returns:
        Protobuf message corresponding to the Trade object.
    """
    execution_time = timestamp_pb2.Timestamp()
    execution_time.FromDatetime(self.execution_time)

    return electricity_trading_pb2.Trade(
        id=self.id,
        order_id=self.order_id,
        side=electricity_trading_pb2.MarketSide.ValueType(self.side.value),
        delivery_area=self.delivery_area.to_pb(),
        delivery_period=self.delivery_period.to_pb(),
        execution_time=execution_time,
        price=self.price.to_pb(),
        quantity=self.quantity.to_pb(),
        state=electricity_trading_pb2.TradeState.ValueType(self.state.value),
    )

frequenz.client.electricity_trading.TradeState

Bases: Enum

State of a trade.

Source code in frequenz/client/electricity_trading/_types.py

class TradeState(enum.Enum):
    """State of a trade."""

    UNSPECIFIED = electricity_trading_pb2.TradeState.TRADE_STATE_UNSPECIFIED
    """The state is not known."""

    ACTIVE = electricity_trading_pb2.TradeState.TRADE_STATE_ACTIVE
    """The trade has been executed in the market."""

    CANCEL_REQUESTED = electricity_trading_pb2.TradeState.TRADE_STATE_CANCEL_REQUESTED
    """A cancellation request for the trade has been submitted."""

    CANCEL_REJECTED = electricity_trading_pb2.TradeState.TRADE_STATE_CANCEL_REJECTED
    """The trade cancellation request was rejected."""

    CANCELED = electricity_trading_pb2.TradeState.TRADE_STATE_CANCELED
    """The trade has been cancelled. This can occur due to a cancellation request by the market
    participant, system, or market operator."""

    RECALL = electricity_trading_pb2.TradeState.TRADE_STATE_RECALLED
    """The trade has been recalled. This could be due to a system issue or a request from the market
    participant or market operator."""

    RECALL_REQUESTED = electricity_trading_pb2.TradeState.TRADE_STATE_RECALL_REQUESTED
    """A recall request for the trade has been submitted."""

    RECALL_REJECTED = electricity_trading_pb2.TradeState.TRADE_STATE_RECALL_REJECTED
    """The trade recall request was rejected."""

    APPROVAL_REQUESTED = (
        electricity_trading_pb2.TradeState.TRADE_STATE_APPROVAL_REQUESTED
    )
    """An approval has been requested."""

    @classmethod
    @from_pb
    def from_pb(
        cls, trade_state: electricity_trading_pb2.TradeState.ValueType
    ) -> "TradeState":
        """Convert a protobuf TradeState value to TradeState enum.

        Args:
            trade_state: The trade state to convert.

        Returns:
            Enum value corresponding to the protobuf message.
        """
        if not any(e.value == trade_state for e in cls):
            _logger.warning(
                "Unknown trade state %s. Returning UNSPECIFIED.", trade_state
            )
            return cls.UNSPECIFIED

        return cls(trade_state)

    def to_pb(self) -> electricity_trading_pb2.TradeState.ValueType:
        """Convert a TradeState enum to protobuf TradeState value.

        Returns:
            Protobuf message corresponding to the TradeState enum.
        """
        return self.value

Attributes

ACTIVE class-attribute instance-attribute

ACTIVE = TRADE_STATE_ACTIVE

The trade has been executed in the market.

APPROVAL_REQUESTED class-attribute instance-attribute

APPROVAL_REQUESTED = TRADE_STATE_APPROVAL_REQUESTED

An approval has been requested.

CANCELED class-attribute instance-attribute

CANCELED = TRADE_STATE_CANCELED

The trade has been cancelled. This can occur due to a cancellation request by the market participant, system, or market operator.

CANCEL_REJECTED class-attribute instance-attribute

CANCEL_REJECTED = TRADE_STATE_CANCEL_REJECTED

The trade cancellation request was rejected.

CANCEL_REQUESTED class-attribute instance-attribute

CANCEL_REQUESTED = TRADE_STATE_CANCEL_REQUESTED

A cancellation request for the trade has been submitted.

RECALL class-attribute instance-attribute

RECALL = TRADE_STATE_RECALLED

The trade has been recalled. This could be due to a system issue or a request from the market participant or market operator.

RECALL_REJECTED class-attribute instance-attribute

RECALL_REJECTED = TRADE_STATE_RECALL_REJECTED

The trade recall request was rejected.

RECALL_REQUESTED class-attribute instance-attribute

RECALL_REQUESTED = TRADE_STATE_RECALL_REQUESTED

A recall request for the trade has been submitted.

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = TRADE_STATE_UNSPECIFIED

The state is not known.

Functions

from_pb classmethod

from_pb(trade_state: ValueType) -> 'TradeState'

Convert a protobuf TradeState value to TradeState enum.

PARAMETER	DESCRIPTION
trade_state	The trade state to convert. TYPE: ValueType

RETURNS	DESCRIPTION
'TradeState'	Enum value corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls, trade_state: electricity_trading_pb2.TradeState.ValueType
) -> "TradeState":
    """Convert a protobuf TradeState value to TradeState enum.

    Args:
        trade_state: The trade state to convert.

    Returns:
        Enum value corresponding to the protobuf message.
    """
    if not any(e.value == trade_state for e in cls):
        _logger.warning(
            "Unknown trade state %s. Returning UNSPECIFIED.", trade_state
        )
        return cls.UNSPECIFIED

    return cls(trade_state)

to_pb

to_pb() -> ValueType

Convert a TradeState enum to protobuf TradeState value.

RETURNS	DESCRIPTION
ValueType	Protobuf message corresponding to the TradeState enum.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.TradeState.ValueType:
    """Convert a TradeState enum to protobuf TradeState value.

    Returns:
        Protobuf message corresponding to the TradeState enum.
    """
    return self.value

frequenz.client.electricity_trading.UpdateOrder dataclass

Represents the order properties that can be updated after an order has been placed.

At least one of the optional fields must be set for an update to take place.

Source code in frequenz/client/electricity_trading/_types.py

@dataclass()
class UpdateOrder:  # pylint: disable=too-many-instance-attributes
    """
    Represents the order properties that can be updated after an order has been placed.

    At least one of the optional fields must be set for an update to take place.
    """

    price: Price | None = None
    """The updated limit price at which the contract is to be traded.
    This is the maximum price for a BUY order or the minimum price for a SELL order."""

    quantity: Power | None = None
    """The updated quantity of the contract being traded, specified in MW."""

    stop_price: Price | None = None
    """Applicable for STOP_LIMIT orders. This is the updated stop price that triggers
    the limit order."""

    peak_price_delta: Price | None = None
    """Applicable for ICEBERG orders. This is the updated price difference
    between the peak price and the limit price."""

    display_quantity: Power | None = None
    """Applicable for ICEBERG orders. This is the updated quantity of the order
    to be displayed in the order book."""

    execution_option: OrderExecutionOption | None = None
    """Updated execution options such as All or None, Fill or Kill, etc."""

    valid_until: datetime | None = None
    """This is an updated timestamp defining the time after which the order should
    be cancelled if not filled. The timestamp is in UTC."""

    payload: dict[str, struct_pb2.Value] | None = None
    """Updated user-defined payload individual to a specific order. This can be any data
    that the user wants to associate with the order."""

    tag: str | None = None
    """Updated user-defined tag to group related orders."""

    def __post_init__(self) -> None:
        """Post initialization checks to ensure that all datetimes are UTC."""
        if self.valid_until is not None:
            if self.valid_until.tzinfo is None:
                raise ValueError("Valid until must be a UTC datetime.")
            if self.valid_until.tzinfo != timezone.utc:
                _logger.warning("Valid until is not a UTC datetime. Converting to UTC.")
                self.valid_until = self.valid_until.astimezone(timezone.utc)

    @classmethod
    @from_pb
    def from_pb(
        cls,
        update_order: electricity_trading_pb2.UpdateGridpoolOrderRequest.UpdateOrder,
    ) -> Self:
        """Convert a protobuf UpdateOrder to UpdateOrder object.

        Args:
            update_order: UpdateOrder to convert.

        Returns:
            UpdateOrder object corresponding to the protobuf message.
        """
        return cls(
            price=(
                Price.from_pb(update_order.price)
                if update_order.HasField("price")
                else None
            ),
            quantity=(
                Power.from_pb(update_order.quantity)
                if update_order.HasField("quantity")
                else None
            ),
            stop_price=(
                Price.from_pb(update_order.stop_price)
                if update_order.HasField("stop_price")
                else None
            ),
            peak_price_delta=(
                Price.from_pb(update_order.peak_price_delta)
                if update_order.HasField("peak_price_delta")
                else None
            ),
            display_quantity=(
                Power.from_pb(update_order.display_quantity)
                if update_order.HasField("display_quantity")
                else None
            ),
            execution_option=(
                OrderExecutionOption.from_pb(update_order.execution_option)
                if update_order.HasField("execution_option")
                else None
            ),
            valid_until=(
                update_order.valid_until.ToDatetime(tzinfo=timezone.utc)
                if update_order.HasField("valid_until")
                else None
            ),
            payload=(
                json_format.MessageToDict(update_order.payload)
                if update_order.payload
                else None
            ),
            tag=update_order.tag if update_order.HasField("tag") else None,
        )

    def to_pb(self) -> electricity_trading_pb2.UpdateGridpoolOrderRequest.UpdateOrder:
        """Convert a UpdateOrder object to protobuf UpdateOrder.

        Returns:
            Protobuf UpdateOrder corresponding to the object.
        """
        if self.valid_until:
            valid_until = timestamp_pb2.Timestamp()
            valid_until.FromDatetime(self.valid_until)
        else:
            valid_until = None
        return electricity_trading_pb2.UpdateGridpoolOrderRequest.UpdateOrder(
            price=self.price.to_pb() if self.price else None,
            quantity=self.quantity.to_pb() if self.quantity else None,
            stop_price=self.stop_price.to_pb() if self.stop_price else None,
            peak_price_delta=(
                self.peak_price_delta.to_pb() if self.peak_price_delta else None
            ),
            display_quantity=(
                self.display_quantity.to_pb() if self.display_quantity else None
            ),
            execution_option=(
                electricity_trading_pb2.OrderExecutionOption.ValueType(
                    self.execution_option.value
                )
                if self.execution_option
                else None
            ),
            valid_until=valid_until if self.valid_until else None,
            payload=struct_pb2.Struct(fields=self.payload) if self.payload else None,
            tag=self.tag if self.tag else None,
        )

Attributes

display_quantity class-attribute instance-attribute

display_quantity: Power | None = None

Applicable for ICEBERG orders. This is the updated quantity of the order to be displayed in the order book.

execution_option class-attribute instance-attribute

execution_option: OrderExecutionOption | None = None

Updated execution options such as All or None, Fill or Kill, etc.

payload class-attribute instance-attribute

payload: dict[str, Value] | None = None

Updated user-defined payload individual to a specific order. This can be any data that the user wants to associate with the order.

peak_price_delta class-attribute instance-attribute

peak_price_delta: Price | None = None

Applicable for ICEBERG orders. This is the updated price difference between the peak price and the limit price.

price class-attribute instance-attribute

price: Price | None = None

The updated limit price at which the contract is to be traded. This is the maximum price for a BUY order or the minimum price for a SELL order.

quantity class-attribute instance-attribute

quantity: Power | None = None

The updated quantity of the contract being traded, specified in MW.

stop_price class-attribute instance-attribute

stop_price: Price | None = None

Applicable for STOP_LIMIT orders. This is the updated stop price that triggers the limit order.

tag class-attribute instance-attribute

tag: str | None = None

Updated user-defined tag to group related orders.

valid_until class-attribute instance-attribute

valid_until: datetime | None = None

This is an updated timestamp defining the time after which the order should be cancelled if not filled. The timestamp is in UTC.

Functions

__post_init__

__post_init__() -> None

Post initialization checks to ensure that all datetimes are UTC.

Source code in frequenz/client/electricity_trading/_types.py

def __post_init__(self) -> None:
    """Post initialization checks to ensure that all datetimes are UTC."""
    if self.valid_until is not None:
        if self.valid_until.tzinfo is None:
            raise ValueError("Valid until must be a UTC datetime.")
        if self.valid_until.tzinfo != timezone.utc:
            _logger.warning("Valid until is not a UTC datetime. Converting to UTC.")
            self.valid_until = self.valid_until.astimezone(timezone.utc)

from_pb classmethod

from_pb(update_order: UpdateOrder) -> Self

Convert a protobuf UpdateOrder to UpdateOrder object.

PARAMETER	DESCRIPTION
update_order	UpdateOrder to convert. TYPE: UpdateOrder

RETURNS	DESCRIPTION
Self	UpdateOrder object corresponding to the protobuf message.

Source code in frequenz/client/electricity_trading/_types.py

@classmethod
@from_pb
def from_pb(
    cls,
    update_order: electricity_trading_pb2.UpdateGridpoolOrderRequest.UpdateOrder,
) -> Self:
    """Convert a protobuf UpdateOrder to UpdateOrder object.

    Args:
        update_order: UpdateOrder to convert.

    Returns:
        UpdateOrder object corresponding to the protobuf message.
    """
    return cls(
        price=(
            Price.from_pb(update_order.price)
            if update_order.HasField("price")
            else None
        ),
        quantity=(
            Power.from_pb(update_order.quantity)
            if update_order.HasField("quantity")
            else None
        ),
        stop_price=(
            Price.from_pb(update_order.stop_price)
            if update_order.HasField("stop_price")
            else None
        ),
        peak_price_delta=(
            Price.from_pb(update_order.peak_price_delta)
            if update_order.HasField("peak_price_delta")
            else None
        ),
        display_quantity=(
            Power.from_pb(update_order.display_quantity)
            if update_order.HasField("display_quantity")
            else None
        ),
        execution_option=(
            OrderExecutionOption.from_pb(update_order.execution_option)
            if update_order.HasField("execution_option")
            else None
        ),
        valid_until=(
            update_order.valid_until.ToDatetime(tzinfo=timezone.utc)
            if update_order.HasField("valid_until")
            else None
        ),
        payload=(
            json_format.MessageToDict(update_order.payload)
            if update_order.payload
            else None
        ),
        tag=update_order.tag if update_order.HasField("tag") else None,
    )

to_pb

to_pb() -> UpdateOrder

Convert a UpdateOrder object to protobuf UpdateOrder.

RETURNS	DESCRIPTION
UpdateOrder	Protobuf UpdateOrder corresponding to the object.

Source code in frequenz/client/electricity_trading/_types.py

def to_pb(self) -> electricity_trading_pb2.UpdateGridpoolOrderRequest.UpdateOrder:
    """Convert a UpdateOrder object to protobuf UpdateOrder.

    Returns:
        Protobuf UpdateOrder corresponding to the object.
    """
    if self.valid_until:
        valid_until = timestamp_pb2.Timestamp()
        valid_until.FromDatetime(self.valid_until)
    else:
        valid_until = None
    return electricity_trading_pb2.UpdateGridpoolOrderRequest.UpdateOrder(
        price=self.price.to_pb() if self.price else None,
        quantity=self.quantity.to_pb() if self.quantity else None,
        stop_price=self.stop_price.to_pb() if self.stop_price else None,
        peak_price_delta=(
            self.peak_price_delta.to_pb() if self.peak_price_delta else None
        ),
        display_quantity=(
            self.display_quantity.to_pb() if self.display_quantity else None
        ),
        execution_option=(
            electricity_trading_pb2.OrderExecutionOption.ValueType(
                self.execution_option.value
            )
            if self.execution_option
            else None
        ),
        valid_until=valid_until if self.valid_until else None,
        payload=struct_pb2.Struct(fields=self.payload) if self.payload else None,
        tag=self.tag if self.tag else None,
    )

Functions

frequenz.client.electricity_trading.quantize_quantity

quantize_quantity(value: Decimal | float) -> Decimal

Convert a decimal to a quantity with the correct precision for the API.

Simply rounds the value to the correct precision using HALF_EVEN rounding.

PARAMETER	DESCRIPTION
value	The value to convert in float or Decimal. TYPE: Decimal | float

RETURNS	DESCRIPTION
Decimal	The quantity with the correct precision as a Decimal.

Source code in frequenz/client/electricity_trading/_utils.py

def quantize_quantity(value: Decimal | float) -> Decimal:
    """Convert a decimal to a quantity with the correct precision for the API.

    Simply rounds the value to the correct precision using HALF_EVEN rounding.

    Args:
        value: The value to convert in float or Decimal.

    Returns:
        The quantity with the correct precision as a Decimal.
    """
    dec = Decimal(str(value)) if isinstance(value, float) else value
    quantity_step = Decimal(f"1e-{PRECISION_DECIMAL_QUANTITY}")
    quantized = Decimal(dec).quantize(quantity_step, rounding="ROUND_HALF_EVEN")
    return quantized