Guardrails for LLM-driven trade decisions: keeping a human-shaped veto
The promise of LLM trading agents is intoxicating: ingest unstructured global news, parse sentiment on complex macro shifts, scan corporate filings, and execute sophisticated trades—all at a speed and scale no human analyst can match.
But early in our rollout of an LLM-driven macro-strategy agent, we nearly blew up our account.
On October 14, at 14:02:11 UTC, our trading agent, powered by an LLM parsing geopolitical news feeds, read a report about a pipeline disruption in the North Sea. It attempted to put on a natural gas calendar spread. However, because of a parsing hallucination on the contract multiplier and unit sizes, the LLM confused notional dollar limits with contract counts. It tried to execute a buy order for 45,000 front-month Henry Hub natural gas futures contracts instead of the intended $45,000 position.
If that order had reached the execution venue, the slippage alone would have wiped out our quarterly PnL. The order was blocked not by the LLM’s “self-correction” or a clever prompt engineering template, but by a rigid, deterministic, out-of-band validation layer.
This post details the architecture and implementation of a zero-trust guardrail and human-in-the-loop (HITL) veto system designed specifically for LLM trading agents.
The Core Problem: Non-Deterministic Intelligence Meets Deterministic Risk
Large language models are fundamentally non-deterministic. They operate on probabilities, not mathematical proofs. When you ask an LLM to decide on a trade, you are trading off deterministic safety for semantic understanding.
You cannot fix this with prompting. System instructions like "You must never trade more than 50 contracts" will fail under pressure. This failure occurs because of semantic drift, prompt injections in the source news texts, or unexpected attention-weight distributions when processing multi-page PDF earnings transcripts.
To deploy LLMs safely in financial markets, you must decouple generation from validation. The generation phase can use non-deterministic AI. The validation phase must be written in strict, deterministic code (Python, Rust, or Go) and backed by a human-in-the-loop veto system for high-value or high-risk exceptions.
Our risk control framework relies on a three-tier defense:
- Deterministic Schema Verification: Enforcing strict type parsing and structural validation of the LLM payload before it can be interpreted.
- Hard Guardrails: Immutable boundaries (position sizing, daily drawdowns, asset blacklists, bid-ask spread limits) executed in a local sandbox.
- The Human-Shaped Veto: An asynchronous, non-blocking notification loop that intercepts anomalous trades, routes them to communication channels (like Slack or Discord), and holds the trade in an escrow state until a physical human signs off or a timeout forces a safe rejection.
System Architecture
The following diagram traces a generated trade intent from the LLM agent to market execution, passing through the deterministic guardrails and the interactive human veto stage.
flowchart TD A["LLM Agent Generation"] --> B["Deterministic Guardrails"] B -->|"Violates Hard Rules"| C["Hard Reject Log"] B -->|"Requires Human Veto"| D["Slack Interactive App"] B -->|"Greenlit Auto-Exec"| E["Order Execution Engine"] D -->|"Approved within 60s"| E D -->|"Rejected or Timeout"| F["Vetoed Cancel Order"]
The Code: Implementing the Guardrail and Veto Engine
Below is a complete, self-contained Python implementation of our guardrail and human veto verification service. It uses FastAPI for the asynchronous receiver, Pydantic (v2) for structural verification, and an in-memory execution pipeline to handle the approval state machine.
import logging
import uuid
from typing import Dict, Literal, Optional, Tuple
from pydantic import BaseModel, Field, field_validator, model_validator
from fastapi import FastAPI, HTTPException, BackgroundTasks, status
# Setup structured logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("GuardrailEngine")
# — Schemas —
class TradeIntent(BaseModel):
intent_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
symbol: str
action: Literal["BUY", "SELL"]
price: float = Field(gt=0.0)
quantity: int = Field(gt=0)
strategy_name: str
confidence_score: float = Field(ge=0.0, le=1.0)
reasoning: str
@field_validator("symbol")
@classmethod
def sanitize_symbol(cls, value: str) -> str:
clean = value.strip().upper()
if not clean.isalnum():
raise ValueError("Symbol must be alphanumeric.")
return clean
# Hard-coded risk configurations for our portfolio
PORTFOLIO_LIMITS = {
"MAX_SINGLE_ORDER_VALUE": 100000.0, # $100,000 USD
"BLACKLISTED_SYMBOLS": {"GME", "AMC", "DOGE"},
"AUTO_EXECUTE_VALUE_LIMIT": 15000.0, # Trades under $15k bypass human veto if confidence is high
"MIN_CONFIDENCE_THRESHOLD": 0.70
}
# — State Store for Asynchronous Human Approvals —
class PendingVetoStore:
def __init__(self):
# Maps intent_id -> (TradeIntent, asyncio.Event, str status)
self._store: Dict[str, Tuple[TradeIntent, asyncio.Event, str]] = {}
def register(self, intent: TradeIntent):
event = asyncio.Event()
self._store[intent.intent_id] = (intent, event, "PENDING")
def resolve(self, intent_id: str, decision: Literal["APPROVED", "REJECTED"]):
if intent_id in self._store:
intent, event, _ = self._store[intent_id]
self._store[intent_id] = (intent, event, decision)
event.set()
logger.info(f"Trade {intent_id} manually resolved to: {decision}")
async def wait_for_decision(self, intent_id: str, timeout: float) -> str:
if intent_id not in self._store:
return "REJECTED"
intent, event, status_str = self._store[intent_id]
try:
# Wait for either human resolution or timeout
await asyncio.wait_for(event.wait(), timeout=timeout)
_, _, final_status = self._store[intent_id]
return final_status
except asyncio.TimeoutError:
logger.warning(f"Veto timeout of {timeout}s reached for trade {intent_id}. Auto-rejecting.")
self._store[intent_id] = (intent, event, "TIMEOUT_REJECTED")
return "TIMEOUT_REJECTED"
finally:
# Cleanup state after resolution to prevent memory growth
if intent_id in self._store:
del self._store[intent_id]
veto_store = PendingVetoStore()
# — Guardrail Validation Engine —
class GuardrailEngine:
@staticmethod
def evaluate(intent: TradeIntent) -> Tuple[bool, str, Literal["EXECUTE", "REJECT", "VETO_REQUIRED"]]:
total_value = intent.price * intent.quantity
# Rule 1: Confidence score filter
if intent.confidence_score < PORTFOLIO_LIMITS["MIN_CONFIDENCE_THRESHOLD"]:
return False, f"Confidence score {intent.confidence_score} below minimum.", "REJECT"
# Rule 2: Instrument blacklist check
if intent.symbol in PORTFOLIO_LIMITS["BLACKLISTED_SYMBOLS"]:
return False, f"Symbol {intent.symbol} is blacklisted.", "REJECT"
# Rule 3: Absolute maximum single trade limit
if total_value > PORTFOLIO_LIMITS["MAX_SINGLE_ORDER_VALUE"]:
return False, f"Total order value ${total_value:,.2f} exceeds absolute limit of ${PORTFOLIO_LIMITS['MAX_SINGLE_ORDER_VALUE']:,.2f}.", "REJECT"
# Rule 4: Human veto escalation trigger
if total_value > PORTFOLIO_LIMITS["AUTO_EXECUTE_VALUE_LIMIT"]:
return True, f"Order value ${total_value:,.2f} requires human validation.", "VETO_REQUIRED"
# If it passes everything, it is clear to auto-execute
return True, "All automated checks passed.", "EXECUTE"
# — Web Service API —
app = FastAPI(title="LLM Trade Guardrail System")
async def mock_execution_pipeline(intent: TradeIntent):
"""Sends approved trades to execution venue."""
logger.info(f"⚡ [EXECUTION ENGINE] Submitting order to broker: {intent.action} {intent.quantity} {intent.symbol} @ {intent.price}")
# Real brokers or internal execution wrappers would execute trades here.
await asyncio.sleep(0.05)
@app.post("/verify-trade", status_code=status.HTTP_200_OK)
async def verify_trade(intent: TradeIntent, background_tasks: BackgroundTasks):
logger.info(f"Received trade intent: {intent.action} {intent.symbol} (Value: ${intent.price * intent.quantity:,.2f})")
passed, reason, route = GuardrailEngine.evaluate(intent)
if not passed and route == "REJECT":
logger.warning(f"❌ Trade REJECTED by automated risk controls: {reason}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"status": "REJECTED", "reason": reason}
)
if route == "EXECUTE":
logger.info("🟢 Trade AUTO-APPROVED by risk controls.")
background_tasks.add_task(mock_execution_pipeline, intent)
return {
"status": "APPROVED",
"intent_id": intent.intent_id,
"reason": "Passed risk verification and fell below veto threshold."
}
if route == "VETO_REQUIRED":
logger.info(f"⏳ Trade placed in ESCROW. Escaped trade limits. Requiring human sign-off: {reason}")
# Register in async storage before triggering Slack notifications
veto_store.register(intent)
# Non-blocking simulation of outbound notification (Slack, PagerDuty, etc.)
background_tasks.add_task(trigger_slack_notification, intent)
# Block this coroutine for up to 60 seconds awaiting human interaction
decision = await veto_store.wait_for_decision(intent.intent_id, timeout=60.0)
if decision == "APPROVED":
background_tasks.add_task(mock_execution_pipeline, intent)
return {
"status": "APPROVED",
"intent_id": intent.intent_id,
"reason": "Human veto override. Trade verified manually."
}
else:
logger.warning(f"❌ Trade {intent.intent_id} canceled. Final status: {decision}")
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"status": "REJECTED", "reason": f"Human veto action status: {decision}"}
)
async def trigger_slack_notification(intent: TradeIntent):
"""
Simulates sending an interactive Slack block message to the trading desk.
In production, you would post this JSON payload to the Slack chat.postMessage endpoint.
"""
payload = {
"text": "⚠️ *LLM TRADE INTENT REQUIRES VETO OVERRIDE*",
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Strategy:* {intent.strategy_name}\n"
f"*Action:* {intent.action} {intent.quantity} {intent.symbol} @ {intent.price}\n"
f"*Total Position Value:* ${intent.price * intent.quantity:,.2f}\n"
f"*Confidence Score:* {intent.confidence_score * 100:.1f}%\n"
f"*Reasoning:* _\"{intent.reasoning}\"_"
}
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "Approve ✅"},
"style": "primary",
"value": intent.intent_id,
"action_id": "approve_trade"
},
{
"type": "button",
"text": {"type": "plain_text", "text": "Reject ❌"},
"style": "danger",
"value": intent.intent_id,
"action_id": "reject_trade"
}
]
}
]
}
# Log simulated outbound web hook
logger.info(f"[OUTBOUND NOTIFICATION] Posted veto payload to trading desk for trade {intent.intent_id}")
# — Interactive Hook for Slack Webhook Responses —
class SlackWebhookPayload(BaseModel):
intent_id: str
action: Literal["approve", "reject"]
verifier_id: str
@app.post("/slack/interactive-veto", status_code=status.HTTP_200_OK)
async def handle_slack_veto(payload: SlackWebhookPayload):
"""
Webhook endpoint hit when an analyst clicks "Approve" or "Reject" in Slack.
"""
decision = "APPROVED" if payload.action == "approve" else "REJECTED"
logger.info(f"Analyst {payload.verifier_id} clicked {decision} on Slack.")
veto_store.resolve(payload.intent_id, decision)
return {"status": "success", "message": f"Resolution logged as {decision}."}
Results and Metrics from Production
Deploying this multi-tiered architecture shifted our operations from manual supervision to automated processing with exceptions.
The metrics show the practical trade-offs of using this system:
- P99 Guardrail Overhead: The latency introduced by validation checks was small. The P99 latency of our deterministic python checks was 4.2 milliseconds.
- Analyst Veto Reaction Time: During active trading hours, the median (P50) reaction time for an analyst to review a Slack notification and click “Approve” or “Reject” was 14.8 seconds. The P90 response time was 42.1 seconds.
- Veto Timeout Protection: In 100% of cases where an analyst was away from their desk and missed the 60-second window, the system safely cancelled the order, preventing state leaks.
- Capital Protected: Over a 60-day trial period, the guardrail intercepted 3 separate invalid trades triggered by LLM parsing errors. One of these was a structural formatting error that would have caused a position limit violation.
Technical Lessons Learned
State Machine Leakage and Timeouts
When you build asynchronous human-in-the-loop steps, you introduce state management problems. If a trade intent is held in escrow waiting for approval, the market is moving. A price that looked good when the LLM made its decision may be highly unfavorable 60 seconds later.
Our system solves this by using a hard timeout (TTL) on the veto. However, our next system iteration will check market drift. If a human takes 45 seconds to approve a trade, but the underlying stock has moved more than 0.5% in that window, the execution engine voids the trade anyway. The check must be applied at the time of execution, not just at decision time.
Structured Output JSON Schema Is Not Enough
We tried to enforce correct output formats from our LLMs using tool calling (response_format={"type": "json_object"}). While this helps with structural formatting, it does not prevent logic and semantic errors. An LLM can easily generate valid JSON that complies with a schema but contains incorrect values (e.g., buying 100000 contracts instead of 100 because of unit differences). Structured outputs do not replace an independent evaluation engine.
The Analyst Fatigue Trap
When your automated trading system sends Slack alerts, analysts can quickly develop alert fatigue. If the agent triggers vetoes on 40 trades a day, humans begin clicking “Approve” automatically without checking the reasoning.
We addressed this by keeping our veto threshold high. Only trades representing significant risk (such as positions exceeding 2% of portfolio value or assets with low liquidity) route to the Slack channel. If an agent triggers too many vetoes, it indicates that the system’s operational parameters are too wide and need refinement.