fix(loaders): bound ccxt + okx fetch (timeout, retry, budget) (HKUDS#121)

* fix(loaders): bound ccxt fetch timeout + retry

ccxt_loader set no request timeout and paginated fetch_ohlcv in an
unbounded loop with no retry budget, so a transient disconnect hung
get_market_data for 10+ minutes (P12-b — it stalled a crypto E2E).

Set an explicit per-call timeout on the exchange, retry the transient
ccxt.NetworkError family with bounded backoff, and enforce a hard
wall-clock budget that raises a clear TimeoutError instead of
hanging. Non-network errors (e.g. bad symbol) are not retried; the
happy path still issues one call per page (no behavior change).
okx.py has the same shape — noted as a follow-up, out of scope here.

(cherry picked from commit ef934fbd70209d74a110e81757dbe322a2b717ab)

* fix(loaders): bound okx fetch retry + budget

okx.py set a per-request timeout but had no retry and no wall-clock
budget, so a transient blip dropped the whole symbol via the outer
silent skip, and a slow keyless tier could stall roughly
max_pages * timeout (P12-b parity with the ccxt loader).

Retry the transient requests.RequestException family with bounded
backoff and enforce a hard wall-clock budget that raises a clear
TimeoutError; non-network errors are not retried; the happy path
still issues exactly one request per page (no behavior change).

(cherry picked from commit e5030bf3edc3e44cc27003b3ac23e9e4fefc3c97)

Author: Teerapat-Vatpitak

agent/backtest/loaders/ccxt_loader.py

@@ -9,6 +9,7 @@
 
 import logging
 import os
+import time
 from typing import Dict, List, Optional
 
 import pandas as pd
@@ -23,6 +24,15 @@
     "1H": "1h", "4H": "4h", "1D": "1d",
 }
 
+# P12-b: ccxt had no request timeout and an unbounded paginated fetch with
+# no retry budget, so a transient disconnect hung get_market_data for 10+
+# minutes. Cap each HTTP call, bound transient retries, and enforce a hard
+# wall-clock budget so the fetch fails fast instead of hanging.
+_CCXT_TIMEOUT_MS = int(os.getenv("CCXT_TIMEOUT_MS", "15000"))
+_CCXT_FETCH_BUDGET_S = float(os.getenv("CCXT_FETCH_BUDGET_S", "60"))
+_CCXT_MAX_RETRIES = 3
+_CCXT_BACKOFF = (0.5, 1.5, 4.0)  # seconds; len == _CCXT_MAX_RETRIES
+
 
 @register
 class DataLoader:
@@ -51,7 +61,7 @@ def _get_exchange(self):
         if exchange_cls is None:
             logger.warning("Unknown CCXT exchange %s, falling back to binance", exchange_id)
             exchange_cls = ccxt.binance
-        return exchange_cls({"enableRateLimit": True})
+        return exchange_cls({"enableRateLimit": True, "timeout": _CCXT_TIMEOUT_MS})
 
     def fetch(
         self,
@@ -97,12 +107,37 @@ def _fetch_one(
         exchange, symbol: str, timeframe: str, since_ms: int, end_ms: int,
     ) -> Optional[pd.DataFrame]:
         """Paginated OHLCV fetch for one symbol."""
+        import ccxt
+
         all_rows: list = []
         cursor = since_ms
         limit = 1000
+        deadline = time.monotonic() + _CCXT_FETCH_BUDGET_S
 
         for _ in range(200):
-            ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since=cursor, limit=limit)
+            if time.monotonic() > deadline:
+                raise TimeoutError(
+                    f"ccxt fetch for {symbol} exceeded "
+                    f"{_CCXT_FETCH_BUDGET_S:.0f}s budget"
+                )
+            ohlcv = None
+            for attempt in range(_CCXT_MAX_RETRIES + 1):
+                try:
+                    ohlcv = exchange.fetch_ohlcv(
+                        symbol, timeframe, since=cursor, limit=limit
+                    )
+                    break
+                except ccxt.NetworkError as exc:
+                    # NetworkError covers RequestTimeout / DDoSProtection /
+                    # ExchangeNotAvailable — the transient family. Anything
+                    # else (e.g. ExchangeError: bad symbol) is not retried.
+                    remaining = deadline - time.monotonic()
+                    if attempt == _CCXT_MAX_RETRIES or remaining <= 0:
+                        raise TimeoutError(
+                            f"ccxt fetch for {symbol} failed after "
+                            f"{attempt + 1} attempt(s): {exc}"
+                        ) from exc
+                    time.sleep(min(_CCXT_BACKOFF[attempt], max(0.0, remaining)))
             if not ohlcv:
                 break
             all_rows.extend(ohlcv)

agent/backtest/loaders/okx.py

