electricity_trading

frequenz.client.electricity_trading

The Electricity Trading API client.

Classes

frequenz.client.electricity_trading.Client

Electricity trading client.

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

class Client:
    """Electricity trading client."""

    def __init__(self, grpc_channel: grpc.aio.Channel) -> None:
        """Initialize the client.

        Args:
            grpc_channel: gRPC channel to use for communication with the API.
        """
        self._stub = electricity_trading_pb2_grpc.ElectricityTradingServiceStub(
            grpc_channel
        )

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

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

    async def stream_gridpool_orders(  # pylint: disable=too-many-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,
    ) -> Receiver[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.
        """
        gridpool_filter = GridpoolOrderFilter(
            states=order_states,
            side=market_side,
            delivery_area=delivery_area,
            delivery_period=delivery_period,
            tag=tag,
        )

        stream_key = (gridpool_id, gridpool_filter)

        if stream_key not in self._gridpool_orders_streams:
            self._gridpool_orders_streams[stream_key] = GrpcStreamingHelper(
                f"electricity-trading-{stream_key}",
                lambda: self._stub.ReceiveGridpoolOrdersStream(  # type: ignore
                    electricity_trading_pb2.ReceiveGridpoolOrdersStreamRequest(
                        gridpool_id=gridpool_id,
                        filter=gridpool_filter.to_pb(),
                    )
                ),
                lambda response: OrderDetail.from_pb(response.order_detail),
            )
        return self._gridpool_orders_streams[stream_key].new_receiver()

    async def stream_public_trades(
        self,
        states: list[OrderState] | None = None,
        delivery_period: DeliveryPeriod | None = None,
        buy_delivery_area: DeliveryArea | None = None,
        sell_delivery_area: DeliveryArea | None = None,
    ) -> Receiver[PublicTrade]:
        """
        Stream public orders.

        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.
        """
        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:
            self._public_trades_streams[public_trade_filter] = GrpcStreamingHelper(
                f"electricity-trading-{public_trade_filter}",
                lambda: self._stub.ReceivePublicTradesStream(  # type: ignore
                    electricity_trading_pb2.ReceivePublicTradesStreamRequest(
                        filter=public_trade_filter.to_pb(),
                    )
                ),
                lambda response: PublicTrade.from_pb(response.public_trade),
            )
        return self._public_trades_streams[public_trade_filter].new_receiver()

    async def create_gridpool_order(  # pylint: disable=too-many-arguments, too-many-locals
        self,
        gridpool_id: int,
        delivery_area: DeliveryArea,
        delivery_period: DeliveryPeriod,
        order_type: OrderType,
        side: MarketSide,
        price: Price,
        quantity: Energy,
        stop_price: Price | None = None,
        peak_price_delta: Price | None = None,
        display_quantity: Energy | None = None,
        execution_option: OrderExecutionOption | None = None,
        valid_until: datetime | None = None,
        payload: dict[str, struct_pb2.Value] | None = None,
        tag: str | 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.

        Returns:
            The created order.
        """
        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,
        )

        response = await cast(
            Awaitable[electricity_trading_pb2.CreateGridpoolOrderResponse],
            self._stub.CreateGridpoolOrder(
                electricity_trading_pb2.CreateGridpoolOrderRequest(
                    gridpool_id=gridpool_id,
                    order=order.to_pb(),
                )
            ),
        )

        return OrderDetail.from_pb(response.order_detail)

    async def update_gridpool_order(  # pylint: disable=too-many-arguments, too-many-locals
        self,
        gridpool_id: int,
        order_id: int,
        price: Price | None | _Sentinel = NO_VALUE,
        quantity: Energy | None | _Sentinel = NO_VALUE,
        stop_price: Price | None | _Sentinel = NO_VALUE,
        peak_price_delta: Price | None | _Sentinel = NO_VALUE,
        display_quantity: Energy | 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,
    ) -> 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 MWh.
            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.

        Returns:
            The updated order.

        Raises:
            ValueError: If no fields to update are provided.
        """
        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,
            quantity=None if quantity is NO_VALUE else quantity,
            stop_price=None if stop_price is NO_VALUE else stop_price,
            peak_price_delta=None if peak_price_delta is NO_VALUE else peak_price_delta,
            display_quantity=None if display_quantity is NO_VALUE else display_quantity,
            execution_option=None if execution_option is NO_VALUE else execution_option,
            valid_until=None if valid_until is NO_VALUE else valid_until,
            payload=None if payload is NO_VALUE else payload,
            tag=None if tag is NO_VALUE else tag,
        )

        response = await cast(
            Awaitable[electricity_trading_pb2.UpdateGridpoolOrderResponse],
            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,
                )
            ),
        )

        return OrderDetail.from_pb(response.order_detail)

    async def cancel_gridpool_order(
        self, gridpool_id: int, order_id: int
    ) -> 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.

        Returns:
            The cancelled order.
        """
        response = await cast(
            Awaitable[electricity_trading_pb2.CancelGridpoolOrderResponse],
            self._stub.CancelGridpoolOrder(
                electricity_trading_pb2.CancelGridpoolOrderRequest(
                    gridpool_id=gridpool_id, order_id=order_id
                )
            ),
        )

        return OrderDetail.from_pb(response.order_detail)

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

        Args:
            gridpool_id: The Gridpool to cancel the orders for.

        Returns:
            The ID of the Gridpool for which the orders were cancelled.
        """
        response = await cast(
            Awaitable[electricity_trading_pb2.CancelAllGridpoolOrdersResponse],
            self._stub.CancelAllGridpoolOrders(
                electricity_trading_pb2.CancelAllGridpoolOrdersRequest(
                    gridpool_id=gridpool_id
                )
            ),
        )

        return response.gridpool_id

    async def get_gridpool_order(self, gridpool_id: int, order_id: int) -> 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.

        Returns:
            The order.
        """
        response = await cast(
            Awaitable[electricity_trading_pb2.GetGridpoolOrderResponse],
            self._stub.GetGridpoolOrder(
                electricity_trading_pb2.GetGridpoolOrderRequest(
                    gridpool_id=gridpool_id, order_id=order_id
                )
            ),
        )

        return OrderDetail.from_pb(response.order_detail)

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

        Args:
            gridpool_id: The Gridpool to retrieve the orders for.
            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.
            max_nr_orders: The maximum number of orders to return.
            page_token: The page token to use for pagination.

        Returns:
            The list of orders for that gridpool.
        """
        gridpool_order_filer = GridpoolOrderFilter(
            states=states,
            side=side,
            delivery_period=delivery_period,
            delivery_area=delivery_area,
            tag=tag,
        )

        pagination_params = PaginationParams(
            page_size=max_nr_orders,
            page_token=page_token,
        )

        response = await cast(
            Awaitable[electricity_trading_pb2.ListGridpoolOrdersResponse],
            self._stub.ListGridpoolOrders(
                electricity_trading_pb2.ListGridpoolOrdersRequest(
                    gridpool_id=gridpool_id,
                    filter=gridpool_order_filer.to_pb(),
                    pagination_params=pagination_params.to_pb(),
                )
            ),
        )

        return [
            OrderDetail.from_pb(order_detail)
            for order_detail in response.order_detail_lists
        ]

    async def list_public_trades(  # pylint: disable=too-many-arguments
        self,
        states: list[OrderState] | None = None,
        delivery_period: DeliveryPeriod | None = None,
        buy_delivery_area: DeliveryArea | None = None,
        sell_delivery_area: DeliveryArea | None = None,
        max_nr_orders: int | None = None,
        page_token: str | None = None,
    ) -> list[PublicTrade]:
        """
        List all executed public orders with optional filters.

        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.
            max_nr_orders: The maximum number of orders to return.
            page_token: The page token to use for pagination.

        Returns:
            The list of public trades.
        """
        public_trade_filter = PublicTradeFilter(
            states=states,
            delivery_period=delivery_period,
            buy_delivery_area=buy_delivery_area,
            sell_delivery_area=sell_delivery_area,
        )

        pagination_params = PaginationParams(
            page_size=max_nr_orders,
            page_token=page_token,
        )

        response = await cast(
            Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
            self._stub.ListPublicTrades(
                electricity_trading_pb2.ListPublicTradesRequest(
                    filter=public_trade_filter.to_pb(),
                    pagination_params=pagination_params.to_pb(),
                )
            ),
        )

        return [
            PublicTrade.from_pb(public_trade)
            for public_trade in response.public_trade_lists
        ]

Functions

__init__

__init__(grpc_channel: Channel) -> None

Initialize the client.

PARAMETER	DESCRIPTION
grpc_channel	gRPC channel to use for communication with the API. TYPE: Channel

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

def __init__(self, grpc_channel: grpc.aio.Channel) -> None:
    """Initialize the client.

    Args:
        grpc_channel: gRPC channel to use for communication with the API.
    """
    self._stub = electricity_trading_pb2_grpc.ElectricityTradingServiceStub(
        grpc_channel
    )

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

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

cancel_all_gridpool_orders async

cancel_all_gridpool_orders(gridpool_id: int) -> int

Cancel all orders for a specific Gridpool.

PARAMETER	DESCRIPTION
gridpool_id	The Gridpool to cancel the orders for. TYPE: int

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

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

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

    Args:
        gridpool_id: The Gridpool to cancel the orders for.

    Returns:
        The ID of the Gridpool for which the orders were cancelled.
    """
    response = await cast(
        Awaitable[electricity_trading_pb2.CancelAllGridpoolOrdersResponse],
        self._stub.CancelAllGridpoolOrders(
            electricity_trading_pb2.CancelAllGridpoolOrdersRequest(
                gridpool_id=gridpool_id
            )
        ),
    )

    return response.gridpool_id

