Skip to content
Home
finance-nepal
nepse

How I Built a Resilient Real‑Time NEPSE Ticker with `websockets` and `asyncio`

Mausam

When I tried to stream NEPSE price updates with the simple websocket-client library, a single network hiccup dropped the connection and my data pipeline stalled for minutes. You, a data engineer who already knows the basics of WebSocket ingestion, need a client that stays up, validates every JSON blob, and never loses a tick—even when the upstream server misbehaves. Below I walk through the exact architecture I settled on: an async consumer, a tenacious reconnection loop, schema‑driven validation, and a producer‑consumer queue that decouples ingestion from processing.

Key Takeaways

  • Async `websockets` + `asyncio` provide non‑blocking I/O essential for high‑frequency ticker streams.
  • Exponential backoff with jitter (via tenacity) prevents thundering‑herd reconnections and keeps latency predictable.
  • Pydantic models catch malformed payloads before they reach downstream storage, protecting the whole pipeline.
  • An asyncio.Queue cleanly separates the network consumer from CPU‑heavy processing or persistence logic.
  • Running a lightweight mock server locally lets you develop and test resilience patterns without a paid NEPSE feed.

The Problem

NEPSE’s public WebSocket endpoint streams price, volume, and change information for dozens of symbols. In production I needed:

  • Zero‑downtime reconnection after network blips.
  • Guarantees that a malformed message never crashes the consumer.
  • A way to buffer incoming ticks while a separate worker writes them to disk.

Standard synchronous clients force you to juggle threads or risk blocking the event loop. The solution is to redesign the client around asyncio and add disciplined retry‑validation‑queue layers.

Data and Sources

Real‑time NEPSE data is proprietary, so the demo uses a local mock server that mimics the official payload shape. For symbol metadata I pull a tiny list of Nepali company names from the public Random User API (just to illustrate an external HTTP call).

Data accessed on 2026-10-14.

Loading the Data

The mock server generates ticker JSON every 1–2 seconds. Before we start the client we fetch a short list of symbols from the Random User API to demonstrate a real HTTP request.

import requests, itertools

def fetch_symbols(limit=5):
    try:
        resp = requests.get("https://randomuser.me/api/?results={}".format(limit), timeout=5)
        resp.raise_for_status()
        users = resp.json()["results"]
        # Use the first name as a fake stock symbol (e.g., "Veer")
        return [u["name"]["first"].upper() for u in users]
    except (requests.RequestException, KeyError) as e:
        print(f"Failed to fetch symbols: {e}")
        # Fallback to a hard‑coded list
        return ["NABIL", "NTC", "KBL", "CIM", "NEPSE"]

The Core Logic

Three building blocks make the client resilient:

1. Async connection with tenacious retries

from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
import websockets, json, asyncio

@retry(
    wait=wait_exponential(multiplier=1, min=4, max=60, jitter=0.1),
    stop=stop_after_attempt(10),
    retry=retry_if_exception_type((websockets.exceptions.ConnectionClosedError,
                                   websockets.exceptions.InvalidURI,
                                   OSError)),
    reraise=True
)
async def connect(uri):
    return await websockets.connect(uri)

The decorator automatically backs off when the connection drops, adding a small random jitter to avoid synchronized retries across many clients.

2. Pydantic schema for each tick

from pydantic import BaseModel, ValidationError
from datetime import datetime

class NepseTick(BaseModel):
    symbol: str
    price: float
    volume: int
    timestamp: datetime
    change: float
    percent_change: float

If the upstream sends a missing field or a string where a number is expected, ValidationError is raised and we simply log the offending payload.

3. Producer‑consumer pipeline

async def producer(ws, queue):
    async for raw_msg in ws:
        try:
            data = json.loads(raw_msg)
            tick = NepseTick(**data)          # validation step
            await queue.put(tick)             # non‑blocking hand‑off
        except (json.JSONDecodeError, ValidationError) as e:
            print(f"Bad message ignored: {e}")

async def consumer(queue):
    history = {}
    while True:
        tick = await queue.get()
        # Simple aggregation: keep last price per symbol
        history[tick.symbol] = tick.price
        print(f"[{tick.timestamp.isoformat()}] {tick.symbol}: {tick.price:.2f} ({tick.percent_change:+.2f}%)")
        queue.task_done()

The consumer can be swapped out for a DB writer, a Kafka producer, or a Matplotlib plotter without touching the network code.

