多 Provider 路由与 Agent 决策

学习目标

理解 OpenBB 的 Provider Interface 如何动态聚合 35+ 数据源，以及 AI Agent 如何利用 Provider 选择能力做出最优数据源决策。

---

项目实践

Provider Interface 的聚合机制

ProviderInterface 是一个单例，在运行时扫描所有已注册的 Provider，动态构建：

openbb_core/app/provider_interface.py

class ProviderInterface(metaclass=SingletonMeta):
    def __init__(self):
        self._registry_map = RegistryMap()  # 扫描所有 Provider
        self._map = self._registry_map.standard_extra        # 模型 → Provider 映射
        self._model_providers_map = self._generate_model_providers_dc(self._map)
        self._params = self._generate_params_dc(self._map)   # 动态生成 dataclass
        self._data = self._generate_data_dc(self._map)

动态生成的 Dataclass：

# ProviderInterface 在运行时生成类似这样的 dataclass：
@dataclass
class EquityHistorical(StandardParams):
    symbol: str = Query(...)         # 所有 Provider 都有的标准字段
    start_date: date = Query(...)
    end_date: date = Query(...)

@dataclass
class EquityHistorical(ExtraParams):
    interval: str = Query(..., title="yfinance")      # yfinance 独有
    adjustment: str = Query(..., title="yfinance")    # yfinance 独有
    page: int = Query(..., title="fmp")               # fmp 独有

Provider 选择注入

在 Router 的 @command 装饰器中，ProviderChoices 被作为 FastAPI 依赖注入：

@dataclass
class EquityHistorical(ProviderChoices):
    provider: Literal["yfinance", "fmp", "alpha_vantage", "polygon", "intrinio", ...]

用户/Agent 调用时：

# 指定 Provider
obb.equity.price.historical("AAPL", provider="yfinance")

# 不指定时，使用默认（通常是第一个可用的）
obb.equity.price.historical("AAPL")

QueryExecutor 的执行路径

openbb_core/provider/query_executor.py

class QueryExecutor:
    async def execute(self, provider_name, model_name, params, credentials, **kwargs):
        provider = self.get_provider(provider_name)       # 从注册表获取 Provider
        fetcher = self.get_fetcher(provider, model_name)  # 获取对应 Fetcher
        filtered_credentials = self.filter_credentials(credentials, provider, fetcher.require_credentials)
        return await fetcher.fetch_data(params, filtered_credentials, **kwargs)

Agent 决策策略

AI Agent 可以根据以下因素选择 Provider：

因素	选择策略	示例
数据覆盖	选择支持该标的的 Provider	美股 → yfinance，A股 → 特定 Provider
API 配额	选择有剩余配额的 Provider	FMP 配额用尽 → 切换 Benzinga
字段需求	选择提供所需字段的 Provider	需要 split_ratio → yfinance
响应速度	选择已知延迟最低的 Provider	实时数据 → 低延迟 Provider
认证可用性	选择已配置凭据的 Provider	未配置 FMP key → 用 yfinance

Agent 自动切换示例

# Agent 可以实现的自动切换逻辑
def smart_fetch(symbol, model, preferred_providers=None):
    providers = preferred_providers or ["yfinance", "fmp", "alpha_vantage"]
    for provider in providers:
        try:
            result = getattr(obb, model)(symbol, provider=provider)
            if result.results:
                return result, provider
        except Exception:
            continue
    raise ValueError(f"No provider returned data for {symbol}")

---

问题与规避

Provider 字段不完全一致

• 即使有 Standard Models，不同 Provider 的 ExtraParams 可能差异很大
• 对策：Agent 应先通过 ProviderInterface.map 查询每个 Provider 支持的字段

凭据缺失

• 某些 Provider 需要 API Key，未配置时会抛出 OpenBBError
• 对策：Agent 应先检查 ProviderInterface.credentials，选择已配置的 Provider

---

设计取舍

动态类型聚合 vs 静态定义

• OpenBB 选择在运行时动态生成 dataclass，而非手动定义每个组合
• 优势：新增 Provider 自动聚合，无需修改核心代码
• 代价：IDE 类型提示受限，调试时需要理解动态生成逻辑

Literal 类型做 Provider 选择

• 使用 Literal["yf", "fmp", ...] 而非字符串
• 优势：类型安全，IDE 自动补全
• 代价：Provider 数量变化时需要重新生成类型

---

参考来源

• 源码验证: openbb_core/app/provider_interface.py — ProviderInterface 完整实现
• 源码验证: openbb_core/provider/query_executor.py — QueryExecutor
• 源码验证: openbb_core/provider/registry_map.py — RegistryMap