Options Analytics Agent with LangGraph

Skill by ara.so — Data Skills collection.

A sophisticated LangGraph-based agent that automates financial options analysis with real-time data from Polygon.io, smart caching via ChromaDB, persistent memory, and professional-grade analysis. Built for creating intelligent trading assistants with RAG capabilities and microservice architecture.

What It Does

This project provides a complete AI agent system for:

Real-time options data retrieval from Polygon.io with intelligent caching
RAG-powered knowledge base using ChromaDB for semantic search
Persistent conversation memory across sessions via SQLite
Professional options analysis with Greeks, sentiment, and anomaly detection
Multi-format exports (CSV, charts, reports)
LangGraph orchestration for multi-agent workflows
FastAPI microservice deployment

Installation

Prerequisites

bash

# Python 3.10+
python --version

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

Install Dependencies

bash

pip install -r requirements.txt

Key dependencies:

langchain>=0.3.0
langgraph>=0.2.45
langchain-openai>=0.2.6
langchain-chroma>=0.1.4
chromadb>=0.5.20
fastapi>=0.115.5
uvicorn>=0.32.1
pandas>=2.2.3
matplotlib>=3.9.2
tavily-python>=0.5.0

Environment Configuration

Create

.env

file in project root:

bash

# Required
OPENAI_API_KEY=your_openai_api_key
POLYGON_API_KEY=your_polygon_io_api_key

# Optional
TAVILY_API_KEY=your_tavily_api_key  # For web search
LANGCHAIN_API_KEY=your_langchain_api_key  # For tracing
LANGCHAIN_TRACING_V2=true

Verify Installation

python

# Test import
from agent_main import create_agent_workflow
from config.settings import validate_api_keys

# Validate API keys
validate_api_keys()
print("✓ Installation successful")

Project Structure

project/
├── agent_main.py              # Main agent entry point
├── config/settings.py         # Configuration management
├── tools/
│   ├── search/               # Options search tools
│   ├── export/               # Data export tools
│   └── analysis/             # Analysis tools
├── rag/                      # RAG knowledge base
├── monitoring/               # Performance tracking
└── microservice/             # FastAPI deployment

Core Usage

Basic Agent Interaction

python

from agent_main import create_agent_workflow
from langchain_core.messages import HumanMessage

# Create agent
workflow = create_agent_workflow()
app = workflow.compile()

# Simple query
config = {"configurable": {"thread_id": "session_1"}}
query = "Search for AAPL options expiring this week"

result = app.invoke(
    {"messages": [HumanMessage(content=query)]},
    config=config
)

print(result["messages"][-1].content)

Interactive Chat Loop

python

from agent_main import create_agent_workflow
from langchain_core.messages import HumanMessage

def chat():
    workflow = create_agent_workflow()
    app = workflow.compile()
    session_id = "user_session_1"
    
    print("Options Analytics Agent (type 'exit' to quit)")
    
    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in ['exit', 'quit']:
            break
            
        config = {"configurable": {"thread_id": session_id}}
        result = app.invoke(
            {"messages": [HumanMessage(content=user_input)]},
            config=config
        )
        
        response = result["messages"][-1].content
        print(f"\nAgent: {response}")

if __name__ == "__main__":
    chat()

Key Tools & Commands

1. Options Search Tool

Search for options data with automatic caching:

python

from tools.search.options_search import OptionsSearchTool

tool = OptionsSearchTool()

# Search with automatic caching
result = tool._run(
    ticker="NVDA",
    expiration_date="2024-12-20",
    option_type="call",
    force_refresh=False  # Use cache if available
)

# Force fresh API call
result = tool._run(
    ticker="NVDA",
    expiration_date="2024-12-20",
    option_type="call",
    force_refresh=True
)

2. Batch Search Tool

Search multiple tickers efficiently:

python

from tools.search.batch_search import BatchOptionsSearchTool

tool = BatchOptionsSearchTool()

result = tool._run(
    tickers=["AAPL", "MSFT", "GOOGL"],
    expiration_date="2024-12-31",
    option_type="call"
)

3. RAG Knowledge Base

Query cached options data semantically:

python

from rag.rag_tools import RAGQueryTool

