Custom Document Loaders for High-Velocity Financial Data in Multi-Agent LangGraph

The Batch Loader Fallacy in Financial RAG

LangChain’s document loader ecosystem is fundamentally batch-oriented. PyPDFLoader, CSVLoader, and UnstructuredLoader assume documents arrive as complete, static files. This assumption collapses in financial services where data arrives as high-velocity streams: Bloomberg terminal feeds, SEC EDGAR filings, broker-dealer trade confirmations, and real-time market microstructure data. In these environments, waiting for a "complete document" means missing the alpha window entirely. A 10-K filing is a discrete artifact; a stream of earnings revisions, insider transactions, and options flow is not. You need loaders that treat streaming data as first-class documents with temporal semantics, partial state management, and backpressure-aware ingestion—all integrated into a stateful multi-agent RAG system. This article demonstrates building custom streaming document loaders for a real-time institutional trading desk, integrated into a LangGraph multi-agent architecture with persistent memory and state.

Real-Time Use Case: Institutional Event-Driven Trading Desk

The Scenario

A quantitative hedge fund’s event-driven desk trades around corporate events. Analysts must synthesize signals from four concurrent high-velocity streams:

SEC EDGAR Stream: 8-K filings, Form 4 insider transactions (arrives within seconds of filing)
Earnings Revision Stream: Broker estimates updated in real-time via Refinitiv API
Options Flow Stream: Unusual block trades detected by proprietary scanner
News Wire Stream: PR Newswire/Business Wire filtered for covered universe

An analyst asks: "What's driving the sudden SPX put skew in NVDA? Cross-reference with any recent insider activity and estimate revisions."

The system must ingest, index, and reason over data arriving at 50-200 events/second while maintaining conversation state across multiple analyst queries.

Why Standard Loaders Fail

Requirement	Batch Loader Limitation	Streaming Loader Solution
Sub-second latency	Wait for file completion	Process events as they arrive
Partial document assembly	All-or-nothing loading	Incremental chunking with sequence tracking
Temporal ordering	File modification time only	Event-level timestamps with causal ordering
Backpressure handling	OOM on burst traffic	Bounded buffers with adaptive throttling
Deduplication across streams	Post-hoc dedup	Content-hash dedup at ingestion
State continuity	Stateless reload	Cursor-based resumption after restart
Schema evolution	Fixed schema per loader	Adaptive parsing with version detection

Architecture: Streaming Loaders Inside LangGraph

Implementation

Step 1: Define Streaming-Aware State

Standard RAG state assumes static documents. Streaming RAG requires cursors, sequence numbers, and buffer health metrics.

from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from datetime import datetime
import operator

class StreamCursor(TypedDict):
    """Resumable position marker for each data stream."""
    stream_id: str
    last_sequence: int
    last_timestamp: datetime
    checkpoint_hash: str  # For integrity verification

class IngestedEvent(TypedDict):
    """Normalized event envelope across all streams."""
    event_id: str
    stream_id: str
    sequence: int
    timestamp: datetime
    ticker: str
    event_type: str  # "8k_filing", "estimate_revision", "options_block", "news_wire"
    content: str
    metadata: Dict[str, Any]
    content_hash: str

class TradingDeskState(TypedDict):
    messages: Annotated[list, add_messages]
    
    # Resumable cursors for each stream
    stream_cursors: Dict[str, StreamCursor]
    
    # Recently ingested events (bounded window)
    ingested_events: Annotated[List[IngestedEvent], operator.add]
    
    # Buffer health metrics for backpressure
    buffer_health: Dict[str, Dict[str, Any]]
    
    # Cross-stream signal cache (avoids re-computation)
    signal_cache: Dict[str, Any]
    
    # Current analyst query
    current_query: str
    
    # Generated trade thesis
    trade_thesis: Optional[str]
    
    # Audit trail
    audit_trail: Annotated[List[Dict], operator.add]

Step 2: Base Streaming Loader Infrastructure

All custom loaders inherit from this base class that handles the hard parts: async iteration, backpressure, deduplication, and cursor management.

import asyncio
import hashlib
from abc import ABC, abstractmethod
from collections import deque
from typing import AsyncIterator, Set, Callable, Optional
from langchain_core.documents import Document

