amazing-data-service/backend/app/services/finance_service.py

"""
财务数据服务
"""
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()