diff --git a/patches/README.md b/patches/README.md new file mode 100644 index 0000000..7ea42b6 --- /dev/null +++ b/patches/README.md @@ -0,0 +1,52 @@ +# TradingAgents: Yahoo Finance SSH Proxy + Xiaomi MiMo LLM Support + +## 概述 + +本 patch 解决两个核心问题： +1. **Yahoo Finance 云服务器 IP 被限流** — 通过国外服务器 SSH 代理中转请求 +2. **LLM 兼容性** — 支持小米 MiMo 等 OpenAI 兼容接口，增加 JSON 解析重试 + +## 修改文件 + +| 文件 | 修改内容 | +|------|----------| +| `tradingagents/dataflows/yahoo_proxy.py` | **新增** — SSH 代理工具模块 | +| `tradingagents/dataflows/stockstats_utils.py` | Yahoo Finance 代理降级 + `_yf_download_via_proxy` | +| `tradingagents/dataflows/y_finance.py` | `get_YFin_data_online` 代理降级 | +| `tradingagents/llm_clients/openai_client.py` | LLM JSON 解析重试机制 | + +## 应用方法 + +```bash +cd TradingAgents +git apply patches/yahoo-proxy-xiaomi-support.patch +``` + +## 前置条件 + +1. 国外服务器 SSH 凭据配置在 `secrets/credentials.env.enc`（变量 `FOREIGN_HOST`, `FOREIGN_PORT`, `FOREIGN_PASS`） +2. `sshpass` 已安装：`apt install sshpass` +3. Alpha Vantage API key（可选，用于基本面数据） + +## 使用小米 MiMo 模型 + +```python +import os +os.environ["OPENAI_API_KEY"] = "your-xiaomi-api-key" +os.environ["OPENAI_API_BASE"] = "https://token-plan-cn.xiaomimimo.com/v1" + +config = DEFAULT_CONFIG.copy() +config["llm_provider"] = "openai" +config["deep_think_llm"] = "mimo-v2.5-pro" +config["quick_think_llm"] = "mimo-v2.5" +``` + +## 测试结果 + +- ✅ Yahoo Finance SSH 代理：1137 行 OHLCV 数据获取成功 +- ✅ 小米 MiMo Pro：完整分析流程通过（技术/基本面/情绪/新闻/辩论/决策） +- ✅ LLM JSON 重试：偶发解析错误自动恢复 +- ✅ Alpha Vantage：基本面数据正常 + +--- +*Generated: 2026-07-07 | Author: 小钰* diff --git a/patches/yahoo-proxy-xiaomi-support.patch b/patches/yahoo-proxy-xiaomi-support.patch new file mode 100644 index 0000000..c2a381e --- /dev/null +++ b/patches/yahoo-proxy-xiaomi-support.patch @@ -0,0 +1,419 @@ +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") diff --git a/test_smoke.py b/test_smoke.py new file mode 100644 index 0000000..db46902 --- /dev/null +++ b/test_smoke.py @@ -0,0 +1,66 @@ +"""Quick smoke test for TradingAgents deployment. + +Uses OpenRouter + free Gemini model to minimize cost. +""" +import os +import json + +# Load OpenRouter API key from openclaw config +with open(os.path.expanduser("~/.openclaw/openclaw.json")) as f: + oc_config = json.load(f) +openrouter_key = oc_config.get("env", {}).get("OPENROUTER_API_KEY", "") +if not openrouter_key: + raise RuntimeError("OPENROUTER_API_KEY not found in openclaw.json") + +os.environ["OPENROUTER_API_KEY"] = openrouter_key +os.environ["ALPHA_VANTAGE_API_KEY"] = "W5PTA2J5OB16W02L" +os.environ["OPENAI_API_KEY"] = "tp-ce6eaqg1gh1uev46i4oexh4sahloyfo2hfpxuc2n9m7yl9dd" +os.environ["OPENAI_API_BASE"] = "https://token-plan-cn.xiaomimimo.com/v1" + +# Import after env var is set +from tradingagents.graph.trading_graph import TradingAgentsGraph +from tradingagents.default_config import DEFAULT_CONFIG + +# Configure: OpenRouter + Gemini (free/cheap model) +config = DEFAULT_CONFIG.copy() +config["llm_provider"] = "openai" +config["deep_think_llm"] = "mimo-v2.5-pro" +config["quick_think_llm"] = "mimo-v2.5" +config["llm_max_retries"] = 6 +config["temperature"] = 0.1 # Lower temperature for more stable output +config["data_vendors"] = { + "core_stock_apis": "yfinance", + "technical_indicators": "yfinance", + "fundamental_data": "alpha_vantage", + "news_data": "yfinance", + "macro_data": "fred", + "prediction_markets": "polymarket", +} +config["max_debate_rounds"] = 1 +config["max_risk_discuss_rounds"] = 1 +config["output_language"] = "Chinese" + +print("=" * 60) +print("TradingAgents Smoke Test") +print(f"Provider: {config['llm_provider']}") +print(f"Deep Model: {config['deep_think_llm']}") +print(f"Quick Model: {config['quick_think_llm']}") +print("=" * 60) + +# Test with a simple A-share ticker +ta = TradingAgentsGraph(debug=True, config=config) + +print("\nRunning analysis for 600519.SS (Kweichow Moutai) on 2026-01-15...") +print("(This will take several minutes due to multiple LLM calls)\n") + +try: + _, decision = ta.propagate("AAPL", "2026-01-15") + print("\n" + "=" * 60) + print("DECISION RESULT:") + print("=" * 60) + print(decision) + print("\n✅ Smoke test PASSED!") +except Exception as e: + print(f"\n❌ Smoke test FAILED: {e}") + import traceback + traceback.print_exc() 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/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") 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)