class AsyncEventBuffer:
    """
    Backpressure-aware bounded buffer for streaming events.
    Prevents OOM during burst traffic while maintaining ordering.
    """
    def __init__(self, max_size: int = 10000, flush_threshold: float = 0.8):
        self._buffer: deque = deque(maxlen=max_size)
        self._max_size = max_size
        self._flush_threshold = flush_threshold
        self._lock = asyncio.Lock()
        self._seen_hashes: Set[str] = set()
        self._dropped_count = 0
    
    @property
    def utilization(self) -> float:
        return len(self._buffer) / self._max_size
    
    @property
    def should_flush(self) -> bool:
        return self.utilization >= self._flush_threshold
    
    async def put(self, event: IngestedEvent) -> bool:
        """Add event with dedup. Returns False if dropped due to backpressure."""
        async with self._lock:
            # Content-hash deduplication across streams
            if event["content_hash"] in self._seen_hashes:
                return True  # Duplicate, silently skip
            
            if len(self._buffer) >= self._max_size:
                self._dropped_count += 1
                return False  # Backpressure: drop oldest
            
            self._buffer.append(event)
            self._seen_hashes.add(event["content_hash"])
            
            # Evict old hashes to prevent unbounded growth
            if len(self._seen_hashes) > self._max_size * 2:
                # Keep only hashes for events still in buffer
                active_hashes = {e["content_hash"] for e in self._buffer}
                self._seen_hashes = active_hashes
            
            return True
    
    async def drain(self) -> List[IngestedEvent]:
        """Atomically drain all buffered events."""
        async with self._lock:
            events = list(self._buffer)
            self._buffer.clear()
            return events
    
    def health_metrics(self) -> Dict[str, Any]:
        return {
            "utilization": self.utilization,
            "buffered": len(self._buffer),
            "dropped": self._dropped_count,
            "unique_hashes": len(self._seen_hashes)
        }


class BaseStreamingLoader(ABC):
    """
    Abstract base for all streaming financial data loaders.
    Handles async iteration, cursor persistence, and normalization.
    Subclasses implement only source-specific parsing.
    """
    
    def __init__(
        self,
        stream_id: str,
        buffer: AsyncEventBuffer,
        cursor_store: Any,  # Redis/Postgres cursor persistence
        vectorstore: Any,
        batch_size: int = 50,
        poll_interval_ms: int = 100
    ):
        self.stream_id = stream_id
        self.buffer = buffer
        self.cursor_store = cursor_store
        self.vectorstore = vectorstore
        self.batch_size = batch_size
        self.poll_interval_ms = poll_interval_ms
        self._running = False
    
    @abstractmethod
    async def _fetch_batch(self, cursor: StreamCursor) -> List[Dict]:
        """Source-specific: fetch raw events since cursor position."""
        ...
    
    @abstractmethod
    def _normalize_event(self, raw: Dict, sequence: int) -> IngestedEvent:
        """Source-specific: normalize raw event to common schema."""
        ...
    
    def _compute_content_hash(self, content: str, ticker: str, event_type: str) -> str:
        """Deterministic hash for cross-stream deduplication."""
        canonical = f"{ticker}|{event_type}|{content.strip()}"
        return hashlib.sha256(canonical.encode()).hexdigest()[:32]
    
    async def get_cursor(self) -> StreamCursor:
        """Load persisted cursor or initialize."""
        cursor = await self.cursor_store.get(f"cursor:{self.stream_id}")
        if cursor is None:
            return {
                "stream_id": self.stream_id,
                "last_sequence": 0,
                "last_timestamp": datetime.min,
                "checkpoint_hash": ""
            }
        return cursor
    
    async def save_cursor(self, cursor: StreamCursor):
        """Persist cursor for resumability."""
        await self.cursor_store.set(f"cursor:{self.stream_id}", cursor)
    
    async def ingest_loop(self):
        """Main ingestion loop with backpressure awareness."""
        self._running = True
        cursor = await self.get_cursor()
        
        while self._running:
            try:
                # Fetch new events from source
                raw_events = await self._fetch_batch(cursor)
                
                if not raw_events:
                    await asyncio.sleep(self.poll_interval_ms / 1000)
                    continue
                
                # Normalize and buffer
                accepted = 0
                for i, raw in enumerate(raw_events):
                    seq = cursor["last_sequence"] + i + 1
                    event = self._normalize_event(raw, seq)
                    
                    success = await self.buffer.put(event)
                    if success:
                        accepted += 1
                        cursor = {
                            "stream_id": self.stream_id,
                            "last_sequence": seq,
                            "last_timestamp": event["timestamp"],
                            "checkpoint_hash": event["content_hash"]
                        }
                
                # Flush to vectorstore when buffer is hot
                if self.buffer.should_flush:
                    events = await self.buffer.drain()
                    docs = [
                        Document(
                            page_content=e["content"],
                            metadata={
                                "event_id": e["event_id"],
                                "stream_id": e["stream_id"],
                                "ticker": e["ticker"],
                                "event_type": e["event_type"],
                                "timestamp": e["timestamp"].isoformat(),
                                "sequence": e["sequence"],
                                "content_hash": e["content_hash"]
                            }
                        )
                        for e in events
                    ]
                    await self.vectorstore.aadd_documents(docs)
                
                # Persist cursor after successful batch
                await self.save_cursor(cursor)
                
            except Exception as e:
                # Log but don't crash; retry on next cycle
                print(f"[{self.stream_id}] Ingestion error: {e}")
                await asyncio.sleep(1)
    
    async def stop(self):
        self._running = False

