Update data_fetcher.py

data_fetcher.py

@@ -1,15 +1,24 @@
-import requests
-import pandas as pd
-import numpy as np
 import time
 import logging
-from typing import Optional, List
-from config import OKX_BASE_URL, TIMEFRAME, CANDLE_LIMIT
+from typing import Dict, List, Optional
+
+import requests
+import numpy as np
+import pandas as pd
+
+from config import (
+    OKX_BASE_URL,
+    TIMEFRAME,
+    CANDLE_LIMIT,
+    REQUEST_DELAY,
+    REQUEST_RETRIES,
+    REQUEST_TIMEOUT,
+)
 
 logger = logging.getLogger(__name__)
 
 
-def _bar_to_series(bar: list) -> dict:
+def _parse_candle(bar: list) -> dict:
     return {
         "timestamp": pd.to_datetime(int(bar[0]), unit="ms", utc=True),
         "open": float(bar[1]),
@@ -25,29 +34,38 @@ def fetch_ohlcv(
     symbol: str,
     timeframe: str = TIMEFRAME,
     limit: int = CANDLE_LIMIT,
-    retries: int = 3,
-    sleep_between: float = 0.2,
 ) -> Optional[pd.DataFrame]:
     url = f"{OKX_BASE_URL}/api/v5/market/candles"
     params = {"instId": symbol, "bar": timeframe, "limit": str(limit)}
 
-    for attempt in range(retries):
+    for attempt in range(REQUEST_RETRIES):
         try:
-            resp = requests.get(url, params=params, timeout=10)
+            resp = requests.get(url, params=params, timeout=REQUEST_TIMEOUT)
             resp.raise_for_status()
-            data = resp.json()
-            if data.get("code") != "0" or not data.get("data"):
-                logger.warning(f"{symbol}: API error or empty data: {data.get('msg')}")
+            payload = resp.json()
+
+            if payload.get("code") != "0":
+                logger.warning(f"{symbol}: API error — {payload.get('msg', 'unknown')}")
                 return None
-            records = [_bar_to_series(bar) for bar in reversed(data["data"])]
-            df = pd.DataFrame(records)
-            df.set_index("timestamp", inplace=True)
-            df.sort_index(inplace=True)
-            time.sleep(sleep_between)
+
+            raw = payload.get("data")
+            if not raw:
+                logger.warning(f"{symbol}: empty data returned")
+                return None
+
+            records = [_parse_candle(bar) for bar in reversed(raw)]
+            df = pd.DataFrame(records).set_index("timestamp").sort_index()
+            time.sleep(REQUEST_DELAY)
             return df
-        except requests.RequestException as e:
-            logger.warning(f"{symbol}: Attempt {attempt+1} failed: {e}")
-            time.sleep(1.0 * (attempt + 1))
+
+        except requests.exceptions.Timeout:
+            logger.warning(f"{symbol}: timeout on attempt {attempt + 1}")
+        except requests.exceptions.RequestException as exc:
+            logger.warning(f"{symbol}: request error on attempt {attempt + 1}: {exc}")
+
+        time.sleep(REQUEST_DELAY * (attempt + 2))
+
+    logger.error(f"{symbol}: all {REQUEST_RETRIES} fetch attempts failed")
     return None
 
 
@@ -55,19 +73,18 @@ def fetch_instruments(inst_type: str = "SPOT") -> List[str]:
     url = f"{OKX_BASE_URL}/api/v5/public/instruments"
     params = {"instType": inst_type}
     try:
-        resp = requests.get(url, params=params, timeout=10)
+        resp = requests.get(url, params=params, timeout=REQUEST_TIMEOUT)
         resp.raise_for_status()
-        data = resp.json()
-        if data.get("code") != "0":
+        payload = resp.json()
+        if payload.get("code") != "0":
             return []
-        symbols = [
+        return [
             item["instId"]
-            for item in data["data"]
+            for item in payload.get("data", [])
             if item.get("quoteCcy") == "USDT" and item.get("state") == "live"
         ]
-        return symbols
-    except requests.RequestException as e:
-        logger.error(f"Failed to fetch instruments: {e}")
+    except requests.exceptions.RequestException as exc:
+        logger.error(f"Failed to fetch instruments: {exc}")
         return []
 
 
@@ -75,17 +92,22 @@ def fetch_multiple(
     symbols: List[str],
     timeframe: str = TIMEFRAME,
     limit: int = CANDLE_LIMIT,
-    max_symbols: Optional[int] = None,
-) -> dict:
-    if max_symbols:
-        symbols = symbols[:max_symbols]
-    results = {}
+    min_bars: int = 50,
+    progress_callback=None,
+) -> Dict[str, pd.DataFrame]:
+    results: Dict[str, pd.DataFrame] = {}
     total = len(symbols)
+
     for i, sym in enumerate(symbols, 1):
-        logger.info(f"Fetching {sym} ({i}/{total})")
+        if progress_callback:
+            progress_callback(i, total, sym)
+        else:
+            logger.info(f"Fetching {sym} ({i}/{total})")
+
         df = fetch_ohlcv(sym, timeframe=timeframe, limit=limit)
-        if df is not None and len(df) >= 50:
+        if df is not None and len(df) >= min_bars:
             results[sym] = df
         else:
-            logger.warning(f"{sym}: insufficient data, skipping")
+            logger.warning(f"{sym}: skipped (insufficient data)")
+
     return results