rag_tool = RAGQueryTool()

# Semantic search
results = rag_tool._run(
    query="high volume AAPL calls near the money",
    top_k=5
)

# Date-based retrieval
from rag.rag_collection_tools import DateRangeCollectionTool

date_tool = DateRangeCollectionTool()
data = date_tool._run(
    ticker="AAPL",
    start_date="2024-12-01",
    end_date="2024-12-31"
)

4. Options Analysis

Professional-grade analysis:

python

from tools.analysis.analysis_tools import AnalyzeOptionsTool

analysis_tool = AnalyzeOptionsTool()

result = analysis_tool._run(
    ticker="TSLA",
    expiration_date="2024-12-20",
    analysis_type="sentiment"  # or "greeks", "anomaly"
)

print(result)

5. Data Export

Export to CSV or charts:

python

from tools.export.csv_export import CSVExportTool
from tools.export.visualization import ChartVisualizationTool

# CSV export
csv_tool = CSVExportTool()
csv_tool._run(
    ticker="AAPL",
    expiration_date="2024-12-20",
    option_type="call",
    output_filename="aapl_calls.csv"
)

# Chart generation
chart_tool = ChartVisualizationTool()
chart_tool._run(
    ticker="AAPL",
    expiration_date="2024-12-20",
    chart_type="volume_oi"
)

Configuration

Settings Management

python

# config/settings.py
from config.settings import (
    OPENAI_API_KEY,
    POLYGON_API_KEY,
    MODEL_NAME,
    CHROMA_PERSIST_DIR,
    validate_api_keys
)

# Validate all keys
validate_api_keys()

# Access configuration
print(f"Model: {MODEL_NAME}")
print(f"ChromaDB: {CHROMA_PERSIST_DIR}")

Custom Agent Configuration

python

from agent_main import create_agent_workflow
from langgraph.checkpoint.memory import MemorySaver

# Create with custom checkpointer
memory = MemorySaver()
workflow = create_agent_workflow()
app = workflow.compile(checkpointer=memory)

# Or use SQLite checkpointer
from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = workflow.compile(checkpointer=checkpointer)

Advanced Patterns

Custom Tool Integration

python

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def custom_options_analyzer(ticker: str, metric: str) -> str:
    """Analyze specific option metric.
    
    Args:
        ticker: Stock symbol
        metric: Metric to analyze (volatility, skew, etc.)
    """
    # Your custom logic
    return f"Analysis for {ticker}: {metric}"

# Add to agent
from config.settings import get_llm

llm = get_llm()
tools = [custom_options_analyzer]
agent = create_react_agent(llm, tools)

RAG Knowledge Base Setup

python

from rag.rag_knowledge_base import RAGKnowledgeBase

# Initialize
kb = RAGKnowledgeBase(
    persist_directory="./data/chroma_db",
    collection_name="options_data"
)

# Add documents
kb.add_documents([
    {
        "ticker": "AAPL",
        "expiration": "2024-12-20",
        "strike": 180.0,
        "type": "call",
        "volume": 5000,
        "open_interest": 10000
    }
])

# Query
results = kb.query(
    query_text="high volume Apple calls",
    n_results=5
)

Persistent Memory Across Sessions

python

from langgraph.checkpoint.sqlite import SqliteSaver

# Create persistent checkpointer
checkpointer = SqliteSaver.from_conn_string("./data/conversation_memory.db")

workflow = create_agent_workflow()
app = workflow.compile(checkpointer=checkpointer)

# Session 1
config1 = {"configurable": {"thread_id": "user_123"}}
app.invoke({"messages": [HumanMessage("Search AAPL options")]}, config1)

# Session 2 (remembers previous context)
app.invoke({"messages": [HumanMessage("Show me the calls")]}, config1)

Streaming Responses

python

from langchain_core.messages import HumanMessage

workflow = create_agent_workflow()
app = workflow.compile()

config = {"configurable": {"thread_id": "session_1"}}
query = HumanMessage(content="Analyze TSLA options")

# Stream tokens
for chunk in app.stream({"messages": [query]}, config):
    if "messages" in chunk:
        print(chunk["messages"][-1].content, end="", flush=True)

