You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
114 lines
4.2 KiB
114 lines
4.2 KiB
"""
|
|
财务数据路由
|
|
"""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import get_db
|
|
from app.schemas.base import ResponseModel
|
|
from app.services.finance_service import FinanceService
|
|
from app.core.security import get_current_user
|
|
from app.models.user import User
|
|
from app.utils.date_utils import parse_date
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/balance-sheet", response_model=ResponseModel)
|
|
async def get_balance_sheet(
|
|
codes: str = Query(..., description="股票代码，多个用逗号分隔"),
|
|
start_date: str = Query(..., description="开始报告期(YYYYMMDD)"),
|
|
end_date: str = Query(..., description="结束报告期(YYYYMMDD)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""获取资产负债表"""
|
|
service = FinanceService(db)
|
|
code_list = [c.strip() for c in codes.split(",")]
|
|
start = parse_date(start_date)
|
|
end = parse_date(end_date)
|
|
|
|
data = service.get_balance_sheet(code_list, start, end)
|
|
return ResponseModel(data=data)
|
|
|
|
|
|
@router.get("/cash-flow", response_model=ResponseModel)
|
|
async def get_cash_flow(
|
|
codes: str = Query(..., description="股票代码，多个用逗号分隔"),
|
|
start_date: str = Query(..., description="开始报告期(YYYYMMDD)"),
|
|
end_date: str = Query(..., description="结束报告期(YYYYMMDD)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""获取现金流量表"""
|
|
service = FinanceService(db)
|
|
code_list = [c.strip() for c in codes.split(",")]
|
|
start = parse_date(start_date)
|
|
end = parse_date(end_date)
|
|
|
|
data = service.get_cash_flow(code_list, start, end)
|
|
return ResponseModel(data=data)
|
|
|
|
|
|
@router.get("/income", response_model=ResponseModel)
|
|
async def get_income(
|
|
codes: str = Query(..., description="股票代码，多个用逗号分隔"),
|
|
start_date: str = Query(..., description="开始报告期(YYYYMMDD)"),
|
|
end_date: str = Query(..., description="结束报告期(YYYYMMDD)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""获取利润表"""
|
|
service = FinanceService(db)
|
|
code_list = [c.strip() for c in codes.split(",")]
|
|
start = parse_date(start_date)
|
|
end = parse_date(end_date)
|
|
|
|
data = service.get_income_statement(code_list, start, end)
|
|
return ResponseModel(data=data)
|
|
|
|
|
|
@router.get("/profit-express", response_model=ResponseModel)
|
|
async def get_profit_express(
|
|
codes: str = Query(..., description="股票代码，多个用逗号分隔"),
|
|
start_date: str = Query(..., description="开始报告期(YYYYMMDD)"),
|
|
end_date: str = Query(..., description="结束报告期(YYYYMMDD)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""获取业绩快报"""
|
|
service = FinanceService(db)
|
|
code_list = [c.strip() for c in codes.split(",")]
|
|
start = parse_date(start_date)
|
|
end = parse_date(end_date)
|
|
|
|
# 从SDK获取业绩快报
|
|
try:
|
|
adapter = service.base_service._get_adapter()
|
|
data = adapter.get_profit_express(code_list, start, end)
|
|
return ResponseModel(data=data.to_dict("records") if not data.empty else [])
|
|
except Exception as e:
|
|
return ResponseModel(data=[], message=str(e))
|
|
|
|
|
|
@router.get("/profit-notice", response_model=ResponseModel)
|
|
async def get_profit_notice(
|
|
codes: str = Query(..., description="股票代码，多个用逗号分隔"),
|
|
start_date: str = Query(..., description="开始报告期(YYYYMMDD)"),
|
|
end_date: str = Query(..., description="结束报告期(YYYYMMDD)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""获取业绩预告"""
|
|
service = FinanceService(db)
|
|
code_list = [c.strip() for c in codes.split(",")]
|
|
start = parse_date(start_date)
|
|
end = parse_date(end_date)
|
|
|
|
try:
|
|
adapter = service.base_service._get_adapter()
|
|
data = adapter.get_profit_notice(code_list, start, end)
|
|
return ResponseModel(data=data.to_dict("records") if not data.empty else [])
|
|
except Exception as e:
|
|
return ResponseModel(data=[], message=str(e))
|