feat(recommendations): 添加现货推荐扫描与API支持

在后端API中新增现货推荐扫描功能,定时将数据写入Redis缓存,并提供相应的API接口以获取现货推荐。前端组件更新以支持现货推荐的展示与切换,提升用户体验与决策支持。此改动为用户提供了实时的现货推荐信息,增强了系统的功能性与灵活性。
This commit is contained in:
薇薇安 2026-02-25 08:40:52 +08:00
parent 1dea3df84a
commit 3389e0aafc
7 changed files with 767 additions and 85 deletions

View File

@ -5,6 +5,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from api.routes import config, trades, stats, dashboard, account, recommendations, system, accounts, auth, admin, public, data_management from api.routes import config, trades, stats, dashboard, account, recommendations, system, accounts, auth, admin, public, data_management
import os import os
import sys
import logging import logging
from pathlib import Path from pathlib import Path
from logging.handlers import RotatingFileHandler from logging.handlers import RotatingFileHandler
@ -165,6 +166,32 @@ app = FastAPI(
redirect_slashes=False # 禁用自动重定向,避免307重定向问题 redirect_slashes=False # 禁用自动重定向,避免307重定向问题
) )
# 现货推荐定时扫描间隔(秒),默认 15 分钟;设为 0 关闭定时扫描
SPOT_SCAN_INTERVAL_SEC = int(os.getenv("SPOT_SCAN_INTERVAL_SEC", "900"))
async def _spot_scan_loop():
"""后台循环:每隔 SPOT_SCAN_INTERVAL_SEC 执行一次现货扫描并写入 Redis。"""
if SPOT_SCAN_INTERVAL_SEC <= 0:
logger.info("现货推荐定时扫描已关闭(SPOT_SCAN_INTERVAL_SEC=0)")
return
import asyncio
backend_dir = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(backend_dir))
try:
from spot_scanner import run_spot_scan_and_cache
except Exception as e:
logger.warning("现货扫描模块加载失败,跳过定时任务: %s", e)
return
logger.info("现货推荐定时扫描已启动,间隔 %d 秒", SPOT_SCAN_INTERVAL_SEC)
while True:
try:
await run_spot_scan_and_cache(ttl_sec=900)
except Exception as e:
logger.warning("现货扫描执行失败: %s", e)
await asyncio.sleep(SPOT_SCAN_INTERVAL_SEC)
# 启动时:确保存在一个初始管理员(通过环境变量配置) # 启动时:确保存在一个初始管理员(通过环境变量配置)
@app.on_event("startup") @app.on_event("startup")
async def _ensure_initial_admin(): async def _ensure_initial_admin():
@ -201,6 +228,14 @@ async def _ensure_initial_admin():
except Exception as e: except Exception as e:
logger.warning(f"初始化管理员失败(可忽略): {e}") logger.warning(f"初始化管理员失败(可忽略): {e}")
# 启动现货推荐定时扫描(后台任务)
try:
import asyncio
asyncio.create_task(_spot_scan_loop())
except Exception as e:
logger.warning("启动现货扫描定时任务失败(可忽略): %s", e)
# CORS配置(允许React前端访问) # CORS配置(允许React前端访问)
# 默认包含:本地开发端口、主前端域名、推荐查看器域名 # 默认包含:本地开发端口、主前端域名、推荐查看器域名
cors_origins_str = os.getenv('CORS_ORIGINS', 'http://localhost:3000,http://localhost:3001,http://localhost:5173,http://as.deepx1.com,http://asapi.deepx1.com,http://r.deepx1.com,https://r.deepx1.com,http://asapi-new.deepx1.com') cors_origins_str = os.getenv('CORS_ORIGINS', 'http://localhost:3000,http://localhost:3001,http://localhost:5173,http://as.deepx1.com,http://asapi.deepx1.com,http://r.deepx1.com,https://r.deepx1.com,http://asapi-new.deepx1.com')

View File