@@ -5,6 +5,8 @@
 Up to 300 bars per request; paginates with ``after`` for longer history.
 """
 
+import os
+import time
 from typing import Dict, List, Optional
 
 import pandas as pd
@@ -15,6 +17,13 @@
 
 BASE_URL = "https://www.okx.com/api/v5"
 _MAX_PER_PAGE = 300
+# P12-b parity: OKX already sets a per-request timeout but had no retry
+# budget, so a transient blip dropped the whole symbol and a slow tier
+# could stall ~max_pages*timeout. Bound it like the ccxt loader.
+_OKX_TIMEOUT = int(os.getenv("OKX_TIMEOUT_S", "15"))
+_OKX_FETCH_BUDGET_S = float(os.getenv("OKX_FETCH_BUDGET_S", "60"))
+_OKX_MAX_RETRIES = 3
+_OKX_BACKOFF = (0.5, 1.5, 4.0)  # seconds; len == _OKX_MAX_RETRIES
 
 
 @register
@@ -98,16 +107,38 @@ def _fetch_candles(
         """
         all_rows: list = []
         after = str(end_ts)
+        deadline = time.monotonic() + _OKX_FETCH_BUDGET_S
 
         for _ in range(max_pages):
+            if time.monotonic() > deadline:
+                raise TimeoutError(
+                    f"OKX fetch for {inst_id} exceeded "
+                    f"{_OKX_FETCH_BUDGET_S:.0f}s budget"
+                )
             params = {
                 "instId": inst_id,
                 "bar": bar,
                 "limit": str(_MAX_PER_PAGE),
                 "after": after,
             }
-            resp = requests.get(f"{BASE_URL}/market/candles", params=params, timeout=15)
-            data = resp.json()
+            data = None
+            for attempt in range(_OKX_MAX_RETRIES + 1):
+                try:
+                    resp = requests.get(
+                        f"{BASE_URL}/market/candles",
+                        params=params,
+                        timeout=_OKX_TIMEOUT,
+                    )
+                    data = resp.json()
+                    break
+                except requests.RequestException as exc:
+                    remaining = deadline - time.monotonic()
+                    if attempt == _OKX_MAX_RETRIES or remaining <= 0:
+                        raise TimeoutError(
+                            f"OKX fetch for {inst_id} failed after "
+                            f"{attempt + 1} attempt(s): {exc}"
+                        ) from exc
+                    time.sleep(min(_OKX_BACKOFF[attempt], max(0.0, remaining)))
             if data.get("code") != "0" or not data.get("data"):
                 break
 

agent/tests/test_ccxt_loader_bounded.py

