| """Base provider interface for data sources""" |
| from __future__ import annotations |
| from abc import ABC, abstractmethod |
| from typing import List, Optional |
| from datetime import datetime |
| import time |
| import httpx |
| import sys |
| import os |
| sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) |
|
|
| from core.models import OHLCV, Price, ProviderHealth |
|
|
|
|
| class CircuitBreaker: |
| """Circuit breaker for provider failures""" |
|
|
| def __init__(self, threshold: int = 5, timeout: int = 60): |
| self.threshold = threshold |
| self.timeout = timeout |
| self.failures = 0 |
| self.last_failure_time: Optional[float] = None |
| self.is_open = False |
|
|
| def record_success(self): |
| """Record successful request""" |
| self.failures = 0 |
| self.is_open = False |
|
|
| def record_failure(self): |
| """Record failed request""" |
| self.failures += 1 |
| self.last_failure_time = time.time() |
|
|
| if self.failures >= self.threshold: |
| self.is_open = True |
|
|
| def can_attempt(self) -> bool: |
| """Check if we can attempt a request""" |
| if not self.is_open: |
| return True |
|
|
| |
| if self.last_failure_time: |
| elapsed = time.time() - self.last_failure_time |
| if elapsed >= self.timeout: |
| self.is_open = False |
| self.failures = 0 |
| return True |
|
|
| return False |
|
|
|
|
| class BaseProvider(ABC): |
| """Base class for all data providers""" |
|
|
| def __init__(self, name: str, base_url: str, timeout: int = 10): |
| self.name = name |
| self.base_url = base_url |
| self.timeout = timeout |
| self.circuit_breaker = CircuitBreaker() |
| self.last_latency: Optional[int] = None |
| self.last_check: Optional[datetime] = None |
| self.last_error: Optional[str] = None |
| self.client: Optional[httpx.AsyncClient] = None |
|
|
| async def get_client(self) -> httpx.AsyncClient: |
| """Get or create HTTP client""" |
| if self.client is None: |
| self.client = httpx.AsyncClient(timeout=self.timeout) |
| return self.client |
|
|
| async def close(self): |
| """Close HTTP client""" |
| if self.client: |
| await self.client.aclose() |
| self.client = None |
|
|
| async def _make_request(self, url: str, params: Optional[dict] = None) -> dict: |
| """Make HTTP request with timing and error handling""" |
| if not self.circuit_breaker.can_attempt(): |
| raise Exception(f"Circuit breaker open for {self.name}") |
|
|
| client = await self.get_client() |
| start_time = time.time() |
|
|
| try: |
| response = await client.get(url, params=params) |
| response.raise_for_status() |
|
|
| self.last_latency = int((time.time() - start_time) * 1000) |
| self.last_check = datetime.now() |
| self.last_error = None |
| self.circuit_breaker.record_success() |
|
|
| return response.json() |
|
|
| except Exception as e: |
| self.last_error = str(e) |
| self.circuit_breaker.record_failure() |
| raise |
|
|
| @abstractmethod |
| async def fetch_ohlcv(self, symbol: str, interval: str, limit: int) -> List[OHLCV]: |
| """Fetch OHLCV data""" |
| pass |
|
|
| @abstractmethod |
| async def fetch_prices(self, symbols: List[str]) -> List[Price]: |
| """Fetch current prices""" |
| pass |
|
|
| async def get_health(self) -> ProviderHealth: |
| """Get provider health status""" |
| if self.circuit_breaker.is_open: |
| status = "offline" |
| elif self.last_error: |
| status = "degraded" |
| else: |
| status = "online" |
|
|
| return ProviderHealth( |
| name=self.name, |
| status=status, |
| latency=self.last_latency, |
| lastCheck=self.last_check.isoformat() if self.last_check else datetime.now().isoformat(), |
| errorMessage=self.last_error |
| ) |
|
|