@ -495,6 +495,72 @@ async def get_recommendations(
raise HTTPException(status_code=500, detail=f"获取推荐列表失败: {str(e)}") raise HTTPException(status_code=500, detail=f"获取推荐列表失败: {str(e)}")
REDIS_KEY_SPOT_SNAPSHOT = "recommendations:spot:snapshot"
@router.get("/spot")
async def get_spot_recommendations(
limit: int = Query(50, ge=1, le=200, description="返回数量限制"),
):
"""
获取现货推荐(只做多)。数据来自定时任务扫描并写入的 Redis 缓存。
"""
try:
rds = await _get_redis()
if rds is None:
raise HTTPException(status_code=503, detail="Redis 不可用,无法读取现货推荐缓存")
snapshot = await _get_cached_json(rds, REDIS_KEY_SPOT_SNAPSHOT)
if not isinstance(snapshot, dict):
return {
"success": True,
"count": 0,
"type": "spot",
"from_cache": False,
"meta": {"generated_at": None, "message": "暂无现货推荐数据,请等待定时扫描更新"},
"data": [],
}
items = snapshot.get("items") or []
if not isinstance(items, list):
items = []
items = items[:limit]
return {
"success": True,
"count": len(items),
"type": "spot",
"from_cache": True,
"meta": {
"generated_at": snapshot.get("generated_at"),
"generated_at_ms": snapshot.get("generated_at_ms"),
"ttl_sec": snapshot.get("ttl_sec"),
},
"data": items,
}
except HTTPException:
raise
except Exception as e:
logger.error(f"获取现货推荐失败: {e}")
raise HTTPException(status_code=500, detail=f"获取现货推荐失败: {str(e)}")
@router.post("/spot/scan")
async def trigger_spot_scan():
"""
手动触发一次现货扫描并更新 Redis 缓存(供定时任务或管理员调用)。
"""
try:
import sys
from pathlib import Path
backend_dir = Path(__file__).resolve().parent.parent.parent
if str(backend_dir) not in sys.path:
sys.path.insert(0, str(backend_dir))
from spot_scanner import run_spot_scan_and_cache
count = await run_spot_scan_and_cache(ttl_sec=900)
return {"success": True, "message": f"已扫描并缓存 {count} 条现货推荐", "count": count}
except Exception as e:
logger.error(f"现货扫描失败: {e}")
raise HTTPException(status_code=500, detail=f"现货扫描失败: {str(e)}")
@router.get("/active") @router.get("/active")
async def get_active_recommendations(): async def get_active_recommendations():
""" """

280
backend/spot_scanner.py Normal file
View File

@ -0,0 +1,280 @@
"""
现货推荐扫描:拉取币安现货行情,仅做多信号,写入 Redis 供 /api/recommendations/spot 使用。
使用公开 API,无需 API Key。定时任务调用 run_spot_scan_and_cache()。
"""
import asyncio
import json
import logging
import os
import sys
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import aiohttp
# 可选的 Redis 写入(与 recommendations 路由共用连接方式)
try:
import redis.asyncio as redis_async
except Exception:
redis_async = None
logger = logging.getLogger(__name__)
BINANCE_SPOT_BASE = "https://api.binance.com"
SPOT_KLINES_LIMIT = 60
SPOT_TOP_N = 80
SPOT_MIN_STRENGTH = 4
SPOT_MAX_RECS = 30
def _beijing_now_iso() -> str:
from datetime import timedelta
return datetime.now(tz=timezone(timedelta(hours=8))).isoformat()
async def _http_get(session: aiohttp.ClientSession, url: str, params: Optional[Dict] = None) -> Optional[Any]:
try:
async with session.get(url, params=params or {}, timeout=aiohttp.ClientTimeout(total=15)) as resp:
if resp.status != 200:
return None
return await resp.json()
except Exception as e:
logger.warning("spot_scanner _http_get %s: %s", url[:60], e)
return None
def _technical_indicators():
"""延迟导入 trading_system.indicators,避免 backend 强依赖 trading_system 路径。"""
project_root = __import__("pathlib").Path(__file__).resolve().parent.parent
trading_system = project_root / "trading_system"
if str(trading_system) not in sys.path:
sys.path.insert(0, str(trading_system))
try:
from indicators import TechnicalIndicators
return TechnicalIndicators
except ImportError:
from trading_system.indicators import TechnicalIndicators
return TechnicalIndicators
async def _fetch_spot_symbols(session: aiohttp.ClientSession) -> List[str]:
"""获取所有 USDT 现货交易对(status=TRADING)。"""
data = await _http_get(session, f"{BINANCE_SPOT_BASE}/api/v3/exchangeInfo")
if not data or "symbols" not in data:
return []
symbols = []
for s in data["symbols"]:
if s.get("status") != "TRADING":
continue
if s.get("quoteAsset") != "USDT":
continue
sym = s.get("symbol")
if sym:
symbols.append(sym)
return symbols
async def _fetch_spot_ticker_24h(session: aiohttp.ClientSession) -> List[Dict]:
"""获取 24h ticker,返回 list of dict (symbol, lastPrice, priceChangePercent, volume, ...)。"""
data = await _http_get(session, f"{BINANCE_SPOT_BASE}/api/v3/ticker/24hr")
if not isinstance(data, list):
return []
return data
async def _fetch_spot_klines(session: aiohttp.ClientSession, symbol: str, interval: str = "15m", limit: int = 60) -> Optional[List[List]]:
"""现货 K 线,格式与合约一致 [open_time, o, h, l, c, volume, ...]。"""
data = await _http_get(
session,
f"{BINANCE_SPOT_BASE}/api/v3/klines",
{"symbol": symbol, "interval": interval, "limit": limit},
)
return data if isinstance(data, list) else None
def _compute_spot_signal(klines: List[List], ticker: Dict, TechnicalIndicators) -> Optional[Dict]:
"""
基于 K 线计算只做多信号。返回 None 或 { direction: 'BUY', strength: int, ... }。
"""
if not klines or len(klines) < 50:
return None
closes = [float(k[4]) for k in klines]
highs = [float(k[2]) for k in klines]
lows = [float(k[3]) for k in klines]
current_price = closes[-1]
rsi = TechnicalIndicators.calculate_rsi(closes, period=14)
macd = TechnicalIndicators.calculate_macd(closes)
bollinger = TechnicalIndicators.calculate_bollinger_bands(closes, period=20)
ema20 = TechnicalIndicators.calculate_ema(closes, period=20)
ema50 = TechnicalIndicators.calculate_ema(closes, period=50)
strength = 0
# 只做多:RSI 超卖、价格在下轨附近、MACD 金叉、价格在均线上方等
if rsi is not None and rsi < 35:
strength += 3
elif rsi is not None and rsi < 50:
strength += 1
if bollinger and current_price <= bollinger["lower"] * 1.002:
strength += 3
elif bollinger and current_price < bollinger["middle"]:
strength += 1
if macd and macd["histogram"] > 0 and macd["macd"] > macd["signal"]:
strength += 2
if ema20 and ema50 and current_price > ema20 > ema50:
strength += 2
elif ema20 and current_price > ema20:
strength += 1
strength = max(0, min(strength, 10))
if strength < SPOT_MIN_STRENGTH:
return None
return {
"direction": "BUY",
"strength": strength,
"rsi": rsi,
"current_price": current_price,
}
def _build_spot_recommendation(
symbol: str,
ticker: Dict,
signal: Dict,
) -> Dict[str, Any]:
"""构造单条现货推荐(与合约推荐结构兼容,便于前端复用)。"""
current_price = float(ticker.get("lastPrice") or signal.get("current_price") or 0)
change_percent = float(ticker.get("priceChangePercent") or 0)
ts = time.time()
entry = current_price * 0.995
stop_pct = 0.05
tp1_pct = 0.08
tp2_pct = 0.15
if current_price <= 0:
return None
stop_loss = entry * (1 - stop_pct)
tp1 = entry * (1 + tp1_pct)
tp2 = entry * (1 + tp2_pct)
return {
"symbol": symbol,
"direction": "BUY",
"market": "spot",
"current_price": current_price,
"signal_strength": signal.get("strength", 0),
"change_percent": change_percent,
"suggested_limit_price": entry,
"planned_entry_price": entry,
"suggested_stop_loss": stop_loss,
"suggested_take_profit_1": tp1,
"suggested_take_profit_2": tp2,
"suggested_position_percent": 0.05,
"recommendation_time": _beijing_now_iso(),
"timestamp": ts,
"recommendation_reason": "现货做多信号(RSI/布林带/MACD/均线)",
"user_guide": f"现货建议在 {entry:.4f} USDT 附近买入,止损 {stop_loss:.4f},目标1 {tp1:.4f},目标2 {tp2:.4f}。仅供参考,请自行判断。",
}
async def run_spot_scan() -> List[Dict[str, Any]]:
"""执行一次现货扫描,返回推荐列表(不写 Redis)。"""
TechnicalIndicators = _technical_indicators()
recommendations = []
async with aiohttp.ClientSession() as session:
symbols = await _fetch_spot_symbols(session)
if not symbols:
logger.warning("spot_scanner: 未获取到现货交易对")
return []
tickers = await _fetch_spot_ticker_24h(session)
ticker_map = {t["symbol"]: t for t in tickers if isinstance(t.get("symbol"), str)}
# 按 24h 成交量排序,取前 SPOT_TOP_N 再按涨跌幅取部分
def volume_key(t):
try:
return float(t.get("volume") or 0) * float(t.get("lastPrice") or 0)
except Exception:
return 0
sorted_tickers = sorted(
[t for t in tickers if t.get("symbol") in symbols],
key=volume_key,
reverse=True,
)[: SPOT_TOP_N * 2]
# 按涨跌幅取前 N 个(偏强势或超跌反弹)
with_change = [(t, float(t.get("priceChangePercent") or 0)) for t in sorted_tickers]
with_change.sort(key=lambda x: -abs(x[1]))
to_scan = [t[0]["symbol"] for t in with_change[: SPOT_TOP_N]]
for symbol in to_scan:
try:
klines = await _fetch_spot_klines(session, symbol, "15m", SPOT_KLINES_LIMIT)
ticker = ticker_map.get(symbol, {})
if not klines or not ticker:
continue
signal = _compute_spot_signal(klines, ticker, TechnicalIndicators)
if not signal:
continue
rec = _build_spot_recommendation(symbol, ticker, signal)
if rec:
recommendations.append(rec)
if len(recommendations) >= SPOT_MAX_RECS:
break
except Exception as e:
logger.debug("spot_scanner %s: %s", symbol, e)
await asyncio.sleep(0.05)
recommendations.sort(key=lambda x: x.get("signal_strength", 0), reverse=True)
return recommendations[: SPOT_MAX_RECS]
def _redis_connection_kwargs():
redis_url = (os.getenv("REDIS_URL", "") or "").strip() or "redis://localhost:6379"
kwargs = {"decode_responses": True}
if os.getenv("REDIS_USERNAME"):
kwargs["username"] = os.getenv("REDIS_USERNAME")
if os.getenv("REDIS_PASSWORD"):
kwargs["password"] = os.getenv("REDIS_PASSWORD")
if redis_url.startswith("rediss://") or os.getenv("REDIS_USE_TLS", "").lower() == "true":
if redis_url.startswith("redis://"):
redis_url = redis_url.replace("redis://", "rediss://", 1)
kwargs.setdefault("ssl_cert_reqs", os.getenv("REDIS_SSL_CERT_REQS", "required"))
if os.getenv("REDIS_SSL_CA_CERTS"):
kwargs["ssl_ca_certs"] = os.getenv("REDIS_SSL_CA_CERTS")
return redis_url, kwargs
async def run_spot_scan_and_cache(ttl_sec: int = 900) -> int:
"""
执行现货扫描并写入 Redis。返回写入的推荐数量。
Redis key: recommendations:spot:snapshot
"""
items = await run_spot_scan()
now_ms = int(time.time() * 1000)
payload = {
"items": items,
"generated_at": _beijing_now_iso(),
"generated_at_ms": now_ms,
"ttl_sec": ttl_sec,
"count": len(items),
}
if redis_async is None:
logger.warning("spot_scanner: redis 不可用,跳过写入")
return len(items)
redis_url, kwargs = _redis_connection_kwargs()
try:
client = redis_async.from_url(redis_url, **kwargs)
await client.ping()
key = "recommendations:spot:snapshot"
await client.setex(key, ttl_sec, json.dumps(payload, ensure_ascii=False))
logger.info("spot_scanner: 已写入 %d 条现货推荐到 %s", len(items), key)
await client.aclose()
return len(items)
except Exception as e:
logger.warning("spot_scanner: Redis 写入失败 %s", e)
return len(items)

View File

@ -73,6 +73,41 @@
cursor: pointer; cursor: pointer;
} }
.market-type-tabs {
display: flex;
gap: 0;
margin-bottom: 12px;
border-bottom: 1px solid #e0e0e0;
}
.market-type-tabs .tab {
padding: 10px 20px;
border: none;
background: transparent;
font-size: 15px;
color: #666;
cursor: pointer;
border-bottom: 2px solid transparent;
margin-bottom: -1px;
transition: color 0.2s, border-color 0.2s;
}
.market-type-tabs .tab:hover {
color: #333;
}
.market-type-tabs .tab.active {
color: #2196F3;
font-weight: 500;
border-bottom-color: #2196F3;
}
.spot-hint {
margin: 0 0 16px 0;
font-size: 13px;
color: #666;
}
.error-message { .error-message {
padding: 12px; padding: 12px;
background-color: #ffebee; background-color: #ffebee;

View File

@ -10,86 +10,23 @@ function RecommendationsViewer() {
const [error, setError] = useState(null); const [error, setError] = useState(null);
const [directionFilter, setDirectionFilter] = useState(''); const [directionFilter, setDirectionFilter] = useState('');
const [showDetails, setShowDetails] = useState({}); const [showDetails, setShowDetails] = useState({});
const [marketType, setMarketType] = useState('futures'); // 'futures' | 'spot'
useEffect(() => {
loadRecommendations();
// 每10秒静默更新价格(不触发loading状态)
const interval = setInterval(async () => {
try {
const result = await api.getRecommendations({
type: 'realtime',
direction: directionFilter,
limit: 50,
min_signal_strength: 5
});
const newData = result.data || [];
// 使用setState直接更新,不触发loading状态
setRecommendations(prevRecommendations => {
if (newData.length === 0) {
return prevRecommendations;
}
// 实时推荐没有id,使用symbol作为key
const newDataMap = new Map(newData.map(rec => [rec.symbol, rec]));
const prevMap = new Map(prevRecommendations.map(rec => [rec.symbol || rec.id, rec]));
// 合并数据:优先使用新数据(包含实时价格更新)
const updated = prevRecommendations.map(prevRec => {
const key = prevRec.symbol || prevRec.id;
const newRec = newDataMap.get(key);
if (newRec) {
return newRec;
}
return prevRec;
});
// 添加新出现的推荐
const newItems = newData.filter(newRec => !prevMap.has(newRec.symbol));
// 合并并去重(按symbol)
const merged = [...updated, ...newItems];
const uniqueMap = new Map();
merged.forEach(rec => {
const key = rec.symbol || rec.id;
if (!uniqueMap.has(key)) {
uniqueMap.set(key, rec);
}
});
return Array.from(uniqueMap.values());
});
} catch (err) {
// 静默失败,不显示错误
console.debug('静默更新价格失败:', err);
}
}, 10000); // 每10秒刷新
return () => {
clearInterval(interval);
};
}, [directionFilter]);
const loadRecommendations = async () => { const loadRecommendations = async () => {
try { try {
setLoading(true); setLoading(true);
setError(null); setError(null);
if (marketType === 'spot') {
const params = { const result = await api.getSpotRecommendations({ limit: 50 });
type: 'realtime', const data = result.data || [];
limit: 50, setRecommendations(data);
min_signal_strength: 5 } else {
}; const params = { type: 'realtime', limit: 50, min_signal_strength: 5 };
if (directionFilter) params.direction = directionFilter;
if (directionFilter) { const result = await api.getRecommendations(params);
params.direction = directionFilter; const data = result.data || [];
setRecommendations(data);
} }
const result = await api.getRecommendations(params);
const data = result.data || [];
setRecommendations(data);
} catch (err) { } catch (err) {
setError(err.message); setError(err.message);
console.error('加载推荐失败:', err); console.error('加载推荐失败:', err);
@ -98,6 +35,58 @@ function RecommendationsViewer() {
} }
}; };
useEffect(() => {
loadRecommendations();
}, [marketType]);
useEffect(() => {
if (marketType !== 'futures') return;
// 合约:每10秒静默更新价格(不触发loading状态)
const interval = setInterval(async () => {
try {
const result = await api.getRecommendations({
type: 'realtime',
direction: directionFilter,
limit: 50,
min_signal_strength: 5
});
const newData = result.data || [];
if (newData.length === 0) return;
const newDataMap = new Map(newData.map(rec => [rec.symbol, rec]));
setRecommendations(prev => {
const prevKeys = new Set(prev.map(rec => rec.symbol || rec.id));
const updated = prev.map(prevRec => {
const key = prevRec.symbol || prevRec.id;
return newDataMap.get(key) || prevRec;
});
const newItems = newData.filter(newRec => !prevKeys.has(newRec.symbol));
const merged = [...updated, ...newItems];
const uniqueMap = new Map();
merged.forEach(rec => { uniqueMap.set(rec.symbol || rec.id, rec); });
return Array.from(uniqueMap.values());
});
} catch (err) {
console.debug('静默更新价格失败:', err);
}
}, 10000);
return () => clearInterval(interval);
}, [marketType, directionFilter]);
// 现货:每 30 秒静默刷新列表
useEffect(() => {
if (marketType !== 'spot') return;
const interval = setInterval(async () => {
try {
const result = await api.getSpotRecommendations({ limit: 50 });
const data = result.data || [];
if (data.length) setRecommendations(data);
} catch (err) {
console.debug('静默更新现货推荐失败:', err);
}
}, 30000);
return () => clearInterval(interval);
}, [marketType]);
const toggleDetails = (key) => { const toggleDetails = (key) => {
setShowDetails(prev => ({ setShowDetails(prev => ({
...prev, ...prev,
@ -204,7 +193,9 @@ function RecommendationsViewer() {
<div className="viewer-header"> <div className="viewer-header">
<h1>交易推荐</h1> <h1>交易推荐</h1>
<div className="header-info"> <div className="header-info">
<span className="update-info">每10秒自动更新</span> <span className="update-info">
{marketType === 'spot' ? '现货推荐每30秒更新' : '合约每10秒自动更新'}
</span>
<button <button
className="btn-refresh" className="btn-refresh"
onClick={loadRecommendations} onClick={loadRecommendations}
@ -215,17 +206,39 @@ function RecommendationsViewer() {
</div> </div>
</div> </div>
<div className="filters"> <div className="market-type-tabs">
<select <button
value={directionFilter} type="button"
onChange={(e) => setDirectionFilter(e.target.value)} className={`tab ${marketType === 'futures' ? 'active' : ''}`}
className="filter-select" onClick={() => setMarketType('futures')}
> >
<option value="">全部方向</option> 合约
<option value="BUY">做多</option> </button>
<option value="SELL">做空</option> <button
</select> type="button"
className={`tab ${marketType === 'spot' ? 'active' : ''}`}
onClick={() => setMarketType('spot')}
>
现货
</button>
</div> </div>
{marketType === 'spot' && (
<p className="spot-hint">现货推荐(只做多),数据来自定时扫描,仅供参考。</p>
)}
{marketType === 'futures' && (
<div className="filters">
<select
value={directionFilter}
onChange={(e) => setDirectionFilter(e.target.value)}
className="filter-select"
>
<option value="">全部方向</option>
<option value="BUY">做多</option>
<option value="SELL">做空</option>
</select>
</div>
)}
{error && ( {error && (
<div className="error-message"> <div className="error-message">

View File

@ -13,7 +13,7 @@ const buildUrl = (path) => {
}; };
export const api = { export const api = {
// 获取实时推荐 // 获取实时推荐(合约)
getRecommendations: async (params = {}) => { getRecommendations: async (params = {}) => {
// 默认使用实时推荐 // 默认使用实时推荐
if (!params.type) { if (!params.type) {
@ -27,5 +27,17 @@ export const api = {
throw new Error(error.detail || '获取推荐失败'); throw new Error(error.detail || '获取推荐失败');
} }
return response.json(); return response.json();
},
// 获取现货推荐
getSpotRecommendations: async (params = {}) => {
const query = new URLSearchParams(params).toString();
const url = query ? `${buildUrl('/api/recommendations/spot')}?${query}` : buildUrl('/api/recommendations/spot');
const response = await fetch(url);
if (!response.ok) {
const error = await response.json().catch(() => ({ detail: '获取现货推荐失败' }));
throw new Error(error.detail || '获取现货推荐失败');
}
return response.json();
} }
}; };

View File

@ -0,0 +1,241 @@
"""
现货推荐扫描:使用币安现货公开 API,只做多信号,供定时任务写入 Redis。
不依赖合约 API 和 API Key,仅使用 api.binance.com 公开接口。
"""
import asyncio
import logging
import time
from datetime import datetime, timezone
from typing import List, Dict, Any, Optional
import aiohttp
try:
from .indicators import TechnicalIndicators
except ImportError:
from indicators import TechnicalIndicators
logger = logging.getLogger(__name__)
BINANCE_SPOT_BASE = "https://api.binance.com"
BEIJING_TZ = timezone(offset=__import__("datetime").timedelta(hours=8))
# 默认只扫描成交量靠前的现货 USDT 对数量
DEFAULT_TOP_N = 80
# 每个 symbol 的 K 线请求间隔,避免触发限频
KLINES_DELAY = 0.15
async def _http_get(session: aiohttp.ClientSession, url: str, params: Optional[Dict] = None, timeout: int = 15) -> Any:
try:
async with session.get(url, params=params or {}, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
if resp.status != 200:
text = await resp.text()
logger.warning(f"现货 API 请求失败 {url} status={resp.status} body={text[:200]}")
return None
return await resp.json()
except Exception as e:
logger.warning(f"现货 API 请求异常 {url}: {e}")
return None
async def get_spot_tickers_24h(session: aiohttp.ClientSession) -> List[Dict]:
"""获取现货 24h 行情,用于按成交量排序、取涨跌幅。"""
url = f"{BINANCE_SPOT_BASE}/api/v3/ticker/24hr"
data = await _http_get(session, url)
if not isinstance(data, list):
return []
# 只保留 USDT 交易对
out = [t for t in data if isinstance(t, dict) and str(t.get("symbol", "")).endswith("USDT")]
return out
async def get_spot_klines(session: aiohttp.ClientSession, symbol: str, interval: str = "15m", limit: int = 100) -> Optional[List[List]]:
"""获取现货 K 线,格式与合约一致 [open_time, o, h, l, c, volume, ...]。"""
url = f"{BINANCE_SPOT_BASE}/api/v3/klines"
params = {"symbol": symbol, "interval": interval, "limit": limit}
data = await _http_get(session, url, params)
if not isinstance(data, list) or len(data) < 20:
return None
return data
def _score_spot_long(ticker: Dict, klines: Optional[List], rsi: Optional[float], macd: Optional[Dict]) -> int:
"""
现货做多评分 0-10。
考虑:24h 涨跌幅不过度下跌、RSI 非超买、MACD 偏多或金叉。
"""
score = 5 # 中性起点
change = 0.0
try:
change = float(ticker.get("priceChangePercent", 0) or 0)
except (TypeError, ValueError):
pass
# 24h 涨跌幅:正加分,负减分
if change >= 2:
score += 1
elif change >= 0:
score += 0
elif change >= -2:
score -= 0
else:
score -= 1
if rsi is not None:
if rsi < 30:
score += 1 # 超卖反弹机会
elif rsi < 45:
score += 0
elif rsi > 70:
score -= 2 # 超买不推荐
elif rsi > 55:
score -= 0
if macd and isinstance(macd, dict):
hist = macd.get("histogram") or 0
if hist > 0:
score += 1
elif hist < 0:
score -= 0
return max(0, min(10, int(score)))
async def run_spot_scan(
top_n: int = DEFAULT_TOP_N,
min_signal_strength: int = 5,
kline_interval: str = "15m",
kline_limit: int = 100,
) -> List[Dict[str, Any]]:
"""
执行现货扫描,返回推荐列表(仅做多)。
按 24h 成交量取前 top_n 个 USDT 交易对,拉 K 线算 RSI/MACD,评分后过滤。
"""
recommendations = []
async with aiohttp.ClientSession() as session:
tickers = await get_spot_tickers_24h(session)
if not tickers:
logger.warning("现货 24h ticker 未获取到数据")
return []
# 按 quoteVolume 排序,取前 top_n
def vol_key(t):
try:
return float(t.get("quoteVolume") or 0)
except (TypeError, ValueError):
return 0.0
tickers.sort(key=vol_key, reverse=True)
symbols_to_scan = [t["symbol"] for t in tickers[: top_n * 2]] # 多取一些,后面按分数再筛
for i, ticker in enumerate(tickers[:top_n]):
symbol = ticker.get("symbol")
if not symbol or not str(symbol).endswith("USDT"):
continue
try:
klines = await get_spot_klines(session, symbol, kline_interval, kline_limit)
await asyncio.sleep(KLINES_DELAY)
except Exception as e:
logger.debug(f"获取 {symbol} K线失败: {e}")
klines = None
if not klines or len(klines) < 30:
continue
closes = [float(k[4]) for k in klines]
highs = [float(k[2]) for k in klines]
lows = [float(k[3]) for k in klines]
rsi = TechnicalIndicators.calculate_rsi(closes, period=14)
macd = TechnicalIndicators.calculate_macd(closes, 12, 26, 9)
score = _score_spot_long(ticker, klines, rsi, macd)
if score < min_signal_strength:
continue
try:
price = float(ticker.get("lastPrice") or ticker.get("weightedAvgPrice") or closes[-1])
except (TypeError, ValueError):
price = closes[-1]
try:
change_percent = float(ticker.get("priceChangePercent") or 0)
except (TypeError, ValueError):
change_percent = 0.0
recommendation_time = datetime.now(BEIJING_TZ).isoformat()
rec = {
"symbol": symbol,
"direction": "BUY",
"side": "BUY",
"signal_strength": score,
"current_price": price,
"change_percent": change_percent,
"recommendation_time": recommendation_time,
"market": "spot",
"recommendation_reason": f"现货做多信号 RSI={rsi:.1f if rsi else '-'} 24h涨跌{change_percent:+.2f}%",
"user_guide": f"现货建议在 {price:.4f} 附近分批买入,并自行设置止损止盈。",
"timestamp": time.time(),
}
recommendations.append(rec)
# 按信号强度排序
recommendations.sort(key=lambda x: (x.get("signal_strength", 0), -(x.get("change_percent") or 0)), reverse=True)
return recommendations[:50] # 最多返回 50 条
async def run_spot_scan_and_cache_redis(ttl_sec: int = 3600) -> int:
"""
执行现货扫描并将结果写入 Redis(key: recommendations:spot:snapshot)。
需要环境变量 REDIS_URL 等与后端一致。
返回写入的推荐条数。
"""
import os
import json
try:
import redis.asyncio as redis_async
except Exception:
logger.warning("redis.asyncio 不可用,无法写入现货推荐缓存")
return 0
redis_url = (os.getenv("REDIS_URL", "") or "").strip() or "redis://localhost:6379"
kwargs = {"decode_responses": True}
if os.getenv("REDIS_USERNAME"):
kwargs["username"] = os.getenv("REDIS_USERNAME")
if os.getenv("REDIS_PASSWORD"):
kwargs["password"] = os.getenv("REDIS_PASSWORD")
if redis_url.startswith("rediss://") or os.getenv("REDIS_USE_TLS", "").lower() == "true":
if not redis_url.startswith("rediss://"):
redis_url = redis_url.replace("redis://", "rediss://", 1)
kwargs.setdefault("ssl_cert_reqs", os.getenv("REDIS_SSL_CERT_REQS", "required"))
try:
rds = redis_async.from_url(redis_url, **kwargs)
await rds.ping()
except Exception as e:
logger.warning(f"Redis 连接失败: {e}")
return 0
try:
items = await run_spot_scan(top_n=DEFAULT_TOP_N, min_signal_strength=5)
now_ms = int(time.time() * 1000)
payload = {
"items": items,
"generated_at_ms": now_ms,
"generated_at": datetime.now(BEIJING_TZ).strftime("%Y-%m-%d %H:%M:%S"),
"ttl_sec": ttl_sec,
"market": "spot",
}
await rds.setex(
"recommendations:spot:snapshot",
ttl_sec,
json.dumps(payload, ensure_ascii=False),
)
logger.info(f"现货推荐已写入 Redis,共 {len(items)} 条,TTL={ttl_sec}s")
await rds.aclose()
return len(items)
except Exception as e:
logger.exception(f"现货扫描或写入 Redis 失败: {e}")
try:
await rds.aclose()
except Exception:
pass
return 0