cancel_gridpool_order async

cancel_gridpool_order(
    gridpool_id: int, order_id: int
) -> 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

RETURNS	DESCRIPTION
OrderDetail	The cancelled order.

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

async def cancel_gridpool_order(
    self, gridpool_id: int, order_id: int
) -> 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.

    Returns:
        The cancelled order.
    """
    response = await cast(
        Awaitable[electricity_trading_pb2.CancelGridpoolOrderResponse],
        self._stub.CancelGridpoolOrder(
            electricity_trading_pb2.CancelGridpoolOrderRequest(
                gridpool_id=gridpool_id, order_id=order_id
            )
        ),
    )

    return OrderDetail.from_pb(response.order_detail)

create_gridpool_order async

create_gridpool_order(
    gridpool_id: int,
    delivery_area: DeliveryArea,
    delivery_period: DeliveryPeriod,
    order_type: OrderType,
    side: MarketSide,
    price: Price,
    quantity: Energy,
    stop_price: Price | None = None,
    peak_price_delta: Price | None = None,
    display_quantity: Energy | None = None,
    execution_option: OrderExecutionOption | None = None,
    valid_until: datetime | None = None,
    payload: dict[str, Value] | None = None,
    tag: str | 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: Energy
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: Energy | 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

RETURNS	DESCRIPTION
OrderDetail	The created order.

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

async def create_gridpool_order(  # pylint: disable=too-many-arguments, too-many-locals
    self,
    gridpool_id: int,
    delivery_area: DeliveryArea,
    delivery_period: DeliveryPeriod,
    order_type: OrderType,
    side: MarketSide,
    price: Price,
    quantity: Energy,
    stop_price: Price | None = None,
    peak_price_delta: Price | None = None,
    display_quantity: Energy | None = None,
    execution_option: OrderExecutionOption | None = None,
    valid_until: datetime | None = None,
    payload: dict[str, struct_pb2.Value] | None = None,
    tag: str | 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.

    Returns:
        The created order.
    """
    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,
    )

    response = await cast(
        Awaitable[electricity_trading_pb2.CreateGridpoolOrderResponse],
        self._stub.CreateGridpoolOrder(
            electricity_trading_pb2.CreateGridpoolOrderRequest(
                gridpool_id=gridpool_id,
                order=order.to_pb(),
            )
        ),
    )

    return OrderDetail.from_pb(response.order_detail)

