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.
This commit is contained in:
eezd committed 2026-10-14 18:16:54 +08:00
1 parent b549f3f441
commit 61486d0c5b
6 files changed
+152 -14

No files matched your search

+21 -6
View File
@@ -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}")
+1 -1
View File
@@ -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、波段高低点,并说明失效条件。
- 数据不足或互相矛盾时如实说明缺什么,不编造。
+114 -1
View File
@@ -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 必须是完整内容。"),
+8 -5
View File
@@ -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 表示不传思考强度
+7 -1
View File
@@ -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"}
@@ -30,6 +30,7 @@ const TOOL_LABELS: Record<string, string> = {
search_symbols: "查找交易对",
get_klines: "获取 K 线",
get_ticker: "获取 24h 行情",
get_derivatives: "获取衍生品数据",
list_skills: "列出 skills",
read_skill: "读取 skill",
propose_skill_edit: "提交 skill 草稿",