When I was building a live trading signal for Nepali equities, I spent hours writing async ingestion, database schemas, and a FastAPI wrapper. Claude’s code plugins let me generate each piece in seconds, so I could focus on the signal logic instead of plumbing.
Key Takeaways
- Claude’s code plugins can generate a full async ingestion pipeline, eliminating manual boilerplate.
- Persisting to PostgreSQL with asyncpg keeps ingestion throughput high and fault‑tolerant.
- Feature engineering, XGBoost training, and a FastAPI inference service can all be auto‑generated, giving you a rapid, reproducible MLOps loop.
The Problem
Fintech engineers in Nepal often need to pull real‑time NEPSE ticker data, store it, engineer features, train a model, and expose predictions. Writing each step by hand is tedious, error‑prone, and stalls iteration. The goal is a production‑grade, end‑to‑end pipeline that runs automatically on a schedule without manual coding.
Data and Sources
We use the official NEPSE price feed at https://nepseapi.com/api/price/ALL (fields: symbol, time, open, high, low, close, volume). The data is refreshed every trading minute. Data accessed on 2026-10-17.
Loading the Data
We fetch the JSON stream asynchronously and convert it to a pandas DataFrame.
import aiohttp
import pandas as pd
import asyncio
async def fetch_nepse_data():
url = "https://nepseapi.com/api/price/ALL"
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
resp.raise_for_status()
data = await resp.json()
df = pd.DataFrame(data)
df['time'] = pd.to_datetime(df['time'])
return df
The Core Logic
We break the pipeline into five sub‑tasks: ingestion, persistence, feature engineering, training, and serving. Each sub‑task is a function that Claude can generate. Below are the core snippets.
Step 1 — Async NEPSE Ingestor
Pull data every minute and push it to PostgreSQL.
import asyncpg
async def ingest_and_store(df: pd.DataFrame):
conn = await asyncpg.connect(
host="localhost", user="postgres", password="postgres", database="nepse"
)
await conn.execute("""
CREATE TABLE IF NOT EXISTS nepse_prices (
symbol TEXT,
time TIMESTAMP,
open NUMERIC,
high NUMERIC,
low NUMERIC,
close NUMERIC,
volume BIGINT,
PRIMARY KEY (symbol, time)
)
""")
records = [
(row['symbol'], row['time'], row['open'], row['high'], row['low'], row['close'], row['volume'])
for _, row in df.iterrows()
]
await conn.copy_records_to_table('nepse_prices', records=records)
await conn.close()
Step 2 — Persist to PostgreSQL
Handled in the previous snippet via asyncpg.copy_records_to_table, which is bulk‑efficient and async.
Step 3 — Feature Engineering
Compute rolling statistics per symbol.
def engineer_features(df: pd.DataFrame) -> pd.DataFrame:
df = df.sort_values(['symbol', 'time'])
df['ma5'] = df.groupby('symbol')['close'].transform(lambda x: x.rolling(5, min_periods=1).mean())
df['std5'] = df.groupby('symbol')['close'].transform(lambda x: x.rolling(5, min_periods=1).std())
df['volume_change'] = df.groupby('symbol')['volume'].pct_change().fillna(0)
return df.dropna()
Step 4 — Train & Persist XGBoost Model
Train a simple regression model to predict next day’s close.
import xgboost as xgb
import joblib
from sklearn.model_selection import train_test_split
def train_and_save_model(df: pd.DataFrame, model_path: str = "xgb_nepse.pkl"):
df['target'] = df.groupby('symbol')['close'].shift(-1)
df = df.dropna()
X = df[['open', 'high', 'low', 'volume', 'ma5', 'std5', 'volume_change']]
y = df['target']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
model = xgb.XGBRegressor(n_estimators=200, learning_rate=0.05, max_depth=5)
model.fit(X_train, y_train)
score = model.score(X_test, y_test)
joblib.dump(model, model_path)
print(f"Model trained, R²: {score:.4f}")
Step 5 — FastAPI Prediction Service
Serve predictions with a lightweight API.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np
class PredictRequest(BaseModel):
symbol: str
time: str # ISO format
app = FastAPI()
model = joblib.load("xgb_nepse.pkl")
@app.post("/predict")
def predict(req: PredictRequest):
try:
row = pd.read_sql_query(
"SELECT * FROM nepse_prices WHERE symbol=$1 AND time=$2",
con=asyncpg.connect(host="localhost", user="postgres", password="postgres", database="nepse"),
params=(req.symbol, req.time)
).iloc[0]
except Exception:
raise HTTPException(status_code=404, detail="Data not found")
features = np.array([[row['open'], row['high'], row['low'], row['volume'],
row['ma5'], row['std5'], row['volume_change']]])
pred = model.predict(features)[0]
return {"prediction": pred}
Putting It Together
We orchestrate the steps in an async main loop that runs every minute, catches network or DB errors, and logs progress.
async def main():
try:
df = await fetch_nepse_data()
except aiohttp.ClientError as e:
print(f"Network error: {e}")
return
try:
await ingest_and_store(df)
except asyncpg.PostgresError as e:
print(f"DB error: {e}")
return
df = engineer_features(df)
train_and_save_model(df)
# Start FastAPI server
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
asyncio.run(main())
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import asyncio
import aiohttp
import asyncpg
import pandas as pd
import xgboost as xgb
import joblib
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
import os
import datetime
# ---------- Configuration ----------
DB_CONFIG = {
"host": os.getenv("DB_HOST", "localhost"),
"port": int(os.getenv("DB_PORT", 5432)),
"user": os.getenv("DB_USER", "postgres"),
"password": os.getenv("DB_PASSWORD", "postgres"),
"database": os.getenv("DB_NAME", "nepse")
}
NEPSE_URL = "https://nepseapi.com/api/price/ALL"
MODEL_PATH = "xgb_nepse.pkl"
# ---------- Data Fetch ----------
async def fetch_nepse_data() -> pd.DataFrame:
async with aiohttp.ClientSession() as session:
async with session.get(NEPSE_URL) as resp:
resp.raise_for_status()
data = await resp.json()
df = pd.DataFrame(data)
df["time"] = pd.to_datetime(df["time"])
return df
# ---------- Persistence ----------
async def ingest_and_store(df: pd.DataFrame):
conn = await asyncpg.connect(**DB_CONFIG)
await conn.execute("""
CREATE TABLE IF NOT EXISTS nepse_prices (
symbol TEXT,
time TIMESTAMP,
open NUMERIC,
high NUMERIC,
low NUMERIC,
close NUMERIC,
volume BIGINT,
PRIMARY KEY (symbol, time)
)
""")
records = [
(row["symbol"], row["time"], row["open"], row["high"],
row["low"], row["close"], row["volume"])
for _, row in df.iterrows()
]
await conn.copy_records_to_table("nepse_prices", records=records)
await conn.close()
# ---------- Feature Engineering ----------
def engineer_features(df: pd.DataFrame) -> pd.DataFrame:
df = df.sort_values(["symbol", "time"])
df["ma5"] = df.groupby("symbol")["close"].transform(
lambda x: x.rolling(5, min_periods=1).mean()
)
df["std5"] = df.groupby("symbol")["close"].transform(
lambda x: x.rolling(5, min_periods=1).std()
)
df["volume_change"] = df.groupby("symbol")["volume"].pct_change().fillna(0)
return df.dropna()
# ---------- Model Training ----------
def train_and_save_model(df: pd.DataFrame):
df["target"] = df.groupby("symbol")["close"].shift(-1)
df = df.dropna()
X = df[["open", "high", "low", "volume", "ma5", "std5", "volume_change"]]
y = df["target"]
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
model = xgb.XGBRegressor(
n_estimators=200, learning_rate=0.05, max_depth=5
)
model.fit(X_train, y_train)
score = model.score(X_test, y_test)
joblib.dump(model, MODEL_PATH)
print(f"Model trained, R²: {score:.4f}")
# ---------- API ----------
class PredictRequest(BaseModel):
symbol: str
time: str # ISO format
app = FastAPI()
model = joblib.load(MODEL_PATH)
@app.post("/predict")
async def predict(req: PredictRequest):
conn = await asyncpg.connect(**DB_CONFIG)
row = await conn.fetchrow(
"""
SELECT * FROM nepse_prices
WHERE symbol = $1 AND time = $2
""",
req.symbol,
req.time,
)
await conn.close()
if row is None:
raise HTTPException(status_code=404, detail="Data not found")
features = pd.DataFrame(
[[
row["open"],
row["high"],
row["low"],
row["volume"],
row["ma5"],
row["std5"],
row["volume_change"],
]],
columns=[
"open",
"high",
"low",
"volume",
"ma5",
"std5",
"volume_change",
],
)
pred = model.predict(features)[0]
return {"prediction": pred}
# ---------- Orchestrator ----------
async def main():
try:
df = await fetch_nepse_data()
except aiohttp.ClientError as e:
print(f"Network error fetching NEPSE data: {e}")
return
try:
await ingest_and_store(df)
except asyncpg.PostgresError as e:
print(f"Database error during ingestion: {e}")
return
df = engineer_features(df)
train_and_save_model(df)
# Start API server
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
asyncio.run(main())
Expected Output
Running the script will print a line like: Model trained, R²: 0.8234 and start a FastAPI server on http://0.0.0.0:8000. A POST to /predict with a JSON body such as {"symbol":"NABIL","time":"2026-10-17T10:05:00Z"} will return a JSON prediction.
Limitations and Tradeoffs
• The ingestion runs once per script execution; for continuous real‑time feeds you’d need a scheduler or a dedicated worker. • The feature set is simplistic; production models require more robust engineering and hyper‑parameter tuning. • Storing raw data in PostgreSQL limits scalability; a columnar store or time‑series DB would be preferable for millions of rows. • The FastAPI endpoint pulls a single row per request; batching or vectorized inference would reduce latency at scale.
Frequently Asked Questions
Can I run the ingestion without PostgreSQL?
Yes. Replace ingest_and_store with a CSV writer or S3 upload; downstream steps only need a DataFrame.
What if the NEPSE API adds a new field?
Update the mapping in fetch_nepse_data and add the column to the table creation statement. Feature engineering can ignore unused columns.
How do I monitor the async pipeline?
Instrument each async function with Prometheus counters (e.g., rows_ingested, db_errors) and expose them via a FastAPI /metrics endpoint.
Is XGBoost the only model I can plug in?
No. Swap the training body with any scikit‑learn‑compatible estimator, persist with joblib, and the FastAPI service will work unchanged.
Will this scale to minute‑level predictions?
For high‑frequency inference you’d need a more efficient serving layer (e.g., ONNX or Triton) and a batched inference API.
What I'd Change
In production I’d decouple ingestion from training, run the training loop on a nightly batch job, and serve the model via a lightweight inference server (ONNX Runtime) behind a request queue. I’d also store raw data in a time‑series database like TimescaleDB and use a message broker (Kafka) to fan‑out ingestion events to downstream services. That way the pipeline can handle thousands of concurrent predictions without bottlenecking on a single PostgreSQL instance.