380 lines
14 KiB
Python
380 lines
14 KiB
Python
import logging
|
|
import os
|
|
import time
|
|
import json
|
|
import subprocess
|
|
from typing import Annotated
|
|
|
|
import pandas as pd
|
|
import yfinance as yf
|
|
from stockstats import wrap
|
|
from yfinance.exceptions import YFRateLimitError
|
|
|
|
from .config import get_config
|
|
from .symbol_utils import NoMarketDataError, normalize_symbol
|
|
from .utils import safe_ticker_component
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# A vendor's latest OHLCV row this many calendar days before the requested date
|
|
# is treated as stale. Generous enough to span long holiday weekends, tight
|
|
# 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 (uses connection pool)."""
|
|
from .ssh_proxy_pool import ssh_proxy_fetch
|
|
return ssh_proxy_fetch(url, timeout)
|
|
|
|
|
|
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.
|
|
|
|
yfinance raises YFRateLimitError on HTTP 429 responses but does not
|
|
retry them internally. This wrapper adds retry logic specifically
|
|
for rate limits. Other exceptions propagate immediately.
|
|
"""
|
|
# Skip local Yahoo Finance entirely - always use SSH proxy
|
|
raise YFRateLimitError("Local Yahoo Finance disabled, use SSH proxy")
|
|
|
|
|
|
def _ensure_date_column(data: pd.DataFrame) -> pd.DataFrame:
|
|
"""Normalize the date column to ``Date``.
|
|
|
|
Some yfinance builds leave the index unnamed (so ``reset_index()`` yields
|
|
``index``) or use ``Datetime`` for intraday data. Rename the first
|
|
date-like column so indicators don't silently drop when it isn't ``Date``.
|
|
"""
|
|
if "Date" in data.columns:
|
|
return data
|
|
for candidate in ("index", "Datetime", "date"):
|
|
if candidate in data.columns:
|
|
return data.rename(columns={candidate: "Date"})
|
|
return data
|
|
|
|
|
|
def _clean_dataframe(data: pd.DataFrame) -> pd.DataFrame:
|
|
"""Normalize a stock DataFrame for stockstats: parse dates, drop invalid rows, fill price gaps."""
|
|
data = _ensure_date_column(data)
|
|
data["Date"] = pd.to_datetime(data["Date"], errors="coerce")
|
|
data = data.dropna(subset=["Date"])
|
|
|
|
price_cols = [c for c in ["Open", "High", "Low", "Close", "Volume"] if c in data.columns]
|
|
data[price_cols] = data[price_cols].apply(pd.to_numeric, errors="coerce")
|
|
data = data.dropna(subset=["Close"])
|
|
data[price_cols] = data[price_cols].ffill().bfill()
|
|
|
|
return data
|
|
|
|
|
|
def _coerce_ohlcv_dates(data: pd.DataFrame) -> pd.Series:
|
|
"""Return parsed dates from an OHLCV frame, whether Date is a column or the index."""
|
|
if "Date" in data.columns:
|
|
return pd.to_datetime(data["Date"], errors="coerce").dropna()
|
|
# yfinance keeps the dates in the index (a DatetimeIndex, sometimes unnamed).
|
|
if isinstance(data.index, pd.DatetimeIndex):
|
|
return pd.Series(pd.to_datetime(data.index, errors="coerce")).dropna()
|
|
# Fallback: expose the index and look for any date-like column.
|
|
df = data.reset_index()
|
|
for col in ("Date", "Datetime", "date", "index"):
|
|
if col in df.columns:
|
|
parsed = pd.to_datetime(df[col], errors="coerce").dropna()
|
|
if not parsed.empty:
|
|
return parsed
|
|
return pd.Series(dtype="datetime64[ns]")
|
|
|
|
|
|
def _assert_ohlcv_not_stale(
|
|
data: pd.DataFrame,
|
|
curr_date: str,
|
|
symbol: str,
|
|
canonical: str | None = None,
|
|
*,
|
|
max_stale_days: int = MAX_OHLCV_STALE_DAYS,
|
|
) -> None:
|
|
"""Reject OHLCV whose latest row is far older than curr_date.
|
|
|
|
Raises NoMarketDataError (with a stale-specific detail) so the router treats
|
|
it like any other "no usable data from this vendor" — try the next vendor,
|
|
then emit one clear unavailable signal. Empty frames are left to the
|
|
caller's existing no-data handling; this guards only the dangerous case of
|
|
present-but-stale rows (a vendor returning a year-old frame that would
|
|
otherwise feed wrong prices to the agent, #1021).
|
|
"""
|
|
if data is None or data.empty:
|
|
return
|
|
requested = pd.to_datetime(curr_date, errors="coerce")
|
|
if pd.isna(requested):
|
|
return
|
|
requested = requested.normalize()
|
|
dates = _coerce_ohlcv_dates(data)
|
|
if dates.empty:
|
|
return
|
|
latest = dates.max().normalize()
|
|
stale_days = (requested - latest).days
|
|
if stale_days > max_stale_days:
|
|
raise NoMarketDataError(
|
|
symbol,
|
|
canonical,
|
|
f"latest row is {latest.date()}, {stale_days} days before the "
|
|
f"requested {requested.date()} (stale) — refusing to use it",
|
|
)
|
|
|
|
|
|
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.
|
|
|
|
Downloads 5 years of data up to today and caches per symbol. On
|
|
subsequent calls the cache is reused. Rows after curr_date are
|
|
filtered out so backtests never see future prices.
|
|
"""
|
|
# Resolve broker/forex symbols (XAUUSD+ -> GC=F) to Yahoo's convention,
|
|
# then reject values that would escape the cache directory when
|
|
# interpolated into the cache filename (e.g. ``../../tmp/x``).
|
|
canonical = normalize_symbol(symbol)
|
|
safe_symbol = safe_ticker_component(canonical)
|
|
|
|
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")
|
|
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}-{cache_prefix}-data-{start_str}-{end_str}.csv",
|
|
)
|
|
|
|
# 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")
|
|
if not cached.empty and "Close" in cached.columns:
|
|
# Ensure Date column is datetime after loading from cache
|
|
if "Date" in cached.columns:
|
|
cached["Date"] = pd.to_datetime(cached["Date"], errors="coerce")
|
|
cached = cached.dropna(subset=["Date"])
|
|
data = cached
|
|
|
|
if data is None:
|
|
if use_alpha_vantage:
|
|
downloaded = _load_ohlcv_alpha_vantage(symbol, curr_date)
|
|
else:
|
|
# Skip local Yahoo Finance, go directly through SSH proxy
|
|
logger.info(f"Yahoo Finance: fetching {canonical} via 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 failed: {proxy_err}"
|
|
) from proxy_err
|
|
if downloaded.empty:
|
|
raise NoMarketDataError(
|
|
symbol, canonical, f"Yahoo Finance proxy 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
|
|
|
|
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
|
|
_assert_ohlcv_not_stale(data, curr_date, symbol, canonical)
|
|
|
|
return data
|
|
|
|
|
|
def filter_financials_by_date(data: pd.DataFrame, curr_date: str) -> pd.DataFrame:
|
|
"""Drop financial statement columns (fiscal period timestamps) after curr_date.
|
|
|
|
yfinance financial statements use fiscal period end dates as columns.
|
|
Columns after curr_date represent future data and are removed to
|
|
prevent look-ahead bias.
|
|
"""
|
|
if not curr_date or data.empty:
|
|
return data
|
|
cutoff = pd.Timestamp(curr_date)
|
|
mask = pd.to_datetime(data.columns, errors="coerce") <= cutoff
|
|
return data.loc[:, mask]
|
|
|
|
|
|
class StockstatsUtils:
|
|
@staticmethod
|
|
def get_stock_stats(
|
|
symbol: Annotated[str, "ticker symbol for the company"],
|
|
indicator: Annotated[
|
|
str, "quantitative indicators based off of the stock data for the company"
|
|
],
|
|
curr_date: Annotated[
|
|
str, "curr date for retrieving stock price data, YYYY-mm-dd"
|
|
],
|
|
):
|
|
data = load_ohlcv(symbol, curr_date)
|
|
df = wrap(data)
|
|
df["Date"] = df["Date"].dt.strftime("%Y-%m-%d")
|
|
curr_date_str = pd.to_datetime(curr_date).strftime("%Y-%m-%d")
|
|
|
|
df[indicator] # trigger stockstats to calculate the indicator
|
|
matching_rows = df[df["Date"].str.startswith(curr_date_str)]
|
|
|
|
if not matching_rows.empty:
|
|
indicator_value = matching_rows[indicator].values[0]
|
|
return indicator_value
|
|
else:
|
|
return "N/A: Not a trading day (weekend or holiday)"
|