import yfinance as yf import akshare as ak import pandas as pd import numpy as np import ta from datetime import datetime, timedelta import requests import json class StockDataFetcher: """股票数据获取类""" def __init__(self): self.data = None self.info = None self.financial_data = None def get_stock_info(self, symbol): """获取股票基本信息""" try: # 处理中国股票代码 if self._is_chinese_stock(symbol): return self._get_chinese_stock_info(symbol) else: return self._get_us_stock_info(symbol) except Exception as e: return {"error": f"获取股票信息失败: {str(e)}"} def get_stock_data(self, symbol, period="1y", interval="1d"): """获取股票历史数据""" try: if self._is_chinese_stock(symbol): return self._get_chinese_stock_data(symbol, period) else: return self._get_us_stock_data(symbol, period, interval) except Exception as e: return {"error": f"获取股票数据失败: {str(e)}"} def _is_chinese_stock(self, symbol): """判断是否为中国股票""" # 简单判断：包含数字且长度为6位的认为是中国股票 return symbol.isdigit() and len(symbol) == 6 def _get_chinese_stock_info(self, symbol): """获取中国股票基本信息""" try: # 初始化基本信息 info = { "symbol": symbol, "name": "未知", "current_price": "N/A", "change_percent": "N/A", "pe_ratio": "N/A", "pb_ratio": "N/A", "market_cap": "N/A", "market": "中国A股", "exchange": "上海/深圳证券交易所" } # 方法1: 尝试获取个股详细信息 try: stock_info = ak.stock_individual_info_em(symbol=symbol) if stock_info is not None and not stock_info.empty: for _, row in stock_info.iterrows(): key = row['item'] value = row['value'] if key == '股票简称': info['name'] = value elif key == '总市值': try: if value and value != '-': info['market_cap'] = float(value) except: pass elif key == '市盈率-动态': try: if value and value != '-': pe_value = float(value) if 0 < pe_value <= 1000: info['pe_ratio'] = pe_value except: pass elif key == '市净率': try: if value and value != '-': pb_value = float(value) if 0 < pb_value <= 100: info['pb_ratio'] = pb_value except: pass except Exception as e: print(f"获取个股详细信息失败: {e}") # 方法2: 尝试获取实时价格和涨跌幅（如果网络允许） try: # 使用更简单的接口获取实时价格 real_time_data = ak.stock_zh_a_spot_em() if real_time_data is not None and not real_time_data.empty: stock_real_time = real_time_data[real_time_data['代码'] == symbol] if not stock_real_time.empty: row = stock_real_time.iloc[0] info['current_price'] = row.get('最新价', 'N/A') info['change_percent'] = row.get('涨跌幅', 'N/A') if info['name'] == '未知': info['name'] = row.get('名称', '未知') # 如果实时数据中有市盈率和市净率，优先使用 if '市盈率-动态' in row and info['pe_ratio'] == 'N/A': try: pe_val = row['市盈率-动态'] if pe_val and pe_val != '-': pe_val = float(pe_val) if 0 < pe_val <= 1000: info['pe_ratio'] = pe_val except: pass if '市净率' in row and info['pb_ratio'] == 'N/A': try: pb_val = row['市净率'] if pb_val and pb_val != '-': pb_val = float(pb_val) if 0 < pb_val <= 100: info['pb_ratio'] = pb_val except: pass except Exception as e: print(f"获取实时数据失败: {e}") # 如果实时数据获取失败，尝试使用历史数据获取价格 try: hist_data = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=(datetime.now() - timedelta(days=5)).strftime('%Y%m%d'), end_date=datetime.now().strftime('%Y%m%d'), adjust="qfq") if hist_data is not None and not hist_data.empty: latest = hist_data.iloc[-1] info['current_price'] = latest['收盘'] # 计算涨跌幅 if len(hist_data) > 1: prev_close = hist_data.iloc[-2]['收盘'] change_pct = ((latest['收盘'] - prev_close) / prev_close) * 100 info['change_percent'] = round(change_pct, 2) except Exception as e2: print(f"获取历史数据也失败: {e2}") # 方法3: 使用百度估值数据获取市盈率和市净率 if info['pe_ratio'] == 'N/A': try: pe_data = ak.stock_zh_valuation_baidu(symbol=symbol, indicator="市盈率(TTM)") if pe_data is not None and not pe_data.empty: latest_pe = pe_data.iloc[-1]['value'] if latest_pe and latest_pe != '-': pe_val = float(latest_pe) if 0 < pe_val <= 1000: info['pe_ratio'] = pe_val except Exception as e: print(f"获取市盈率失败: {e}") if info['pb_ratio'] == 'N/A': try: pb_data = ak.stock_zh_valuation_baidu(symbol=symbol, indicator="市净率") if pb_data is not None and not pb_data.empty: latest_pb = pb_data.iloc[-1]['value'] if latest_pb and latest_pb != '-': pb_val = float(latest_pb) if 0 < pb_val <= 100: info['pb_ratio'] = pb_val except Exception as e: print(f"获取市净率失败: {e}") return info except Exception as e: print(f"获取中国股票信息完全失败: {e}") # 返回基本信息，避免完全失败 return { "symbol": symbol, "name": f"股票{symbol}", "current_price": "N/A", "change_percent": "N/A", "pe_ratio": "N/A", "pb_ratio": "N/A", "market_cap": "N/A", "market": "中国A股", "exchange": "上海/深圳证券交易所" } def _get_us_stock_info(self, symbol): """获取美股基本信息""" import time try: # 添加延迟避免频率限制 time.sleep(1) ticker = yf.Ticker(symbol) # 先尝试获取历史数据（通常更稳定） try: hist = ticker.history(period="2d") if not hist.empty: current_price = hist['Close'].iloc[-1] if len(hist) > 1: prev_close = hist['Close'].iloc[-2] change_percent = ((current_price - prev_close) / prev_close) * 100 else: change_percent = 'N/A' else: current_price = 'N/A' change_percent = 'N/A' except: current_price = 'N/A' change_percent = 'N/A' # 获取基本信息 try: info = ticker.info # 获取市盈率，优先使用trailing PE，其次forward PE pe_ratio = info.get('trailingPE', info.get('forwardPE', 'N/A')) if pe_ratio == 'N/A' or pe_ratio is None or (isinstance(pe_ratio, float) and np.isnan(pe_ratio)): pe_ratio = 'N/A' # 获取市净率 pb_ratio = info.get('priceToBook', 'N/A') if pb_ratio == 'N/A' or pb_ratio is None or (isinstance(pb_ratio, float) and np.isnan(pb_ratio)): pb_ratio = 'N/A' # 如果历史数据没有获取到价格，尝试从info获取 if current_price == 'N/A': current_price = info.get('currentPrice', info.get('regularMarketPrice', 'N/A')) if change_percent == 'N/A': change_percent = info.get('regularMarketChangePercent', 'N/A') if change_percent != 'N/A' and change_percent is not None: change_percent = change_percent * 100 # 转换为百分比 return { "symbol": symbol, "name": info.get('longName', info.get('shortName', 'N/A')), "current_price": current_price, "change_percent": change_percent, "market_cap": info.get('marketCap', 'N/A'), "pe_ratio": pe_ratio, "pb_ratio": pb_ratio, "dividend_yield": info.get('dividendYield', 'N/A'), "beta": info.get('beta', 'N/A'), "52_week_high": info.get('fiftyTwoWeekHigh', 'N/A'), "52_week_low": info.get('fiftyTwoWeekLow', 'N/A'), "sector": info.get('sector', 'N/A'), "industry": info.get('industry', 'N/A'), "market": "美股", "exchange": info.get('exchange', 'N/A') } except Exception as e: # 如果获取详细信息失败，返回基本价格信息 return { "symbol": symbol, "name": f"美股{symbol}", "current_price": current_price, "change_percent": change_percent, "market_cap": 'N/A', "pe_ratio": 'N/A', "pb_ratio": 'N/A', "dividend_yield": 'N/A', "beta": 'N/A', "52_week_high": 'N/A', "52_week_low": 'N/A', "sector": 'N/A', "industry": 'N/A', "market": "美股", "exchange": 'N/A' } except Exception as e: return {"error": f"获取美股信息失败: {str(e)}"} def _get_chinese_stock_data(self, symbol, period="1y"): """获取中国股票历史数据""" try: # 计算日期范围 end_date = datetime.now().strftime('%Y%m%d') if period == "1y": start_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d') elif period == "6mo": start_date = (datetime.now() - timedelta(days=180)).strftime('%Y%m%d') elif period == "3mo": start_date = (datetime.now() - timedelta(days=90)).strftime('%Y%m%d') else: start_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d') # 获取历史数据 df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq") if df is not None and not df.empty: # 重命名列以匹配标准格式 df = df.rename(columns={ '日期': 'Date', '开盘': 'Open', '收盘': 'Close', '最高': 'High', '最低': 'Low', '成交量': 'Volume' }) df['Date'] = pd.to_datetime(df['Date']) df.set_index('Date', inplace=True) return df else: return {"error": "无法获取历史数据"} except Exception as e: return {"error": f"获取中国股票数据失败: {str(e)}"} def _get_us_stock_data(self, symbol, period="1y", interval="1d"): """获取美股历史数据""" try: ticker = yf.Ticker(symbol) df = ticker.history(period=period, interval=interval) if not df.empty: return df else: return {"error": "无法获取历史数据"} except Exception as e: return {"error": f"获取美股数据失败: {str(e)}"} def calculate_technical_indicators(self, df): """计算技术指标""" try: if isinstance(df, dict) and "error" in df: return df # 移动平均线 df['MA5'] = ta.trend.sma_indicator(df['Close'], window=5) df['MA10'] = ta.trend.sma_indicator(df['Close'], window=10) df['MA20'] = ta.trend.sma_indicator(df['Close'], window=20) df['MA60'] = ta.trend.sma_indicator(df['Close'], window=60) # RSI df['RSI'] = ta.momentum.rsi(df['Close'], window=14) # MACD macd = ta.trend.MACD(df['Close']) df['MACD'] = macd.macd() df['MACD_signal'] = macd.macd_signal() df['MACD_histogram'] = macd.macd_diff() # 布林带 bollinger = ta.volatility.BollingerBands(df['Close']) df['BB_upper'] = bollinger.bollinger_hband() df['BB_middle'] = bollinger.bollinger_mavg() df['BB_lower'] = bollinger.bollinger_lband() # KDJ指标 df['K'] = ta.momentum.stoch(df['High'], df['Low'], df['Close']) df['D'] = ta.momentum.stoch_signal(df['High'], df['Low'], df['Close']) # 成交量指标 df['Volume_MA5'] = ta.trend.sma_indicator(df['Volume'], window=5) df['Volume_ratio'] = df['Volume'] / df['Volume_MA5'] return df except Exception as e: return {"error": f"计算技术指标失败: {str(e)}"} def get_latest_indicators(self, df): """获取最新的技术指标值""" try: if isinstance(df, dict) and "error" in df: return df latest = df.iloc[-1] return { "price": latest['Close'], "ma5": latest['MA5'], "ma10": latest['MA10'], "ma20": latest['MA20'], "ma60": latest['MA60'], "rsi": latest['RSI'], "macd": latest['MACD'], "macd_signal": latest['MACD_signal'], "bb_upper": latest['BB_upper'], "bb_lower": latest['BB_lower'], "k_value": latest['K'], "d_value": latest['D'], "volume_ratio": latest['Volume_ratio'] } except Exception as e: return {"error": f"获取最新指标失败: {str(e)}"} def get_financial_data(self, symbol): """获取详细财务数据""" try: if self._is_chinese_stock(symbol): return self._get_chinese_financial_data(symbol) else: return self._get_us_financial_data(symbol) except Exception as e: return {"error": f"获取财务数据失败: {str(e)}"} def _get_chinese_financial_data(self, symbol): """获取中国股票财务数据""" financial_data = { "symbol": symbol, "balance_sheet": None, # 资产负债表 "income_statement": None, # 利润表 "cash_flow": None, # 现金流量表 "financial_ratios": {}, # 财务比率 "quarter_data": None, # 季度数据 } try: # 1. 获取资产负债表 try: balance_sheet = ak.stock_financial_abstract_ths(symbol=symbol, indicator="资产负债表") if balance_sheet is not None and not balance_sheet.empty: financial_data["balance_sheet"] = balance_sheet.head(8).to_dict('records') except Exception as e: print(f"获取资产负债表失败: {e}") # 2. 获取利润表 try: income_statement = ak.stock_financial_abstract_ths(symbol=symbol, indicator="利润表") if income_statement is not None and not income_statement.empty: financial_data["income_statement"] = income_statement.head(8).to_dict('records') except Exception as e: print(f"获取利润表失败: {e}") # 3. 获取现金流量表 try: cash_flow = ak.stock_financial_abstract_ths(symbol=symbol, indicator="现金流量表") if cash_flow is not None and not cash_flow.empty: financial_data["cash_flow"] = cash_flow.head(8).to_dict('records') except Exception as e: print(f"获取现金流量表失败: {e}") # 4. 获取主要财务指标 try: financial_indicators = ak.stock_financial_analysis_indicator(symbol=symbol) if financial_indicators is not None and not financial_indicators.empty: latest_data = financial_indicators.iloc[0] financial_data["financial_ratios"] = { "报告期": latest_data.get('报告期', 'N/A'), "净资产收益率ROE": latest_data.get('净资产收益率', 'N/A'), "总资产收益率ROA": latest_data.get('总资产收益率', 'N/A'), "销售毛利率": latest_data.get('销售毛利率', 'N/A'), "销售净利率": latest_data.get('销售净利率', 'N/A'), "资产负债率": latest_data.get('资产负债率', 'N/A'), "流动比率": latest_data.get('流动比率', 'N/A'), "速动比率": latest_data.get('速动比率', 'N/A'), "存货周转率": latest_data.get('存货周转率', 'N/A'), "应收账款周转率": latest_data.get('应收账款周转率', 'N/A'), "总资产周转率": latest_data.get('总资产周转率', 'N/A'), "营业收入同比增长": latest_data.get('营业收入同比增长', 'N/A'), "净利润同比增长": latest_data.get('净利润同比增长', 'N/A'), } except Exception as e: print(f"获取财务指标失败: {e}") # 5. 获取季度业绩（尝试不同API） try: # 尝试获取业绩预告 quarter_data = ak.stock_profit_forecast_em(symbol=symbol) if quarter_data is not None and not quarter_data.empty: financial_data["quarter_data"] = quarter_data.head(4).to_dict('records') except: try: # 备用方案：获取季度财报 quarter_data = ak.stock_financial_report_sina(stock=symbol, symbol="季报") if quarter_data is not None and not quarter_data.empty: financial_data["quarter_data"] = quarter_data.head(4).to_dict('records') except Exception as e: print(f"获取季度数据失败: {e}") return financial_data except Exception as e: print(f"获取中国股票财务数据失败: {e}") return financial_data def _get_us_financial_data(self, symbol): """获取美股财务数据""" financial_data = { "symbol": symbol, "balance_sheet": None, "income_statement": None, "cash_flow": None, "financial_ratios": {}, "quarter_data": None, } try: stock = yf.Ticker(symbol) info = stock.info # 1. 资产负债表 try: balance_sheet = stock.balance_sheet if balance_sheet is not None and not balance_sheet.empty: financial_data["balance_sheet"] = balance_sheet.iloc[:, :4].to_dict('index') except Exception as e: print(f"获取资产负债表失败: {e}") # 2. 利润表 try: income_stmt = stock.income_stmt if income_stmt is not None and not income_stmt.empty: financial_data["income_statement"] = income_stmt.iloc[:, :4].to_dict('index') except Exception as e: print(f"获取利润表失败: {e}") # 3. 现金流量表 try: cash_flow = stock.cashflow if cash_flow is not None and not cash_flow.empty: financial_data["cash_flow"] = cash_flow.iloc[:, :4].to_dict('index') except Exception as e: print(f"获取现金流量表失败: {e}") # 4. 财务比率（从info中提取） financial_data["financial_ratios"] = { "ROE": info.get('returnOnEquity', 'N/A'), "ROA": info.get('returnOnAssets', 'N/A'), "毛利率": info.get('grossMargins', 'N/A'), "营业利润率": info.get('operatingMargins', 'N/A'), "净利率": info.get('profitMargins', 'N/A'), "资产负债率": info.get('debtToEquity', 'N/A'), "流动比率": info.get('currentRatio', 'N/A'), "速动比率": info.get('quickRatio', 'N/A'), "EPS": info.get('trailingEps', 'N/A'), "每股账面价值": info.get('bookValue', 'N/A'), "股息率": info.get('dividendYield', 'N/A'), "派息率": info.get('payoutRatio', 'N/A'), "收入增长": info.get('revenueGrowth', 'N/A'), "盈利增长": info.get('earningsGrowth', 'N/A'), } return financial_data except Exception as e: print(f"获取美股财务数据失败: {e}") return financial_data