使用Python高效抓取Yahoo Finance历史财报数据

import requests
from bs4 import BeautifulSoup as bs
from datetime import date, timedelta

today = date.today()
# 尝试获取前一个交易日的数据，例如如果今天是周日，则尝试获取周五的数据
# 这里简单计算两天前，可能需要更复杂的逻辑来确保是交易日
yesterday = today - timedelta(days=2) 

url = f"https://finance.yahoo.com/calendar/earnings?day={yesterday.strftime('%Y-%m-%d')}"
page = requests.get(url)
soup = bs(page.text, 'html.parser')

# 尝试查找财报符号
# print(soup.find('a', 'Fw(600) C($linkColor)').text)
# 这种方法经常会因为数据未加载而失败，或者获取到不相关的信息

import requests
from datetime import date, timedelta
import json # 导入json模块以便更好地打印输出

# 定义请求头，模拟浏览器行为
headers = {
    "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:120.0) Gecko/20100101 Firefox/120.0",
}

# API端点
url = "https://query2.finance.yahoo.com/v1/finance/visualization"

# GET请求参数
# 注意：'crumb'参数通常是动态生成的，可能需要从Yahoo Finance页面的某个地方获取，
# 或者通过模拟登录等方式获取。这里使用一个示例值，实际使用时可能需要更新。
params = {
    "crumb": "EwuCwsPbKM2", # 示例值，可能需要更新
    "lang": "en-US",
    "region": "US",
    "corsDomain": "finance.yahoo.com",
}

# POST请求的JSON体 (payload)
# 这是定义查询条件的核心部分
query_payload = {
    "entityIdType": "earnings", # 指定查询实体类型为财报
    "includeFields": [          # 指定需要返回的数据字段
        "ticker",               # 股票代码
        "companyshortname",     # 公司简称
        "eventname",            # 事件名称 (如财报发布、财报电话会议)
        "startdatetime",        # 事件开始时间 (ISO格式)
        "startdatetimetype",    # 事件开始时间类型
        "epsestimate",          # EPS估值
        "epsactual",            # 实际EPS
        "epssurprisepct",       # EPS惊喜百分比
        "timeZoneShortName",    # 时区简称
        "gmtOffsetMilliSeconds",# GMT偏移毫秒数
    ],
    "offset": 0,                # 结果偏移量 (用于分页)
    "query": {                  # 查询条件
        "operands": [
            # 日期范围查询：大于等于开始日期，小于结束日期
            # 示例：获取2023-12-15的财报，则开始日期为2023-12-15，结束日期为2023-12-16
            {"operands": ["startdatetime", "2023-12-15"], "operator": "gte"}, # 大于等于此日期
            {"operands": ["startdatetime", "2023-12-16"], "operator": "lt"},  # 小于此日期
            {"operands": ["region", "us"], "operator": "eq"},                  # 限定地区为美国
        ],
        "operator": "and", # 多个条件之间的逻辑关系
    },
    "size": 100,            # 返回结果数量上限
    "sortField": "companyshortname", # 排序字段
    "sortType": "ASC",      # 排序方式 (升序)
}

# Cookie：'A3' cookie也通常是会话相关的，可能需要更新
# 这是一个示例值，实际使用时很可能需要从浏览器中获取最新的Cookie
cookie = "d=AQABBK8KXmQCEA8-VE0dBLqG5QEpQ7OglmEFEgABCAHCeWWpZfNtb2UB9qMAAAcIqgpeZJj7vK8&S=AQAAAqhyTAOrxcxONc4ktfzCOkg" # 示例值，可能需要更新

# 使用requests.session保持会话，并发送POST请求
with requests.Session() as s:
    s.headers.update(headers) # 更新会话的请求头
    s.cookies["A3"] = cookie  # 设置会话的A3 cookie

    try:
        response = s.post(url, params=params, json=query_payload)
        response.raise_for_status() # 检查HTTP请求是否成功 (2xx状态码)
        data = response.json()      # 解析JSON响应

        # 检查并处理API响应数据
        if data and data["finance"]["result"][0]["documents"][0]["rows"]:
            print(f"成功获取到 {query_payload['query']['operands'][0]['operands'][1]} 的财报数据：")
            print(f"{'公司简称':<40} {'事件名称':<40} {'开始时间':<30}")
            print("-" * 110)
            for r in data["finance"]["result"][0]["documents"][0]["rows"]:
                # r[1] 是 companyshortname, r[2] 是 eventname, r[3] 是 startdatetime
                print(f"{r[1] or 'N/A':<40} {r[2] or 'N/A':<40} {r[3] or 'N/A':<30}")
        else:
            print("未找到指定日期的财报数据。")

    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
    except json.JSONDecodeError:
        print("API响应不是有效的JSON格式。")
    except KeyError as e:
        print(f"解析JSON数据时出错，缺少键: {e}")
        print(f"完整响应数据: {json.dumps(data, indent=2)}") # 打印完整响应以便调试

# 获取昨天的日期
target_date = date.today() - timedelta(days=1)
# 将日期格式化为 YYYY-MM-DD
start_date_str = target_date.strftime("%Y-%m-%d")
# 结束日期设置为目标日期的后一天，以实现精确匹配一天
end_date_str = (target_date + timedelta(days=1)).strftime("%Y-%m-%d")

# 更新query_payload中的日期条件
query_payload["query"]["operands"][0]["operands"][1] = start_date_str
query_payload["query"]["operands"][1]["operands"][1] = end_date_str

# ... (其余代码与上面示例相同，重新执行请求即可获取动态日期的财报)

# 示例：获取上一个交易日
current_date = date.today()
if current_date.weekday() == 0:  # 如果今天是周一 (0代表周一)
    last_trading_day = current_date - timedelta(days=3) # 回溯3天到上周五
elif current_date.weekday() == 6: # 如果今天是周日 (6代表周日)
    last_trading_day = current_date - timedelta(days=2) # 回溯2天到上周五
else: # 其他工作日
    last_trading_day = current_date - timedelta(days=1) # 回溯1天到昨天

start_date_str = last_trading_day.strftime("%Y-%m-%d")
end_date_str = (last_trading_day + timedelta(days=1)).strftime("%Y-%m-%d")

# 更新query_payload并执行请求
# ...