Python 75.9%
TeX 24%
1"""2================================================================3Auteur : Simon-Pierre Boucher4Contact : contact@spboucher.ai5Projet : Prévision de volatilité réalisée multi-actifs6 (HAR-RV vs GARCH vs Machine Learning)7Fichier : api_client.py8Description : Client réutilisable pour l'API HF Market Data —9 rate limiting, retry avec backoff exponentiel,10 pagination et cache local parquet.11================================================================12"""1314from __future__ import annotations1516import io17import logging18import time19from pathlib import Path20from typing import Any2122import pandas as pd23import requests2425from . import config2627logger = logging.getLogger(__name__)2829_RETRYABLE = {429, 500, 502, 503, 504}303132class HFMarketDataClient:33 """Client for https://www.hfmarketdata.io with caching and retries.3435 Parameters36 ----------37 base_url : str, optional38 API root; defaults to the value in ``config.yaml``.39 cache_dir : pathlib.Path, optional40 Directory for the local parquet cache; defaults to ``data/raw``.4142 Notes43 -----44 Bars are fetched in ascending order and paginated by advancing the45 ``start`` parameter one minute past the last received timestamp until46 an empty page is returned — this is robust to any server-side row cap.47 """4849 def __init__(self, base_url: str | None = None, cache_dir: Path | None = None) -> None:50 cfg = config.load_config()["api"]51 self.base_url = (base_url or cfg["base_url"]).rstrip("/")52 self.cache_dir = Path(cache_dir) if cache_dir else config.path("raw")53 self.min_interval = 1.0 / float(cfg.get("rate_limit_per_sec", 4))54 self.max_retries = int(cfg.get("max_retries", 5))55 self.backoff_base = float(cfg.get("backoff_base_sec", 1.0))56 self.timeout = float(cfg.get("timeout_sec", 60))57 self.page_limit = int(cfg.get("page_limit", 50_000))58 self._last_request_ts = 0.059 self._session = requests.Session()6061 # ------------------------------------------------------------- low level62 def _throttle(self) -> None:63 """Sleep as needed to respect the configured request rate."""64 wait = self._last_request_ts + self.min_interval - time.monotonic()65 if wait > 0:66 time.sleep(wait)67 self._last_request_ts = time.monotonic()6869 def _get(self, path: str, params: dict[str, Any] | None = None) -> requests.Response:70 """Issue a GET with throttling and exponential-backoff retries.7172 Parameters73 ----------74 path : str75 Path relative to the API root (e.g. ``/v1/status``).76 params : dict, optional77 Query-string parameters.7879 Returns80 -------81 requests.Response82 The successful response (status code < 400).8384 Raises85 ------86 requests.HTTPError87 If the request keeps failing after ``max_retries`` attempts.88 """89 url = f"{self.base_url}{path}"90 last_exc: Exception | None = None91 for attempt in range(self.max_retries + 1):92 self._throttle()93 try:94 resp = self._session.get(url, params=params, timeout=self.timeout)95 if resp.status_code in _RETRYABLE:96 raise requests.HTTPError(f"HTTP {resp.status_code}", response=resp)97 resp.raise_for_status()98 return resp99 except (requests.ConnectionError, requests.Timeout, requests.HTTPError) as exc:100 resp_obj = getattr(exc, "response", None)101 if resp_obj is not None and resp_obj.status_code not in _RETRYABLE:102 raise103 last_exc = exc104 delay = self.backoff_base * (2**attempt)105 logger.warning(106 "GET %s failed (%s), retry %d/%d in %.1fs",107 path, exc, attempt + 1, self.max_retries, delay,108 )109 time.sleep(delay)110 raise requests.HTTPError(f"GET {url} failed after {self.max_retries} retries") from last_exc111112 def get_json(self, path: str, params: dict[str, Any] | None = None) -> Any:113 """GET a JSON endpoint and return the decoded payload."""114 return self._get(path, params).json()115116 # ------------------------------------------------------------- endpoints117 def status(self) -> dict[str, Any]:118 """Return the ``/v1/status`` dataset inventory."""119 return self.get_json("/v1/status")120121 def tickers(self, asset: str, search: str | None = None, limit: int = 100) -> list[str]:122 """List tickers available for an asset class.123124 Parameters125 ----------126 asset : str127 API asset class (``stock``, ``etf``, ``futures``, ``crypto``,128 ``index``, ``fx``).129 search : str, optional130 Substring filter applied server-side.131 limit : int132 Maximum number of tickers returned.133 """134 params: dict[str, Any] = {"limit": limit}135 if search:136 params["search"] = search137 payload = self.get_json(f"/v1/{asset}/tickers", params)138 return payload.get("tickers", payload)139140 def fetch_bars(141 self,142 asset: str,143 ticker: str,144 timeframe: str = "1min",145 adjustment: str | None = None,146 start: str | None = None,147 end: str | None = None,148 ) -> pd.DataFrame:149 """Download OHLCV bars, transparently handling pagination.150151 Parameters152 ----------153 asset, ticker : str154 Instrument identification.155 timeframe : str156 ``1min``, ``5min``, ``30min``, ``1hour`` or ``1day``.157 adjustment : str, optional158 Price-adjustment scheme; defaults to the class setting in159 ``config.yaml``.160 start, end : str, optional161 Inclusive date bounds (``YYYY-MM-DD`` or full datetime).162163 Returns164 -------165 pandas.DataFrame166 Columns ``datetime`` (naive exchange-local timestamps),167 ``open``, ``high``, ``low``, ``close``, ``volume``; sorted,168 duplicate timestamps dropped.169 """170 adjustment = adjustment or config.adjustment_for(asset)171 frames: list[pd.DataFrame] = []172 cursor = start173 n_pages = 0174 while True:175 params: dict[str, Any] = {176 "timeframe": timeframe,177 "adjustment": adjustment,178 "order": "asc",179 "limit": self.page_limit,180 "format": "csv",181 }182 if cursor:183 params["start"] = cursor184 if end:185 params["end"] = end186 resp = self._get(f"/v1/bars/{asset}/{ticker}", params)187 page = pd.read_csv(io.StringIO(resp.text)) if resp.text.strip() else pd.DataFrame()188 if page.empty:189 break190 page["datetime"] = pd.to_datetime(page["datetime"])191 frames.append(page)192 n_pages += 1193 last_dt = page["datetime"].iloc[-1]194 logger.debug("%s/%s %s page %d: %d rows, up to %s",195 asset, ticker, timeframe, n_pages, len(page), last_dt)196 if len(page) < self.page_limit:197 break198 cursor = (last_dt + pd.Timedelta(minutes=1)).strftime("%Y-%m-%d %H:%M:%S")199 if not frames:200 logger.warning("%s/%s %s: no data returned", asset, ticker, timeframe)201 return pd.DataFrame(columns=["datetime", "open", "high", "low", "close", "volume"])202 df = pd.concat(frames, ignore_index=True)203 df = (204 df.drop(columns=["ticker"], errors="ignore")205 .drop_duplicates(subset="datetime")206 .sort_values("datetime")207 .reset_index(drop=True)208 )209 if "volume" not in df.columns: # some index series carry no volume210 df["volume"] = float("nan")211 logger.info("%s/%s %s: %d bars (%s -> %s)", asset, ticker, timeframe,212 len(df), df["datetime"].iloc[0], df["datetime"].iloc[-1])213 return df214215 # ------------------------------------------------------------- cache216 def _cache_file(self, asset: str, ticker: str, timeframe: str) -> Path:217 safe = ticker.replace("/", "-")218 return self.cache_dir / f"{asset}_{safe}_{timeframe}.parquet"219220 def get_bars(221 self,222 asset: str,223 ticker: str,224 timeframe: str = "1min",225 start: str | None = None,226 end: str | None = None,227 refresh: bool = False,228 ) -> pd.DataFrame:229 """Return bars from the local parquet cache, downloading if absent.230231 Parameters232 ----------233 asset, ticker, timeframe : str234 Instrument identification and bar frequency.235 start, end : str, optional236 Date bounds used when the series must be downloaded; when the237 cache is hit the full cached range is returned (filter at the238 call site if needed).239 refresh : bool240 Force a re-download even when a cache file exists.241242 Returns243 -------244 pandas.DataFrame245 Same layout as :meth:`fetch_bars`.246 """247 f = self._cache_file(asset, ticker, timeframe)248 if f.exists() and not refresh:249 logger.debug("cache hit: %s", f.name)250 return pd.read_parquet(f)251 df = self.fetch_bars(asset, ticker, timeframe, start=start, end=end)252 if not df.empty:253 f.parent.mkdir(parents=True, exist_ok=True)254 df.to_parquet(f, index=False)255 logger.info("cached %s (%.1f MB)", f.name, f.stat().st_size / 1e6)256 return df257