etrading

frequenz.client.electricity_trading.cli.etrading

CLI tool to interact with the trading API.

Classes

Functions

frequenz.client.electricity_trading.cli.etrading.cancel_order async

cancel_order(
    url: str,
    auth_key: str,
    *,
    gridpool_id: int,
    order_id: int | None,
    sign_secret: str | None = None
) -> None

Cancel an order by order ID.

If order_id is None, cancel all orders in the gridpool.

PARAMETER	DESCRIPTION
url	URL of the trading API. TYPE: str
auth_key	API key. TYPE: str
gridpool_id	Gridpool ID. TYPE: int
order_id	Order ID to cancel or None to cancel all orders. TYPE: int | None
sign_secret	The cryptographic secret to use for HMAC generation. TYPE: str | None DEFAULT: None

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

async def cancel_order(
    url: str,
    auth_key: str,
    *,
    gridpool_id: int,
    order_id: int | None,
    sign_secret: str | None = None,
) -> None:
    """Cancel an order by order ID.

    If order_id is None, cancel all orders in the gridpool.

    Args:
        url: URL of the trading API.
        auth_key: API key.
        gridpool_id: Gridpool ID.
        order_id: Order ID to cancel or None to cancel all orders.
        sign_secret: The cryptographic secret to use for HMAC generation.
    """
    client = Client(server_url=url, auth_key=auth_key, sign_secret=sign_secret)
    if order_id is None:
        await client.cancel_all_gridpool_orders(gridpool_id)
    else:
        await client.cancel_gridpool_order(gridpool_id, order_id=order_id)

frequenz.client.electricity_trading.cli.etrading.check_delivery_start

check_delivery_start(
    ts: datetime,
    duration: timedelta = timedelta(minutes=15),
) -> None

Validate that the delivery start is a multiple of duration.

PARAMETER	DESCRIPTION
ts	Delivery start timestamp. TYPE: datetime
duration	Delivery period duration. TYPE: timedelta DEFAULT: timedelta(minutes=15)

RAISES	DESCRIPTION
ValueError	If ts is not a multiple of duration.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def check_delivery_start(
    ts: datetime, duration: timedelta = timedelta(minutes=15)
) -> None:
    """Validate that the delivery start is a multiple of duration.

    Args:
        ts: Delivery start timestamp.
        duration: Delivery period duration.

    Raises:
        ValueError: If `ts` is not a multiple of `duration`.
    """
    if int(ts.timestamp()) % int(duration.total_seconds()) != 0:
        raise ValueError("Delivery period must be a multiple of `duration`.")

frequenz.client.electricity_trading.cli.etrading.create_order async

create_order(
    url: str,
    auth_key: str,
    *,
    gid: int,
    delivery_start: datetime,
    delivery_area: str,
    price: str,
    quantity_mw: str,
    currency: str,
    duration: timedelta,
    sign_secret: str | None = None,
    tag: str | None = None
) -> None

Create a limit order for a given price and quantity (in MW).

The market side is determined by the sign of the quantity, positive for buy orders and negative for sell orders. The delivery area code is expected to be in EUROPE_EIC format.

PARAMETER	DESCRIPTION
url	URL of the trading API. TYPE: str
auth_key	API key. TYPE: str
gid	Gridpool ID. TYPE: int
delivery_start	Start of the delivery period. TYPE: datetime
delivery_area	Delivery area code. TYPE: str
price	Price of the order. TYPE: str
quantity_mw	Quantity in MW, positive for buy orders and negative for sell orders. TYPE: str
currency	Currency of the price. TYPE: str
duration	Duration of the delivery period. TYPE: timedelta
sign_secret	The cryptographic secret to use for HMAC generation. TYPE: str | None DEFAULT: None
tag	Optional tag to attach to the order. TYPE: str | None DEFAULT: None

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

