Spaces:
Sleeping
Sleeping
| import os | |
| from fastapi import FastAPI | |
| import yfinance as yf | |
| import math | |
| import pandas as pd | |
| import uvicorn | |
| app = FastAPI() | |
| def clean_num(val): | |
| try: | |
| if val is None: return None | |
| f_val = float(val) | |
| if math.isnan(f_val) or math.isinf(f_val): return None | |
| return f_val | |
| except: | |
| return None | |
| def home(): | |
| return {"status": "Stock API is running", "endpoint": "/stock/{ticker}"} | |
| def get_stock(ticker: str): | |
| try: | |
| stock = yf.Ticker(ticker.upper()) | |
| # 1. 尝试从 stock.info 获取数据 | |
| info = stock.info | |
| # 2. 尝试获取详细的分析师预测表 | |
| try: | |
| est = stock.earnings_estimate | |
| except Exception: | |
| est = None | |
| eps_est_current = None | |
| eps_est_next = None | |
| if est is not None and not est.empty: | |
| try: | |
| # 方法 A: 使用位置索引 iloc | |
| eps_est_current = est.iloc[2]['avg'] # 当前财年 | |
| eps_est_next = est.iloc[3]['avg'] # 下一财年 | |
| except Exception as e: | |
| print(f"Error parsing dataframe: {e}") | |
| # 3. 如果从表格获取失败,使用 info 中的备用数据 | |
| if eps_est_next is None and info: | |
| eps_est_next = info.get('forwardEps') | |
| if not info or len(info) < 5: | |
| return {"error": "No data found", "ticker": ticker} | |
| try: | |
| ept = stock.eps_trend | |
| except Exception: | |
| ept = None | |
| eps_current = None | |
| if ept is not None and not ept.empty: | |
| try: | |
| eps_current = ept.iloc[2]['current'] # 当前财年 | |
| except Exception as e: | |
| print(f"Error parsing dataframe: {e}") | |
| return { | |
| "ticker": ticker.upper(), | |
| "pe_static": clean_num(info.get('trailingPE')), | |
| "pe_dynamic": clean_num(info.get('forwardPE')), | |
| "eps_est_avg_current_year": clean_num(eps_est_current), | |
| "eps_est_avg_next_year": clean_num(eps_est_next), | |
| "eps_avg_current_year": clean_num(eps_current), | |
| } | |
| except Exception as e: | |
| return {"error": str(e)} | |
| if __name__ == "__main__": | |
| uvicorn.run(app, host="0.0.0.0", port=7860) |