""" 财务数据服务 """ from typing import List, Dict from datetime import date from sqlalchemy.orm import Session from sqlalchemy import and_ import pandas as pd import logging from app.models.finance import FinanceBalanceSheet, FinanceCashFlow, FinanceIncome from app.services.sdk_manager import sdk_manager from app.services.base_data_service import BaseDataService from app.utils.date_utils import parse_date, format_date logger = logging.getLogger(__name__) class FinanceService: """财务数据服务""" def __init__(self, db: Session): self.db = db self.base_service = BaseDataService(db) def _get_adapter(self): """获取SDK适配器（使用连接管理器）""" return sdk_manager.get_default_connection() def get_balance_sheet( self, codes: List[str], start_date: date, end_date: date ) -> Dict[str, List[dict]]: """获取资产负债表""" result = {} for code in codes: try: # 查询本地缓存 records = self.db.query(FinanceBalanceSheet).filter( and_( FinanceBalanceSheet.code == code, FinanceBalanceSheet.report_date >= start_date, FinanceBalanceSheet.report_date <= end_date ) ).order_by(FinanceBalanceSheet.report_date.desc()).all() # 如果本地没有数据，从SDK获取 if not records: adapter = self._get_adapter() if adapter: sdk_data = adapter.get_balance_sheet([code], start_date, end_date) if code in sdk_data and not sdk_data[code].empty: self._save_balance_sheet(code, sdk_data[code]) records = self.db.query(FinanceBalanceSheet).filter( and_( FinanceBalanceSheet.code == code, FinanceBalanceSheet.report_date >= start_date, FinanceBalanceSheet.report_date <= end_date ) ).order_by(FinanceBalanceSheet.report_date.desc()).all() result[code] = [ { "report_date": format_date(r.report_date), "report_type": r.report_type, "statement_type": r.statement_type, "total_assets": float(r.total_assets) if r.total_assets else None, "total_cur_assets": float(r.total_cur_assets) if r.total_cur_assets else None, "total_noncur_assets": float(r.total_noncur_assets) if r.total_noncur_assets else None, "currency_cap": float(r.currency_cap) if r.currency_cap else None, "notes_receivable": float(r.notes_receivable) if r.notes_receivable else None, "acct_receivable": float(r.acct_receivable) if r.acct_receivable else None, "inventory": float(r.inventory) if r.inventory else None, "fix_assets": float(r.fix_assets) if r.fix_assets else None, "total_liab": float(r.total_liab) if r.total_liab else None, "total_cur_liab": float(r.total_cur_liab) if r.total_cur_liab else None, "total_noncur_liab": float(r.total_noncur_liab) if r.total_noncur_liab else None, "notes_payable": float(r.notes_payable) if r.notes_payable else None, "acct_payable": float(r.acct_payable) if r.acct_payable else None, "st_borrowing": float(r.st_borrowing) if r.st_borrowing else None, "lt_loan": float(r.lt_loan) if r.lt_loan else None, "tot_share_equity": float(r.tot_share_equity) if r.tot_share_equity else None } for r in records ] except Exception as e: logger.error(f"获取{code}资产负债表失败: {str(e)}") result[code] = [] return result def get_cash_flow( self, codes: List[str], start_date: date, end_date: date ) -> Dict[str, List[dict]]: """获取现金流量表""" result = {} for code in codes: try: records = self.db.query(FinanceCashFlow).filter( and_( FinanceCashFlow.code == code, FinanceCashFlow.report_date >= start_date, FinanceCashFlow.report_date <= end_date ) ).order_by(FinanceCashFlow.report_date.desc()).all() if not records: adapter = self._get_adapter() if adapter: sdk_data = adapter.get_cash_flow([code], start_date, end_date) if code in sdk_data and not sdk_data[code].empty: self._save_cash_flow(code, sdk_data[code]) records = self.db.query(FinanceCashFlow).filter( and_( FinanceCashFlow.code == code, FinanceCashFlow.report_date >= start_date, FinanceCashFlow.report_date <= end_date ) ).order_by(FinanceCashFlow.report_date.desc()).all() result[code] = [ { "report_date": format_date(r.report_date), "report_type": r.report_type, "statement_type": r.statement_type, "net_cash_flows_opera_act": float(r.net_cash_flows_opera_act) if r.net_cash_flows_opera_act else None, "net_cash_flows_inv_act": float(r.net_cash_flows_inv_act) if r.net_cash_flows_inv_act else None, "net_cash_flows_fin_act": float(r.net_cash_flows_fin_act) if r.net_cash_flows_fin_act else None, "net_incr_cash_and_cash_equ": float(r.net_incr_cash_and_cash_equ) if r.net_incr_cash_and_cash_equ else None, "cash_recp_sg_and_rs": float(r.cash_recp_sg_and_rs) if r.cash_recp_sg_and_rs else None, "cash_pay_goods_services": float(r.cash_pay_goods_services) if r.cash_pay_goods_services else None } for r in records ] except Exception as e: logger.error(f"获取{code}现金流量表失败: {str(e)}") result[code] = [] return result def get_income_statement( self, codes: List[str], start_date: date, end_date: date ) -> Dict[str, List[dict]]: """获取利润表""" result = {} for code in codes: try: records = self.db.query(FinanceIncome).filter( and_( FinanceIncome.code == code, FinanceIncome.report_date >= start_date, FinanceIncome.report_date <= end_date ) ).order_by(FinanceIncome.report_date.desc()).all() if not records: adapter = self._get_adapter() if adapter: sdk_data = adapter.get_income_statement([code], start_date, end_date) if code in sdk_data and not sdk_data[code].empty: self._save_income(code, sdk_data[code]) records = self.db.query(FinanceIncome).filter( and_( FinanceIncome.code == code, FinanceIncome.report_date >= start_date, FinanceIncome.report_date <= end_date ) ).order_by(FinanceIncome.report_date.desc()).all() result[code] = [ { "report_date": format_date(r.report_date), "report_type": r.report_type, "statement_type": r.statement_type, "tot_opera_rev": float(r.tot_opera_rev) if r.tot_opera_rev else None, "opera_rev": float(r.opera_rev) if r.opera_rev else None, "tot_opera_cost": float(r.tot_opera_cost) if r.tot_opera_cost else None, "opera_profit": float(r.opera_profit) if r.opera_profit else None, "total_profit": float(r.total_profit) if r.total_profit else None, "net_pro_incl_min_int_inc": float(r.net_pro_incl_min_int_inc) if r.net_pro_incl_min_int_inc else None, "basic_eps": float(r.basic_eps) if r.basic_eps else None, "diluted_eps": float(r.diluted_eps) if r.diluted_eps else None, "rd_exp": float(r.rd_exp) if r.rd_exp else None, "selling_exp": float(r.selling_exp) if r.selling_exp else None, "admin_exp": float(r.admin_exp) if r.admin_exp else None, "fin_exp": float(r.fin_exp) if r.fin_exp else None } for r in records ] except Exception as e: logger.error(f"获取{code}利润表失败: {str(e)}") result[code] = [] return result def _save_balance_sheet(self, code: str, df: pd.DataFrame): """保存资产负债表""" if df.empty: return for idx, row in df.iterrows(): report_date = idx if isinstance(idx, date) else parse_date(str(idx)) statement_type = int(row.get("statement_type", 0)) existing = self.db.query(FinanceBalanceSheet).filter( and_( FinanceBalanceSheet.code == code, FinanceBalanceSheet.report_date == report_date, FinanceBalanceSheet.statement_type == statement_type ) ).first() def get_float(val): if pd.isna(val): return None return float(val) if existing: existing.total_assets = get_float(row.get("total_assets")) existing.total_cur_assets = get_float(row.get("total_cur_assets")) existing.total_noncur_assets = get_float(row.get("total_noncur_assets")) existing.tot_share_equity = get_float(row.get("tot_share_equity")) else: record = FinanceBalanceSheet( code=code, report_date=report_date, report_type=get_float(row.get("report_type")), statement_type=statement_type, total_assets=get_float(row.get("total_assets")), total_cur_assets=get_float(row.get("total_cur_assets")), tot_share_equity=get_float(row.get("tot_share_equity")) ) self.db.add(record) self.db.commit() def _save_cash_flow(self, code: str, df: pd.DataFrame): """保存现金流量表""" if df.empty: return for idx, row in df.iterrows(): report_date = idx if isinstance(idx, date) else parse_date(str(idx)) statement_type = int(row.get("statement_type", 0)) existing = self.db.query(FinanceCashFlow).filter( and_( FinanceCashFlow.code == code, FinanceCashFlow.report_date == report_date, FinanceCashFlow.statement_type == statement_type ) ).first() if not existing: def get_float(val): if pd.isna(val): return None return float(val) record = FinanceCashFlow( code=code, report_date=report_date, report_type=get_float(row.get("report_type")), statement_type=statement_type, net_cash_flows_opera_act=get_float(row.get("net_cash_flows_opera_act")) ) self.db.add(record) self.db.commit() def _save_income(self, code: str, df: pd.DataFrame): """保存利润表""" if df.empty: return for idx, row in df.iterrows(): report_date = idx if isinstance(idx, date) else parse_date(str(idx)) statement_type = int(row.get("statement_type", 0)) existing = self.db.query(FinanceIncome).filter( and_( FinanceIncome.code == code, FinanceIncome.report_date == report_date, FinanceIncome.statement_type == statement_type ) ).first() if not existing: def get_float(val): if pd.isna(val): return None return float(val) record = FinanceIncome( code=code, report_date=report_date, report_type=get_float(row.get("report_type")), statement_type=statement_type, tot_opera_rev=get_float(row.get("tot_opera_rev")), net_pro_incl_min_int_inc=get_float(row.get("net_pro_incl_min_int_inc")), basic_eps=get_float(row.get("basic_eps")) ) self.db.add(record) self.db.commit()