FMP 数据

通过自然语言查询访问金融市场数据。

概述

FMP（金融建模准备）LangChain 集成提供了一种通过自然语言查询访问金融市场数据的无缝方式。此集成提供两个主要组件

• FMPDataToolkit：根据自然语言查询创建工具集合
• FMPDataTool：一个统一的工具，可自动选择和使用适当的端点

该集成利用 LangChain 的语义搜索功能，将用户查询与最相关的 FMP API 端点进行匹配，从而使金融数据访问更加直观和高效。

设置

!pip install -U langchain-fmp-data

import os

# Replace with your actual API keys
os.environ["FMP_API_KEY"] = "your-fmp-api-key"  # pragma: allowlist secret
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"  # pragma: allowlist secret

设置 LangSmith 以获得一流的可观察性也很有帮助（但不是必需的）

# os.environ["LANGSMITH_TRACING"] = "true"
# os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

实例化

实例化 FMP LangChain 集成主要有两种方法

1. 使用 FMPDataToolkit

from langchain_fmp_data import FMPDataToolkit

query = "Get stock market prices and technical indicators"
# Basic instantiation
toolkit = FMPDataToolkit(query=query)

# Instantiation with specific query focus
market_toolkit = FMPDataToolkit(
    query=query,
    num_results=5,
)

# Instantiation with custom configuration
custom_toolkit = FMPDataToolkit(
    query="Financial analysis",
    num_results=3,
    similarity_threshold=0.4,
    cache_dir="/custom/cache/path",
)

2. 使用 FMPDataTool

from langchain_fmp_data import FMPDataTool
from langchain_fmp_data.tools import ResponseFormat

# Basic instantiation
tool = FMPDataTool()

# Advanced instantiation with custom settings
advanced_tool = FMPDataTool(
    max_iterations=50,
    temperature=0.2,
)

调用

可以通过多种方式调用这些工具

直接调用

# Using FMPDataTool
tool_direct = FMPDataTool()

# Basic query
# fmt: off
result = tool.invoke({"query": "What's Apple's current stock price?"})
# fmt: on

# Advanced query with specific format
# fmt: off
detailed_result = tool_direct.invoke(
    {
        "query": "Compare Tesla and Ford's profit margins",
        "response_format": ResponseFormat.BOTH,
    }
)
# fmt: on

与 LangChain Agents 一起使用

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

# Setup
llm = ChatOpenAI(temperature=0)
toolkit = FMPDataToolkit(
    query="Stock analysis",
    num_results=3,
)
tools = toolkit.get_tools()

# Create agent
prompt = "You are a helpful assistant. Answer the user's questions based on the provided context."
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
)

# Run query
# fmt: off
response = agent_executor.invoke({"input": "What's the PE ratio of Microsoft?"})
# fmt: on

API 参考：AgentExecutor | create_openai_functions_agent | ChatOpenAI

高级用法

您可以自定义工具的行为

# Initialize with custom settings
advanced_tool = FMPDataTool(
    max_iterations=50,  # Increase max iterations for complex queries
    temperature=0.2,  # Adjust temperature for more/less focused responses
)

# Example of a complex multi-part analysis
query = """
Analyze Apple's financial health by:
1. Examining current ratios and debt levels
2. Comparing profit margins to industry average
3. Looking at cash flow trends
4. Assessing growth metrics
"""
# fmt: off
response = advanced_tool.invoke(
    {
        "query": query,
        "response_format": ResponseFormat.BOTH}
)
# fmt: on
print("Detailed Financial Analysis:")
print(response)

链式调用

您可以像其他工具一样链接该工具，只需使用所需的模型创建链即可。

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# Setup
llm = ChatOpenAI(temperature=0)
toolkit = FMPDataToolkit(query="Stock analysis", num_results=3)
tools = toolkit.get_tools()

llm_with_tools = llm.bind(functions=tools)
output_parser = StrOutputParser()
# Create chain
runner = llm_with_tools | output_parser

# Run chain
# fmt: off
response = runner.invoke(
    {
        "input": "What's the PE ratio of Microsoft?"
    }
)
# fmt: on

API 参考：StrOutputParser | ChatOpenAI

API 参考

FMPDataToolkit

用于创建 FMP API 工具集合的主类

from typing import Any

from langchain.tools import Tool


class FMPDataToolkit:
    """Creates a collection of FMP data tools based on queries."""

    def __init__(
        self,
        query: str | None = None,
        num_results: int = 3,
        similarity_threshold: float = 0.3,
        cache_dir: str | None = None,
    ): ...

    def get_tools(self) -> list[Tool]:
        """Returns a list of relevant FMP API tools based on the query."""
        ...

API 参考：Tool

FMPDataTool

统一工具，可自动选择适当的 FMP 端点

# fmt: off
class FMPDataTool:
    """Single unified tool for accessing FMP data through natural language."""

    def __init__(
            self,
            max_iterations: int = 3,
            temperature: float = 0.0,
    ): ...

    def invoke(
            self,
            input: dict[str, Any],
    ) -> str | dict[str, Any]:
        """Execute a natural language query against FMP API."""
        ...

# fmt: on

ResponseFormat

用于控制响应格式的枚举

from enum import Enum


class ResponseFormat(str, Enum):
    RAW = "raw"  # Raw API response
    ANALYSIS = "text"  # Natural language analysis
    BOTH = "both"  # Both raw data and analysis

相关内容

• 工具 概念指南
• 工具 操作指南