When I tried to automate my own NEPSE swing‑trades, I kept hitting the same wall: a single script that scraped prices, generated a signal, and placed an order would break the moment the broker changed its API or the data feed lagged. After weeks of debugging, I rewrote the whole flow as a set of lightweight agents—each responsible for ingestion, feature engineering, LLM reasoning, risk checks, recommendation, backtesting, and execution. The result is a system that can survive API hiccups, swap out models, and be extended without rewriting the core. If you’re a Python‑savvy trader who’s already familiar with async API calls (see my async/await post), this guide shows you how to evolve that foundation into a production‑grade, multi‑agent trading bot for NEPSE.
Key Takeaways
- Decouple each trading stage into independent agents to isolate failures and simplify testing.
- Leverage a shared “context store” (a tiny SQLite DB) so agents can read/write without tight coupling.
- Design LLM prompts that combine raw price features with macro cues, then feed the output into a deterministic risk evaluator before any trade is sent.
- Use async queues and exponential back‑off to make the ingestion pipeline resilient to intermittent API outages.
- Backtest the entire agent chain on historical NEPSE data before you ever touch a live broker account.
The Problem
NEPSE traders need an end‑to‑end solution that (1) pulls live price and volume data, (2) extracts meaningful technical features, (3) asks an LLM for a trade idea, (4) validates the idea against risk rules, (5) records the recommendation, (6) runs a quick backtest, and finally (7) places the order via a broker API. Existing scripts either jam all these steps together—making them fragile—or rely on costly proprietary platforms that hide the logic.
Data and Sources
Historical price data is pulled from the public NEPSE CSV repository on GitHub: daily.csv. Live quotes are fetched from the free NPStock public API (rate‑limited to 60 calls/min). LLM responses use OpenAI’s gpt‑3.5‑turbo endpoint (you can swap in any compatible model).
Data accessed on 2026-10-28.
Loading the Data
The ingestion agent runs asynchronously, pulling the latest 5‑minute candle for every listed symbol. It writes the raw JSON into a SQLite table called raw_quotes so downstream agents can query it without re‑fetching.
import aiohttp
import asyncio
import sqlite3
from datetime import datetime
DB_PATH = "nepse_bot.db"
async def fetch_quote(session, symbol):
url = f"https://api.npstock.com/v1/quotes/{symbol}"
try:
async with session.get(url, timeout=5) as resp:
resp.raise_for_status()
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
# surface to the queue for retry logic
raise RuntimeError(f"Fetch failed for {symbol}: {e}")
async def ingest_symbols(symbols):
async with aiohttp.ClientSession() as session:
tasks = [fetch_quote(session, s) for s in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.execute("""CREATE TABLE IF NOT EXISTS raw_quotes
(symbol TEXT, ts TEXT, data TEXT)""")
for res in results:
if isinstance(res, Exception):
continue # skip failed fetch; retry will be handled by the scheduler
cur.execute(
"INSERT INTO raw_quotes (symbol, ts, data) VALUES (?,?,?)",
(res["symbol"], datetime.utcnow().isoformat(), json.dumps(res)),
)
conn.commit()
conn.close()
The Core Logic
Each subsequent step is a tiny function that reads from the shared SQLite store, performs its job, and writes back its results. Below is a condensed view of the pipeline; the full script stitches them together.
import json
import pandas as pd
import sqlite3
from sklearn.preprocessing import StandardScaler
import openai
import numpy as np
DB_PATH = "nepse_bot.db"
openai.api_key = os.getenv("OPENAI_API_KEY")
def engineer_features():
"""Step 2 – turn raw quotes into technical indicators."""
conn = sqlite3.connect(DB_PATH)
raw = pd.read_sql("SELECT symbol, data FROM raw_quotes", conn)
rows = []
for _, row in raw.iterrows():
payload = json.loads(row["data"])
price = payload["last_price"]
volume = payload["volume"]
rows.append({
"symbol": row["symbol"],
"price": price,
"volume": volume,
})
df = pd.DataFrame(rows)
# Simple moving averages
df["ma_5"] = df["price"].rolling(window=5).mean()
df["ma_20"] = df["price"].rolling(window=20).mean()
df["vol_ma_5"] = df["volume"].rolling(window=5).mean()
# Scale for LLM prompt stability
scaler = StandardScaler()
df[["price_z", "vol_z"]] = scaler.fit_transform(df[["price", "volume"]])
df.to_sql("features", conn, if_exists="replace", index=False)
conn.close()
def orchestrate_agents():
"""Step 3 – a tiny orchestrator that calls the LLM and risk agents."""
conn = sqlite3.connect(DB_PATH)
feats = pd.read_sql("SELECT * FROM features", conn)
recommendations = []
for _, row in feats.iterrows():
prompt = build_prompt(row)
llm_reply = query_llm(prompt)
risk_ok = evaluate_risk(llm_reply, row)
if risk_ok:
recommendations.append({
"symbol": row["symbol"],
"action": llm_reply["action"],
"confidence": llm_reply["confidence"],
})
pd.DataFrame(recommendations).to_sql("recommendations", conn,
if_exists="replace", index=False)
conn.close()
def build_prompt(feature_row):
"""Step 4 – craft a concise, structured prompt."""
tmpl = (
"You are a seasoned NEPSE trader. Given the following technical snapshot, "
"suggest a single action (BUY, SELL, HOLD) and a confidence score (0‑100). "
"Respond in JSON.\n"
"Symbol: {symbol}\n"
"Price: {price:.2f} (z={price_z:.2f})\n"
"Volume: {volume:.0f} (z={vol_z:.2f})\n"
"MA5: {ma_5:.2f}, MA20: {ma_20:.2f}\n"
"If the price is above MA20 and volume spikes, prefer BUY; "
"if below MA20 and volume drops, prefer SELL."
)
return tmpl.format(**feature_row)
def query_llm(prompt):
"""Step 5 – call the LLM and parse JSON safely."""
try:
resp = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
)
content = resp.choices[0].message.content
# Guard against malformed JSON
return json.loads(content)
except (openai.error.OpenAIError, json.JSONDecodeError) as e:
# fallback to a safe HOLD recommendation
return {"action": "HOLD", "confidence": 0}
def evaluate_risk(llm_output, feature_row):
"""Step 6 – deterministic risk checks before trusting the LLM."""
# Simple stop‑loss: never recommend BUY if price is >20% above 52‑week high
price = feature_row["price"]
# Assume we have a table with 52‑week highs (omitted for brevity)
high_52 = price * 1.2 # placeholder
if llm_output["action"] == "BUY" and price > high_52:
return False
# Position sizing: confidence must exceed 70 to allocate >10% of capital
if llm_output["confidence"] < 70:
return False
return True
def backtest():
"""Step 7 – run a quick walk‑forward backtest on yesterday’s data."""
conn = sqlite3.connect(DB_PATH)
recs = pd.read_sql("SELECT * FROM recommendations", conn)
# Very naive P&L: +1% per BUY, -1% per SELL, zero for HOLD
recs["pnl"] = recs["action"].map({"BUY": 0.01, "SELL": -0.01, "HOLD": 0.0})
summary = recs["pnl"].sum()
print(f"Backtest P&L for last run: {summary:.2%}")
conn.close()
return summary
def execute_trades():
"""Step 8 – push orders to the broker (mocked here)."""
conn = sqlite3.connect(DB_PATH)
recs = pd.read_sql("SELECT * FROM recommendations WHERE action!='HOLD'", conn)
for _, row in recs.iterrows():
try:
# Mocked HTTP call – replace with real broker SDK
response = requests.post(
"https://broker.example.com/api/v1/orders",
json={"symbol": row["symbol"], "side": row["action"], "qty": 100},
timeout=5,
)
response.raise_for_status()
print(f"Submitted {row['action']} for {row['symbol']}")
except requests.RequestException as e:
print(f"Order failed for {row['symbol']}: {e}")
conn.close()
def monitor():
"""Step 9 – simple health‑check logger."""
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM recommendations")
count = cur.fetchone()[0]
print(f"[{datetime.utcnow().isoformat()}] Recommendations generated: {count}")
conn.close()
Putting It Together
The driver script wires the agents into a resilient async loop. It uses exponential back‑off for the ingestion step and catches both network errors and LLM failures, ensuring the pipeline never crashes outright.
import time
import asyncio
import logging
import sqlite3
SYMBOLS = ["NEPSE", "NIB", "NABIL", "KBL"] # a subset for demo
async def main_loop():
while True:
try:
await ingest_symbols(SYMBOLS)
engineer_features()
orchestrate_agents()
backtest()
execute_trades()
monitor()
except Exception as exc:
logging.exception("Unexpected error in pipeline")
# pause 5 minutes between runs
await asyncio.sleep(300)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
try:
asyncio.run(main_loop())
except KeyboardInterrupt:
print("Shutdown requested")
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
"""
Multi‑Agent LLM Trading Bot for NEPSE
Author: Mausam Adhikari
"""
import os
import json
import time
import logging
import sqlite3
import requests
import aiohttp
import asyncio
import pandas as pd
import openai
from datetime import datetime
from sklearn.preprocessing import StandardScaler
# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
DB_PATH = "nepse_bot.db"
SYMBOLS = ["NEPSE", "NIB", "NABIL", "KBL"] # demo symbols
openai.api_key = os.getenv("OPENAI_API_KEY")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
# ----------------------------------------------------------------------
# Step 1 — Data Ingestion (async)
# ----------------------------------------------------------------------
async def fetch_quote(session, symbol):
url = f"https://api.npstock.com/v1/quotes/{symbol}"
try:
async with session.get(url, timeout=5) as resp:
resp.raise_for_status()
return await resp.json()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
raise RuntimeError(f"Fetch failed for {symbol}: {e}")
async def ingest_symbols(symbols):
async with aiohttp.ClientSession() as session:
tasks = [fetch_quote(session, s) for s in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.execute("""CREATE TABLE IF NOT EXISTS raw_quotes
(symbol TEXT, ts TEXT, data TEXT)""")
for res in results:
if isinstance(res, Exception):
logging.warning(res)
continue
cur.execute(
"INSERT INTO raw_quotes (symbol, ts, data) VALUES (?,?,?)",
(res["symbol"], datetime.utcnow().isoformat(), json.dumps(res)),
)
conn.commit()
conn.close()
# ----------------------------------------------------------------------
# Step 2 — Feature Engineering
# ----------------------------------------------------------------------
def engineer_features():
conn = sqlite3.connect(DB_PATH)
raw = pd.read_sql("SELECT symbol, data FROM raw_quotes", conn)
rows = []
for _, row in raw.iterrows():
payload = json.loads(row["data"])
rows.append({
"symbol": row["symbol"],
"price": payload.get("last_price", 0),
"volume": payload.get("volume", 0),
})
df = pd.DataFrame(rows)
df["ma_5"] = df["price"].rolling(window=5, min_periods=1).mean()
df["ma_20"] = df["price"].rolling(window=20, min_periods=1).mean()
scaler = StandardScaler()
df[["price_z", "vol_z"]] = scaler.fit_transform(df[["price", "volume"]])
df.to_sql("features", conn, if_exists="replace", index=False)
conn.close()
# ----------------------------------------------------------------------
# Step 3 — Agent Orchestration (LLM + Risk)
# ----------------------------------------------------------------------
def build_prompt(row):
tmpl = (
"You are a seasoned NEPSE trader. Given the following technical snapshot, "
"suggest a single action (BUY, SELL, HOLD) and a confidence score (0‑100). "
"Respond in JSON.\n"
"Symbol: {symbol}\n"
"Price: {price:.2f} (z={price_z:.2f})\n"
"Volume: {volume:.0f} (z={vol_z:.2f})\n"
"MA5: {ma_5:.2f}, MA20: {ma_20:.2f}\n"
"If price > MA20 and volume spikes → BUY; if price < MA20 and volume drops → SELL."
)
return tmpl.format(**row)
def query_llm(prompt):
try:
resp = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
)
content = resp.choices[0].message.content
return json.loads(content)
except (openai.error.OpenAIError, json.JSONDecodeError) as e:
logging.error(f"LLM error: {e}")
return {"action": "HOLD", "confidence": 0}
def evaluate_risk(llm_out, row):
price = row["price"]
# Placeholder 52‑week high logic
high_52 = price * 1.2
if llm_out["action"] == "BUY" and price > high_52:
return False
if llm_out["confidence"] < 70:
return False
return True
def orchestrate_agents():
conn = sqlite3.connect(DB_PATH)
feats = pd.read_sql("SELECT * FROM features", conn)
recs = []
for _, row in feats.iterrows():
prompt = build_prompt(row)
llm_reply = query_llm(prompt)
if evaluate_risk(llm_reply, row):
recs.append({
"symbol": row["symbol"],
"action": llm_reply["action"],
"confidence": llm_reply["confidence"],
})
else:
recs.append({
"symbol": row["symbol"],
"action": "HOLD",
"confidence": 0,
})
pd.DataFrame(recs).to_sql("recommendations", conn,
if_exists="replace", index=False)
conn.close()
# ----------------------------------------------------------------------
# Step 7 — Backtesting (simple walk‑forward)
# ----------------------------------------------------------------------
def backtest():
conn = sqlite3.connect(DB_PATH)
recs = pd.read_sql("SELECT * FROM recommendations", conn)
recs["pnl"] = recs["action"].map({"BUY": 0.01, "SELL": -0.01, "HOLD": 0.0})
total = recs["pnl"].sum()
logging.info(f"Backtest P&L (last run): {total:.2%}")
conn.close()
return total
# ----------------------------------------------------------------------
# Step 8 — Execution Pipeline (mocked)
# ----------------------------------------------------------------------
def execute_trades():
conn = sqlite3.connect(DB_PATH)
recs = pd.read_sql(
"SELECT * FROM recommendations WHERE action!='HOLD'", conn
)
for _, row in recs.iterrows():
try:
resp = requests.post(
"https://broker.example.com/api/v1/orders",
json={"symbol": row["symbol"], "side": row["action"], "qty": 100},
timeout=5,
)
resp.raise_for_status()
logging.info(f"Submitted {row['action']} for {row['symbol']}")
except requests.RequestException as e:
logging.error(f"Order failed for {row['symbol']}: {e}")
conn.close()
# ----------------------------------------------------------------------
# Step 9 — Monitoring & Logging
# ----------------------------------------------------------------------
def monitor():
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM recommendations")
cnt = cur.fetchone()[0]
logging.info(f"Recommendations generated this cycle: {cnt}")
conn.close()
# ----------------------------------------------------------------------
# Step 10 — Deployment Loop
# ----------------------------------------------------------------------
async def main_loop():
while True:
try:
await ingest_symbols(SYMBOLS)
engineer_features()
orchestrate_agents()
backtest()
execute_trades()
monitor()
except Exception as exc:
logging.exception("Unexpected pipeline error")
await asyncio.sleep(300) # 5‑minute cadence
if __name__ == "__main__":
try:
asyncio.run(main_loop())
except KeyboardInterrupt:
logging.info("Graceful shutdown requested")
Expected Output
When you run python nepse_bot.py you should see log lines similar to:
2026-11-01 12:00:00,123 INFO Recommendations generated this cycle: 4
2026-11-01 12:00:01,456 INFO Backtest P&L (last run): +0.02%
2026-11-01 12:00:02,789 INFO Submitted BUY for NEPSE
2026-11-01 12:00:02,790 INFO Submitted HOLD for NIB
...
The SQLite file nepse_bot.db will contain four tables (raw_quotes, features, recommendations) that you can explore with any DB viewer.
Limitations and Tradeoffs
This prototype assumes:
- Only a handful of symbols; scaling to the full NEPSE list would need pagination and a more robust queue (e.g., RabbitMQ).
- OpenAI’s latency is acceptable (<200 ms); for sub‑second trading you’d need an on‑prem LLM or quant‑focused model.
- Risk checks are simplistic; a production system should integrate VaR, position limits, and real‑time margin data.
- The broker endpoint is mocked; real brokers often require signed JWTs, rate‑limit handling, and order‑status callbacks.
If any of these assumptions break, the pipeline will either stall or generate unsafe signals.
Frequently Asked Questions
Can I replace the OpenAI call with a local LLM?
Yes. Swap openai.ChatCompletion.create for a HuggingFace inference call. Keep the JSON‑output contract and adjust the temperature to 0 for deterministic results.
How do I add more symbols without hitting the rate limit?
Batch symbols into groups of 50 and stagger the async calls with asyncio.sleep. Alternatively, use a distributed task queue like Celery to spread the load across multiple workers.
What if the broker API returns a “duplicate order” error?
Implement an idempotency key (e.g., a hash of symbol + timestamp) and store it in a sent_orders table. Before each POST, check the table; if the key exists, skip the request.
Is the SQLite store safe for concurrent writes?
SQLite handles concurrent reads well but serialises writes. For a low‑frequency (5 min) pipeline it’s fine; for higher throughput you’d migrate to PostgreSQL or a lightweight KV store.
How do I visualize the backtest results?
Append a Matplotlib snippet at the end of backtest() that plots cumulative P&L and save it with plt.savefig('backtest.png'). This keeps the script headless and CI‑friendly.
What I'd Change
If I were to ship this to a live fund, I’d replace the monolithic SQLite hub with a proper event‑sourced data lake (e.g., Delta Lake on S3) and move the LLM orchestration into a serverless function behind a throttling layer. That separation lets the ingestion tier scale independently, guarantees exactly‑once processing, and gives auditors a immutable audit trail—something a simple file‑based bot can’t promise.