diff --git a/backend/pyproject.toml b/backend/pyproject.toml index a8b707f..2b75246 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -27,3 +27,8 @@ dependencies = [ [build-system] requires = ["uv_build>=0.12.23,<0.13.0"] build-backend = "uv_build" + +[dependency-groups] +dev = [ + "pytest>=9.1.1", +] diff --git a/backend/src/alphapulse/ai/llm.py b/backend/src/alphapulse/ai/llm.py index 55af8ae..66b38fe 100644 --- a/backend/src/alphapulse/ai/llm.py +++ b/backend/src/alphapulse/ai/llm.py @@ -31,6 +31,8 @@ MAX_RETRIES = 2 # 这些都由这里再重试，且只在还没有输出内容时重试 STREAM_RETRIES = 3 STREAM_RETRY_DELAY_SECONDS = 2 +# 输出被截断时回给模型的工具结果：参数不完整，不能执行 +TRUNCATED_RESULT = json.dumps({"error": "输出达到最大输出上限被截断，工具参数不完整，未执行。请精简内容后重新调用"}, ensure_ascii=False) logger = logging.getLogger("alphapulse.ai.llm") diff --git a/backend/src/alphapulse/ai/prompt.py b/backend/src/alphapulse/ai/prompt.py index 19a2e3d..825b8eb 100644 --- a/backend/src/alphapulse/ai/prompt.py +++ b/backend/src/alphapulse/ai/prompt.py @@ -50,14 +50,18 @@ async def data_overview(session: AsyncSession) -> str: return "\n".join(lines) -async def build_system_prompt(session: AsyncSession) -> str: - parts = [BASE_INSTRUCTIONS, "## 数据概况", await data_overview(session)] +async def skills_section(session: AsyncSession) -> str: skills = await published_skills(session) - parts.append("## Skills（按优先级从高到低）") + parts = ["## Skills（按优先级从高到低）"] if skills: parts += [f"### 【优先级 {info.node.priority}】{info.path}\n\n{content}" for info, content in skills] else: parts.append("（暂无已发布的 skill）") + return "\n\n".join(parts) + + +async def build_system_prompt(session: AsyncSession) -> str: + parts = [BASE_INSTRUCTIONS, "## 数据概况", await data_overview(session), await skills_section(session)] context = await current_context(session) parts += ["## CONTEXT.md", context.content if context and context.content.strip() else "（空）"] return "\n\n".join(parts) diff --git a/backend/src/alphapulse/api/chat.py b/backend/src/alphapulse/api/chat.py index 5089f4f..ba9fcb4 100644 --- a/backend/src/alphapulse/api/chat.py +++ b/backend/src/alphapulse/api/chat.py @@ -27,7 +27,6 @@ MAX_TOOL_ROUNDS = 8 # 发给前端展示的工具结果长度 TOOL_PREVIEW_CHARS = 2000 STOPPED_RESULT = json.dumps({"error": "用户已停止生成，工具未执行"}, ensure_ascii=False) -TRUNCATED_RESULT = json.dumps({"error": "输出达到最大输出上限被截断，工具参数不完整，未执行。请精简内容后重新调用"}, ensure_ascii=False) # api 只有一个进程：同一时间只允许一轮生成，Reset 也要等生成结束 generating = asyncio.Lock() @@ -195,7 +194,7 @@ async def send_message(body: SendIn) -> AsyncIterator[ServerSentEvent]: call = pending[0] yield event("tool_call", {"id": call["id"], "name": call["name"], "arguments": call["arguments"]}) # 截断时不执行，把原因告诉模型，让它精简后重试 - outcome = ToolOutcome(TRUNCATED_RESULT) if completion.truncated else await run_tool(call["name"], call["arguments"]) + outcome = ToolOutcome(llm.TRUNCATED_RESULT) if completion.truncated else await run_tool(call["name"], call["arguments"]) result = {"tool_call_id": call["id"], "name": call["name"], "result": outcome.result} await save_messages(ChatMessage(session_id=current.id, role="tool", content=result)) pending.pop(0) diff --git a/backend/src/alphapulse/api/trading.py b/backend/src/alphapulse/api/trading.py index a4dc160..e307936 100644 --- a/backend/src/alphapulse/api/trading.py +++ b/backend/src/alphapulse/api/trading.py @@ -18,8 +18,9 @@ from alphapulse.core.auth import require_admin from alphapulse.core.db import get_session from alphapulse.core.redis import KEY_PREFIX, redis from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingSymbol -from alphapulse.trading.client import ExchangeError, public_get -from alphapulse.trading.engine import DAILY_ANALYSIS_LIMIT, OWN_CLIENT_ID, STATUS_KEY +from alphapulse.trading.client import OWN_CLIENT_ID, ExchangeError, public_get +from alphapulse.trading.plan import DAILY_ANALYSIS_LIMIT +from alphapulse.trading.snapshot import read_snapshot router = APIRouter(prefix="/api/trading", tags=["trading"], dependencies=[Depends(require_admin)]) @@ -82,6 +83,36 @@ class SymbolPatch(BaseModel): angle: Angle +class AnalysisOut(BaseModel): + id: int + symbol: str + angle: str + reasons: list[str] + dry_run: bool + status: str + action: str | None + decision: dict | None + sizing: dict | None + error: str | None + model: str | None + reasoning_effort: str | None + usage: dict | None + started_at: datetime + finished_at: datetime | None + plan_status: str | None + valid_until: datetime | None + first_fill_at: datetime | None + close_deadline: datetime | None + closed_at: datetime | None + close_reason: str | None + result: dict | None + + +class AnalysisDetail(AnalysisOut): + text: str | None + transcript: list | None + + class EventOut(BaseModel): id: int created_at: datetime @@ -97,11 +128,6 @@ async def get_account(session: AsyncSession) -> TradingAccount: return await session.get(TradingAccount, 1) or TradingAccount(id=1, running=False) -async def read_snapshot() -> dict | None: - raw = await redis.get(STATUS_KEY) - return json.loads(raw) if raw else None - - @router.get("/status") async def get_status(session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut: account = await get_account(session) @@ -292,3 +318,39 @@ async def list_events( if before: query = query.where(TradingEvent.id < before) return [EventOut.model_validate(e, from_attributes=True) for e in await session.scalars(query)] + + +@router.post("/symbols/{symbol}/analyze", status_code=status.HTTP_202_ACCEPTED) +async def request_analysis(symbol: str, session: Annotated[AsyncSession, Depends(get_session)]) -> None: + """排队一次手动分析：引擎停止时为试运行，计划只记录、不下单""" + s = await session.get(TradingSymbol, symbol) + if s is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, f"{symbol} 不在交易列表中") + if (await get_account(session)).capital_usdt is None: + raise HTTPException(status.HTTP_400_BAD_REQUEST, "还没有设置交易资金") + running = await session.scalar(select(func.count()).where(TradingAnalysis.symbol == symbol, TradingAnalysis.status == "running")) + if running or s.analysis_requested_at is not None: + raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 已在分析或排队中") + s.analysis_requested_at = datetime.now(LOCAL_TZ) + s.analysis_reasons = ["手动分析"] + await session.commit() + + +@router.get("/analyses") +async def list_analyses( + session: Annotated[AsyncSession, Depends(get_session)], + symbol: str | None = None, + limit: Annotated[int, Query(ge=1, le=50)] = 10, +) -> list[AnalysisOut]: + query = select(TradingAnalysis).order_by(TradingAnalysis.id.desc()).limit(limit) + if symbol: + query = query.where(TradingAnalysis.symbol == symbol) + return [AnalysisOut.model_validate(a, from_attributes=True) for a in await session.scalars(query)] + + +@router.get("/analyses/{analysis_id}") +async def get_analysis(analysis_id: int, session: Annotated[AsyncSession, Depends(get_session)]) -> AnalysisDetail: + analysis = await session.get(TradingAnalysis, analysis_id) + if analysis is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "分析不存在") + return AnalysisDetail.model_validate(analysis, from_attributes=True) diff --git a/backend/src/alphapulse/trading/analyst.py b/backend/src/alphapulse/trading/analyst.py new file mode 100644 index 0000000..c1840ea --- /dev/null +++ b/backend/src/alphapulse/trading/analyst.py @@ -0,0 +1,351 @@ +"""模拟盘分析器：为一个币的一个角度调用模型，产出经程序校验、可自动执行的交易计划。 + +只开放只读工具和 submit_trade_plan；行情预先取好放进用户消息，减少工具往返。 +""" + +import asyncio +import json +import logging +import time +from dataclasses import dataclass +from datetime import UTC, datetime +from decimal import Decimal + +from pydantic import ValidationError +from sqlalchemy import select + +from alphapulse.ai import llm +from alphapulse.ai.prompt import data_overview, estimate_tokens, skills_section +from alphapulse.ai.tools import TOOLS, dump, local, run_tool +from alphapulse.core.db import SessionLocal +from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingSymbol +from alphapulse.trading import plan as rules +from alphapulse.trading.client import public_get +from alphapulse.trading.plan import PlanContext, PlanError, SymbolRules, TradePlan, size_plan +from alphapulse.trading.snapshot import read_snapshot + +logger = logging.getLogger("alphapulse.trading.analyst") + +READ_TOOLS = ("search_news", "get_news", "search_symbols", "get_klines", "get_ticker", "get_derivatives") +# 最后一轮只开放 submit_trade_plan，逼模型给出结论 +MAX_ROUNDS = 6 +RECENT_DECISIONS = 2 +EXCHANGE_INFO_SECONDS = 3600 +ANGLE_LABELS = {"short": "短线（24 小时内平仓）", "long": "长线（12–48 小时内平仓）"} +DIRECTION_LABELS = {"long": "做多", "short": "做空"} +CONFIDENCE_LABELS = {"high": "高", "medium": "中", "low": "低"} +ACTION_LABELS = {"plan": "新计划", "keep": "维持当前计划", "exit": "结束当前计划", "wait": "观望"} +# 与 README 第 1 节一致：定方向和找区的周期、衍生品参数 +KLINE_INTERVALS = {"short": ("4h", "1h", "15m"), "long": ("1d", "4h", "1h")} +DERIVATIVES_PARAMS = {"short": ("1h", 24), "long": ("4h", 18)} + + +def _pct(value: Decimal) -> str: + return f"{(value * 100).normalize()}%" + + +INSTRUCTIONS = f"""你是 AlphaPulse 的模拟盘交易分析器：为一个币种的一个角度给出由程序自动执行的交易计划。 + +## 任务 +- 只分析用户消息指定的币种和角度（短线或长线），不给另一个角度。 +- 下方 Skills 是交易规范。以下几处按本节执行，与 Skills 冲突时以本节为准： + - main/README 第 1 节「每次同时给出两个角度」不适用，只给指定角度。 + - main/README 第 8 节的输出表格不适用：用简短中文写出结论、用到的支撑区与压力区、依据（新闻、结构、衍生品）和风险；计划的数字只通过 submit_trade_plan 提交。 + - main/README 第 9 节不适用：没有 CONTEXT.md，不要写。之前的结论见用户消息里的「最近决策」，按 README 第 6 节做实时复核，不直接沿用旧点位。 + - main/README 5.4 的条件计划（突破回踩、事件落地后再入场等）不能直接挂单，改写成 reevaluate 条件：价格到达后程序会重新请你分析。 + - 数量由程序按交易资金计算，你只给价格、档位比例和置信度。 +- 用户消息里已预先取好实盘行情（ticker、K 线、衍生品），不要用相同参数重复调用；需要其他周期或其他币时再调用。 +- 新闻没有预取，必须按 main/README 第 3 节第 3 步用 search_news 检索相关新闻，并查清持仓时限内的重大数据及公布时间。 +- 最后必须调用一次 submit_trade_plan。校验不通过会返回原因，按原因修正后重新提交。 + +## 程序如何执行计划 +- 入场：每档一张限价单；到 entry_valid_hours 仍未成交的档位撤销。买单价高于现价（卖单价低于现价）会立即成交。 +- 止损：与入场单同时在交易所挂只减仓止损单（按标记价触发），首档成交即受保护。 +- 止盈：短线止盈 1 平一半、止盈 2 平剩余；长线 3 个止盈依次平 50% / 30% / 20%。止盈 1 成交后撤销剩余入场单，止损移到保本价。 +- invalidation（失效条件）：短线用 1m 或 15m 收盘，触发后程序立即市价平仓；长线只能用 4h 收盘，触发后程序请你复核。 +- reevaluate：任一条件满足时程序重新请你分析（最多 {rules.MAX_REEVALUATE} 个）。用来表达另一侧方向、条件计划和需要复核的位置。 +- max_hold_hours 从首档成交起算，到时程序市价平仓。 + +## 程序的硬性校验 +- 短线 1 档入场、2 个止盈；长线 1–3 档入场、3 个止盈；入场比例之和为 1。 +- 做多：止损 < 失效价 < 所有入场价 < 止盈 1 < 止盈 2 < 止盈 3，且止损低于现价、止盈 1 高于现价；做空反过来。 +- 盈亏比 ≥ {rules.MIN_REWARD_RISK}：短线按止盈 1，长线按 50/30/20 加权（全部档位成交时）。 +- entry_valid_hours：短线 ≤ {rules.ENTRY_VALID_HOURS['short']}，长线 ≤ {rules.ENTRY_VALID_HOURS['long']}；max_hold_hours：短线 ≤ {rules.MAX_HOLD_HOURS['short']}，长线 ≤ {rules.MAX_HOLD_HOURS['long']}。 +- 每档入场的名义价值不低于交易所最小下单金额，不足时合并或删档。 +- 单笔风险为交易资金的 {_pct(rules.RISK_PCT['short'])}（短线）或 {_pct(rules.RISK_PCT['long'])}（长线），置信度低时减半；名义仓位合计不超过 {rules.MAX_LEVERAGE} 倍交易资金。这两条由程序计算数量来保证。 +- 有持仓时新计划的方向必须与持仓一致；要反手先用 exit。 +""" + +SUBMIT_SPEC = llm.ToolSpec( + "submit_trade_plan", + "提交本次分析的结论和计划。程序会校验并计算仓位；不通过时返回原因，修正后重新提交。", + TradePlan.model_json_schema(), +) + + +class AnalysisError(Exception): + """分析无法完成；message 写入分析记录和时间线""" + + +@dataclass +class Prepared: + provider: AiProvider + effort: str | None + dry_run: bool + angle: str + system: str + user_text: str + context: PlanContext + + +_exchange_info: tuple[float, dict[str, SymbolRules]] | None = None + + +async def symbol_rules(symbol: str) -> SymbolRules: + """模拟盘的价格精度、数量精度、最小下单金额，按小时缓存""" + global _exchange_info + if _exchange_info is None or time.monotonic() - _exchange_info[0] > EXCHANGE_INFO_SECONDS: + info = await public_get("/fapi/v1/exchangeInfo") + parsed = {} + for s in info["symbols"]: + filters = {f["filterType"]: f for f in s["filters"]} + parsed[s["symbol"]] = SymbolRules( + tick_size=Decimal(filters["PRICE_FILTER"]["tickSize"]).normalize(), + step_size=Decimal(filters["LOT_SIZE"]["stepSize"]).normalize(), + min_notional=Decimal(filters["MIN_NOTIONAL"]["notional"]), + ) + _exchange_info = (time.monotonic(), parsed) + if symbol not in _exchange_info[1]: + raise AnalysisError(f"模拟盘没有交易对 {symbol}") + return _exchange_info[1][symbol] + + +async def mark_price(symbol: str) -> Decimal: + return Decimal((await public_get("/fapi/v1/premiumIndex", {"symbol": symbol}))["markPrice"]) + + +def plan_brief(decision: dict, sizing: dict | None) -> dict: + """给模型看的计划摘要：只留价格和状态，不带长段文字""" + brief = {k: decision.get(k) for k in ("action", "summary", "direction", "confidence", "stop", "entry_valid_hours", "max_hold_hours")} + brief["entries"] = [{"price": e["price"], "ratio": e["ratio"]} for e in decision.get("entries") or []] + brief["take_profits"] = [t["price"] for t in decision.get("take_profits") or []] + brief["invalidation"] = decision.get("invalidation") + brief["reevaluate"] = [{k: c[k] for k in ("direction", "price", "timeframe")} for c in decision.get("reevaluate") or []] + if sizing: + brief["sizing"] = {k: sizing.get(k) for k in ("quantity", "notional", "risk_at_stop", "reward_risk")} + return brief + + +async def prefetch(symbol: str, angle: str) -> str: + """预取行情：工具出错时把错误原样放进消息，不中断分析""" + period, limit = DERIVATIVES_PARAMS[angle] + calls = [("get_ticker", {"symbol": symbol})] + calls += [("get_klines", {"symbol": symbol, "interval": i, "limit": 200}) for i in KLINE_INTERVALS[angle]] + calls.append(("get_derivatives", {"symbol": symbol, "period": period, "limit": limit})) + outcomes = await asyncio.gather(*(run_tool(name, args) for name, args in calls)) + return "\n\n".join(f"### {name} {json.dumps(args, ensure_ascii=False)}\n{o.result}" for (name, args), o in zip(calls, outcomes, strict=True)) + + +async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared: + async with SessionLocal() as session: + account = await session.get(TradingAccount, 1) + target = await session.get(TradingSymbol, symbol) + if target is None: + raise AnalysisError(f"{symbol} 已不在交易列表中") + if account is None or account.capital_usdt is None: + raise AnalysisError("还没有设置交易资金") + provider = await session.get(AiProvider, account.provider_id) if account.provider_id else await session.scalar( + select(AiProvider).where(AiProvider.is_default) + ) + if provider is None: + raise AnalysisError("没有可用的模型配置，请先到「环境变量」页面添加") + session.expunge(provider) + symbols = list((await session.scalars(select(TradingSymbol.symbol))).all()) + recent = (await session.scalars( + select(TradingAnalysis) + .where(TradingAnalysis.symbol == symbol, TradingAnalysis.status == "done", TradingAnalysis.id != analysis_id) + .order_by(TradingAnalysis.id.desc()).limit(RECENT_DECISIONS) + )).all() + current = await session.scalar( + select(TradingAnalysis) + .where(TradingAnalysis.symbol == symbol, TradingAnalysis.plan_status.in_(("pending", "open")), TradingAnalysis.dry_run.is_(False)) + .order_by(TradingAnalysis.id.desc()).limit(1) + ) + system = "\n\n".join([INSTRUCTIONS, "## 数据概况", await data_overview(session), await skills_section(session)]) + angle, dry_run = target.angle, not account.running + effort = account.reasoning_effort or provider.reasoning_effort + + snapshot = await read_snapshot() + if not snapshot or not snapshot["account"]: + raise AnalysisError("取不到模拟盘账户快照（引擎未同步到币安），无法计算仓位") + equity = Decimal(snapshot["account"]["margin_balance"]) + capital = min(account.capital_usdt, equity) + position = next((p for p in snapshot["positions"] if p["symbol"] == symbol), None) + other_notional = sum((Decimal(p["notional"]) for p in snapshot["positions"] if p["symbol"] != symbol and p["symbol"] in symbols), Decimal(0)) + symbol_rules_ = await symbol_rules(symbol) + mark = await mark_price(symbol) + context = PlanContext( + angle=angle, mark_price=mark, capital=capital, other_notional=other_notional, rules=symbol_rules_, + has_plan=current is not None, position_side=("long" if position["side"] == "LONG" else "short") if position else None, + ) + + orders = [o for o in snapshot["orders"] + snapshot["algo_orders"] if o["symbol"] == symbol] + risk = capital * rules.RISK_PCT[angle] + lines = [ + "## 任务", + f"- 币种：{symbol}（币安 U 本位永续）", + f"- 角度：{ANGLE_LABELS[angle]}", + f"- 触发原因：{'；'.join(reasons)}", + f"- 运行模式：{'试运行（引擎已停止，计划只记录、不下单）' if dry_run else '交易中（计划会在模拟盘执行）'}", + "", + "## 账户（程序据此计算数量）", + f"- 交易资金：{capital:.2f} USDT（设定 {account.capital_usdt} USDT，账户权益 {equity:.2f} USDT，取较小者）", + f"- 本次单笔风险金额：{risk:.2f} USDT（{_pct(rules.RISK_PCT[angle])}；置信度低时 {risk * rules.LOW_CONFIDENCE_FACTOR:.2f}）", + f"- 可用名义仓位：{capital * rules.MAX_LEVERAGE - other_notional:.2f} USDT（{rules.MAX_LEVERAGE} 倍交易资金减去交易列表其他币的持仓 {other_notional:.2f}）", + f"- 模拟盘交易规则：价格精度 {symbol_rules_.tick_size}，数量精度 {symbol_rules_.step_size}，最小下单金额 {symbol_rules_.min_notional} USDT", + f"- 模拟盘标记价：{mark}（校验以此为准；行情数据来自实盘，两者通常相差不到 0.1%）", + "", + "## 当前持仓与挂单（模拟盘）", + f"- 持仓：{dump(position) if position else '无'}", + f"- 挂单：{dump(orders) if orders else '无'}", + "", + "## 当前计划", + dump({"id": current.id, "plan_status": current.plan_status, **plan_brief(current.decision, current.sizing)}) if current else "无", + "", + "## 最近决策（新的在前）", + *([f"- {local(a.started_at)}{'（试运行）' if a.dry_run else ''}：{dump(plan_brief(a.decision, a.sizing))}" for a in recent] or ["无"]), + "", + "## 预取的行情（实盘数据，已取，不要用相同参数重复调用）", + await prefetch(symbol, angle), + ] + return Prepared(provider=provider, effort=effort, dry_run=dry_run, angle=angle, system=system, user_text="\n".join(lines), context=context) + + +async def complete(prepared: Prepared, messages: list[dict], tools: list[llm.ToolSpec]) -> llm.Completion: + async for item in llm.stream(prepared.provider, prepared.system, messages, tools, prepared.effort): + if isinstance(item, llm.Completion): + return item + raise AssertionError("llm.stream 没有产出 Completion") + + +async def add_event(symbol: str, analysis_id: int, kind: str, message: str, level: str = "info", data: dict | None = None) -> None: + async with SessionLocal() as session: + session.add(TradingEvent(symbol=symbol, analysis_id=analysis_id, kind=kind, level=level, message=message, data=data)) + await session.commit() + + +def describe(plan: TradePlan, sizing: dict, dry_run: bool) -> str: + prefix = "试运行" if dry_run else "" + if plan.action != "plan": + return f"{prefix}{ACTION_LABELS[plan.action]}：{plan.summary}" + entries = "、".join(f"{e['price']}×{e['quantity']}" for e in sizing["entries"]) + targets = " / ".join(t["price"] for t in sizing["take_profits"]) + return ( + f"{prefix}计划：{DIRECTION_LABELS[plan.direction]}，置信度{CONFIDENCE_LABELS[plan.confidence]}；入场 {entries}，" + f"止损 {sizing['stop']}，止盈 {targets}；止损亏损 {sizing['risk_at_stop']} USDT（交易资金的 {sizing['risk_pct']}%），" + f"盈亏比 {sizing['reward_risk']}（仅首档 {sizing['reward_risk_first_only']}）。{plan.summary}" + ) + + +async def submit(prepared: Prepared, symbol: str, analysis_id: int, arguments: dict) -> tuple[str, tuple[TradePlan, dict] | None]: + try: + plan = TradePlan.model_validate(arguments) + except ValidationError as e: + return dump({"error": "参数无效", "details": e.errors(include_url=False)}), None + # 分析可能持续几分钟，按提交时的标记价校验 + prepared.context.mark_price = await mark_price(symbol) + try: + sizing = size_plan(plan, prepared.context) + except PlanError as e: + await add_event(symbol, analysis_id, "plan_rejected", f"计划未通过校验，已退回模型修正：{'；'.join(e.errors)}", "warn") + return dump({"error": "计划未通过校验", "details": e.errors, "mark_price": str(prepared.context.mark_price)}), None + return dump({"status": "已接受", "sizing": sizing}), (plan, sizing) + + +async def run_analysis(symbol: str, reasons: list[str]) -> None: + """一次完整的分析；结果和失败原因都写进 trading_analyses 和时间线，不向外抛出""" + async with SessionLocal() as session: + account = await session.get(TradingAccount, 1) + target = await session.get(TradingSymbol, symbol) + analysis = TradingAnalysis( + symbol=symbol, angle=target.angle if target else "long", reasons=reasons, + dry_run=not (account and account.running), status="running", + ) + session.add(analysis) + await session.commit() + analysis_id, dry_run = analysis.id, analysis.dry_run + await add_event(symbol, analysis_id, "analysis_started", f"开始{'试运行' if dry_run else ''}分析：{'；'.join(reasons)}") + + messages: list[dict] = [] + texts: list[str] = [] + usage = {"input_tokens": 0, "output_tokens": 0} + accepted: tuple[TradePlan, dict] | None = None + prepared: Prepared | None = None + try: + prepared = await prepare(symbol, reasons, analysis_id) + estimated = estimate_tokens(prepared.system) + estimate_tokens(prepared.user_text) + if estimated + prepared.provider.max_tokens > prepared.provider.context_window: + raise AnalysisError(f"提示词约 {estimated} token，加上最大输出超过模型上下文上限 {prepared.provider.context_window}") + messages.append({"role": "user", "text": prepared.user_text}) + specs = [llm.ToolSpec(name, TOOLS[name].description, TOOLS[name].args_model.model_json_schema()) for name in READ_TOOLS] + for round_no in range(MAX_ROUNDS): + last = round_no == MAX_ROUNDS - 1 + completion = await complete(prepared, messages, [SUBMIT_SPEC] if last else [*specs, SUBMIT_SPEC]) + for key in usage: + usage[key] += completion.usage.get(key, 0) + if completion.text.strip(): + texts.append(completion.text.strip()) + assistant = {"role": "assistant", "text": completion.text, "tool_calls": completion.tool_calls} + if completion.thinking_blocks: + assistant["thinking_blocks"] = completion.thinking_blocks + messages.append(assistant) + if not completion.tool_calls: + if completion.truncated: + raise AnalysisError(f"模型输出达到最大输出上限（{prepared.provider.max_tokens} token）被截断") + messages.append({"role": "user", "text": "请调用 submit_trade_plan 提交结论。"}) + continue + for call in completion.tool_calls: + if completion.truncated: + result = llm.TRUNCATED_RESULT + elif call["name"] == "submit_trade_plan": + result, outcome = await submit(prepared, symbol, analysis_id, call["arguments"]) + accepted = accepted or outcome + elif call["name"] in READ_TOOLS: + result = (await run_tool(call["name"], call["arguments"])).result + else: + result = dump({"error": f"分析时不能使用工具 {call['name']}"}) + messages.append({"role": "tool", "tool_call_id": call["id"], "name": call["name"], "result": result}) + if accepted: + break + if accepted is None: + raise AnalysisError(f"模型在 {MAX_ROUNDS} 轮内没有提交通过校验的结论") + except (AnalysisError, llm.LlmError, PlanError) as e: + error = str(e) + except Exception as e: + logger.exception("%s 分析出错", symbol) + error = f"程序出错：{e!r}" + else: + error = None + + async with SessionLocal() as session: + analysis = await session.get(TradingAnalysis, analysis_id) + analysis.status = "failed" if error else "done" + analysis.error = error + analysis.text = "\n\n".join(texts) or None + # thinking 块只在同一轮工具循环内需要回传，存档时去掉，减小体积 + analysis.transcript = [{k: v for k, v in m.items() if k != "thinking_blocks"} for m in messages] or None + analysis.usage = usage + analysis.finished_at = datetime.now(UTC) + if prepared: + analysis.provider_id, analysis.model, analysis.reasoning_effort = prepared.provider.id, prepared.provider.model, prepared.effort + if accepted: + plan, sizing = accepted + analysis.action = plan.action + analysis.decision = plan.model_dump(mode="json") + analysis.sizing = sizing or None + await session.commit() + if error: + await add_event(symbol, analysis_id, "analysis_failed", f"分析失败：{error}", "error") + else: + plan, sizing = accepted + await add_event(symbol, analysis_id, "analysis_done", describe(plan, sizing, dry_run), data={"action": plan.action}) diff --git a/backend/src/alphapulse/trading/client.py b/backend/src/alphapulse/trading/client.py index 715783d..3b48589 100644 --- a/backend/src/alphapulse/trading/client.py +++ b/backend/src/alphapulse/trading/client.py @@ -5,6 +5,7 @@ import hashlib import hmac +import re import time from typing import Any from urllib.parse import urlencode @@ -14,6 +15,8 @@ import httpx BASE_URL = "https://demo-fapi.binance.com" RECV_WINDOW_MS = 5000 TIMEOUT_SECONDS = 10 +# 本程序下的单 clientOrderId / clientAlgoId 为 ap{流水 id}；其他（含用户手动下的）不动 +OWN_CLIENT_ID = re.compile(r"^ap\d+$") class ExchangeError(Exception): diff --git a/backend/src/alphapulse/trading/engine.py b/backend/src/alphapulse/trading/engine.py index f0b2e3f..bbdf985 100644 --- a/backend/src/alphapulse/trading/engine.py +++ b/backend/src/alphapulse/trading/engine.py @@ -1,156 +1,75 @@ -"""模拟盘交易进程：定时把账户、持仓、挂单快照写入 Redis，供页面读取。 +"""模拟盘交易进程：同步账户快照、调度分析。 用法：python -m alphapulse.trading.engine """ import asyncio -import json import logging -import re -import time from datetime import UTC, datetime -from sqlalchemy import select +from sqlalchemy import select, update -from alphapulse.core.crypto import decrypt from alphapulse.core.db import SessionLocal -from alphapulse.core.redis import KEY_PREFIX, redis -from alphapulse.models import TradingAccount, TradingSymbol -from alphapulse.trading.client import DemoClient, ExchangeError +from alphapulse.models import TradingAnalysis, TradingSymbol +from alphapulse.trading.analyst import run_analysis +from alphapulse.trading.snapshot import snapshot_loop logger = logging.getLogger("alphapulse.trading.engine") -# 进程挂掉后快照自动过期，页面据此显示「引擎离线」 -STATUS_KEY = f"{KEY_PREFIX}trader:status" -STATUS_TTL_SECONDS = 15 -SNAPSHOT_SECONDS = 3 -# 杠杆和保证金模式很少变，按币缓存 -SYMBOL_CONFIG_SECONDS = 60 -# 所有币合计每天（北京时间）最多自动分析的次数 -DAILY_ANALYSIS_LIMIT = 60 -# 本程序下的单 clientOrderId / clientAlgoId 为 ap{流水 id}；其他（含用户手动下的）不动 -OWN_CLIENT_ID = re.compile(r"^ap\d+$") +ANALYSIS_POLL_SECONDS = 5 +# 模型调用耗时长，并发过多会撞上游限流 +MAX_CONCURRENT_ANALYSES = 2 -class Snapshot: - def __init__(self) -> None: - self.client: DemoClient | None = None - self.key_ciphertext: bytes | None = None - # 最近一次成功同步的数据；同步失败时保留并标注错误，避免页面闪空 - self.data: dict | None = None - self.synced_at: datetime | None = None - self.symbol_config: dict[str, tuple[float, dict]] = {} - - async def ensure_client(self, account: TradingAccount | None) -> DemoClient | None: - ciphertext = account.api_key_encrypted if account else None - if ciphertext != self.key_ciphertext: - # key 被修改或删除：旧连接作废，旧数据属于另一个账户，一并清掉 - if self.client is not None: - await self.client.http.aclose() - self.client = None - self.data = self.synced_at = None - self.symbol_config.clear() - self.key_ciphertext = ciphertext - if account is not None and ciphertext is not None: - self.client = DemoClient(decrypt(ciphertext), decrypt(account.api_secret_encrypted)) - return self.client - - async def leverage_and_margin(self, client: DemoClient, symbol: str) -> dict: - cached = self.symbol_config.get(symbol) - if cached is None or time.monotonic() - cached[0] > SYMBOL_CONFIG_SECONDS: - config = (await client.request("GET", "/fapi/v1/symbolConfig", {"symbol": symbol}))[0] - cached = (time.monotonic(), {"leverage": config["leverage"], "margin_type": config["marginType"]}) - self.symbol_config[symbol] = cached - return cached[1] - - async def fetch(self, client: DemoClient, symbols: list[str]) -> dict: - account = await client.request("GET", "/fapi/v3/account") - positions = [p for p in await client.request("GET", "/fapi/v3/positionRisk") if float(p["positionAmt"]) != 0] - # 挂单只查交易列表里的币和有持仓的币：不带 symbol 查全部的权重是 40，3 秒一次太贵 - watched = sorted(set(symbols) | {p["symbol"] for p in positions}) - orders, algo_orders = [], [] - for symbol in watched: - orders += await client.request("GET", "/fapi/v1/openOrders", {"symbol": symbol}) - algo_orders += await client.request("GET", "/fapi/v1/openAlgoOrders", {"symbol": symbol}) - return { - "account": { - "wallet_balance": account["totalWalletBalance"], - "margin_balance": account["totalMarginBalance"], - "available_balance": account["availableBalance"], - "unrealized_pnl": account["totalUnrealizedProfit"], - }, - "positions": [ - { - "symbol": p["symbol"], - "side": "LONG" if float(p["positionAmt"]) > 0 else "SHORT", - "amount": p["positionAmt"].lstrip("-"), - "entry_price": p["entryPrice"], - "break_even_price": p["breakEvenPrice"], - "mark_price": p["markPrice"], - "unrealized_pnl": p["unRealizedProfit"], - "liquidation_price": p["liquidationPrice"], - "notional": p["notional"].lstrip("-"), - "margin": p["isolatedMargin"] if float(p["isolatedMargin"]) else p["initialMargin"], - **await self.leverage_and_margin(client, p["symbol"]), - } - for p in positions - ], - "orders": [ - { - "symbol": o["symbol"], "client_id": o["clientOrderId"], "side": o["side"], "type": o["type"], - "price": o["price"], "quantity": o["origQty"], "filled": o["executedQty"], "reduce_only": o["reduceOnly"], - "time": o["time"], - } - for o in orders - ], - "algo_orders": [ - { - "symbol": o["symbol"], "client_id": o["clientAlgoId"], "side": o["side"], "type": o["orderType"], - "trigger_price": o["triggerPrice"], "quantity": o["quantity"], "close_position": o["closePosition"], - "reduce_only": o["reduceOnly"], "time": o["createTime"], - } - for o in algo_orders - ], - } - - async def publish(self) -> None: - async with SessionLocal() as session: - account = await session.get(TradingAccount, 1) - symbols = list((await session.scalars(select(TradingSymbol.symbol))).all()) - error = None - client = await self.ensure_client(account) - if client is None: - error = "还没有填写模拟盘 key" - else: - try: - self.data = await self.fetch(client, symbols) - self.synced_at = datetime.now(UTC) - except ExchangeError as e: - logger.warning("同步模拟盘失败：%s", e) - error = str(e) - status = { - "updated_at": datetime.now(UTC).isoformat(), - "synced_at": self.synced_at.isoformat() if self.synced_at else None, - "error": error, - **(self.data or {"account": None, "positions": [], "orders": [], "algo_orders": []}), - } - await redis.set(STATUS_KEY, json.dumps(status), ex=STATUS_TTL_SECONDS) +async def fail_interrupted() -> None: + """上次进程退出时还在进行的分析不会再有结果""" + async with SessionLocal() as session: + await session.execute( + update(TradingAnalysis).where(TradingAnalysis.status == "running") + .values(status="failed", error="引擎重启，分析中断", finished_at=datetime.now(UTC)) + ) + await session.commit() -async def snapshot_loop() -> None: - snapshot = Snapshot() +async def claim_requests(limit: int, busy: set[str]) -> list[tuple[str, list[str]]]: + """取出到期的分析请求并清空，同一请求里的多个触发原因合并为一次分析""" + async with SessionLocal() as session: + symbols = (await session.scalars( + select(TradingSymbol) + .where(TradingSymbol.analysis_requested_at <= datetime.now(UTC), TradingSymbol.symbol.not_in(busy)) + .order_by(TradingSymbol.analysis_requested_at).limit(limit) + )).all() + claimed = [(s.symbol, list(s.analysis_reasons)) for s in symbols] + for s in symbols: + s.analysis_requested_at = None + s.analysis_reasons = [] + await session.commit() + return claimed + + +async def analysis_loop() -> None: + await fail_interrupted() + tasks: dict[str, asyncio.Task] = {} while True: try: - await snapshot.publish() + for symbol, task in list(tasks.items()): + if task.done(): + del tasks[symbol] + if not task.cancelled() and task.exception(): + logger.error("%s 分析任务异常退出", symbol, exc_info=task.exception()) + free = MAX_CONCURRENT_ANALYSES - len(tasks) + if free > 0: + for symbol, reasons in await claim_requests(free, set(tasks)): + tasks[symbol] = asyncio.create_task(run_analysis(symbol, reasons)) except Exception: - logger.exception("发布快照失败，稍后重试") - await asyncio.sleep(SNAPSHOT_SECONDS) + logger.exception("调度分析出错，稍后重试") + await asyncio.sleep(ANALYSIS_POLL_SECONDS) async def main() -> None: - # 分析器和执行器在后续里程碑加入同一个 TaskGroup async with asyncio.TaskGroup() as tg: tg.create_task(snapshot_loop()) + tg.create_task(analysis_loop()) if __name__ == "__main__": diff --git a/backend/src/alphapulse/trading/plan.py b/backend/src/alphapulse/trading/plan.py new file mode 100644 index 0000000..feed357 --- /dev/null +++ b/backend/src/alphapulse/trading/plan.py @@ -0,0 +1,285 @@ +"""交易计划的纯计算：模型提交的计划参数、校验、价格取整、仓位计算、止盈拆分。 + +下方常量与 main/README 第 1、2 节的数字重复，并写进了分析器的提示词；改 README 时要同步修改。 +""" + +from dataclasses import dataclass +from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal +from typing import Annotated, Literal + +from pydantic import BaseModel, Field + +Angle = Literal["short", "long"] +Direction = Literal["long", "short"] + +# 所有币合计每天（北京时间）最多自动分析的次数 +DAILY_ANALYSIS_LIMIT = 60 +RISK_PCT = {"short": Decimal("0.01"), "long": Decimal("0.02")} +LOW_CONFIDENCE_FACTOR = Decimal("0.5") +MAX_LEVERAGE = 3 +MIN_REWARD_RISK = Decimal("1.5") +ENTRY_VALID_HOURS = {"short": 4, "long": 24} +MAX_HOLD_HOURS = {"short": 24, "long": 48} +# 入场档数（最少，最多） +ENTRY_COUNT = {"short": (1, 1), "long": (1, 3)} +# 止盈档数由角度决定：短线止盈 1 平一半、止盈 2 平剩余；长线 50/30/20 +TAKE_PROFIT_RATIOS = {"short": (Decimal("0.5"), Decimal("0.5")), "long": (Decimal("0.5"), Decimal("0.3"), Decimal("0.2"))} +# 短线失效立即市价平仓，要用短周期；长线等 4h 收盘后复核 +INVALIDATION_TIMEFRAMES = {"short": ("1m", "15m"), "long": ("4h",)} +MAX_REEVALUATE = 4 +# 开仓、止损都按吃单手续费估算（C1 实测 taker 0.04%） +TAKER_FEE_RATE = Decimal("0.0004") +# 模拟盘 PERCENT_PRICE：买单价不高于标记价 × 1.05，卖单价不低于标记价 × 0.95 +PERCENT_PRICE_LIMIT = Decimal("0.05") +RATIO_TOLERANCE = Decimal("0.001") + +Price = Annotated[Decimal, Field(gt=0)] + + +class Condition(BaseModel): + direction: Annotated[Literal["above", "below"], Field(description="above：价格高于 price 时满足；below：低于时满足")] + price: Price + timeframe: Annotated[ + Literal["mark", "1m", "15m", "1h", "4h"], + Field(description="mark：标记价触及即满足；其他：该周期 K 线收盘价满足才算"), + ] + basis: Annotated[str, Field(description="依据，如对应的支撑区、压力区")] + + +class Entry(BaseModel): + price: Price + ratio: Annotated[Decimal, Field(gt=0, le=1, description="占总仓位的比例，所有档位之和为 1")] + basis: str + + +class TakeProfit(BaseModel): + price: Price + basis: str + + +class TradePlan(BaseModel): + """submit_trade_plan 的参数""" + + action: Annotated[Literal["plan", "keep", "exit", "wait"], Field(description=( + "plan：给出新计划（替换当前计划）；keep：当前计划不变；" + "exit：结束当前计划，撤销未成交的入场单，有持仓则市价平仓；wait：观望，不给计划（仅在没有持仓和当前计划时）" + ))] + summary: Annotated[str, Field(description="一句话结论，含方向和置信度")] + reasons: Annotated[str, Field(description="依据要点：新闻、结构、衍生品，引用新闻写 [新闻 id]")] + risks: Annotated[str, Field(description="持仓时限内的重大事件及时间、与结论相反的信号")] + reevaluate: Annotated[list[Condition], Field(max_length=MAX_REEVALUATE, description=( + "满足任一条件时程序重新请你分析：用来表达另一侧方向、突破回踩等条件计划、需要复核的位置。提交时不能已经满足" + ))] = [] + direction: Direction | None = None + confidence: Literal["high", "medium", "low"] | None = None + entries: Annotated[list[Entry] | None, Field(description="入场限价单，每档一张")] = None + stop: Annotated[Price | None, Field(description="统一止损价")] = None + stop_basis: str | None = None + take_profits: Annotated[list[TakeProfit] | None, Field(description="按价格由近到远排列")] = None + entry_valid_hours: Annotated[Decimal | None, Field(gt=0, description="入场单有效小时数，到期未成交的档位撤销")] = None + max_hold_hours: Annotated[Decimal | None, Field(gt=0, description="首档成交起最长持仓小时数，到时市价平仓")] = None + invalidation: Annotated[Condition | None, Field(description="失效条件：做多为 below，做空为 above")] = None + + +@dataclass +class SymbolRules: + tick_size: Decimal + step_size: Decimal + min_notional: Decimal + + +@dataclass +class PlanContext: + angle: Angle + mark_price: Decimal + # min(设定的交易资金, 账户权益) + capital: Decimal + # 交易列表中其他币的持仓名义价值合计 + other_notional: Decimal + rules: SymbolRules + has_plan: bool + position_side: Direction | None + + +class PlanError(Exception): + """计划不通过校验；errors 逐条返回给模型修正""" + + def __init__(self, errors: list[str]) -> None: + super().__init__("；".join(errors)) + self.errors = errors + + +def round_to_step(value: Decimal, step: Decimal, rounding: str) -> Decimal: + # quantize 统一小数位，否则可能得到 8.05E+4 这类指数形式，转成字符串下单会被拒 + return ((value / step).to_integral_value(rounding) * step).quantize(step) + + +def split_take_profits(quantity: Decimal, ratios: tuple[Decimal, ...], step: Decimal) -> list[Decimal]: + """按比例拆分止盈数量：前几档向下取整，最后一档取余数，保证合计等于 quantity。取整后为 0 的档位由调用方跳过""" + parts = [round_to_step(quantity * r, step, ROUND_FLOOR) for r in ratios[:-1]] + return [*parts, quantity - sum(parts, Decimal(0))] + + +def reward_risk(entry: Decimal, stop: Decimal, targets: list[Decimal], ratios: tuple[Decimal, ...]) -> Decimal: + """综合盈亏比 = Σ(止盈比例 × |止盈价 − 入场价|) ÷ |入场价 − 止损价|，与 README 第 4 节一致（不含手续费）""" + reward = sum((r * abs(t - entry) for t, r in zip(targets, ratios, strict=True)), Decimal(0)) + return reward / abs(entry - stop) + + +def check_action(plan: TradePlan, ctx: PlanContext) -> list[str]: + errors = [] + if plan.action == "keep" and not ctx.has_plan: + errors.append("当前没有计划，不能用 keep") + if plan.action == "exit" and not ctx.has_plan and ctx.position_side is None: + errors.append("当前没有计划也没有持仓，不能用 exit，观望请用 wait") + if plan.action == "wait" and (ctx.has_plan or ctx.position_side is not None): + errors.append("有当前计划或持仓时不能用 wait：保留用 keep，结束用 exit") + for i, c in enumerate(plan.reevaluate, 1): + satisfied = ctx.mark_price > c.price if c.direction == "above" else ctx.mark_price < c.price + if satisfied: + errors.append(f"reevaluate 第 {i} 个条件（{c.direction} {c.price}）按现价 {ctx.mark_price} 已经满足，提交后会立即触发") + return errors + + +def check_plan_fields(plan: TradePlan, ctx: PlanContext) -> list[str]: + """action=plan 时的字段完整性、档数、比例、有效期""" + required = ("direction", "confidence", "entries", "stop", "take_profits", "entry_valid_hours", "max_hold_hours", "invalidation") + missing = [name for name in required if getattr(plan, name) is None] + if missing: + return [f"action=plan 时缺少：{', '.join(missing)}"] + errors = [] + angle = ctx.angle + label = "短线" if angle == "short" else "长线" + low, high = ENTRY_COUNT[angle] + if not low <= len(plan.entries) <= high: + errors.append(f"{label}入场档数须为 {low}–{high}，现为 {len(plan.entries)}" if low != high else f"{label}只能 {low} 档入场，现为 {len(plan.entries)}") + if abs(sum((e.ratio for e in plan.entries), Decimal(0)) - 1) > RATIO_TOLERANCE: + errors.append("入场各档比例之和须为 1") + if len(plan.take_profits) != len(TAKE_PROFIT_RATIOS[angle]): + errors.append(f"{label}须给 {len(TAKE_PROFIT_RATIOS[angle])} 个止盈，现为 {len(plan.take_profits)}") + if plan.entry_valid_hours > ENTRY_VALID_HOURS[angle]: + errors.append(f"{label} entry_valid_hours 不能超过 {ENTRY_VALID_HOURS[angle]}") + if plan.max_hold_hours > MAX_HOLD_HOURS[angle]: + errors.append(f"{label} max_hold_hours 不能超过 {MAX_HOLD_HOURS[angle]}") + if plan.invalidation.timeframe not in INVALIDATION_TIMEFRAMES[angle]: + errors.append(f"{label}失效条件的周期只能是 {'/'.join(INVALIDATION_TIMEFRAMES[angle])}") + if ctx.position_side is not None and plan.direction != ctx.position_side: + errors.append(f"已有{'多' if ctx.position_side == 'long' else '空'}头持仓，新计划方向必须一致；要反手请先用 exit") + return errors + + +def check_prices(direction: Direction, entries: list[Decimal], stop: Decimal, targets: list[Decimal], plan: TradePlan, ctx: PlanContext) -> list[str]: + """价格顺序：做多 止损 < 失效价 < 入场 < 止盈 1 < 止盈 2 < …，止损低于现价、止盈 1 高于现价；做空镜像""" + errors = [] + sign = 1 if direction == "long" else -1 + mark = ctx.mark_price + invalidation = plan.invalidation.price + + def below(a: Decimal, b: Decimal) -> bool: + # 做多时 a < b；做空时 a > b + return sign * (b - a) > 0 + + side = "做多" if direction == "long" else "做空" + if plan.invalidation.direction != ("below" if direction == "long" else "above"): + errors.append(f"{side}的失效条件方向应为 {'below' if direction == 'long' else 'above'}") + if not below(stop, invalidation): + errors.append(f"{side}时止损 {stop} 应在失效价 {invalidation} 之外") + nearest_entry = max(entries) if direction == "long" else min(entries) + farthest_entry = min(entries) if direction == "long" else max(entries) + if not below(invalidation, farthest_entry): + errors.append(f"{side}时失效价 {invalidation} 应在所有入场价之外") + if not below(stop, mark): + errors.append(f"{side}时止损 {stop} 应在现价 {mark} 之外，否则挂单会立即触发") + if len(set(entries)) != len(entries): + errors.append("入场档位价格重复") + if targets and not below(nearest_entry, targets[0]): + errors.append(f"{side}时止盈 1 {targets[0]} 应在所有入场价之外的盈利方向") + if targets and not below(mark, targets[0]): + errors.append(f"止盈 1 {targets[0]} 已越过现价 {mark}，挂单会立即成交") + if any(not below(a, b) for a, b in zip(targets, targets[1:], strict=False)): + errors.append("止盈价须按由近到远排列且不重复") + limit = mark * (1 + PERCENT_PRICE_LIMIT) if direction == "long" else mark * (1 - PERCENT_PRICE_LIMIT) + if any(below(limit, e) for e in entries): + errors.append(f"入场价超出交易所限价范围（{'不高于' if direction == 'long' else '不低于'} {limit:.2f}）") + return errors + + +def size_plan(plan: TradePlan, ctx: PlanContext) -> dict: + """校验计划并计算仓位；不通过时抛 PlanError。返回值可直接存 JSONB（数值为字符串）""" + errors = check_action(plan, ctx) + if plan.action != "plan": + if errors: + raise PlanError(errors) + return {} + errors += check_plan_fields(plan, ctx) + if errors: + raise PlanError(errors) + + angle, direction, rules = ctx.angle, plan.direction, ctx.rules + # 向保守方向取整：做多的入场、止损、止盈都向下（入场更低、止损更远、止盈更近），做空都向上 + rounding = ROUND_FLOOR if direction == "long" else ROUND_CEILING + entry_prices = [round_to_step(e.price, rules.tick_size, rounding) for e in plan.entries] + stop = round_to_step(plan.stop, rules.tick_size, rounding) + targets = [round_to_step(t.price, rules.tick_size, rounding) for t in plan.take_profits] + ratios = [e.ratio / sum((e.ratio for e in plan.entries), Decimal(0)) for e in plan.entries] + tp_ratios = TAKE_PROFIT_RATIOS[angle] + + errors = check_prices(direction, entry_prices, stop, targets, plan, ctx) + if errors: + raise PlanError(errors) + + average = sum((r * p for r, p in zip(ratios, entry_prices, strict=True)), Decimal(0)) + rr = reward_risk(average, stop, targets, tp_ratios) if angle == "long" else reward_risk(average, stop, targets[:1], (Decimal(1),)) + if rr < MIN_REWARD_RISK: + basis = "50/30/20 加权" if angle == "long" else "止盈 1" + raise PlanError([f"盈亏比 {rr:.2f}（按{basis}）低于 {MIN_REWARD_RISK}。止盈必须放在有依据的区，不得为凑盈亏比倒推；凑不够就不给这个方向"]) + nearest_entry = max(entry_prices) if direction == "long" else min(entry_prices) + rr_first = reward_risk(nearest_entry, stop, targets, tp_ratios) if angle == "long" else rr + + def unit_loss(price: Decimal) -> Decimal: + return abs(price - stop) + (price + stop) * TAKER_FEE_RATE + + risk_amount = ctx.capital * RISK_PCT[angle] * (LOW_CONFIDENCE_FACTOR if plan.confidence == "low" else 1) + max_notional = ctx.capital * MAX_LEVERAGE - ctx.other_notional + if max_notional <= 0: + raise PlanError([f"交易列表其他币的名义仓位 {ctx.other_notional:.2f} 已达到 {MAX_LEVERAGE} 倍交易资金，不能开新仓"]) + by_risk = risk_amount / unit_loss(average) + by_leverage = max_notional / average + total = min(by_risk, by_leverage) + quantities = [round_to_step(total * r, rules.step_size, ROUND_FLOOR) for r in ratios] + small = [ + f"第 {i} 档 {p} × {q} = {p * q:.2f} USDT" + for i, (p, q) in enumerate(zip(entry_prices, quantities, strict=True), 1) if p * q < rules.min_notional + ] + if small: + raise PlanError([ + f"入场档位名义价值低于最小下单金额 {rules.min_notional} USDT：{'、'.join(small)}。" + f"本次可用风险金额 {risk_amount:.2f} USDT、总数量约 {total:.6f}，请合并或删档" + ]) + + quantity = sum(quantities, Decimal(0)) + notional = sum((p * q for p, q in zip(entry_prices, quantities, strict=True)), Decimal(0)) + risk_at_stop = sum((q * unit_loss(p) for p, q in zip(entry_prices, quantities, strict=True)), Decimal(0)) + tp_quantities = split_take_profits(quantity, tp_ratios, rules.step_size) + return { + "direction": direction, + "entries": [ + {"price": str(p), "ratio": str(r.quantize(Decimal("0.0001"))), "quantity": str(q), "notional": str((p * q).quantize(Decimal("0.01")))} + for p, r, q in zip(entry_prices, ratios, quantities, strict=True) + ], + "stop": str(stop), + "take_profits": [ + {"price": str(t), "ratio": str(r), "quantity": str(q)} for t, r, q in zip(targets, tp_ratios, tp_quantities, strict=True) + ], + "average_price": str(average.quantize(rules.tick_size)), + "quantity": str(quantity), + "notional": str(notional.quantize(Decimal("0.01"))), + "capital": str(ctx.capital), + "leverage": str((notional / ctx.capital).quantize(Decimal("0.01"))), + "risk_amount": str(risk_amount.quantize(Decimal("0.01"))), + "risk_at_stop": str(risk_at_stop.quantize(Decimal("0.01"))), + "risk_pct": str((risk_at_stop / ctx.capital * 100).quantize(Decimal("0.01"))), + "capped_by_leverage": by_leverage < by_risk, + "reward_risk": str(rr.quantize(Decimal("0.01"))), + "reward_risk_first_only": str(rr_first.quantize(Decimal("0.01"))), + } diff --git a/backend/src/alphapulse/trading/snapshot.py b/backend/src/alphapulse/trading/snapshot.py new file mode 100644 index 0000000..264260f --- /dev/null +++ b/backend/src/alphapulse/trading/snapshot.py @@ -0,0 +1,146 @@ +"""账户、持仓、挂单快照：定时从模拟盘同步并写入 Redis，供页面和分析器读取""" + +import asyncio +import json +import logging +import time +from datetime import UTC, datetime + +from sqlalchemy import select + +from alphapulse.core.crypto import decrypt +from alphapulse.core.db import SessionLocal +from alphapulse.core.redis import KEY_PREFIX, redis +from alphapulse.models import TradingAccount, TradingSymbol +from alphapulse.trading.client import DemoClient, ExchangeError + +logger = logging.getLogger("alphapulse.trading.snapshot") + +# 进程挂掉后快照自动过期，页面据此显示「引擎离线」 +STATUS_KEY = f"{KEY_PREFIX}trader:status" +STATUS_TTL_SECONDS = 15 +SNAPSHOT_SECONDS = 3 +# 杠杆和保证金模式很少变，按币缓存 +SYMBOL_CONFIG_SECONDS = 60 + + +class Snapshot: + def __init__(self) -> None: + self.client: DemoClient | None = None + self.key_ciphertext: bytes | None = None + # 最近一次成功同步的数据；同步失败时保留并标注错误，避免页面闪空 + self.data: dict | None = None + self.synced_at: datetime | None = None + self.symbol_config: dict[str, tuple[float, dict]] = {} + + async def ensure_client(self, account: TradingAccount | None) -> DemoClient | None: + ciphertext = account.api_key_encrypted if account else None + if ciphertext != self.key_ciphertext: + # key 被修改或删除：旧连接作废，旧数据属于另一个账户，一并清掉 + if self.client is not None: + await self.client.http.aclose() + self.client = None + self.data = self.synced_at = None + self.symbol_config.clear() + self.key_ciphertext = ciphertext + if account is not None and ciphertext is not None: + self.client = DemoClient(decrypt(ciphertext), decrypt(account.api_secret_encrypted)) + return self.client + + async def leverage_and_margin(self, client: DemoClient, symbol: str) -> dict: + cached = self.symbol_config.get(symbol) + if cached is None or time.monotonic() - cached[0] > SYMBOL_CONFIG_SECONDS: + config = (await client.request("GET", "/fapi/v1/symbolConfig", {"symbol": symbol}))[0] + cached = (time.monotonic(), {"leverage": config["leverage"], "margin_type": config["marginType"]}) + self.symbol_config[symbol] = cached + return cached[1] + + async def fetch(self, client: DemoClient, symbols: list[str]) -> dict: + account = await client.request("GET", "/fapi/v3/account") + positions = [p for p in await client.request("GET", "/fapi/v3/positionRisk") if float(p["positionAmt"]) != 0] + # 挂单只查交易列表里的币和有持仓的币：不带 symbol 查全部的权重是 40，3 秒一次太贵 + watched = sorted(set(symbols) | {p["symbol"] for p in positions}) + orders, algo_orders = [], [] + for symbol in watched: + orders += await client.request("GET", "/fapi/v1/openOrders", {"symbol": symbol}) + algo_orders += await client.request("GET", "/fapi/v1/openAlgoOrders", {"symbol": symbol}) + return { + "account": { + "wallet_balance": account["totalWalletBalance"], + "margin_balance": account["totalMarginBalance"], + "available_balance": account["availableBalance"], + "unrealized_pnl": account["totalUnrealizedProfit"], + }, + "positions": [ + { + "symbol": p["symbol"], + "side": "LONG" if float(p["positionAmt"]) > 0 else "SHORT", + "amount": p["positionAmt"].lstrip("-"), + "entry_price": p["entryPrice"], + "break_even_price": p["breakEvenPrice"], + "mark_price": p["markPrice"], + "unrealized_pnl": p["unRealizedProfit"], + "liquidation_price": p["liquidationPrice"], + "notional": p["notional"].lstrip("-"), + "margin": p["isolatedMargin"] if float(p["isolatedMargin"]) else p["initialMargin"], + **await self.leverage_and_margin(client, p["symbol"]), + } + for p in positions + ], + "orders": [ + { + "symbol": o["symbol"], "client_id": o["clientOrderId"], "side": o["side"], "type": o["type"], + "price": o["price"], "quantity": o["origQty"], "filled": o["executedQty"], "reduce_only": o["reduceOnly"], + "time": o["time"], + } + for o in orders + ], + "algo_orders": [ + { + "symbol": o["symbol"], "client_id": o["clientAlgoId"], "side": o["side"], "type": o["orderType"], + "trigger_price": o["triggerPrice"], "quantity": o["quantity"], "close_position": o["closePosition"], + "reduce_only": o["reduceOnly"], "time": o["createTime"], + } + for o in algo_orders + ], + } + + async def publish(self) -> None: + async with SessionLocal() as session: + account = await session.get(TradingAccount, 1) + symbols = list((await session.scalars(select(TradingSymbol.symbol))).all()) + error = None + client = await self.ensure_client(account) + if client is None: + error = "还没有填写模拟盘 key" + else: + try: + self.data = await self.fetch(client, symbols) + self.synced_at = datetime.now(UTC) + except ExchangeError as e: + logger.warning("同步模拟盘失败：%s", e) + error = str(e) + status = { + "updated_at": datetime.now(UTC).isoformat(), + "synced_at": self.synced_at.isoformat() if self.synced_at else None, + "error": error, + **(self.data or {"account": None, "positions": [], "orders": [], "algo_orders": []}), + } + await redis.set(STATUS_KEY, json.dumps(status), ex=STATUS_TTL_SECONDS) + + +async def snapshot_loop() -> None: + snapshot = Snapshot() + while True: + try: + await snapshot.publish() + except Exception: + logger.exception("发布快照失败，稍后重试") + await asyncio.sleep(SNAPSHOT_SECONDS) + + + + +async def read_snapshot() -> dict | None: + raw = await redis.get(STATUS_KEY) + return json.loads(raw) if raw else None diff --git a/backend/tests/test_plan.py b/backend/tests/test_plan.py new file mode 100644 index 0000000..02c8fca --- /dev/null +++ b/backend/tests/test_plan.py @@ -0,0 +1,139 @@ +from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal + +import pytest + +from alphapulse.trading.plan import PlanContext, PlanError, SymbolRules, TradePlan, round_to_step, size_plan, split_take_profits + +BTC = SymbolRules(tick_size=Decimal("0.1"), step_size=Decimal("0.0001"), min_notional=Decimal("50")) + + +def context(**overrides) -> PlanContext: + values = { + "angle": "long", "mark_price": Decimal("82000"), "capital": Decimal("1000"), "other_notional": Decimal("0"), + "rules": BTC, "has_plan": False, "position_side": None, + } + return PlanContext(**{**values, **overrides}) + + +def long_plan(**overrides) -> TradePlan: + values = { + "action": "plan", "summary": "长线做多", "reasons": "支撑区", "risks": "无", + "direction": "long", "confidence": "medium", + "entries": [{"price": "81000", "ratio": "0.4", "basis": "区1"}, {"price": "80000", "ratio": "0.6", "basis": "区2"}], + "stop": "79500", "stop_basis": "区下沿外", + "take_profits": [{"price": "83000", "basis": "压1"}, {"price": "84000", "basis": "压2"}, {"price": "85000", "basis": "压3"}], + "entry_valid_hours": "24", "max_hold_hours": "48", + "invalidation": {"direction": "below", "price": "79800", "timeframe": "4h", "basis": "跌破区"}, + } + return TradePlan.model_validate({**values, **overrides}) + + +def short_term_plan(**overrides) -> TradePlan: + values = { + "action": "plan", "summary": "短线做空", "reasons": "压力区", "risks": "无", + "direction": "short", "confidence": "medium", + "entries": [{"price": "82500", "ratio": "1", "basis": "压力区"}], + "stop": "82900", "take_profits": [{"price": "81800", "basis": "支1"}, {"price": "81300", "basis": "支2"}], + "entry_valid_hours": "4", "max_hold_hours": "24", + "invalidation": {"direction": "above", "price": "82800", "timeframe": "15m", "basis": "站上区"}, + } + return TradePlan.model_validate({**values, **overrides}) + + +def errors_of(plan: TradePlan, ctx: PlanContext) -> str: + with pytest.raises(PlanError) as e: + size_plan(plan, ctx) + return "；".join(e.value.errors) + + +def test_round_to_step(): + assert round_to_step(Decimal("81000.05"), Decimal("0.1"), ROUND_FLOOR) == Decimal("81000.0") + # 整数输入不能变成指数形式 + assert str(round_to_step(Decimal("80500"), Decimal("0.1"), ROUND_FLOOR)) == "80500.0" + assert round_to_step(Decimal("82500.01"), Decimal("0.1"), ROUND_CEILING) == Decimal("82500.1") + assert round_to_step(Decimal("0.00829908"), Decimal("0.0001"), ROUND_FLOOR) == Decimal("0.0082") + + +def test_split_take_profits_keeps_total(): + parts = split_take_profits(Decimal("0.0206"), (Decimal("0.5"), Decimal("0.3"), Decimal("0.2")), Decimal("0.0001")) + assert parts == [Decimal("0.0103"), Decimal("0.0061"), Decimal("0.0042")] + assert sum(parts) == Decimal("0.0206") + assert split_take_profits(Decimal("0.001"), (Decimal("0.5"), Decimal("0.3"), Decimal("0.2")), Decimal("0.001")) == [0, 0, Decimal("0.001")] + + +def test_long_plan_sizing_matches_hand_calculation(): + sizing = size_plan(long_plan(), context()) + # 均价 0.4×81000 + 0.6×80000 = 80400；单位亏损 900 + (80400+79500)×0.0004 = 963.96；20 ÷ 963.96 = 0.020747 + assert sizing["average_price"] == "80400.0" + assert [e["quantity"] for e in sizing["entries"]] == ["0.0082", "0.0124"] + assert sizing["quantity"] == "0.0206" + assert sizing["notional"] == "1656.20" + assert sizing["risk_amount"] == "20.00" + assert sizing["risk_at_stop"] == "19.82" + # (0.5×2600 + 0.3×3600 + 0.2×4600) ÷ 900；只成交首档 (0.5×2000 + 0.3×3000 + 0.2×4000) ÷ 1500 + assert sizing["reward_risk"] == "3.67" + assert sizing["reward_risk_first_only"] == "1.80" + assert [t["quantity"] for t in sizing["take_profits"]] == ["0.0103", "0.0061", "0.0042"] + assert sizing["capped_by_leverage"] is False + + +def test_low_confidence_halves_risk(): + sizing = size_plan(long_plan(confidence="low"), context()) + assert sizing["risk_amount"] == "10.00" + # 10 ÷ 963.96 = 0.010374：0.4 档 0.0041，0.6 档 0.0062 + assert sizing["quantity"] == "0.0103" + + +def test_leverage_cap_counts_other_positions(): + sizing = size_plan(long_plan(), context(other_notional=Decimal("2000"))) + assert sizing["capped_by_leverage"] is True + assert Decimal(sizing["notional"]) <= Decimal("1000") + + +def test_short_term_short_plan(): + sizing = size_plan(short_term_plan(), context(angle="short")) + # 止盈 1 盈亏比 700 ÷ 400 + assert sizing["reward_risk"] == "1.75" + assert sizing["risk_amount"] == "10.00" + assert len(sizing["take_profits"]) == 2 + + +def test_short_prices_round_up(): + sizing = size_plan(short_term_plan(entries=[{"price": "82500.01", "ratio": "1", "basis": "x"}]), context(angle="short")) + assert sizing["entries"][0]["price"] == "82500.1" + + +def test_rejects_wrong_counts_and_timeframe(): + message = errors_of(short_term_plan(entries=[{"price": "82500", "ratio": "0.5", "basis": "a"}, {"price": "82600", "ratio": "0.5", "basis": "b"}]), context(angle="short")) + assert "短线只能 1 档入场" in message + assert "长线失效条件的周期只能是 4h" in errors_of(long_plan(invalidation={"direction": "below", "price": "79800", "timeframe": "1h", "basis": "x"}), context()) + assert "长线须给 3 个止盈" in errors_of(long_plan(take_profits=[{"price": "83000", "basis": "a"}]), context()) + + +def test_rejects_bad_price_order(): + message = errors_of(long_plan(stop="80500"), context()) + assert "做多时止损 80500.0 应在失效价 79800 之外" in message + assert "止盈 1 81000.0 已越过现价" in errors_of(long_plan(take_profits=[{"price": "81000", "basis": "a"}, {"price": "84000", "basis": "b"}, {"price": "85000", "basis": "c"}], entries=[{"price": "80500", "ratio": "1", "basis": "a"}]), context()) + + +def test_rejects_low_reward_risk(): + # (0.5×400 + 0.3×600 + 0.2×800) ÷ 500 = 1.08 + plan = long_plan(take_profits=[{"price": "80400", "basis": "a"}, {"price": "80600", "basis": "b"}, {"price": "80800", "basis": "c"}], + entries=[{"price": "80000", "ratio": "1", "basis": "a"}]) + assert "盈亏比 1.08" in errors_of(plan, context(mark_price=Decimal("80300"))) + + +def test_rejects_entries_below_min_notional(): + message = errors_of(long_plan(entries=[{"price": "81000", "ratio": "0.02", "basis": "a"}, {"price": "80000", "ratio": "0.98", "basis": "b"}]), context()) + assert "最小下单金额 50" in message + assert "第 1 档" in message + + +def test_action_rules(): + assert "不能用 keep" in errors_of(TradePlan(action="keep", summary="s", reasons="r", risks="x"), context()) + assert "不能用 wait" in errors_of(TradePlan(action="wait", summary="s", reasons="r", risks="x"), context(position_side="long")) + assert size_plan(TradePlan(action="wait", summary="s", reasons="r", risks="x"), context()) == {} + satisfied = TradePlan(action="wait", summary="s", reasons="r", risks="x", + reevaluate=[{"direction": "above", "price": "81000", "timeframe": "1h", "basis": "x"}]) + assert "已经满足" in errors_of(satisfied, context()) + assert "方向必须一致" in errors_of(long_plan(), context(position_side="short")) diff --git a/backend/uv.lock b/backend/uv.lock index b194a70..2a11f9b 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -186,6 +186,11 @@ dependencies = [ { name = "uvicorn", extra = ["standard"] }, ] +[package.dev-dependencies] +dev = [ + { name = "pytest" }, +] + [package.metadata] requires-dist = [ { name = "alembic", specifier = ">=1.20.0" }, @@ -205,6 +210,9 @@ requires-dist = [ { name = "uvicorn", extras = ["standard"], specifier = ">=0.54.0" }, ] +[package.metadata.requires-dev] +dev = [{ name = "pytest", specifier = ">=9.1.1" }] + [[package]] name = "annotated-doc" version = "0.0.5" @@ -665,6 +673,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/58/50/6c0d534c5f134586a8e1ba4e330569e32f057e33372ae556463212fb4cd3/click-8.5.0-py3-none-any.whl", hash = "sha256:255bc9599cf7748b4b1a446ccc735421bd08a2ae529a8b88597d3de5664ee360", size = 125251, upload-time = "2026-08-26T13:33:12.928Z" }, ] +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + [[package]] name = "constantly" version = "23.10.4" @@ -1101,6 +1118,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1d/55/0f4df2a44053867ea9cbea73fc588b03c55605cd695cee0a3d86f0029cb2/incremental-24.11.0-py3-none-any.whl", hash = "sha256:a34450716b1c4341fe6676a0598e88a39e04189f4dce5dc96f656e040baa10b3", size = 21109, upload-time = "2025-11-28T02:30:16.442Z" }, ] +[[package]] +name = "iniconfig" +version = "2.3.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/01/e1/2069291243c926a2ff1cd706c7f3eeb9b62144bf60f77c9fb9ff2fb26bd3/iniconfig-2.3.1.tar.gz", hash = "sha256:67f4b9c50da0dedf52af349e7749a80a9057a5031199791b906c3bb3ae878960", size = 21209, upload-time = "2026-10-06T22:48:38.076Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/56/43/4ca9e49d27a1fcf6bece6f6aec0ea46bb9112489b93d4b688fb415457bdb/iniconfig-2.3.1-py3-none-any.whl", hash = "sha256:9121e2c1fdb355232495be3194c8dfe87ccc2d5dee45947b78e68f499790d7a7", size = 7552, upload-time = "2026-10-06T22:48:36.959Z" }, +] + [[package]] name = "itemadapter" version = "0.13.1" @@ -1592,6 +1618,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8d/15/1633010b26e88e872c93b67c0b6c5e174fb74cb6fb5c1472b4d51d4a8f22/platformdirs-4.13.0-py3-none-any.whl", hash = "sha256:3dbcf4cd708f21cf876c4eaa90e58412bc4f033d87143f41b1493ff77c25b7e1", size = 32724, upload-time = "2026-10-11T02:05:22.776Z" }, ] +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + [[package]] name = "propcache" version = "0.5.4" @@ -1829,6 +1864,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/66/0e/9ee7bc0b48ec45d93b302fa2d787830dca4dc454d31a237faa5815995988/PyDispatcher-2.0.7-py3-none-any.whl", hash = "sha256:96543bea04115ffde08f851e1d45cacbfd1ee866ac42127d9b476dc5aefa7de0", size = 12040, upload-time = "2023-02-17T20:11:11.991Z" }, ] +[[package]] +name = "pygments" +version = "2.21.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/49/2e/ced460408999b33da6b31b0021b0f37d329e202d4169aeb164493778f25b/pygments-2.21.0.tar.gz", hash = "sha256:610ca751c9bc2492b38eb9a38a7fbc93edbbb2d7182edaf34e66ae493dee5c8c", size = 5005329, upload-time = "2026-08-17T08:02:48.824Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" }, +] + [[package]] name = "pyjwt" version = "2.15.1" @@ -1856,6 +1900,22 @@ version = "2.1.2" source = { registry = "https://pypi.org/simple" } sdist = { url = "https://files.pythonhosted.org/packages/d5/7b/65f55513d3c769fd677f90032d8d8703e3dc17e88a41b6074d2177548bca/PyPyDispatcher-2.1.2.tar.gz", hash = "sha256:b6bec5dfcff9d2535bca2b23c80eae367b1ac250a645106948d315fcfa9130f2", size = 23224, upload-time = "2017-07-03T14:20:51.806Z" } +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.4" diff --git a/web/src/apis/trading.ts b/web/src/apis/trading.ts index ae83100..d48b08c 100644 --- a/web/src/apis/trading.ts +++ b/web/src/apis/trading.ts @@ -140,3 +140,95 @@ export function deleteSymbol(symbol: string): Promise { export function listEvents(params: { symbol?: string, before?: number, limit?: number }): Promise { return http.get("/trading/events", { params }) } + +export interface PlanCondition { + direction: "above" | "below" + price: string + timeframe: "mark" | "1m" | "15m" | "1h" | "4h" + basis: string +} + +/** submit_trade_plan 的参数原样保存；数值为字符串 */ +export interface TradeDecision { + action: "plan" | "keep" | "exit" | "wait" + summary: string + reasons: string + risks: string + reevaluate: PlanCondition[] + direction: "long" | "short" | null + confidence: "high" | "medium" | "low" | null + entries: { price: string, ratio: string, basis: string }[] | null + stop: string | null + stop_basis: string | null + take_profits: { price: string, basis: string }[] | null + entry_valid_hours: string | null + max_hold_hours: string | null + invalidation: PlanCondition | null +} + +/** 程序按交易资金算出的数量和风险 */ +export interface Sizing { + direction: "long" | "short" + entries: { price: string, ratio: string, quantity: string, notional: string }[] + stop: string + take_profits: { price: string, ratio: string, quantity: string }[] + average_price: string + quantity: string + notional: string + capital: string + leverage: string + risk_amount: string + risk_at_stop: string + risk_pct: string + capped_by_leverage: boolean + reward_risk: string + reward_risk_first_only: string +} + +export interface Analysis { + id: number + symbol: string + angle: Angle + reasons: string[] + dry_run: boolean + status: "running" | "done" | "failed" + action: TradeDecision["action"] | null + decision: TradeDecision | null + sizing: Sizing | null + error: string | null + model: string | null + reasoning_effort: string | null + usage: { input_tokens: number, output_tokens: number } | null + started_at: string + finished_at: string | null + plan_status: "pending" | "open" | "closed" | "canceled" | "replaced" | null + valid_until: string | null + first_fill_at: string | null + close_deadline: string | null + closed_at: string | null + close_reason: string | null + result: Record | null +} + +/** 与协议无关的消息格式，同 Chat */ +export type TranscriptMessage + = | { role: "user", text: string } + | { role: "assistant", text: string, tool_calls: { id: string, name: string, arguments: Record }[] } + | { role: "tool", tool_call_id: string, name: string, result: string } + +export interface AnalysisDetail extends Analysis { + text: string | null + transcript: TranscriptMessage[] | null +} + +export function requestAnalysis(symbol: string): Promise { + return http.post(`/trading/symbols/${symbol}/analyze`) +} + +export function listAnalyses(params: { symbol?: string, limit?: number }): Promise { + return http.get("/trading/analyses", { params }) +} + +export function getAnalysis(id: number): Promise { + return http.get(`/trading/analyses/${id}`) +} diff --git a/web/src/pages/trading/components/AnalysisDrawer.vue b/web/src/pages/trading/components/AnalysisDrawer.vue new file mode 100644 index 0000000..fa4604b --- /dev/null +++ b/web/src/pages/trading/components/AnalysisDrawer.vue @@ -0,0 +1,68 @@ + + + diff --git a/web/src/pages/trading/components/PlanCard.vue b/web/src/pages/trading/components/PlanCard.vue new file mode 100644 index 0000000..d7049ef --- /dev/null +++ b/web/src/pages/trading/components/PlanCard.vue @@ -0,0 +1,148 @@ + + + + + diff --git a/web/src/pages/trading/index.vue b/web/src/pages/trading/index.vue index 2271da7..f4e5a1a 100644 --- a/web/src/pages/trading/index.vue +++ b/web/src/pages/trading/index.vue @@ -1,11 +1,13 @@