Putting It Together

We spin up the mock server, fetch symbols, then launch the producer and consumer tasks inside a single event loop. Two explicit try/except blocks guard the connection handshake and the JSON parsing stage.

async def run_client():
    symbols = fetch_symbols()
    uri = "ws://localhost:8765"
    queue = asyncio.Queue(maxsize=100)

    async with await connect(uri) as ws:
        producer_task = asyncio.create_task(producer(ws, queue))
        consumer_task = asyncio.create_task(consumer(queue))
        await asyncio.gather(producer_task, consumer_task)

if __name__ == "__main__":
    try:
        asyncio.run(run_client())
    except (websockets.exceptions.InvalidURI, OSError) as e:
        print(f"Unable to start client: {e}")

Complete Script

The full, runnable script combines the mock server, the resilient client, and a tiny CSV logger. Save it as nepse_ticker.py and run python nepse_ticker.py.

#!/usr/bin/env python3
"""
Resilient NEPSE ticker client demo.
Runs a local mock WebSocket server and connects with async retries,
Pydantic validation, and a producer‑consumer queue.
"""

import asyncio
import json
import random
import datetime
import csv
import pathlib
import requests

import websockets
from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
from pydantic import BaseModel, ValidationError

# ----------------------------------------------------------------------
# Mock NEPSE server (runs in the same process for demo purposes)
# ----------------------------------------------------------------------
MOCK_SYMBOLS = ["NABIL", "NTC", "KBL", "CIM", "NEPSE"]


async def mock_nepse_server(websocket, path):
    """Emit a realistic ticker JSON every 1‑2 seconds, occasionally corrupt."""
    while True:
        # Randomly decide to send a malformed payload
        if random.random() < 0.1:
            await websocket.send("MALFORMED_MESSAGE")
        else:
            symbol = random.choice(MOCK_SYMBOLS)
            price = round(random.uniform(10, 150), 2)
            volume = random.randint(100, 5000)
            change = round(random.uniform(-5, 5), 2)
            percent = round(change / price * 100, 2)
            payload = {
                "symbol": symbol,
                "price": price,
                "volume": volume,
                "timestamp": datetime.datetime.utcnow().isoformat(),
                "change": change,
                "percent_change": percent,
            }
            await websocket.send(json.dumps(payload))
        await asyncio.sleep(random.uniform(1, 2))


# ----------------------------------------------------------------------
# Data model
# ----------------------------------------------------------------------
class NepseTick(BaseModel):
    symbol: str
    price: float
    volume: int
    timestamp: datetime.datetime
    change: float
    percent_change: float


# ----------------------------------------------------------------------
# Helper: fetch a fake symbol list from Random User API
# ----------------------------------------------------------------------
def fetch_symbols(limit: int = 5):
    try:
        resp = requests.get(f"https://randomuser.me/api/?results={limit}", timeout=5)
        resp.raise_for_status()
        users = resp.json()["results"]
        return [u["name"]["first"].upper() for u in users]
    except Exception as e:
        print(f"[symbol fetch] fallback due to: {e}")
        return MOCK_SYMBOLS


# ----------------------------------------------------------------------
# Retry‑wrapped connection
# ----------------------------------------------------------------------
@retry(
    wait=wait_exponential(multiplier=1, min=4, max=60, jitter=0.1),
    stop=stop_after_attempt(10),
    retry=retry_if_exception_type((ConnectionClosedError, OSError)),
    reraise=True,
)
async def connect(uri: str):
    return await websockets.connect(uri)


# ----------------------------------------------------------------------
# Producer: receive raw messages, validate, push to queue
# ----------------------------------------------------------------------
async def producer(ws, queue: asyncio.Queue):
    async for raw_msg in ws:
        try:
            data = json.loads(raw_msg)
            tick = NepseTick(**data)
            await queue.put(tick)
        except (json.JSONDecodeError, ValidationError) as e:
            print(f"[producer] ignored bad payload: {e}")


