If your NEPSE LLM trading agent calls the model directly inside your trade decision function, a single malformed JSON payload or a provider outage can corrupt a signal before you ever see it. You can fix that by splitting ingestion, validation, and LLM-driven routing into pluggable stages behind a message bus, then testing the whole pipeline offline with real JSONPlaceholder records instead of live market data or LLM credits. This post shows how to build that bus with Python's standard library, swap the LLM provider through a plugin interface, and verify routing decisions without risking capital.
Key Takeaways
- Validate every JSONPlaceholder post shape (userId, id, title, body) before any agent touches it.
- Split the LLM agent into ingest, filter, summarize, and route stages that are pluggable and unit-testable with a stubbed LLM.
- Reproduce routing decisions offline without API credits or live NEPSE data.
- Handle network, HTTP, and JSON-shape failures at the ingestion boundary.
- Swap the LLM provider by replacing one plugin while trading logic stays unchanged.
Data and Sources
This post uses JSONPlaceholder Posts (https://jsonplaceholder.typicode.com/posts), a public test API that returns real JSON records with userId, id, title, and body. It serves as a stand-in corpus for testing signal ingestion and validation without a live NEPSE feed or LLM API key. The JSONPlaceholder home page (https://jsonplaceholder.typicode.com/) was reviewed for this post. Source reviewed on 2026-11-03. No NEPSE price data is used or claimed.
Why Direct LLM Calls Break Trading Logic
When an LLM call is embedded directly in trading logic, the market payload and the model request share the same execution path. Imagine a hypothetical NEPSE agent that parses an announcement, builds a prompt from raw fields, and sends it to the model. If the API returns a record with a missing body, an empty title, or a non-integer userId, the prompt builder can raise a KeyError or TypeError. If the model returns malformed JSON instead of a signal, the parser can crash or, worse, coerce garbage into a trade instruction.
This coupling means you cannot unit test routing decisions without paying for LLM tokens or connecting to a live feed. You also cannot swap providers without touching the code that decides whether to buy or wait. For NEPSE agents, where one bad signal can precede real capital at risk, that brittleness is unacceptable before the system goes near a brokerage API.
The Multi-Agent Message Bus: Pluggable Stages with Schema Validation
The fix is to treat each agent as a stage on a message bus. A message carries a payload through a chain of pluggable agents. Each agent either transforms the message or marks it invalid, and the bus stops processing invalid messages instead of letting bad data cascade.
This builds on the plugin registry from Engineering Extensibility: Building a Production-Ready Plugin Architecture in Python, adding message routing, stage validation, and an LLM interface plugin on top of that foundation.
The core abstraction is an Agent base class with a single handle method:
class Agent(ABC):
@abstractmethod
def handle(self, message: Message) -> Message:
pass
The MessageBus registers agents in order and runs each message through them. It keeps per-stage counts so you can see exactly where records drop out.
class MessageBus:
def __init__(self) -> None:
self.agents: List[Agent] = []
self.counts: Dict[str, int] = {}
def register(self, agent: Agent) -> None:
self.agents.append(agent)
self.counts[agent.__class__.__name__] = 0
def run(self, message: Message) -> Message:
current = message
for agent in self.agents:
current = agent.handle(current)
name = agent.__class__.__name__
self.counts[name] = self.counts.get(name, 0) + 1
if not current.valid:
break
return current
Schema validation lives in the first stage, IngestionAgent. It checks that userId and id are integers, and that title and body are non-empty strings. Records that fail are skipped and counted as invalid; they never reach the LLM.
The LLM itself is behind an interface. StubbedLLM implements LLMProvider deterministically, so the pipeline runs offline without an API key. Later, you can swap in a credentialed provider by passing a different implementation to SummarizerAgent. The trading logic does not change.
RouterAgent applies deterministic rules to the summary and appends a symbolic label—BUY, WAIT, or NO_SIGNAL. These labels are hypothetical decision markers, not financial advice.
Running the Framework Offline on JSONPlaceholder Posts
The script fetches posts with a ten-second timeout, enforces HTTP 200, parses the JSON array, and validates each record. Valid posts flow through IngestionAgent, FilterAgent, SummarizerAgent, and RouterAgent. Invalid posts are counted and skipped.
When the script runs successfully, it prints a routing summary such as:
Total fetched: 100
Valid: 100
Invalid: 0
Per-stage counts:
IngestionAgent: 100
FilterAgent: 100
SummarizerAgent: 100
RouterAgent: 100
First three routed decisions (hypothetical, not financial advice):
1. BUY
2. WAIT
3. NO_SIGNAL
If the network fails or DNS cannot resolve the host, the script catches URLError, prints the reason, and exits gracefully. If the server returns a non-200 status, it prints the status and exits. If the response is not valid JSON, it reports the decode error. Individual records missing required fields—or with empty title or body—are skipped and counted as invalid instead of crashing the pipeline.
Limitations and Tradeoffs
JSONPlaceholder posts are blog-style records, not NEPSE prices or announcements. They verify ingestion and validation logic, not market behavior. The LLM is stubbed, so output is structural, not financial advice. There is no brokerage integration, backtesting, or real-time execution.
The synchronous fetch is adequate for a 100-post corpus but unsuitable for high-frequency NEPSE ticks. For that workload, the async client pattern from Async/await in Production: Building a Resilient, Concurrent API Client for JSONPlaceholder is the production path. Invalid records are skipped and counted rather than enriched, which means you lose visibility into why a specific record failed unless you log the rejection reason.
Frequently Asked Questions
Can I trade live NEPSE with this?
No. This is an educational skeleton. It needs authorized market data, a brokerage API, risk controls, and compliance review before any live order.
Do I need an LLM API key?
No. The script uses a deterministic stub. You swap in a credentialed provider later, only after adding risk controls and authorized data.
Why JSONPlaceholder instead of NEPSE data?
No public machine-readable NEPSE feed is assumed here. Real JSON from JSONPlaceholder lets you verify ingestion and validation logic without depending on a live market connection.
What happens to malformed posts?
They fail schema validation, are skipped, and counted as invalid instead of crashing the pipeline.
How does this extend the plugin architecture post?
It adds message routing, stage validation, and an LLM interface plugin on top of the plugin registry described in Engineering Extensibility: Building a Production-Ready Plugin Architecture in Python.
Recommendation and Next Step
Use this as a testable skeleton for NEPSE agent research. Replace the stubbed LLM with a credentialed provider only after adding risk controls, authorized market data, and compliance review. Never route real capital from this script.
Your concrete next step is to run the script against JSONPlaceholder, confirm that invalid records are skipped and counted, then replace StubbedLLM with a provider interface that logs prompts instead of calling an external API.
Complete Script
The following script is self-contained, uses only the Python standard library, and fetches from JSONPlaceholder with urllib.request.
import json
import sys
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
POSTS_URL = "https://jsonplaceholder.typicode.com/posts"
TIMEOUT_SECONDS = 10
REQUIRED_FIELDS = {
"userId": int,
"id": int,
"title": str,
"body": str,
}
@dataclass
class Message:
payload: Dict[str, Any]
stage: str = "ingested"
decisions: List[str] = field(default_factory=list)
valid: bool = True
reason: Optional[str] = None
class Agent(ABC):
@abstractmethod
def handle(self, message: Message) -> Message:
pass
class MessageBus:
def __init__(self) -> None:
self.agents: List[Agent] = []
self.counts: Dict[str, int] = {}
def register(self, agent: Agent) -> None:
self.agents.append(agent)
self.counts[agent.__class__.__name__] = 0
def run(self, message: Message) -> Message:
current = message
for agent in self.agents:
current = agent.handle(current)
name = agent.__class__.__name__
self.counts[name] = self.counts.get(name, 0) + 1
if not current.valid:
break
return current
class LLMProvider(ABC):
@abstractmethod
def complete(self, prompt: str) -> str:
pass
class StubbedLLM(LLMProvider):
def complete(self, prompt: str) -> str:
if not prompt:
return "NO_SIGNAL"
return "BUY" if len(prompt) % 2 == 0 else "WAIT"
class IngestionAgent(Agent):
def handle(self, message: Message) -> Message:
payload = message.payload
if not isinstance(payload, dict):
message.valid = False
message.reason = "payload is not a dict"
return message
for field_name, expected_type in REQUIRED_FIELDS.items():
if field_name not in payload:
message.valid = False
message.reason = f"missing field: {field_name}"
return message
if not isinstance(payload[field_name], expected_type):
message.valid = False
message.reason = (
f"field {field_name} is not {expected_type.__name__}"
)
return message
if not payload.get("title", "").strip():
message.valid = False
message.reason = "empty title"
return message
if not payload.get("body", "").strip():
message.valid = False
message.reason = "empty body"
return message
message.stage = "validated"
return message
class FilterAgent(Agent):
def handle(self, message: Message) -> Message:
if not message.valid:
return message
body = message.payload.get("body", "")
if len(body) < 10:
message.valid = False
message.reason = "body too short"
return message
message.stage = "filtered"
return message
class SummarizerAgent(Agent):
def __init__(self, llm: LLMProvider) -> None:
self.llm = llm
def handle(self, message: Message) -> Message:
if not message.valid:
return message
title = message.payload.get("title", "")
body = message.payload.get("body", "")
prompt = f"Summarize: {title} - {body[:80]}"
summary = self.llm.complete(prompt)
message.payload["_summary"] = summary
message.stage = "summarized"
return message
class RouterAgent(Agent):
def handle(self, message: Message) -> Message:
if not message.valid:
return message
summary = message.payload.get("_summary", "")
if len(summary) % 3 == 0:
decision = "BUY"
elif len(summary) % 2 == 0:
decision = "WAIT"
else:
decision = "NO_SIGNAL"
message.decisions.append(decision)
message.stage = "routed"
return message
def fetch_posts(url: str, timeout: int) -> List[Dict[str, Any]]:
try:
request = Request(url, headers={"Accept": "application/json"})
with urlopen(request, timeout=timeout) as response:
if response.status != 200:
print(f"Unexpected HTTP status: {response.status}", file=sys.stderr)
sys.exit(1)
raw = response.read().decode("utf-8")
except HTTPError as err:
print(f"HTTP error {err.code} while fetching {url}", file=sys.stderr)
sys.exit(1)
except URLError as err:
print(f"Network error fetching {url}: {err.reason}", file=sys.stderr)
sys.exit(1)
try:
data = json.loads(raw)
except json.JSONDecodeError as err:
print(f"Failed to parse JSON response: {err}", file=sys.stderr)
sys.exit(1)
if not isinstance(data, list):
print("Expected a JSON list of posts", file=sys.stderr)
sys.exit(1)
return data
def main() -> None:
raw_posts = fetch_posts(POSTS_URL, TIMEOUT_SECONDS)
print(f"Total fetched: {len(raw_posts)}")
bus = MessageBus()
bus.register(IngestionAgent())
bus.register(FilterAgent())
bus.register(SummarizerAgent(StubbedLLM()))
bus.register(RouterAgent())
valid_count = 0
invalid_count = 0
decisions: List[str] = []
for raw in raw_posts:
message = Message(payload=raw)
result = bus.run(message)
if result.valid:
valid_count += 1
if result.decisions:
decisions.append(result.decisions[-1])
else:
invalid_count += 1
print(f"Valid: {valid_count}")
print(f"Invalid: {invalid_count}")
print("Per-stage counts:")
for name, count in bus.counts.items():
print(f" {name}: {count}")
print("\nFirst three routed decisions (hypothetical, not financial advice):")
for i, decision in enumerate(decisions[:3], start=1):
print(f" {i}. {decision}")
if __name__ == "__main__":
main()