Microservice Deployment

FastAPI Server

python

# microservice/app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent_main import create_agent_workflow
from langchain_core.messages import HumanMessage

app = FastAPI(title="Options Analytics API")
workflow = create_agent_workflow()
agent_app = workflow.compile()

class QueryRequest(BaseModel):
    query: str
    session_id: str = "default"

@app.post("/query")
async def query_agent(request: QueryRequest):
    try:
        config = {"configurable": {"thread_id": request.session_id}}
        result = agent_app.invoke(
            {"messages": [HumanMessage(content=request.query)]},
            config=config
        )
        return {
            "response": result["messages"][-1].content,
            "session_id": request.session_id
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# Run: uvicorn microservice.app:app --reload

Docker Deployment

dockerfile

# microservice/Dockerfile
FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "microservice.app:app", "--host", "0.0.0.0", "--port", "8000"]

yaml

# docker-compose.yml
version: '3.8'

services:
  options-agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - POLYGON_API_KEY=${POLYGON_API_KEY}
    volumes:
      - ./data:/app/data
      - ./outputs:/app/outputs

Start service:

bash

docker-compose up -d

Common Workflows

Complete Options Analysis Pipeline

python

from agent_main import create_agent_workflow
from langchain_core.messages import HumanMessage

def analyze_options_workflow(ticker: str, expiration: str):
    workflow = create_agent_workflow()
    app = workflow.compile()
    
    config = {"configurable": {"thread_id": f"analysis_{ticker}"}}
    
    # Step 1: Search options
    query1 = f"Search {ticker} options expiring {expiration}"
    result1 = app.invoke({"messages": [HumanMessage(query1)]}, config)
    
    # Step 2: Analyze sentiment
    query2 = f"Analyze sentiment for {ticker} options"
    result2 = app.invoke({"messages": [HumanMessage(query2)]}, config)
    
    # Step 3: Export to CSV
    query3 = f"Export {ticker} options to CSV"
    result3 = app.invoke({"messages": [HumanMessage(query3)]}, config)
    
    return {
        "search": result1["messages"][-1].content,
        "analysis": result2["messages"][-1].content,
        "export": result3["messages"][-1].content
    }

# Run pipeline
results = analyze_options_workflow("NVDA", "2024-12-31")

Batch Processing Multiple Tickers

python

from tools.search.batch_search import BatchOptionsSearchTool
from tools.analysis.analysis_tools import AnalyzeOptionsTool

def batch_analysis(tickers: list, expiration: str):
    search_tool = BatchOptionsSearchTool()
    analysis_tool = AnalyzeOptionsTool()
    
    results = {}
    
    # Batch search
    search_result = search_tool._run(
        tickers=tickers,
        expiration_date=expiration,
        option_type="call"
    )
    
    # Individual analysis
    for ticker in tickers:
        analysis = analysis_tool._run(
            ticker=ticker,
            expiration_date=expiration,
            analysis_type="sentiment"
        )
        results[ticker] = analysis
    
    return results

# Process watchlist
watchlist = ["AAPL", "MSFT", "GOOGL", "NVDA", "TSLA"]
results = batch_analysis(watchlist, "2024-12-31")

Real-time Monitoring with Callbacks

python

from langchain.callbacks.base import BaseCallbackHandler
from langchain_core.messages import HumanMessage

class OptionsMonitorCallback(BaseCallbackHandler):
    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f"🔧 Tool: {serialized['name']}")
    
    def on_tool_end(self, output, **kwargs):
        print(f"✅ Result: {output[:100]}...")

# Use callback
workflow = create_agent_workflow()
app = workflow.compile()

config = {
    "configurable": {"thread_id": "monitor_session"},
    "callbacks": [OptionsMonitorCallback()]
}

result = app.invoke(
    {"messages": [HumanMessage("Search AAPL options")]},
    config=config
)

Troubleshooting

API Key Issues

python

# Validate API keys
from config.settings import validate_api_keys

try:
    validate_api_keys()
    print("✓ All API keys valid")
except ValueError as e:
    print(f"✗ Missing: {e}")
    # Set missing keys in .env file

ChromaDB Connection Errors

