Add proxy support for market truth clients

.env.example

@@ -19,6 +19,7 @@ MARKET_TRUTH_FALLBACK_LIMIT=20
 MARKET_TRUTH_FALLBACK_TIMEOUT=5
 MARKET_TRUTH_FALLBACK_COOLDOWN_SECONDS=60
 MARKET_TRUTH_GAP_MULTIPLIER=2
+MARKET_TRUTH_PROXY_URL=
 NEWS_PROVIDER_ENABLED=1
 NEWS_PROVIDER_URL=https://nfs.faireconomy.media/ff_calendar_thisweek.json
 NEWS_PROVIDER_TIMEOUT=15

src/config/settings.py

@@ -24,6 +24,7 @@ class Settings(BaseSettings):
     market_truth_fallback_timeout: int = 5
     market_truth_fallback_cooldown_seconds: int = 60
     market_truth_gap_multiplier: int = 2
+    market_truth_proxy_url: str = ""
     news_provider_enabled: bool = True
     news_provider_url: str = "https://nfs.faireconomy.media/ff_calendar_thisweek.json"
     news_provider_timeout: int = 15

src/services/market_data/okx_client.py

@@ -1,6 +1,8 @@
 from dataclasses import dataclass
+from os import getenv
 from typing import Any, Optional
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlparse
+from urllib.request import ProxyHandler, build_opener
 from urllib.request import Request, urlopen
 from urllib.error import URLError, HTTPError
 from http.client import IncompleteRead
@@ -24,9 +26,11 @@ class OKXCandle:
 
 
 class OKXClient:
-    def __init__(self, base_url: Optional[str] = None, timeout: Optional[int] = None) -> None:
+    def __init__(self, base_url: Optional[str] = None, timeout: Optional[int] = None, proxy_url: Optional[str] = None) -> None:
         self.base_url = base_url or settings.okx_base_url
         self.timeout = timeout or settings.okx_timeout
+        self.proxy_url = proxy_url or settings.market_truth_proxy_url or getenv("HTTPS_PROXY") or getenv("HTTP_PROXY") or ""
+        self._opener = _build_proxy_opener(self.proxy_url)
 
     def _get(self, path: str, params: dict[str, Any]) -> dict[str, Any]:
         query = urlencode(params)
@@ -35,7 +39,8 @@ class OKXClient:
         last_error: Optional[Exception] = None
         for attempt in range(3):
             try:
-                with urlopen(request, timeout=self.timeout) as response:
+                open_fn = self._opener.open if self._opener is not None else urlopen
+                with open_fn(request, timeout=self.timeout) as response:
                     payload = json.loads(response.read().decode("utf-8"))
                 if payload.get("code") not in ("0", 0, None):
                     raise RuntimeError(f"OKX API error: {payload}")
@@ -75,3 +80,12 @@ class OKXClient:
                 )
             )
         return candles
+
+
+def _build_proxy_opener(proxy_url: str):
+    if not proxy_url:
+        return None
+    parsed = urlparse(proxy_url)
+    if not parsed.scheme or not parsed.netloc:
+        return None
+    return build_opener(ProxyHandler({"http": proxy_url, "https": proxy_url}))

src/services/market_state/market_data_stream_service.py

@@ -1,9 +1,11 @@
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from datetime import datetime, timezone
 import json
+from os import getenv
 import threading
 import time
 from typing import Optional
+from urllib.parse import urlparse
 
 from websocket import WebSocketApp
 
@@ -113,11 +115,13 @@ class MarketDataStreamService:
         heartbeat_seconds: Optional[int] = None,
         max_lag_seconds: Optional[int] = None,
         reconnect_backoff_seconds: Optional[int] = None,
+        proxy_url: Optional[str] = None,
     ) -> None:
         self.url = url or settings.okx_ws_public_url
         self.heartbeat_seconds = heartbeat_seconds or settings.market_data_stream_heartbeat_seconds
         self.max_lag_seconds = max_lag_seconds or settings.market_data_stream_max_lag_seconds
         self.reconnect_backoff_seconds = reconnect_backoff_seconds or settings.market_data_stream_reconnect_backoff_seconds
