From 3dc4214b7879b6bf38af45b96b893fd018209b57 Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 7 May 2026 01:05:02 +0800 Subject: [PATCH] Add proxy support for market truth clients --- .env.example | 1 + src/config/settings.py | 1 + src/services/market_data/okx_client.py | 20 +++++++++++-- .../market_data_stream_service.py | 30 +++++++++++++++++-- .../market_state/news_event_provider.py | 9 +++++- 5 files changed, 55 insertions(+), 6 deletions(-) diff --git a/.env.example b/.env.example index c730da1..432317f 100644 --- a/.env.example +++ b/.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 diff --git a/src/config/settings.py b/src/config/settings.py index 3149dcc..19a7b43 100644 --- a/src/config/settings.py +++ b/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 diff --git a/src/services/market_data/okx_client.py b/src/services/market_data/okx_client.py index bcc35d3..8645926 100644 --- a/src/services/market_data/okx_client.py +++ b/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})) diff --git a/src/services/market_state/market_data_stream_service.py b/src/services/market_state/market_data_stream_service.py index 068547f..c4c0cc7 100644 --- a/src/services/market_state/market_data_stream_service.py +++ b/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 diff --git a/src/services/market_state/news_event_provider.py b/src/services/market_state/news_event_provider.py index f57e0e9..de5ef87 100644 --- a/src/services/market_state/news_event_provider.py +++ b/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()