Skip to content
Home
finance-nepal
nepse

From Zero to Live: Building a Production‑Ready NEPSE Ticker with WebSockets

Mausam

When I first tried to feed live NEPSE quotes into my back‑testing framework, the only option was a noisy REST poll that missed sub‑second price moves and ate my rate‑limit quota. After a few missed arbitrage windows I decided to stitch together the official NEPSE WebSocket, a Redis cache, and a tiny FastAPI server that streams Server‑Sent Events (SSE) to browsers. If you are a quant or data engineer in Nepal who needs millisecond‑level market data without paying a third‑party, this walk‑through shows exactly how I built a production‑ready ticker from zero to live.

Key Takeaways

  • Bootstrapping state from the REST snapshot guarantees a complete view even if the WebSocket drops.
  • Redis hashes give O(1) reads/writes per symbol and survive process restarts.
  • SSE is a browser‑friendly, one‑way push protocol that lets you avoid the complexity of a full WebSocket client stack.
  • Explicit error handling (reconnect loops, JSON schema validation) turns a flaky feed into a resilient service.
  • Minimal monitoring (heartbeat logs and Prometheus metrics) surfaces latency spikes before they break your trading bot.

The Problem

NEPSE publishes a public WebSocket at wss://api.nepse.com.np/stream/tickers that pushes JSON updates for every listed security. The feed is fast, but it’s also stateful: when you connect you only receive deltas, not the full order book. A naïve consumer that starts mid‑stream will see an incomplete picture, and any network hiccup wipes out all in‑flight updates. I needed a service that (1) starts with a full snapshot, (2) keeps the latest quote for every symbol in durable memory, and (3) offers a simple HTTP endpoint that any front‑end can consume without dealing with WebSocket quirks.

Data and Sources

Data accessed on 2026-10-21.

Loading the Data

The first step is to pull the full market snapshot so we can seed Redis with a known good state. This runs once at startup and also on a daily schedule to recover from any drift.

import requests, json, redis, logging

def bootstrap_snapshot(r: redis.Redis):
    try:
        resp = requests.get("https://api.nepse.com.np/market/quotes", timeout=5)
        resp.raise_for_status()
        snapshot = resp.json()  # list of {"symbol": "...", "price": ..., "volume": ...}
        pipe = r.pipeline()
        for quote in snapshot:
            key = f"ticker:{quote['symbol']}"
            pipe.hset(key, mapping=quote)
        pipe.execute()
        logging.info("Bootstrap completed: %d symbols loaded", len(snapshot))
    except requests.RequestException as e:
        logging.error("Failed to fetch snapshot: %s", e)
        raise

The Core Logic

From here we spin up three async tasks: (1) maintain the WebSocket connection, (2) validate incoming messages, and (3) push updates into Redis. I use pydantic for schema validation because it catches malformed payloads early, but I keep it optional for ultra‑low‑latency paths.

import asyncio, json, websockets
from pydantic import BaseModel, ValidationError

class TickerUpdate(BaseModel):
    symbol: str
    price: float
    volume: int
    timestamp: int

async def ws_consumer(r: redis.Redis, url: str):
    while True:  # reconnect loop
        try:
            async with websockets.connect(url, ping_interval=20, ping_timeout=10) as ws:
                logging.info("WebSocket connected")
                async for msg in ws:
                    try:
                        data = json.loads(msg)
                        upd = TickerUpdate(**data)  # validation
                        key = f"ticker:{upd.symbol}"
                        r.hset(key, mapping=upd.dict())
                        r.lpush("updates", msg)  # raw log for audit
                    except (json.JSONDecodeError, ValidationError) as e:
                        logging.warning("Bad message skipped: %s", e)
        except (websockets.exceptions.ConnectionClosedError,
                websockets.exceptions.InvalidStatusCode) as e:
            logging.error("WebSocket error (%s), retrying in 5s", e)
            await asyncio.sleep(5)

Putting It Together

Now we wire the pieces: start Redis, bootstrap, fire the consumer, and launch a FastAPI app that streams SSE to any client. The SSE endpoint reads the latest hash for each requested symbol and yields a newline‑delimited JSON payload whenever a new update arrives.

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

def sse_generator(r: redis.Redis, symbols: list):
    last_ts = {sym: 0 for sym in symbols}
    while True:
        for sym in symbols:
            key = f"ticker:{sym}"
            data = r.hgetall(key)
            if not data:
                continue
            ts = int(data.get(b'timestamp', 0))
            if ts > last_ts[sym]:
                last_ts[sym] = ts
                payload = json.dumps({k.decode(): v.decode() for k, v in data.items()})
                yield f"data: {payload}\n\n"
        time.sleep(0.1)  # throttle a bit

@app.get("/sse/{symbol}")
async def stream_symbol(symbol: str, request: Request):
    if not redis_client.exists(f"ticker:{symbol}"):
        return {"error": "unknown symbol"}
    generator = sse_generator(redis_client, [symbol])
    return StreamingResponse(generator, media_type="text/event-stream")