get_gridpool_order async

get_gridpool_order(
    gridpool_id: int, order_id: int
) -> 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

RETURNS	DESCRIPTION
OrderDetail	The order.

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

async def get_gridpool_order(self, gridpool_id: int, order_id: int) -> 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.

    Returns:
        The order.
    """
    response = await cast(
        Awaitable[electricity_trading_pb2.GetGridpoolOrderResponse],
        self._stub.GetGridpoolOrder(
            electricity_trading_pb2.GetGridpoolOrderRequest(
                gridpool_id=gridpool_id, order_id=order_id
            )
        ),
    )

    return OrderDetail.from_pb(response.order_detail)

list_gridpool_orders async

list_gridpool_orders(
    gridpool_id: int,
    states: list[OrderState] | None = None,
    side: MarketSide | None = None,
    delivery_period: DeliveryPeriod | None = None,
    delivery_area: DeliveryArea | None = None,
    tag: str | None = None,
    max_nr_orders: int | None = None,
    page_token: str | None = None,
) -> list[OrderDetail]

List orders for a specific Gridpool with optional filters.

PARAMETER	DESCRIPTION
gridpool_id	The Gridpool to retrieve the orders for. TYPE: int
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
max_nr_orders	The maximum number of orders to return. TYPE: int | None DEFAULT: None
page_token	The page token to use for pagination. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
list[OrderDetail]	The list of orders for that gridpool.

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

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

    Args:
        gridpool_id: The Gridpool to retrieve the orders for.
        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.
        max_nr_orders: The maximum number of orders to return.
        page_token: The page token to use for pagination.

    Returns:
        The list of orders for that gridpool.
    """
    gridpool_order_filer = GridpoolOrderFilter(
        states=states,
        side=side,
        delivery_period=delivery_period,
        delivery_area=delivery_area,
        tag=tag,
    )

    pagination_params = PaginationParams(
        page_size=max_nr_orders,
        page_token=page_token,
    )

    response = await cast(
        Awaitable[electricity_trading_pb2.ListGridpoolOrdersResponse],
        self._stub.ListGridpoolOrders(
            electricity_trading_pb2.ListGridpoolOrdersRequest(
                gridpool_id=gridpool_id,
                filter=gridpool_order_filer.to_pb(),
                pagination_params=pagination_params.to_pb(),
            )
        ),
    )

    return [
        OrderDetail.from_pb(order_detail)
        for order_detail in response.order_detail_lists
    ]

