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")