Tech Tutorial - February 20 2026 173007
Welcome to today’s deep‑dive tutorial! We’ll walk through building a real‑time stock price dashboard that streams live market data using FastAPI, WebSockets, and Plotly. By the end of this guide you’ll have a fully functional web app you can host locally or on the cloud, and you’ll understand the key patterns that make real‑time Python services tick. Grab a cup of coffee, fire up your IDE, and let’s turn raw price ticks into an interactive visual experience.
Why Real‑Time Dashboards Matter
In finance, every millisecond counts—traders need instant feedback to make informed decisions. A real‑time dashboard eliminates the latency of manual refreshes and lets users spot trends as they happen. Beyond finance, the same architecture powers IoT monitoring, live sports scores, and collaborative editing tools. Understanding the building blocks here gives you a reusable template for any streaming use case.
FastAPI shines because it blends high performance with an intuitive, type‑hint‑driven API design. Its native support for asynchronous endpoints makes it a natural fit for WebSocket communication. Pair that with Plotly’s rich, browser‑based visualizations, and you have a stack that’s both developer‑friendly and production‑ready.
Setting Up the Development Environment
First, let’s create an isolated Python environment. Open a terminal and run:
python -m venv venv
source venv/bin/activate # On Windows use `venv\Scripts\activate`
pip install fastapi[all] uvicorn python‑websockets pandas plotly
We’re pulling in FastAPI (with all optional extras), uvicorn as the ASGI server, websockets for low‑level socket handling, pandas for data manipulation, and plotly for the front‑end charts. Keep your requirements.txt tidy for future deployments.
Next, create a project structure that separates concerns:
app/– core FastAPI codefrontend/– static HTML/JS assetsdata/– sample CSV files or mock data generators
This layout mirrors real‑world microservice patterns, making it easier to scale or replace components later.
Crafting the FastAPI Backend
Inside app/main.py we’ll define the API, a mock data generator, and a WebSocket endpoint. The generator simulates a live feed by pulling random rows from a CSV of historical prices.
import asyncio
import random
import pandas as pd
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
# Load historical data once at startup
prices_df = pd.read_csv("data/sp500_sample.csv")
timestamps = prices_df["timestamp"].tolist()
values = prices_df["price"].tolist()
async def price_stream():
"""Yield a new price every 0.5 seconds, mimicking a live feed."""
while True:
idx = random.randint(0, len(timestamps) - 1)
yield {
"timestamp": timestamps[idx],
"price": round(values[idx] + random.uniform(-0.5, 0.5), 2)
}
await asyncio.sleep(0.5)
The above coroutine is deliberately simple; in production you’d replace it with a connection to a market data provider or a Kafka topic. Now let’s expose a health‑check endpoint and the WebSocket route.
@app.get("/health")
async def health_check():
return {"status": "ok"}
@app.websocket("/ws/prices")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
async for price in price_stream():
await websocket.send_json(price)
except WebSocketDisconnect:
print("Client disconnected")
Notice the use of await websocket.send_json—FastAPI automatically serializes the dictionary to JSON, keeping the client side lightweight.
Pro tip: If you anticipate many concurrent connections, consider usinguvicorn[standard]with--workersor deploying behind a reverse proxy like Nginx that supports HTTP/2 and WebSocket upgrades.
Building the Front‑End with Plotly and JavaScript
The front‑end lives in frontend/index.html. We’ll embed a Plotly line chart that updates in real time as new price messages arrive. First, include the Plotly CDN and a minimal CSS reset.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Live S&P 500 Dashboard</title>
<script src="https://cdn.plot.ly/plotly-2.24.1.min.js"></script>
<style>
body {font-family: Arial, sans-serif; margin: 2rem;}
#chart {width: 100%; height: 500px;}
</style>
</head>
<body>
<h1>Live S&P 500 Prices</h1>
<div id="chart"></div>
<script>
const socket = new WebSocket("ws://localhost:8000/ws/prices");
const timestamps = [];
const prices = [];
const layout = {
title: "Real‑Time S&P 500",
xaxis: {title: "Time"},
yaxis: {title: "Price"},
margin: {l:50, r:20, t:50, b:50}
};
Plotly.newPlot("chart", [{x: timestamps, y: prices, mode: "lines", line: {color: "#1f77b4"}}], layout);
socket.onmessage = function(event) {
const data = JSON.parse(event.data);
timestamps.push(data.timestamp);
prices.push(data.price);
// Keep only the latest 50 points for readability
if (timestamps.length > 50) {
timestamps.shift();
prices.shift();
}
Plotly.update("chart", {x: [timestamps], y: [prices]});
};
socket.onerror = function(err) {
console.error("WebSocket error:", err);
};
</script>
</body>
</html>
The JavaScript establishes a WebSocket connection to ws://localhost:8000/ws/prices. Each incoming JSON payload updates two arrays that Plotly uses to redraw the line chart. By capping the arrays at 50 entries we avoid memory bloat and keep the UI snappy.
Pro tip: For production, switch the hard‑coded URL to a relative path (e.g., new WebSocket(`${location.origin.replace(/^http/, "ws")}/ws/prices`)) so the app works behind reverse proxies and on HTTPS.
Adding a Simple Data Persistence Layer
While the mock generator is fine for demos, most real‑world dashboards need to store incoming data for later analysis. Let’s quickly integrate SQLite using SQLModel, a thin wrapper around SQLAlchemy that plays nicely with FastAPI.
from sqlmodel import Field, SQLModel, create_engine, Session
class PriceTick(SQLModel, table=True):
id: int = Field(default=None, primary_key=True)
timestamp: str
price: float
engine = create_engine("sqlite:///prices.db")
SQLModel.metadata.create_all(engine)
async def price_stream():
while True:
idx = random.randint(0, len(timestamps) - 1)
tick = {
"timestamp": timestamps[idx],
"price": round(values[idx] + random.uniform(-0.5, 0.5), 2)
}
# Persist to DB
with Session(engine) as session:
session.add(PriceTick(**tick))
session.commit()
yield tick
await asyncio.sleep(0.5)
Now each tick is written to prices.db. You can expose an endpoint to query historical data, enabling users to scrub back in time or download CSV exports.
@app.get("/history")
async def get_history(limit: int = 100):
with Session(engine) as session:
records = session.query(PriceTick).order_by(PriceTick.id.desc()).limit(limit).all()
return records
This endpoint returns a JSON array of the most recent limit records, which the front‑end could fetch on load to pre‑populate the chart.
Testing and Debugging the Real‑Time Flow
Before we ship, let’s verify the WebSocket pipeline. FastAPI’s TestClient works for HTTP routes, but for WebSockets we need WebSocketTestClient from starlette.testclient. Add a test file tests/test_ws.py:
import pytest
from starlette.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_websocket_receives_data():
with client.websocket_connect("/ws/prices") as websocket:
data = websocket.receive_json()
assert "timestamp" in data
assert "price" in data
# Ensure price is a float
assert isinstance(data["price"], float)
Run pytest -q to see the test pass. For live debugging, open the browser’s DevTools Network tab, filter for “WS”, and watch frames as they arrive. Any latency spikes will show up as delayed frames, helping you tune the await asyncio.sleep interval.
Deploying to Production
When you’re ready to go beyond localhost, containerization is the most portable path. Create a Dockerfile that builds the app and serves static assets via uvicorn:
# syntax=docker/dockerfile:1
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# Expose port 8000 for FastAPI
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
Build and run:
docker build -t live-dashboard .
docker run -d -p 8000:8000 live-dashboard
Now the dashboard is reachable at http://your-server-ip:8000. For HTTPS, terminate TLS at a reverse proxy (Nginx or Traefik) and forward to the container. Don’t forget to set proxy_set_header Upgrade $http_upgrade; and proxy_set_header Connection "upgrade"; so WebSocket upgrades work correctly.
Pro tip: Enableuvicorn --workers 4in production to leverage multiple CPU cores. Pair it withGunicornusing theuvicorn.workers.UvicornWorkerfor graceful restarts.
Extending the Dashboard: Alerts and Analytics
Real‑time dashboards become truly valuable when they surface actionable insights. Let’s add a simple threshold alert: if the price moves more than 2% within a 10‑second window, push a browser notification.
# In app/main.py, add a global state holder
from collections import deque
price_window = deque(maxlen=20) # 20 samples ≈ 10 seconds at 0.5s interval
async def price_stream():
while True:
idx = random.randint(0, len(timestamps) - 1)
new_price = round(values[idx] + random.uniform(-0.5, 0.5), 2)
tick = {"timestamp": timestamps[idx], "price": new_price}
price_window.append(new_price)
# Simple percent change check
if len(price_window) == price_window.maxlen:
change = (price_window[-1] - price_window[0]) / price_window[0] * 100
if abs(change) >= 2:
tick["alert"] = f"{change:.2f}% move!"
# Persist and broadcast
with Session(engine) as session:
session.add(PriceTick(**tick))
session.commit()
yield tick
await asyncio.sleep(0.5)
On the front‑end, listen for the alert field and trigger the Notification API:
<script>
if (Notification.permission !== "granted") {
Notification.requestPermission();
}
socket.onmessage = function(event) {
const data = JSON.parse(event.data);
// Existing chart update logic...
if (data.alert) {
new Notification("Price Alert", {body: data.alert});
}
};
</script>
This lightweight approach gives users immediate visual and auditory cues without needing a full‑blown analytics engine.
Performance Considerations and Scaling
WebSocket connections are stateful, so each client consumes a small amount of memory on the server. For a few dozen users, a single FastAPI instance suffices. Once you cross the 100‑connection threshold, think about a message broker like Redis Pub/Sub or NATS to fan‑out price updates.
Here’s a sketch of how you could replace the in‑process price_stream with a Redis channel:
import aioredis
redis = await aioredis.from_url("redis://localhost")
async def price_producer():
while True:
# generate tick...
await redis.publish("prices", json.dumps(tick))
await asyncio.sleep(0.5)
@app.websocket("/ws/prices")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
pubsub = redis.pubsub()
await pubsub.subscribe("prices")
try:
async for message in pubsub.listen():
if message["type"] == "message":
await websocket.send_text(message["data"])
finally:
await pubsub.unsubscribe("prices")
By decoupling the producer from the consumer, you can horizontally scale both sides independently. The broker also offers durability options if you need to replay missed messages for late‑joining clients.
Security Best Practices
Never expose raw market data without authentication in a production environment. FastAPI integrates seamlessly with OAuth2, JWT, or API keys. Below is a minimal JWT dependency you can plug into the WebSocket route.
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
security = HTTPBearer()
SECRET_KEY = "supersecretkey"
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
try:
payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=["HS256"])
return payload
except jwt.PyJWTError:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid token")
@app.websocket("/ws/prices")
async def websocket_endpoint(websocket: WebSocket, token: dict = Depends(verify_token)):
await websocket.accept()
# rest of the logic...