list_public_trades async

list_public_trades(
    states: list[OrderState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
    max_nr_orders: int | None = None,
    page_token: str | None = None,
) -> list[PublicTrade]

List all executed public orders with optional filters.

PARAMETER	DESCRIPTION
states	List of order states to filter by. TYPE: list[OrderState] | 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
max_nr_orders	The maximum number of orders to return. TYPE: int | None DEFAULT: None
page_token	The page token to use for pagination. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
list[PublicTrade]	The list of public trades.

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

async def list_public_trades(  # pylint: disable=too-many-arguments
    self,
    states: list[OrderState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
    max_nr_orders: int | None = None,
    page_token: str | None = None,
) -> list[PublicTrade]:
    """
    List all executed public orders with optional filters.

    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.
        max_nr_orders: The maximum number of orders to return.
        page_token: The page token to use for pagination.

    Returns:
        The list of public trades.
    """
    public_trade_filter = PublicTradeFilter(
        states=states,
        delivery_period=delivery_period,
        buy_delivery_area=buy_delivery_area,
        sell_delivery_area=sell_delivery_area,
    )

    pagination_params = PaginationParams(
        page_size=max_nr_orders,
        page_token=page_token,
    )

    response = await cast(
        Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
        self._stub.ListPublicTrades(
            electricity_trading_pb2.ListPublicTradesRequest(
                filter=public_trade_filter.to_pb(),
                pagination_params=pagination_params.to_pb(),
            )
        ),
    )

    return [
        PublicTrade.from_pb(public_trade)
        for public_trade in response.public_trade_lists
    ]

stream_gridpool_orders async

stream_gridpool_orders(
    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,
) -> Receiver[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
Receiver[OrderDetail]	Async generator of orders.

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

async def stream_gridpool_orders(  # pylint: disable=too-many-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,
) -> Receiver[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.
    """
    gridpool_filter = GridpoolOrderFilter(
        states=order_states,
        side=market_side,
        delivery_area=delivery_area,
        delivery_period=delivery_period,
        tag=tag,
    )

    stream_key = (gridpool_id, gridpool_filter)

    if stream_key not in self._gridpool_orders_streams:
        self._gridpool_orders_streams[stream_key] = GrpcStreamingHelper(
            f"electricity-trading-{stream_key}",
            lambda: self._stub.ReceiveGridpoolOrdersStream(  # type: ignore
                electricity_trading_pb2.ReceiveGridpoolOrdersStreamRequest(
                    gridpool_id=gridpool_id,
                    filter=gridpool_filter.to_pb(),
                )
            ),
            lambda response: OrderDetail.from_pb(response.order_detail),
        )
    return self._gridpool_orders_streams[stream_key].new_receiver()

stream_public_trades async

stream_public_trades(
    states: list[OrderState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
) -> Receiver[PublicTrade]

Stream public orders.

PARAMETER	DESCRIPTION
states	List of order states to filter for. TYPE: list[OrderState] | 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
Receiver[PublicTrade]	Async generator of orders.

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

async def stream_public_trades(
    self,
    states: list[OrderState] | None = None,
    delivery_period: DeliveryPeriod | None = None,
    buy_delivery_area: DeliveryArea | None = None,
    sell_delivery_area: DeliveryArea | None = None,
) -> Receiver[PublicTrade]:
    """
    Stream public orders.

    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.
    """
    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:
        self._public_trades_streams[public_trade_filter] = GrpcStreamingHelper(
            f"electricity-trading-{public_trade_filter}",
            lambda: self._stub.ReceivePublicTradesStream(  # type: ignore
                electricity_trading_pb2.ReceivePublicTradesStreamRequest(
                    filter=public_trade_filter.to_pb(),
                )
            ),
            lambda response: PublicTrade.from_pb(response.public_trade),
        )
    return self._public_trades_streams[public_trade_filter].new_receiver()

update_gridpool_order async

update_gridpool_order(
    gridpool_id: int,
    order_id: int,
    price: Price | None | _Sentinel = NO_VALUE,
    quantity: Energy | None | _Sentinel = NO_VALUE,
    stop_price: Price | None | _Sentinel = NO_VALUE,
    peak_price_delta: Price | None | _Sentinel = NO_VALUE,
    display_quantity: Energy | 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,
) -> 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 MWh. TYPE: Energy | 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: Energy | 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

RETURNS	DESCRIPTION
OrderDetail	The updated order.

RAISES	DESCRIPTION
ValueError	If no fields to update are provided.

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

async def update_gridpool_order(  # pylint: disable=too-many-arguments, too-many-locals
    self,
    gridpool_id: int,
    order_id: int,
    price: Price | None | _Sentinel = NO_VALUE,
    quantity: Energy | None | _Sentinel = NO_VALUE,
    stop_price: Price | None | _Sentinel = NO_VALUE,
    peak_price_delta: Price | None | _Sentinel = NO_VALUE,
    display_quantity: Energy | 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,
) -> 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 MWh.
        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.

    Returns:
        The updated order.

    Raises:
        ValueError: If no fields to update are provided.
    """
    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,
        quantity=None if quantity is NO_VALUE else quantity,
        stop_price=None if stop_price is NO_VALUE else stop_price,
        peak_price_delta=None if peak_price_delta is NO_VALUE else peak_price_delta,
        display_quantity=None if display_quantity is NO_VALUE else display_quantity,
        execution_option=None if execution_option is NO_VALUE else execution_option,
        valid_until=None if valid_until is NO_VALUE else valid_until,
        payload=None if payload is NO_VALUE else payload,
        tag=None if tag is NO_VALUE else tag,
    )

    response = await cast(
        Awaitable[electricity_trading_pb2.UpdateGridpoolOrderResponse],
        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,
            )
        ),
    )

    return OrderDetail.from_pb(response.order_detail)

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: delivery_area_pb2.EnergyMarketCodeType.ValueType
    """Type of code used for identifying the delivery area itself."""

    @classmethod
    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=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)