async def create_order(
    url: str,
    auth_key: str,
    *,
    gid: int,
    delivery_start: datetime,
    delivery_area: str,
    price: str,
    quantity_mw: str,
    currency: str,
    duration: timedelta,
    sign_secret: str | None = None,
    tag: str | None = None,
) -> None:
    """Create a limit order for a given price and quantity (in MW).

    The market side is determined by the sign of the quantity, positive for buy orders
    and negative for sell orders. The delivery area code is expected to be in
    EUROPE_EIC format.

    Args:
        url: URL of the trading API.
        auth_key: API key.
        gid: Gridpool ID.
        delivery_start: Start of the delivery period.
        delivery_area: Delivery area code.
        price: Price of the order.
        quantity_mw: Quantity in MW, positive for buy orders and negative for sell orders.
        currency: Currency of the price.
        duration: Duration of the delivery period.
        sign_secret: The cryptographic secret to use for HMAC generation.
        tag: Optional tag to attach to the order.
    """
    client = Client(server_url=url, auth_key=auth_key, sign_secret=sign_secret)

    side = MarketSide.SELL if quantity_mw[0] == "-" else MarketSide.BUY
    quantity = Power(mw=Decimal(quantity_mw.lstrip("-")))
    check_delivery_start(delivery_start)
    order = await client.create_gridpool_order(
        gridpool_id=gid,
        delivery_area=DeliveryArea(
            code=delivery_area,
            code_type=EnergyMarketCodeType.EUROPE_EIC,
        ),
        delivery_period=DeliveryPeriod(
            start=delivery_start,
            duration=DeliveryDuration.from_timedelta(duration),
        ),
        order_type=OrderType.LIMIT,
        side=side,
        price=Price(
            amount=Decimal(price),
            currency=Currency[currency],
        ),
        quantity=quantity,
        tag=tag,
    )

    print_order(order, gid)

frequenz.client.electricity_trading.cli.etrading.list_gridpool_orders async

list_gridpool_orders(
    url: str,
    auth_key: str,
    *,
    delivery_from: datetime | None,
    delivery_to: datetime | None,
    gid: int,
    sign_secret: str | None = None
) -> None

List orders and stream new gridpool orders.

Optionally orders can be filtered by delivery period.

Note that retrieved sort order for listed orders (starting from the newest) is reversed in chunks trying to bring more recent orders to the bottom.

PARAMETER	DESCRIPTION
url	URL of the trading API. TYPE: str
auth_key	API key. TYPE: str
delivery_from	Start timestamp (inclusive) to filter delivery start times or None. TYPE: datetime | None
delivery_to	End timestamp (exclusive) to filter delivery start times or None. TYPE: datetime | None
gid	Gridpool ID. TYPE: int
sign_secret	The cryptographic secret to use for HMAC generation. TYPE: str | None DEFAULT: None

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

async def list_gridpool_orders(
    url: str,
    auth_key: str,
    *,
    delivery_from: datetime | None,
    delivery_to: datetime | None,
    gid: int,
    sign_secret: str | None = None,
) -> None:
    """List orders and stream new gridpool orders.

    Optionally orders can be filtered by delivery period.

    Note that retrieved sort order for listed orders (starting from the newest)
    is reversed in chunks trying to bring more recent orders to the bottom.

    Args:
        url: URL of the trading API.
        auth_key: API key.
        delivery_from: Start timestamp (inclusive) to filter delivery start times or None.
        delivery_to: End timestamp (exclusive) to filter delivery start times or None.
        gid: Gridpool ID.
        sign_secret: The cryptographic secret to use for HMAC generation.
    """
    client = Client(server_url=url, auth_key=auth_key, sign_secret=sign_secret)

    print_order_header()

    delivery_time_filter = DeliveryTimeFilter(Interval(delivery_from, delivery_to))

    lst = client.list_gridpool_orders(
        gid,
        delivery_time_filter=delivery_time_filter,
    )

    stream = client.gridpool_orders_stream(
        gid,
        delivery_time_filter=delivery_time_filter,
    ).new_receiver()

    async for order in reverse_iterator(lst):
        print_order(order, gid)

    async for order in stream:
        print_order(order, gid)

frequenz.client.electricity_trading.cli.etrading.list_gridpool_trades async

list_gridpool_trades(
    url: str,
    auth_key: str,
    gid: int,
    *,
    delivery_from: datetime | None,
    delivery_to: datetime | None,
    sign_secret: str | None = None
) -> None

List gridpool trades and stream new gridpool trades.

Optionally trades can be filtered by delivery period.

