网站 首页 关键词企业线上推广方案

网站 首页 关键词,企业线上推广方案,html网页源代码查看,网站 二级域名7.3.4 从Yahoo Finance抓取和解析金融数据文件yahoo_finance_utils.py定义了一个名为yahoo_finance_utils.py的模块#xff0c;它包含了一系列用于从Yahoo Finance网站抓取和解析金融数据的函数。这些函数可以异步地获取收益日历数据、市场动态#xff08;如涨幅最大、跌幅最…7.3.4 从Yahoo Finance抓取和解析金融数据文件yahoo_finance_utils.py定义了一个名为yahoo_finance_utils.py的模块它包含了一系列用于从Yahoo Finance网站抓取和解析金融数据的函数。这些函数可以异步地获取收益日历数据、市场动态如涨幅最大、跌幅最大和交易量最大的股票等信息。在代码中使用了httpx库来发送HTTP请求并利用pandas和BeautifulSoup库来解析和处理数据。此外代码中还包含了一个装饰器函数retry_on_rate_limit用于在遇到速率限制错误时重试函数调用。1下面代码的功能是定义一组HTTP请求头YAHOO_HEADERS这些请求头模拟了一个常见的浏览器环境用于在向Yahoo Finance网站发起请求时伪装成浏览器访问以避免被网站识别为爬虫而阻止访问。YAHOO_HEADERS { User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: en-US,en;q0.5, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Upgrade-Insecure-Requests: 1, }2下面代码的功能是创建一个异步HTTP客户端create_cached_async_client该客户端通过hishel库启用了缓存功能可以提高重复请求相同资源时的效率。同时该函数允许传入自定义的请求头。def create_cached_async_client(headers: dict | None None) - httpx.AsyncClient: 创建一个启用了 hishel 缓存的 httpx.AsyncClient。 hishel.install_cache() return httpx.AsyncClient( timeout30.0, follow_redirectsTrue, headersheaders, )3下面代码的功能是定义一个convert_to_numeric函数用于将包含单位后缀如K、M、B、T的字符串转换为数值类型以便进行后续的数值计算和分析。def convert_to_numeric(value_str): 将类似 1.2M、3.45B、123.4K 这样的字符串值转换为数值。 if pd.isna(value_str) or value_str in (, -): return None value_str str(value_str).strip().replace(,, ) try: return float(value_str) except ValueError: pass # 处理带后缀的值K、M、B、T multipliers {K: 1_000, M: 1_000_000, B: 1_000_000_000, T: 1_000_000_000_000} for suffix, multiplier in multipliers.items(): if value_str.upper().endswith(suffix): try: return float(value_str[:-1]) * multiplier except ValueError: pass return value_str4下面代码的功能是定义一个异步函数_parse_earnings_json用于从Yahoo Finance的特定URL解析收益相关的JSON数据。该函数尝试使用预定义的正则表达式模式从网页内容中提取JSON数据并将其转换为Python字典。async def _parse_earnings_json(url: str) - dict: 使用现有的异步基础架构从雅虎财经 URL 解析收益 JSON。 async with create_cached_async_client(headersYAHOO_HEADERS) as client: response await client.get(url) response.raise_for_status() content response.text # 尝试过去有效的模式 patterns [ rroot\.App\.main\s*\s*({.*?});, rwindow\.App\.main\s*\s*({.*?}); ] for pattern_name, pattern in zip([root.App.main, window.App.main], patterns): match re.search(pattern, content, re.DOTALL) if match: try: data json.loads(match.group(1)) logger.info(f使用 {pattern_name} 模式成功解析收益数据) return data except json.JSONDecodeError as e: logger.warning(f解析 {pattern_name} JSON 失败{e}) raise ValueError(无法使用原始模式找到收益数据)5下面代码的功能是定义一个异步函数get_earnings_for_date用于获取特定日期的收益数据支持分页功能。该函数通过递归调用自身逐步获取所有分页的收益数据并将它们合并为一个列表。async def get_earnings_for_date(date, offset0, count1): 获取特定日期的收益带分页是原有有效代码的异步版本。 base_earnings_url https://finance.yahoo.com/calendar/earnings if offset count: return [] temp pd.Timestamp(date) date temp.strftime(%Y-%m-%d) dated_url {0}?day{1}offset{2}size{3}.format( base_earnings_url, date, offset, 100) result await _parse_earnings_json(dated_url) stores result[context][dispatcher][stores] earnings_count stores[ScreenerCriteriaStore][meta][total] new_offset offset 100 more_earnings await get_earnings_for_date(date, new_offset, earnings_count) current_earnings stores[ScreenerResultsStore][results][rows] total_earnings current_earnings more_earnings return total_earnings6下面代码的功能是定义一个异步函数get_earnings_in_date_range用于获取指定日期范围内的收益数据。该函数通过计算起始日期和结束日期之间的天数逐天调用get_earnings_for_date函数来获取每一天的收益数据并将它们合并。async def get_earnings_in_date_range(start_date, end_date): 获取日期范围内的收益是原有有效代码的异步版本。 import datetime earnings_data [] days_diff pd.Timestamp(end_date) - pd.Timestamp(start_date) days_diff days_diff.days current_date pd.Timestamp(start_date) dates [current_date datetime.timedelta(diff) for diff in range(days_diff 1)] dates [d.strftime(%Y-%m-%d) for d in dates] i 0 while i len(dates): try: earnings_data await get_earnings_for_date(dates[i]) except Exception: pass i 1 return earnings_data7下面代码的功能是定义一个异步函数get_earnings_calendar_data使用Playwright浏览器自动化工具获取收益日历数据并支持分页。该函数通过模拟浏览器访问Yahoo Finance的收益日历页面提取页面中的表格数据并将其转换为结构化的收益数据。async def get_earnings_calendar_data( start_date: str None, end_date: str None, limit: int 100 ) - dict: 使用 Playwright 并支持分页来获取收益日历数据。 try: from playwright.async_api import async_playwright except ImportError: raise ImportError(Playwright 不可用。请使用uvx investor-agent[playwright] 安装) # 设置默认日期 start_date start_date or datetime.date.today().strftime(%Y-%m-%d) end_date end_date or (pd.Timestamp(start_date) pd.DateOffset(days7)).strftime(%Y-%m-%d) all_earnings [] offset 0 async with async_playwright() as p: async with await p.chromium.launch(headlessTrue) as browser: context await browser.new_context( user_agentMozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 ) page await context.new_page() # 持续获取页面直到获取足够数据或没有更多页面 while len(all_earnings) limit: url fhttps://finance.yahoo.com/calendar/earnings?from{start_date}to{end_date}offset{offset}size100 logger.info(f使用 Playwright 加载收益页面 {offset//100 1}) await page.goto(url, wait_untildomcontentloaded, timeout60000) await page.wait_for_timeout(3000) # 从当前页面提取收益数据 earnings_data await page.eval_on_selector_all(table tr, (rows) rows.slice(1).map(row { const cells Array.from(row.querySelectorAll(td, th)).map(c c.textContent.trim()); // 只需要有效的代码——让其他字段自然设为空 const symbol cells[0] || ; if (!symbol || symbol Symbol || cells.length 0) return null; return { symbol, company: cells[1] || , event_name: cells[2] || , time: cells[3] || , eps_estimate: cells[4] || , eps_actual: cells[5] || , surprise_percent: cells[6] || , market_cap: cells[7] || }; }).filter(Boolean) ) if not earnings_data: logger.info(未找到更多收益数据——已到结果末尾) break all_earnings.extend(earnings_data) logger.info(f从第 {offset//100 1} 页获取了 {len(earnings_data)} 条收益数据) # 如果获取的数据少于页面大小说明已到最后一页 if len(earnings_data) 100: logger.info(f获取了 {len(earnings_data)} 条 100 条收益数据——已到最后一页) break offset 100 await page.wait_for_timeout(1000) # 转换为结构化收益数据限制为请求的数量 earnings_list [ { symbol: str(item.get(symbol, )), company: str(item.get(company, )), event_name: str(item.get(event_name, )), time: str(item.get(time, )), eps_estimate: convert_to_numeric(item.get(eps_estimate, )), eps_actual: convert_to_numeric(item.get(eps_actual, )), surprise_percent: convert_to_numeric(item.get(surprise_percent, )), market_cap: str(item.get(market_cap, )) } for item in all_earnings[:limit] if item.get(symbol) ] return { metadata: { start_date: start_date, end_date: end_date, count: len(earnings_list), total_found: len(all_earnings), pages_fetched: (offset // 100) 1, timestamp: pd.Timestamp.now().isoformat(), source: Yahoo Finance Playwright Scraping (Paginated) }, earnings: earnings_list }8下面代码的功能是定义一个异步函数get_market_movers_data用于从Yahoo Finance获取市场动态数据如涨幅最大、跌幅最大和交易量最大的股票。该函数根据传入的类别和市场时段构建相应的URL然后使用httpx库异步获取网页内容并通过pandas库解析HTML表格数据最后将解析后的数据转换为字典格式返回。async def get_market_movers_data( category: Literal[gainers, losers, most-active] most-active, count: int 25, market_session: Literal[regular, pre-market, after-hours] regular ) - dict: 从雅虎财经获取并解析市场动态数据。 from bs4 import BeautifulSoup count min(max(count, 1), 100) # 根据类别和市场时段构建 URL if category most-active: if market_session regular: url fhttps://finance.yahoo.com/most-active?count{count}offset0 elif market_session pre-market: url fhttps://finance.yahoo.com/markets/stocks/pre-market?count{count}offset0 elif market_session after-hours: url fhttps://finance.yahoo.com/markets/stocks/after-hours?count{count}offset0 else: raise ValueError(f无效的市场时段{market_session}) else: # 涨幅榜和跌幅榜仅适用于常规时段 url_map { gainers: fhttps://finance.yahoo.com/gainers?count{count}offset0, losers: fhttps://finance.yahoo.com/losers?count{count}offset0 } url url_map.get(category) if not url: raise ValueError(f无效的类别{category}) async with create_cached_async_client(headersYAHOO_HEADERS) as client: logger.info(f从以下地址获取 {category}{market_session} 时段数据{url}) response await client.get(url) response.raise_for_status() # 用 pandas 解析 tables pd.read_html(response.content) if not tables: raise ValueError(f未找到 {category} 的数据) df tables[0].copy() # 清理数据 df df.drop(52 Week Range, axis1, errorsignore) # 清理涨跌幅列 if % Change in df.columns: df[% Change] df[% Change].astype(str).str.replace([%,], , regexTrue) df[% Change] pd.to_numeric(df[% Change], errorscoerce) # 清理数值列 numeric_cols [col for col in df.columns if any(x in col for x in [Vol, Volume, Market Cap, Market])] for col in numeric_cols: df[col] df[col].astype(str).apply(convert_to_numeric) return { metadata: { category: category, market_session: market_session if category most-active else regular, count: len(df), timestamp: pd.Timestamp.now().isoformat(), source: Yahoo Finance }, stocks: df.head(count).to_dict(records) }