132 lines
3.2 KiB
Python
132 lines
3.2 KiB
Python
"""
|
|
美股相关 API 路由
|
|
"""
|
|
from fastapi import APIRouter, HTTPException
|
|
from typing import Dict, Any, List
|
|
from app.utils.logger import logger
|
|
from app.config import get_settings
|
|
|
|
router = APIRouter()
|
|
|
|
# 全局变量，用于访问智能体实例
|
|
_stock_agent_instance = None
|
|
|
|
|
|
def set_stock_agent(agent):
|
|
"""设置智能体实例（由 main.py 调用）"""
|
|
global _stock_agent_instance
|
|
_stock_agent_instance = agent
|
|
|
|
|
|
@router.get("/status")
|
|
async def get_stock_status() -> Dict[str, Any]:
|
|
"""
|
|
获取美股智能体状态
|
|
|
|
Returns:
|
|
智能体状态信息
|
|
"""
|
|
try:
|
|
if _stock_agent_instance is None:
|
|
return {
|
|
"enabled": False,
|
|
"message": "美股智能体未启用"
|
|
}
|
|
|
|
status = _stock_agent_instance.get_status()
|
|
status["enabled"] = True
|
|
return status
|
|
|
|
except Exception as e:
|
|
logger.error(f"获取美股状态失败: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/symbols")
|
|
async def get_stock_symbols() -> Dict[str, Any]:
|
|
"""
|
|
获取当前监控的股票列表
|
|
|
|
Returns:
|
|
股票列表
|
|
"""
|
|
try:
|
|
settings = get_settings()
|
|
symbols = settings.stock_symbols.split(',') if settings.stock_symbols else []
|
|
|
|
return {
|
|
"symbols": symbols,
|
|
"count": len(symbols)
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"获取股票列表失败: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post("/analyze/{symbol}")
|
|
async def analyze_stock(symbol: str) -> Dict[str, Any]:
|
|
"""
|
|
手动触发分析指定股票
|
|
|
|
Args:
|
|
symbol: 股票代码，如 'AAPL'
|
|
|
|
Returns:
|
|
分析结果
|
|
"""
|
|
try:
|
|
if _stock_agent_instance is None:
|
|
raise HTTPException(status_code=400, detail="美股智能体未启用")
|
|
|
|
# 执行单次分析
|
|
result = await _stock_agent_instance.analyze_once(symbol)
|
|
|
|
if "error" in result:
|
|
raise HTTPException(status_code=400, detail=result["error"])
|
|
|
|
return {
|
|
"success": True,
|
|
"symbol": symbol,
|
|
"result": result
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"分析 {symbol} 失败: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.get("/quote/{symbol}")
|
|
async def get_stock_quote(symbol: str) -> Dict[str, Any]:
|
|
"""
|
|
获取股票实时行情
|
|
|
|
Args:
|
|
symbol: 股票代码
|
|
|
|
Returns:
|
|
行情数据
|
|
"""
|
|
try:
|
|
from app.services.yfinance_service import get_yfinance_service
|
|
|
|
yf_service = get_yfinance_service()
|
|
quote = yf_service.get_ticker(symbol.upper())
|
|
|
|
if quote is None:
|
|
raise HTTPException(status_code=404, detail=f"无法获取 {symbol} 的行情")
|
|
|
|
return {
|
|
"success": True,
|
|
"symbol": symbol.upper(),
|
|
"quote": quote
|
|
}
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"获取 {symbol} 行情失败: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|