When I first let a single LLM decide whether to buy NBL, the answer felt lucky rather than reasoned. The market’s nuance—technical over‑bought signals, a fresh earnings beat, and a swirl of social‑media chatter—needs more than one perspective. If you’re building a production‑grade financial AI for Nepal’s exchange, you’ll quickly discover that a single “expert” can’t capture all the angles. In this tutorial I’ll show you how to assemble a council of specialist agents (fundamental, technical, sentiment, and risk) that talk to each other, surface conflicts, and emerge with a consensus‑driven recommendation. By the end you’ll have a runnable Python script that mimics a mini‑boardroom debate and a clear judgment on why a collaborative LLM beats a lone one.
Key Takeaways
- Define narrow, purpose‑built LLM agents to inject domain expertise (fundamentals, technicals, sentiment, risk).
- Wrap reusable analytics (historical data loaders, indicator calculators, sentiment extractors) as tools that agents can invoke on demand.
- Use an orchestrator to collect initial reports, let a risk manager surface contradictions, and synthesize a final consensus via a dedicated LLM call.
- Anticipate latency and cost: parallelize tool execution, cache expensive LLM responses, and set timeouts.
- Validate the pipeline with real NEPSE historical CSV and static news samples before moving to live feeds.
The Problem
Individual LLM agents excel at context‑aware analysis, but they lack the breadth needed for high‑stakes NEPSE decisions. A single model may overlook a technical over‑bought condition or dismiss a subtle regulatory rumor. The challenge is to break a monolithic decision into a structured debate where each specialist contributes, a risk manager flags inconsistencies, and a consensus engine produces a justified BUY/SELL/HOLD signal.
Data and Sources
- nepse_nbl_historical_data.csv – Daily OHLCV for Nepal Bank Limited (NBL) from 2022‑01‑01 to 2024‑06‑30.
- OpenAI Chat Completion API – official docs: OpenAI Chat API.
- LangChain Agent Executors & Tools documentation: LangChain Agents.
- Previous post “Beyond Extraction: Engineering an LLM Agent for Context‑Aware NEPSE Trade Decisions” – foundational single‑agent patterns.
- Sharesansar.com – market commentary and macro context for NEPSE (reference only).
Loading the Data
We start by reading the CSV, filtering by ticker and date range, and converting the date column to a proper datetime. Errors such as missing files or malformed dates are caught early.
import pandas as pd
import os
def load_stock_data(symbol: str, start: str, end: str) -> pd.DataFrame:
path = os.path.join(os.path.dirname(__file__), "nepse_nbl_historical_data.csv")
try:
df = pd.read_csv(path, parse_dates=["Date"])
except FileNotFoundError as e:
raise FileNotFoundError("Historical data file not found.") from e
mask = (df["Symbol"] == symbol) & (df["Date"] >= start) & (df["Date"] <= end)
filtered = df.loc[mask].sort_values("Date")
if filtered.empty:
raise ValueError(f"No data for {symbol} between {start} and {end}.")
return filtered
Step 1 — Defining the Analyst Council
Each analyst is a thin wrapper around an LLM call with a role‑specific system prompt. The prompt tells the model what to focus on and how to format its output.
from langchain.agents import AgentExecutor, Tool
from langchain.llms import OpenAI
# Base LLM (you can swap models later)
llm = OpenAI(temperature=0.0, model_name="gpt-3.5-turbo")
def make_agent(name: str, system_prompt: str, tools: list[Tool]) -> AgentExecutor:
return AgentExecutor.from_agent_and_tools(
agent=llm,
tools=tools,
verbose=False,
agent_kwargs={"system_message": system_prompt, "name": name},
)
# System prompts (concise, role‑focused)
FUNDAMENTAL_PROMPT = """You are a Fundamental Analyst for Nepal Bank Limited.
Assess earnings, dividend yield, and macro factors. Return a JSON with:
{
"recommendation": "BUY|SELL|HOLD",
"justification": "...",
"key_metrics": {...}
}
"""
TECHNICAL_PROMPT = """You are a Technical Analyst for Nepal Bank Limited.
Interpret moving averages, RSI, MACD, and price patterns. Return a JSON with the same schema."""
SENTIMENT_PROMPT = """You are a Sentiment Analyst.
Summarize the tone of the supplied news list and output a JSON recommendation."""
RISK_PROMPT = """You are a Risk Manager.
Given the three analyst reports, flag contradictions, highlight market‑level risks, and suggest a risk‑adjusted stance."""
Step 2 — Equipping Agents with Tools
Tools are pure Python functions wrapped so the LLM can call them. They handle data loading, indicator calculation, and sentiment extraction.
import numpy as np
def calculate_technical_indicators(df: pd.DataFrame) -> dict:
df = df.copy()
df["MA_50"] = df["Close"].rolling(50).mean()
df["MA_200"] = df["Close"].rolling(200).mean()
delta = df["Close"].diff()
up = delta.clip(lower=0)
down = -delta.clip(upper=0)
roll_up = up.rolling(14).mean()
roll_down = down.rolling(14).mean()
df["RSI_14"] = 100 - (100 / (1 + roll_up / roll_down))
# Simple MACD
df["EMA_12"] = df["Close"].ewm(span=12, adjust=False).mean()
df["EMA_26"] = df["Close"].ewm(span=26, adjust=False).mean()
df["MACD"] = df["EMA_12"] - df["EMA_26"]
df["MACD_signal"] = df["MACD"].ewm(span=9, adjust=False).mean()
latest = df.iloc[-1]
return {
"MA_50": round(latest["MA_50"], 2),
"MA_200": round(latest["MA_200"], 2),
"RSI_14": round(latest["RSI_14"], 1),
"MACD": round(latest["MACD"], 3),
"MACD_signal": round(latest["MACD_signal"], 3),
}
def read_historical_data(symbol: str, start: str, end: str) -> pd.DataFrame:
return load_stock_data(symbol, start, end)
def analyze_news_sentiment(news: list[dict]) -> str:
# Simple LLM summarizer – returns "positive|negative|neutral"
prompt = "Summarize the overall sentiment of the following news items about Nepal Bank Limited. Respond with only one word: Positive, Negative, or Neutral.\n\n"
for item in news:
prompt += f"- {item['title']}: {item['summary']}\n"
response = llm(prompt)
sentiment = response.strip().lower()
if sentiment not in {"positive", "negative", "neutral"}:
sentiment = "neutral"
return sentiment
# Wrap as LangChain Tools
technical_tool = Tool(
name="TechnicalIndicators",
func=lambda sym, s, e: calculate_technical_indicators(read_historical_data(sym, s, e)),
description="Calculate MA, RSI, MACD for a given symbol and date range."
)
sentiment_tool = Tool(
name="NewsSentiment",
func=analyze_news_sentiment,
description="Return overall sentiment (positive/negative/neutral) for a list of news dicts."
)
Step 3 — Orchestrating the Debate
The Orchestrator runs each analyst, collects their JSON reports, and stores them in a dict. Errors from any agent are captured so the pipeline never crashes mid‑day.
import json
from typing import Dict
class Orchestrator:
def __init__(self, agents: Dict[str, AgentExecutor], risk_agent: AgentExecutor):
self.agents = agents
self.risk_agent = risk_agent
def run(self, symbol: str, start: str, end: str, news: list[dict]) -> dict:
reports = {}
# Fundamental
try:
fund_prompt = f"{FUNDAMENTAL_PROMPT}\nData: earnings grew 15% YoY."
reports["fundamental"] = json.loads(self.agents["fundamental"].run(fund_prompt))
except Exception as e:
reports["fundamental"] = {"error": str(e)}
# Technical
try:
tech_data = technical_tool.func(symbol, start, end)
tech_prompt = f"{TECHNICAL_PROMPT}\nIndicators: {tech_data}"
reports["technical"] = json.loads(self.agents["technical"].run(tech_prompt))
except Exception as e:
reports["technical"] = {"error": str(e)}
# Sentiment
try:
sentiment = sentiment_tool.func(news)
sent_prompt = f"{SENTIMENT_PROMPT}\nOverall sentiment: {sentiment}"
reports["sentiment"] = json.loads(self.agents["sentiment"].run(sent_prompt))
except Exception as e:
reports["sentiment"] = {"error": str(e)}
# Risk assessment
try:
risk_input = json.dumps(reports, indent=2)
risk_prompt = f"{RISK_PROMPT}\nReports:\n{risk_input}"
risk_report = json.loads(self.risk_agent.run(risk_prompt))
reports["risk"] = risk_report
except Exception as e:
reports["risk"] = {"error": str(e)}
# Consensus synthesis (final LLM call)
consensus_prompt = (
"You are a Consensus Agent. Given the four JSON reports above, produce a single "
"recommendation (BUY/SELL/HOLD) with a concise justification and list the top three "
"risks identified. Return a JSON with keys: recommendation, justification, risks."
)
final_output = json.loads(self.agents["consensus"].run(consensus_prompt))
return {"reports": reports, "final": final_output}
Step 4 — Towards Consensus: Inter‑Agent Review and Risk Assessment
The risk manager’s job is to surface contradictions. For example, a technical over‑bought signal (RSI > 70) clashes with a bullish fundamental outlook. The risk prompt explicitly asks the LLM to flag such mismatches.
RISK_PROMPT = """You are a Risk Manager.
Given the JSON reports from Fundamental, Technical, and Sentiment analysts, identify:
1. Direct contradictions (e.g., BUY vs SELL).
2. Market‑level risks (liquidity, regulatory, macro).
Return a JSON:
{
"conflicts": [...],
"key_risks": [...],
"risk_adjusted_recommendation": "BUY|SELL|HOLD"
}
"""
Putting It Together
We wire everything up, create the agents, feed a static news list, and let the orchestrator run. Two explicit error handlers protect against missing data and OpenAI API failures.
import openai
import os
def main():
# Simulated news for NBL
news = [
{"title": "NBL posts Q3 profit surge", "summary": "Earnings up 15% YoY driven by loan growth."},
{"title": "Regulatory tweak on banking fees", "summary": "Potential impact on net interest margin."},
{"title": "Investor sentiment turns cautious", "summary": "Market fears a slowdown in consumer credit."},
]
# Build agents
agents = {
"fundamental": make_agent("FundamentalAnalyst", FUNDAMENTAL_PROMPT, []),
"technical": make_agent("TechnicalAnalyst", TECHNICAL_PROMPT, [technical_tool]),
"sentiment": make_agent("SentimentAnalyst", SENTIMENT_PROMPT, [sentiment_tool]),
"consensus": make_agent("ConsensusAgent", "", []), # system prompt supplied later
}
risk_agent = make_agent("RiskManager", RISK_PROMPT, [])
orchestrator = Orchestrator(agents, risk_agent)
try:
result = orchestrator.run("NBL", "2024-01-01", "2024-06-30", news)
print("=== Final Consensus Recommendation ===")
print(json.dumps(result["final"], indent=2))
except openai.error.OpenAIError as api_err:
print(f"OpenAI API error: {api_err}")
except Exception as exc:
print(f"Unexpected error: {exc}")
if __name__ == "__main__":
main()
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import os
import json
import pandas as pd
import openai
from langchain.agents import AgentExecutor, Tool
from langchain.llms import OpenAI
# ---------- Configuration ----------
openai.api_key = os.getenv("OPENAI_API_KEY")
if not openai.api_key:
raise EnvironmentError("Set the OPENAI_API_KEY environment variable.")
# ---------- Data Loading ----------
def load_stock_data(symbol: str, start: str, end: str) -> pd.DataFrame:
path = os.path.join(os.path.dirname(__file__), "nepse_nbl_historical_data.csv")
try:
df = pd.read_csv(path, parse_dates=["Date"])
except FileNotFoundError as e:
raise FileNotFoundError("Historical data file not found.") from e
mask = (df["Symbol"] == symbol) & (df["Date"] >= start) & (df["Date"] <= end)
filtered = df.loc[mask].sort_values("Date")
if filtered.empty:
raise ValueError(f"No data for {symbol} between {start} and {end}.")
return filtered
# ---------- Tools ----------
def calculate_technical_indicators(df: pd.DataFrame) -> dict:
df = df.copy()
df["MA_50"] = df["Close"].rolling(50).mean()
df["MA_200"] = df["Close"].rolling(200).mean()
delta = df["Close"].diff()
up = delta.clip(lower=0)
down = -delta.clip(upper=0)
roll_up = up.rolling(14).mean()
roll_down = down.rolling(14).mean()
df["RSI_14"] = 100 - (100 / (1 + roll_up / roll_down))
df["EMA_12"] = df["Close"].ewm(span=12, adjust=False).mean()
df["EMA_26"] = df["Close"].ewm(span=26, adjust=False).mean()
df["MACD"] = df["EMA_12"] - df["EMA_26"]
df["MACD_signal"] = df["MACD"].ewm(span=9, adjust=False).mean()
latest = df.iloc[-1]
return {
"MA_50": round(latest["MA_50"], 2),
"MA_200": round(latest["MA_200"], 2),
"RSI_14": round(latest["RSI_14"], 1),
"MACD": round(latest["MACD"], 3),
"MACD_signal": round(latest["MACD_signal"], 3),
}
def read_historical_data(symbol: str, start: str, end: str) -> pd.DataFrame:
return load_stock_data(symbol, start, end)
def analyze_news_sentiment(news: list[dict]) -> str:
prompt = "Summarize the overall sentiment of the following news items about Nepal Bank Limited. Respond with only one word: Positive, Negative, or Neutral.\n\n"
for item in news:
prompt += f"- {item['title']}: {item['summary']}\n"
response = OpenAI(temperature=0.0, model_name="gpt-3.5-turbo")(prompt)
sentiment = response.strip().lower()
return sentiment if sentiment in {"positive", "negative", "neutral"} else "neutral"
technical_tool = Tool(
name="TechnicalIndicators",
func=lambda sym, s, e: calculate_technical_indicators(read_historical_data(sym, s, e)),
description="Calculate MA, RSI, MACD for a given symbol and date range."
)
sentiment_tool = Tool(
name="NewsSentiment",
func=analyze_news_sentiment,
description="Return overall sentiment (positive/negative/neutral) for a list of news dicts."
)
# ---------- Agent Prompts ----------
FUNDAMENTAL_PROMPT = """You are a Fundamental Analyst for Nepal Bank Limited.
Assess earnings, dividend yield, and macro factors. Return a JSON with:
{
"recommendation": "BUY|SELL|HOLD",
"justification": "...",
"key_metrics": {...}
}
"""
TECHNICAL_PROMPT = """You are a Technical Analyst for Nepal Bank Limited.
Interpret moving averages, RSI, MACD, and price patterns. Return a JSON with the same schema."""
SENTIMENT_PROMPT = """You are a Sentiment Analyst.
Summarize the tone of the supplied news list and output a JSON recommendation."""
RISK_PROMPT = """You are a Risk Manager.
Given the JSON reports from Fundamental, Technical, and Sentiment analysts, identify:
1. Direct contradictions (e.g., BUY vs SELL).
2. Market‑level risks (liquidity, regulatory, macro).
Return a JSON:
{
"conflicts": [...],
"key_risks": [...],
"risk_adjusted_recommendation": "BUY|SELL|HOLD"
}
"""
# ---------- Agent Factory ----------
def make_agent(name: str, system_prompt: str, tools: list[Tool]) -> AgentExecutor:
llm = OpenAI(temperature=0.0, model_name="gpt-3.5-turbo")
return AgentExecutor.from_agent_and_tools(
agent=llm,
tools=tools,
verbose=False,
agent_kwargs={"system_message": system_prompt, "name": name},
)
# ---------- Orchestrator ----------
class Orchestrator:
def __init__(self, agents: dict, risk_agent: AgentExecutor):
self.agents = agents
self.risk_agent = risk_agent
def run(self, symbol: str, start: str, end: str, news: list[dict]) -> dict:
reports = {}
# Fundamental
try:
fund_prompt = f"{FUNDAMENTAL_PROMPT}\nData: earnings grew 15% YoY."
reports["fundamental"] = json.loads(self.agents["fundamental"].run(fund_prompt))
except Exception as e:
reports["fundamental"] = {"error": str(e)}
# Technical
try:
tech_data = technical_tool.func(symbol, start, end)
tech_prompt = f"{TECHNICAL_PROMPT}\nIndicators: {tech_data}"
reports["technical"] = json.loads(self.agents["technical"].run(tech_prompt))
except Exception as e:
reports["technical"] = {"error": str(e)}
# Sentiment
try:
sentiment = sentiment_tool.func(news)
sent_prompt = f"{SENTIMENT_PROMPT}\nOverall sentiment: {sentiment}"
reports["sentiment"] = json.loads(self.agents["sentiment"].run(sent_prompt))
except Exception as e:
reports["sentiment"] = {"error": str(e)}
# Risk
try:
risk_input = json.dumps(reports, indent=2)
risk_prompt = f"{RISK_PROMPT}\nReports:\n{risk_input}"
reports["risk"] = json.loads(self.risk_agent.run(risk_prompt))
except Exception as e:
reports["risk"] = {"error": str(e)}
# Consensus
try:
consensus_prompt = (
"You are a Consensus Agent. Given the four JSON reports above, produce a single "
"recommendation (BUY/SELL/HOLD) with a concise justification and list the top three "
"risks identified. Return a JSON with keys: recommendation, justification, risks."
)
final = json.loads(self.agents["consensus"].run(consensus_prompt))
except Exception as e:
final = {"error": str(e)}
return {"reports": reports, "final": final}
# ---------- Main ----------
def main():
news = [
{"title": "NBL posts Q3 profit surge", "summary": "Earnings up 15% YoY driven by loan growth."},
{"title": "Regulatory tweak on banking fees", "summary": "Potential impact on net interest margin."},
{"title": "Investor sentiment turns cautious", "summary": "Market fears a slowdown in consumer credit."},
]
agents = {
"fundamental": make_agent("FundamentalAnalyst", FUNDAMENTAL_PROMPT, []),
"technical": make_agent("TechnicalAnalyst", TECHNICAL_PROMPT, [technical_tool]),
"sentiment": make_agent("SentimentAnalyst", SENTIMENT_PROMPT, [sentiment_tool]),
"consensus": make_agent("ConsensusAgent", "", []),
}
risk_agent = make_agent("RiskManager", RISK_PROMPT, [])
orchestrator = Orchestrator(agents, risk_agent)
try:
result = orchestrator.run("NBL", "2024-01-01", "2024-06-30", news)
print("=== Final Consensus Recommendation ===")
print(json.dumps(result["final"], indent=2))
except openai.error.OpenAIError as api_err:
print(f"OpenAI API error: {api_err}")
except Exception as exc:
print(f"Unexpected error: {exc}")
if __name__ == "__main__":
main()
Expected Output
Running python script.py should print something akin to:
{
"recommendation": "HOLD",
"justification": "Fundamentals are strong (earnings +15% YoY) but technical indicators show RSI 72 (overbought) and price approaching MA_200 resistance. Sentiment is mixed, and risk manager flags regulatory fee changes.",
"risks": [
"Potential regulatory fee increase",
"Technical overbought condition",
"Mixed market sentiment"
]
}
Limitations and Tradeoffs
- Hallucinations & bias: Each LLM still invents numbers if the prompt is ambiguous; rigorous prompt engineering and validation are required.
- Cost & latency: Six separate LLM calls per decision (three analysts, risk, consensus, plus any tool‑driven calls) can add seconds of latency and noticeable API spend.
- Orchestration complexity: Managing state, retries, and timeouts across agents quickly becomes non‑trivial; a workflow engine (e.g., Airflow or Prefect) is advisable for production.
- Data freshness: The static CSV limits the system to back‑testing. Real‑time pipelines need incremental ingestion, versioned storage, and cache invalidation.
- Regulatory exposure: Automated recommendations must be vetted against Nepal’s securities regulations; a compliance layer is mandatory before any live trade execution.
Frequently Asked Questions
How do I handle conflicting recommendations between agents?
Let the Risk Manager surface contradictions and feed them into the Consensus Agent. You can also weight agents (e.g., give technical a higher weight during volatile periods) by adjusting the consensus prompt.
Can I add more specialized agents without rewriting the orchestrator?
Yes. Define a new system prompt, wrap any required tools, register the agent in the agents dict, and extend the orchestrator’s run method to collect its report. The pattern scales linearly.
Which LLM model balances cost and reasoning for multi‑agent setups?
For most Nepal‑focused prototypes, gpt-3.5-turbo offers adequate chain‑of‑thought capabilities at low cost. If you need deeper financial reasoning, consider gpt-4 for the Consensus Agent only, keeping the other agents on the cheaper model.
What steps are needed to turn this prototype into a live trading system?
Replace the CSV loader with a streaming market data connector, implement a message queue (Kafka/RabbitMQ) for news, add authentication & order‑routing modules, and enforce rate‑limit and fail‑over logic. Finally, embed a compliance audit trail.
How should I evaluate the performance of this multi‑agent system?
Beyond traditional back‑test metrics (Sharpe, max drawdown), track consensus stability (how often agents agree), latency per decision, and API cost per trade. A/B test against a single‑agent baseline to quantify the robustness gain.
What I'd Change
If I were to rebuild this for production tomorrow, I'd decouple the LLM layer from the orchestration engine entirely—using a lightweight RPC framework (gRPC) to call stateless LLM micro‑services. That lets you swap models per agent, cache tool results, and scale each analyst independently behind its own autoscaling group. The extra engineering overhead pays off when you move from daily batch runs to sub‑minute trade signals, and it gives you the observability needed to audit every “boardroom” discussion before a trade hits the exchange.