Attributes

code instance-attribute

code: str

Code representing the unique identifier for the delivery area.

code_type instance-attribute

code_type: ValueType

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
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=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)

frequenz.client.electricity_trading.DeliveryPeriod dataclass

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

@dataclass(frozen=True)
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 __post_init__(self) -> None:
        """Validate the parameters."""
        if not isinstance(self.start, datetime):
            raise ValueError("Start must be a datetime object.")
        if not isinstance(self.duration, DeliveryDuration):
            raise ValueError("Duration must be a DeliveryDuration object.")

    @classmethod
    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.
        """
        return cls(
            start=delivery_period.start.ToDatetime(),
            duration=DeliveryDuration.from_pb(delivery_period.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

The length of the delivery period.

start instance-attribute

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.

Functions

__post_init__

__post_init__() -> None

Validate the parameters.

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

def __post_init__(self) -> None:
    """Validate the parameters."""
    if not isinstance(self.start, datetime):
        raise ValueError("Start must be a datetime object.")
    if not isinstance(self.duration, DeliveryDuration):
        raise ValueError("Duration must be a DeliveryDuration object.")

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.

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

@classmethod
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.
    """
    return cls(
        start=delivery_period.start.ToDatetime(),
        duration=DeliveryDuration.from_pb(delivery_period.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.Energy dataclass

Represents energy unit in Megawatthours (MWh).

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

@dataclass(frozen=True)
class Energy:
    """Represents energy unit in Megawatthours (MWh)."""

    mwh: Decimal

    @classmethod
    def from_pb(cls, energy: energy_pb2.Energy) -> Self:
        """Convert a protobuf Energy to Energy object.

        Args:
            energy: Energy to convert.

        Returns:
            Energy object corresponding to the protobuf message.
        """
        return cls(mwh=Decimal(energy.mwh.value))

    def to_pb(self) -> energy_pb2.Energy:
        """Convert a Energy object to protobuf Energy.

        Returns:
            Protobuf message corresponding to the Energy object.
        """
        decimal_mwh = decimal_pb2.Decimal()
        decimal_mwh.value = str(self.mwh)
        return energy_pb2.Energy(mwh=decimal_mwh)

Functions

from_pb classmethod

from_pb(energy: Energy) -> Self

Convert a protobuf Energy to Energy object.

PARAMETER	DESCRIPTION
energy	Energy to convert. TYPE: Energy

RETURNS	DESCRIPTION
Self	Energy object corresponding to the protobuf message.

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

@classmethod
def from_pb(cls, energy: energy_pb2.Energy) -> Self:
    """Convert a protobuf Energy to Energy object.

    Args:
        energy: Energy to convert.

    Returns:
        Energy object corresponding to the protobuf message.
    """
    return cls(mwh=Decimal(energy.mwh.value))

to_pb

to_pb() -> Energy

Convert a Energy object to protobuf Energy.

RETURNS	DESCRIPTION
Energy	Protobuf message corresponding to the Energy object.

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

def to_pb(self) -> energy_pb2.Energy:
    """Convert a Energy object to protobuf Energy.

    Returns:
        Protobuf message corresponding to the Energy object.
    """
    decimal_mwh = decimal_pb2.Decimal()
    decimal_mwh.value = str(self.mwh)
    return energy_pb2.Energy(mwh=decimal_mwh)

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

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

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

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

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

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

    @classmethod
    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(
            states=[
                OrderState.from_pb(state) for state in gridpool_order_filter.states
            ],
            side=MarketSide.from_pb(gridpool_order_filter.side),
            delivery_period=DeliveryPeriod.from_pb(
                gridpool_order_filter.delivery_period
            ),
            delivery_area=DeliveryArea.from_pb(gridpool_order_filter.delivery_area),
            tag=gridpool_order_filter.tag,
        )

    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.states
            ]
            if self.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 instance-attribute

delivery_area: DeliveryArea | None

Delivery area to filter for.

delivery_period instance-attribute

delivery_period: DeliveryPeriod | None

Delivery period to filter for.

side instance-attribute

side: MarketSide | None

Market side to filter for.

states instance-attribute

states: list[OrderState] | None

List of order states to filter for.

tag instance-attribute

tag: str | None

Tag associated with the orders to be filtered.

Functions

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
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(
        states=[
            OrderState.from_pb(state) for state in gridpool_order_filter.states
        ],
        side=MarketSide.from_pb(gridpool_order_filter.side),
        delivery_period=DeliveryPeriod.from_pb(
            gridpool_order_filter.delivery_period
        ),
        delivery_area=DeliveryArea.from_pb(gridpool_order_filter.delivery_area),
        tag=gridpool_order_filter.tag,
    )

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.states
        ]
        if self.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.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
    def from_pb(cls, market_side: electricity_trading_pb2.MarketSide.ValueType) -> Self:
        """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) -> Self

Convert a protobuf MarketSide value to MarketSide enum.

PARAMETER	DESCRIPTION
market_side	Market side to convert. TYPE: ValueType

RETURNS	DESCRIPTION
Self	Enum value corresponding to the protobuf message.

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

@classmethod
def from_pb(cls, market_side: electricity_trading_pb2.MarketSide.ValueType) -> Self:
    """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(frozen=True)
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: Energy
    """The quantity of the contract being traded."""

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

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

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

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

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

    payload: dict[str, struct_pb2.Value] | 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
    """User-defined tag to group related orders."""

    @classmethod
    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),
            quantity=Energy.from_pb(order.quantity),
            stop_price=Price.from_pb(order.stop_price) if order.stop_price else None,
            peak_price_delta=Price.from_pb(order.peak_price_delta)
            if order.peak_price_delta
            else None,
            display_quantity=Energy.from_pb(order.display_quantity)
            if order.display_quantity
            else None,
            execution_option=OrderExecutionOption.from_pb(order.execution_option)
            if order.execution_option
            else None,
            valid_until=order.valid_until.ToDatetime() if order.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,
        )

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 instance-attribute

