Live order stream misses orders

[martinschlueter91]

What happened?

When comparing the live order stream with the historic order stream, the live order stream is missing some orders. Below is a script that first starts a live stream for 5 minutes and then requests the same time frame again as historic stream. Both should give identical results, but there are always more orders in the historic stream than in the live stream. I.e., the live stream is missing some orders.

import asyncio
import os
import pathlib

import frequenz.client.electricity_trading as fcet
import pandas as pd
from dotenv import load_dotenv


async def stream_public_order_book() -> None:
    load_dotenv()
    api_key = os.getenv("FREQUENZ_API_KEY")
    if api_key is None:
        raise ValueError("FREQUENZ_API_KEY environment variable is not set")

    trading_test_url = os.getenv("FREQUENZ_TRADING_TEST_URL")
    if trading_test_url is None:
        raise ValueError("FREQUENZ_TRADING_TEST_URL environment variable is not set")

    client = fcet.Client(server_url=trading_test_url, auth_key=api_key)

    delivery_area = fcet.DeliveryArea(code="10YDE-RWENET---I", code_type=fcet.EnergyMarketCodeType.EUROPE_EIC)

    live_public_orders = client.receive_public_order_book(delivery_area=delivery_area)
    live_stream_start = pd.Timestamp.now(tz="UTC")
    live_stream_end = live_stream_start + pd.Timedelta(minutes=5)

    # Live order stream
    live_file_path = pathlib.Path(__file__).with_name("live_order_stream.txt")
    live_orders = []
    with live_file_path.open("w", encoding="utf-8") as output_file:
        start_msg = (
            "Live stream of public order book with start time: "
            + str(live_stream_start)
            + " and end time: "
            + str(live_stream_end)
        )
        print(start_msg, file=output_file)

        async for batch in live_public_orders.new_receiver():
            for order in batch:
                if order.delivery_area == delivery_area:
                    live_orders.append(order)
                    print(f"{order.update_time} - {order.public_order_id}", file=output_file)

            if pd.Timestamp.now(tz="UTC") > live_stream_end:
                break

    # Historic order stream
    hist_start_time = live_orders[0].update_time
    hist_end_time = live_orders[-1].update_time
    historic_public_orders = client.receive_public_order_book(
        delivery_area=delivery_area, start_time=hist_start_time, end_time=hist_end_time
    )

    historic_file_path = pathlib.Path(__file__).with_name("historic_order_stream.txt")
    historic_orders = []
    async for batch in historic_public_orders.new_receiver():
        for order in batch:
            if order.delivery_area == delivery_area:
                historic_orders.append(order)

    historic_orders_sorted = sorted(historic_orders, key=lambda x: (x.update_time, x.public_order_id))

    with historic_file_path.open("w", encoding="utf-8") as output_file:
        start_msg = (
            "Historic stream of public order book with start time: "
            + str(hist_start_time)
            + " and end time: "
            + str(hist_end_time)
        )
        print(start_msg, file=output_file)
        for order in historic_orders_sorted:
            print(f"{order.update_time} - {order.public_order_id}", file=output_file)

    print(f"Live stream: {len(live_orders)} orders")
    print(f"Historic stream: {len(historic_orders_sorted)} orders")


if __name__ == "__main__":
    asyncio.run(stream_public_order_book())

What did you expect instead?

That both streams give the same orders.

Affected version(s)

0.12.1

Affected part(s)

I don't know (part:❓)

Extra information

No response

[matthias-wende-frequenz]

That's a know internal bug. We're heavily working on updating the internal stream architecture to circumvent this problem.