diff --git a/.env.example b/.env.example index 49b91db..077a180 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,12 @@ OKX_ORDER_BASE_URL=https://www.okx.com OKX_API_KEY= OKX_API_SECRET= OKX_API_PASSPHRASE= +BROKER_READ_ENABLED=1 +BROKER_READ_BASE_URL=https://www.okx.com +BROKER_READ_TIMEOUT=30 +BROKER_READ_PROXY_URL= +BROKER_READ_RETRIES=2 +BROKER_READ_RETRY_BACKOFF_SECONDS=1.0 SIGNAL_EVIDENCE_MODE=strict MARKET_DATA_STREAM_ENABLED=1 MARKET_DATA_STREAM_SYMBOLS=BTC-USDT,ETH-USDT diff --git a/src/config/settings.py b/src/config/settings.py index 628fa73..889afdf 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -16,6 +16,8 @@ class Settings(BaseSettings): broker_read_base_url: str = "https://www.okx.com" broker_read_timeout: int = 30 broker_read_proxy_url: str = "" + broker_read_retries: int = 2 + broker_read_retry_backoff_seconds: float = 1.0 market_data_stream_enabled: bool = True market_data_stream_symbols: str = "BTC-USDT,ETH-USDT" market_data_stream_timeframes: str = "1m" diff --git a/src/services/broker/broker_read_model_service.py b/src/services/broker/broker_read_model_service.py index 3ad1364..43dedb2 100644 --- a/src/services/broker/broker_read_model_service.py +++ b/src/services/broker/broker_read_model_service.py @@ -8,6 +8,7 @@ import hmac from http.client import IncompleteRead import json from os import getenv +import time from typing import Optional from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlparse @@ -105,6 +106,8 @@ class BrokerReadModelService: self.base_url = (base_url or settings.broker_read_base_url or settings.okx_base_url).rstrip("/") self.proxy_url = proxy_url or settings.broker_read_proxy_url or settings.market_truth_proxy_url or getenv("HTTPS_PROXY") or getenv("HTTP_PROXY") or "" self.timeout = timeout or settings.broker_read_timeout or settings.okx_timeout + self.retries = max(int(settings.broker_read_retries or 1), 1) + self.retry_backoff_seconds = max(float(settings.broker_read_retry_backoff_seconds or 0), 0.0) self._opener = _build_proxy_opener(self.proxy_url) def fetch_snapshot( @@ -213,20 +216,29 @@ class BrokerReadModelService: }, ) open_fn = self._opener.open if self._opener is not None else urlopen - try: - with open_fn(request, timeout=self.timeout) as response: - payload = json.loads(response.read().decode("utf-8")) - except HTTPError as exc: - body = exc.read().decode("utf-8", errors="replace") + last_transport_error = None + for attempt in range(1, self.retries + 1): try: - parsed = json.loads(body) - if isinstance(parsed, dict): - body = f"{parsed.get('code')}:{parsed.get('msg')}" - except Exception: - pass - raise ValueError(f"okx_read_http_{exc.code}:{body}") from exc - except (IncompleteRead, URLError, TimeoutError, OSError) as exc: - raise ValueError(f"okx_read_transport_error:{exc}") from exc + with open_fn(request, timeout=self.timeout) as response: + payload = json.loads(response.read().decode("utf-8")) + break + except HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + try: + parsed = json.loads(body) + if isinstance(parsed, dict): + body = f"{parsed.get('code')}:{parsed.get('msg')}" + except Exception: + pass + raise ValueError(f"okx_read_http_{exc.code}:{body}") from exc + except (IncompleteRead, URLError, TimeoutError, OSError) as exc: + last_transport_error = exc + if attempt >= self.retries: + raise ValueError(f"okx_read_transport_error:{exc}") from exc + if self.retry_backoff_seconds: + time.sleep(self.retry_backoff_seconds) + else: + raise ValueError(f"okx_read_transport_error:{last_transport_error}") if not isinstance(payload, dict): raise ValueError("okx_read_invalid_payload") if str(payload.get("code", "")) not in {"", OKX_CODE_OK}: diff --git a/tests/unit/test_broker_read_model_service.py b/tests/unit/test_broker_read_model_service.py index 07a9726..24286f2 100644 --- a/tests/unit/test_broker_read_model_service.py +++ b/tests/unit/test_broker_read_model_service.py @@ -102,6 +102,48 @@ class BrokerReadModelServiceTests(unittest.TestCase): self.assertEqual(snapshot.status, BROKER_STATE_UNAVAILABLE) self.assertIn("okx_read_http_401:50113:Invalid signature", snapshot.error) + def test_private_get_retries_transport_errors(self) -> None: + from unittest.mock import patch + from urllib.error import URLError + + class _Response: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return None + + def read(self): + return b'{"code":"0","data":[]}' + + class _FlakyOpener: + def __init__(self): + self.calls = 0 + + def open(self, request, timeout=None): + self.calls += 1 + if self.calls == 1: + raise URLError("temporary handshake timeout") + return _Response() + + opener = _FlakyOpener() + with patch("src.services.broker.broker_read_model_service._build_proxy_opener", return_value=opener), patch( + "src.services.broker.broker_read_model_service.settings.okx_api_key", "k" + ), patch( + "src.services.broker.broker_read_model_service.settings.okx_api_secret", "s" + ), patch( + "src.services.broker.broker_read_model_service.settings.okx_api_passphrase", "p" + ), patch( + "src.services.broker.broker_read_model_service.settings.broker_read_retries", 2 + ), patch( + "src.services.broker.broker_read_model_service.settings.broker_read_retry_backoff_seconds", 0 + ): + service = BrokerReadModelService(proxy_url="http://127.0.0.1:10808") + payload = service._private_get("/api/v5/account/balance") + + self.assertEqual(payload["code"], "0") + self.assertEqual(opener.calls, 2) + if __name__ == "__main__": unittest.main()