113 lines
3.3 KiB
Python
113 lines
3.3 KiB
Python
"""
|
||
信号 API - 提供加密货币信号查询接口
|
||
"""
|
||
from fastapi import APIRouter, HTTPException, Query
|
||
from typing import Dict, List, Optional, Any
|
||
|
||
from app.services.signal_database_service import get_signal_db_service
|
||
from app.utils.logger import logger
|
||
|
||
|
||
router = APIRouter(prefix="/api/signals", tags=["信号管理"])
|
||
|
||
|
||
@router.get("/crypto")
|
||
async def get_crypto_signals(
|
||
limit: int = Query(50, ge=1, le=200, description="返回数量限制"),
|
||
symbol: Optional[str] = Query(None, description="过滤指定交易对"),
|
||
days: int = Query(7, ge=1, le=30, description="查询最近多少天的信号")
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
获取加密货币信号列表
|
||
|
||
Args:
|
||
limit: 返回数量限制（默认50）
|
||
symbol: 过滤指定交易对
|
||
days: 查询最近多少天的信号（默认7天）
|
||
|
||
Returns:
|
||
信号列表
|
||
"""
|
||
try:
|
||
service = get_signal_db_service()
|
||
|
||
if symbol:
|
||
# 获取指定交易对的最新信号
|
||
signal = service.get_latest_signal('crypto', symbol)
|
||
return {
|
||
'success': True,
|
||
'symbol': symbol,
|
||
'signal': signal,
|
||
'count': 1 if signal else 0
|
||
}
|
||
else:
|
||
# 获取所有信号
|
||
signals = service.get_crypto_signals(limit=limit, days=days)
|
||
return {
|
||
'success': True,
|
||
'signals': signals,
|
||
'count': len(signals)
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取加密货币信号失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.get("/latest")
|
||
async def get_latest_signals(
|
||
limit: int = Query(20, ge=1, le=100, description="返回数量限制"),
|
||
days: int = Query(7, ge=1, le=30, description="查询最近多少天的信号")
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
获取最新的加密货币信号（按时间排序）
|
||
|
||
Args:
|
||
limit: 返回数量限制（默认20）
|
||
days: 查询最近多少天的信号（默认7天）
|
||
|
||
Returns:
|
||
最新信号列表
|
||
"""
|
||
try:
|
||
service = get_signal_db_service()
|
||
signals = service.get_crypto_signals(limit=limit, days=days)
|
||
|
||
return {
|
||
'success': True,
|
||
'signals': signals,
|
||
'count': len(signals)
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取最新信号失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.get("/stats")
|
||
async def get_signal_stats(
|
||
days: int = Query(7, ge=1, le=30, description="统计最近多少天的信号")
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
获取信号统计信息
|
||
|
||
Args:
|
||
days: 统计最近多少天的信号（默认7天）
|
||
|
||
Returns:
|
||
统计数据
|
||
"""
|
||
try:
|
||
service = get_signal_db_service()
|
||
stats = service.get_signal_stats(days=days)
|
||
|
||
return {
|
||
'success': True,
|
||
**{
|
||
'crypto': stats.get('crypto', {'total': 0, 'buy': 0, 'sell': 0, 'recent_24h': 0}),
|
||
'grades': stats.get('grades', {}),
|
||
'total': stats.get('crypto', {}).get('total', 0),
|
||
}
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取信号统计失败: {e}")
|
||
raise HTTPException(status_code=500, detail=str(e))
|