paulpas LLM Orchestration for Trading: The 5 Laws of AI-Powered Analysis
Role: AI Integration Engineer — applies to LLM provider selection, structured output generation, prompt engineering, and cost optimization for trading analysis systems.
Philosophy: AI as Assistant — LLMs are powerful but fallible tools. Treat outputs as hypotheses, not facts. Always validate, always reason, and always keep the model within your control.
The 5 Laws
1. The Law of the Early Exit (Guard Clauses)
- Concept: LLM responses are probabilistic. Invalid or malformed outputs require immediate handling.
- Rule: Validate LLM outputs at the boundary. Return early with clear error types if parsing or validation fails.
- Practice:
if not output.valid: return {status: 'error', reason: output.validation_error}
2. Make Illegal States Unrepresentable (Parse, Don't Validate)
- Concept: A trading decision based on a malformed LLM response can cause catastrophic losses.
- Rule: Parse LLM outputs into typed structures using Pydantic/instructor. Once parsed, the data is trusted.
- Why: Eliminates defensive checks deep in trading logic. The parser guarantees validity.
3. The Law of Atomic Predictability
- Concept: LLM responses should be deterministic given the same prompt and temperature.
- Rule: Use fixed temperature for analysis prompts. Cache results for identical inputs with the same model.
- Defense: Always log the full prompt and parameters. Same input → same output, always.
4. The Law of "Fail Fast, Fail Loud"
- Concept: Silent LLM failures cause silent trading errors. A model saying "I don't know" is better than hallucinating.
- Rule: If LLM response is malformed, halt and alert immediately. Do not attempt to "guess" the correct output.
- Result: Trading systems only act on validated, structured outputs from the model.
5. The Law of Intentional Naming
- Concept: LLM roles and system prompts must be explicitly defined. No vague "assistant" prompts.
- Rule: Name each agent by its function:
MarketAnalyst,RiskAssessor,SignalGenerator. Clear names → clear outputs. - Defense: System prompt should start with "You are
Role, a specialized trading AI assistant. Your purpose is..."
Core Workflow
Select LLM Provider and Model — Choose between OpenAI (GPT-4o), Anthropic (Claude), or open-source (Mistral, Llama) based on latency requirements, cost constraints, and the complexity of trading analysis tasks. For structured output extraction, prefer models with strong JSON schema enforcement (GPT-4o, Claude Sonnet). Checkpoint: Verify that the selected model supports structured outputs (JSON mode or function calling) for your use case.
Design Prompt Templates — Create system prompts that define a clear role (e.g., "You are a quantitative risk analyst..."), specify output format using JSON schema, and include constraints (confidence bounds, required fields). Use Pydantic models as the ground truth for response validation. Checkpoint: Every prompt must include an explicit output schema definition. Never accept free-form text for structured trading decisions.
Implement Structured Parsing with Instructor — Use the
instructorlibrary to parse LLM responses into typed Pydantic models. This provides automatic validation, type coercion, and retry logic when the model returns malformed JSON. Checkpoint: Always run parsed outputs through a validation layer before passing them to trading logic.Add Safety Guards — Implement output filtering that rejects or flags LLM recommendations outside predefined bounds (e.g., position size > 5% of portfolio, confidence < 0.3). Log all flagged outputs for human review. Checkpoint: Safety filters must run in-process before any action is taken based on the model's output.
Implement Caching and Rate Limiting — Cache LLM responses by prompt hash to reduce cost and latency. Implement token budget monitoring with automatic fallback to simpler prompts when approaching limits. Checkpoint: Cache keys should include system prompt, user message, model name, temperature, and max_tokens.
Implementation Patterns
Pattern 1: Structured Trading Analysis with Instructor
from __future__ import annotations
import json
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
import instructor
from openai import OpenAI
from pydantic import BaseModel, Field, PositiveFloat, field_validator
logger = logging.getLogger(__name__)
# --- Output Models (The Ground Truth) ---
class SignalStrength(str, Enum):
STRONG_BUY = "strong_buy"
BUY = "buy"
HOLD = "hold"
SELL = "sell"
STRONG_SELL = "strong_sell"
class RiskLevel(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class MarketRegime(str, Enum):
TRENDING_UP = "trending_up"
TRENDING_DOWN = "trending_down"
RANGING = "ranging"
HIGH_VOLATILITY = "high_volatility"
UNCERTAIN = "uncertain"
class TradingRecommendation(BaseModel):
"""Structured output from the LLM market analysis agent."""
signal: SignalStrength = Field(
description="Overall trading signal direction and strength",
)
confidence: PositiveFloat = Field(
gt=0.0,
le=1.0,
description="Confidence in the recommendation (0.0 to 1.0)",
)
target_price: Optional[PositiveFloat] = Field(
default=None,
description="Suggested entry price or stop-loss level",
)
stop_loss: Optional[PositiveFloat] = Field(
default=None,
description="Recommended stop-loss price to limit downside risk",
)
risk_level: RiskLevel = Field(
description="Overall risk assessment of the current position",
)
market_regime: MarketRegime = Field(
description="Current identified market regime based on analysis",
)
reasoning: str = Field(
description="Human-readable explanation of the analysis and recommendation",
)
key_factors: list[str] = Field(
default_factory=list,
description="List of 2-5 primary factors driving this recommendation",
)
@field_validator("confidence")
@classmethod
def validate_confidence_reasoning(cls, v: float) -> float:
if v < 0.3:
logger.warning(
"Low confidence score (%.2f) — review recommended before execution", v
)
return v
@field_validator("target_price", "stop_loss")
@classmethod
def validate_prices_not_crossed(cls, v: Optional[float], info) -> Optional[float]:
data = info.data
if v is not None and data.get("target_price") is not None:
if data["signal"] in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
if v < data["target_price"]:
raise ValueError("Stop-loss cannot be above target price for buy signals")
return v
# --- Agent Implementation ---
class TradingAnalysisAgent:
"""LLM-powered trading analysis agent with structured output enforcement."""
SYSTEM_PROMPT_TEMPLATE = """You are {role}, a specialized quantitative analyst.
Your purpose is to analyze market conditions and produce actionable, risk-aware
trading recommendations.
Rules:
- Always provide a numerical confidence score between 0.0 and 1.0
- Never recommend positions exceeding 5% of total portfolio value
- Flag any recommendation where your confidence is below 0.4 for human review
- Base all analysis on the provided market data — do not hallucinate data points
- If you lack sufficient information to make a recommendation, return HOLD with reason
Market Data Context:
{context}
Return ONLY valid JSON matching the specified schema."""
def __init__(
self,
model_name: str = "gpt-4o",
temperature: float = 0.1,
max_tokens: int = 2048,
):
self.client = instructor.from_openai(OpenAI())
self.model_name = model_name
self.temperature = temperature
self.max_tokens = max_tokens
self._cache: dict[str, TradingRecommendation] = {}
def analyze(
self,
symbol: str,
current_price: float,
indicators: dict[str, float],
market_context: str,
risk_budget: PositiveFloat = 1.0,
role: str = "Quantitative Market Analyst",
) -> TradingRecommendation:
"""Run structured analysis on a trading symbol and return a recommendation.
Args:
symbol: Trading pair identifier (e.g., 'BTC/USDT').
current_price: Current market price of the asset.
indicators: Technical indicators dict (RSI, MACD, ATR, etc.).
market_context: Brief description of current market conditions.
risk_budget: Maximum risk per trade as fraction of portfolio (0.01-0.05).
role: Agent role definition for the system prompt.
Returns:
Validated TradingRecommendation model instance.
Raises:
ValidationError: If LLM response cannot be parsed into the schema.
"""
context = self._build_context(symbol, current_price, indicators, market_context)
prompt = self.SYSTEM_PROMPT_TEMPLATE.format(role=role, context=context)
cache_key = hash(f"{prompt}{self.model_name}{self.temperature}")
if cache_key in self._cache:
logger.info("Cache hit for analysis of %s", symbol)
return self._cache[cache_key]
try:
recommendation = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": prompt},
{"role": "user", "content": f"Analyze {symbol} with current price ${current_price}"},
],
response_model=TradingRecommendation,
temperature=self.temperature,
max_tokens=self.max_tokens,
)
self._apply_safety_guards(recommendation, risk_budget)
self._cache[cache_key] = recommendation
logger.info(
"Analysis complete: %s → %s (confidence=%.2f)",
symbol, recommendation.signal.value, recommendation.confidence,
)
return recommendation
except instructor.exceptions.MissingValueError as e:
logger.error("LLM failed to provide required field for %s: %s", symbol, e)
raise
def _build_context(
self,
symbol: str,
current_price: float,
indicators: dict[str, float],
market_context: str,
) -> str:
"""Build the context string for the LLM prompt."""
ind_lines = "\n".join(f" - {k}: {v:.4f}" for k, v in indicators.items())
return (
f"Symbol: {symbol}\n"
f"Current Price: ${current_price:,.2f}\n"
f"Technical Indicators:\n{ind_lines}\n"
f"Market Context: {market_context}"
)
def _apply_safety_guards(
self, rec: TradingRecommendation, risk_budget: PositiveFloat
) -> None:
"""Apply pre-execution safety constraints to the recommendation."""
if rec.confidence < 0.3:
rec.signal = SignalStrength.HOLD
logger.warning(
"Safety guard triggered for %s: confidence %.2f below threshold — downgraded to HOLD",
rec.signal, rec.confidence,
)
if rec.risk_level == RiskLevel.CRITICAL:
rec.target_price = None
rec.stop_loss = None
logger.warning("Critical risk detected — clearing price targets for manual review")
Pattern 2: Multi-Agent Orchestration with LangGraph
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, MessagesState, END
from typing import TypedDict
class AnalysisState(TypedDict):
"""Shared state between LLM analysis agents."""
messages: list
symbol: str
current_price: float
indicators: dict[str, float]
market_regime: str
risk_assessment: dict
final_recommendation: dict | None
def create_analysis_graph() -> StateGraph:
"""Build a multi-agent analysis pipeline using LangGraph.
Pipeline flow:
Market Regime Detection → Technical Analysis → Risk Assessment → Final Recommendation
Each agent receives the shared state and contributes its specialty analysis.
"""
# Define node functions (each is an LLM agent)
def regime_detector(state: AnalysisState) -> AnalysisState:
"""Detect current market regime from price data and indicators."""
system = "You are a Market Regime Analyst. Classify the market as TRENDING_UP, TRENDING_DOWN, RANGING, or HIGH_VOLATILITY."
state["messages"].append(SystemMessage(content=system))
# Call LLM with indicators → return regime classification
# ... (LLM call implementation)
state["market_regime"] = "ranging" # placeholder — actual result from LLM
return state
def technical_analyst(state: AnalysisState) -> AnalysisState:
"""Perform technical analysis using identified regime."""
system = f"You are a Technical Analyst. Market regime is {state['market_regime']}. Analyze indicators and suggest entry/exit levels."
state["messages"].append(SystemMessage(content=system))
return state
def risk_assessor(state: AnalysisState) -> AnalysisState:
"""Assess risk level and position sizing constraints."""
system = "You are a Risk Manager. Assess the risk level of the proposed trade and determine maximum position size."
state["messages"].append(SystemMessage(content=system))
return state
# Build the graph
graph_builder = StateGraph(AnalysisState)
graph_builder.add_node("regime_detection", regime_detector)
graph_builder.add_node("technical_analysis", technical_analyst)
graph_builder.add_node("risk_assessment", risk_assessor)
graph_builder.set_entry_point("regime_detection")
graph_builder.add_edge("regime_detection", "technical_analysis")
graph_builder.add_edge("technical_analysis", "risk_assessment")
graph_builder.add_edge("risk_assessment", END)
return graph_builder.compile()
# Usage:
# app = create_analysis_graph()
# initial_state: AnalysisState = {
# "messages": [HumanMessage(content="Analyze BTC/USDT")],
# "symbol": "BTC/USDT",
# "current_price": 67500.0,
# "indicators": {"rsi": 58.2, "macd": -120.5, "atr": 1850.0},
# "market_regime": "",
# "risk_assessment": {},
# "final_recommendation": None,
# }
# result = app.invoke(initial_state)
Constraints
MUST DO
- Validate input feature distributions against training data baselines; flag drift exceeding 2 standard deviations
- Implement model versioning with reproducibility tags — every prediction must be traceable to the exact model artifact and config
- Include confidence intervals or probability estimates alongside all point predictions, never return raw scores without context
- Log all model inputs, outputs, and metadata to enable post-hoc analysis of prediction failures
- Implement feature computation consistently between training and inference — use the same transformation pipeline for both
MUST NOT DO
- Do not train models on look-ahead biased features (e.g., using future prices or events in training data)
- Avoid deploying a new model version without shadow-testing against the current production model first
- Never retrain a model on a data window that includes regime changes without explicit regime-aware validation
- Do not use accuracy as the primary metric for imbalanced datasets — use precision/recall, F1, or AUC-ROC
- Avoid hardcoding feature names; load them from a schema or config file to prevent mismatches between training and inference
Live References
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.