Backtesting an LLM trading agent built with LangGraph

Backtesting an LLM trading agent requires more than replaying prompts against historical prices. You need a harness that simulates execution reality — slippage, fees, latency, and the path-dependence of multi-step decisions — while preventing lookahead bias at every layer. This guide walks through building that harness with LangGraph, from state design through walk-forward validation.

Step 1: Define the agent state with explicit temporal boundaries

LangGraph’s StateGraph forces you to be precise about what the agent knows and when. For backtesting, the state must carry a current_bar index and a frozen snapshot of market data up to that point — nothing beyond it.

# state.py
from typing import TypedDict, List, Optional, Literal
from dataclasses import dataclass, field
from datetime import datetime
import pandas as pd

@dataclass(frozen=True)
class Bar:
    timestamp: datetime
    open: float
    high: float
    low: float
    close: float
    volume: float

class AgentState(TypedDict):
    # Immutable market context — set once per step by the harness
    bars: List[Bar]                    # All bars up to current_bar (inclusive)
    current_bar: int                   # Index into bars
    symbol: str
    
    # Agent's evolving state
    position: float                    # Current position size (positive=long, negative=short)
    cash: float                        # Available cash
    equity: float                      # Total portfolio value
    trade_log: List[dict]              # Executed trades with fill prices
    
    # LLM interaction
    messages: List[dict]               # Conversation history
    last_signal: Optional[Literal["long", "short", "flat", "hold"]]
    reasoning: str                     # LLM's explanation for audit trail
    
    # Risk controls (set by policy nodes, not LLM)
    max_position_pct: float            # e.g., 0.10 = 10% of equity per trade
    stop_loss_pct: Optional[float]     # e.g., 0.02 = 2% stop

Verify: Instantiate AgentState with sample bars and confirm bars[current_bar].timestamp matches your expected simulation time. No future bars should be accessible.

Step 2: Build the data feeder node — the only source of market truth

The data feeder node advances current_bar and injects the new bar into state. It never calls the LLM. This separation prevents the model from “peeking” at future data through prompt construction.

# nodes/data_feeder.py
from langgraph.graph import StateGraph
from state import AgentState, Bar
import pandas as pd

def load_bars(symbol: str, start: str, end: str) -> List[Bar]:
    # Replace with your data source (Parquet, TimescaleDB, etc.)
    df = pd.read_parquet(f"data/{symbol}.parquet")
    df = df[(df.index >= start) & (df.index <= end)]
    return [
        Bar(timestamp=ts, open=row.open, high=row.high, 
            low=row.low, close=row.close, volume=row.volume)
        for ts, row in df.iterrows()
    ]

def data_feeder_node(state: AgentState) -> AgentState:
    bars = state["bars"]
    idx = state["current_bar"]
    
    if idx >= len(bars) - 1:
        return {**state, "is_done": True}
    
    next_idx = idx + 1
    current_bar = bars[next_idx]
    
    # Inject only the new bar's OHLCV into messages for the LLM
    bar_summary = (
        f"Bar {next_idx}/{len(bars)-1} | {current_bar.timestamp.isoformat()} | "
        f"O:{current_bar.open:.4f} H:{current_bar.high:.4f} "
        f"L:{current_bar.low:.4f} C:{current_bar.close:.4f} V:{current_bar.volume:.0f}"
    )
    
    return {
        **state,
        "current_bar": next_idx,
        "messages": state["messages"] + [{"role": "system", "content": bar_summary}],
        "is_done": False
    }

Verify: Run the feeder in a loop on a 10-bar dataset. Confirm current_bar increments 0→9 and is_done flips true only after the last bar.

Step 3: Implement the LLM decision node with structured output

Use a Pydantic model to force the LLM into a parseable schema. This eliminates regex parsing failures during backtest runs. The prompt receives only historical bars up to current_bar — enforced by the feeder node.

# nodes/llm_decision.py
from pydantic import BaseModel, Field
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from state import AgentState