Step 3: Concrete Loader — SEC EDGAR Stream

import aiohttp
from lxml import etree

class EDGARStreamLoader(BaseStreamingLoader):
    """
    Streams SEC 8-K and Form 4 filings from EDGAR's Atom feed.
    Handles XML parsing, exhibit extraction, and rate limiting.
    """
    
    EDGAR_ATOM_URL = "https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&type=8-K&dateb=&owner=include&count=100&search_text=&action=getcompany&output=atom"
    RATE_LIMIT_DELAY = 0.1  # SEC allows 10 req/sec
    
    def __init__(self, buffer, cursor_store, vectorstore, **kwargs):
        super().__init__("edgar_stream", buffer, cursor_store, vectorstore, **kwargs)
        self._session: Optional[aiohttp.ClientSession] = None
        self._rate_limiter = asyncio.Semaphore(10)
    
    async def _get_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            headers = {"User-Agent": "TradingDesk RAG v1.0 [email protected]"}
            self._session = aiohttp.ClientSession(headers=headers)
        return self._session
    
    async def _fetch_batch(self, cursor: StreamCursor) -> List[Dict]:
        async with self._rate_limiter:
            session = await self._get_session()
            params = {
                "start": cursor["last_sequence"],
                "count": self.batch_size
            }
            async with session.get(self.EDGAR_ATOM_URL, params=params) as resp:
                resp.raise_for_status()
                xml_content = await resp.text()
        
        # Parse Atom feed entries
        root = etree.fromstring(xml_content.encode())
        nsmap = {"atom": "http://www.w3.org/2005/Atom"}
        entries = root.findall("atom:entry", nsmap)
        
        results = []
        for entry in entries:
            title = entry.findtext("atom:title", "", nsmap)
            link = entry.find("atom:link", nsmap).get("href")
            updated = entry.findtext("atom:updated", "", nsmap)
            
            # Extract CIK and form type from title
            # Format: "1000275 - 8-K - NVIDIA CORP"
            parts = title.split(" - ")
            cik = parts[0].strip() if len(parts) > 0 else ""
            form_type = parts[1].strip() if len(parts) > 1 else ""
            company = parts[2].strip() if len(parts) > 2 else ""
            
            results.append({
                "cik": cik,
                "form_type": form_type,
                "company": company,
                "filing_url": link,
                "filed_at": updated,
                "raw_xml": etree.tostring(entry, encoding="unicode")
            })
            
            await asyncio.sleep(self.RATE_LIMIT_DELAY)
        
        return results
    
    def _normalize_event(self, raw: Dict, sequence: int) -> IngestedEvent:
        content = f"SEC {raw['form_type']} Filing: {raw['company']} (CIK: {raw['cik']})\nFiled: {raw['filed_at']}\nURL: {raw['filing_url']}"
        
        # Map company name to ticker (in production, use a lookup service)
        ticker = self._resolve_ticker(raw["company"])
        
        return {
            "event_id": f"edgar_{raw['cik']}_{sequence}",
            "stream_id": "edgar_stream",
            "sequence": sequence,
            "timestamp": datetime.fromisoformat(raw["filed_at"].replace("Z", "+00:00")),
            "ticker": ticker,
            "event_type": "8k_filing" if "8-K" in raw["form_type"] else "form4",
            "content": content,
            "metadata": {
                "cik": raw["cik"],
                "form_type": raw["form_type"],
                "filing_url": raw["filing_url"]
            },
            "content_hash": self._compute_content_hash(content, ticker, raw["form_type"])
        }
    
    def _resolve_ticker(self, company_name: str) -> str:
        # Production: use SEC company tickers dataset or mapping service
        mappings = {"NVIDIA CORP": "NVDA", "APPLE INC": "AAPL", "TESLA INC": "TSLA"}
        return mappings.get(company_name.upper(), "UNKNOWN")
    
    async def close(self):
        if self._session and not self._session.closed:
            await self._session.close()

