129 lines
3.6 KiB
Python
129 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Centralized data source configuration and fallback ordering."""
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import Literal, Optional
|
|
|
|
DataSource = Literal["finnhub", "financial_datasets", "yfinance", "local_csv"]
|
|
_KNOWN_SOURCES: tuple[DataSource, ...] = (
|
|
"finnhub",
|
|
"financial_datasets",
|
|
"yfinance",
|
|
"local_csv",
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class DataSourceConfig:
|
|
"""Resolved data source configuration."""
|
|
|
|
source: DataSource
|
|
api_key: str
|
|
sources: list[DataSource]
|
|
|
|
|
|
# Module-level cache for the resolved configuration
|
|
_config_cache: Optional[DataSourceConfig] = None
|
|
|
|
|
|
def _parse_enabled_sources() -> list[DataSource]:
|
|
"""Parse optional enabled source allowlist from the environment."""
|
|
raw_value = os.getenv("ENABLED_DATA_SOURCES", "").strip().lower()
|
|
if not raw_value:
|
|
return []
|
|
|
|
enabled: list[DataSource] = []
|
|
for item in raw_value.split(","):
|
|
candidate = item.strip()
|
|
if not candidate or candidate not in _KNOWN_SOURCES:
|
|
continue
|
|
if candidate not in enabled:
|
|
enabled.append(candidate)
|
|
return enabled
|
|
|
|
|
|
def _ordered_sources() -> list[DataSource]:
|
|
"""Resolve source preference and available fallbacks."""
|
|
preferred = os.getenv("FIN_DATA_SOURCE", "").strip().lower()
|
|
finnhub_key = os.getenv("FINNHUB_API_KEY", "").strip()
|
|
fd_key = os.getenv("FINANCIAL_DATASETS_API_KEY", "").strip()
|
|
enabled_sources = _parse_enabled_sources()
|
|
wants_yfinance = preferred == "yfinance" or "yfinance" in enabled_sources
|
|
|
|
available: list[DataSource] = []
|
|
if finnhub_key:
|
|
available.append("finnhub")
|
|
if fd_key:
|
|
available.append("financial_datasets")
|
|
if wants_yfinance:
|
|
available.append("yfinance")
|
|
available.append("local_csv")
|
|
|
|
if enabled_sources:
|
|
filtered = [source for source in enabled_sources if source in available]
|
|
if filtered:
|
|
available = filtered
|
|
|
|
if preferred in available:
|
|
ordered = [preferred]
|
|
ordered.extend(source for source in available if source != preferred)
|
|
return ordered
|
|
return available
|
|
|
|
|
|
def _resolve_config() -> DataSourceConfig:
|
|
"""
|
|
Resolve data source configuration based on available API keys.
|
|
|
|
The effective source should always match the first item in the resolved
|
|
ordered source list.
|
|
"""
|
|
sources = _ordered_sources()
|
|
source = sources[0] if sources else "local_csv"
|
|
|
|
api_key = ""
|
|
if source == "finnhub":
|
|
api_key = os.getenv("FINNHUB_API_KEY", "").strip()
|
|
elif source == "financial_datasets":
|
|
api_key = os.getenv("FINANCIAL_DATASETS_API_KEY", "").strip()
|
|
|
|
return DataSourceConfig(source=source, api_key=api_key, sources=sources)
|
|
|
|
|
|
def get_config() -> DataSourceConfig:
|
|
"""
|
|
Get the resolved data source configuration (cached).
|
|
|
|
Returns:
|
|
DataSourceConfig with source and api_key
|
|
|
|
Raises:
|
|
ValueError: If no API key is configured
|
|
"""
|
|
global _config_cache
|
|
if _config_cache is None:
|
|
_config_cache = _resolve_config()
|
|
return _config_cache
|
|
|
|
|
|
def get_data_source() -> DataSource:
|
|
"""Get the configured data source name."""
|
|
return get_config().source
|
|
|
|
|
|
def get_data_sources() -> list[DataSource]:
|
|
"""Get preferred source ordering including fallbacks."""
|
|
return get_config().sources
|
|
|
|
|
|
def get_api_key() -> str:
|
|
"""Get the API key for the configured data source."""
|
|
return get_config().api_key
|
|
|
|
|
|
def reset_config() -> None:
|
|
"""Reset the cached configuration (useful for testing)."""
|
|
global _config_cache
|
|
_config_cache = None
|