python

# Reset ChromaDB
import shutil
import os

chroma_path = "./data/chroma_db"
if os.path.exists(chroma_path):
    shutil.rmtree(chroma_path)
    print("ChromaDB reset")

# Reinitialize
from rag.rag_knowledge_base import RAGKnowledgeBase
kb = RAGKnowledgeBase()

Clear Conversation Memory

python

# Clear SQLite memory
import os

memory_file = "./data/conversation_memory.db"
if os.path.exists(memory_file):
    os.remove(memory_file)
    print("Conversation memory cleared")

Polygon.io Rate Limits

python

# Use caching to reduce API calls
from tools.search.options_search import OptionsSearchTool

tool = OptionsSearchTool()

# Always try cache first
result = tool._run(
    ticker="AAPL",
    expiration_date="2024-12-20",
    option_type="call",
    force_refresh=False  # Use cached data
)

# Only force refresh when absolutely necessary

Debug Agent State

python

# Inspect agent state
from agent_main import create_agent_workflow

workflow = create_agent_workflow()
app = workflow.compile()

config = {"configurable": {"thread_id": "debug_session"}}
result = app.invoke(
    {"messages": [HumanMessage("Search AAPL options")]},
    config=config
)

# Print full state
print("Messages:", result["messages"])
print("Tools called:", [m.additional_kwargs for m in result["messages"]])

Enable Tracing

python

# Set in .env
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_API_KEY=your_key

# Or in code
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your_key"

# View traces at https://smith.langchain.com

Performance Optimization

Batch Embeddings

python

from rag.rag_knowledge_base import RAGKnowledgeBase

kb = RAGKnowledgeBase()

# Batch add for efficiency
documents = [
    {"ticker": "AAPL", "data": "..."},
    {"ticker": "MSFT", "data": "..."},
    # ... more documents
]

kb.add_documents(documents)  # Processes in batches

Parallel Tool Execution

python

from concurrent.futures import ThreadPoolExecutor
from tools.search.options_search import OptionsSearchTool

def search_ticker(ticker, date):
    tool = OptionsSearchTool()
    return tool._run(ticker=ticker, expiration_date=date, option_type="call")

tickers = ["AAPL", "MSFT", "GOOGL", "NVDA"]
date = "2024-12-31"

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(lambda t: search_ticker(t, date), tickers))

Testing

Unit Test Example

python

import unittest
from tools.search.options_search import OptionsSearchTool

class TestOptionsSearch(unittest.TestCase):
    def setUp(self):
        self.tool = OptionsSearchTool()
    
    def test_search_call_options(self):
        result = self.tool._run(
            ticker="AAPL",
            expiration_date="2024-12-20",
            option_type="call"
        )
        self.assertIn("AAPL", result)
        self.assertIn("call", result.lower())

if __name__ == "__main__":
    unittest.main()

This skill provides comprehensive coverage of the Options Analytics Agent project, enabling AI coding agents to effectively assist developers in building sophisticated financial analysis systems with LangGraph, RAG, and persistent memory.

options-analytics-agent-langgraph

NPX Install

Tags

SKILL.md Content

Options Analytics Agent with LangGraph

What It Does

Installation

Prerequisites

Install Dependencies

Environment Configuration

Verify Installation

Project Structure

Core Usage

Basic Agent Interaction

Interactive Chat Loop

Key Tools & Commands

1. Options Search Tool

2. Batch Search Tool

3. RAG Knowledge Base

4. Options Analysis

5. Data Export

Configuration

Settings Management

Custom Agent Configuration

Advanced Patterns

Custom Tool Integration

RAG Knowledge Base Setup

Persistent Memory Across Sessions

Streaming Responses

Microservice Deployment

FastAPI Server

Docker Deployment

Common Workflows

Complete Options Analysis Pipeline

Batch Processing Multiple Tickers

Real-time Monitoring with Callbacks

Troubleshooting

API Key Issues

ChromaDB Connection Errors

Clear Conversation Memory

Polygon.io Rate Limits

Debug Agent State

Enable Tracing

Performance Optimization

Batch Embeddings

Parallel Tool Execution

Testing

Unit Test Example