class TradingSignal(BaseModel):
    action: Literal["long", "short", "flat", "hold"] = Field(
        description="Target position direction"
    )
    confidence: float = Field(ge=0.0, le=1.0, description="Conviction 0-1")
    reasoning: str = Field(description="Concise rationale referencing specific price levels")
    size_pct: float = Field(ge=0.0, le=1.0, description="Position size as fraction of equity")

SYSTEM_PROMPT = """You are a systematic futures trader. You receive sequential price bars.
Decide your target position for the NEXT bar. You cannot trade the current bar's close.
Output ONLY the structured signal. No markdown."""

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).with_structured_output(TradingSignal)

def llm_decision_node(state: AgentState) -> AgentState:
    # Build context: last N bars + position + equity
    lookback = 20
    bars = state["bars"]
    idx = state["current_bar"]
    recent = bars[max(0, idx-lookback):idx+1]
    
    context = "\n".join([
        f"{b.timestamp:%H:%M} O:{b.open:.2f} H:{b.high:.2f} L:{b.low:.2f} C:{b.close:.2f} V:{b.volume:.0f}"
        for b in recent
    ])
    
    prompt = f"""Current position: {state['position']:.4f} | Equity: ${state['equity']:,.0f} | Cash: ${state['cash']:,.0f}
Max position: {state['max_position_pct']*100:.0f}% of equity

Recent bars (most recent last):
{context}

What is your target position for the NEXT bar?"""
    
    messages = [
        SystemMessage(content=SYSTEM_PROMPT),
        HumanMessage(content=prompt)
    ]
    
    signal: TradingSignal = llm.invoke(messages)
    
    return {
        **state,
        "last_signal": signal.action,
        "reasoning": signal.reasoning,
        "messages": state["messages"] + messages + [{"role": "assistant", "content": signal.model_dump_json()}]
    }

Verify: Run the node with a frozen state snapshot. Confirm the output parses to TradingSignal without exceptions. Log the raw response for 100 consecutive calls — zero parse failures is the target.

Step 4: Build the execution simulator with realistic slippage and fees

This node translates the LLM’s target position into fills. It uses the next bar’s open (or a VWAP model) as the reference price, applies slippage, and updates cash/equity. Never fill at the bar the LLM just saw — that’s lookahead bias.

# nodes/executor.py
from state import AgentState
from dataclasses import dataclass

@dataclass
class Fill:
    timestamp: str
    symbol: str
    side: str
    qty: float
    price: float
    commission: float
    slippage: float

def executor_node(state: AgentState) -> AgentState:
    if state.get("is_done"):
        return state
    
    bars = state["bars"]
    idx = state["current_bar"]
    current_bar = bars[idx]
    next_bar = bars[idx + 1] if idx + 1 < len(bars) else None
    
    if not next_bar:
        return state
    
    target_action = state["last_signal"]
    current_pos = state["position"]
    equity = state["equity"]
    max_pos_value = equity * state["max_position_pct"]
    
    # Map signal to target position size (simplified: full size or flat)
    target_pos = 0.0
    if target_action == "long":
        target_pos = max_pos_value / next_bar.open
    elif target_action == "short":
        target_pos = -max_pos_value / next_bar.open
    # "flat" and "hold" keep target_pos = 0
    
    delta = target_pos - current_pos
    
    if abs(delta) < 1e-8:
        return state  # No trade
    
    # Execution model: next bar open + slippage
    # Slippage: 1 bps + 0.1% of spread for market orders
    spread = next_bar.high - next_bar.low
    slippage_bps = 1.0 + 0.1 * (spread / next_bar.open) * 10000
    slippage = next_bar.open * slippage_bps / 10000
    
    fill_price = next_bar.open + (slippage if delta > 0 else -slippage)
    commission = abs(delta) * fill_price * 0.0002  # 2 bps commission
    
    fill = Fill(
        timestamp=next_bar.timestamp.isoformat(),
        symbol=state["symbol"],
        side="buy" if delta > 0 else "sell",
        qty=abs(delta),
        price=fill_price,
        commission=commission,
        slippage=slippage
    )
    
    new_cash = state["cash"] - delta * fill_price - commission
    new_position = current_pos + delta
    new_equity = new_cash + new_position * next_bar.close
    
    trade_log = state["trade_log"] + [{
        "timestamp": fill.timestamp,
        "side": fill.side,
        "qty": fill.qty,
        "price": fill.price,
        "commission": fill.commission,
        "slippage": fill.slippage,
        "reasoning": state["reasoning"]
    }]
    
    return {
        **state,
        "position": new_position,
        "cash": new_cash,
        "equity": new_equity,
        "trade_log": trade_log
    }

