CI / tests (py3.10) (push) Has been cancelled
CI / tests (py3.11) (push) Has been cancelled
CI / tests (py3.12) (push) Has been cancelled
CI / tests (py3.13) (push) Has been cancelled
CI / clean-install smoke (push) Has been cancelled
CI / ruff (strict, full repo) (push) Has been cancelled
- Add SSH proxy via foreign server for Yahoo Finance bypass - Add fallback in load_ohlcv and get_YFin_data_online - Add LLM JSON decode retry (3 attempts) in NormalizedChatOpenAI - Add yahoo_proxy.py utility module - Add test_smoke.py and patches/
420 lines
17 KiB
Diff
420 lines
17 KiB
Diff
diff --git a/tradingagents/dataflows/stockstats_utils.py b/tradingagents/dataflows/stockstats_utils.py
|
|
index db880b0..cb97dc1 100644
|
|
--- a/tradingagents/dataflows/stockstats_utils.py
|
|
+++ b/tradingagents/dataflows/stockstats_utils.py
|
|
@@ -1,6 +1,8 @@
|
|
import logging
|
|
import os
|
|
import time
|
|
+import json
|
|
+import subprocess
|
|
from typing import Annotated
|
|
|
|
import pandas as pd
|
|
@@ -19,6 +21,82 @@ logger = logging.getLogger(__name__)
|
|
# enough to catch the year-old frames yfinance occasionally returns (#1021).
|
|
MAX_OHLCV_STALE_DAYS = 10
|
|
|
|
+_ssh_config_cache = None
|
|
+
|
|
+def _get_ssh_config():
|
|
+ """Load SSH credentials for foreign server from vault (cached)."""
|
|
+ global _ssh_config_cache
|
|
+ if _ssh_config_cache is not None:
|
|
+ return _ssh_config_cache
|
|
+ result = subprocess.run(
|
|
+ ["bash", "-c",
|
|
+ 'source <(bash /root/.openclaw/workspace/scripts/vault.sh decrypt '
|
|
+ '/root/.openclaw/workspace/secrets/credentials.env.enc) '
|
|
+ '&& echo "$FOREIGN_HOST $FOREIGN_PORT $FOREIGN_PASS"'],
|
|
+ capture_output=True, text=True, timeout=10
|
|
+ )
|
|
+ parts = result.stdout.strip().split()
|
|
+ if len(parts) != 3:
|
|
+ raise RuntimeError(f"Failed to load SSH config: {result.stderr}")
|
|
+ _ssh_config_cache = (parts[0], int(parts[1]), parts[2])
|
|
+ return _ssh_config_cache
|
|
+
|
|
+
|
|
+def _fetch_via_ssh(url: str, timeout: int = 30) -> str:
|
|
+ """Fetch a URL through the foreign server via SSH tunnel."""
|
|
+ host, port, password = _get_ssh_config()
|
|
+ cmd = [
|
|
+ "sshpass", "-p", password,
|
|
+ "ssh", "-o", "StrictHostKeyChecking=no",
|
|
+ "-o", "ConnectTimeout=15",
|
|
+ "-p", str(port), f"root@{host}",
|
|
+ f'curl -s "{url}" -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"'
|
|
+ ]
|
|
+ result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
|
+ if result.returncode != 0:
|
|
+ raise RuntimeError(f"SSH fetch failed: {result.stderr}")
|
|
+ return result.stdout
|
|
+
|
|
+
|
|
+def _yf_download_via_proxy(symbol: str, start: str, end: str) -> pd.DataFrame:
|
|
+ """Download OHLCV data via SSH proxy when local Yahoo Finance is rate-limited."""
|
|
+ import datetime
|
|
+ # Convert dates to timestamps
|
|
+ start_ts = int(pd.Timestamp(start).timestamp())
|
|
+ end_ts = int(pd.Timestamp(end).timestamp())
|
|
+
|
|
+ url = (f"https://query2.finance.yahoo.com/v8/finance/chart/{symbol}"
|
|
+ f"?period1={start_ts}&period2={end_ts}&interval=1d")
|
|
+
|
|
+ data_str = _fetch_via_ssh(url)
|
|
+ parsed = json.loads(data_str)
|
|
+
|
|
+ result = parsed.get("chart", {}).get("result", [])
|
|
+ if not result:
|
|
+ return pd.DataFrame()
|
|
+
|
|
+ ts = result[0].get("timestamp", [])
|
|
+ indicators = result[0].get("indicators", {}).get("quote", [{}])[0]
|
|
+
|
|
+ if not ts:
|
|
+ return pd.DataFrame()
|
|
+
|
|
+ df = pd.DataFrame({
|
|
+ "Date": pd.to_datetime(ts, unit="s"),
|
|
+ "Open": indicators.get("open", []),
|
|
+ "High": indicators.get("high", []),
|
|
+ "Low": indicators.get("low", []),
|
|
+ "Close": indicators.get("close", []),
|
|
+ "Volume": indicators.get("volume", []),
|
|
+ })
|
|
+
|
|
+ # Convert None values to NaN
|
|
+ for col in ["Open", "High", "Low", "Close", "Volume"]:
|
|
+ df[col] = pd.to_numeric(df[col], errors="coerce")
|
|
+
|
|
+ df = df.dropna(subset=["Close"])
|
|
+ return df
|
|
+
|
|
|
|
def yf_retry(func, max_retries=3, base_delay=2.0):
|
|
"""Execute a yfinance call with exponential backoff on rate limits.
|
|
@@ -36,6 +114,7 @@ def yf_retry(func, max_retries=3, base_delay=2.0):
|
|
logger.warning(f"Yahoo Finance rate limited, retrying in {delay:.0f}s (attempt {attempt + 1}/{max_retries})")
|
|
time.sleep(delay)
|
|
else:
|
|
+ logger.warning("Yahoo Finance local exhausted, will let caller handle fallback")
|
|
raise
|
|
|
|
|
|
@@ -122,6 +201,82 @@ def _assert_ohlcv_not_stale(
|
|
)
|
|
|
|
|
|
+def _load_ohlcv_alpha_vantage(symbol: str, curr_date: str) -> pd.DataFrame:
|
|
+ """Fetch OHLCV data from Alpha Vantage, filtered to prevent look-ahead bias."""
|
|
+ from .alpha_vantage_common import _make_api_request
|
|
+ from io import StringIO
|
|
+
|
|
+ canonical = normalize_symbol(symbol)
|
|
+ curr_date_dt = pd.to_datetime(curr_date)
|
|
+ today_date = pd.Timestamp.today()
|
|
+ start_date = today_date - pd.DateOffset(years=5)
|
|
+ start_str = start_date.strftime("%Y-%m-%d")
|
|
+
|
|
+ # Determine outputsize based on how far back we need
|
|
+ days_from_today = (today_date - start_date).days
|
|
+ outputsize = "compact" if days_from_today < 100 else "full"
|
|
+
|
|
+ params = {
|
|
+ "symbol": canonical,
|
|
+ "outputsize": outputsize,
|
|
+ "datatype": "csv",
|
|
+ }
|
|
+
|
|
+ try:
|
|
+ csv_text = _make_api_request("TIME_SERIES_DAILY", params)
|
|
+ except Exception as e:
|
|
+ raise NoMarketDataError(symbol, canonical, f"Alpha Vantage error: {e}") from e
|
|
+
|
|
+ if not csv_text or not csv_text.strip():
|
|
+ raise NoMarketDataError(symbol, canonical, "Alpha Vantage returned empty data")
|
|
+
|
|
+ try:
|
|
+ df = pd.read_csv(StringIO(csv_text))
|
|
+ except Exception as e:
|
|
+ raise NoMarketDataError(symbol, canonical, f"Failed to parse Alpha Vantage CSV: {e}") from e
|
|
+
|
|
+ if df.empty:
|
|
+ raise NoMarketDataError(symbol, canonical, "Alpha Vantage returned no rows")
|
|
+
|
|
+ # Alpha Vantage CSV has 'timestamp' column; normalize to 'Date'
|
|
+ date_col = df.columns[0] # usually 'timestamp'
|
|
+ df = df.rename(columns={date_col: "Date"})
|
|
+
|
|
+ # Keep only standard OHLCV columns
|
|
+ keep_cols = ["Date"]
|
|
+ col_map = {}
|
|
+ for av_col, std_col in [
|
|
+ ("open", "Open"), ("high", "High"), ("low", "Low"),
|
|
+ ("close", "Close"), ("adjusted_close", "Close"), ("volume", "Volume"),
|
|
+ ]:
|
|
+ for c in df.columns:
|
|
+ if c.lower().replace(" ", "_") == av_col:
|
|
+ if std_col not in keep_cols:
|
|
+ keep_cols.append(std_col)
|
|
+ col_map[c] = std_col
|
|
+ break
|
|
+ df = df.rename(columns=col_map)
|
|
+ # Prefer adjusted_close over close for "Close"
|
|
+ if "adjusted_close" in [c.lower().replace(" ", "_") for c in df.columns]:
|
|
+ for c in df.columns:
|
|
+ if c.lower().replace(" ", "_") == "adjusted_close":
|
|
+ df["Close"] = pd.to_numeric(df[c], errors="coerce")
|
|
+ break
|
|
+
|
|
+ available = [c for c in ["Date", "Open", "High", "Low", "Close", "Volume"] if c in df.columns]
|
|
+ df = df[available]
|
|
+
|
|
+ # Filter to curr_date
|
|
+ df["Date"] = pd.to_datetime(df["Date"], errors="coerce")
|
|
+ df = df.dropna(subset=["Date"])
|
|
+ df = df[df["Date"] <= curr_date_dt].sort_values("Date")
|
|
+
|
|
+ if df.empty:
|
|
+ raise NoMarketDataError(symbol, canonical, "No data on or before " + curr_date)
|
|
+
|
|
+ return df
|
|
+
|
|
+
|
|
def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
|
|
"""Fetch OHLCV data with caching, filtered to prevent look-ahead bias.
|
|
|
|
@@ -138,24 +293,24 @@ def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
|
|
config = get_config()
|
|
curr_date_dt = pd.to_datetime(curr_date)
|
|
|
|
+ # Check which vendor to use for OHLCV
|
|
+ vendor_config = config.get("data_vendors", {}).get("core_stock_apis", "yfinance")
|
|
+ use_alpha_vantage = "alpha_vantage" in vendor_config
|
|
+
|
|
# Cache uses a fixed window (5y to today) so one file per symbol.
|
|
today_date = pd.Timestamp.today()
|
|
start_date = today_date - pd.DateOffset(years=5)
|
|
start_str = start_date.strftime("%Y-%m-%d")
|
|
- # yfinance ``end`` is EXCLUSIVE; request tomorrow so today's row is included
|
|
- # when curr_date is the current day (#986). Look-ahead is still prevented by
|
|
- # the curr_date filter below.
|
|
end_str = (today_date + pd.Timedelta(days=1)).strftime("%Y-%m-%d")
|
|
|
|
+ cache_prefix = "AV" if use_alpha_vantage else "YFin"
|
|
os.makedirs(config["data_cache_dir"], exist_ok=True)
|
|
data_file = os.path.join(
|
|
config["data_cache_dir"],
|
|
- f"{safe_symbol}-YFin-data-{start_str}-{end_str}.csv",
|
|
+ f"{safe_symbol}-{cache_prefix}-data-{start_str}-{end_str}.csv",
|
|
)
|
|
|
|
- # A cached file may be empty if a prior fetch failed (unknown symbol,
|
|
- # transient rate limit). Treat an empty/columnless cache as a miss and
|
|
- # re-fetch rather than serving the poisoned file forever.
|
|
+ # A cached file may be empty if a prior fetch failed. Treat empty cache as miss.
|
|
data = None
|
|
if os.path.exists(data_file):
|
|
cached = pd.read_csv(data_file, on_bad_lines="skip", encoding="utf-8")
|
|
@@ -163,30 +318,53 @@ def load_ohlcv(symbol: str, curr_date: str) -> pd.DataFrame:
|
|
data = cached
|
|
|
|
if data is None:
|
|
- downloaded = yf_retry(lambda: yf.download(
|
|
- canonical,
|
|
- start=start_str,
|
|
- end=end_str,
|
|
- multi_level_index=False,
|
|
- progress=False,
|
|
- auto_adjust=True,
|
|
- ))
|
|
- downloaded = _ensure_date_column(downloaded.reset_index())
|
|
- # Only cache real data — never persist an empty frame.
|
|
- if downloaded.empty or "Close" not in downloaded.columns:
|
|
- raise NoMarketDataError(
|
|
- symbol, canonical, "Yahoo Finance returned no rows"
|
|
- )
|
|
+ if use_alpha_vantage:
|
|
+ downloaded = _load_ohlcv_alpha_vantage(symbol, curr_date)
|
|
+ else:
|
|
+ downloaded = None
|
|
+ try:
|
|
+ raw = yf_retry(lambda: yf.download(
|
|
+ canonical,
|
|
+ start=start_str,
|
|
+ end=end_str,
|
|
+ multi_level_index=False,
|
|
+ progress=False,
|
|
+ auto_adjust=True,
|
|
+ ))
|
|
+ downloaded = _ensure_date_column(raw.reset_index())
|
|
+ except Exception as yf_err:
|
|
+ pass # Will try proxy below
|
|
+
|
|
+ # If local yfinance failed or returned empty, try SSH proxy
|
|
+ if downloaded is None or downloaded.empty or "Close" not in downloaded.columns:
|
|
+ if downloaded is None or downloaded.empty:
|
|
+ logger.warning(f"Yahoo Finance local failed, trying SSH proxy...")
|
|
+ try:
|
|
+ downloaded = _yf_download_via_proxy(canonical, start_str, end_str)
|
|
+ except Exception as proxy_err:
|
|
+ raise NoMarketDataError(
|
|
+ symbol, canonical, f"Yahoo Finance proxy also failed: {proxy_err}"
|
|
+ ) from proxy_err
|
|
+ if downloaded.empty:
|
|
+ raise NoMarketDataError(
|
|
+ symbol, canonical, f"Yahoo Finance proxy also returned no data"
|
|
+ )
|
|
+ if downloaded.empty or "Close" not in downloaded.columns:
|
|
+ raise NoMarketDataError(
|
|
+ symbol, canonical, "Yahoo Finance returned no rows"
|
|
+ )
|
|
+ downloaded = _clean_dataframe(downloaded)
|
|
+
|
|
downloaded.to_csv(data_file, index=False, encoding="utf-8")
|
|
data = downloaded
|
|
|
|
- data = _clean_dataframe(data)
|
|
+ if not use_alpha_vantage:
|
|
+ data = _clean_dataframe(data)
|
|
|
|
# Filter to curr_date to prevent look-ahead bias in backtesting
|
|
data = data[data["Date"] <= curr_date_dt]
|
|
|
|
- # Reject a stale frame (latest row far older than curr_date) rather than
|
|
- # feeding year-old prices into indicators (#1021).
|
|
+ # Reject a stale frame
|
|
_assert_ohlcv_not_stale(data, curr_date, symbol, canonical)
|
|
|
|
return data
|
|
diff --git a/tradingagents/dataflows/y_finance.py b/tradingagents/dataflows/y_finance.py
|
|
index fdb49d5..9e36c6d 100644
|
|
--- a/tradingagents/dataflows/y_finance.py
|
|
+++ b/tradingagents/dataflows/y_finance.py
|
|
@@ -1,13 +1,17 @@
|
|
+import logging
|
|
from datetime import datetime
|
|
from typing import Annotated
|
|
|
|
import pandas as pd
|
|
import yfinance as yf
|
|
+
|
|
+logger = logging.getLogger(__name__)
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
from .stockstats_utils import (
|
|
StockstatsUtils,
|
|
_assert_ohlcv_not_stale,
|
|
+ _yf_download_via_proxy,
|
|
filter_financials_by_date,
|
|
load_ohlcv,
|
|
yf_retry,
|
|
@@ -32,7 +36,19 @@ def get_YFin_data_online(
|
|
# end_date row (and the current day when end_date is today). Request one day
|
|
# past end_date so the requested range is actually inclusive (#986/#987).
|
|
end_inclusive = (end_dt + relativedelta(days=1)).strftime("%Y-%m-%d")
|
|
- data = yf_retry(lambda: ticker.history(start=start_date, end=end_inclusive))
|
|
+ try:
|
|
+ data = yf_retry(lambda: ticker.history(start=start_date, end=end_inclusive))
|
|
+ except Exception:
|
|
+ # Fallback: use SSH proxy via foreign server
|
|
+ logger.warning(f"Yahoo Finance local failed for {symbol}, trying SSH proxy...")
|
|
+ proxy_df = _yf_download_via_proxy(canonical, start_date, end_inclusive)
|
|
+ if proxy_df.empty:
|
|
+ raise NoMarketDataError(
|
|
+ symbol, canonical, f"no rows between {start_date} and {end_date} (proxy also failed)"
|
|
+ )
|
|
+ # Convert to match expected format
|
|
+ proxy_df = proxy_df.set_index("Date")
|
|
+ data = proxy_df
|
|
|
|
# Empty result means the symbol is unknown/delisted. Raise a typed error
|
|
# instead of returning prose: the routing layer turns it into a single
|
|
diff --git a/tradingagents/llm_clients/openai_client.py b/tradingagents/llm_clients/openai_client.py
|
|
index bf6e554..717750d 100644
|
|
--- a/tradingagents/llm_clients/openai_client.py
|
|
+++ b/tradingagents/llm_clients/openai_client.py
|
|
@@ -33,7 +33,21 @@ class NormalizedChatOpenAI(ChatOpenAI):
|
|
"""
|
|
|
|
def invoke(self, input, config=None, **kwargs):
|
|
- return normalize_content(super().invoke(input, config, **kwargs))
|
|
+ import json as _json
|
|
+ import time as _time
|
|
+ import logging as _logging
|
|
+ _logger = _logging.getLogger(__name__)
|
|
+ max_retries = 3
|
|
+ for attempt in range(max_retries):
|
|
+ try:
|
|
+ return normalize_content(super().invoke(input, config, **kwargs))
|
|
+ except (_json.JSONDecodeError, Exception) as e:
|
|
+ if 'JSONDecodeError' in type(e).__name__ or 'Expecting value' in str(e):
|
|
+ if attempt < max_retries - 1:
|
|
+ _logger.warning(f"LLM returned invalid JSON, retrying ({attempt+1}/{max_retries}): {e}")
|
|
+ _time.sleep(2 * (attempt + 1))
|
|
+ continue
|
|
+ raise
|
|
|
|
def with_structured_output(self, schema, *, method=None, **kwargs):
|
|
caps = get_capabilities(self.model_name)
|
|
diff --git a/tradingagents/dataflows/yahoo_proxy.py b/tradingagents/dataflows/yahoo_proxy.py
|
|
new file mode 100644
|
|
index 0000000..c06e0ad
|
|
--- /dev/null
|
|
+++ b/tradingagents/dataflows/yahoo_proxy.py
|
|
@@ -0,0 +1,53 @@
|
|
+#!/usr/bin/env python3
|
|
+"""Yahoo Finance proxy via foreign server.
|
|
+
|
|
+Routes yfinance requests through the foreign server to bypass local IP rate limiting.
|
|
+"""
|
|
+import subprocess
|
|
+import json
|
|
+import os
|
|
+import sys
|
|
+
|
|
+def _get_ssh_config():
|
|
+ """Load SSH credentials from vault."""
|
|
+ result = subprocess.run(
|
|
+ ["bash", "-c", 'source <(bash /root/.openclaw/workspace/scripts/vault.sh decrypt /root/.openclaw/workspace/secrets/credentials.env.enc) && echo "$FOREIGN_HOST $FOREIGN_PORT $FOREIGN_PASS"'],
|
|
+ capture_output=True, text=True, timeout=10
|
|
+ )
|
|
+ parts = result.stdout.strip().split()
|
|
+ if len(parts) != 3:
|
|
+ raise RuntimeError(f"Failed to load SSH config: {result.stderr}")
|
|
+ return parts[0], int(parts[1]), parts[2]
|
|
+
|
|
+def fetch_yahoo_via_ssh(url: str, timeout: int = 30) -> str:
|
|
+ """Fetch a Yahoo Finance URL through the foreign server via SSH."""
|
|
+ host, port, password = _get_ssh_config()
|
|
+ cmd = [
|
|
+ "sshpass", "-p", password,
|
|
+ "ssh", "-o", "StrictHostKeyChecking=no",
|
|
+ "-o", f"ConnectTimeout=15",
|
|
+ "-p", str(port), f"root@{host}",
|
|
+ f'curl -s "{url}" -H "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"'
|
|
+ ]
|
|
+ result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
|
+ if result.returncode != 0:
|
|
+ raise RuntimeError(f"SSH fetch failed: {result.stderr}")
|
|
+ return result.stdout
|
|
+
|
|
+def test_connection() -> bool:
|
|
+ """Test if Yahoo Finance is accessible via foreign server."""
|
|
+ try:
|
|
+ data = fetch_yahoo_via_ssh(
|
|
+ "https://query2.finance.yahoo.com/v8/finance/chart/AAPL?interval=1d&range=5d"
|
|
+ )
|
|
+ parsed = json.loads(data)
|
|
+ return "chart" in parsed and "result" in parsed.get("chart", {})
|
|
+ except Exception as e:
|
|
+ print(f"Connection test failed: {e}", file=sys.stderr)
|
|
+ return False
|
|
+
|
|
+if __name__ == "__main__":
|
|
+ if test_connection():
|
|
+ print("✅ Yahoo Finance proxy via foreign server: OK")
|
|
+ else:
|
|
+ print("❌ Yahoo Finance proxy via foreign server: FAILED")
|