PARAMETER	DESCRIPTION
url	URL of the trading API. TYPE: str
auth_key	API key. TYPE: str
gid	Gridpool ID. TYPE: int
delivery_from	Start timestamp (inclusive) to filter delivery start times or None. TYPE: datetime | None
delivery_to	End timestamp (exclusive) to filter delivery start times or None. TYPE: datetime | None
sign_secret	The cryptographic secret to use for HMAC generation. TYPE: str | None DEFAULT: None

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

async def list_gridpool_trades(
    url: str,
    auth_key: str,
    gid: int,
    *,
    delivery_from: datetime | None,
    delivery_to: datetime | None,
    sign_secret: str | None = None,
) -> None:
    """List gridpool trades and stream new gridpool trades.

    Optionally trades can be filtered by delivery period.

    Args:
        url: URL of the trading API.
        auth_key: API key.
        gid: Gridpool ID.
        delivery_from: Start timestamp (inclusive) to filter delivery start times or None.
        delivery_to: End timestamp (exclusive) to filter delivery start times or None.
        sign_secret: The cryptographic secret to use for HMAC generation.
    """
    client = Client(server_url=url, auth_key=auth_key, sign_secret=sign_secret)

    print_trade_header()

    delivery_time_filter = DeliveryTimeFilter(Interval(delivery_from, delivery_to))
    lst = client.list_gridpool_trades(
        gid,
        delivery_time_filter=delivery_time_filter,
    )

    # Initialize the stream before printing to minimize the gap between the two
    stream = client.gridpool_trades_stream(
        gid,
        delivery_time_filter=delivery_time_filter,
    ).new_receiver()

    async for trade in lst:
        print_trade(trade, gid)

    async for trade in stream:
        print_trade(trade, gid)

frequenz.client.electricity_trading.cli.etrading.print_order

print_order(order: OrderDetail, gid: int) -> None

Print order details to stdout in CSV format.

All fields except the following are printed: - order.stop_price - order.peak_price_delta - order.display_quantity - order.execution_option - order.valid_until - order.payload - state_detail.state_reason - state_detail.market_actor - open_quantity

PARAMETER	DESCRIPTION
order	OrderDetail object TYPE: OrderDetail
gid	Gridpool ID TYPE: int

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_order(order: OrderDetail, gid: int) -> None:
    """
    Print order details to stdout in CSV format.

    All fields except the following are printed:
    - order.stop_price
    - order.peak_price_delta
    - order.display_quantity
    - order.execution_option
    - order.valid_until
    - order.payload
    - state_detail.state_reason
    - state_detail.market_actor
    - open_quantity

    Args:
        order: OrderDetail object
        gid: Gridpool ID
    """
    values = [
        order.order_id,
        order.create_time.isoformat(),
        order.modification_time.isoformat(),
        order.order.delivery_period.start.isoformat(),
        order.order.delivery_period.duration,
        order.order.delivery_area.code,
        order.order.delivery_area.code_type,
        order.order.type,
        order.order.quantity.mw,
        order.filled_quantity.mw,
        order.order.side,
        order.order.price.currency,
        order.order.price.amount,
        order.state_detail.state,
        order.order.tag,
        gid,
    ]
    print(",".join(v.name if isinstance(v, Enum) else str(v) for v in values))

frequenz.client.electricity_trading.cli.etrading.print_order_header

print_order_header() -> None

Print order header in CSV format.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_order_header() -> None:
    """Print order header in CSV format."""
    header = (
        "order_id,"
        "create_time,"
        "modification_time,"
        "delivery_period_start,"
        "delivery_period_duration,"
        "delivery_area_code,"
        "delivery_area_code_type,"
        "order_type,"
        "quantity_mw,"
        "filled_quantity_mw,"
        "side,"
        "currency,"
        "price,"
        "state,"
        "tag,"
        "gridpool_id"
    )
    print(header)

frequenz.client.electricity_trading.cli.etrading.print_public_order

print_public_order(order: PublicOrder) -> None

Print public order details to stdout in CSV format.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_public_order(order: PublicOrder) -> None:
    """Print public order details to stdout in CSV format."""
    values = (
        order.public_order_id,
        order.create_time.isoformat(),
        order.update_time.isoformat(),
        order.delivery_period.start.isoformat(),
        order.delivery_period.duration,
        order.delivery_area.code,
        order.quantity.mw,
        order.side,
        order.price.amount,
        order.price.currency,
        order.type,
        order.execution_option,
    )
    print(",".join(v.name if isinstance(v, Enum) else str(v) for v in values))