# ----------------------------------------------------------------------
# Consumer: simple CSV logger + console output
# ----------------------------------------------------------------------
async def consumer(queue: asyncio.Queue, csv_path: pathlib.Path):
    csv_path.parent.mkdir(parents=True, exist_ok=True)
    with csv_path.open("a", newline="") as f:
        writer = csv.writer(f)
        # Write header if file is empty
        if f.tell() == 0:
            writer.writerow(["timestamp", "symbol", "price", "volume", "change", "percent_change"])
        while True:
            tick = await queue.get()
            writer.writerow(
                [
                    tick.timestamp.isoformat(),
                    tick.symbol,
                    tick.price,
                    tick.volume,
                    tick.change,
                    tick.percent_change,
                ]
            )
            print(
                f"[{tick.timestamp.time()}] {tick.symbol:<5} {tick.price:>7.2f} "
                f"({tick.percent_change:+5.2f}%) vol:{tick.volume}"
            )
            queue.task_done()


# ----------------------------------------------------------------------
# Main orchestration
# ----------------------------------------------------------------------
async def main():
    # Start mock server in background
    server = await websockets.serve(mock_nepse_server, "localhost", 8765)
    print("[server] mock NEPSE server listening on ws://localhost:8765")

    symbols = fetch_symbols()
    print(f"[client] using symbols: {', '.join(symbols)}")

    queue = asyncio.Queue(maxsize=200)
    csv_file = pathlib.Path("nepse_ticks.csv")

    try:
        async with await connect("ws://localhost:8765") as ws:
            prod = asyncio.create_task(producer(ws, queue))
            cons = asyncio.create_task(consumer(queue, csv_file))
            await asyncio.gather(prod, cons)
    except (ConnectionClosedOK, ConnectionClosedError) as e:
        print(f"[client] connection closed gracefully: {e}")
    finally:
        server.close()
        await server.wait_closed()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"[fatal] {e}")

Expected Output

When you run the script you should see something like:

[server] mock NEPSE server listening on ws://localhost:8765
[client] using symbols: VEER, SANDHA, KUMAR, RAI, SHRESTHA
[12:00:01.123456] NABIL  112.45 (+0.87%) vol:3421
[12:00:02.789012] NTC     45.78 (-1.23%) vol:1289
[producer] ignored bad payload: Expecting value: line 1 column 1 (char 0)
[12:00:04.345678] KBL     78.90 (+0.15%) vol:2100
...

The CSV file nepse_ticks.csv will contain the same rows for later analysis.

Limitations and Tradeoffs

This demo focuses on connection resilience and data validation. It does NOT address:

  • Historical catch‑up: If the client disconnects for longer than a few seconds, missed ticks are lost. Production systems usually supplement the stream with a REST snapshot or a Kafka replay.
  • Scalability: A single async process can comfortably handle a few hundred messages per second. For thousands of symbols you’d shard across multiple processes or move deserialization to a compiled library.
  • Security: The mock server runs without TLS or authentication. Real NEPSE APIs require API keys or OAuth, which adds token refresh logic.
  • Back‑pressure handling: The queue size is fixed; if the consumer falls behind, the producer will pause when queue.put blocks, which is acceptable for most pipelines but may need fine‑tuning.

Frequently Asked Questions

How does websockets differ from websocket-client for production use?

websockets is built on asyncio, giving you true non‑blocking I/O, native cancellation, and back‑pressure via queues. websocket-client is synchronous, forcing you to manage threads or processes manually, which complicates graceful reconnection and scaling.

Can I handle multiple message types (price updates vs. order‑book snapshots) on the same stream?

Yes. Add a type discriminator field to the JSON, create a base Pydantic model, and subclass it for each message type. In the producer you can route validated objects to different consumer queues based on msg.type.

What storage solution works best for high‑frequency ticker data?

For analytics, a time‑series database such as InfluxDB or TimescaleDB offers efficient range queries. For downstream ML pipelines, ingest into Kafka (Avro schema) and use a consumer that writes Parquet files to S3.

How do I avoid a “thundering herd” when many instances reconnect simultaneously?

Tenacity’s exponential backoff with jitter (the jitter=0.1 argument) randomizes each client’s retry interval, spreading reconnection attempts over time.

Is it safe to run multiple WebSocket clients in the same event loop?

Absolutely. Each client should own its own asyncio.Queue and be launched as a separate task. Just be mindful of total socket descriptors and CPU usage.

What I'd Change

If I were to ship this into production tomorrow, I would replace the in‑process mock server with the real NEPSE endpoint, add a TLS‑enabled connection, and introduce a small state‑reconciliation service that pulls a snapshot via REST after every reconnect. That way the system guarantees no data gaps, while still enjoying the lightweight async architecture that kept my dev loop fast.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.