Verify: Feed a known signal sequence (long → flat → short) through the executor with synthetic bars. Manually compute expected fills and compare cash/equity at each step. Confirm slippage increases with wider spreads.

Step 5: Add risk policy nodes that the LLM cannot override

Risk controls belong in deterministic nodes, not the prompt. Add a stop-loss node that forces flat when equity drawdown exceeds a threshold, and a position-limit node that clamps size.

# nodes/risk.py
from state import AgentState

def stop_loss_node(state: AgentState) -> AgentState:
    if not state.get("stop_loss_pct"):
        return state
    
    bars = state["bars"]
    idx = state["current_bar"]
    current_price = bars[idx].close
    
    if state["position"] > 0:
        entry_price = state["cash"] / (state["equity"] - state["cash"]) * current_price  # simplified
        # Better: track entry price in trade_log
        drawdown = (current_price - entry_price) / entry_price
        if drawdown < -state["stop_loss_pct"]:
            return {**state, "last_signal": "flat", "reasoning": "Stop loss triggered"}
    
    elif state["position"] < 0:
        # Symmetric logic for shorts
        pass
    
    return state

def position_limit_node(state: AgentState) -> AgentState:
    # Clamp position to max_position_pct (defense against LLM hallucination)
    max_pos_value = state["equity"] * state["max_position_pct"]
    current_price = state["bars"][state["current_bar"]].close
    max_qty = max_pos_value / current_price
    
    if abs(state["position"]) > max_qty * 1.001:  # Small tolerance
        # Force reduce
        target_qty = max_qty if state["position"] > 0 else -max_qty
        delta = target_qty - state["position"]
        # Reuse executor logic or emit a forced trade
        pass
    
    return state

Verify: Inject a state with a 3% losing long position and a 2% stop loss. Confirm the node flips last_signal to "flat".

Step 6: Wire the graph with conditional edges for walk-forward steps

The graph cycles: data_feeder → risk_checks → llm_decision → executor → (loop). Use StateGraph.add_conditional_edges to stop at is_done.

# graph.py
from langgraph.graph import StateGraph, END
from state import AgentState
from nodes.data_feeder import data_feeder_node
from nodes.llm_decision import llm_decision_node
from nodes.executor import executor_node
from nodes.risk import stop_loss_node, position_limit_node

def build_backtest_graph() -> StateGraph:
    workflow = StateGraph(AgentState)
    
    workflow.add_node("data_feeder", data_feeder_node)
    workflow.add_node("stop_loss", stop_loss_node)
    workflow.add_node("position_limit", position_limit_node)
    workflow.add_node("llm_decision", llm_decision_node)
    workflow.add_node("executor", executor_node)
    
    workflow.set_entry_point("data_feeder")
    
    workflow.add_edge("data_feeder", "stop_loss")
    workflow.add_edge("stop_loss", "position_limit")
    workflow.add_edge("position_limit", "llm_decision")
    workflow.add_edge("llm_decision", "executor")
    
    def should_continue(state: AgentState) -> str:
        return END if state.get("is_done") else "data_feeder"
    
    workflow.add_conditional_edges("executor", should_continue)
    
    return workflow.compile()