frequenz.client.electricity_trading.cli.etrading.print_public_orders_header

print_public_orders_header() -> None

Print public order header in CSV format.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_public_orders_header() -> None:
    """Print public order header in CSV format."""
    header = (
        "public_order_id,"
        "create_time,"
        "update_time,"
        "delivery_period_start,"
        "delivery_period_duration,"
        "delivery_area_code,"
        "quantity_mw,"
        "side,"
        "price_amount,"
        "price_currency,"
        "type,"
        "execution_option"
    )
    print(header)

frequenz.client.electricity_trading.cli.etrading.print_public_trade

print_public_trade(trade: PublicTrade) -> None

Print trade details to stdout in CSV format.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_public_trade(trade: PublicTrade) -> None:
    """Print trade details to stdout in CSV format."""
    values = (
        trade.public_trade_id,
        trade.execution_time.isoformat(),
        trade.delivery_period.start.isoformat(),
        trade.delivery_period.duration,
        trade.buy_delivery_area.code,
        trade.sell_delivery_area.code,
        trade.buy_delivery_area.code_type,
        trade.sell_delivery_area.code_type,
        trade.quantity.mw,
        trade.price.currency,
        trade.price.amount,
        trade.state,
    )
    print(",".join(v.name if isinstance(v, Enum) else str(v) for v in values))

frequenz.client.electricity_trading.cli.etrading.print_public_trade_header

print_public_trade_header() -> None

Print trade header in CSV format.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_public_trade_header() -> None:
    """Print trade header in CSV format."""
    header = (
        "public_trade_id,"
        "execution_time,"
        "delivery_period_start,"
        "delivery_period_duration,"
        "buy_delivery_area_code,"
        "sell_delivery_area_code,"
        "buy_delivery_area_code_type,"
        "sell_delivery_area_code_type,"
        "quantity_mw,"
        "currency,"
        "price,"
        "state "
    )
    print(header)

frequenz.client.electricity_trading.cli.etrading.print_trade

print_trade(trade: Trade, gid: int) -> None

Print trade details to stdout in CSV format.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_trade(trade: Trade, gid: int) -> None:
    """Print trade details to stdout in CSV format."""
    values = (
        trade.id,
        trade.order_id,
        trade.execution_time.isoformat(),
        trade.delivery_period.start.isoformat(),
        trade.delivery_period.duration,
        trade.delivery_area.code,
        trade.delivery_area.code_type,
        trade.side,
        trade.quantity.mw,
        trade.price.currency,
        trade.price.amount,
        trade.state,
        gid,
    )
    print(",".join(v.name if isinstance(v, Enum) else str(v) for v in values))

frequenz.client.electricity_trading.cli.etrading.print_trade_header

print_trade_header() -> None

Print trade header in CSV format.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

def print_trade_header() -> None:
    """Print trade header in CSV format."""
    header = (
        "trade_id,"
        "order_id,"
        "execution_time,"
        "delivery_period_start,"
        "delivery_period_duration,"
        "delivery_area_code,"
        "delivery_area_code_type,"
        "side,"
        "quantity_mw,"
        "currency,"
        "price,"
        "state,"
        "gridpool_id"
    )
    print(header)

frequenz.client.electricity_trading.cli.etrading.receive_public_orders async

receive_public_orders(
    url: str,
    auth_key: str,
    *,
    delivery_start: datetime | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    sign_secret: str | None = None
) -> None

List trades and stream new public trades.

PARAMETER	DESCRIPTION
url	URL of the trading API. TYPE: str
auth_key	API key. TYPE: str
delivery_start	Start of the delivery period or None. TYPE: datetime | None DEFAULT: None
start	First execution time to list trades from. TYPE: datetime | None DEFAULT: None
end	Last execution time to list trades until. TYPE: datetime | None DEFAULT: None
sign_secret	The cryptographic secret to use for HMAC generation. TYPE: str | None DEFAULT: None

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

