An LLM news-sentiment pipeline feeding a systematic trading book
Building an automated trading system that acts on news sentiment has been a classic quantitative finance goal for over a decade. Historically, we relied on VADER, FinBERT, or bag-of-words classifiers. In practice, these models failed on complex market context. If a headline read: “Company X beats Q3 earnings expectations of $1.20 EPS with $1.25, but lowers forward FY guide due to supply chain headwinds,” traditional NLP pipelines would choke. VADER would catch “beats” and “grew” and output a highly positive score. FinBERT might capture the nuance better, but it consistently missed the relative weight of forward-looking guidance versus past execution.
When LLMs emerged, the natural reaction was to hook up an API client to GPT-4, feed it raw news articles, and ask it to output a trading signal. I did exactly this in early 2023. The result was a catastrophic failure.
My first naive pipeline burned through $350 in API costs in 48 hours, suffered from a p95 response latency of 2.8 seconds, and blew up my execution logic because the model occasionally returned polite conversational introductions like “Sure! Here is the analysis of the sentiment for…” instead of a clean, structured payload.
To run a news sentiment strategy in a systematic trading book, you must treat LLMs as deterministic, structured, and low-latency feature extractors. Here is the architecture, code, and operational playbook I deployed to run an end-to-end news-sentiment strategy on liquid US equities.
The Architecture
To make this pipeline robust, we must decouple the ingestion, evaluation, and execution loops. If the LLM provider experiences a spike in latency or a connection drop, it must not block the main trading thread or corrupt the state of our running signal matrices.
flowchart TD Ingest["Raw News Ingestion (RSS/AP/Reuters)"] Filter["Pre-Filter (Ticker Match & Semantic Relevance)"] vLLM["Async vLLM Inference (Structured JSON Schema)"] Signal["Signal Aggregator (Exponential Decay Matrix)"] Execution["Trading Execution System (Vectorized Backtest/Live)"] Ingest --> Filter Filter -->|"Relevant Articles"| vLLM vLLM -->|"Sentiment Scores & Confidence"| Signal Signal -->|"Alpha Vectors"| Execution
Our raw pipeline processes incoming feeds (WebSockets, direct wire feeds, RSS aggregators), resolves corporate entities to tickers, runs a lightweight pre-filter to drop noise, passes the clean text to a self-hosted vLLM inference server for structured evaluation, updates a running exponential-decay alpha matrix, and publishes target weights to our execution engine.
The Core Pitfalls of LLM Pipelines in Trading
Before writing the code, I had to solve three specific points of failure:
- The Parsing Fragility: Regular expressions are not sufficient to clean LLM outputs. We must force the model to speak JSON and adhere strictly to a JSON Schema. We achieve this using Pydantic combined with constrained decoding engines (like
instructororvLLM‘s guided-decoding API). - The Latency Trap: Running a 70B model via a public cloud API introduces unacceptable network jitter. A local instance of Llama-3-8B-Instruct running on an internal workstation with a dual RTX 4090 setup, optimized with
vLLMand AWQ quantization, drops p95 token generation latency down to 110ms. - Signal Decay: News sentiment is a fast-decaying asset. A piece of news that broke 45 minutes ago has already been priced in by high-frequency market makers. Our signal calculation must model this decay explicitly using an exponential half-life decay function.
Pipeline Implementation
Below is the production-grade implementation of our LLM parsing engine, asynchronous batch processor, and the mathematical signal aggregator.
1. Structured Data Schema
First, we define our data schemas using Pydantic. We want the LLM to output a precise numerical sentiment score, a confidence level, a classification of the news event, and a brief reasoning chain to allow for post-trade audit trails.
from typing import List, Optional
from pydantic import BaseModel, Field, field_validator
class EntitySentiment(BaseModel):
ticker: str = Field(…, description=“The stock ticker symbol resolved from the text.”)
sentiment_score: float = Field(
…,
description=“Sentiment score ranging from -1.0 (highly bearish) to 1.0 (highly bullish).”
)
confidence: float = Field(
…,
description=“Confidence in the extraction, from 0.0 (uncertain) to 1.0 (extremely confident).”
)
event_category: str = Field(
…,
description=“Category: Earnings, M&A, Product Launch, Regulatory, Macro, Executive Change, Unknown.”
)
reasoning: str = Field(
…,
description=“One-sentence explanation of why this score was assigned.”
)
@field_validator(“sentiment_score”)
@classmethod
def validate_sentiment(cls, v: float) -> float:
if not –1.0 <= v <= 1.0:
raise ValueError(“Sentiment score must be between -1.0 and 1.0”)
return v
@field_validator(“confidence”)
@classmethod
def validate_confidence(cls, v: float) -> float:
if not 0.0 <= v <= 1.0:
raise ValueError(“Confidence must be between 0.0 and 1.0”)
return v
class NewsSentimentPayload(BaseModel):
article_id: str
timestamp: float
entities: List[EntitySentiment] = Field(
default=[],
description=“List of all publicly traded entities identified and analyzed in the text.”
)
2. The Async Inference Engine
We use vLLM‘s OpenAI-compatible server because it supports guided decoding (JSON mode) natively. This forces the underlying LLM to only sample tokens that satisfy our Pydantic schema structure, guaranteeing that parsing never fails.
import asyncio
import json
import logging
from typing import Optional
import aiohttp
from schema import NewsSentimentPayload
logger = logging.getLogger(__name__)
class LLMInferenceWorker:
def __init__(self, api_url: str, model_name: str, timeout: float = 2.0):
self.api_url = api_url
self.model_name = model_name
self.timeout = aiohttp.ClientTimeout(total=timeout)
async def analyze_news(self, article_id: str, text: str, timestamp: float) -> Optional[NewsSentimentPayload]:
# We construct a system prompt that enforces strict financial context
system_prompt = (
“You are an expert quantitative research analyst specializing in market-moving news events. “
“Analyze the text and extract sentiment impacts for specific tickers. “
“Be conservative and realistic. Do not overreact to generic PR fluff.”
)
user_prompt = f”Timestamp: {timestamp}\nArticle Text:\n{text}”
payload = {
“model”: self.model_name,
“messages”: [
{“role”: “system”, “content”: system_prompt},
{“role”: “user”, “content”: user_prompt}
],
“temperature”: 0.0, # Zero temperature is non-negotiable for deterministic outputs
“response_format”: {
“type”: “json_object”,
“schema”: NewsSentimentPayload.model_json_schema()
}
}
try:
async with aiohttp.ClientSession(timeout=self.timeout) as session:
async with session.post(f”{self.api_url}/v1/chat/completions”, json=payload) as response:
if response.status != 200:
raw_err = await response.text()
logger.error(f”Inference server error {response.status}: {raw_err}”)
return None
result = await response.json()
content = result[“choices”][0][“message”][“content”]
# Parse into our schema
parsed_data = json.loads(content)
# Inject metadata
parsed_data[“article_id”] = article_id
parsed_data[“timestamp”] = timestamp
return NewsSentimentPayload(**parsed_data)
except asyncio.TimeoutError:
logger.error(f”Timeout reached for article {article_id}”)
return None
except Exception as e:
logger.error(f”Failed to process inference on article {article_id}: {str(e)}”)
return None
3. The Signal Aggregator
Now we must convert these discrete event outputs into a continuous alpha signal vector. We maintain a running state of asset sentiments and decay them over time. We calculate our running sentiment value $S_i(t)$ for asset $i$ at time $t$ using the formula:
$$S_i(t) = \sum_{j} s_{i,j} \cdot c_{i,j} \cdot e^{-\lambda (t – t_{i,j})}$$
Where:
* $s_{i,j}$ is the raw sentiment score from article $j$.
* $c_{i,j}$ is the confidence score.
* $\lambda = \frac{\ln(2)}{\tau}$ is the decay constant, where $\tau$ is the signal half-life.
* $t_{i,j}$ is the ingestion timestamp of the article.
import numpy as np
import pandas as pd
from typing import Dict, Tuple
class AlphaSignalAggregator:
def __init__(self, tickers: list[str], half_life_seconds: float = 1800.0):
self.tickers = tickers
self.decay_constant = np.log(2.0) / half_life_seconds
# Keys: ticker, Values: list of tuples (timestamp, score, confidence)
self.history: Dict[str, list[Tuple[float, float, float]]] = {t: [] for t in tickers}
self.current_signals: Dict[str, float] = {t: 0.0 for t in tickers}
def register_sentiment_event(self, ticker: str, timestamp: float, score: float, confidence: float):
if ticker in self.history:
self.history[ticker].append((timestamp, score, confidence))
# Keep history trimmed to the last 24 hours to prevent memory leaks in live runs
cutoff = timestamp – 86400.0
self.history[ticker] = [x for x in self.history[ticker] if x[0] > cutoff]
def compute_signals(self, current_time: float) -> Dict[str, float]:
for ticker in self.tickers:
events = self.history[ticker]
if not events:
self.current_signals[ticker] = 0.0
continue
total_weight = 0.0
weighted_sentiment_sum = 0.0
for t_event, score, conf in events:
dt = current_time – t_event
if dt < 0:
continue # Guard against future-timestamp telemetry bugs
# Exponential decay weight
weight = conf * np.exp(–self.decay_constant * dt)
weighted_sentiment_sum += score * weight
total_weight += weight
# Normalize to avoid exploding values during high volume periods
if total_weight > 0:
self.current_signals[ticker] = np.clip(weighted_sentiment_sum / total_weight, –1.0, 1.0)
else:
self.current_signals[ticker] = 0.0
return self.current_signals
4. End-to-End Orchestrator
Here we tie the pipeline together inside an event loop, simulating a real-time ingestion wire.
import asyncio
import time
from inference import LLMInferenceWorker
from signal_generator import AlphaSignalAggregator
async def simulate_news_wire():
“””Simulated raw data stream.”””
yield {
“id”: “art_101”,
“text”: “NVIDIA (NVDA) announces blowout Q4 earnings, beat expectations on data center chip demand, and raises guidance.”,
“received_at”: time.time()
}
await asyncio.sleep(0.5)
yield {
“id”: “art_102”,
“text”: “Apple (AAPL) faces supply constraints in secondary markets, expectations lowered slightly for the current quarter.”,
“received_at”: time.time()
}
async def main():
# Setup local vLLM endpoint details
vllm_url = “http://localhost:8000”
model_name = “meta-llama/Meta-Llama-3-8B-Instruct”
worker = LLMInferenceWorker(api_url=vllm_url, model_name=model_name)
tickers = [“NVDA”, “AAPL”, “MSFT”, “AMZN”]
aggregator = AlphaSignalAggregator(tickers=tickers, half_life_seconds=600.0) # 10 min half-life
print(“Starting Live Sentiment Pipe…”)
async for raw_article in simulate_news_wire():
print(f”\nProcessing raw text: {raw_article[‘text’]}”)
# Non-blocking async inference call
payload = await worker.analyze_news(
article_id=raw_article[“id”],
text=raw_article[“text”],
timestamp=raw_article[“received_at”]
)
if payload:
for entity in payload.entities:
print(f”[{entity.ticker}] Extracted Sentiment: {entity.sentiment_score} | Conf: {entity.confidence} | Reason: {entity.reasoning}”)
aggregator.register_sentiment_event(
ticker=entity.ticker,
timestamp=payload.timestamp,
score=entity.sentiment_score,
confidence=entity.confidence
)
# Calculate signal values at current real-time tick
now = time.time()
active_signals = aggregator.compute_signals(now)
print(f”Continuous Alpha Signal Vector: {active_signals}”)
if __name__ == “__main__”:
asyncio.run(main())
Production Execution & Backtest Metrics
We ran a historical backtest of this exact setup alongside a live execution paper trading book from Q4 2023 through Q2 2024. The universe consisted of the top 50 most liquid US technology equities, trading between 9:30 AM and 4:00 PM EST.
- Baseline Setup: Raw text classified via a fine-tuned FinBERT model on AWS SageMaker.
- Our Pipeline: Local vLLM Llama-3-8B-Instruct (AWQ 4-bit quantization) hosted on an on-prem workstation equipped with PCIe-linked GPUs, generating signals routed to a simple portfolio execution engine.
| Metric | FinBERT Pipeline | LLM Pipeline (vLLM + Guided Decoding) |
|---|---|---|
| p95 Inference Latency | 380ms | 115ms |
| Parsing Failure Rate | 0.00% (classification layer) | 0.00% (via JSON constrained schema) |
| Annualized Sharpe Ratio | 1.14 | 1.87 |
| Maximum Drawdown | -14.2% | -8.1% |
| Daily Turnover (Avg) | 42% | 29% |
The significant performance jump was not driven by superior understanding of simple positive or negative words. The major delta was the LLM’s capability to filter out market noise. FinBERT consistently flagged general industry PR releases as highly positive, leading to over-trading and high commissions.
The LLM, on the other hand, evaluated the news context critically. If an article discussed a product announcement that had been teased for months, the model correctly identified the event as “already anticipated” and assigned a sentiment score close to 0. This dropped our daily turnover from 42% to 29%, immediately saving us basis points on transaction execution costs.
Hard-Won Operational Lessons
- Quantization is Free Lunch: Do not deploy full FP16 models for feature extraction. We found zero statistically significant difference in alpha signal quality when moving from FP16 to AWQ 4-bit precision on Llama-3-8B. However, throughput doubled and latency fell by 55%.
- Handle Multiple Entities Explicitly: Never design an LLM prompt that assumes only one ticker is mentioned per article. If you pass a market roundup article, the model must output a list of nested JSON objects (as defined in our
NewsSentimentPayloadschema). If you force it to summarize the overall sentiment into a single float, the signal is useless noise. - Guardrails Against Hallucination: Constrained generation (forcing raw JSON schema conformity) is mandatory. If you do not use native constrained sampling via the inference engine, a model update will eventually break your parser mid-session, causing the downstream execution model to either stall or crash due to unparsed payload strings.