分享1个美股数据接口(附源码)

一个轻量级 Python 工具类，整合新浪财经、老虎证券、雅虎财经三个数据源，提供美股实时行情、深度基本面和历史K线数据的获取能力。

环境要求

• Python 3.8+
• 依赖：requests

pip install requests

数据源概览

API 方法说明

get_sina_quote(symbol)

获取新浪财经实时行情快照。

参数：symbol — 股票代码，如 "NVDA"

返回字段：

get_laohu_quote(symbol)

获取老虎证券详细行情，包含盘前/盘后交易数据。

参数：symbol — 股票代码，如 "NVDA"

返回字段：

get_laohu_fundamental(symbol)

获取老虎证券基本面数据。

参数：symbol — 股票代码，如 "NVDA"

返回字段：

get_laohu_profile(symbol)

获取老虎证券公司基本信息。

参数：symbol — 股票代码，如 "NVDA"

返回字段：

get_yahoo_history(symbol, range_str, interval)

获取雅虎财经历史K线数据。

参数：

返回字段：

注意：分时数据（1m/5m）仅支持最近 7 天，日线可拉取全部历史。

使用示例

from test_us_stock_apis import USStockAPIapi = USStockAPI()# 实时报价sina = api.get_sina_quote("NVDA")print(f"最新价: {sina['latest_price']}")# 盘前盘后quote = api.get_laohu_quote("NVDA")print(f"盘后价: {quote['extended_hours_price']}")# 基本面筛选fund = api.get_laohu_fundamental("NVDA")print(f"PE: {fund['pe_ttm']}, ROE: {fund['roe']}")# 历史K线history = api.get_yahoo_history("NVDA", range_str="1y", interval="1d")print(f"获取 {history['count']} 根日K线")

注意事项

1. 老虎证券接口建议控制请求频率在每秒 1 次以内
2. 未登录状态下部分数据可能存在 15 分钟延迟
3. 接口均为公开网页逆向提取，存在变动可能
4. 仅限个人研究使用

附录：完整源代码