async def receive_public_orders(  # pylint: disable=too-many-arguments
    url: str,
    auth_key: str,
    *,
    delivery_start: datetime | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    sign_secret: str | None = None,
) -> None:
    """List trades and stream new public trades.

    Args:
        url: URL of the trading API.
        auth_key: API key.
        delivery_start: Start of the delivery period or None.
        start: First execution time to list trades from.
        end: Last execution time to list trades until.
        sign_secret: The cryptographic secret to use for HMAC generation.
    """
    client = Client(server_url=url, auth_key=auth_key, sign_secret=sign_secret)

    print_public_orders_header()

    delivery_period = None
    # If delivery period is selected, list historical trades also
    if delivery_start is not None:
        check_delivery_start(delivery_start)
        delivery_period = DeliveryPeriod(
            start=delivery_start,
            duration=DeliveryDuration.MINUTES_15,
        )
    stream = client.receive_public_order_book(
        delivery_period=delivery_period,
        start_time=start,
        end_time=end,
    )
    async for orders in stream.new_receiver():
        for order in orders:
            print_public_order(order)

frequenz.client.electricity_trading.cli.etrading.receive_public_trades async

receive_public_trades(
    url: str,
    auth_key: str,
    *,
    delivery_start: datetime | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    sign_secret: str | None = None
) -> None

List trades and stream new public trades.

PARAMETER	DESCRIPTION
url	URL of the trading API. TYPE: str
auth_key	API key. TYPE: str
delivery_start	Start of the delivery period or None. TYPE: datetime | None DEFAULT: None
start	First execution time to list trades from. TYPE: datetime | None DEFAULT: None
end	Last execution time to list trades until. TYPE: datetime | None DEFAULT: None
sign_secret	The cryptographic secret to use for HMAC generation. TYPE: str | None DEFAULT: None

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

async def receive_public_trades(  # pylint: disable=too-many-arguments
    url: str,
    auth_key: str,
    *,
    delivery_start: datetime | None = None,
    start: datetime | None = None,
    end: datetime | None = None,
    sign_secret: str | None = None,
) -> None:
    """List trades and stream new public trades.

    Args:
        url: URL of the trading API.
        auth_key: API key.
        delivery_start: Start of the delivery period or None.
        start: First execution time to list trades from.
        end: Last execution time to list trades until.
        sign_secret: The cryptographic secret to use for HMAC generation.
    """
    client = Client(server_url=url, auth_key=auth_key, sign_secret=sign_secret)

    print_public_trade_header()

    delivery_period = None
    # If delivery period is selected, list historical trades also
    if delivery_start is not None:
        check_delivery_start(delivery_start)
        delivery_period = DeliveryPeriod(
            start=delivery_start,
            duration=DeliveryDuration.MINUTES_15,
        )
    stream = client.receive_public_trades(
        delivery_period=delivery_period,
        start_time=start,
        end_time=end,
    )
    async for trade in stream.new_receiver():
        print_public_trade(trade)

frequenz.client.electricity_trading.cli.etrading.reverse_iterator async

reverse_iterator(
    iterator: AsyncIterator[OrderDetail],
    chunk_size: int = 100000,
) -> AsyncIterator[OrderDetail]

Reverse an async iterator in chunks to avoid loading all elements into memory.

PARAMETER	DESCRIPTION
iterator	Async iterator to reverse. TYPE: AsyncIterator[OrderDetail]
chunk_size	Size of the buffer to store elements. TYPE: int DEFAULT: 100000

YIELDS	DESCRIPTION
AsyncIterator[OrderDetail]	Elements of the iterator in reverse order.

Source code in src/frequenz/client/electricity_trading/cli/etrading.py

async def reverse_iterator(
    iterator: AsyncIterator[OrderDetail], chunk_size: int = 100_000
) -> AsyncIterator[OrderDetail]:
    """Reverse an async iterator in chunks to avoid loading all elements into memory.

    Args:
        iterator: Async iterator to reverse.
        chunk_size: Size of the buffer to store elements.

    Yields:
        Elements of the iterator in reverse order.
    """
    buffer: deque[OrderDetail] = deque(maxlen=chunk_size)
    async for item in iterator:
        buffer.append(item)
        if len(buffer) == chunk_size:
            for item in reversed(buffer):
                yield item
            buffer.clear()
    if buffer:
        for item in reversed(buffer):
            yield item