Event-stream (SSE)

[Intershade]

Any plans to implement Event-stream (SSE)?

[Qluxzz]

When you say implement SSE, do you mean that Avanza today uses SSE and thus this library should add support for reading those events?

This project only wraps Avanza's existing functionality and exposes it through a Python API.

If you need real time data there is support for that by using the websocket connection, there's an example in the README for it.

[svennystrom]

The browser platform now uses EventStreams for the streaming data - in fact I think the order and trade flows are not working in the current socket implementation (but they do if you implement EventStreams, whereas the price and orderdepth flows work in the socket case but probably require additional steps in the stream case)

[swechencheng]

Now it is becoming an issue since Avanza seems to have removed the WS.
See #141

[Intershade]

Here is a really simple example of receiving data for orders. This also works for "https://www.avanza.se/_push/trading/deals".
The data streams for quotes etc does however not work with this example "https://www.avanza.se/_push/order-depth-web-push/{orderbook_id}". It will receive "429 Too Many Requests".

import httpx
import asyncio

cookies = {
    "AZACSRF": "...",
    "csid": "...",
    "cstoken": "...",
}

SSE_URL = "https://www.avanza.se/_push/trading/orders/"


async def run_sse():
    while True:
        try:
            async with httpx.AsyncClient(timeout=None) as client:
                async with client.stream("GET", SSE_URL, cookies=cookies) as response:
                    async for line in response.aiter_lines():
                        print(line)
        except httpx.RequestError as e:
            print("Connection error, retrying in 5s:", e)
            await asyncio.sleep(5)
        except Exception as e:
            print("Unexpected error, retrying in 5s:", e)
            await asyncio.sleep(5)


asyncio.run(run_sse())

[dichai1983]

Thanks for your post. I'm new to Python. Is it possible that it's because you indeed try too many times to connect to the SSE? Maybe it only needs to subscribe the SSE once, then it handles reconnection automatically.

import json
from sseclient import SSEClient

url = "https://www.avanza.se/_push/quote-web-push"

for event in SSEClient(url):
    print(f"Event type: {event.event}")
    print(f"Data: {event.data}")
    
    if event.data:
        try:
            data = json.loads(event.data)
            print(f"Parsed JSON: {data}")
        except json.JSONDecodeError:
            print("Non-JSON data received")
    
    if event.data == 'END':
        break

For more controls:

import requests
from sseclient import SSEClient

url = "https://www.avanza.se/_push/quote-web-push/"
payload = '{"T":"/_push/quote-web-push", "orderbookIds":["1234"], "forceWebsocket":false}'
headers = {
    "Content-Type": "application/json"
}

response = requests.post(url, data=payload, headers=headers, stream=True)
client = SSEClient(response)

for event in client.events():
    print(event.event, event.data)

[Intershade]

No, the problem is that quote-web-push and other event-streams are handled differently by their servers. I have never worked with this kind of problem so I am really fumbling in the dark here...

[Intershade]

I have boiled it down to a TLS/JA3/HTTP2 fingerprints issue. Using curl_cffi we can impersonate a browser with correct fingerprints. The code below works fine if you enter your csid, cstoken and AZACSRF.

import asyncio
from curl_cffi.requests import AsyncSession

async def main():
    async with AsyncSession(impersonate="chrome136") as session:
        async with session.stream(
            "GET",
            "https://www.avanza.se/_push/quote-web-push/1300498",
            cookies={
                "AZACSRF": "...",
                "csid": "...",
                "cstoken": "...",
            },
        ) as resp:
            async for line in resp.aiter_lines():
                print(line)


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

b'event:QUOTE'
b'data:{"orderbookId":"1300498","buyPrice":2.91,"sellPrice":2.91,"closingPrice":1.87,"highestPrice":4.17,"lowestPrice":2.92,"lastPrice":2.92,"totalValueTraded":114077798.0515,"totalVolumeTraded":100797072,"change":1.05,"changePercent":0.5615,"spreadPercent":0.0000,"volumeWeightedAveragePrice":3.99,"updated":"2025-10-09T19:17:26.499Z","lastPriceUpdated":"2025-10-09T19:17:26.499Z"}'

[dichai1983]

Great job!

So we need to login in the browser first and copy the csid, cstoken and AZACSRF from the browser cookie? Thanks.

[dichai1983]

Or from avanza._session.cookies.get_dict()? I'll try. Thanks.

[Intershade]

Yes, you will find it in the session cookies

[dichai1983]

@Intershade it works, thanks

[dichai1983]

And then parse the binary line:

async for line in resp.aiter_lines():    
    # 1. Decode from bytes → string
    text = line.decode("utf-8")
    
    # 2. Check if it starts with 'data:{'
    if text.startswith("data:{"):
        json_part = text[
            len("data:") :
        ].strip()  # remove 'data:' prefix
        data = json.loads(json_part)
        print(data)

[dichai1983]

And if you want to run multiple sessions at the same time:

async def gather():
        # Run both concurrently
        await asyncio.gather(
            load_quotes(orderbook_id_1),
            load_quotes(orderbook_id_2),
        )

asyncio.run(gather())

[Intershade]

Yes and it is also possible to replace:

import requests
self._session = requests.Session() 

with:

from curl_cffi.requests import Session
self._session = Session(impersonate="firefox")

Then use that session to stream data, in a separate thread. In that case all post/get requests using the session will also be correct with fingerprints.

with self._session.stream("GET", url, headers=headers) as resp:
    for raw_line in resp.iter_lines():
                line: str = raw_line.decode()

[dichai1983]

Sometimes this happens:

curl_cffi.requests.exceptions.RequestException: Failed to perform, curl: (28) Connection timed out after 30000 milliseconds.

Let me test set timeout = 0, basically turn off the timeout.