小钰
51d057e2c9
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/
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
#!/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")
|