@@ -0,0 +1,91 @@
+"""Regression tests for P12-b — the ccxt loader must fail fast instead of
+hanging on a transient disconnect.
+
+Pre-fix: `_fetch_one` called `exchange.fetch_ohlcv` with no per-call timeout,
+no retry, and no wall-clock budget, so a flaky connection hung
+`get_market_data` for 10+ minutes. Post-fix: bounded retry on the transient
+`ccxt.NetworkError` family + a hard budget that raises a clear `TimeoutError`;
+the happy path is unchanged (one call per page).
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import ccxt
+
+import backtest.loaders.ccxt_loader as cl
+from backtest.loaders.ccxt_loader import DataLoader
+
+SINCE = int(pd.Timestamp("2026-05-01").timestamp() * 1000)
+END = int((pd.Timestamp("2026-05-05") + pd.Timedelta(days=1)).timestamp() * 1000)
+
+
+def _bars(n: int = 4) -> list:
+    base = int(pd.Timestamp("2026-05-01").timestamp() * 1000)
+    day = 86_400_000
+    return [[base + i * day, 100 + i, 101 + i, 99 + i, 100 + i, 10 + i] for i in range(n)]
+
+
+class _FakeEx:
+    """Scripted exchange: each fetch_ohlcv call consumes the next script item;
+    an Exception item is raised, a list item is returned."""
+
+    def __init__(self, script: list) -> None:
+        self.script = script
+        self.calls = 0
+
+    def fetch_ohlcv(self, symbol, timeframe, since=None, limit=None):
+        item = self.script[min(self.calls, len(self.script) - 1)]
+        self.calls += 1
+        if isinstance(item, BaseException):
+            raise item
+        return item
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep(monkeypatch):
+    monkeypatch.setattr(cl.time, "sleep", lambda *_a, **_k: None)
+
+
+def test_transient_networkerror_retried_then_succeeds():
+    ex = _FakeEx([ccxt.NetworkError("blip"), ccxt.NetworkError("blip"), _bars(), []])
+    df = DataLoader._fetch_one(ex, "BTC/USDT", "1d", SINCE, END)
+    assert ex.calls >= 3
+    assert df is not None and not df.empty
+
+
+def test_persistent_disconnect_is_bounded_not_a_hang():
+    """The old 10-min hang: now a bounded TimeoutError after a fixed budget."""
+    ex = _FakeEx([ccxt.NetworkError("down")])  # always fails
+    with pytest.raises(TimeoutError):
+        DataLoader._fetch_one(ex, "BTC/USDT", "1d", SINCE, END)
+    assert ex.calls == cl._CCXT_MAX_RETRIES + 1  # bounded, not range(200)/forever
+
+
+def test_non_network_error_is_not_retried():
+    ex = _FakeEx([ccxt.ExchangeError("bad symbol")])
+    with pytest.raises(ccxt.ExchangeError):
+        DataLoader._fetch_one(ex, "BTC/USDT", "1d", SINCE, END)
+    assert ex.calls == 1
+
+
+def test_happy_path_single_call_unchanged():
+    ex = _FakeEx([_bars(), []])
+    df = DataLoader._fetch_one(ex, "BTC/USDT", "1d", SINCE, END)
+    assert ex.calls == 1  # short page (< limit) -> exactly one call, as before
+    assert list(df.columns) == ["open", "high", "low", "close", "volume"]
+
+
+def test_wallclock_budget_enforced(monkeypatch):
+    seq = iter([1000.0, 1000.0, 1_000_000.0])  # deadline blown by the retry check
+    monkeypatch.setattr(cl.time, "monotonic", lambda: next(seq, 1_000_000.0))
+    ex = _FakeEx([ccxt.NetworkError("slow")])
+    with pytest.raises(TimeoutError):
+        DataLoader._fetch_one(ex, "BTC/USDT", "1d", SINCE, END)
+
+
+def test_get_exchange_sets_explicit_timeout():
+    ex = DataLoader()._get_exchange()
+    assert ex.timeout == cl._CCXT_TIMEOUT_MS

agent/tests/test_okx_loader_bounded.py

@@ -0,0 +1,93 @@
+"""Regression tests for the P12-b okx.py parity fix — the OKX loader must
+fail fast on a transient disconnect instead of silently dropping the symbol
+or stalling ~max_pages*timeout.
+
+Pre-fix: `_fetch_candles` called requests.get once per page with no retry and
+no wall-clock budget. Post-fix: bounded retry on the transient
+requests.RequestException family + a hard budget raising a clear TimeoutError;
+the happy path still issues one request per page (no behavior change).
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+import requests
+
+import backtest.loaders.okx as okx
+from backtest.loaders.okx import DataLoader
+
+S = int(pd.Timestamp("2026-05-01").timestamp() * 1000)
+E = int((pd.Timestamp("2026-05-05") + pd.Timedelta(days=1)).timestamp() * 1000)
+
+
+class _Resp:
+    def __init__(self, payload):
+        self._p = payload
+
+    def json(self):
+        return self._p
+
+
+def _ok_page():
+    # one short page (< _MAX_PER_PAGE) so the loop breaks after one call
+    ts = int(pd.Timestamp("2026-05-02").timestamp() * 1000)
+    return _Resp({"code": "0", "data": [[ts, "1", "2", "0.5", "1.5", "10", "0", "0", "1"]]})
+
+
+class _Seq:
+    def __init__(self, script):
+        self.script = script
+        self.calls = 0
+
+    def __call__(self, *a, **k):
+        item = self.script[min(self.calls, len(self.script) - 1)]
+        self.calls += 1
+        if isinstance(item, BaseException):
+            raise item
+        return item
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep(monkeypatch):
+    monkeypatch.setattr(okx.time, "sleep", lambda *_a, **_k: None)
+
+
+def test_transient_then_success(monkeypatch):
+    seq = _Seq([requests.ConnectionError("blip"), requests.ConnectionError("blip"), _ok_page()])
+    monkeypatch.setattr(okx.requests, "get", seq)
+    df = DataLoader()._fetch_candles("BTC-USDT", S, E, "1D", 20)
+    assert seq.calls >= 3
+    assert df is not None and not df.empty
+
+
+def test_persistent_disconnect_is_bounded(monkeypatch):
+    seq = _Seq([requests.ConnectionError("down")])
+    monkeypatch.setattr(okx.requests, "get", seq)
+    with pytest.raises(TimeoutError):
+        DataLoader()._fetch_candles("BTC-USDT", S, E, "1D", 20)
+    assert seq.calls == okx._OKX_MAX_RETRIES + 1  # bounded, not max_pages/forever
+
+
+def test_non_network_error_not_retried(monkeypatch):
+    seq = _Seq([KeyError("logic bug")])
+    monkeypatch.setattr(okx.requests, "get", seq)
+    with pytest.raises(KeyError):
+        DataLoader()._fetch_candles("BTC-USDT", S, E, "1D", 20)
+    assert seq.calls == 1
+
+
+def test_happy_path_single_call(monkeypatch):
+    seq = _Seq([_ok_page()])
+    monkeypatch.setattr(okx.requests, "get", seq)
+    df = DataLoader()._fetch_candles("BTC-USDT", S, E, "1D", 20)
+    assert seq.calls == 1
+    assert list(df.columns) == ["open", "high", "low", "close", "volume"]
+
+
+def test_wallclock_budget_enforced(monkeypatch):
+    seq = iter([1000.0, 1000.0, 1_000_000.0])
+    monkeypatch.setattr(okx.time, "monotonic", lambda: next(seq, 1_000_000.0))
+    monkeypatch.setattr(okx.requests, "get", _Seq([requests.ConnectionError("slow")]))
+    with pytest.raises(TimeoutError):
+        DataLoader()._fetch_candles("BTC-USDT", S, E, "1D", 20)