display_quantity: Energy | None

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

execution_option instance-attribute

execution_option: OrderExecutionOption | None

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

payload instance-attribute

payload: dict[str, Value] | 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 instance-attribute

peak_price_delta: Price | 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: Energy

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 instance-attribute

stop_price: Price | None

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

tag instance-attribute

tag: str | None

User-defined tag to group related orders.

type instance-attribute

type: OrderType

The type of order.

valid_until instance-attribute

valid_until: datetime | None

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

Functions

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
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),
        quantity=Energy.from_pb(order.quantity),
        stop_price=Price.from_pb(order.stop_price) if order.stop_price else None,
        peak_price_delta=Price.from_pb(order.peak_price_delta)
        if order.peak_price_delta
        else None,
        display_quantity=Energy.from_pb(order.display_quantity)
        if order.display_quantity
        else None,
        execution_option=OrderExecutionOption.from_pb(order.execution_option)
        if order.execution_option
        else None,
        valid_until=order.valid_until.ToDatetime() if order.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: Energy
filled_quantity	Filled quantity for this order. TYPE: Energy
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(frozen=True)
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: Energy
    filled_quantity: Energy
    create_time: datetime
    modification_time: datetime

    @classmethod
    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.
        """
        return 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=Energy.from_pb(order_detail.open_quantity),
            filled_quantity=Energy.from_pb(order_detail.filled_quantity),
            create_time=order_detail.create_time.ToDatetime(),
            modification_time=order_detail.modification_time.ToDatetime(),
        )

Functions

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.

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

@classmethod
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.
    """
    return 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=Energy.from_pb(order_detail.open_quantity),
        filled_quantity=Energy.from_pb(order_detail.filled_quantity),
        create_time=order_detail.create_time.ToDatetime(),
        modification_time=order_detail.modification_time.ToDatetime(),
    )

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.

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

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

    NONE = electricity_trading_pb2.OrderExecutionOption.ORDER_EXECUTION_OPTION_NONE
    """Order remains open until it's fully filled, cancelled by the client,
    `valid_until` timestamp is reached, or the end of the trading session."""

    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
    def from_pb(
        cls,
        order_execution_option: electricity_trading_pb2.OrderExecutionOption.ValueType,
    ) -> Self:
        """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)

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.

NONE class-attribute instance-attribute

NONE = ORDER_EXECUTION_OPTION_NONE

Order remains open until it's fully filled, cancelled by the client, valid_until timestamp is reached, or the end of the trading session.

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) -> Self

Convert a protobuf OrderExecutionOption value to OrderExecutionOption enum.

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

RETURNS	DESCRIPTION
Self	Enum value corresponding to the protobuf message.

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

@classmethod
def from_pb(
    cls,
    order_execution_option: electricity_trading_pb2.OrderExecutionOption.ValueType,
) -> Self:
    """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)

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

    CANCEL_REQUESTED = electricity_trading_pb2.OrderState.ORDER_STATE_CANCEL_REQUESTED
    """A cancellation request for the order has been submitted but the order is not yet removed
    from the order book."""

    CANCEL_REJECTED = electricity_trading_pb2.OrderState.ORDER_STATE_CANCEL_REJECTED
    """The order cancellation request was rejected, likely due to it having already been filled or
    expired."""

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

    RECALL = electricity_trading_pb2.OrderState.ORDER_STATE_RECALL
    """The order has been recalled. This could be due to a system issue or a request from the
    market participant or market operator."""

    @classmethod
    def from_pb(cls, order_state: electricity_trading_pb2.OrderState.ValueType) -> Self:
        """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.

CANCEL_REJECTED class-attribute instance-attribute

CANCEL_REJECTED = ORDER_STATE_CANCEL_REJECTED

The order cancellation request was rejected, likely due to it having already been filled or expired.

CANCEL_REQUESTED class-attribute instance-attribute

CANCEL_REQUESTED = ORDER_STATE_CANCEL_REQUESTED

A cancellation request for the order has been submitted but the order is not yet removed from the order book.

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.

RECALL class-attribute instance-attribute

RECALL = ORDER_STATE_RECALL

The order has been recalled. This could be due to a system issue or a request from the market participant or market operator.

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) -> Self

Convert a protobuf OrderState value to OrderState enum.

PARAMETER	DESCRIPTION
order_state	Order state to convert. TYPE: ValueType

RETURNS	DESCRIPTION
Self	Enum value corresponding to the protobuf message.

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

@classmethod
def from_pb(cls, order_state: electricity_trading_pb2.OrderState.ValueType) -> Self:
    """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
    def from_pb(cls, order_type: electricity_trading_pb2.OrderType.ValueType) -> Self:
        """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) -> Self

Convert a protobuf OrderType value to OrderType enum.

PARAMETER	DESCRIPTION
order_type	Order type to convert. TYPE: ValueType

RETURNS	DESCRIPTION
Self	Enum value corresponding to the protobuf message.

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

@classmethod
def from_pb(cls, order_type: electricity_trading_pb2.OrderType.ValueType) -> Self:
    """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.PaginationParams dataclass

