用Python抓取Yahoo Finance：行情报价与历史数据

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install selenium pandas yfinance requests

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install selenium pandas yfinance requests

import sys
import pandas as pd
import requests
import selenium
import yfinance as yf

print("Python:", sys.version.split()[0])
print("pandas:", pd.__version__)
print("requests:", requests.__version__)
print("selenium:", selenium.__version__)
print("yfinance:", yf.__version__)

from datetime import datetime, timezone
import math
import pandas as pd
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.common.exceptions import TimeoutException, WebDriverException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

URL = "https://finance.yahoo.com/quote/AAPL/history/"
SYMBOL = "AAPL"
DATE_RANGE = "REPLACE_WITH_CAPTURED_DATE_RANGE"
FREQUENCY = "REPLACE_WITH_CAPTURED_FREQUENCY"
CURRENCY = "REPLACE_WITH_CAPTURED_CURRENCY"
COLUMNS = ["Date", "Open", "High", "Low", "Close", "Adj Close", "Volume"]

def parse_number(text):
    value = text.strip().replace(",", "")
    if value in {"", "-", "--", "N/A"}:
        return None
    number = float(value)
    if not math.isfinite(number):
        raise ValueError(f"Invalid number: {text}")
    return number

options = webdriver.ChromeOptions()
options.add_argument("--lang=en-US")
driver = None
try:
    driver = webdriver.Chrome(options=options)
    driver.set_page_load_timeout(30)
    driver.get(URL)
    wait = WebDriverWait(driver, 30)
    table = wait.until(EC.presence_of_element_located((
        By.XPATH,
        "//table[.//th[contains(normalize-space(.),'Date')] "
        "and .//th[contains(normalize-space(.),'Adj Close')]]"
    )))

    prices, events = [], []
    for row in table.find_elements(By.CSS_SELECTOR, "tbody tr"):
        cells = [c.text.strip() for c in row.find_elements(By.CSS_SELECTOR, "td")]
        if not cells:
            continue
        date = datetime.strptime(cells[0], "%b %d, %Y").date().isoformat()
        if len(cells) == 2:
            events.append({"Date": date, "Symbol": SYMBOL, "Event": cells[1]})
            continue
        if len(cells) != 7:
            raise ValueError(f"Unexpected row with {len(cells)} cells")
        values = [parse_number(value) for value in cells[1:]]
        op, high, low, close, adj_close, volume = values
        if all(v is not None for v in (op, high, low, close)):
            if not low <= min(op, close) <= max(op, close) <= high:
                raise ValueError(f"OHLC check failed on {date}")
        prices.append(dict(zip(COLUMNS, [date] + values)))

    data = pd.DataFrame(prices, columns=COLUMNS)
    if data.empty or data["Date"].duplicated().any():
        raise ValueError("Empty data or duplicate dates")
    data["Volume"] = data["Volume"].astype("Int64")
    data = data.sort_values("Date").reset_index(drop=True)
    event_data = pd.DataFrame(events, columns=["Date", "Symbol", "Event"])
    metadata = {
        "symbol": SYMBOL, "source": URL, "currency": CURRENCY,
        "date_range": DATE_RANGE, "frequency": FREQUENCY,
        "retrieved_at": datetime.now(timezone.utc).isoformat(),
    }
except (TimeoutException, WebDriverException) as exc:
    raise RuntimeError("Yahoo Finance未加载出可用的历史数据表，请检查同意弹窗、访问策略和页面结构") from exc
finally:
    if driver is not None:
        driver.quit()

print(data.tail())
print(event_data)
print(metadata)

from datetime import datetime, timezone
import yfinance as yf

ticker = yf.Ticker("AAPL")
try:
    history = ticker.history(
        period="3mo", interval="1d", auto_adjust=False,
        actions=True, timeout=20,
    )
except Exception as exc:
    if "429" in str(exc) or "rate limit" in str(exc).lower():
        raise RuntimeError("Yahoo Finance触发限流，请停止请求并执行规定的冷却时间") from exc
    raise

if history.empty:
    raise RuntimeError("Yahoo Finance没有返回历史记录")
required = {"Open", "High", "Low", "Close", "Adj Close", "Volume"}
missing = required.difference(history.columns)
if missing:
    raise RuntimeError(f"缺少字段: {sorted(missing)}")

print(history.tail())
metadata = {
    "symbol": "AAPL", "source": "yfinance", "date_range": "3mo",
    "frequency": "1d", "adjustment": "auto_adjust=False",
    "retrieved_at": datetime.now(timezone.utc).isoformat(),
}
print(metadata)

import json
from pathlib import Path

output = Path("output")
output.mkdir(exist_ok=True)
data.to_csv(output / "AAPL-history.csv", index=False, encoding="utf-8-sig")
event_data.to_csv(output / "AAPL-events.csv", index=False, encoding="utf-8-sig")

if "metadata" not in globals():
    raise RuntimeError("请先运行一条采集路径以生成metadata")
with (output / "AAPL-metadata.json").open("w", encoding="utf-8") as handle:
    json.dump(metadata, handle, ensure_ascii=False, indent=2)

check = pd.read_csv(output / "AAPL-history.csv")
if len(check) != len(data):
    raise RuntimeError("导出校验失败")
print(check.tail())

import os
from urllib.parse import quote
import requests

host = os.environ["ROLA_PROXY_HOST"]
port = os.environ["ROLA_PROXY_PORT"]
user = quote(os.environ["ROLA_PROXY_USER"], safe="")
password = quote(os.environ["ROLA_PROXY_PASSWORD"], safe="")
proxy = f"http://{user}:{password}@{host}:{port}"

session = requests.Session()
session.trust_env = False
session.proxies.update({"http": proxy, "https": proxy})
response = session.get("https://ipinfo.io/json", timeout=20)
if response.status_code == 429:
    retry_after = response.headers.get("Retry-After", "未提供")
    raise RuntimeError(f"代理检查触发429，请停止并遵守Retry-After: {retry_after}")
if response.status_code in {401, 403, 407}:
    raise RuntimeError(f"代理检查失败，HTTP {response.status_code}，请核对授权和配置")
response.raise_for_status()
print(response.json().get("country"))