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.
74 lines
2.6 KiB
74 lines
2.6 KiB
from fastapi import APIRouter, Query, Depends, Path
|
|
from typing import List, Optional
|
|
from loguru import logger
|
|
|
|
from models import AIAnalysis, KLineData, Stock, ApiResponse, DataSourceType
|
|
from adapters import DataAdapterFactory
|
|
|
|
router = APIRouter(prefix="/stock", tags=["个股分析"])
|
|
|
|
|
|
def get_data_source(source: str = Query(None)) -> DataSourceType:
|
|
if source:
|
|
try:
|
|
return DataSourceType(source)
|
|
except ValueError:
|
|
pass
|
|
return DataSourceType.EASTMONEY
|
|
|
|
|
|
@router.get("/{code}", response_model=ApiResponse, summary="获取股票详情")
|
|
async def get_stock_detail(
|
|
code: str = Path(..., description="股票代码"),
|
|
source: DataSourceType = Depends(get_data_source)
|
|
):
|
|
try:
|
|
adapter = DataAdapterFactory.get_adapter(source)
|
|
data = await adapter.fetch_stock_detail(code)
|
|
if data:
|
|
return ApiResponse(
|
|
success=True,
|
|
data=data.model_dump(),
|
|
message=f"数据来源: {adapter.name}",
|
|
)
|
|
return ApiResponse(success=False, error="股票不存在")
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch stock detail: {e}")
|
|
return ApiResponse(success=False, error=str(e))
|
|
|
|
|
|
@router.get("/{code}/kline", response_model=ApiResponse, summary="获取K线数据")
|
|
async def get_kline_data(
|
|
code: str = Path(..., description="股票代码"),
|
|
days: int = Query(30, ge=5, le=120, description="天数"),
|
|
source: DataSourceType = Depends(get_data_source)
|
|
):
|
|
try:
|
|
adapter = DataAdapterFactory.get_adapter(source)
|
|
data = await adapter.fetch_kline_data(code, days)
|
|
return ApiResponse(
|
|
success=True,
|
|
data=[item.model_dump() for item in data],
|
|
message=f"数据来源: {adapter.name}",
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch kline data: {e}")
|
|
return ApiResponse(success=False, error=str(e))
|
|
|
|
|
|
@router.get("/{code}/analysis", response_model=ApiResponse, summary="获取AI分析")
|
|
async def get_ai_analysis(
|
|
code: str = Path(..., description="股票代码"),
|
|
source: DataSourceType = Depends(get_data_source)
|
|
):
|
|
try:
|
|
adapter = DataAdapterFactory.get_adapter(source)
|
|
data = await adapter.fetch_ai_analysis(code)
|
|
return ApiResponse(
|
|
success=True,
|
|
data=data.model_dump(),
|
|
message=f"数据来源: {adapter.name}",
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch AI analysis: {e}")
|
|
return ApiResponse(success=False, error=str(e)) |