Parameters for paginating list requests.

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

@dataclass(frozen=True)
class PaginationParams:
    """Parameters for paginating list requests."""

    page_size: int | None = None
    """The maximum number of results to be returned per request."""

    page_token: str | None = None
    """The token identifying a specific page of the list results."""

    @classmethod
    def from_pb(cls, pagination_params: pagination_params_pb2.PaginationParams) -> Self:
        """Convert a protobuf PaginationParams to PaginationParams object.

        Args:
            pagination_params: PaginationParams to convert.

        Returns:
            PaginationParams object corresponding to the protobuf message.
        """
        return cls(
            page_size=pagination_params.page_size,
            page_token=pagination_params.page_token,
        )

    def to_pb(self) -> pagination_params_pb2.PaginationParams:
        """Convert a PaginationParams object to protobuf PaginationParams.

        Returns:
            Protobuf message corresponding to the PaginationParams object.
        """
        return pagination_params_pb2.PaginationParams(
            page_size=self.page_size,
            page_token=self.page_token,
        )

Attributes

page_size class-attribute instance-attribute

page_size: int | None = None

The maximum number of results to be returned per request.

page_token class-attribute instance-attribute

page_token: str | None = None

The token identifying a specific page of the list results.

Functions

from_pb classmethod

from_pb(pagination_params: PaginationParams) -> Self

Convert a protobuf PaginationParams to PaginationParams object.

PARAMETER	DESCRIPTION
pagination_params	PaginationParams to convert. TYPE: PaginationParams

RETURNS	DESCRIPTION
Self	PaginationParams object corresponding to the protobuf message.

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

