From 61486d0c5b0ce42dc12b77df55b0feb42ac6da06 Mon Sep 17 00:00:00 2001 From: eezd Date: Wed, 14 Oct 2026 18:16:54 +0800 Subject: [PATCH] Add derivatives tool, trim provider fields, widen model call retries - New chat tool get_derivatives: funding rate (next estimate and recent settlements), premium, open interest change vs price, global and top trader long/short ratios, taker buy/sell ratio from Binance public endpoints. Each part fails independently. - Strip whitespace from provider name and model; a leading space in the model name made the upstream return 404. - Retry connection failures and mid-stream disconnects (raw httpx / httpx2 errors are not wrapped by the SDKs) up to 3 times while no output has been emitted, and surface them as chat errors. - Return 30 recent bars from get_klines instead of 60 to cut tokens. --- backend/src/alphapulse/ai/llm.py | 27 +++- backend/src/alphapulse/ai/prompt.py | 2 +- backend/src/alphapulse/ai/tools.py | 115 +++++++++++++++++- backend/src/alphapulse/api/settings.py | 13 +- backend/src/alphapulse/market/binance.py | 8 +- web/src/pages/chat/components/MessageList.vue | 1 + 6 files changed, 152 insertions(+), 14 deletions(-) diff --git a/backend/src/alphapulse/ai/llm.py b/backend/src/alphapulse/ai/llm.py index 4c67308..17858b6 100644 --- a/backend/src/alphapulse/ai/llm.py +++ b/backend/src/alphapulse/ai/llm.py @@ -14,6 +14,8 @@ from dataclasses import dataclass, field from typing import Any, Literal import anthropic +import httpx +import httpx2 # anthropic SDK 使用的 HTTP 客户端，异常类与 httpx 不通用 import openai from alphapulse.core.crypto import decrypt @@ -25,7 +27,8 @@ TIMEOUT_SECONDS = 600 Effort = Literal["low", "medium", "high", "xhigh", "max"] # SDK 自带重试只针对连接错误、429 和 5xx MAX_RETRIES = 2 -# 流内错误（上游过载等）SDK 不重试，由这里重试；只在还没有输出内容时重试 +# 流内错误（上游过载等）和流式读取中的网络中断 SDK 不重试；连接失败 SDK 重试后仍可能失败（上游短暂不可达）。 +# 这些都由这里再重试，且只在还没有输出内容时重试 STREAM_RETRIES = 3 STREAM_RETRY_DELAY_SECONDS = 2 @@ -84,11 +87,12 @@ async def stream( output = True yield event return - except (openai.APIError, anthropic.APIError) as e: - # 已经推送过内容再重试会重复输出，只重试还没有输出时的流内错误 - if output or attempt == STREAM_RETRIES or not _in_stream_error(e): + # 流式读取过程中的网络错误 SDK 不包装，以原始的 httpx / httpx2 异常抛出 + except (openai.APIError, anthropic.APIError, httpx.TransportError, httpx2.TransportError) as e: + # 已经推送过内容再重试会重复输出 + if output or attempt == STREAM_RETRIES or not _retryable(e): raise _llm_error(e) from e - logger.warning("模型流内出错，%ss 后第 %s 次重试：%s", STREAM_RETRY_DELAY_SECONDS, attempt + 1, _llm_error(e)) + logger.warning("模型调用失败，%ss 后第 %s 次重试：%s", STREAM_RETRY_DELAY_SECONDS, attempt + 1, _llm_error(e)) await asyncio.sleep(STREAM_RETRY_DELAY_SECONDS) @@ -100,6 +104,15 @@ def _in_stream_error(e: Exception) -> bool: return isinstance(e, anthropic.APIStatusError) and e.status_code == 200 +def _retryable(e: Exception) -> bool: + """流内错误事件、连接失败、读取中断可重试；超时不重试（单次已等 TIMEOUT_SECONDS）""" + if isinstance(e, (openai.APITimeoutError, anthropic.APITimeoutError, httpx.TimeoutException, httpx2.TimeoutException)): + return False + return _in_stream_error(e) or isinstance( + e, (openai.APIConnectionError, anthropic.APIConnectionError, httpx.TransportError, httpx2.TransportError), + ) + + def _llm_error(e: Exception) -> LlmError: if _in_stream_error(e): error = e.body.get("error") if isinstance(e.body, dict) else None @@ -107,10 +120,12 @@ def _llm_error(e: Exception) -> LlmError: return LlmError(f"模型接口出错：{message or e.message}") if isinstance(e, (openai.APIStatusError, anthropic.APIStatusError)): return LlmError(f"模型接口返回 {e.status_code}：{e.message}") - if isinstance(e, (openai.APITimeoutError, anthropic.APITimeoutError)): + if isinstance(e, (openai.APITimeoutError, anthropic.APITimeoutError, httpx.TimeoutException, httpx2.TimeoutException)): return LlmError(f"模型接口超时（{TIMEOUT_SECONDS}s）") if isinstance(e, (openai.APIConnectionError, anthropic.APIConnectionError)): return LlmError(f"无法连接模型接口：{e}") + if isinstance(e, (httpx.TransportError, httpx2.TransportError)): + return LlmError(f"模型接口连接中断：{e!r}") return LlmError(f"模型接口出错：{e}") diff --git a/backend/src/alphapulse/ai/prompt.py b/backend/src/alphapulse/ai/prompt.py index ad3dd28..985d045 100644 --- a/backend/src/alphapulse/ai/prompt.py +++ b/backend/src/alphapulse/ai/prompt.py @@ -12,7 +12,7 @@ from alphapulse.models import ContextVersion, MarketSymbol, NewsItem BASE_INSTRUCTIONS = """你是用户的个人投资研究助手，服务于 AlphaPulse 系统。 ## 工作方式 -- 先用工具取证据，再下结论：新闻用 search_news / get_news，行情用 search_symbols / get_klines / get_ticker。不要凭记忆给出价格、日期或新闻内容。 +- 先用工具取证据，再下结论：新闻用 search_news / get_news，行情用 search_symbols / get_klines / get_ticker / get_derivatives。不要凭记忆给出价格、日期或新闻内容。 - 引用新闻时写明 [新闻 id] 和发布时间；引用 K 线时写明交易对和周期。 - 点位（抄底区间、止损、止盈）必须给出依据：来自 get_klines 的高低点、均线、ATR、波段高低点，并说明失效条件。 - 数据不足或互相矛盾时如实说明缺什么，不编造。 diff --git a/backend/src/alphapulse/ai/tools.py b/backend/src/alphapulse/ai/tools.py index 684a6a4..3f17f38 100644 --- a/backend/src/alphapulse/ai/tools.py +++ b/backend/src/alphapulse/ai/tools.py @@ -4,6 +4,7 @@ 工具出错时把错误文本返回给模型，不编造数据。 """ +import asyncio import json from collections.abc import Awaitable, Callable from dataclasses import dataclass @@ -26,7 +27,7 @@ from alphapulse.models.market import INTERVALS LOCAL_TZ = ZoneInfo("Asia/Shanghai") NEWS_BODY_LIMIT = 4000 # 返回给模型的逐根 K 线数量；统计指标基于全部取到的 K 线计算 -BARS_IN_RESULT = 60 +BARS_IN_RESULT = 30 SOURCE_NAMES = {"cls_telegraph": "财联社电报", "cls_depth": "财联社深度", "wscn_live": "华尔街见闻快讯", "wscn_news": "华尔街见闻文章"} @@ -195,6 +196,114 @@ async def get_ticker(args: GetTicker) -> ToolOutcome: })) +# ---------- 衍生品数据 ---------- + +# 每个序列返回给模型的最近点数；窗口统计基于全部取到的数据 +POINTS_IN_RESULT = 12 +FUNDING_HISTORY = 30 + + +class GetDerivatives(BaseModel): + symbol: Annotated[str, Field(description="U 本位永续交易对，如 BTCUSDT")] + period: Annotated[ + Literal["5m", "15m", "30m", "1h", "2h", "4h", "6h", "12h", "1d"], + Field(description="持仓量、多空比、主动买卖比的统计周期"), + ] = "1h" + limit: Annotated[int, Field(ge=2, le=500, description="统计窗口的周期数，如 1h × 24 为最近一天；币安只提供最近 30 天")] = 24 + + +def pct(value: float) -> float: + return round(value * 100, 4) + + +def ratio_summary(rows: list[dict], key: str) -> dict: + values = [float(r[key]) for r in rows] + return { + "latest": values[-1], "window_avg": round(sum(values) / len(values), 4), "window_min": min(values), "window_max": max(values), + "recent": [[ms_local(r["timestamp"]), float(r[key])] for r in rows[-POINTS_IN_RESULT:]], + } + + +async def funding_part(symbol: str) -> dict: + premium, history = await asyncio.gather( + call_binance(client.get("um", "premium_index", {"symbol": symbol})), + call_binance(client.get("um", "funding_rate", {"symbol": symbol, "limit": FUNDING_HISTORY})), + ) + mark, index = float(premium["markPrice"]), float(premium["indexPrice"]) + rates = [float(r["fundingRate"]) for r in history] + return { + "unit": "%", + # premiumIndex 的 lastFundingRate 随行情实时变化，是下次结算的预估值 + "next_rate_estimate_pct": pct(float(premium["lastFundingRate"])), + "next_funding_time": ms_local(premium["nextFundingTime"]), + "interval_hours": round((history[-1]["fundingTime"] - history[-2]["fundingTime"]) / 3_600_000) if len(history) >= 2 else None, + "mark_price": mark, "index_price": index, "premium_pct": pct((mark - index) / index), + "history_count": len(rates), + "history_avg_pct": pct(sum(rates) / len(rates)) if rates else None, + "history_max_pct": pct(max(rates)) if rates else None, + "history_min_pct": pct(min(rates)) if rates else None, + "recent_settled": [[ms_local(r["fundingTime"]), pct(float(r["fundingRate"]))] for r in history[-8:]], + } + + +async def open_interest_part(symbol: str, period: str, limit: int) -> dict: + rows = await call_binance(client.get("um", "open_interest_hist", {"symbol": symbol, "period": period, "limit": limit})) + if not rows: + return {"error": "无数据"} + first, last = rows[0], rows[-1] + + def price(r: dict) -> float: + return float(r["sumOpenInterestValue"]) / float(r["sumOpenInterest"]) + + return { + "latest_contracts": float(last["sumOpenInterest"]), "latest_value_usdt": round(float(last["sumOpenInterestValue"])), + "window_change_pct": pct(float(last["sumOpenInterest"]) / float(first["sumOpenInterest"]) - 1), + # 由持仓价值 / 持仓量推算的同窗口价格变化，用于对照持仓量与价格的方向 + "window_price_change_pct": pct(price(last) / price(first) - 1), + "recent": [[ms_local(r["timestamp"]), float(r["sumOpenInterest"])] for r in rows[-POINTS_IN_RESULT:]], + } + + +async def ratio_part(path_key: str, symbol: str, period: str, limit: int, key: str) -> dict: + rows = await call_binance(client.get("um", path_key, {"symbol": symbol, "period": period, "limit": limit})) + return ratio_summary(rows, key) if rows else {"error": "无数据"} + + +async def taker_part(symbol: str, period: str, limit: int) -> dict: + rows = await call_binance(client.get("um", "taker_buy_sell", {"symbol": symbol, "period": period, "limit": limit})) + if not rows: + return {"error": "无数据"} + buy, sell = sum(float(r["buyVol"]) for r in rows), sum(float(r["sellVol"]) for r in rows) + return {**ratio_summary(rows, "buySellRatio"), "window_buy_vol": buy, "window_sell_vol": sell} + + +async def settle(request: Awaitable[dict]) -> dict: + """单项失败只标记该项，其余照常返回""" + try: + return await request + except HTTPException as e: + return {"error": e.detail} + + +async def get_derivatives(args: GetDerivatives) -> ToolOutcome: + symbol = args.symbol.strip().upper() + funding, oi, accounts, top, taker = await asyncio.gather( + settle(funding_part(symbol)), + settle(open_interest_part(symbol, args.period, args.limit)), + settle(ratio_part("global_long_short", symbol, args.period, args.limit, "longShortRatio")), + settle(ratio_part("top_long_short_position", symbol, args.period, args.limit, "longShortRatio")), + settle(taker_part(symbol, args.period, args.limit)), + ) + return ToolOutcome(dump({ + "symbol": symbol, "period": args.period, "window": f"最近 {args.limit} 个 {args.period}", + "funding_rate": funding, + "open_interest": oi, + "long_short_account_ratio": accounts, + "top_trader_position_ratio": top, + "taker_buy_sell_ratio": taker, + })) + + # ---------- Skills 与 CONTEXT.md ---------- class NoArgs(BaseModel): @@ -286,6 +395,10 @@ TOOLS: dict[str, Tool] = { "search_symbols": Tool(SearchSymbols, search_symbols, "在币安 U 本位永续（含美股/韩股等 TradFi 永续）中查找交易对代码。"), "get_klines": Tool(GetKlines, get_klines, "取 U 本位永续 K 线和确定性统计（涨跌幅、高低点、MA20/60、ATR14、波段高低点、量比），用于判断支撑阻力和止损位。"), "get_ticker": Tool(GetTicker, get_ticker, "取 U 本位永续 24h 行情和最后成交时间。"), + "get_derivatives": Tool(GetDerivatives, get_derivatives, ( + "取 U 本位永续的衍生品数据，用于判断多空拥挤程度：资金费率（下次预估、近 30 次结算，单位 %）、" + "溢价、持仓量（窗口变化及同期价格变化）、全市场账户多空比、大户持仓多空比、主动买卖比（>1 表示主动买入多）。" + )), "list_skills": Tool(NoArgs, list_skills, "列出全部 skills 的路径、优先级和版本状态。"), "read_skill": Tool(ReadSkill, read_skill, "读取 skill 的某个版本（默认最新，可能是草稿）。"), "propose_skill_edit": Tool(ProposeSkillEdit, propose_skill_edit, "修改已有 skill：保存为草稿版本，用户发布后才生效。content 必须是完整内容。"), diff --git a/backend/src/alphapulse/api/settings.py b/backend/src/alphapulse/api/settings.py index 8d81270..836d558 100644 --- a/backend/src/alphapulse/api/settings.py +++ b/backend/src/alphapulse/api/settings.py @@ -5,7 +5,7 @@ from datetime import datetime from typing import Annotated, Literal from fastapi import APIRouter, Depends, HTTPException, status -from pydantic import BaseModel, Field, HttpUrl +from pydantic import BaseModel, Field, HttpUrl, StringConstraints from sqlalchemy import select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -19,6 +19,9 @@ from alphapulse.models import AiProvider router = APIRouter(prefix="/api/settings", tags=["settings"], dependencies=[Depends(require_admin)]) Protocol = Literal["openai", "anthropic"] +# 从页面粘贴时容易带上首尾空格，模型名带空格会被上游当成另一个模型（404） +Name = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=64)] +Model = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=128)] class ProviderOut(BaseModel): @@ -36,11 +39,11 @@ class ProviderOut(BaseModel): class ProviderIn(BaseModel): - name: Annotated[str, Field(min_length=1, max_length=64)] + name: Name protocol: Protocol base_url: HttpUrl api_key: Annotated[str, Field(min_length=1)] - model: Annotated[str, Field(min_length=1, max_length=128)] + model: Model max_tokens: Annotated[int, Field(ge=256, le=128000)] = 8192 context_window: Annotated[int, Field(ge=4096, le=2000000)] = 200000 reasoning_effort: llm.Effort | None = None @@ -48,12 +51,12 @@ class ProviderIn(BaseModel): class ProviderPatch(BaseModel): - name: Annotated[str | None, Field(min_length=1, max_length=64)] = None + name: Name | None = None protocol: Protocol | None = None base_url: HttpUrl | None = None # 为空表示不修改 api_key: str | None = None - model: Annotated[str | None, Field(min_length=1, max_length=128)] = None + model: Model | None = None max_tokens: Annotated[int | None, Field(ge=256, le=128000)] = None context_window: Annotated[int | None, Field(ge=4096, le=2000000)] = None # 显式传 null 表示不传思考强度 diff --git a/backend/src/alphapulse/market/binance.py b/backend/src/alphapulse/market/binance.py index a4f4419..818d77a 100644 --- a/backend/src/alphapulse/market/binance.py +++ b/backend/src/alphapulse/market/binance.py @@ -13,7 +13,13 @@ REST = { "spot": {"base": "https://api.binance.com", "klines": "/api/v3/klines", "exchange_info": "/api/v3/exchangeInfo", "ticker": "/api/v3/ticker/24hr", "weight_limit": 6000, "kline_limit": 1000}, "um": {"base": "https://fapi.binance.com", "klines": "/fapi/v1/klines", "exchange_info": "/fapi/v1/exchangeInfo", - "ticker": "/fapi/v1/ticker/24hr", "weight_limit": 2400, "kline_limit": 1500}, + "ticker": "/fapi/v1/ticker/24hr", "weight_limit": 2400, "kline_limit": 1500, + # 衍生品数据：/futures/data 下的统计只保留最近 30 天 + "premium_index": "/fapi/v1/premiumIndex", "funding_rate": "/fapi/v1/fundingRate", + "open_interest_hist": "/futures/data/openInterestHist", + "global_long_short": "/futures/data/globalLongShortAccountRatio", + "top_long_short_position": "/futures/data/topLongShortPositionRatio", + "taker_buy_sell": "/futures/data/takerlongshortRatio"}, } # 合约旧路由 fstream.binance.com/stream 能连上但不推送数据，K 线/行情流在 /market 路由下 WS = {"spot": "wss://stream.binance.com:9443/stream", "um": "wss://fstream.binance.com/market/stream"} diff --git a/web/src/pages/chat/components/MessageList.vue b/web/src/pages/chat/components/MessageList.vue index 5f7c572..e949cac 100644 --- a/web/src/pages/chat/components/MessageList.vue +++ b/web/src/pages/chat/components/MessageList.vue @@ -30,6 +30,7 @@ const TOOL_LABELS: Record = { search_symbols: "查找交易对", get_klines: "获取 K 线", get_ticker: "获取 24h 行情", + get_derivatives: "获取衍生品数据", list_skills: "列出 skills", read_skill: "读取 skill", propose_skill_edit: "提交 skill 草稿",