+        self.proxy_url = proxy_url or settings.market_truth_proxy_url or getenv("HTTPS_PROXY") or getenv("HTTP_PROXY") or ""
         self._lock = threading.RLock()
         self._subscriptions: dict[tuple[int, str], StreamSubscription] = {}
         self._snapshots: dict[tuple[int, str], StreamSnapshot] = {}
@@ -211,7 +215,12 @@ class MarketDataStreamService:
             with self._lock:
                 self._ws = ws
             try:
-                ws.run_forever(ping_interval=max(1, self.heartbeat_seconds), ping_timeout=max(1, self.heartbeat_seconds // 2))
+                kwargs = {
+                    "ping_interval": max(1, self.heartbeat_seconds),
+                    "ping_timeout": max(1, self.heartbeat_seconds // 2),
+                }
+                kwargs.update(_proxy_run_forever_kwargs(self.proxy_url))
+                ws.run_forever(**kwargs)
             except Exception as exc:
                 self._mark_disconnected(reason=str(exc), mark_error=True)
             finally:
@@ -436,3 +445,20 @@ def _to_optional_float(value) -> Optional[float]:
     if value is None or value == "":
         return None
     return float(value)
+
+
+def _proxy_run_forever_kwargs(proxy_url: str) -> dict:
+    if not proxy_url:
+        return {}
+    parsed = urlparse(proxy_url)
+    if not parsed.hostname or not parsed.port:
+        return {}
+    proxy_type = "http" if parsed.scheme in {"http", "https"} else parsed.scheme
+    kwargs = {
+        "http_proxy_host": parsed.hostname,
+        "http_proxy_port": parsed.port,
+        "proxy_type": proxy_type,
+    }
+    if parsed.username:
+        kwargs["http_proxy_auth"] = (parsed.username, parsed.password or "")
+    return kwargs

src/services/market_state/news_event_provider.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
+from os import getenv
 from typing import Optional
 
 import httpx
@@ -67,6 +68,7 @@ class EconomicCalendarProvider:
         post_window_minutes: Optional[int] = None,
         max_age_seconds: Optional[int] = None,
         high_impact_only: Optional[bool] = None,
+        proxy_url: Optional[str] = None,
     ) -> None:
         self.url = url or settings.news_provider_url
         self.timeout = timeout or settings.news_provider_timeout
@@ -74,6 +76,7 @@ class EconomicCalendarProvider:
         self.post_window_minutes = post_window_minutes if post_window_minutes is not None else settings.news_event_post_window_minutes
         self.max_age_seconds = max_age_seconds if max_age_seconds is not None else settings.news_provider_max_age_seconds
         self.high_impact_only = settings.news_event_high_impact_only if high_impact_only is None else high_impact_only
+        self.proxy_url = proxy_url or settings.market_truth_proxy_url or getenv("HTTPS_PROXY") or getenv("HTTP_PROXY") or ""
 
     def fetch_snapshot(self, *, now: Optional[datetime] = None, rows: Optional[list[dict]] = None) -> NewsProviderSnapshot:
         now = _ensure_aware(now or datetime.now(timezone.utc))
@@ -139,7 +142,11 @@ class EconomicCalendarProvider:
         return snapshot
 
     def _fetch_rows(self) -> list[dict]:
-        with httpx.Client(timeout=self.timeout, follow_redirects=True) as client:
+        client_kwargs = {"timeout": self.timeout, "follow_redirects": True}
+        if self.proxy_url:
+            client_kwargs["proxy"] = self.proxy_url
+            client_kwargs["trust_env"] = False
+        with httpx.Client(**client_kwargs) as client:
             response = client.get(self.url, headers={"User-Agent": "ai-ict/market-truth"})
             response.raise_for_status()
             payload = response.json()