from .base import BaseDataProvider from typing import Any, Dict, List, Optional, Callable import logging import asyncio import tushare as ts import math import datetime logger = logging.getLogger(__name__) class TushareProvider(BaseDataProvider): def _initialize(self): if not self.token: raise ValueError("Tushare API token not provided.") # 使用官方 SDK 客户端 self._pro = ts.pro_api(self.token) # 交易日历缓存：key=(exchange, start, end) -> List[Dict] self._trade_cal_cache: Dict[str, List[Dict[str, Any]]] = {} async def _resolve_trade_dates(self, dates: List[str], exchange: str = "SSE") -> Dict[str, str]: """ 将任意日期映射为“该日若非交易日，则取不晚于该日的最近一个交易日”。 返回映射：requested_date -> resolved_trade_date。 """ if not dates: return {} start_date = min(dates) end_date = max(dates) cache_key = f"{exchange}:{start_date}:{end_date}" if cache_key in self._trade_cal_cache: cal_rows = self._trade_cal_cache[cache_key] else: cal_rows = await self._query( api_name="trade_cal", params={ "exchange": exchange, "start_date": start_date, "end_date": end_date, }, fields=["cal_date", "is_open", "pretrade_date"], ) self._trade_cal_cache[cache_key] = cal_rows by_date: Dict[str, Dict[str, Any]] = {str(r.get("cal_date")): r for r in cal_rows} # 同时准备已开放的交易日期序列，便于兜底搜索 open_dates = sorted([d for d, r in by_date.items() if int(r.get("is_open", 0)) == 1]) def _prev_open(d: str) -> Optional[str]: # 找到 <= d 的最大开市日 lo, hi = 0, len(open_dates) - 1 ans = None while lo <= hi: mid = (lo + hi) // 2 if open_dates[mid] <= d: ans = open_dates[mid] lo = mid + 1 else: hi = mid - 1 return ans resolved: Dict[str, str] = {} for d in dates: row = by_date.get(d) if row is None: # 不在本段日历（极少数情况），做一次兜底：使用区间内最近开市日 prev_d = _prev_open(d) if prev_d: resolved[d] = prev_d else: # 最后兜底，仍找不到则原样返回 resolved[d] = d continue is_open = int(row.get("is_open", 0)) if is_open == 1: resolved[d] = d else: prev = str(row.get("pretrade_date") or "") if prev: resolved[d] = prev else: prev_d = _prev_open(d) resolved[d] = prev_d or d return resolved async def _query( self, api_name: str, params: Optional[Dict[str, Any]] = None, fields: Optional[List[str]] = None, ) -> List[Dict[str, Any]]: """ 使用官方 tushare SDK 统一查询，返回字典列表。 为避免阻塞事件循环，内部通过 asyncio.to_thread 在线程中执行同步调用。 """ params = params or {} def _call() -> List[Dict[str, Any]]: # 将字段列表转换为逗号分隔的字符串（SDK 推荐方式） fields_arg: Optional[str] = ",".join(fields) if isinstance(fields, list) else None # 优先使用属性方式（pro.fina_indicator 等）；若不存在则回退到通用 query func: Optional[Callable] = getattr(self._pro, api_name, None) try: if callable(func): df = func(**params, fields=fields_arg) if fields_arg else func(**params) else: # 通用回退：pro.query(name, params=..., fields=...) if fields_arg: df = self._pro.query(api_name, params=params, fields=fields_arg) else: df = self._pro.query(api_name, params=params) except Exception as exc: # 将 SDK 抛出的异常包装为统一日志 raise RuntimeError(f"tushare.{api_name} failed: {exc}") if df is None or df.empty: return [] # DataFrame -> List[Dict] return df.to_dict(orient="records") try: rows: List[Dict[str, Any]] = await asyncio.to_thread(_call) # 清洗 NaN/Inf，避免 JSON 序列化错误 DATE_KEYS = { "cal_date", "pretrade_date", "trade_date", "trade_dt", "date", "end_date", "ann_date", "f_ann_date", "period" } def _sanitize_value(key: str, v: Any) -> Any: if v is None: return None # 保持日期/期末字段为字符串（避免 20231231 -> 20231231.0 导致匹配失败） if key in DATE_KEYS: try: s = str(v) # 去除意外的小数点形式 if s.endswith(".0"): s = s[:-2] return s except Exception: return str(v) try: # 处理 numpy.nan / numpy.inf / Decimal / numpy 数值等，统一为 Python float fv = float(v) return fv if math.isfinite(fv) else None except Exception: # 利用自反性判断 NaN（NaN != NaN） try: if v != v: return None except Exception: pass return v for row in rows: for k, v in list(row.items()): row[k] = _sanitize_value(k, v) # logger.info(f"Tushare '{api_name}' returned {len(rows)} rows.") return rows except Exception as e: logger.error(f"Exception calling tushare '{api_name}': {e}") raise async def get_stock_basic(self, stock_code: str) -> Optional[Dict[str, Any]]: try: rows = await self._query( api_name="stock_basic", params={"ts_code": stock_code}, ) return rows[0] if rows else None except Exception as e: logger.error(f"Tushare get_stock_basic failed for {stock_code}: {e}") return None async def get_daily_price(self, stock_code: str, start_date: str, end_date: str) -> List[Dict[str, Any]]: try: rows = await self._query( api_name="daily", params={ "ts_code": stock_code, "start_date": start_date, "end_date": end_date, }, ) return rows or [] except Exception as e: logger.error(f"Tushare get_daily_price failed for {stock_code}: {e}") return [] async def get_daily_basic_points(self, stock_code: str, trade_dates: List[str]) -> List[Dict[str, Any]]: """ 获取指定交易日列表的 daily_basic 数据（例如 total_mv、pe、pb）。 """ try: if not trade_dates: return [] # 将请求日期映射到不晚于该日的最近交易日 mapping = await self._resolve_trade_dates(trade_dates, exchange="SSE") resolved_dates = list(set(mapping.values())) start_date = min(resolved_dates) end_date = max(resolved_dates) # 一次性取区间内数据，再按解析后的交易日过滤 all_rows = await self._query( api_name="daily_basic", params={ "ts_code": stock_code, "start_date": start_date, "end_date": end_date, }, ) wanted = set(resolved_dates) rows = [r for r in all_rows if str(r.get("trade_date")) in wanted] logger.info(f"Tushare daily_basic returned {len(rows)} rows for {stock_code} on {len(trade_dates)} requested dates (resolved to {len(wanted)} trading dates)") return rows except Exception as e: logger.error(f"Tushare get_daily_basic_points failed for {stock_code}: {e}") return [] async def get_daily_points(self, stock_code: str, trade_dates: List[str]) -> List[Dict[str, Any]]: """ 获取指定交易日列表的日行情（例如 close）。 """ try: if not trade_dates: return [] mapping = await self._resolve_trade_dates(trade_dates, exchange="SSE") resolved_dates = list(set(mapping.values())) start_date = min(resolved_dates) end_date = max(resolved_dates) all_rows = await self._query( api_name="daily", params={ "ts_code": stock_code, "start_date": start_date, "end_date": end_date, }, ) wanted = set(resolved_dates) rows = [r for r in all_rows if str(r.get("trade_date")) in wanted] logger.info(f"Tushare daily returned {len(rows)} rows for {stock_code} on {len(trade_dates)} requested dates (resolved to {len(wanted)} trading dates)") return rows except Exception as e: logger.error(f"Tushare get_daily_points failed for {stock_code}: {e}") return [] def _calculate_derived_metrics(self, series: Dict[str, List[Dict]], periods: List[str]) -> Dict[str, List[Dict]]: """ 在 Tushare provider 内部计算派生指标。 """ # --- Helper Functions --- def _get_value(key: str, period: str) -> Optional[float]: if key not in series: return None point = next((p for p in series[key] if p.get("period") == period), None) if point is None or point.get("value") is None: return None try: return float(point["value"]) except (ValueError, TypeError): return None def _get_avg_value(key: str, period: str) -> Optional[float]: current_val = _get_value(key, period) try: # 总是和上一年度的年报值（如果存在）进行平均 current_year = int(period[:4]) prev_year_end_period = str(current_year - 1) + "1231" prev_val = _get_value(key, prev_year_end_period) except (ValueError, TypeError): prev_val = None if current_val is None: return None if prev_val is None: return current_val return (current_val + prev_val) / 2 def _get_cogs(period: str) -> Optional[float]: revenue = _get_value('revenue', period) gp_margin_raw = _get_value('grossprofit_margin', period) if revenue is None or gp_margin_raw is None: return None gp_margin = gp_margin_raw / 100.0 if abs(gp_margin_raw) > 1 else gp_margin_raw return revenue * (1 - gp_margin) def add_series(key: str, data: List[Dict]): if data: series[key] = data # --- Calculations --- fcf_data = [] for period in periods: op_cashflow = _get_value('n_cashflow_act', period) capex = _get_value('c_pay_acq_const_fiolta', period) if op_cashflow is not None and capex is not None: fcf_data.append({"period": period, "value": op_cashflow - capex}) add_series('__free_cash_flow', fcf_data) fee_calcs = [ ('__sell_rate', 'sell_exp', 'revenue'), ('__admin_rate', 'admin_exp', 'revenue'), ('__rd_rate', 'rd_exp', 'revenue'), ('__depr_ratio', 'depr_fa_coga_dpba', 'revenue'), ] for key, num_key, den_key in fee_calcs: data = [] for period in periods: numerator = _get_value(num_key, period) denominator = _get_value(den_key, period) if numerator is not None and denominator is not None and denominator != 0: data.append({"period": period, "value": (numerator / denominator) * 100}) add_series(key, data) tax_rate_data = [] for period in periods: tax_to_ebt = _get_value('tax_to_ebt', period) if tax_to_ebt is not None: rate = tax_to_ebt * 100 if abs(tax_to_ebt) <= 1 else tax_to_ebt tax_rate_data.append({"period": period, "value": rate}) add_series('__tax_rate', tax_rate_data) other_fee_data = [] for period in periods: gp_raw = _get_value('grossprofit_margin', period) np_raw = _get_value('netprofit_margin', period) rev = _get_value('revenue', period) sell_exp = _get_value('sell_exp', period) admin_exp = _get_value('admin_exp', period) rd_exp = _get_value('rd_exp', period) if all(v is not None for v in [gp_raw, np_raw, rev, sell_exp, admin_exp, rd_exp]) and rev != 0: gp = gp_raw / 100 if abs(gp_raw) > 1 else gp_raw np = np_raw / 100 if abs(np_raw) > 1 else np_raw sell_rate = sell_exp / rev admin_rate = admin_exp / rev rd_rate = rd_exp / rev other_rate = (gp - np - sell_rate - admin_rate - rd_rate) * 100 other_fee_data.append({"period": period, "value": other_rate}) add_series('__other_fee_rate', other_fee_data) asset_ratio_keys = [ ('__money_cap_ratio', 'money_cap'), ('__inventories_ratio', 'inventories'), ('__ar_ratio', 'accounts_receiv_bill'), ('__prepay_ratio', 'prepayment'), ('__fix_assets_ratio', 'fix_assets'), ('__lt_invest_ratio', 'lt_eqt_invest'), ('__goodwill_ratio', 'goodwill'), ('__ap_ratio', 'accounts_pay'), ('__st_borr_ratio', 'st_borr'), ('__lt_borr_ratio', 'lt_borr'), ] for key, num_key in asset_ratio_keys: data = [] for period in periods: numerator = _get_value(num_key, period) denominator = _get_value('total_assets', period) if numerator is not None and denominator is not None and denominator != 0: data.append({"period": period, "value": (numerator / denominator) * 100}) add_series(key, data) adv_data = [] for period in periods: adv = _get_value('adv_receipts', period) or 0 contract = _get_value('contract_liab', period) or 0 total_assets = _get_value('total_assets', period) if total_assets is not None and total_assets != 0: adv_data.append({"period": period, "value": ((adv + contract) / total_assets) * 100}) add_series('__adv_ratio', adv_data) other_assets_data = [] known_assets_keys = ['money_cap', 'inventories', 'accounts_receiv_bill', 'prepayment', 'fix_assets', 'lt_eqt_invest', 'goodwill'] for period in periods: total_assets = _get_value('total_assets', period) if total_assets is not None and total_assets != 0: sum_known = sum(_get_value(k, period) or 0 for k in known_assets_keys) other_assets_data.append({"period": period, "value": ((total_assets - sum_known) / total_assets) * 100}) add_series('__other_assets_ratio', other_assets_data) op_assets_data = [] for period in periods: total_assets = _get_value('total_assets', period) if total_assets is not None and total_assets != 0: inv = _get_value('inventories', period) or 0 ar = _get_value('accounts_receiv_bill', period) or 0 pre = _get_value('prepayment', period) or 0 ap = _get_value('accounts_pay', period) or 0 adv = _get_value('adv_receipts', period) or 0 contract_liab = _get_value('contract_liab', period) or 0 operating_assets = inv + ar + pre - ap - adv - contract_liab op_assets_data.append({"period": period, "value": (operating_assets / total_assets) * 100}) add_series('__operating_assets_ratio', op_assets_data) debt_ratio_data = [] for period in periods: total_assets = _get_value('total_assets', period) if total_assets is not None and total_assets != 0: st_borr = _get_value('st_borr', period) or 0 lt_borr = _get_value('lt_borr', period) or 0 debt_ratio_data.append({"period": period, "value": ((st_borr + lt_borr) / total_assets) * 100}) add_series('__interest_bearing_debt_ratio', debt_ratio_data) payturn_data = [] for period in periods: avg_ap = _get_avg_value('accounts_pay', period) cogs = _get_cogs(period) if avg_ap is not None and cogs is not None and cogs != 0: payturn_data.append({"period": period, "value": (365 * avg_ap) / cogs}) add_series('payturn_days', payturn_data) per_capita_calcs = [ ('__rev_per_emp', 'revenue', 10000), ('__profit_per_emp', 'n_income', 10000), ('__salary_per_emp', 'c_paid_to_for_empl', 10000), ] for key, num_key, divisor in per_capita_calcs: data = [] for period in periods: numerator = _get_value(num_key, period) employees = _get_value('employees', period) if numerator is not None and employees is not None and employees != 0: data.append({"period": period, "value": (numerator / employees) / divisor}) add_series(key, data) return series async def get_financial_statements(self, stock_code: str, report_dates: Optional[List[str]] = None) -> Dict[str, List[Dict[str, Any]]]: # 1) 一次性拉取所需四表（尽量齐全字段），再按指定 report_dates 过滤 # 字段列表基于官方示例，避免超量请求可按需精简 bs_fields = [ "ts_code","ann_date","f_ann_date","end_date","report_type","comp_type","end_type", "money_cap","inventories","prepayment","accounts_receiv","accounts_receiv_bill","goodwill", "lt_eqt_invest","fix_assets","total_assets","accounts_pay","adv_receipts","contract_liab", "st_borr","lt_borr","total_cur_assets","total_cur_liab","total_ncl","total_liab","total_hldr_eqy_exc_min_int", ] ic_fields = [ "ts_code","ann_date","f_ann_date","end_date","report_type","comp_type","end_type", "total_revenue","revenue","sell_exp","admin_exp","rd_exp","operate_profit","total_profit", "income_tax","n_income","n_income_attr_p","ebit","ebitda","netprofit_margin","grossprofit_margin", ] cf_fields = [ "ts_code","ann_date","f_ann_date","end_date","comp_type","report_type","end_type", "n_cashflow_act","c_pay_acq_const_fiolta","c_paid_to_for_empl","depr_fa_coga_dpba", ] fi_fields = [ "ts_code","end_date","ann_date","grossprofit_margin","netprofit_margin","tax_to_ebt","roe","roa","roic", "invturn_days","arturn_days","fa_turn","tr_yoy","dt_netprofit_yoy","assets_turn", ] try: bs_rows, ic_rows, cf_rows, fi_rows, rep_rows, div_rows, holder_rows, company_rows = await asyncio.gather( self._query("balancesheet", params={"ts_code": stock_code, "report_type": 1}, fields=bs_fields), self._query("income", params={"ts_code": stock_code, "report_type": 1}, fields=ic_fields), self._query("cashflow", params={"ts_code": stock_code, "report_type": 1}, fields=cf_fields), self._query("fina_indicator", params={"ts_code": stock_code}, fields=fi_fields), # 回购公告 self._query( "repurchase", params={"ts_code": stock_code}, fields=[ "ts_code","ann_date","end_date","proc","exp_date","vol","amount","high_limit","low_limit", ], ), # 分红公告（仅取必要字段） self._query( "dividend", params={"ts_code": stock_code}, fields=[ "ts_code","end_date","cash_div_tax","pay_date","base_share", ], ), # 股东户数（按报告期） self._query( "stk_holdernumber", params={"ts_code": stock_code}, fields=[ "ts_code","ann_date","end_date","holder_num", ], ), # 公司基本信息（包含员工数） self._query( "stock_company", params={"ts_code": stock_code}, fields=[ "ts_code","employees", ], ), ) try: logger.info(f"[Dividend] fetched {len(div_rows)} rows for {stock_code}") except Exception: pass except Exception as e: logger.error(f"Tushare bulk fetch failed for {stock_code}: {e}") bs_rows, ic_rows, cf_rows, fi_rows, rep_rows, div_rows, holder_rows, company_rows = [], [], [], [], [], [], [], [] # 2) 以 end_date 聚合合并四表 by_date: Dict[str, Dict[str, Any]] = {} def _merge_rows(rows: List[Dict[str, Any]]): for r in rows or []: end_date = str(r.get("end_date") or r.get("period") or "") if not end_date: continue if end_date not in by_date: by_date[end_date] = {"ts_code": stock_code, "end_date": end_date} by_date[end_date].update(r) _merge_rows(bs_rows) _merge_rows(ic_rows) _merge_rows(cf_rows) _merge_rows(fi_rows) # 3) 筛选报告期：今年的最新报告期 + 往年所有年报 current_year = str(datetime.date.today().year) all_available_dates = sorted(by_date.keys(), reverse=True) latest_current_year_report = None for d in all_available_dates: if d.startswith(current_year): latest_current_year_report = d break previous_years_annual_reports = [ d for d in all_available_dates if d.endswith("1231") and not d.startswith(current_year) ] wanted_dates = [] if latest_current_year_report: wanted_dates.append(latest_current_year_report) wanted_dates.extend(previous_years_annual_reports) all_statements = [by_date[d] for d in wanted_dates if d in by_date] logger.info(f"Successfully prepared {len(all_statements)} merged statement(s) for {stock_code} from {len(by_date)} available reports.") # Transform to series format series: Dict[str, List[Dict]] = {} if all_statements: for report in all_statements: period = report.get("end_date", "") if not period: continue for key, value in report.items(): if key in ['ts_code', 'end_date', 'ann_date', 'f_ann_date', 'report_type', 'comp_type', 'end_type', 'update_flag', 'period']: continue # 仅保留可转为有限 float 的数值，避免 JSON 序列化错误 try: fv = float(value) except (TypeError, ValueError): continue if value is not None and math.isfinite(fv): if key not in series: series[key] = [] if not any(d['period'] == period for d in series[key]): series[key].append({"period": period, "value": fv}) # 汇总回购信息为年度序列：按报告期 end_date 年份分组； # 其中 repurchase_amount 取该年内“最后一个 ann_date”的 amount 值。 if 'rep_rows' in locals() and rep_rows: rep_by_year: Dict[str, Dict[str, Any]] = {} for r in rep_rows: endd = str(r.get("end_date") or r.get("ann_date") or "") if not endd: continue y = endd[:4] bucket = rep_by_year.setdefault(y, { "amount_sum": 0.0, "vol": 0.0, "high_limit": None, "low_limit": None, "last_ann_date": None, "amount_last": None, }) amt = r.get("amount") vol = r.get("vol") hi = r.get("high_limit") lo = r.get("low_limit") ann = str(r.get("ann_date") or "") if isinstance(amt, (int, float)) and amt is not None: bucket["amount_sum"] += float(amt) if ann and ann[:4] == y: last = bucket["last_ann_date"] if last is None or ann > last: bucket["last_ann_date"] = ann bucket["amount_last"] = float(amt) if isinstance(vol, (int, float)) and vol is not None: bucket["vol"] += float(vol) if isinstance(hi, (int, float)) and hi is not None: bucket["high_limit"] = float(hi) if isinstance(lo, (int, float)) and lo is not None: bucket["low_limit"] = float(lo) if rep_by_year: amt_series = [] vol_series = [] hi_series = [] lo_series = [] for y, v in rep_by_year.items(): # 当年数据放在当前年最新报告期，否则放在年度报告期 if y == current_year and latest_current_year_report: period_key = latest_current_year_report else: period_key = f"{y}1231" if v.get("amount_last") is not None: amt_series.append({"period": period_key, "value": v["amount_last"]}) if v.get("vol"): vol_series.append({"period": period_key, "value": v["vol"]}) if v.get("high_limit") is not None: hi_series.append({"period": period_key, "value": v["high_limit"]}) if v.get("low_limit") is not None: lo_series.append({"period": period_key, "value": v["low_limit"]}) if amt_series: series["repurchase_amount"] = amt_series if vol_series: series["repurchase_vol"] = vol_series if hi_series: series["repurchase_high_limit"] = hi_series if lo_series: series["repurchase_low_limit"] = lo_series # 汇总分红信息为年度序列：以真实派息日 pay_date 的年份分组； # 每条记录金额= 每股分红(cash_div_tax) * 基准股本(base_share)，其中 base_share 单位为“万股”， # 金额以“亿”为单位返回，因此需再除以 10000。 if 'div_rows' in locals() and div_rows: div_by_year: Dict[str, float] = {} for r in div_rows: pay = str(r.get("pay_date") or "") # 仅统计存在数字年份的真实派息日 if not pay or len(pay) < 4 or not any(ch.isdigit() for ch in pay): continue y = pay[:4] cash_div = r.get("cash_div_tax") base_share = r.get("base_share") if isinstance(cash_div, (int, float)) and isinstance(base_share, (int, float)): # 现金分红总额（万元）= 每股分红(元) * 基准股本(万股) # 转为“亿”需除以 10000 amount_billion = (float(cash_div) * float(base_share)) / 10000.0 div_by_year[y] = div_by_year.get(y, 0.0) + amount_billion if div_by_year: div_series = [] for y, v in sorted(div_by_year.items()): # 当年数据放在当前年最新报告期，否则放在年度报告期 if y == current_year and latest_current_year_report: period_key = latest_current_year_report else: period_key = f"{y}1231" div_series.append({"period": period_key, "value": v}) series["dividend_amount"] = div_series # try: # logger.info(f"[Dividend] Series dividend_amount(period) for {stock_code}: {div_series}") # except Exception: # pass # 汇总股东户数信息：按报告期 end_date 分组，取最新的 holder_num if 'holder_rows' in locals() and holder_rows: # 按 end_date 分组，取最新的 ann_date 的 holder_num holder_by_period: Dict[str, Dict[str, Any]] = {} for r in holder_rows: end_date = str(r.get("end_date") or "") if not end_date: continue ann_date = str(r.get("ann_date") or "") holder_num = r.get("holder_num") if end_date not in holder_by_period: holder_by_period[end_date] = { "holder_num": holder_num, "latest_ann_date": ann_date } else: # 比较 ann_date，取最新的 current_latest = holder_by_period[end_date]["latest_ann_date"] if ann_date and (not current_latest or ann_date > current_latest): holder_by_period[end_date] = { "holder_num": holder_num, "latest_ann_date": ann_date } # 使用与财务报表相同的报告期筛选逻辑 # 股东户数应该与财务报表的报告期时间点对应 holder_series = [] for end_date in wanted_dates: if end_date in holder_by_period: data = holder_by_period[end_date] holder_num = data["holder_num"] if isinstance(holder_num, (int, float)) and holder_num is not None: holder_series.append({"period": end_date, "value": float(holder_num)}) if holder_series: series["holder_num"] = holder_series # 汇总员工数信息：员工数放在去年的年末（上一年的12月31日） if 'company_rows' in locals() and company_rows: # 员工数通常是静态数据，取最新的一个值 latest_employees = None for r in company_rows: employees = r.get("employees") if isinstance(employees, (int, float)) and employees is not None: latest_employees = float(employees) break # 取第一个有效值 if latest_employees is not None: # 将员工数放在去年的年末（上一年的12月31日） previous_year = str(datetime.date.today().year - 1) period_key = f"{previous_year}1231" series["employees"] = [{"period": period_key, "value": latest_employees}] # Calculate derived metrics periods = sorted(list(set(d['period'] for s in series.values() for d in s))) series = self._calculate_derived_metrics(series, periods) return series