Finally, the orchestrator runs everything inside an asyncio event loop. We also add a simple Prometheus gauge to expose the current number of connected WebSocket clients.

from prometheus_client import start_http_server, Gauge

ws_clients = Gauge("nepse_ws_clients", "Number of active WebSocket connections")

async def monitor():
    while True:
        ws_clients.set(len(asyncio.all_tasks()))  # crude but illustrative
        await asyncio.sleep(10)

if __name__ == "__main__":
    import uvicorn
    redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

    # Step 1: Bootstrap
    bootstrap_snapshot(redis_client)

    # Step 2‑4: Run async workers
    loop = asyncio.get_event_loop()
    loop.create_task(ws_consumer(redis_client, "wss://api.nepse.com.np/stream/tickers"))
    loop.create_task(monitor())
    # Step 5: Start FastAPI server (runs in the same process)
    start_http_server(8001)  # Prometheus metrics
    uvicorn.run(app, host="0.0.0.0", port=8000)

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
Production‑ready NEPSE ticker service:
- Bootstrap from REST snapshot
- Ingest live deltas via WebSocket
- Persist latest quote per symbol in Redis
- Serve SSE endpoint for browsers
- Light monitoring with Prometheus
"""

import asyncio
import json
import logging
import time
from typing import List

import redis
import requests
import websockets
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, ValidationError
from prometheus_client import Gauge, start_http_server
import uvicorn

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
WS_URL = "wss://api.nepse.com.np/stream/tickers"
SNAPSHOT_URL = "https://api.nepse.com.np/market/quotes"
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
PROMETHEUS_PORT = 8001
FASTAPI_HOST = "0.0.0.0"
FASTAPI_PORT = 8000

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

# ----------------------------------------------------------------------
# Redis client (single‑node, decode strings for convenience)
# ----------------------------------------------------------------------
redis_client = redis.Redis(
    host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True
)

# ----------------------------------------------------------------------
# Data models
# ----------------------------------------------------------------------
class TickerUpdate(BaseModel):
    symbol: str
    price: float
    volume: int
    timestamp: int

# ----------------------------------------------------------------------
# Step 1 – Bootstrapping the State
# ----------------------------------------------------------------------
def bootstrap_snapshot(r: redis.Redis) -> None:
    """Fetch full market snapshot and store each symbol as a Redis hash."""
    try:
        resp = requests.get(SNAPSHOT_URL, timeout=5)
        resp.raise_for_status()
        snapshot = resp.json()  # Expected: list of dicts
        pipe = r.pipeline()
        for quote in snapshot:
            key = f"ticker:{quote['symbol']}"
            pipe.hset(key, mapping=quote)
        pipe.execute()
        logging.info("Bootstrap completed with %d symbols", len(snapshot))
    except requests.RequestException as exc:
        logging.error("Snapshot fetch failed: %s", exc)
        raise

# ----------------------------------------------------------------------
# Step 2 – Connecting to the NEPSE WebSocket
# ----------------------------------------------------------------------
async def ws_consumer(r: redis.Redis, url: str) -> None:
    """Maintain a persistent WebSocket connection, reconnect on failure."""
    while True:
        try:
            async with websockets.connect(url, ping_interval=20, ping_timeout=10) as ws:
                logging.info("WebSocket connected")
                async for raw_msg in ws:
                    await handle_message(r, raw_msg)
        except (websockets.exceptions.ConnectionClosedError,
                websockets.exceptions.InvalidStatusCode,
                OSError) as exc:
            logging.error("WebSocket error (%s); retrying in 5s", exc)
            await asyncio.sleep(5)

# ----------------------------------------------------------------------
# Step 3 – Parsing and Validating Updates
# ----------------------------------------------------------------------
async def handle_message(r: redis.Redis, raw_msg: str) -> None:
    """Decode JSON, validate schema, and persist to Redis."""
    try:
        payload = json.loads(raw_msg)
        upd = TickerUpdate(**payload)  # validation
        key = f"ticker:{upd.symbol}"
        # Store the latest quote as a hash
        r.hset(key, mapping=upd.dict())
        # Keep a raw audit trail (trimmed later if needed)
        r.lpush("updates", raw_msg)
        # Optional: trim list to last 10k entries
        r.ltrim("updates", 0, 9999)
    except (json.JSONDecodeError, ValidationError) as exc:
        logging.warning("Skipped malformed message: %s", exc)

# ----------------------------------------------------------------------
# Step 4 – Serving Subscribers via SSE
# ----------------------------------------------------------------------
app = FastAPI()

def sse_generator(r: redis.Redis, symbols: List[str]):
    """Yield Server‑Sent Events whenever a newer timestamp appears."""
    last_ts = {sym: 0 for sym in symbols}
    while True:
        for sym in symbols:
            key = f"ticker:{sym}"
            data = r.hgetall(key)
            if not data:
                continue
            ts = int(data.get("timestamp", 0))
            if ts > last_ts[sym]:
                last_ts[sym] = ts
                payload = json.dumps(data)
                yield f"data: {payload}\n\n"
        time.sleep(0.1)  # small back‑off to avoid busy loop

@app.get("/sse/{symbol}")
async def stream_symbol(symbol: str, request: Request):
    """SSE endpoint for a single symbol."""
    if not redis_client.exists(f"ticker:{symbol}"):
        return {"error": f"symbol {symbol} not found"}
    generator = sse_generator(redis_client, [symbol])
    return StreamingResponse(generator, media_type="text/event-stream")

# ----------------------------------------------------------------------
# Step 5 – Monitoring and Alerting
# ----------------------------------------------------------------------
ws_clients = Gauge("nepse_ws_clients", "Number of active asyncio tasks handling WS")

async def monitor():
    """Publish simple metrics every 10 seconds."""
    while True:
        # Rough estimate: count of tasks that are not the main loop
        ws_clients.set(len([t for t in asyncio.all_tasks() if not t.done()]))
        await asyncio.sleep(10)

# ----------------------------------------------------------------------
# Orchestration
# ----------------------------------------------------------------------
def main():
    # 1️⃣ Bootstrap state
    bootstrap_snapshot(redis_client)

    # 2️⃣ Launch async workers
    loop = asyncio.get_event_loop()
    loop.create_task(ws_consumer(redis_client, WS_URL))
    loop.create_task(monitor())

    # 3️⃣ Start Prometheus exporter
    start_http_server(PROMETHEUS_PORT)
    logging.info("Prometheus metrics available on :%d/metrics", PROMETHEUS_PORT)

    # 4️⃣ Run FastAPI (uvicorn) – runs in the same process
    uvicorn.run(app, host=FASTAPI_HOST, port=FASTAPI_PORT, log_level="info")

if __name__ == "__main__":
    main()

Expected Output

When you start the script you should see log lines similar to:

2026-10-21 12:00:01,234 INFO Bootstrap completed with 245 symbols
2026-10-21 12:00:02,001 INFO WebSocket connected
2026-10-21 12:00:02,015 INFO Prometheus metrics available on :8001/metrics

Visiting http://localhost:8000/sse/NABIL in a browser will stream lines like:

data: {"symbol":"NABIL","price":123.45,"volume":5678,"timestamp":1729545601}

Limitations and Tradeoffs

  • WebSocket vs REST polling: WS gives sub‑second latency but requires a stable TCP connection; a network outage forces a reconnection delay. Polling is simpler but can miss rapid price spikes and quickly hits rate limits.
  • Redis persistence: In a single‑node setup the Redis AOF file protects against crashes, but it won’t survive a full data‑center outage. For horizontal scaling you’d need Redis Cluster or a log‑based system like Kafka.
  • Schema validation cost: Pydantic adds ~0.3 ms per message. If you push >10 k updates/sec you may want to drop validation after the bootstrap phase.
  • SSE vs WebSocket for clients: SSE works out‑of‑the‑box in browsers and respects HTTP proxies, but it’s one‑way only. If you ever need to acknowledge receipt or send commands back, switch to a full WebSocket client.

Frequently Asked Questions

Can I run this on a cloud VM without Docker?

Yes. The script only depends on system‑installed Redis and Python 3.10+. A virtualenv or poetry environment is sufficient.

What happens if the WebSocket sends a duplicate update?

The timestamp field is monotonic per symbol; the Redis hash will be overwritten with the same data, which is harmless. If you need deduplication you can store a hash of the payload in a Redis SET.

How do I persist the raw audit trail longer than 10 k entries?

Replace the lpush/ltrim pair with a Redis Stream (XADD) or ship the messages to an external log store (e.g., Loki or S3) via a background worker.

Is the SSE endpoint scalable to thousands of concurrent browsers?

SSE uses a single HTTP connection per client, so you’ll need a reverse proxy (NGINX or Traefik) with keep‑alive tuning. For massive fan‑out, consider a Pub/Sub system (Redis Pub/Sub or NATS) and let the proxy handle connection multiplexing.

Can I add authentication to the API?

FastAPI supports OAuth2, API keys, or JWT out of the box. Wrap the /sse/{symbol} route with a dependency that validates the token before returning the generator.

What I'd Change

If I were to re‑architect this for a multi‑team trading desk, I’d replace the raw Redis hash with a time‑series module (RedisTimeSeries) so that historical candles are persisted automatically. I’d also decouple the WebSocket ingest from the HTTP layer by feeding updates into a Kafka topic; the FastAPI service would then consume from Kafka, giving us true horizontal scalability and replay capability. For now, the single‑process design wins on simplicity and cost, but the next iteration should embrace a streaming backbone.

Post a Comment

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