使用说明
IsoFinancial MCP
一个增强的开源MCP(模型上下文协议)服务器,为定量交易机会检测和分析提供全面的金融市场数据端点。提供高级数据源,包括美国证券交易委员会文件、美国金融业监管局空头交易量、收益日历、新闻情绪和谷歌趋势分析。
🚀 增强功能
核心市场数据(雅虎财经)
- 实时市场数据:实时股票价格、交易量和市场统计数据
- 财务报表:资产负债表、损益表和现金流量数据
- 期权分析:带有到期日期和希腊语的完整期权链
- 公司行动:股息、股票分割和历史行为
- 公司信息:详细资料、主要持有人和机构投资者
- 分析师建议:专业分析师评级和价格目标
🆕 用于定量分析的增强数据源
- SEC文件集成:实时EDGAR API访问8-K、S-3、424B、10-Q、10-K文件,6小时缓存
- FINRA短交易量:每日短体积比和压力指标,带趋势分析
- 收益日历:每股收益估计值、实际值、BMO/AMC时间安排的意外百分比
- 新闻标题:雅虎财经RSS集成源归因和重复检测
- 谷歌趋势:使用动量指标和相关查询进行搜索量分析
- 多源战略:SerpAPI(如果已配置)→ DuckDuckGo(免费)→ PyTrends(回退)
- 备注PyTrends经常失败,出现429个错误;DuckDuckGo提供可靠的估计趋势
- 看 TRENDS_SOURCES.md 用于详细的源比较
🔧 高级技术功能
- 智能高速缓存:多层缓存系统,每个数据源具有可配置的TTL
- 速率限制:内置速率限制,带指数退避,用于API保护
- 错误处理:优雅的降级,详细的错误报告
- 性能优化:通过连接池实现异步/等待
- 数据验证:全面的输入验证和净化
🎯 代理优化的元工具
v0.3.0中的新功能:专门为具有迭代预算的LLM代理设计的整合元工具。这些工具大大减少了所需的API调用的数量,确保您的代理可以在不耗尽迭代的情况下生成完整的报告。
为什么选择元工具?
传统方法要求 每个自动收报机10-15次单独通话,消耗宝贵的代理迭代:
# ❌ OLD WAY: 7+ separate calls per ticker
info = await get_info("AAPL")
prices = await get_historical_prices("AAPL")
news = await get_news_headlines("AAPL")
sec = await get_sec_filings("AAPL")
earnings = await get_earnings_calendar("AAPL")
short = await get_finra_short_volume("AAPL")
trends = await get_google_trends("AAPL")
# For 3 tickers: 21+ calls = 25+ agent iterations ❌
元工具将所有内容整合到 总共1-2个电话:
# ✅ NEW WAY: 1 call for everything
analysis = await get_multi_ticker_analysis("AAPL,MSFT,GOOGL")
# All data for 3 tickers in parallel = 1-2 agent iterations ✅
可用元工具
🎯 get_ticker_complete_analysis
获取 所有财务数据 一次通话只需一个股票。
# Single comprehensive call
uv run python -c "
from iso_financial_mcp.server import get_ticker_complete_analysis
import asyncio
result = asyncio.run(get_ticker_complete_analysis('AAPL', lookback_days=30))
print(result)
"
包括:
- 公司信息(行业、行业、市值)
- 历史价格(最近5天+30天变化)
- 最近的新闻(5篇文章和摘要)
- 美国证券交易委员会文件(最近3份)
- 收益数据(未来+3个最近季度)
- FINRA短交易量(汇总指标)
- 谷歌趋势(搜索势头)
参数:
ticker(str):股票代码(例如“AAPL”)include_options(bool):包含选项数据(默认值:False)lookback_days(int):历史数据窗口(默认值:30)
🎯 get_multi_ticker_analysis
分析 多个并行自动收报机 只需一个电话。
# Parallel analysis of multiple tickers
uv run python -c "
from iso_financial_mcp.server import get_multi_ticker_analysis
import asyncio
result = asyncio.run(get_multi_ticker_analysis('NVDA,AMD,INTC', lookback_days=30))
print(result)
"
特征:
- 并行数据检索(比顺序快5-10x)
- 自动收报机限制(最多10个,以防止超时)
- 每个自动收报机都有优雅的错误处理
- 预格式化、紧凑的输出,针对LLM进行了优化
参数:
tickers(str):逗号分隔的股票代码(例如,“AAPL、MSFT、GOOGL”)include_options(bool):包含选项数据(默认值:False)lookback_days(int):历史数据窗口(默认值:30)
性能比较
| 场景 | 之前(单个工具) | 之后(元工具) | 改进 |
|---|---|---|---|
| 单票分析 | 7次通话,~15s | 1次通话,~3 s | 速度提高5倍 |
| 3票务分析 | 21次通话,~45秒 | 1次通话,~5秒 | 快9倍 |
| 5票务分析 | 35次通话,约75秒 | 1次通话,~7秒 | 速度提高10倍以上 |
| 代理迭代 | 25+次迭代 | <20次迭代 | 保证HTML |
| 代币消费 | 约10000个代币 | 约3000个代币 | 减少70% |
现实世界示例:时事通讯生成
元工具之前(❌ 通常无法生成HTML):
Iteration 1-7: Analyze NVDA (7 individual calls)
Iteration 8-14: Analyze AMD (7 individual calls)
Iteration 15-21: Analyze INTC (7 individual calls)
Iteration 22-25: Try to start HTML generation...
Iteration 26+: ❌ TIMEOUT - No report generated
元工具之后(✅ 始终生成HTML):
Iteration 1-2: get_multi_ticker_analysis("NVDA,AMD,INTC")
→ All data retrieved in parallel
Iteration 3: Analyze consolidated results
Iteration 4-15: Generate comprehensive HTML report ✅
Iteration 16+: Refinements (optional)
令牌优化
元工具返回 预格式化的紧凑数据 针对LLM消耗进行了优化:
- 公司概述:截断为300个字符(而500多个单词)
- 历史价格:仅过去5天(与30天相比)
- 新闻文章:最多5篇文章,150个字符摘要
- 行情走势:仅最近3次
- 收益:仅前三个季度
- 汇总指标:统计摘要与原始数据
结果:在保持数据质量的同时减少50-70%的令牌。
与AI代理集成
非常适合有迭代预算的代理(例如,时事通讯生成器、交易分析师):
# Example: Newsletter agent with 30 iteration budget
from iso_financial_mcp.server import get_multi_ticker_analysis
# Reserve 10 iterations for HTML generation
# Use meta-tools for data gathering (1-2 iterations)
async def generate_newsletter(tickers: list):
# Single call gets ALL data
analysis = await get_multi_ticker_analysis(",".join(tickers))
# Agent now has 25+ iterations left for:
# - Analysis and insights
# - HTML report generation
# - Refinements and polish
return analysis
从单个工具迁移
从单个工具迁移到元工具很简单:
之前(单个工具):
# Multiple separate calls
info = await get_info("AAPL")
prices = await get_historical_prices("AAPL")
news = await get_news_headlines("AAPL")
# ... 7+ more calls
之后(元工具):
# Single consolidated call
analysis = await get_ticker_complete_analysis("AAPL")
# All data retrieved in parallel
最佳实践:
- 使用
get_ticker_complete_analysis单人深潜 - 使用
get_multi_ticker_analysis用于比较多个自动收报机 - 集
include_options=False如果不需要选项数据,则减少令牌使用 - 调整
lookback_days根据您的分析需求(默认值:30)
备注:所有单独的工具都可以向后兼容,但强烈建议将元工具用于基于代理的工作流。
📋 需求
- Python 3.10+ (建议使用Python 3.13+以获得最佳性能)
- UV包装管理器 (安装指南)
- Internet连接 用于API访问多个数据源
- 不需要API密钥 -所有数据源都使用免费/公共API
🔧 安装
使用紫外线(推荐)
# Install UV if not already installed
curl -LsSf https://astral.sh/uv/install.sh | sh
# Add to your project
uv add iso-financial-mcp
# Or install globally
uvx iso-financial-mcp
使用pip
pip install iso-financial-mcp
用于AI交易系统集成
IsoFinancial MCP服务器可以轻松集成到任何AI交易系统或定量分析管道中:
# Add to your trading system dependencies
uv add iso-financial-mcp
# Or include in your pyproject.toml
dependencies = ["iso-financial-mcp>=0.2.0"]
🚀 快速开始
作为MCP服务器(推荐用于AI代理)
# Test the server directly
uv run python -m iso_financial_mcp
# Or run specific endpoints
uv run python -c "
from iso_financial_mcp.server import get_info
import asyncio
result = asyncio.run(get_info('AAPL'))
print(result)
"
与人工智能交易系统集成
该服务器可以轻松地与任何AI交易系统或定量分析框架集成:
# Example integration with MCP-compatible AI agents
from fastmcp.agent import StdioServerParams, mcp_server_tools
finance_server_params = StdioServerParams(
command="python",
args=["-m", "iso_financial_mcp"],
)
# Get available financial tools
finance_tools = await mcp_server_tools(finance_server_params)
作为独立的HTTP服务器
# Start HTTP server with UV
uv run uvicorn iso_financial_mcp.server:server.app --host 0.0.0.0 --port 8000
# Test endpoints
curl http://localhost:8000/health
测试单个端点
# Test SEC filings
uv run python -c "
from iso_financial_mcp.server import get_sec_filings
import asyncio
result = asyncio.run(get_sec_filings('AAPL', '8-K,S-3', 30))
print(result)
"
# Test FINRA short volume
uv run python -c "
from iso_financial_mcp.server import get_finra_short_volume
import asyncio
result = asyncio.run(get_finra_short_volume('GME'))
print(result)
"
# Test earnings calendar
uv run python -c "
from iso_financial_mcp.server import get_earnings_calendar
import asyncio
result = asyncio.run(get_earnings_calendar('NVDA'))
print(result)
"
📊 可用端点
📈 核心市场数据(雅虎财经)
get_info(ticker)-公司简介和基本信息get_historical_prices(ticker, period, interval)-OHLCV的历史价格数据get_actions(ticker)-股息和股票分割历史get_earnings_dates(ticker)-即将公布的收益日期和历史收益日期get_isin(ticker)-国际证券识别号
💰 财务报表
get_balance_sheet(ticker, freq)-资产负债表数据(年度/季度)get_financials(ticker, freq)-损益表数据(年度/季度)get_cash_flow(ticker, freq)-现金流量表(年度/季度)
📊 期权分析
get_options_expirations(ticker)-可用的到期日期get_option_chain(ticker, expiration_date)-与希腊人建立完整的期权链
🏢 公司信息
get_major_holders(ticker)-主要股东和内部人持股get_institutional_holders(ticker)-机构投资者头寸get_recommendations(ticker)-分析师建议和价格目标
🆕 用于定量分析的增强数据源
📋 美国证券交易委员会文件(EDGAR API)
get_sec_filings(ticker, form_types="8-K,S-3,424B,10-Q,10-K", lookback_days=30)
- 表单类型:8-K(重大事件)、S-3(货架注册)、424B(招股说明书)、10-Q/10-K(季度/年度报告)
- 缓存TTL:6小时
- 特性:EDGAR API直接集成、登录号、提交URL
📊 FINRA短交易量
get_finra_short_volume(ticker, start_date="", end_date="")
- 数据源:FINRA每日短容量CSV文件
- 缓存TTL:24小时
- 特性:短期比率、趋势分析、综合指标、5天滚动平均值
📅 收益日历
get_earnings_calendar(ticker)
- 数据源:雅虎财经/纳斯达克盈利数据
- 缓存TTL:24小时
- 特性:每股收益估计值、实际值、意外百分比、BMO/AMC时间、即将到来与历史
📰 新闻标题
get_news_headlines(ticker, limit=10, lookback_days=3)
- 数据源:雅虎财经RSS订阅源
- 缓存TTL:2小时
- 特性:来源归因、重复检测、摘要提取、发布时间戳
📈 谷歌趋势
get_google_trends(term, window_days=30)
- 数据源 (优先级顺序):
- SerpAPI (如果已配置):完整的谷歌趋势数据,高度可靠
- 鸭鸭搜 (默认):估计趋势,免费可靠
- PyTrends (回退):完整数据,但经常失败,出现429个错误
- 缓存TTL:24小时
- 特性:搜索量趋势、动量分析、相关查询(仅限SerpAPI/PyTrends)
- 配置:设置
SERPAPI_KEY为了获得最佳效果,或依靠免费的DuckDuckGo - 看: TRENDS_SOURCES.md 进行详细比较
🔧 技术特性
缓存系统
- 多层缓存 每个端点可配置TTL
- 内存效率高 具有自动清理功能
- 缓存预热 用于频繁访问的数据
速率限制
- 每终点速率限制 指数回退
- 突发保护 使用令牌桶算法
- API特定限制 尊重提供者约束
错误处理
- 优雅降级 当数据源不可用时
- 详细的错误报告 附上下文和建议
- 自动重试 具有智能退避策略
📖 用法示例
基本市场数据
# Get company information
uv run python -c "
from iso_financial_mcp.server import get_info
import asyncio
result = asyncio.run(get_info('AAPL'))
print(result)
"
# Get historical prices with custom period
uv run python -c "
from iso_financial_mcp.server import get_historical_prices
import asyncio
result = asyncio.run(get_historical_prices('TSLA', '6mo', '1d'))
print(result)
"
财务分析
# Get quarterly financials
uv run python -c "
from iso_financial_mcp.server import get_financials
import asyncio
result = asyncio.run(get_financials('NVDA', 'quarterly'))
print(result)
"
# Get balance sheet
uv run python -c "
from iso_financial_mcp.server import get_balance_sheet
import asyncio
result = asyncio.run(get_balance_sheet('AAPL', 'yearly'))
print(result)
"
期权分析
# Get option expirations
uv run python -c "
from iso_financial_mcp.server import get_options_expirations
import asyncio
result = asyncio.run(get_options_expirations('SPY'))
print(result)
"
# Get complete option chain
uv run python -c "
from iso_financial_mcp.server import get_option_chain
import asyncio
result = asyncio.run(get_option_chain('SPY', '2024-12-20'))
print(result)
"
🆕 增强的数据源
SEC文件分析
# Get recent 8-K and S-3 filings
uv run python -c "
from iso_financial_mcp.server import get_sec_filings
import asyncio
result = asyncio.run(get_sec_filings('GME', '8-K,S-3', 30))
print(result)
"
# Get all major filing types
uv run python -c "
from iso_financial_mcp.server import get_sec_filings
import asyncio
result = asyncio.run(get_sec_filings('AAPL', '8-K,S-3,424B,10-Q,10-K', 60))
print(result)
"
FINRA短交易量分析
# Get short volume with trend analysis
uv run python -c "
from iso_financial_mcp.server import get_finra_short_volume
import asyncio
result = asyncio.run(get_finra_short_volume('AMC'))
print(result)
"
# Get short volume for specific date range
uv run python -c "
from iso_financial_mcp.server import get_finra_short_volume
import asyncio
result = asyncio.run(get_finra_short_volume('GME', '2024-01-01', '2024-01-31'))
print(result)
"
惊喜的收益日历
# Get comprehensive earnings data
uv run python -c "
from iso_financial_mcp.server import get_earnings_calendar
import asyncio
result = asyncio.run(get_earnings_calendar('NVDA'))
print(result)
"
# Analyze earnings surprises
uv run python -c "
from iso_financial_mcp.server import get_earnings_calendar
import asyncio
result = asyncio.run(get_earnings_calendar('TSLA'))
print(result)
"
新闻情绪分析
# Get recent news headlines
uv run python -c "
from iso_financial_mcp.server import get_news_headlines
import asyncio
result = asyncio.run(get_news_headlines('AAPL', 15, 5))
print(result)
"
# Monitor breaking news
uv run python -c "
from iso_financial_mcp.server import get_news_headlines
import asyncio
result = asyncio.run(get_news_headlines('TSLA', 5, 1))
print(result)
"
谷歌趋势社交动力
# Analyze search trends for ticker
uv run python -c "
from iso_financial_mcp.server import get_google_trends
import asyncio
result = asyncio.run(get_google_trends('GameStop', 30))
print(result)
"
# Monitor company name trends
uv run python -c "
from iso_financial_mcp.server import get_google_trends
import asyncio
result = asyncio.run(get_google_trends('Tesla Motors', 60))
print(result)
"
与人工智能交易代理集成
IsoFinancial MCP服务器可以与任何兼容MCP的AI交易代理集成:
# Example: Using with AI trading agents
from your_trading_agent import TradingAgent
agent = TradingAgent()
agent.add_mcp_server("iso-financial-mcp")
# Agent can now use all financial endpoints
# Example agent query: "Get SEC filings and short volume data for GME"
🔧 配置
不需要API密钥
服务器使用完全免费的公共API:
- 雅虎财经:市场数据、财务、期权(无身份验证)
- 第二埃德加:美国证券交易委员会正式文件API(公开访问)
- 美国金融业监管局:每日短容量CSV文件(公共数据)
- 谷歌趋势:多源(DuckDuckGo免费,SerpAPI可选,PyTrends回退)
- RSS 订阅源:雅虎财经新闻头条RSS(公共)
可选配置
# Copy environment template (optional for advanced configuration)
cp .env.example .env
缓存配置
针对交易应用程序优化的默认缓存TTL设置:
CACHE_TTL = {
'sec_filings': 21600, # 6 hours - SEC filings don't change frequently
'finra_data': 86400, # 24 hours - FINRA data is daily
'earnings': 86400, # 24 hours - Earnings calendar updates daily
'news': 7200, # 2 hours - News updates frequently
'trends': 86400, # 24 hours - Trends data is daily
'market_data': 300, # 5 minutes - Market data for real-time needs
'options_data': 900 # 15 minutes - Options data updates frequently
}
速率限制配置
内置速率限制尊重API提供商限制:
RATE_LIMITS = {
'yahoo_finance': {'calls_per_minute': 120, 'burst_limit': 20},
'sec_edgar': {'calls_per_minute': 10, 'burst_limit': 3},
'google_trends': {'calls_per_minute': 20, 'burst_limit': 5},
'rss_feeds': {'calls_per_minute': 60, 'burst_limit': 10}
}
与交易系统集成
当与AI交易系统集成时,配置可以自动处理:
# Example: Automatic configuration in trading systems
from iso_financial_mcp.server import server
# Server can be embedded directly in your application
app = YourTradingApp()
app.add_mcp_server(server) # All endpoints available automatically
🧪 测试
综合测试套件
# Run all tests with UV
uv run pytest
# Run with coverage reporting
uv run pytest --cov=iso_financial_mcp --cov-report=html
# Run specific test categories
uv run pytest tests/test_yfinance_source.py # Yahoo Finance endpoints
uv run pytest tests/test_sec_source.py # SEC filings
uv run pytest tests/test_finra_source.py # FINRA short volume
uv run pytest tests/test_earnings_source.py # Earnings calendar
uv run pytest tests/test_news_source.py # News headlines
uv run pytest tests/test_trends_source.py # Google Trends
# Integration tests
uv run pytest tests/test_integration.py # End-to-end testing
手动测试
# Test individual endpoints
uv run python -c "
from iso_financial_mcp.server import get_info, get_sec_filings, get_finra_short_volume
import asyncio
async def test_endpoints():
# Test basic market data
info = await get_info('AAPL')
print('✅ Basic info:', 'OK' if info else 'FAILED')
# Test SEC filings
filings = await get_sec_filings('AAPL', '8-K', 30)
print('✅ SEC filings:', 'OK' if 'SEC Filings' in filings else 'FAILED')
# Test FINRA data
short_data = await get_finra_short_volume('GME')
print('✅ FINRA data:', 'OK' if 'Short Volume Data' in short_data else 'FAILED')
asyncio.run(test_endpoints())
"
# Test server startup
uv run python -m iso_financial_mcp
性能测试
# Test with high-volume tickers
uv run python -c "
import asyncio
import time
from iso_financial_mcp.server import get_info
async def performance_test():
tickers = ['AAPL', 'TSLA', 'NVDA', 'GME', 'AMC']
start_time = time.time()
tasks = [get_info(ticker) for ticker in tickers]
results = await asyncio.gather(*tasks)
end_time = time.time()
print(f'Processed {len(tickers)} tickers in {end_time - start_time:.2f} seconds')
print(f'Average: {(end_time - start_time) / len(tickers):.2f} seconds per ticker')
asyncio.run(performance_test())
"
📦 发展
开发设置
# Clone the repository
git clone https://github.com/Niels-8/isofinancial-mcp.git
cd isofinancial-mcp
# Install with development dependencies using UV
uv sync --dev
# Or install in editable mode
uv pip install -e ".[dev]"
代码质量工具
# Format code with Black
uv run black .
# Lint with Ruff
uv run ruff check . --fix
# Type checking with MyPy
uv run mypy .
# Run all quality checks
uv run black . && uv run ruff check . && uv run mypy .
添加新数据源
- 创建数据源模块:在中添加新模块
iso_financial_mcp/datasources/ - 实现异步函数:使用缓存和错误处理遵循现有模式
- 添加服务器终结点:在中注册新端点
server.py随着@server.tool装饰器 - 编写测试:在中添加综合测试
tests/目录 - 更新文档:使用新的端点文档更新README
示例:添加新数据源
# iso_financial_mcp/datasources/new_source.py
import asyncio
import aiohttp
from typing import Optional, List, Dict, Any
from ..utils.cache import cached_request
from ..utils.rate_limiter import rate_limit
@rate_limit('new_api', calls_per_minute=60)
@cached_request(ttl=3600) # 1 hour cache
async def get_new_data(ticker: str) -> Optional[Dict[str, Any]]:
"""Get data from new source"""
try:
async with aiohttp.ClientSession() as session:
url = f"https://api.newsource.com/data/{ticker}"
async with session.get(url) as response:
if response.status == 200:
return await response.json()
return None
except Exception as e:
print(f"Error fetching new data for {ticker}: {e}")
return None
# server.py - Add endpoint
@server.tool
async def get_new_endpoint(ticker: str) -> str:
"""Get new data endpoint"""
data = await new_source.get_new_data(ticker)
return format_data(data) if data else f"No data for {ticker}"
建筑与出版
# Build package
uv build
# Check package
uv run twine check dist/*
# Publish to PyPI (maintainers only)
uv run twine upload dist/*
🤝 贡献
- 分叉存储库
- 创建要素分支(
git checkout -b feature/amazing-feature) - 提交您的更改(
git commit -m 'Add amazing feature') - 推到分支(
git push origin feature/amazing-feature) - 打开拉取请求
📄 许可证
此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。
🔗 与人工智能交易系统集成
此MCP服务器旨在与任何定量交易系统或AI代理无缝协作:
易于集成
- 零配置:使用任何MCP兼容系统进行简单设置
- 优化的缓存:缓存TTL设置,针对交易工作流程进行了优化
- 错误处理:即使某些数据源发生故障,优雅的降级也能确保系统继续工作
支持的交易机会类型
增强的端点支持检测多种机会类型:
- 短挤压分析:FINRA短交易量+SEC文件+社会趋势
- 突破/动量检测:新闻情绪+谷歌趋势+成交量分析
- 收益发挥:收益日历+每股收益惊喜+期权数据
- 生物技术/FDA活动:SEC 8-K文件+新闻头条+社会动力
- 并购/谣言检测:新闻分析+社会趋势+交易量激增
- 稀释/ATM检测:美国证券交易委员会S-3/424号文件+数量分析
- 差距消退机会:新闻情绪+交易量模式+期权数据
在交易系统中的使用
# Example: Integration with AI trading agents
from your_trading_system import TradingAgent
agent = TradingAgent()
agent.add_financial_data_source("iso-financial-mcp")
# Agent can now use IsoFinancial-MCP for:
# - SEC filing analysis for dilution detection
# - FINRA short volume for squeeze signals
# - Earnings calendar for earnings plays
# - News sentiment for momentum analysis
# - Google Trends for social validation
🙏 致谢
- 雅虎财经 获取全面的市场数据
- 第二埃德加 用于官方备案数据
- 美国金融业监管局 用于短体积透明度
- 谷歌趋势 搜索量洞察
- FastMCP 对于MCP框架
- UV包装管理器 用于现代Python打包
- 开源社区 用于贡献和反馈
📞 支持
📊 数据来源和可靠性
数据源概述
| 数据源 | 端点 | 更新频率 | 缓存TTL | 回退源 | 可靠性 |
|---|---|---|---|---|---|
| 雅虎财经 | 市场数据、财务、期权 | 实时到每日 | 5分钟-24小时 | 无 | 高 |
| SEC EDGAR | 官方文件 | 实时 | 6小时 | RSS Feed、XBRL API | 非常高 |
| FINRA | 交易量短 | 每日 | 24小时 | N/A | 高 |
| 谷歌趋势 | 搜索量 | 每日 | 24小时 | DuckDuckGo,SerpAPI | 中高 |
| 收益 | 日历数据 | 每日 | 24小时 | 纳斯达克,阿尔法Vantage | 高点 |
| RSS源 | 新闻标题 | 每小时 | 2小时 | N/A | 中等 |
🛡️ 可靠性特性(v0.4.0+)
多源回退
当主数据源发生故障时,自动回退到其他数据源:
- 行情走势:EDGAR API→ RSS源→ API XBRL→ 过期缓存
- 谷歌趋势:SerpAPI(如果已配置)→ 鸭鸭搜→ PyTrends→ 过期缓存
- 收益:雅虎财经→ 纳斯达克API→ 阿尔法Vantage→ 估计
智能高速缓存
具有过时数据回退功能的两级缓存系统:
- 内存缓存:快速内存缓存(1小时TTL,1000个项目)
- 磁盘缓存:持久磁盘缓存(7天TTL,最大500MB)
- 陈旧的退路:当所有源都失败时,返回过期的缓存
- 自动清理:旧缓存条目将自动清除
自适应速率限制
根据API行为进行调整的智能速率限制:
- 错误检测:监视429个速率限制错误
- 指数退避:10秒→ 20s → 40s延迟与抖动
- 慢速模式:错误率>50%时自动激活
- 每源限制:每个API的独立速率限制
健康监测
持续监测数据源可靠性:
- 成功率跟踪:监视每个源的最后100个请求
- 自动故障切换:跳过成功率<30%的来源
- 指标记录:用于调试和分析的JSONL日志
- 健康状况API:以编程方式查询源运行状况
故障弱化
即使源发生故障,系统也会继续工作:
- 部分数据:返回包含错误详细信息的可用数据
- 错误分类:区分暂时错误和永久错误
- 可操作消息:建议解决常见问题
- 无无声故障:记录并报告所有错误
配置
默认配置(无需设置)
该系统使用合理的默认值即可开箱即用。无需配置!
自定义配置(可选)
对于高级用户,创建 ~/.iso_financial_mcp/config/datasources.yaml:
# Example: Adjust rate limiting for Google Trends
trends:
sources:
- name: pytrends_direct
enabled: true
rate_limit: 0.1 # Slower: 1 req / 10 sec
adaptive_rate_limit:
enabled: true
error_threshold: 0.5
slow_mode_delay: 10
# Example: Enable optional sources with API keys
earnings:
sources:
- name: alpha_vantage
enabled: true
api_key_env: ALPHA_VANTAGE_KEY # Set via environment variable
看 config/datasources.yaml.example 查看完整配置选项。
环境变量
用于增强数据源的可选API密钥:
# Alpha Vantage (for additional earnings data)
export ALPHA_VANTAGE_KEY="your_key_here"
# SerpAPI (for Google Trends fallback)
export SERPAPI_KEY="your_key_here"
数据质量特征
- 重复检测:新闻标题在各个来源中都是重复的
- 数据验证:所有端点的输入验证和净化
- 错误恢复:数据源不可用时的优雅回退
- 缓存预热:频繁访问的数据已预缓存
- 速率限制:尊重API提供商的限制,以确保一致的访问
- 数据融合:合并来自多个来源的数据以获得更好的覆盖率
- 自动重试:具有指数回退的智能重试
🔧 MCP配置工具
v0.4.0中的新功能:直接通过MCP工具配置服务器,无需编辑配置文件。
可用配置工具
configure_api_key
在运行时为可选数据源配置API密钥。
# Configure Alpha Vantage for enhanced earnings data
await configure_api_key("alpha_vantage", "YOUR_API_KEY_HERE")
# Configure SerpAPI for Google Trends fallback
await configure_api_key("serpapi", "YOUR_API_KEY_HERE")
特征:
- 保存前验证API密钥
- 为未来的会话保留配置
- 提供清晰的成功/错误消息
- 支持:
alpha_vantage,serpapi
get_configuration
使用屏蔽的API键查看当前配置。
# Get current configuration
await get_configuration()
退货:
- API密钥(为安全起见,已屏蔽:
...XXXX) - 缓存设置(TTL,最大大小)
- 速率限制配置
- 已启用的数据源
list_data_sources
列出所有可用的数据源及其状态。
# List all data sources
await list_data_sources()
显示:
- ✅ 已启用的源(不需要API密钥)
- ⚠️ 禁用的源(需要API密钥,但未配置)
- 源描述和功能
配置方法
服务器支持三种优先级顺序的配置方法:
- MCP工具 (最高优先级)-通过运行时配置
configure_api_key - 环境变量 -设置
ALPHA_VANTAGE_KEY,SERPAPI_KEY等等。 - YAML文件 (最低优先级)-
~/.iso_financial_mcp/config/datasources.yaml
看 配置指南 以获取详细示例。
🏥 MCP健康检查工具
v0.4.0中的新功能:通过MCP工具监控数据源运行状况并诊断问题。
可用的健康检查工具
get_health_status
获取所有数据源的全面健康状态。
# Check health of all sources
await get_health_status()
每个来源的回报:
- ✅ 状态:健康/⚠️ 退化/❌ 不健康的
- 成功率(最近100个请求)
- 平均延迟(毫秒)
- 已处理的请求总数
- 上次成功请求时间戳
- 最近的错误消息(如果有的话)
输出示例:
🏥 Data Sources Health Status:
✅ Yahoo Finance
Status: healthy
Success Rate: 98.5%
Avg Latency: 245ms
Total Requests: 1,234
Last Success: 2025-01-15 14:32:10
⚠️ Google Trends
Status: degraded
Success Rate: 65.2%
Avg Latency: 1,850ms
Total Requests: 456
Recent Errors: Rate limit exceeded, Timeout
test_data_source
使用示例请求测试特定数据源。
# Test SEC filings source
await test_data_source("sec", "AAPL")
# Test Google Trends source
await test_data_source("trends", "Tesla")
# Test earnings calendar
await test_data_source("earnings", "NVDA")
特征:
- 测量响应时间
- 显示数据预览(前500个字符)
- 提供详细的错误消息
- 使用真实的API调用进行测试
支持的来源:
sec-SEC EDGAR文件trends-谷歌趋势earnings-收益日历finra-FINRA短交易量news-新闻头条
输出示例:
🧪 Testing sec with ticker AAPL...
✅ Test successful!
⏱️ Response time: 1.23s
📊 Data preview:
SEC Filings for AAPL (Last 30 days)
Found 3 filings:
1. 8-K - 2025-01-10 - Material Event...
用例
故障排除:
# Check if a source is having issues
status = await get_health_status()
# Test specific source that's failing
result = await test_data_source("trends", "AAPL")
监控:
# Regular health checks in production
status = await get_health_status()
# Alert if any source has <80% success rate
验证:
# Verify API key configuration
await configure_api_key("alpha_vantage", "YOUR_KEY")
await test_data_source("earnings", "AAPL")
📚 文档
有关该系统的详细信息,请参阅我们的综合文档:
⚠️ 重要免责声明
财务数据免责声明:本软件仅为教育和研究目的提供财务数据。数据来源于公共API,可能包含错误、延迟或不准确之处。在做出投资决策之前,始终独立验证数据。
交易风险预警:交易涉及巨大的损失风险,并不适合所有投资者。过去的表现并不能保证未来的结果。该系统提供的定量分析不应被视为财务建议。
数据准确性:在我们努力提高准确性的同时,财务数据可能会受到来源提供商的修订、延迟和错误的影响。关键的交易决策应始终基于官方消息来源和专业的财务建议。
无担保:本软件按“原样”提供,不提供任何形式的保修。作者和贡献者不对使用本软件造成的任何经济损失负责。