import pandas as pd import numpy as np try: from .config import TS_TOKEN, START_DATE, END_DATE, PRECISION_CONFIG from .stock_utils import * except (ImportError, SystemError): from config import TS_TOKEN, START_DATE, END_DATE, PRECISION_CONFIG from stock_utils import * from .data_source import get_tushare_pro class FinanceData: def __init__(self, token=None): """ 初始化 Tushare 接口 :param token: 可选，不再使用（保留兼容） """ self.pro = get_tushare_pro() self.ts_code = None self.fin_date = None self.unit_factor = 100000000 # 单位换算因子 (元 -> 亿) def get_finance_data(self): """ 获取完整的财务数据并计算指标 :param ts_code: 股票代码 (如 '002273.SZ') :param fin_date: 财报日期 (如 '20220331') :return: 包含财务指标的字典 """ # 获取各类财务数据 (返回 DataFrame) balance = self.get_balance() pre_balance = self.get_pre_balance() income = self.get_income() cash = self.get_cashflow() balance = balance.where(balance.notna(), 0) # NaN替换为0 pre_balance = pre_balance.where(pre_balance.notna(), 0) # NaN替换为0 income = income.where(income.notna(), 0) # NaN替换为0 cash = cash.where(cash.notna(), 0) # NaN替换为0 # 初始化结果字典 data = { 'ts_code': self.ts_code, 'period': self.fin_date, '--运营数据--': '' } # 运营数据计算 cash_equ = cash['c_cash_equ_end_period'].iloc[0] inventories = balance['inventories'].iloc[0] total_assets = balance['total_assets'].iloc[0] # 应收账票 = 应收账款 + 应收票据 accounts_receiv = float(balance['accounts_receiv'].iloc[0]) + float(balance['notes_receiv'].iloc[0]) prepayment = float(balance['prepayment'].iloc[0]) data.update({ '现金额-亿': cash_equ / self.unit_factor, '现金占比率': cash_equ / total_assets, '存货-亿': inventories / self.unit_factor, '存货占比率': inventories / total_assets, '应收账票-亿': accounts_receiv / self.unit_factor, '应收账票占比率': accounts_receiv / total_assets, '预付款-亿': prepayment / self.unit_factor, '预付款占比率': prepayment / total_assets, '运营占比率': (cash_equ + inventories + accounts_receiv + prepayment) / total_assets }) # 资产分布计算 fix_assets = float(balance['fix_assets'].iloc[0]) if 'fix_assets' in balance else 0 intan_assets = float(balance['intan_assets'].iloc[0]) if 'intan_assets' in balance else 0 lt_eqt_invest = float(balance['lt_eqt_invest'].iloc[0]) if 'lt_eqt_invest' in balance else 0 data.update({ '--资产分布--': '', '固定资产-亿': fix_assets / self.unit_factor, '固定资产占比率': fix_assets / total_assets, '无形资产-亿': intan_assets / self.unit_factor, '无形资产占率': intan_assets / total_assets, '股权投资-亿': lt_eqt_invest / self.unit_factor, '股权投资占比率': lt_eqt_invest / total_assets, '投资占比率': (fix_assets + intan_assets + lt_eqt_invest) / total_assets }) # 负债分布计算 acct_payable = float(balance['acct_payable'].iloc[0]) if 'acct_payable' in balance else 0 notes_payable = float(balance['notes_payable'].iloc[0]) if 'notes_payable' in balance else 0 adv_receipts = float(balance['adv_receipts'].iloc[0]) if 'adv_receipts' in balance else 0 st_borr = float(balance['st_borr'].iloc[0]) if 'st_borr' in balance else 0 lt_borr = float(balance['lt_borr'].iloc[0]) if 'lt_borr' in balance else 0 bond_payable = float(balance['bond_payable'].iloc[0]) if 'bond_payable' in balance else 0 biz_liab = acct_payable + notes_payable + adv_receipts fin_liab = st_borr + lt_borr + bond_payable zcfz = (float(balance['total_cur_liab'].iloc[0]) + float(balance['total_ncl'].iloc[0])) / total_assets data.update({ '--负债分布--': '', '经营负债-亿': biz_liab / self.unit_factor, '经营负债占比率': biz_liab / total_assets, '金融负债-亿': fin_liab / self.unit_factor, '金融负债占比率': fin_liab / total_assets, '资产负债率': zcfz }) # 运营能力计算 total_days = self.get_total_days() oper_cost = income['oper_cost'].iloc[0] revenue = income['revenue'].iloc[0] # 存货周转天数 (防除零) #avg_inventories = (float(pre_balance.get('inventories', 0)) + inventories) / 2 avg_inventories = (float(pre_balance['inventories'].iloc[0]) + inventories) / 2 days_1 = total_days / (oper_cost / avg_inventories) if avg_inventories > 0 else 0 # 应收周转天数 (防除零) avg_receiv = (float(pre_balance['accounts_receiv'].iloc[0]) + accounts_receiv) / 2 days_2 = total_days / (revenue / avg_receiv) if avg_receiv > 0 else 0 data.update({ '--运营能力--': '', '存货周转天数': days_1, '应收周转天数': days_2, '营业周期': days_1 + days_2 }) # 管理费分布计算 gross_profit = revenue - oper_cost gross_margin = gross_profit / revenue if revenue > 0 else 0 data.update({ '--管理费分布--': '', '毛利额': gross_profit / self.unit_factor, '毛利率': gross_margin, '营业税金率': float(income['biz_tax_surchg'].iloc[0]) / float(revenue) if revenue > 0 else 0, '销售费用率': float(income['sell_exp'].iloc[0]) / float(revenue) if revenue > 0 else 0, '研发费用率': float(income['rd_exp'].iloc[0]) / float(revenue) if revenue > 0 else 0, '管理费用率': float(income['admin_exp'].iloc[0]) / float(revenue) if revenue > 0 else 0, '净利润': float(income['n_income'].iloc[0]) / self.unit_factor, '净利润率': float(income['n_income'].iloc[0]) / float(revenue) if revenue > 0 else 0 }) # 权益及回报率计算 data.update({ '--权益及回报率--': '', '总资产-亿': total_assets / self.unit_factor, '销售收入-亿': revenue / self.unit_factor, '总资产周转率': revenue / total_assets if total_assets > 0 else 0, '总资产回报率': float(income['n_income'].iloc[0]) / total_assets if total_assets > 0 else 0, '权益乘数': 1 / (1 - zcfz) if zcfz < 1 else 0, }) # 计算ROE (净资产回报率) roa = data['总资产回报率'] data['净资产回报率'] = roa * data['权益乘数'] # 格式化比率数据 for key in list(data.keys()): if isinstance(data[key], float): if key.endswith('率'): data[key] = f"{data[key] * 100:.2f}%" else: data[key] = round(data[key], 2) return pd.Series(data) def get_balance(self): """获取资产负债表数据""" fields = [ 'ts_code', 'end_date', 'total_assets', 'fix_assets', 'intan_assets', 'lt_eqt_invest', 'inventories', 'accounts_receiv', 'notes_receiv', 'prepayment', 'acct_payable', 'notes_payable', 'adv_receipts', 'st_borr', 'lt_borr', 'bond_payable', 'total_cur_liab', 'total_ncl' ] return self.pro.balancesheet( ts_code=self.ts_code, period=self.fin_date, fields=fields ) def get_pre_balance(self): """获取上年度资产负债表数据""" pre_date = f"{int(self.fin_date[:4]) - 1}1231" # 上年末日期 fields = ['inventories', 'accounts_receiv', 'notes_receiv'] return self.pro.balancesheet( ts_code=self.ts_code, period=pre_date, fields=fields ) # 直接返回Series def get_income(self): """获取利润表数据""" fields = [ 'revenue', 'oper_cost', 'biz_tax_surchg', 'sell_exp', 'fin_exp', 'admin_exp', 'n_income', 'rd_exp' ] return self.pro.income( ts_code=self.ts_code, period=self.fin_date, fields=fields ) def get_cashflow(self): """获取现金流量表数据""" #c_cash_equ_end_period 期末现金及现金等价物余额 #end_bal_cash 现金的期末余额 fields = ['c_cash_equ_end_period', 'end_bal_cash'] return self.pro.cashflow( ts_code=self.ts_code, period=self.fin_date, fields=fields ) def get_total_days(self): """根据财报类型返回计算周转率的天数""" quarter = self.fin_date[4:6] return { '03': 90, # Q1 '06': 180, # H1 '09': 270, # Q1-Q3 '12': 360 # 全年 }.get(quarter, 360) #循环调用类 def get_finance_data_range( ts_code, start_date=START_DATE, end_date=END_DATE,TS_TOKEN=TS_TOKEN): """ 获取指定日期范围内的所有财报数据 :param ts_code: 股票代码 :param start_date: 开始日期 (yyyyMMdd) :param end_date: 结束日期 (yyyyMMdd) :return: 合并后的财报数据列表 """ results = pd.DataFrame() # 生成所有可能的财报日期 (季度末) years = range(int(start_date[:4]), int(end_date[:4]) + 1) report_dates = [] for year in years: report_dates.extend([ f"{year}0331", # Q1 f"{year}0630", # Q2 f"{year}0930", # Q3 f"{year}1231" # Q4 ]) # 筛选在指定日期范围内的财报日期 report_dates = [ date for date in report_dates if start_date <= date <= end_date ] report_dates=get_released_report_dates(start_date,end_date) # 按日期顺序获取财报数据 analyzer = FinanceData(TS_TOKEN) analyzer.ts_code=tscodeCheck(ts_code) for fin_date in sorted(report_dates): analyzer.fin_date = fin_date try: data = analyzer.get_finance_data() results = pd.concat([pd.DataFrame(results), data.to_frame().T], ignore_index=True) if len(results) > 0 else data.to_frame().T except Exception as e: print(f"获取 {ts_code} {fin_date} 财报数据失败: {str(e)}") continue return results # 使用示例 if __name__ == "__main__": # 导入当站目录的config文件 try: # 尝试相对导入（作为包的一部分） from .config import TS_TOKEN, START_DATE, END_DATE, PRECISION_CONFIG except (ImportError, SystemError): # 失败则使用绝对导入（直接运行脚本） from config import TS_TOKEN, START_DATE, END_DATE, PRECISION_CONFIG #token = "your_tushare_token" # 替换为实际token #analyzer = FinanceData(TS_TOKEN) # 获取002273.SZ在2022Q1的财务数据 #analyzer.ts_code = '300316.SZ' #analyzer.fin_date = '20240630' #result = analyzer.get_finance_data() result=get_finance_data_range("300316",start_date='20200101',end_date='2025060') print(result) #print(pd.Series(result))