Step 4: Concrete Loader — Options Flow Stream

class OptionsFlowLoader(BaseStreamingLoader):
    """
    Streams unusual options block trades from proprietary scanner.
    Handles WebSocket connection, binary protocol decoding, and reconnection.
    Demonstrates non-HTTP streaming source integration.
    """
    
    def __init__(self, ws_url: str, api_key: str, buffer, cursor_store, vectorstore, **kwargs):
        super().__init__("options_flow", buffer, cursor_store, vectorstore, **kwargs)
        self.ws_url = ws_url
        self.api_key = api_key
        self._ws = None
    
    async def _fetch_batch(self, cursor: StreamCursor) -> List[Dict]:
        """WebSocket streams don't use cursors; we track by sequence."""
        import websockets
        
        if self._ws is None or self._ws.closed:
            self._ws = await websockets.connect(
                self.ws_url,
                extra_headers={"Authorization": f"Bearer {self.api_key}"}
            )
            # Resume from last sequence
            await self._ws.send(f'{{"resume_from": {cursor["last_sequence"]}}}')
        
        events = []
        try:
            for _ in range(self.batch_size):
                msg = await asyncio.wait_for(self._ws.recv(), timeout=1.0)
                data = json.loads(msg)
                events.append(data)
        except asyncio.TimeoutError:
            pass  # No more events in this window
        except websockets.ConnectionClosed:
            self._ws = None  # Will reconnect next cycle
        
        return events
    
    def _normalize_event(self, raw: Dict, sequence: int) -> IngestedEvent:
        side = raw.get("side", "unknown").upper()
        ticker = raw.get("symbol", "UNKNOWN")
        strike = raw.get("strike", 0)
        expiry = raw.get("expiry", "")
        premium = raw.get("premium", 0)
        volume = raw.get("volume", 0)
        oi = raw.get("open_interest", 0)
        
        content = (
            f"OPTIONS BLOCK: {ticker} {expiry} ${strike} {side}\n"
            f"Volume: {volume:,} | OI: {oi:,} | Premium: ${premium:,.0f}\n"
            f"Vol/OI Ratio: {volume/max(oi,1):.2f} | "
            f"Notional: ${premium * volume * 100:,.0f}"
        )
        
        event_type = "options_block_put" if side == "BUY" and raw.get("type") == "P" else "options_block_call"
        
        return {
            "event_id": f"opt_{ticker}_{sequence}",
            "stream_id": "options_flow",
            "sequence": sequence,
            "timestamp": datetime.fromtimestamp(raw.get("ts", 0)),
            "ticker": ticker,
            "event_type": event_type,
            "content": content,
            "metadata": {
                "strike": strike, "expiry": expiry, "side": side,
                "volume": volume, "open_interest": oi, "premium": premium
            },
            "content_hash": self._compute_content_hash(content, ticker, event_type)
        }

Step 5: Integrate Streaming Loaders into LangGraph Agents

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

# Initialize shared infrastructure
shared_buffer = AsyncEventBuffer(max_size=50000)
cursor_store = RedisCursorStore(redis_url="redis://localhost:6379")
vectorstore = PGVectorStore(connection_string="postgresql://...")

# Start streaming loaders as background tasks
loaders = [
    EDGARStreamLoader(shared_buffer, cursor_store, vectorstore, batch_size=20),
    OptionsFlowLoader("wss://flow.scanner.io/ws", API_KEY, shared_buffer, cursor_store, vectorstore),
    # EarningsStreamLoader(...), NewsWireLoader(...)
]

loader_tasks = [asyncio.create_task(loader.ingest_loop()) for loader in loaders]

# --- INGESTION VALIDATION NODE ---
async def ingestion_validation_node(state: TradingDeskState) -> dict:
    """Validates buffer health and routes stale/broken streams."""
    health = shared_buffer.health_metrics()
    
    alerts = []
    if health["utilization"] > 0.95:
        alerts.append("CRITICAL: Buffer near capacity, events being dropped")
    if health["dropped"] > 100:
        alerts.append(f"WARNING: {health['dropped']} events dropped since last check")
    
    # Check stream freshness
    for stream_id, cursor in state.get("stream_cursors", {}).items():
        age = (datetime.utcnow() - cursor["last_timestamp"]).total_seconds()
        if age > 300:  # 5 min staleness threshold
            alerts.append(f"STALE: {stream_id} last updated {age:.0f}s ago")
    
    return {
        "buffer_health": health,
        "audit_trail": [{"event": "health_check", "alerts": alerts, "metrics": health}]
    }

