Add demo trading analyst with plan validation and sizing

The analyst prompts the model with trading rules, published skills,
account sizing inputs and prefetched market data, exposes read-only
tools plus submit_trade_plan, and validates and sizes the submitted plan
in pure functions covered by pytest. Manual analyses run as dry runs
while trading is stopped; the trading page shows the latest plan and an
analysis detail drawer.

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")
 

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)

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)

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)

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})

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):

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__":

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"))),
+    }

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