Yesterday I tried to run my batch‑oriented liquidity forecaster on a live NEPSE feed and watched the latency balloon as soon as the order book turned sparse. If you’ve already built a nightly model for Nepal’s market, you know the next logical step is “intraday” – but the moment you add a real‑time stream you hit three blockers: blocking HTTP calls, rolling feature windows that never finish, and a model that can’t adapt without a full retrain. In this tutorial I’ll show you how to stitch together an asynchronous poller, a sliding‑window feature buffer, an incremental scikit‑learn model, and a lightweight LLM‑style summarizer so you can push predictions every few seconds while the market is open. By the end you’ll have a runnable script that streams JSONPlaceholder todos (our stand‑in for live ticks), computes rolling completion‑rate metrics, continuously updates an SGDRegressor, and prints a human‑readable market brief.
Key Takeaways
- Asynchronous
httpx.AsyncClientlets you poll high‑frequency feeds without freezing the event loop. - A
collections.dequebuffer provides O(1) sliding‑window updates for rolling technical indicators. SGDRegressor.partial_fitturns a batch model into an online learner that adapts to drift on the fly.- Simple prompt templates combined with a local LLM (or a rule‑based stub) can turn numeric forecasts into executive‑level commentary.
- The pattern scales to real market feeds; just replace the placeholder URL and tune the polling interval.
The Problem
Our earlier liquidity forecaster ran once per night on a static CSV dump of NEPSE trades. Intraday traders need a stream that delivers a fresh price‑direction signal every few seconds, yet the exchange only pushes sparse order‑book updates and occasional news blobs. The challenge is to ingest these micro‑batches without blocking, compute rolling indicators on the fly, keep the model current without costly full retraining, and finally surface the prediction as a concise market brief that non‑technical stakeholders can read.
Data and Sources
We simulate a live tick feed with the public JSONPlaceholder todos endpoint: https://jsonplaceholder.typicode.com/todos. Each todo represents a “tick” – userId maps to a stock symbol, completed indicates whether the trade was executed, and title carries a free‑form comment. In production you would swap this URL for the NEPSE WebSocket or HTTP streaming endpoint.
Data accessed on 2026-10-12.
Loading the Data
We fetch the JSON payload asynchronously, convert it to a pandas DataFrame, and normalize the fields we care about. The function also catches network errors and returns an empty list on failure so the pipeline can keep running.
import asyncio, httpx, pandas as pd
async def fetch_batch(session: httpx.AsyncClient, url: str) -> pd.DataFrame:
try:
resp = await session.get(url, timeout=5.0)
resp.raise_for_status()
raw = resp.json()
df = pd.DataFrame(raw)
# Normalization: map userId → symbol, completed → executed flag
df.rename(columns={'userId': 'symbol_id',
'completed': 'executed',
'title': 'note'}, inplace=True)
return df
except (httpx.RequestError, httpx.HTTPStatusError) as exc:
print(f"⚠️ fetch error: {exc}")
return pd.DataFrame(columns=['symbol_id', 'executed', 'note'])
The Core Logic
Four functions compose the streaming pipeline.
from collections import deque
from sklearn.linear_model import SGDRegressor
import numpy as np
# 1️⃣ Sliding‑window buffer for each symbol
class WindowBuffer:
def __init__(self, size: int = 20):
self.size = size
self.buffers = {} # symbol_id → deque
def add(self, symbol_id: int, executed: int):
buf = self.buffers.setdefault(symbol_id, deque(maxlen=self.size))
buf.append(executed)
def ma(self, symbol_id: int) -> float:
buf = self.buffers.get(symbol_id, [])
return np.mean(buf) if buf else 0.0
# 2️⃣ Incremental model – predicts next execution probability
class OnlineForecaster:
def __init__(self):
# SGDRegressor expects 2‑D X, y in {0,1}
self.model = SGDRegressor(loss='log', max_iter=1, learning_rate='optimal')
self.initialized = False
def update(self, X: np.ndarray, y: np.ndarray):
if not self.initialized:
# First call needs a class list for partial_fit
self.model.partial_fit(X, y, classes=np.array([0, 1]))
self.initialized = True
else:
self.model.partial_fit(X, y)
def predict(self, X: np.ndarray) -> float:
return float(self.model.predict(X)[0])
3️⃣ Generative commentary – a lightweight template that could be swapped for llama.cpp or any local LLM. It receives the symbol, the rolling MA, and the model’s probability, then returns a short English sentence.
def generate_comment(symbol_id: int, ma: float, prob: float) -> str:
trend = "upward" if prob > 0.6 else "downward" if prob < 0.4 else "neutral"
return (f"Symbol {symbol_id} shows a {trend} momentum (rolling exec rate "
f"{ma:.2%}, predicted execution probability {prob:.2%}).")
4️⃣ Orchestrator – ties everything together in an async loop that polls every 2 seconds, updates the window, fits the model, and prints the commentary.
async def stream_loop(url: str, interval: float = 2.0):
buffer = WindowBuffer(size=30)
forecaster = OnlineForecaster()
async with httpx.AsyncClient() as client:
while True:
batch = await fetch_batch(client, url)
if batch.empty:
await asyncio.sleep(interval)
continue
# Process each row as a micro‑tick
for _, row in batch.iterrows():
symbol = int(row['symbol_id'])
executed = int(row['executed'])
buffer.add(symbol, executed)
# Feature vector: [rolling MA]
X = np.array([[buffer.ma(symbol)]])
y = np.array([executed])
forecaster.update(X, y)
prob = forecaster.predict(X)
comment = generate_comment(symbol, X[0,0], prob)
print(comment)
await asyncio.sleep(interval)
Putting It Together
The entry point launches the async stream. We also guard against KeyboardInterrupt so the script exits cleanly.
def main():
url = "https://jsonplaceholder.typicode.com/todos"
try:
asyncio.run(stream_loop(url))
except KeyboardInterrupt:
print("\n🛑 Streaming stopped by user.")
if __name__ == "__main__":
main()
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import asyncio
import httpx
import pandas as pd
import numpy as np
from collections import deque
from sklearn.linear_model import SGDRegressor
# ---------- Data Ingestion ----------
async def fetch_batch(session: httpx.AsyncClient, url: str) -> pd.DataFrame:
try:
resp = await session.get(url, timeout=5.0)
resp.raise_for_status()
raw = resp.json()
df = pd.DataFrame(raw)
df.rename(columns={'userId': 'symbol_id',
'completed': 'executed',
'title': 'note'}, inplace=True)
return df
except (httpx.RequestError, httpx.HTTPStatusError) as exc:
print(f"⚠️ fetch error: {exc}")
return pd.DataFrame(columns=['symbol_id', 'executed', 'note'])
# ---------- Sliding Window ----------
class WindowBuffer:
def __init__(self, size: int = 20):
self.size = size
self.buffers = {}
def add(self, symbol_id: int, executed: int):
buf = self.buffers.setdefault(symbol_id, deque(maxlen=self.size))
buf.append(executed)
def ma(self, symbol_id: int) -> float:
buf = self.buffers.get(symbol_id, [])
return np.mean(buf) if buf else 0.0
# ---------- Incremental Model ----------
class OnlineForecaster:
def __init__(self):
self.model = SGDRegressor(loss='log', max_iter=1, learning_rate='optimal')
self.initialized = False
def update(self, X: np.ndarray, y: np.ndarray):
if not self.initialized:
self.model.partial_fit(X, y, classes=np.array([0, 1]))
self.initialized = True
else:
self.model.partial_fit(X, y)
def predict(self, X: np.ndarray) -> float:
return float(self.model.predict(X)[0])
# ---------- Generative Commentary ----------
def generate_comment(symbol_id: int, ma: float, prob: float) -> str:
trend = "upward" if prob > 0.6 else "downward" if prob < 0.4 else "neutral"
return (f"Symbol {symbol_id} shows a {trend} momentum (rolling exec rate "
f"{ma:.2%}, predicted execution probability {prob:.2%}).")
# ---------- Orchestration ----------
async def stream_loop(url: str, interval: float = 2.0):
buffer = WindowBuffer(size=30)
forecaster = OnlineForecaster()
async with httpx.AsyncClient() as client:
while True:
batch = await fetch_batch(client, url)
if batch.empty:
await asyncio.sleep(interval)
continue
for _, row in batch.iterrows():
symbol = int(row['symbol_id'])
executed = int(row['executed'])
buffer.add(symbol, executed)
X = np.array([[buffer.ma(symbol)]])
y = np.array([executed])
forecaster.update(X, y)
prob = forecaster.predict(X)
comment = generate_comment(symbol, X[0,0], prob)
print(comment)
await asyncio.sleep(interval)
def main():
url = "https://jsonplaceholder.typicode.com/todos"
try:
asyncio.run(stream_loop(url))
except KeyboardInterrupt:
print("\n🛑 Streaming stopped by user.")
if __name__ == "__main__":
main()
Expected Output
Running the script prints a line every 2 seconds similar to:
Symbol 3 shows a neutral momentum (rolling exec rate 45.00%, predicted execution probability 48.73%).
Symbol 7 shows an upward momentum (rolling exec rate 60.00%, predicted execution probability 62.10%).
…Because the placeholder data is static, the numbers will converge quickly; with a live NEPSE feed you’ll see the probabilities drift as market conditions change.
Limitations and Tradeoffs
This architecture is great for sub‑second to few‑second latency scenarios, but it isn’t built for high‑frequency trading where microseconds matter – a C++‑based order‑book processor would be required. The rolling‑window buffer lives entirely in memory; a symbol universe of thousands would need a more scalable store (Redis or a time‑series DB). Using SGDRegressor assumes a linear relationship; for highly non‑linear price dynamics you’d swap in an online tree ensemble or a tiny neural net. Finally, the “LLM” commentary is a stub; integrating a true local LLM adds GPU/CPU overhead and model‑version management.
Frequently Asked Questions
Can I replace the placeholder URL with a real WebSocket feed?
Yes. Replace fetch_batch with an async WebSocket client (e.g., websockets) that yields JSON messages, then feed them into the same buffer/model pipeline.
Why use SGDRegressor.partial_fit instead of fit?
partial_fit updates model weights incrementally, avoiding a full retrain on every micro‑batch and keeping CPU usage low – essential for a continuously running service.
What if the incoming JSON schema changes?
Guard the parsing step with try/except KeyError and emit a warning; you can also version‑control the schema mapping in a small config file.
Is the deque buffer thread‑safe?
In a single‑threaded asyncio loop it’s safe. If you move to a multi‑process architecture you’d need a process‑safe queue or external store.
How do I persist model state across restarts?
Serialize the SGDRegressor with joblib.dump after each successful batch and reload it at startup; similarly, pickle the WindowBuffer contents.
What I'd Change
If I were to production‑ize this for NEPSE, I’d swap the in‑memory deque for a Redis Stream so multiple workers can share state, upgrade the linear model to an online gradient‑boosted tree (e.g., river library) for better non‑linear capture, and plug a locally‑served Llama 2 model behind a FastAPI endpoint to generate richer, context‑aware commentaries. Those upgrades raise operational complexity, but they close the gap between a demo and a reliable intraday forecasting service that can survive data‑drift, scale to hundreds of symbols, and deliver analyst‑grade narratives on the fly.