@classmethod
def from_pb(cls, pagination_params: pagination_params_pb2.PaginationParams) -> Self:
    """Convert a protobuf PaginationParams to PaginationParams object.

    Args:
        pagination_params: PaginationParams to convert.

    Returns:
        PaginationParams object corresponding to the protobuf message.
    """
    return cls(
        page_size=pagination_params.page_size,
        page_token=pagination_params.page_token,
    )

to_pb

to_pb() -> PaginationParams

Convert a PaginationParams object to protobuf PaginationParams.

RETURNS	DESCRIPTION
PaginationParams	Protobuf message corresponding to the PaginationParams object.

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

def to_pb(self) -> pagination_params_pb2.PaginationParams:
    """Convert a PaginationParams object to protobuf PaginationParams.

    Returns:
        Protobuf message corresponding to the PaginationParams object.
    """
    return pagination_params_pb2.PaginationParams(
        page_size=self.page_size,
        page_token=self.page_token,
    )

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
    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())

Attributes

amount instance-attribute

amount: Decimal

Amount of the price.

currency instance-attribute

currency: Currency

Currency of the price.

Functions

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
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(frozen=True)
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."""

    modification_time: datetime
    """UTC Timestamp of the last order update or matching."""

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

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

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

    @classmethod
    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),
            modification_time=public_trade.modification_time.ToDatetime(),
            price=Price.from_pb(public_trade.price),
            quantity=Energy.from_pb(public_trade.quantity),
            state=OrderState.from_pb(public_trade.state),
        )

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.

modification_time instance-attribute

modification_time: datetime

UTC Timestamp of the last order update or matching.

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: Energy

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: OrderState

State of the order.

Functions

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
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),
        modification_time=public_trade.modification_time.ToDatetime(),
        price=Price.from_pb(public_trade.price),
        quantity=Energy.from_pb(public_trade.quantity),
        state=OrderState.from_pb(public_trade.state),
    )

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[OrderState] | None
    """List of order states to filter for."""

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

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

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

    @classmethod
    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=[OrderState.from_pb(state) for state in public_trade_filter.states],
            delivery_period=DeliveryPeriod.from_pb(public_trade_filter.delivery_period),
            buy_delivery_area=DeliveryArea.from_pb(
                public_trade_filter.buy_delivery_area
            ),
            sell_delivery_area=DeliveryArea.from_pb(
                public_trade_filter.sell_delivery_area
            ),
        )

    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.OrderState.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 instance-attribute

buy_delivery_area: DeliveryArea | None

Delivery area to filter for on the buy side.

delivery_period instance-attribute

delivery_period: DeliveryPeriod | None

Delivery period to filter for.

sell_delivery_area instance-attribute

sell_delivery_area: DeliveryArea | None

Delivery area to filter for on the sell side.

states instance-attribute

states: list[OrderState] | None

List of order states to filter for.

Functions

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
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=[OrderState.from_pb(state) for state in public_trade_filter.states],
        delivery_period=DeliveryPeriod.from_pb(public_trade_filter.delivery_period),
        buy_delivery_area=DeliveryArea.from_pb(
            public_trade_filter.buy_delivery_area
        ),
        sell_delivery_area=DeliveryArea.from_pb(
            public_trade_filter.sell_delivery_area
        ),
    )

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.OrderState.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.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(frozen=True)
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
    """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: Energy | None
    """The updated quantity of the contract being traded, specified in MWh."""

    stop_price: Price | None
    """Applicable for STOP_LIMIT orders. This is the updated stop price that triggers
    the limit order."""

    peak_price_delta: Price | None
    """Applicable for ICEBERG orders. This is the updated price difference
    between the peak price and the limit price."""

    display_quantity: Energy | None
    """Applicable for ICEBERG orders. This is the updated quantity of the order
    to be displayed in the order book."""

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

    valid_until: datetime | 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
    """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
    """Updated user-defined tag to group related orders."""

    @classmethod
    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=Energy.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=Energy.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()
            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 instance-attribute

display_quantity: Energy | None

Applicable for ICEBERG orders. This is the updated quantity of the order to be displayed in the order book.

execution_option instance-attribute

execution_option: OrderExecutionOption | None

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

payload instance-attribute

payload: dict[str, Value] | 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 instance-attribute

peak_price_delta: Price | None

Applicable for ICEBERG orders. This is the updated price difference between the peak price and the limit price.

price instance-attribute

price: Price | 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 instance-attribute

quantity: Energy | None

The updated quantity of the contract being traded, specified in MWh.

stop_price instance-attribute

stop_price: Price | None

Applicable for STOP_LIMIT orders. This is the updated stop price that triggers the limit order.

tag instance-attribute

tag: str | None

Updated user-defined tag to group related orders.

valid_until instance-attribute

valid_until: datetime | 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

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
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=Energy.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=Energy.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()
        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,
    )