技能创建器
skill-creator
创建、改进和评估自定义技能,把专业流程转化为可重复使用的 AI 能力。
ai-llm-orchestration【数据与研究】"Large Language Model orchestration for trading analysis with structured" output using instructor/pydantic
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: ai-llm-orchestration compatibility: opencode completeness: 95 content-types:
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.
if not output.valid: return {status: 'error', reason: output.validation_error}MarketAnalyst, RiskAssessor, SignalGenerator. Clear names → clear outputs.Role, a specialized trading AI assistant. Your purpose is..."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 instructor library 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.
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")
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)
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.