# Run backtest
def run_backtest(symbol: str, start: str, end: str, initial_cash: float = 100_000):
    from state import AgentState
    from nodes.data_feeder import load_bars
    
    bars = load_bars(symbol, start, end)
    
    initial_state: AgentState = {
        "bars": bars,
        "current_bar": -1,  # Feeder advances to 0 on first call
        "symbol": symbol,
        "position": 0.0,
        "cash": initial_cash,
        "equity": initial_cash,
        "trade_log": [],
        "messages": [],
        "last_signal": None,
        "reasoning": "",
        "max_position_pct": 0.10,
        "stop_loss_pct": 0.02,
        "is_done": False
    }
    
    graph = build_backtest_graph()
    final_state = graph.invoke(initial_state, {"recursion_limit": len(bars) + 10})
    
    return final_state

Verify: Run on a 100-bar dataset. Confirm the graph terminates (doesn’t hit recursion limit) and final_state["trade_log"] length matches expected trade count.

Step 7: Compute performance metrics with statistical rigor

Raw returns are meaningless without risk adjustment. Compute Sharpe, Sortino, max drawdown, and — critically — the distribution of trade PnLs to detect strategy fragility.

# metrics.py
import numpy as np
import pandas as pd
from state import AgentState

def compute_metrics(final_state: AgentState, risk_free_rate: float = 0.02) -> dict:
    trade_log = final_state["trade_log"]
    if not trade_log:
        return {"error": "No trades executed"}
    
    # Reconstruct equity curve from trade log
    # (In production, store equity at each bar in state)
    equity_curve = []
    # ... reconstruct from trade_log timestamps and fills ...
    
    returns = pd.Series(equity_curve).pct_change().dropna()
    
    # Annualization assumes bar frequency (e.g., 1h bars = 24*252)
    periods_per_year = 24 * 252
    ann_return = (1 + returns.mean()) ** periods_per_year - 1
    ann_vol = returns.std() * np.sqrt(periods_per_year)
    sharpe = (ann_return - risk_free_rate) / ann_vol if ann_vol > 0 else 0
    
    # Sortino: only downside deviation
    downside = returns[returns < 0]
    sortino = (ann_return - risk_free_rate) / (downside.std() * np.sqrt(periods_per_year)) if len(downside) > 0 else 0
    
    # Max drawdown
    cum_returns = (1 + returns).cumprod()
    running_max = cum_returns.expanding().max()
    drawdown = (cum_returns - running_max) / running_max
    max_dd = drawdown.min()
    
    # Trade-level stats
    trade_pnls = [t.get("pnl", 0) for t in trade_log]  # Add pnl to executor
    win_rate = sum(1 for p in trade_pnls if p > 0) / len(trade_pnls)
    avg_win = np.mean([p for p in trade_pnls if p > 0]) if any(p > 0 for p in trade_pnls) else 0
    avg_loss = np.mean([p for p in trade_pnls if p < 0]) if any(p < 0 for p in trade_pnls) else 0
    profit_factor = abs(sum(p for p in trade_pnls if p > 0) / sum(p for p in trade_pnls if p < 0)) if any(p < 0 for p in trade_pnls) else float('inf')
    
    return {
        "total_return": cum_returns.iloc[-1] - 1,
        "annualized_return": ann_return,
        "annualized_vol": ann_vol,
        "sharpe": sharpe,
        "sortino": sortino,
        "max_drawdown": max_dd,
        "num_trades": len(trade_log),
        "win_rate": win_rate,
        "avg_win": avg_win,
        "avg_loss": avg_loss,
        "profit_factor": profit_factor,
        "equity_curve": equity_curve
    }

Verify: Generate a synthetic equity curve with known Sharpe (e.g., 1.0). Confirm the computed Sharpe matches within 0.01.

Step 8: Implement walk-forward validation to prevent overfitting

A single backtest is a single data point. Walk-forward splits data into sequential train/test windows, re-optimizing (or re-prompting) on each train window and testing on the subsequent test window. This mimics real deployment.

# walkforward.py
from typing import List, Tuple
from metrics import compute_metrics
from graph import run_backtest