# --- CROSS-STREAM ANALYSIS NODE ---
analysis_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an event-driven trading analyst. 
    Synthesize signals from multiple concurrent financial streams.
    Always cite event_id and timestamp for every claim.
    Identify temporal causality: which event likely triggered another.
    Flag conflicting signals explicitly."""),
    ("human", "Query: {query}\n\nRecent Events:\n{events_context}\n\nBuffer Health: {health}")
])

async def cross_stream_analysis_node(state: TradingDeskState) -> dict:
    # Build temporally-ordered context from ingested events
    events_sorted = sorted(state["ingested_events"], key=lambda e: e["timestamp"], reverse=True)
    
    events_context = "\n\n".join(
        f"[{e['event_id']} | {e['timestamp'].isoformat()} | {e['stream_id']}]\n"
        f"{e['ticker']}: {e['content']}"
        for e in events_sorted[:30]  # Top 30 most recent
    )
    
    response = await analysis_prompt | llm | (lambda x: x.content)
    thesis = await response.ainvoke({
        "query": state["current_query"],
        "events_context": events_context,
        "health": json.dumps(state["buffer_health"])
    })
    
    return {"trade_thesis": thesis}

# --- BUILD GRAPH ---
workflow = StateGraph(TradingDeskState)
workflow.add_node("validate_ingestion", ingestion_validation_node)
workflow.add_node("analyze", cross_stream_analysis_node)

workflow.add_edge(START, "validate_ingestion")
workflow.add_conditional_edges(
    "validate_ingestion",
    lambda s: "analyze" if not any("CRITICAL" in a for a in s.get("audit_trail", [{}])[0].get("alerts", [])) else END,
    {"analyze": "analyze", END: END}
)
workflow.add_edge("analyze", END)

checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = workflow.compile(checkpointer=checkpointer)

Step 6: Execute with Persistent Memory

config = {"configurable": {"thread_id": "trading-desk-nvda-20240805"}}

result = await app.ainvoke({
    "current_query": "What's driving NVDA put skew? Cross-reference insider activity and estimate revisions.",
    "messages": [],
    "stream_cursors": {},
    "ingested_events": [],
    "buffer_health": {},
    "signal_cache": {},
    "trade_thesis": None,
    "audit_trail": []
}, config=config)

print(result["trade_thesis"])
print(f"\nBuffer Utilization: {result['buffer_health']['utilization']:.1%}")
print(f"Events Ingested This Turn: {len(result['ingested_events'])}")

Key Design Principles for Streaming Loaders

1. Decouple Fetch Rate from Processing Rate

The AsyncEventBuffer acts as a shock absorber. Sources may burst at 500 events/sec while vector indexing sustains only 50/sec. The buffer absorbs bursts and signals backpressure when full. Never let source velocity dictate processing cadence.

2. Make Cursors First-Class State

Batch loaders treat position as an implementation detail. Streaming loaders treat cursors as resumable, auditable, versioned state. Every cursor includes a checkpoint hash so you can detect corruption or tampering after restart.

3. Normalize Before Indexing

Each stream has unique schemas. Normalize to a common IngestedEvent envelope before buffering. This enables cross-stream deduplication, unified retrieval, and consistent metadata filtering in the vector store.

4. Hash Everything

Content hashes serve triple duty: deduplication across streams, integrity verification for cursors, and citation anchors for analyst responses. Compute hashes deterministically (ticker|event_type|content) so identical events from different sources produce identical hashes.

5. Monitor Buffer Health as Graph State

Buffer utilization isn't just an ops metric—it's analytical context. When the analysis agent sees 95% utilization, it knows recent events may be incomplete and should qualify its conclusions accordingly. This is impossible with batch loaders.

Conclusion

High-velocity financial RAG demands rethinking document loading from a batch operation to a streaming primitive. The custom loaders demonstrated here—EDGAR filings, options flow, earnings revisions—are not merely faster versions of existing loaders. They are architecturally different: they maintain state, enforce backpressure, normalize across heterogeneous sources, and expose their operational health as first-class graph state. When integrated into a LangGraph multi-agent system, streaming loaders transform RAG from a question-answering tool into a real-time situational awareness platform. The ingestion validation agent monitors data freshness, the analysis agent reasons over temporally-ordered cross-stream events, and the entire system resumes seamlessly after failures. The fundamental insight is that in streaming RAG, the loader is not a preprocessing step—it is a continuous agent. Treat it accordingly: give it state, give it error handling, give it observability, and integrate it deeply into your graph's reasoning lifecycle. The alpha is in the stream; make sure your architecture can actually drink from it.