import requestsimport reimport jsonimport loggingfrom typing import Dict, Any, Optionallogging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')classUSStockAPI:"""US Stock Data Retrieval API wrapper."""def__init__(self):        self.session = requests.Session()        self.session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"        })        self._laohu_token: Optional[str] = Nonedefget_sina_quote(self, symbol: str) -> Dict[str, Any]:"""Fetch basic real-time quote from Sina Finance."""        url = f"https://hq.sinajs.cn/list=gb_{symbol.lower()}"        headers = {"Referer": "https://finance.sina.com.cn/"}try:            res = requests.get(url, headers=headers, timeout=5)            res.raise_for_status()            res.encoding = 'gbk'            raw_data = res.text.strip()if'="'in raw_data:                data_str = raw_data.split('="')[1].strip('";')if data_str:                    fields = data_str.split(',')return {"symbol": fields[0],"latest_price": float(fields[1]),"change_pct": float(fields[2]),"local_time": fields[3],"change_amount": float(fields[4]),"open": float(fields[5]),"high": float(fields[6]),"low": float(fields[7]),"week52_high": float(fields[8]),"week52_low": float(fields[9]),"volume": int(fields[10]),                    }except Exception as e:            logging.error(f"Sina quote fetch failed for {symbol}: {e}")return {}def_refresh_laohu_token(self, symbol: str) -> bool:        url = f"https://www.laohu8.com/stock/{symbol.lower()}"try:            res = self.session.get(url, timeout=10)            res.raise_for_status()            match = re.search(r'(eyJhbGciOi[a-zA-Z0-9_.-]+)', res.text)if match:                self._laohu_token = match.group(1)                self.session.headers.update({"Authorization": f"Bearer {self._laohu_token}","Accept": "application/json"                })returnTrueexcept Exception:passreturnFalsedef_laohu_request(self, url: str, params: Dict[str, Any], symbol: str) -> dict:ifnot self._laohu_token andnot self._refresh_laohu_token(symbol):return {}        params.update({"lang": "zh_CN", "lang_content": "all", "region": "USA", "vendor": "web"})try:            res = self.session.get(url, params=params, timeout=10)if res.status_code == 401:if self._refresh_laohu_token(symbol):                    res = self.session.get(url, params=params, timeout=10)if res.status_code == 200:return res.json()except Exception:passreturn {}defget_laohu_quote(self, symbol: str) -> Dict[str, Any]:"""Fetch detailed quote including pre/post market from Tiger Brokers."""        url = f"https://hq.laohu8.com/stock_info/detail/{symbol.upper()}"        data = self._laohu_request(url, {}, symbol)if data and"items"in data and data["items"]:            stock = data["items"][0]            result = {"name": stock.get('nameCN'),"latest_price": stock.get('latestPrice'),"pre_close": stock.get('preClose'),"volume": stock.get('volume'),"amount": stock.get('amount'),"change_rate": stock.get('changeRate', 0),"amplitude": stock.get('amplitude', 0),            }            ht = stock.get("hourTrading")if ht:                result["extended_hours_tag"] = ht.get('tag')                result["extended_hours_price"] = ht.get('latestPrice')                result["extended_hours_change_rate"] = ht.get('changeRate', 0)return resultreturn {}defget_laohu_fundamental(self, symbol: str) -> Dict[str, Any]:"""Fetch fundamental data from Tiger Brokers."""        url = f"https://hq.laohu8.com/stock_info/fundamental/{symbol.upper()}"        data = self._laohu_request(url, {}, symbol)if data and"items"in data and data["items"]:            fund = data["items"][0]            latest_price = fund.get('latestPrice', 0)            ttm_eps = fund.get('ttmEps', 1)            shares = fund.get('shares', 0)            pe_ttm = round(latest_price / ttm_eps, 2) if ttm_eps elseNone            market_value = (shares * latest_price) if shares and latest_price elseNonereturn {"roa": fund.get('roa'),"roe": fund.get('roe'),"pb_rate": fund.get('pbRate'),"ps_rate": fund.get('psRate'),"institution_held": fund.get('institutionHeld', 0),"ttm_eps": fund.get('ttmEps'),"forward_eps": fund.get('forwardEps'),"shares": fund.get('shares'),"pe_ttm": pe_ttm,"market_value": market_value,"week52_high": fund.get('week52High'),"week52_low": fund.get('week52Low'),            }return {}defget_laohu_profile(self, symbol: str) -> Dict[str, Any]:"""Fetch corporate profile from Tiger Brokers."""        url = "https://hq.laohu8.com/fundamental/company/profile"        params = {"symbol": symbol.upper(), "market": "US"}        data = self._laohu_request(url, params, symbol)if data and"data"in data:            prof = data["data"]return {"year_founded": prof.get('yearFounded'),"employee_num": prof.get('employeeNum'),"address": prof.get('address'),"office_phone": prof.get('officePhone'),"fiscal_year_end": prof.get('fiscalYearEnd')            }return {}defget_yahoo_history(self, symbol: str, range_str: str = "1mo", interval: str = "1d") -> Dict[str, Any]:"""Fetch historical k-line data from Yahoo Finance."""        url = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol.upper()}"        params = {"range": range_str, "interval": interval}try:            res = requests.get(url, params=params, headers={"User-Agent": "Mozilla/5.0"}, timeout=10)            res.raise_for_status()            data = res.json()if"chart"in data and"result"in data["chart"] and data["chart"]["result"]:                result_data = data["chart"]["result"][0]                timestamps = result_data.get("timestamp", [])                indicators = result_data.get("indicators", {}).get("quote", [{}])[0]if timestamps and indicators:return {"count": len(timestamps),"interval": interval,"range": range_str,"timestamps": timestamps,"open": indicators.get("open", []),"high": indicators.get("high", []),"low": indicators.get("low", []),"close": indicators.get("close", []),"volume": indicators.get("volume", [])                    }except Exception as e:            logging.error(f"Yahoo history fetch failed for {symbol}: {e}")return {}if __name__ == "__main__":    api = USStockAPI()    test_symbol = "NVDA"    logging.info(f"Testing APIs for symbol: {test_symbol}")    sina_data = api.get_sina_quote(test_symbol)    logging.info(f"Sina Quote: {json.dumps(sina_data, indent=2, ensure_ascii=False)}")    laohu_q_data = api.get_laohu_quote(test_symbol)    logging.info(f"Tiger Quote: {json.dumps(laohu_q_data, indent=2, ensure_ascii=False)}")    laohu_f_data = api.get_laohu_fundamental(test_symbol)    logging.info(f"Tiger Fundamental: {json.dumps(laohu_f_data, indent=2, ensure_ascii=False)}")    laohu_p_data = api.get_laohu_profile(test_symbol)    logging.info(f"Tiger Profile: {json.dumps(laohu_p_data, indent=2, ensure_ascii=False)}")    yahoo_h_data = api.get_yahoo_history(test_symbol)if yahoo_h_data:        logging.info(f"Yahoo History: {yahoo_h_data['count']} points retrieved.")