def walk_forward_windows(
    symbol: str,
    start: str,
    end: str,
    train_months: int = 6,
    test_months: int = 1,
    step_months: int = 1
) -> List[dict]:
    import pandas as pd
    from dateutil.relativedelta import relativedelta
    
    start_dt = pd.Timestamp(start)
    end_dt = pd.Timestamp(end)
    
    windows = []
    current = start_dt
    
    while current + relativedelta(months=train_months + test_months) <= end_dt:
        train_start = current
        train_end = current + relativedelta(months=train_months)
        test_start = train_end
        test_end = test_start + relativedelta(months=test_months)
        
        # In practice: optimize prompt / few-shot examples on train window
        # For this example, we use the same prompt but evaluate on test
        result = run_backtest(symbol, 
                            train_start.strftime("%Y-%m-%d"),
                            test_end.strftime("%Y-%m-%d"))
        
        # Slice trade_log to test period only
        test_trades = [
            t for t in result["trade_log"]
            if pd.Timestamp(t["timestamp"]) >= test_start
        ]
        result["trade_log"] = test_trades
        
        metrics = compute_metrics(result)
        metrics["window"] = f"{train_start.date()} → {test_end.date()}"
        windows.append(metrics)
        
        current += relativedelta(months=step_months)
    
    return windows

def aggregate_walkforward(results: List[dict]) -> dict:
    # Aggregate across windows: median Sharpe, worst window, consistency
    sharpes = [r["sharpe"] for r in results if "sharpe" in r]
    returns = [r["total_return"] for r in results if "total_return" in r]
    
    return {
        "median_sharpe": np.median(sharpes),
        "mean_sharpe": np.mean(sharpes),
        "sharpe_std": np.std(sharpes),
        "positive_windows": sum(1 for s in sharpes if s > 0) / len(sharpes),
        "worst_window_sharpe": min(sharpes),
        "median_return": np.median(returns),
        "num_windows": len(results)
    }

Verify: Run walk-forward on 2 years of hourly data (6M train / 1M test / 1M step → ~18 windows). Confirm each window’s test trades fall strictly within its test period — no train-data contamination.

Step 9: Stress-test with Monte Carlo permutation

Walk-forward still assumes the test period’s market regime resembles the train period. Permutation testing shuffles trade PnLs to build a null distribution — if your strategy’s Sharpe isn’t in the top 5% of permuted Sharpes, it’s likely noise.

# monte_carlo.py
import numpy as np
from metrics import compute_metrics

def permutation_test(trade_pnls: List[float], n_permutations: int = 10000) -> dict:
    observed_sharpe = compute_sharpe_from_pnls(trade_pnls)
    permuted_sharpes = []
    
    for _ in range(n_permutations):
        shuffled = np.random.permutation(trade_pnls)
        permuted_sharpes.append(compute_sharpe_from_pnls(shuffled))
    
    p_value = sum(s >= observed_sharpe for s in permuted_sharpes) / n_permutations
    
    return {
        "observed_sharpe": observed_sharpe,
        "p_value": p_value,
        "percentile": np.percentile(permuted_sharpes, 95),
        "significant_at_5pct": p_value < 0.05
    }

def compute_sharpe_from_pnls(pnls: List[float], periods_per_year: int = 252) -> float:
    if len(pnls) < 2:
        return 0.0
    returns = np.array(pnls)  # Assumes PnL ≈ return on capital
    return (returns.mean() / returns.std()) * np.sqrt(periods_per_year) if returns.std() > 0 else 0.0

Verify: Feed a strategy with 50 trades, Sharpe 1.5. Run 10k permutations. Confirm p-value < 0.01. Then feed pure noise (random ±1% returns) — p-value should be ~0.5.

Common failure modes to check before trusting results

What this buys you

A backtest harness built this way — temporal state isolation, deterministic execution simulation, walk-forward validation, permutation testing — produces results that survive contact with reality. The LangGraph structure makes each component auditable and swappable: swap the LLM node for a rule-based baseline, swap the executor for a high-fidelity simulator, swap the risk nodes for portfolio-level controls. The graph topology stays the same; only the node implementations change.

When you eventually deploy, the same graph runs in production with a live data feeder and a real execution node. The backtest is the integration test.