Add demo trading executor and analysis scheduling (plan 3 C4)

- Executor reconciles each symbol to its desired orders: journaled ap{id}
  client ids, reduce-only quantity stop placed before entries, make-before-
  break stop moves, breakeven after TP1, orphan cancel, -2021 market close
- Plan lifecycle: proposed plans are adopted only if the position is
  unchanged; holding-mode plans may only tighten the stop
- Program rules: time stop, short-term invalidation close, long-term 4h
  review, reevaluate conditions, takeover of unmanaged positions
- Trigger and routine analysis scheduling with a 10 min gap, daily cap and
  bounded retries
- Plan endpoint and plan status, fills and countdowns on the trading page

backend/src/alphapulse/api/trading.py

@@ -17,7 +17,7 @@ from alphapulse.api.market import call_binance, client as market_client
 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.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol
 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
@@ -30,6 +30,8 @@ SYMBOLS_CACHE_KEY = f"{KEY_PREFIX}trading:symbols"
 SYMBOLS_CACHE_SECONDS = 3600
 # 删除或切换角度前要求快照足够新，否则无法确认没有仓位和挂单
 SNAPSHOT_FRESH_SECONDS = 30
+# 执行器正在管理的计划状态
+LIVE_PLAN_STATUSES = ("pending", "open", "closing")
 
 
 class TodayUsage(BaseModel):
@@ -70,8 +72,10 @@ class SymbolOut(BaseModel):
     added_at: datetime
     analysis_requested_at: datetime | None
     analyzing: bool
-    # 当前计划的状态：pending / open，没有计划为空
+    # 当前计划：pending / open / closing，没有计划为空
+    plan_id: int | None
     plan_status: str | None
+    close_deadline: datetime | None
 
 
 class SymbolIn(BaseModel):
@@ -113,6 +117,28 @@ class AnalysisDetail(AnalysisOut):
     transcript: list | None
 
 
+class OrderOut(BaseModel):
+    id: int
+    purpose: str
+    leg: int | None
+    side: str
+    type: str
+    price: Decimal | None
+    trigger_price: Decimal | None
+    quantity: Decimal
+    status: str
+    filled_qty: Decimal
+    avg_price: Decimal | None
+    error: str | None
+    created_at: datetime
+
+
+class SymbolPlanOut(BaseModel):
+    # 执行器正在管理的计划；没有时为最近一次待采纳的提议
+    plan: AnalysisOut | None
+    orders: list[OrderOut]
+
+
 class EventOut(BaseModel):
     id: int
     created_at: datetime
@@ -236,14 +262,15 @@ async def list_symbol_options() -> list[SymbolOption]:
 async def list_symbols(session: Annotated[AsyncSession, Depends(get_session)]) -> list[SymbolOut]:
     symbols = (await session.scalars(select(TradingSymbol).order_by(TradingSymbol.added_at))).all()
     analyzing = set((await session.scalars(select(TradingAnalysis.symbol).where(TradingAnalysis.status == "running"))).all())
-    plans = dict((await session.execute(
+    plans = {p.symbol: p for p in await session.scalars(
-        select(TradingAnalysis.symbol, TradingAnalysis.plan_status)
+        select(TradingAnalysis).where(TradingAnalysis.plan_status.in_(LIVE_PLAN_STATUSES)).order_by(TradingAnalysis.id)
-        .where(TradingAnalysis.plan_status.in_(("pending", "open")), TradingAnalysis.dry_run.is_(False))
+    )}
-    )).all())
     return [
         SymbolOut(
             symbol=s.symbol, angle=s.angle, added_at=s.added_at, analysis_requested_at=s.analysis_requested_at,
-            analyzing=s.symbol in analyzing, plan_status=plans.get(s.symbol),
+            analyzing=s.symbol in analyzing, plan_id=plans[s.symbol].id if s.symbol in plans else None,
+            plan_status=plans[s.symbol].plan_status if s.symbol in plans else None,
+            close_deadline=plans[s.symbol].close_deadline if s.symbol in plans else None,
         )
         for s in symbols
     ]
@@ -269,8 +296,7 @@ async def ensure_idle(session: AsyncSession, symbol: str, action: str) -> Tradin
         raise HTTPException(status.HTTP_404_NOT_FOUND, f"{symbol} 不在交易列表中")
     busy = await session.scalar(select(func.count()).where(
         TradingAnalysis.symbol == symbol,
-        or_(TradingAnalysis.status == "running",
+        or_(TradingAnalysis.status == "running", TradingAnalysis.plan_status.in_(("proposed", *LIVE_PLAN_STATUSES))),
-            TradingAnalysis.plan_status.in_(("pending", "open")) & TradingAnalysis.dry_run.is_(False)),
     ))
     if busy:
         raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有进行中的分析或计划，不能{action}")
@@ -354,3 +380,18 @@ async def get_analysis(analysis_id: int, session: Annotated[AsyncSession, Depend
     if analysis is None:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "分析不存在")
     return AnalysisDetail.model_validate(analysis, from_attributes=True)
+
+
+@router.get("/symbols/{symbol}/plan")
+async def get_symbol_plan(symbol: str, session: Annotated[AsyncSession, Depends(get_session)]) -> SymbolPlanOut:
+    plan = await session.scalar(
+        select(TradingAnalysis).where(TradingAnalysis.symbol == symbol, TradingAnalysis.plan_status.in_(("proposed", *LIVE_PLAN_STATUSES)))
+        .order_by(TradingAnalysis.id.desc()).limit(1)
+    )
+    if plan is None:
+        return SymbolPlanOut(plan=None, orders=[])
+    orders = await session.scalars(select(TradingOrder).where(TradingOrder.analysis_id == plan.id).order_by(TradingOrder.id))
+    return SymbolPlanOut(
+        plan=AnalysisOut.model_validate(plan, from_attributes=True),
+        orders=[OrderOut.model_validate(o, from_attributes=True) for o in orders],
+    )

backend/src/alphapulse/models/trading.py

@@ -69,7 +69,8 @@ class TradingAnalysis(Base):
     error: Mapped[str | None] = mapped_column(Text)
     started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
     finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
-    # pending（挂单中）/ open（持仓中）/ closed / canceled（未成交即结束）/ replaced（被新计划替换）
+    # proposed（待执行器采纳）/ pending（挂单中）/ open（持仓中）/ closing（平仓中）/ closed / canceled（未成交即结束）
+    # / replaced（被新计划替换）/ stale（采纳前持仓已变，作废）/ applied（exit 决定已执行）
     plan_status: Mapped[str | None] = mapped_column(String(16))
     valid_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     first_fill_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

backend/src/alphapulse/trading/analyst.py

@@ -6,7 +6,6 @@
 import asyncio
 import json
 import logging
-import time
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from decimal import Decimal
@@ -18,10 +17,10 @@ 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.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol
 from alphapulse.trading import plan as rules
-from alphapulse.trading.client import public_get
+from alphapulse.trading.client import OWN_CLIENT_ID, ExchangeError, mark_price, symbol_rules
-from alphapulse.trading.plan import PlanContext, PlanError, SymbolRules, TradePlan, size_plan
+from alphapulse.trading.plan import PlanContext, PlanError, TradePlan, size_plan
 from alphapulse.trading.snapshot import read_snapshot
 
 logger = logging.getLogger("alphapulse.trading.analyst")
@@ -30,7 +29,6 @@ READ_TOOLS = ("search_news", "get_news", "search_symbols", "get_klines", "get_ti
 # 最后一轮只开放 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": "低"}
@@ -74,6 +72,11 @@ INSTRUCTIONS = f"""你是 AlphaPulse 的模拟盘交易分析器：为一个币
 - 每档入场的名义价值不低于交易所最小下单金额，不足时合并或删档。
 - 单笔风险为交易资金的 {_pct(rules.RISK_PCT['short'])}（短线）或 {_pct(rules.RISK_PCT['long'])}（长线），置信度低时减半；名义仓位合计不超过 {rules.MAX_LEVERAGE} 倍交易资金。这两条由程序计算数量来保证。
 - 有持仓时新计划的方向必须与持仓一致；要反手先用 exit。
+
+## 有持仓时
+- action=plan 只用来调整止损（只能收紧）、止盈、失效条件和 reevaluate：entries 留空，不加仓（README 第 10 节禁止计划外加仓）。止盈按当前持仓数量拆分；平仓截止时间沿用首档成交时确定的，不延长。
+- 计划仍然合适就用 keep；要提前结束用 exit（撤销未成交的入场单并市价平仓）。
+- 计划提交后由执行器在下一轮（约 2 秒内）按最新持仓核对后生效；期间持仓变了，计划作废并重新请你分析。
 """
 
 SUBMIT_SPEC = llm.ToolSpec(
@@ -96,32 +99,8 @@ class Prepared:
     system: str
     user_text: str
     context: PlanContext
-
+    # 当前计划的仓位计算结果，提交时重算当前止损用
-
+    current_sizing: dict | None
-_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:
@@ -168,9 +147,15 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared
         )).all()
         current = await session.scalar(
             select(TradingAnalysis)
-            .where(TradingAnalysis.symbol == symbol, TradingAnalysis.plan_status.in_(("pending", "open")), TradingAnalysis.dry_run.is_(False))
+            .where(TradingAnalysis.symbol == symbol, TradingAnalysis.plan_status.in_(("pending", "open", "closing")), TradingAnalysis.dry_run.is_(False))
             .order_by(TradingAnalysis.id.desc()).limit(1)
         )
+        fills = [
+            {"purpose": o.purpose, "leg": o.leg, "filled": str(o.filled_qty), "avg_price": str(o.avg_price), "status": o.status}
+            for o in (await session.scalars(
+                select(TradingOrder).where(TradingOrder.analysis_id == current.id, TradingOrder.filled_qty > 0).order_by(TradingOrder.id)
+            )).all()
+        ] if current else []
         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
@@ -185,9 +170,10 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared
     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_,
+        angle=angle, mark_price=mark, capital=capital, other_notional=other_notional, rules=symbol_rules_, has_plan=current is not None,
-        has_plan=current is not None, position_side=("long" if position["side"] == "LONG" else "short") if position else None,
+        position_side=None,
     )
+    apply_position(context, snapshot, symbol, current.sizing if current else None)
 
     orders = [o for o in snapshot["orders"] + snapshot["algo_orders"] if o["symbol"] == symbol]
     risk = capital * rules.RISK_PCT[angle]
@@ -210,7 +196,15 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared
         f"- 挂单：{dump(orders) if orders else '无'}",
         "",
         "## 当前计划",
-        dump({"id": current.id, "plan_status": current.plan_status, **plan_brief(current.decision, current.sizing)}) if current else "无",
+        dump({
+            "id": current.id, "plan_status": current.plan_status,
+            "valid_until": current.valid_until and local(current.valid_until),
+            "first_fill_at": current.first_fill_at and local(current.first_fill_at),
+            "close_deadline": current.close_deadline and local(current.close_deadline),
+            "current_stop": str(context.current_stop) if context.current_stop else None,
+            "fills": fills,
+            **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 ["无"]),
@@ -218,7 +212,28 @@ async def prepare(symbol: str, reasons: list[str], analysis_id: int) -> Prepared
         "## 预取的行情（实盘数据，已取，不要用相同参数重复调用）",
         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)
+    return Prepared(provider=provider, effort=effort, dry_run=dry_run, angle=angle, system=system, user_text="\n".join(lines),
+                    context=context, current_sizing=current.sizing if current else None)
+
+
+def apply_position(context: PlanContext, snapshot: dict, symbol: str, current_sizing: dict | None) -> None:
+    """按快照填入持仓与当前止损。当前止损取交易所上本程序止损单里最紧的一张（可能已移到保本价），没有则取当前计划的止损"""
+    position = next((p for p in snapshot["positions"] if p["symbol"] == symbol), None)
+    if position is None:
+        context.position_side, context.position_qty, context.position_entry, context.current_stop = None, Decimal(0), None, None
+        return
+    side = "long" if position["side"] == "LONG" else "short"
+    close_side = "SELL" if side == "long" else "BUY"
+    stops = [
+        Decimal(o["trigger_price"]) for o in snapshot["algo_orders"]
+        if o["symbol"] == symbol and o["side"] == close_side and o["type"].startswith("STOP") and OWN_CLIENT_ID.match(o["client_id"])
+    ]
+    if stops:
+        current_stop = max(stops) if side == "long" else min(stops)
+    else:
+        current_stop = Decimal(current_sizing["stop"]) if current_sizing else None
+    context.position_side, context.position_qty = side, Decimal(position["amount"])
+    context.position_entry, context.current_stop = Decimal(position["entry_price"]), current_stop
 
 
 async def complete(prepared: Prepared, messages: list[dict], tools: list[llm.ToolSpec]) -> llm.Completion:
@@ -252,8 +267,11 @@ async def submit(prepared: Prepared, symbol: str, analysis_id: int, arguments: d
         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)
+    snapshot = await read_snapshot()
+    if snapshot:
+        apply_position(prepared.context, snapshot, symbol, prepared.current_sizing)
     try:
         sizing = size_plan(plan, prepared.context)
     except PlanError as e:
@@ -319,7 +337,7 @@ async def run_analysis(symbol: str, reasons: list[str]) -> None:
                 break
         if accepted is None:
             raise AnalysisError(f"模型在 {MAX_ROUNDS} 轮内没有提交通过校验的结论")
-    except (AnalysisError, llm.LlmError, PlanError) as e:
+    except (AnalysisError, ExchangeError, llm.LlmError, PlanError) as e:
         error = str(e)
     except Exception as e:
         logger.exception("%s 分析出错", symbol)
@@ -343,6 +361,9 @@ async def run_analysis(symbol: str, reasons: list[str]) -> None:
             analysis.action = plan.action
             analysis.decision = plan.model_dump(mode="json")
             analysis.sizing = sizing or None
+            # 交易中的新计划和结束决定交给执行器按最新持仓核对后采纳
+            if not dry_run and plan.action in ("plan", "exit"):
+                analysis.plan_status = "proposed"
         await session.commit()
     if error:
         await add_event(symbol, analysis_id, "analysis_failed", f"分析失败：{error}", "error")

backend/src/alphapulse/trading/client.py

@@ -7,14 +7,18 @@ import hashlib
 import hmac
 import re
 import time
+from decimal import Decimal
 from typing import Any
 from urllib.parse import urlencode
 
 import httpx
 
+from alphapulse.trading.plan import SymbolRules
+
 BASE_URL = "https://demo-fapi.binance.com"
 RECV_WINDOW_MS = 5000
 TIMEOUT_SECONDS = 10
+EXCHANGE_INFO_SECONDS = 3600
 # 本程序下的单 clientOrderId / clientAlgoId 为 ap{流水 id}；其他（含用户手动下的）不动
 OWN_CLIENT_ID = re.compile(r"^ap\d+$")
 
@@ -96,3 +100,35 @@ class DemoClient:
                 continue
             raise ExchangeError(resp.status_code, code, body.get("msg", resp.text[:200]) if isinstance(body, dict) else resp.text[:200])
         raise AssertionError("unreachable")
+
+
+_exchange_info: tuple[float, dict[str, SymbolRules]] | None = None
+
+
+def step_decimal(raw: str) -> Decimal:
+    """去掉多余的 0（"0.10" → 0.1），但整数步长不能变成 1E+2 这类指数形式，否则按它取整的价格和数量也会是指数形式"""
+    value = Decimal(raw).normalize()
+    return value.quantize(Decimal(1)) if value.as_tuple().exponent > 0 else value
+
+
+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=step_decimal(filters["PRICE_FILTER"]["tickSize"]),
+                step_size=step_decimal(filters["LOT_SIZE"]["stepSize"]),
+                min_notional=Decimal(filters["MIN_NOTIONAL"]["notional"]),
+            )
+        _exchange_info = (time.monotonic(), parsed)
+    if symbol not in _exchange_info[1]:
+        raise ExchangeError(None, None, 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"])

backend/src/alphapulse/trading/engine.py

@@ -1,4 +1,4 @@
-"""模拟盘交易进程：同步账户快照、调度分析。
+"""模拟盘交易进程：同步账户快照、执行器对账、调度分析。
 
 用法：python -m alphapulse.trading.engine
 """
@@ -12,6 +12,8 @@ from sqlalchemy import select, update
 from alphapulse.core.db import SessionLocal
 from alphapulse.models import TradingAnalysis, TradingSymbol
 from alphapulse.trading.analyst import run_analysis
+from alphapulse.trading.executor import executor_loop
+from alphapulse.trading.schedule import routine_requests
 from alphapulse.trading.snapshot import snapshot_loop
 
 logger = logging.getLogger("alphapulse.trading.engine")
@@ -57,6 +59,7 @@ async def analysis_loop() -> None:
                     del tasks[symbol]
                     if not task.cancelled() and task.exception():
                         logger.error("%s 分析任务异常退出", symbol, exc_info=task.exception())
+            await routine_requests()
             free = MAX_CONCURRENT_ANALYSES - len(tasks)
             if free > 0:
                 for symbol, reasons in await claim_requests(free, set(tasks)):
@@ -69,6 +72,7 @@ async def analysis_loop() -> None:
 async def main() -> None:
     async with asyncio.TaskGroup() as tg:
         tg.create_task(snapshot_loop())
+        tg.create_task(executor_loop())
         tg.create_task(analysis_loop())
 
 

backend/src/alphapulse/trading/executor.py

@@ -0,0 +1,578 @@
+"""执行器：按「应有状态」对账。唯一会下单、撤单、推进计划状态的代码。
+
+每轮对交易列表里的每个币：
+1. 同步订单流水与交易所（成交、撤销、触发）。
+2. 采纳分析器提交的新计划（按最新持仓核对，过时则作废重算）。
+3. 推进计划状态，执行程序规则：到期、时间止损、失效条件、复核条件、无计划持仓。
+4. 算出应有订单与现有订单对比：先挂止损，再撤多余，最后补缺（撤了旧入场单的这一轮不挂新入场单）。
+"""
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from alphapulse.ai.tools import LOCAL_TZ
+from alphapulse.core.crypto import decrypt
+from alphapulse.core.db import SessionLocal
+from alphapulse.market.binance import BinanceClient
+from alphapulse.models import TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol
+from alphapulse.trading.client import OWN_CLIENT_ID, DemoClient, ExchangeError, mark_price, symbol_rules
+from alphapulse.trading.plan import MAX_HOLD_HOURS, OrderSpec, SymbolRules, condition_met, desired_orders, diff_orders, round_to_step
+from alphapulse.trading.schedule import request_analysis
+
+logger = logging.getLogger("alphapulse.trading.executor")
+
+EXECUTOR_SECONDS = 2
+ACTIVE_ORDER_STATUSES = ("submitting", "new", "partially_filled")
+LIVE_PLAN_STATUSES = ("pending", "open", "closing")
+# 刚下的单可能短时间查不到（C1 实测约 0.25s）
+SUBMIT_GRACE_SECONDS = 10
+# 止损挂单连续失败（可重试错误）达到次数后撤单平仓
+STOP_FAILURE_LIMIT = 3
+# 同一错误的时间线事件间隔，避免刷屏
+ERROR_EVENT_SECONDS = 600
+ORDER_STATUSES = {"NEW": "new", "PARTIALLY_FILLED": "partially_filled", "FILLED": "filled", "CANCELED": "canceled",
+                  "EXPIRED": "expired", "EXPIRED_IN_MATCH": "expired", "REJECTED": "rejected"}
+PURPOSE_LABELS = {"entry": "入场", "tp": "止盈", "stop": "止损", "close": "市价平仓"}
+CLOSE_REASON_LABELS = {"stop": "止损", "take_profit": "止盈", "time": "到达持仓时限", "invalidation": "短线失效条件触发",
+                       "exit": "模型决定结束", "external": "外部平仓（如在网页手动平仓）", "stop_failed": "止损挂单失败",
+                       "close": "市价平仓"}
+
+
+@dataclass
+class Position:
+    side: str  # long / short
+    qty: Decimal
+    entry: Decimal
+    break_even: Decimal
+    mark: Decimal
+
+
+def fmt(value: Decimal) -> str:
+    return format(value, "f")
+
+
+def utcnow() -> datetime:
+    return datetime.now(UTC)
+
+
+class Executor:
+    def __init__(self) -> None:
+        self.client: DemoClient | None = None
+        self.key_ciphertext: bytes | None = None
+        self.market = BinanceClient()
+        # 已设置为逐仓 3 倍的币；单向持仓是账户级，只设一次
+        self.configured: set[str] = set()
+        self.one_way_checked = False
+        self.stop_failures: dict[str, int] = {}
+        self.error_events: dict[str, float] = {}
+        # (symbol, 周期) -> (下一根收盘时间 ms, 最近收盘价)
+        self.kline_cache: dict[tuple[str, str], tuple[int, Decimal]] = {}
+        # 已请模型接管的无计划持仓：symbol -> 持仓均价，同一笔持仓只请求一次
+        self.takeover_requested: dict[str, Decimal] = {}
+
+    # ---------- 基础 ----------
+
+    async def ensure_client(self, account: TradingAccount | None) -> DemoClient | None:
+        ciphertext = account.api_key_encrypted if account else None
+        if ciphertext != self.key_ciphertext:
+            if self.client is not None:
+                await self.client.http.aclose()
+            self.client = None
+            self.configured.clear()
+            self.one_way_checked = False
+            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 report(self, session: AsyncSession, symbol: str | None, message: str, level: str = "error", analysis_id: int | None = None) -> None:
+        """同一条错误十分钟内只记一次"""
+        key = f"{symbol}:{message}"
+        if time.monotonic() - self.error_events.get(key, -ERROR_EVENT_SECONDS) < ERROR_EVENT_SECONDS:
+            return
+        self.error_events[key] = time.monotonic()
+        logger.warning("%s %s", symbol, message)
+        session.add(TradingEvent(symbol=symbol, analysis_id=analysis_id, kind="executor_error", level=level, message=message))
+
+    async def last_close(self, symbol: str, interval: str) -> Decimal:
+        """实盘该周期最近一根已收盘 K 线的收盘价；到下一根收盘前用缓存"""
+        now_ms = int(time.time() * 1000)
+        cached = self.kline_cache.get((symbol, interval))
+        if cached and now_ms < cached[0]:
+            return cached[1]
+        rows = await self.market.get("um", "klines", {"symbol": symbol, "interval": interval, "limit": 2})
+        closed = rows[-1] if rows[-1][6] < now_ms else rows[-2]
+        # 下一根收盘时间 = 当前未收盘 K 线的 close_time
+        next_close = rows[-1][6] + 1 if rows[-1][6] >= now_ms else now_ms
+        self.kline_cache[(symbol, interval)] = (next_close, Decimal(closed[4]))
+        return Decimal(closed[4])
+
+    async def condition_met(self, symbol: str, condition: dict, mark: Decimal) -> tuple[bool, Decimal]:
+        price = mark if condition["timeframe"] == "mark" else await self.last_close(symbol, condition["timeframe"])
+        return condition_met(condition["direction"], Decimal(condition["price"]), price), price
+
+    async def ensure_symbol_setup(self, client: DemoClient, symbol: str) -> None:
+        """单向持仓、逐仓、3 倍杠杆。先查再改：有挂单时即使不需要修改，写操作也可能被拒（-4067）"""
+        if not self.one_way_checked:
+            if (await client.request("GET", "/fapi/v1/positionSide/dual"))["dualSidePosition"]:
+                await client.request("POST", "/fapi/v1/positionSide/dual", {"dualSidePosition": False})
+            self.one_way_checked = True
+        if symbol in self.configured:
+            return
+        config = (await client.request("GET", "/fapi/v1/symbolConfig", {"symbol": symbol}))[0]
+        if config["marginType"] != "ISOLATED":
+            await client.request("POST", "/fapi/v1/marginType", {"symbol": symbol, "marginType": "ISOLATED"})
+        if config["leverage"] != 3:
+            await client.request("POST", "/fapi/v1/leverage", {"symbol": symbol, "leverage": 3})
+        self.configured.add(symbol)
+
+    # ---------- 订单 ----------
+
+    async def place(self, session: AsyncSession, client: DemoClient, symbol: str, plan_id: int | None, purpose: str, spec: OrderSpec) -> ExchangeError | None:
+        """先写流水并提交，再用 ap{id} 下单。结果不明（网络、5xx）时保持 submitting，下一轮按 id 查询，绝不换 id 重发"""
+        row = TradingOrder(
+            analysis_id=plan_id, symbol=symbol, purpose=purpose, leg=spec.leg, is_algo=spec.type == "STOP_MARKET",
+            side=spec.side, type=spec.type, price=spec.price, trigger_price=spec.trigger_price, quantity=spec.quantity,
+            reduce_only=purpose != "entry", status="submitting",
+        )
+        session.add(row)
+        await session.commit()
+        client_id = f"ap{row.id}"
+        try:
+            if row.is_algo:
+                result = await client.request("POST", "/fapi/v1/algoOrder", {
+                    "algoType": "CONDITIONAL", "symbol": symbol, "side": spec.side, "type": "STOP_MARKET",
+                    "triggerPrice": fmt(spec.trigger_price), "quantity": fmt(spec.quantity), "reduceOnly": True,
+                    "workingType": "MARK_PRICE", "clientAlgoId": client_id,
+                })
+                row.exchange_id, row.status = str(result["algoId"]), "new"
+            else:
+                params = {"symbol": symbol, "side": spec.side, "type": spec.type, "quantity": fmt(spec.quantity),
+                          "reduceOnly": row.reduce_only or None, "newClientOrderId": client_id}
+                if spec.type == "LIMIT":
+                    params |= {"price": fmt(spec.price), "timeInForce": "GTC"}
+                result = await client.request("POST", "/fapi/v1/order", params)
+                row.exchange_id = str(result["orderId"])
+                row.status = ORDER_STATUSES.get(result["status"], result["status"].lower())
+        except ExchangeError as e:
+            row.error = str(e)
+            if not (e.retryable or e.status is None):
+                row.status = "rejected"
+            await session.commit()
+            return e
+        await session.commit()
+        return None
+
+    async def cancel(self, client: DemoClient, row: TradingOrder) -> None:
+        await self.cancel_client_id(client, row.symbol, f"ap{row.id}", row.is_algo)
+
+    async def cancel_client_id(self, client: DemoClient, symbol: str, client_id: str, is_algo: bool) -> None:
+        """撤单后不直接改状态：可能恰好已成交，交给下一轮同步确认"""
+        try:
+            if is_algo:
+                await client.request("DELETE", "/fapi/v1/algoOrder", {"clientAlgoId": client_id})
+            else:
+                await client.request("DELETE", "/fapi/v1/order", {"symbol": symbol, "origClientOrderId": client_id})
+        except ExchangeError as e:
+            # -2011 / -2013：单已不在，等同步
+            if e.code not in (-2011, -2013):
+                raise
+
+    async def sync_order(self, session: AsyncSession, client: DemoClient, row: TradingOrder, open_orders: dict, open_algos: dict) -> None:
+        client_id = f"ap{row.id}"
+        before = row.filled_qty
+        try:
+            if row.is_algo:
+                if client_id in open_algos:
+                    row.exchange_id, row.status = str(open_algos[client_id]["algoId"]), "new"
+                    return
+                algo = await client.request("GET", "/fapi/v1/algoOrder", {"clientAlgoId": client_id})
+                if algo["algoStatus"] in ("NEW", "TRIGGERING"):
+                    row.status = "new"
+                    return
+                row.exchange_id = str(algo["algoId"])
+                if algo["algoStatus"] == "FINISHED" and algo.get("actualOrderId"):
+                    actual = await client.request("GET", "/fapi/v1/order", {"symbol": row.symbol, "orderId": algo["actualOrderId"]})
+                    row.filled_qty = Decimal(actual["executedQty"])
+                    row.avg_price = Decimal(actual["avgPrice"]) or None
+                    row.status = "filled" if actual["status"] == "FILLED" else ORDER_STATUSES.get(actual["status"], "finished")
+                    if row.status == "new":
+                        # 触发后的市价单还没成交完，下一轮再查
+                        return
+                else:
+                    row.status = algo["algoStatus"].lower()
+            else:
+                order = open_orders.get(client_id) or await client.request("GET", "/fapi/v1/order", {"symbol": row.symbol, "origClientOrderId": client_id})
+                row.exchange_id = str(order["orderId"])
+                row.status = ORDER_STATUSES.get(order["status"], order["status"].lower())
+                row.filled_qty = Decimal(order["executedQty"])
+                row.avg_price = Decimal(order["avgPrice"]) or None
+        except ExchangeError as e:
+            if e.code != -2013:
+                raise
+            if row.status == "submitting" and utcnow() - row.created_at < timedelta(seconds=SUBMIT_GRACE_SECONDS):
+                return
+            row.status, row.error = "failed", "交易所查无此单"
+            return
+        if row.filled_qty > before:
+            delta = row.filled_qty - before
+            leg = f"第 {row.leg + 1} 档" if row.purpose == "entry" else (f" {row.leg + 1} " if row.purpose == "tp" else "")
+            session.add(TradingEvent(
+                symbol=row.symbol, analysis_id=row.analysis_id, kind=f"{row.purpose}_filled",
+                level="warn" if row.purpose == "stop" else "info",
+                message=f"{PURPOSE_LABELS[row.purpose]}{leg}成交 {fmt(delta)} @ {row.avg_price}",
+                data={"order_id": row.id, "quantity": fmt(delta)},
+            ))
+            if row.purpose == "tp":
+                await request_analysis(session, row.symbol, f"止盈 {row.leg + 1} 成交")
+
+    # ---------- 计划 ----------
+
+    async def adopt(self, session: AsyncSession, client: DemoClient, symbol: TradingSymbol, current: TradingAnalysis | None,
+                    position: Position | None) -> TradingAnalysis | None:
+        """采纳最新的提议；较早的提议作废。返回采纳后的当前计划"""
+        proposals = (await session.scalars(
+            select(TradingAnalysis).where(TradingAnalysis.symbol == symbol.symbol, TradingAnalysis.plan_status == "proposed")
+            .order_by(TradingAnalysis.id.desc())
+        )).all()
+        if not proposals:
+            return current
+        proposal, older = proposals[0], proposals[1:]
+        for p in older:
+            p.plan_status = "stale"
+        now = utcnow()
+        if proposal.action == "exit":
+            proposal.plan_status = "applied"
+            if current is not None:
+                current.plan_status, current.close_reason = "closing", "exit"
+                session.add(TradingEvent(symbol=symbol.symbol, analysis_id=current.id, kind="plan_exit", level="warn",
+                                         message=f"按分析 #{proposal.id} 结束计划：撤销入场单，有持仓则市价平仓"))
+            elif position is not None:
+                session.add(TradingEvent(symbol=symbol.symbol, analysis_id=proposal.id, kind="plan_exit", level="warn",
+                                         message=f"按分析 #{proposal.id} 市价平掉无计划的持仓 {fmt(position.qty)}"))
+                close_side = "SELL" if position.side == "long" else "BUY"
+                await self.place(session, client, symbol.symbol, proposal.id, "close", OrderSpec("close", None, close_side, "MARKET", None, None, position.qty))
+            return current
+        qty = position.qty if position else Decimal(0)
+        expected = Decimal(proposal.sizing["position_qty"])
+        if qty != expected or (current is not None and current.plan_status == "closing"):
+            proposal.plan_status = "stale"
+            reason = "计划正在平仓" if current is not None and current.plan_status == "closing" else f"持仓已从 {fmt(expected)} 变为 {fmt(qty)}"
+            session.add(TradingEvent(symbol=symbol.symbol, analysis_id=proposal.id, kind="plan_stale", level="warn",
+                                     message=f"计划未采纳：{reason}"))
+            await request_analysis(session, symbol.symbol, "持仓变化，计划作废重算")
+            return current
+        if current is not None:
+            current.plan_status, current.close_reason, current.closed_at = "replaced", "replaced", now
+        if qty > 0:
+            proposal.plan_status = "open"
+            proposal.first_fill_at = current.first_fill_at if current and current.first_fill_at else now
+            # 持仓时限沿用原计划，不延长；接管无计划持仓时从现在起算
+            proposal.close_deadline = current.close_deadline if current and current.close_deadline else now + timedelta(hours=MAX_HOLD_HOURS[symbol.angle])
+        else:
+            proposal.plan_status = "pending"
+            proposal.valid_until = now + timedelta(hours=float(proposal.decision["entry_valid_hours"]))
+        session.add(TradingEvent(symbol=symbol.symbol, analysis_id=proposal.id, kind="plan_adopted",
+                                 message=f"采纳计划 #{proposal.id}" + (f"，替换 #{current.id}" if current else "")))
+        return proposal
+
+    async def realized(self, client: DemoClient, symbol: str, since: datetime | None) -> dict | None:
+        if since is None:
+            return None
+        trades = await client.request("GET", "/fapi/v1/userTrades", {"symbol": symbol, "startTime": int(since.timestamp() * 1000) - 1000, "limit": 1000})
+        pnl = sum((Decimal(t["realizedPnl"]) for t in trades), Decimal(0))
+        fee = sum((Decimal(t["commission"]) for t in trades if t["commissionAsset"] == "USDT"), Decimal(0))
+        return {"realized_pnl": fmt(pnl), "commission": fmt(fee), "net": fmt(pnl - fee)}
+
+    async def finish(self, session: AsyncSession, client: DemoClient, plan: TradingAnalysis, rows: list[TradingOrder]) -> None:
+        """持仓归零：计划结束，记录原因和盈亏，排队分析"""
+        if plan.close_reason is None or plan.plan_status == "open":
+            filled = sorted((r for r in rows if r.filled_qty > 0 and r.purpose in ("stop", "tp", "close")), key=lambda r: r.updated_at)
+            last = filled[-1].purpose if filled else None
+            plan.close_reason = {"stop": "stop", "tp": "take_profit", "close": plan.close_reason or "close"}.get(last, "external")
+        plan.plan_status = "closed" if plan.first_fill_at else "canceled"
+        plan.closed_at = utcnow()
+        plan.result = await self.realized(client, plan.symbol, plan.first_fill_at)
+        net = f"，净盈亏 {plan.result['net']} USDT（含手续费 {plan.result['commission']}）" if plan.result else ""
+        label = CLOSE_REASON_LABELS.get(plan.close_reason, plan.close_reason)
+        session.add(TradingEvent(symbol=plan.symbol, analysis_id=plan.id, kind="plan_closed", level="warn" if plan.close_reason in ("stop", "stop_failed", "external") else "info",
+                                 message=f"计划 #{plan.id} 结束：{label}{net}", data=plan.result))
+        await request_analysis(session, plan.symbol, f"平仓（{label}）")
+
+    async def start_closing(self, session: AsyncSession, plan: TradingAnalysis, reason: str, detail: str = "") -> None:
+        plan.plan_status, plan.close_reason = "closing", reason
+        session.add(TradingEvent(symbol=plan.symbol, analysis_id=plan.id, kind="plan_closing", level="warn",
+                                 message=f"{CLOSE_REASON_LABELS[reason]}{detail}：撤销入场单并市价平仓"))
+
+    async def review_once(self, session: AsyncSession, plan_id: int, symbol: str, kind: str, index: int, message: str, reason: str) -> None:
+        """失效复核、reevaluate 每个条件只触发一次：以时间线事件去重"""
+        exists = await session.scalar(select(func.count()).where(
+            TradingEvent.analysis_id == plan_id, TradingEvent.kind == kind, TradingEvent.data["index"].as_integer() == index,
+        ))
+        if exists:
+            return
+        session.add(TradingEvent(symbol=symbol, analysis_id=plan_id, kind=kind, level="warn", message=message, data={"index": index}))
+        await request_analysis(session, symbol, reason)
+
+    # ---------- 对账 ----------
+
+    async def reconcile(self, session: AsyncSession, client: DemoClient, symbol: TradingSymbol, running: bool, position: Position | None) -> None:
+        name = symbol.symbol
+        rules = await symbol_rules(name)
+        open_orders = {o["clientOrderId"]: o for o in await client.request("GET", "/fapi/v1/openOrders", {"symbol": name})}
+        open_algos = {o["clientAlgoId"]: o for o in await client.request("GET", "/fapi/v1/openAlgoOrders", {"symbol": name})}
+
+        rows = list((await session.scalars(
+            select(TradingOrder).where(TradingOrder.symbol == name, TradingOrder.status.in_(ACTIVE_ORDER_STATUSES))
+        )).all())
+        for row in rows:
+            await self.sync_order(session, client, row, open_orders, open_algos)
+        await session.commit()
+
+        # 交易所上本程序的单，但流水里不是活动状态：孤儿单，撤掉（用户手动下的单不动）
+        active_ids = {f"ap{r.id}" for r in rows if r.status in ACTIVE_ORDER_STATUSES}
+        for client_id, order in [*open_orders.items(), *open_algos.items()]:
+            if OWN_CLIENT_ID.match(client_id) and client_id not in active_ids:
+                await self.cancel_client_id(client, name, client_id, client_id in open_algos)
+                session.add(TradingEvent(symbol=name, kind="orphan_canceled", level="warn", message=f"撤销孤儿单 {client_id}（{order.get('type') or order.get('orderType')}）"))
+
+        plan = await session.scalar(
+            select(TradingAnalysis).where(TradingAnalysis.symbol == name, TradingAnalysis.plan_status.in_(LIVE_PLAN_STATUSES))
+            .order_by(TradingAnalysis.id.desc()).limit(1)
+        )
+        plan = await self.adopt(session, client, symbol, plan, position)
+        await session.commit()
+
+        # 不属于当前计划的活动订单（被替换、已结束的计划）：入场和止盈先撤；止损等新止损挂上后再撤，不留无保护窗口
+        stale = [r for r in rows if r.status in ACTIVE_ORDER_STATUSES and r.purpose != "close" and (plan is None or r.analysis_id != plan.id)]
+        for row in stale:
+            if row.purpose != "stop":
+                await self.cancel(client, row)
+        stale_entry_canceled = any(r.purpose == "entry" for r in stale)
+
+        if plan is None:
+            for row in stale:
+                if row.purpose == "stop":
+                    await self.cancel(client, row)
+            if position is None:
+                self.takeover_requested.pop(name, None)
+            elif running:
+                await self.unmanaged(session, client, name, position, open_algos)
+            if running:
+                await self.check_reevaluate(session, name, position.mark if position else None)
+            await session.commit()
+            return
+
+        direction = plan.sizing["direction"]
+        if position is not None and position.side != direction:
+            await self.report(session, name, f"持仓方向（{position.side}）与计划 #{plan.id} 方向（{direction}）不一致，暂停该币的下单", analysis_id=plan.id)
+            await session.commit()
+            return
+        qty = position.qty if position else Decimal(0)
+        now = utcnow()
+        plan_rows = list((await session.scalars(select(TradingOrder).where(TradingOrder.analysis_id == plan.id))).all())
+
+        if plan.plan_status == "pending":
+            if qty > 0:
+                plan.plan_status, plan.first_fill_at = "open", now
+                plan.close_deadline = now + timedelta(hours=float(plan.decision["max_hold_hours"]))
+                session.add(TradingEvent(symbol=name, analysis_id=plan.id, kind="plan_open",
+                                         message=f"首档成交，计划 #{plan.id} 进入持仓；最晚 {plan.close_deadline.astimezone(LOCAL_TZ).strftime('%m-%d %H:%M')} 平仓"))
+                await request_analysis(session, name, "首档成交")
+            elif now >= plan.valid_until or not running:
+                plan.plan_status, plan.closed_at = "canceled", now
+                plan.close_reason = "expired" if running else "stopped"
+                session.add(TradingEvent(symbol=name, analysis_id=plan.id, kind="plan_canceled",
+                                         message=f"计划 #{plan.id} 撤销：{'入场单到期未成交' if running else '已停止交易，撤销未成交的入场单'}"))
+                if running:
+                    await request_analysis(session, name, "计划到期未成交")
+        elif plan.plan_status == "open":
+            if qty == 0:
+                await self.finish(session, client, plan, plan_rows)
+            elif now >= plan.close_deadline:
+                await self.start_closing(session, plan, "time")
+            else:
+                await self.check_invalidation(session, plan, position)
+        elif plan.plan_status == "closing" and qty == 0:
+            await self.finish(session, client, plan, plan_rows)
+        await session.commit()
+
+        if plan.plan_status in LIVE_PLAN_STATUSES:
+            await self.sync_orders(session, client, plan, plan_rows, position, running, rules, stale_entry_canceled)
+        else:
+            # 计划已结束：只减仓止损不会被交易所自动撤销，必须主动撤，以免作用到之后的新仓位
+            for row in plan_rows:
+                if row.status in ACTIVE_ORDER_STATUSES and row.purpose != "close":
+                    await self.cancel(client, row)
+        for row in stale:
+            if row.purpose == "stop":
+                await self.cancel(client, row)
+        if running and plan.plan_status != "closing":
+            await self.check_reevaluate(session, name, position.mark if position else None)
+        await session.commit()
+
+    async def check_invalidation(self, session: AsyncSession, plan: TradingAnalysis, position: Position) -> None:
+        condition = plan.decision["invalidation"]
+        met, price = await self.condition_met(plan.symbol, condition, position.mark)
+        if not met:
+            return
+        detail = f"（{condition['timeframe']} 收盘 {fmt(price)} {'高于' if condition['direction'] == 'above' else '低于'} {condition['price']}）"
+        if plan.angle == "short":
+            await self.start_closing(session, plan, "invalidation", detail)
+        else:
+            await self.review_once(session, plan.id, plan.symbol, "invalidation_review", 0,
+                                   f"长线失效条件触发{detail}，请模型复核", f"失效条件触发{detail}")
+
+    async def check_reevaluate(self, session: AsyncSession, symbol: str, mark: Decimal | None) -> None:
+        """最近一次交易中分析给出的 reevaluate 条件：满足任一个就排队分析"""
+        latest = await session.scalar(
+            select(TradingAnalysis).where(TradingAnalysis.symbol == symbol, TradingAnalysis.status == "done", TradingAnalysis.dry_run.is_(False))
+            .order_by(TradingAnalysis.id.desc()).limit(1)
+        )
+        if latest is None or not latest.decision["reevaluate"]:
+            return
+        if mark is None and any(c["timeframe"] == "mark" for c in latest.decision["reevaluate"]):
+            mark = await mark_price(symbol)
+        for i, condition in enumerate(latest.decision["reevaluate"]):
+            met, price = await self.condition_met(symbol, condition, mark)
+            if met:
+                text = f"{condition['timeframe']} {'高于' if condition['direction'] == 'above' else '低于'} {condition['price']}（现 {fmt(price)}）"
+                await self.review_once(session, latest.id, symbol, "reevaluate_triggered", i, f"复核条件满足：{text}", f"复核条件满足：{text}")
+
+    async def unmanaged(self, session: AsyncSession, client: DemoClient, symbol: str, position: Position, open_algos: dict) -> None:
+        """有持仓但没有计划：有止损就请模型接管，没有止损就市价平仓"""
+        close_side = "SELL" if position.side == "long" else "BUY"
+        has_stop = any(o["side"] == close_side and o["orderType"].startswith("STOP") for o in open_algos.values())
+        if has_stop:
+            if self.takeover_requested.get(symbol) != position.entry:
+                self.takeover_requested[symbol] = position.entry
+                session.add(TradingEvent(symbol=symbol, kind="unmanaged_position", level="warn",
+                                         message=f"{symbol} 有持仓 {fmt(position.qty)} 和止损但没有计划，请模型接管"))
+                await request_analysis(session, symbol, "有持仓但没有计划，请接管")
+            return
+        closing = await session.scalar(select(func.count()).where(
+            TradingOrder.symbol == symbol, TradingOrder.purpose == "close", TradingOrder.status.in_(ACTIVE_ORDER_STATUSES),
+        ))
+        if closing:
+            return
+        session.add(TradingEvent(symbol=symbol, kind="unmanaged_closed", level="error",
+                                 message=f"{symbol} 有持仓 {fmt(position.qty)} 但没有计划也没有止损，市价平仓"))
+        await self.place(session, client, symbol, None, "close", OrderSpec("close", None, close_side, "MARKET", None, None, position.qty))
+
+    async def sync_orders(self, session: AsyncSession, client: DemoClient, plan: TradingAnalysis, plan_rows: list[TradingOrder],
+                          position: Position | None, running: bool, rules: SymbolRules, block_entries: bool) -> None:
+        sizing = plan.sizing
+        qty = position.qty if position else Decimal(0)
+        n_entries, n_tps = len(sizing["entries"]), len(sizing["take_profits"])
+        entry_filled = [sum((r.filled_qty for r in plan_rows if r.purpose == "entry" and r.leg == i), Decimal(0)) for i in range(n_entries)]
+        tp_filled = [sum((r.filled_qty for r in plan_rows if r.purpose == "tp" and r.leg == j), Decimal(0)) for j in range(n_tps)]
+        tp1_hit = tp_filled[0] > 0
+        stop = Decimal(sizing["stop"])
+        if tp1_hit and position is not None:
+            # 止盈 1 后止损移到保本价：做多向上取整、做空向下取整，保证不亏；只收紧不放宽
+            long = sizing["direction"] == "long"
+            break_even = round_to_step(position.break_even, rules.tick_size, ROUND_CEILING if long else ROUND_FLOOR)
+            stop = max(stop, break_even) if long else min(stop, break_even)
+
+        active = [r for r in plan_rows if r.status in ACTIVE_ORDER_STATUSES and r.purpose in ("entry", "stop", "tp")]
+        long = sizing["direction"] == "long"
+        if plan.plan_status == "closing":
+            # 平仓中：撤入场和止盈，保留已有止损直到市价平仓成交（止损可能正因越过现价挂不上，不再补挂）
+            for row in active:
+                if row.purpose != "stop":
+                    await self.cancel(client, row)
+            if qty > 0 and not any(r.purpose == "close" and r.status in ACTIVE_ORDER_STATUSES for r in plan_rows):
+                close_side = "SELL" if long else "BUY"
+                await self.place(session, client, plan.symbol, plan.id, "close", OrderSpec("close", None, close_side, "MARKET", None, None, qty))
+            return
+
+        entries_open = running and utcnow() < (plan.valid_until or utcnow()) and not tp1_hit
+        desired = desired_orders(sizing, entries_open=entries_open, entry_filled=entry_filled, tp_filled=tp_filled,
+                                 position_qty=qty, stop_price=stop, step_size=rules.step_size)
+        actual = [(r, OrderSpec(r.purpose, r.leg, r.side, r.type, r.price, r.trigger_price, r.quantity - r.filled_qty)) for r in active]
+        to_cancel, to_place = diff_orders(desired, actual)
+
+        entries = [d for d in to_place if d.purpose == "entry"]
+        if entries:
+            # 保证金模式和杠杆在有挂单时不能改，必须在挂任何单之前设好
+            await self.ensure_symbol_setup(client, plan.symbol)
+        # 1. 止损先挂新的再撤旧的（只减仓止损可以并存），任何时候都不留无保护窗口
+        for spec in [d for d in to_place if d.purpose == "stop"]:
+            error = await self.place(session, client, plan.symbol, plan.id, "stop", spec)
+            if error is not None:
+                await self.stop_failed(session, plan, error)
+                return
+        self.stop_failures.pop(plan.symbol, None)
+        # 2. 撤多余的单
+        canceled_entry = False
+        for row in to_cancel:
+            await self.cancel(client, row)
+            canceled_entry = canceled_entry or row.purpose == "entry"
+        # 3. 补止盈；入场单只在本轮没有撤旧入场单、且止损在场时挂
+        has_stop = any(r.purpose == "stop" for r in active if r not in to_cancel) or any(d.purpose == "stop" for d in to_place)
+        for spec in [d for d in to_place if d.purpose == "tp"]:
+            await self.place(session, client, plan.symbol, plan.id, "tp", spec)
+        if entries and has_stop and not canceled_entry and not block_entries:
+            for spec in entries:
+                await self.place(session, client, plan.symbol, plan.id, "entry", spec)
+
+    async def stop_failed(self, session: AsyncSession, plan: TradingAnalysis, error: ExchangeError) -> None:
+        """会立即触发（-2021）或不可重试的错误：撤单平仓；可重试的错误连续失败多次后同样处理"""
+        if error.retryable or error.status is None:
+            count = self.stop_failures[plan.symbol] = self.stop_failures.get(plan.symbol, 0) + 1
+            if count < STOP_FAILURE_LIMIT:
+                await self.report(session, plan.symbol, f"挂止损失败（第 {count} 次），下一轮重试：{error}", "warn", plan.id)
+                return
+        self.stop_failures.pop(plan.symbol, None)
+        reason = "止损价已越过现价" if error.code == -2021 else f"挂止损失败：{error}"
+        session.add(TradingEvent(symbol=plan.symbol, analysis_id=plan.id, kind="stop_failed", level="error", message=reason))
+        await self.start_closing(session, plan, "stop_failed")
+
+    # ---------- 主循环 ----------
+
+    async def tick(self) -> None:
+        async with SessionLocal() as session:
+            account = await session.get(TradingAccount, 1)
+            symbols = list((await session.scalars(select(TradingSymbol))).all())
+            client = await self.ensure_client(account)
+            if client is None or not symbols:
+                return
+            positions = {}
+            for p in await client.request("GET", "/fapi/v3/positionRisk"):
+                amount = Decimal(p["positionAmt"])
+                if amount != 0:
+                    positions[p["symbol"]] = Position(
+                        side="long" if amount > 0 else "short", qty=abs(amount), entry=Decimal(p["entryPrice"]),
+                        break_even=Decimal(p["breakEvenPrice"]), mark=Decimal(p["markPrice"]),
+                    )
+            for symbol in symbols:
+                try:
+                    await self.reconcile(session, client, symbol, account.running, positions.get(symbol.symbol))
+                except ExchangeError as e:
+                    await session.rollback()
+                    await self.report(session, symbol.symbol, f"对账失败：{e}")
+                    await session.commit()
+                except Exception as e:
+                    logger.exception("%s 对账出错", symbol.symbol)
+                    await session.rollback()
+                    await self.report(session, symbol.symbol, f"对账出错：{e!r}")
+                    await session.commit()
+
+
+async def executor_loop() -> None:
+    executor = Executor()
+    while True:
+        try:
+            await executor.tick()
+        except ExchangeError as e:
+            logger.warning("执行器同步失败：%s", e)
+        except Exception:
+            logger.exception("执行器出错，稍后重试")
+        await asyncio.sleep(EXECUTOR_SECONDS)

backend/src/alphapulse/trading/plan.py

@@ -4,6 +4,7 @@
 """
 
 from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
 from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal
 from typing import Annotated, Literal
 
@@ -32,6 +33,10 @@ TAKER_FEE_RATE = Decimal("0.0004")
 # 模拟盘 PERCENT_PRICE：买单价不高于标记价 × 1.05，卖单价不低于标记价 × 0.95
 PERCENT_PRICE_LIMIT = Decimal("0.05")
 RATIO_TOLERANCE = Decimal("0.001")
+# 例行分析周期（小时）：短线跟 1h、长线跟 4h 收盘
+ROUTINE_HOURS = {"short": 1, "long": 4}
+# 同一个币两次自动分析的最短间隔
+MIN_ANALYSIS_GAP = timedelta(minutes=10)
 
 Price = Annotated[Decimal, Field(gt=0)]
 
@@ -99,6 +104,11 @@ class PlanContext:
     rules: SymbolRules
     has_plan: bool
     position_side: Direction | None
+    position_qty: Decimal = Decimal(0)
+    # 持仓均价
+    position_entry: Decimal | None = None
+    # 当前生效的止损价（可能已移到保本价）；新计划只能收紧
+    current_stop: Decimal | None = None
 
 
 class PlanError(Exception):
@@ -142,14 +152,27 @@ def check_action(plan: TradePlan, ctx: PlanContext) -> list[str]:
 
 
 def check_plan_fields(plan: TradePlan, ctx: PlanContext) -> list[str]:
-    """action=plan 时的字段完整性、档数、比例、有效期"""
+    """action=plan 时的字段完整性、档数、比例、有效期。有持仓时只调整止损止盈等，不加仓"""
-    required = ("direction", "confidence", "entries", "stop", "take_profits", "entry_valid_hours", "max_hold_hours", "invalidation")
+    holding = ctx.position_side is not None
+    required = ("direction", "confidence", "stop", "take_profits", "invalidation")
+    if not holding:
+        required += ("entries", "entry_valid_hours", "max_hold_hours")
     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 "长线"
+    if holding:
+        if plan.entries:
+            errors.append("已有持仓，新计划不加仓：entries 须为空，只调整止损、止盈、失效条件和 reevaluate")
+        if plan.direction != ctx.position_side:
+            errors.append(f"已有{'多' if ctx.position_side == 'long' else '空'}头持仓，新计划方向必须一致；要反手请先用 exit")
+        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.invalidation.timeframe not in INVALIDATION_TIMEFRAMES[angle]:
+            errors.append(f"{label}失效条件的周期只能是 {'/'.join(INVALIDATION_TIMEFRAMES[angle])}")
+        return errors
     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)}")
@@ -163,8 +186,6 @@ def check_plan_fields(plan: TradePlan, ctx: PlanContext) -> list[str]:
         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
 
 
@@ -214,6 +235,8 @@ def size_plan(plan: TradePlan, ctx: PlanContext) -> dict:
     errors += check_plan_fields(plan, ctx)
     if errors:
         raise PlanError(errors)
+    if ctx.position_side is not None:
+        return size_managed(plan, ctx)
 
     angle, direction, rules = ctx.angle, plan.direction, ctx.rules
     # 向保守方向取整：做多的入场、止损、止盈都向下（入场更低、止损更远、止盈更近），做空都向上
@@ -282,4 +305,138 @@ def size_plan(plan: TradePlan, ctx: PlanContext) -> dict:
         "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"))),
+        # 执行器采纳前核对：持仓变了说明计划基于过时状态
+        "position_qty": "0",
+        "managed": False,
     }
+
+
+def size_managed(plan: TradePlan, ctx: PlanContext) -> dict:
+    """有持仓时的新计划：不加仓，止损只能收紧，止盈按当前持仓数量拆分"""
+    direction, rules = plan.direction, ctx.rules
+    rounding = ROUND_FLOOR if direction == "long" else ROUND_CEILING
+    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]
+    sign = 1 if direction == "long" else -1
+    mark, invalidation = ctx.mark_price, plan.invalidation.price
+
+    def below(a: Decimal, b: Decimal) -> bool:
+        return sign * (b - a) > 0
+
+    side = "做多" if direction == "long" else "做空"
+    errors = []
+    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) and below(invalidation, mark)):
+        errors.append(f"{side}时须满足 止损 {stop} → 失效价 {invalidation} → 现价 {mark} 的顺序")
+    if ctx.current_stop is not None and below(stop, ctx.current_stop):
+        errors.append(f"止损只能收紧：当前止损 {ctx.current_stop}，新止损 {stop} 更宽")
+    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("止盈价须按由近到远排列且不重复")
+    if errors:
+        raise PlanError(errors)
+
+    tp_ratios = TAKE_PROFIT_RATIOS[ctx.angle]
+    entry = ctx.position_entry
+    quantity = ctx.position_qty
+    rr = reward_risk(mark, stop, targets, tp_ratios) if ctx.angle == "long" else reward_risk(mark, stop, targets[:1], (Decimal(1),))
+    loss = quantity * (sign * (entry - stop) + (entry + stop) * TAKER_FEE_RATE)
+    return {
+        "direction": direction,
+        "entries": [],
+        "stop": str(stop),
+        "take_profits": [
+            {"price": str(t), "ratio": str(r), "quantity": str(q)}
+            for t, r, q in zip(targets, tp_ratios, split_take_profits(quantity, tp_ratios, rules.step_size), strict=True)
+        ],
+        "average_price": str(entry),
+        "quantity": str(quantity),
+        "notional": str((entry * quantity).quantize(Decimal("0.01"))),
+        "capital": str(ctx.capital),
+        "leverage": str((entry * quantity / ctx.capital).quantize(Decimal("0.01"))),
+        "risk_amount": "0",
+        # 止损已越过成本价时为负，表示锁定盈利
+        "risk_at_stop": str(loss.quantize(Decimal("0.01"))),
+        "risk_pct": str((loss / ctx.capital * 100).quantize(Decimal("0.01"))),
+        "capped_by_leverage": False,
+        # 按现价计算
+        "reward_risk": str(rr.quantize(Decimal("0.01"))),
+        "reward_risk_first_only": str(rr.quantize(Decimal("0.01"))),
+        "position_qty": str(quantity),
+        "managed": True,
+    }
+
+
+# ---------- 执行：应有订单 ----------
+
+@dataclass(frozen=True)
+class OrderSpec:
+    """一张应有（或现有）的订单；quantity 为剩余未成交数量"""
+
+    purpose: Literal["entry", "stop", "tp", "close"]
+    leg: int | None
+    side: Literal["BUY", "SELL"]
+    type: Literal["LIMIT", "STOP_MARKET", "MARKET"]
+    price: Decimal | None
+    trigger_price: Decimal | None
+    quantity: Decimal
+
+
+def desired_orders(
+    sizing: dict, *, entries_open: bool, entry_filled: list[Decimal], tp_filled: list[Decimal],
+    position_qty: Decimal, stop_price: Decimal, step_size: Decimal,
+) -> list[OrderSpec]:
+    """计划在当前成交状态下应挂的单。
+
+    - 入场：entries_open 时各档剩余数量。
+    - 止损：有持仓或有入场单时一张只减仓止损，数量取计划总量与持仓的较大者（数量大于持仓时交易所只平实际持仓）。
+    - 止盈：有持仓时按累计成交数量拆分，减去各档已成交。
+    """
+    long = sizing["direction"] == "long"
+    open_side, close_side = ("BUY", "SELL") if long else ("SELL", "BUY")
+    orders = []
+    if entries_open:
+        for i, e in enumerate(sizing["entries"]):
+            remaining = Decimal(e["quantity"]) - entry_filled[i]
+            if remaining > 0:
+                orders.append(OrderSpec("entry", i, open_side, "LIMIT", Decimal(e["price"]), None, remaining))
+    if position_qty > 0 or orders:
+        quantity = max(Decimal(sizing["quantity"]), position_qty)
+        orders.append(OrderSpec("stop", None, close_side, "STOP_MARKET", None, stop_price, quantity))
+    if position_qty > 0:
+        filled_total = Decimal(sizing["quantity"]) if sizing["managed"] else sum(entry_filled, Decimal(0))
+        ratios = tuple(Decimal(t["ratio"]) for t in sizing["take_profits"])
+        for j, (t, target) in enumerate(zip(sizing["take_profits"], split_take_profits(filled_total, ratios, step_size), strict=True)):
+            remaining = target - tp_filled[j]
+            if remaining > 0:
+                orders.append(OrderSpec("tp", j, close_side, "LIMIT", Decimal(t["price"]), None, remaining))
+    return orders
+
+
+def diff_orders[K](desired: list[OrderSpec], actual: list[tuple[K, OrderSpec]]) -> tuple[list[K], list[OrderSpec]]:
+    """对比应有与现有订单：返回（要撤销的现有订单，要新挂的订单）。价格、数量完全一致才算同一张"""
+    missing = list(desired)
+    cancel = []
+    for key, spec in actual:
+        if spec in missing:
+            missing.remove(spec)
+        else:
+            cancel.append(key)
+    return cancel, missing
+
+
+def condition_met(direction: Literal["above", "below"], threshold: Decimal, price: Decimal) -> bool:
+    return price > threshold if direction == "above" else price < threshold
+
+
+def next_routine_at(last_started: datetime, angle: Angle) -> datetime:
+    """例行分析时间：上次分析后约 3/4 个周期之后的第一个 K 线收盘，再加 1 分钟等收盘数据就绪。
+    任何一次分析（包括触发式）都会让例行分析重新计时"""
+    period = timedelta(hours=ROUTINE_HOURS[angle])
+    # 币安 K 线按 UTC 对齐
+    earliest = (last_started + period * 3 / 4).astimezone(UTC)
+    day = earliest.replace(hour=0, minute=0, second=0, microsecond=0)
+    boundary = day + period * -(-(earliest - day) // period)
+    return boundary + timedelta(minutes=1)

backend/src/alphapulse/trading/schedule.py

@@ -0,0 +1,68 @@
+"""自动分析的排队：触发式（成交、止盈、平仓、到期、条件满足）和例行。只在交易中排队，手动分析走 API"""
+
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from alphapulse.ai.tools import LOCAL_TZ
+from alphapulse.core.db import SessionLocal
+from alphapulse.models import TradingAccount, TradingAnalysis, TradingEvent, TradingSymbol
+from alphapulse.trading.plan import DAILY_ANALYSIS_LIMIT, MIN_ANALYSIS_GAP, next_routine_at
+
+# 自动分析失败后的重试：间隔、连续失败上限
+RETRY_AFTER = timedelta(minutes=5)
+MAX_CONSECUTIVE_FAILURES = 3
+
+
+async def daily_limit_reached(session: AsyncSession) -> bool:
+    today = datetime.now(LOCAL_TZ).replace(hour=0, minute=0, second=0, microsecond=0)
+    count = await session.scalar(select(func.count()).where(TradingAnalysis.started_at >= today))
+    if count < DAILY_ANALYSIS_LIMIT:
+        return False
+    noticed = await session.scalar(select(func.count()).where(TradingEvent.kind == "daily_limit", TradingEvent.created_at >= today))
+    if not noticed:
+        session.add(TradingEvent(kind="daily_limit", level="warn",
+                                 message=f"今日自动分析已达 {DAILY_ANALYSIS_LIMIT} 次上限，之后只执行程序规则（止损、止盈、到期平仓等）"))
+    return True
+
+
+async def request_analysis(session: AsyncSession, symbol: str, reason: str) -> None:
+    """排队一次自动分析：多个原因合并；距上次分析不足最短间隔时顺延。调用方负责 commit"""
+    account = await session.get(TradingAccount, 1)
+    target = await session.get(TradingSymbol, symbol)
+    if account is None or not account.running or target is None or await daily_limit_reached(session):
+        return
+    now = datetime.now(UTC)
+    last_started = await session.scalar(select(func.max(TradingAnalysis.started_at)).where(TradingAnalysis.symbol == symbol))
+    at = max(now, last_started + MIN_ANALYSIS_GAP) if last_started else now
+    target.analysis_requested_at = min(target.analysis_requested_at, at) if target.analysis_requested_at else at
+    if reason not in target.analysis_reasons:
+        # JSONB 列要整体赋值才会被识别为修改
+        target.analysis_reasons = [*target.analysis_reasons, reason]
+
+
+async def routine_requests() -> None:
+    """交易中：到例行时间的币排队例行分析；自动分析失败的隔一段时间重试"""
+    async with SessionLocal() as session:
+        account = await session.get(TradingAccount, 1)
+        if account is None or not account.running:
+            return
+        now = datetime.now(UTC)
+        for target in (await session.scalars(select(TradingSymbol))).all():
+            if target.analysis_requested_at is not None:
+                continue
+            recent = (await session.scalars(
+                select(TradingAnalysis).where(TradingAnalysis.symbol == target.symbol)
+                .order_by(TradingAnalysis.id.desc()).limit(MAX_CONSECUTIVE_FAILURES)
+            )).all()
+            if recent and recent[0].status == "running":
+                continue
+            if not recent or now >= next_routine_at(recent[0].started_at, target.angle):
+                await request_analysis(session, target.symbol, "例行分析")
+                continue
+            last = recent[0]
+            all_failed = len(recent) == MAX_CONSECUTIVE_FAILURES and all(a.status == "failed" for a in recent)
+            if last.status == "failed" and last.reasons != ["手动分析"] and not all_failed and now >= last.finished_at + RETRY_AFTER:
+                await request_analysis(session, target.symbol, f"重试：{'；'.join(last.reasons)}")
+        await session.commit()

backend/tests/test_plan.py

@@ -1,8 +1,12 @@
+from datetime import UTC, datetime, timedelta, timezone
 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
+from alphapulse.trading.plan import (
+    OrderSpec, PlanContext, PlanError, SymbolRules, TradePlan, desired_orders, diff_orders, next_routine_at, round_to_step,
+    size_plan, split_take_profits,
+)
 
 BTC = SymbolRules(tick_size=Decimal("0.1"), step_size=Decimal("0.0001"), min_notional=Decimal("50"))
 
@@ -137,3 +141,99 @@ def test_action_rules():
                           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"))
+
+
+# ---------- 有持仓时的新计划 ----------
+
+def managed_plan(**overrides) -> TradePlan:
+    values = {
+        "action": "plan", "summary": "调整", "reasons": "r", "risks": "x", "direction": "long", "confidence": "medium",
+        "stop": "80000", "take_profits": [{"price": "83000", "basis": "a"}, {"price": "84000", "basis": "b"}, {"price": "85000", "basis": "c"}],
+        "invalidation": {"direction": "below", "price": "80500", "timeframe": "4h", "basis": "x"},
+    }
+    return TradePlan.model_validate({**values, **overrides})
+
+
+def holding(**overrides) -> PlanContext:
+    return context(position_side="long", position_qty=Decimal("0.0100"), position_entry=Decimal("81000"), current_stop=Decimal("79500"), has_plan=True, **overrides)
+
+
+def test_managed_plan_does_not_add_and_splits_position():
+    sizing = size_plan(managed_plan(), holding())
+    assert sizing["managed"] is True
+    assert sizing["entries"] == []
+    assert [t["quantity"] for t in sizing["take_profits"]] == ["0.0050", "0.0030", "0.0020"]
+    assert sizing["position_qty"] == "0.0100"
+    assert "entries 须为空" in errors_of(managed_plan(entries=[{"price": "80000", "ratio": "1", "basis": "a"}]), holding())
+
+
+def test_managed_plan_only_tightens_stop():
+    assert "止损只能收紧" in errors_of(managed_plan(stop="79000", invalidation={"direction": "below", "price": "79400", "timeframe": "4h", "basis": "x"}), holding())
+
+
+# ---------- 应有订单 ----------
+
+LONG_SIZING = {
+    "direction": "long", "managed": False, "quantity": "0.0030",
+    "entries": [{"price": "81000.0", "quantity": "0.0012"}, {"price": "80000.0", "quantity": "0.0018"}],
+    "take_profits": [{"price": "83000.0", "ratio": "0.5"}, {"price": "84000.0", "ratio": "0.3"}, {"price": "85000.0", "ratio": "0.2"}],
+}
+ZERO3 = [Decimal(0)] * 3
+
+
+def test_desired_orders_before_fill_places_entries_and_stop_together():
+    orders = desired_orders(LONG_SIZING, entries_open=True, entry_filled=[Decimal(0)] * 2, tp_filled=ZERO3,
+                            position_qty=Decimal(0), stop_price=Decimal("79500"), step_size=Decimal("0.0001"))
+    assert [(o.purpose, o.leg, o.side, o.quantity) for o in orders] == [
+        ("entry", 0, "BUY", Decimal("0.0012")), ("entry", 1, "BUY", Decimal("0.0018")), ("stop", None, "SELL", Decimal("0.0030")),
+    ]
+
+
+def test_desired_orders_partial_fill_splits_take_profits_by_filled():
+    orders = desired_orders(LONG_SIZING, entries_open=True, entry_filled=[Decimal("0.0012"), Decimal("0.0004")], tp_filled=ZERO3,
+                            position_qty=Decimal("0.0016"), stop_price=Decimal("79500"), step_size=Decimal("0.0001"))
+    assert [(o.purpose, o.leg, o.quantity) for o in orders] == [
+        ("entry", 1, Decimal("0.0014")), ("stop", None, Decimal("0.0030")),
+        ("tp", 0, Decimal("0.0008")), ("tp", 1, Decimal("0.0004")), ("tp", 2, Decimal("0.0004")),
+    ]
+
+
+def test_desired_orders_after_tp1_keeps_remaining_targets_only():
+    orders = desired_orders(LONG_SIZING, entries_open=False, entry_filled=[Decimal("0.0012"), Decimal("0.0018")],
+                            tp_filled=[Decimal("0.0015"), Decimal(0), Decimal(0)], position_qty=Decimal("0.0015"),
+                            stop_price=Decimal("81050"), step_size=Decimal("0.0001"))
+    assert [(o.purpose, o.leg, o.quantity, o.trigger_price) for o in orders] == [
+        ("stop", None, Decimal("0.0030"), Decimal("81050")), ("tp", 1, Decimal("0.0009"), None), ("tp", 2, Decimal("0.0006"), None),
+    ]
+
+
+def test_desired_orders_nothing_when_flat_and_entries_closed():
+    assert desired_orders(LONG_SIZING, entries_open=False, entry_filled=[Decimal(0)] * 2, tp_filled=ZERO3,
+                          position_qty=Decimal(0), stop_price=Decimal("79500"), step_size=Decimal("0.0001")) == []
+
+
+def test_diff_orders_matches_by_value():
+    keep = OrderSpec("stop", None, "SELL", "STOP_MARKET", None, Decimal("79500.0"), Decimal("0.0030"))
+    stale = OrderSpec("entry", 0, "BUY", "LIMIT", Decimal("81000"), None, Decimal("0.0012"))
+    wanted = OrderSpec("entry", 0, "BUY", "LIMIT", Decimal("80900"), None, Decimal("0.0012"))
+    cancel, place = diff_orders([OrderSpec("stop", None, "SELL", "STOP_MARKET", None, Decimal("79500"), Decimal("0.003")), wanted],
+                                [(1, keep), (2, stale)])
+    assert cancel == [2]
+    assert place == [wanted]
+
+
+def test_next_routine_at_aligns_to_candle_close():
+    tz8 = timezone(timedelta(hours=8))
+    # 北京 10:20 = UTC 02:20；短线 +45 分钟后的第一个整点 UTC 04:00，再加 1 分钟
+    assert next_routine_at(datetime(2026, 10, 15, 10, 20, tzinfo=tz8), "short") == datetime(2026, 10, 15, 4, 1, tzinfo=UTC)
+    assert next_routine_at(datetime(2026, 10, 15, 3, 1, tzinfo=UTC), "short") == datetime(2026, 10, 15, 4, 1, tzinfo=UTC)
+    # 长线：UTC 04:01 + 3h = 07:01，下一个 4h 收盘 08:00
+    assert next_routine_at(datetime(2026, 10, 15, 4, 1, tzinfo=UTC), "long") == datetime(2026, 10, 15, 8, 1, tzinfo=UTC)
+
+
+def test_step_decimal_never_uses_exponent():
+    from alphapulse.trading.client import step_decimal
+
+    assert str(step_decimal("0.10")) == "0.1"
+    assert str(step_decimal("100")) == "100"
+    assert str(round_to_step(Decimal("12345"), step_decimal("100"), ROUND_FLOOR)) == "12300"

web/src/apis/trading.ts

@@ -77,7 +77,10 @@ export interface TradingSymbol {
   added_at: string
   analysis_requested_at: string | null
   analyzing: boolean
-  plan_status: "pending" | "open" | null
+  /** 执行器正在管理的计划 */
+  plan_id: number | null
+  plan_status: "pending" | "open" | "closing" | null
+  close_deadline: string | null
 }
 
 export interface SymbolOption {
@@ -183,6 +186,10 @@ export interface Sizing {
   capped_by_leverage: boolean
   reward_risk: string
   reward_risk_first_only: string
+  /** 采纳前核对用：提交时的持仓数量 */
+  position_qty: string
+  /** 持仓中调整的计划：不加仓 */
+  managed: boolean
 }
 
 export interface Analysis {
@@ -201,13 +208,13 @@ export interface Analysis {
   usage: { input_tokens: number, output_tokens: number } | null
   started_at: string
   finished_at: string | null
-  plan_status: "pending" | "open" | "closed" | "canceled" | "replaced" | null
+  plan_status: PlanStatus | null
   valid_until: string | null
   first_fill_at: string | null
   close_deadline: string | null
   closed_at: string | null
   close_reason: string | null
-  result: Record<string, unknown> | null
+  result: { realized_pnl: string, commission: string, net: string } | null
 }
 
 /** 与协议无关的消息格式，同 Chat */
@@ -221,6 +228,41 @@ export interface AnalysisDetail extends Analysis {
   transcript: TranscriptMessage[] | null
 }
 
+export type PlanStatus = "proposed" | "pending" | "open" | "closing" | "closed" | "canceled" | "replaced" | "stale" | "applied"
+
+export const PLAN_STATUS_LABELS: Record<PlanStatus, string> = {
+  proposed: "待采纳",
+  pending: "挂单中",
+  open: "持仓中",
+  closing: "平仓中",
+  closed: "已结束",
+  canceled: "已撤销",
+  replaced: "已被替换",
+  stale: "已作废",
+  applied: "已执行"
+}
+
+/** 执行器的订单流水 */
+export interface OrderRow {
+  id: number
+  purpose: "entry" | "stop" | "tp" | "close"
+  leg: number | null
+  side: "BUY" | "SELL"
+  type: string
+  price: string | null
+  trigger_price: string | null
+  quantity: string
+  status: string
+  filled_qty: string
+  avg_price: string | null
+  error: string | null
+  created_at: string
+}
+
+export function getSymbolPlan(symbol: string): Promise<{ plan: Analysis | null, orders: OrderRow[] }> {
+  return http.get(`/trading/symbols/${symbol}/plan`)
+}
+
 export function requestAnalysis(symbol: string): Promise<void> {
   return http.post(`/trading/symbols/${symbol}/analyze`)
 }

web/src/pages/trading/components/PlanCard.vue

@@ -1,9 +1,10 @@
 <script lang="ts" setup>
-import type { Analysis, PlanCondition, TradingSymbol } from "@/apis/trading"
+import type { Analysis, OrderRow, PlanCondition, TradingSymbol } from "@/apis/trading"
-import { ANGLE_LABELS } from "@/apis/trading"
+import { ANGLE_LABELS, PLAN_STATUS_LABELS } from "@/apis/trading"
 import { formatTime } from "@/utils/text"
 
-const props = defineProps<{ symbol: TradingSymbol, analysis: Analysis | null, requesting: boolean }>()
+/** analysis：执行中的计划，没有时为最近一次分析；latest：最近一次分析（与 analysis 不同时另行提示） */
+const props = defineProps<{ symbol: TradingSymbol, analysis: Analysis | null, latest: Analysis | null, orders: OrderRow[], requesting: boolean }>()
 const emit = defineEmits<{ analyze: [], detail: [id: number] }>()
 
 const ACTION_LABELS = { plan: "新计划", keep: "维持当前计划", exit: "结束当前计划", wait: "观望" } as const
@@ -19,6 +20,20 @@ const sizing = computed(() => props.analysis?.sizing ?? null)
 const busy = computed(() => props.requesting || props.symbol.analyzing || !!props.symbol.analysis_requested_at)
 const elapsed = computed(() => props.analysis ? Math.max(0, Math.round((now.value - new Date(props.analysis.started_at).getTime()) / 1000)) : 0)
 
+function filled(purpose: "entry" | "tp", leg: number) {
+  const total = props.orders.filter(o => o.purpose === purpose && o.leg === leg).reduce((sum, o) => sum + Number(o.filled_qty), 0)
+  return total ? String(Number(total.toFixed(8))) : "-"
+}
+
+function countdown(deadline: string | null) {
+  if (!deadline)
+    return ""
+  const seconds = Math.round((new Date(deadline).getTime() - now.value) / 1000)
+  if (seconds <= 0)
+    return "已到期"
+  return `${Math.floor(seconds / 3600)} 小时 ${Math.floor(seconds % 3600 / 60)} 分`
+}
+
 function condition(c: PlanCondition) {
   return `${TIMEFRAME_LABELS[c.timeframe]}${c.direction === "above" ? "高于" : "低于"} ${c.price}`
 }
@@ -56,6 +71,9 @@ const invalidationAction = computed(() => props.analysis?.angle === "short" ? "
         <el-tag v-if="analysis.dry_run" type="info" size="small">
           试运行
         </el-tag>
+        <el-tag v-if="analysis.plan_status" type="success" effect="dark" size="small">
+          {{ PLAN_STATUS_LABELS[analysis.plan_status] }}
+        </el-tag>
         <el-tag size="small">
           {{ ACTION_LABELS[decision.action] }}
         </el-tag>
@@ -73,10 +91,33 @@ const invalidationAction = computed(() => props.analysis?.angle === "short" ? "
       <div class="font-bold">
         {{ decision.summary }}
       </div>
+      <div v-if="latest && latest.id !== analysis.id" class="text-xs text-[var(--el-text-color-secondary)]">
+        最近一次分析 #{{ latest.id }}（{{ formatTime(latest.started_at) }}）：
+        <template v-if="latest.status === 'done'">
+          {{ latest.decision?.summary }}
+        </template>
+        <template v-else>
+          {{ latest.status === "running" ? "分析中" : `失败：${latest.error}` }}
+        </template>
+        <el-link type="primary" underline="never" class="ml-1 text-xs!" @click="emit('detail', latest.id)">
+          查看
+        </el-link>
+      </div>
+      <div v-if="analysis.valid_until || analysis.close_deadline" class="flex flex-wrap gap-x-6">
+        <span v-if="analysis.plan_status === 'pending' && analysis.valid_until"><span class="label">入场单有效至</span>{{ formatTime(analysis.valid_until, "MM-DD HH:mm") }}（剩 {{ countdown(analysis.valid_until) }}）</span>
+        <span v-if="analysis.first_fill_at"><span class="label">首档成交</span>{{ formatTime(analysis.first_fill_at, "MM-DD HH:mm") }}</span>
+        <span v-if="analysis.close_deadline"><span class="label">最晚平仓</span>{{ formatTime(analysis.close_deadline, "MM-DD HH:mm") }}（剩 {{ countdown(analysis.close_deadline) }}）</span>
+      </div>
+      <div v-if="analysis.result">
+        <span class="label">结果</span>{{ analysis.close_reason }}，净盈亏 {{ analysis.result.net }} USDT（手续费 {{ analysis.result.commission }}）
+      </div>
       <div><span class="label">为什么</span>{{ decision.reasons }}</div>
 
+      <div v-if="sizing?.managed" class="text-xs text-[var(--el-text-color-secondary)]">
+        持仓中调整的计划：不加仓，止盈按当前持仓数量拆分
+      </div>
       <template v-if="decision.action === 'plan' && sizing">
-        <el-table :data="sizing.entries" border size="small">
+        <el-table v-if="sizing.entries.length" :data="sizing.entries" border size="small">
           <el-table-column label="入场档位" min-width="110">
             <template #default="{ $index }">
               第 {{ $index + 1 }} 档
@@ -90,6 +131,11 @@ const invalidationAction = computed(() => props.analysis?.angle === "short" ? "
           </el-table-column>
           <el-table-column prop="quantity" label="数量" min-width="90" />
           <el-table-column prop="notional" label="名义 USDT" min-width="100" />
+          <el-table-column label="已成交" min-width="80">
+            <template #default="{ $index }">
+              {{ filled("entry", $index) }}
+            </template>
+          </el-table-column>
           <el-table-column label="依据" min-width="200">
             <template #default="{ $index }">
               {{ decision.entries?.[$index]?.basis }}
@@ -109,6 +155,11 @@ const invalidationAction = computed(() => props.analysis?.angle === "short" ? "
             </template>
           </el-table-column>
           <el-table-column prop="quantity" label="数量" min-width="90" />
+          <el-table-column label="已成交" min-width="80">
+            <template #default="{ $index }">
+              {{ filled("tp", $index) }}
+            </template>
+          </el-table-column>
           <el-table-column label="依据" min-width="300">
             <template #default="{ $index }">
               {{ decision.take_profits?.[$index]?.basis }}

web/src/pages/trading/index.vue

@@ -1,8 +1,8 @@
 <script lang="ts" setup>
 import type { Effort, Provider } from "@/apis/settings"
-import type { Analysis, Angle, SnapshotPosition, TradingEvent, TradingStatus, TradingSymbol } from "@/apis/trading"
+import type { Analysis, Angle, OrderRow, SnapshotPosition, TradingEvent, TradingStatus, TradingSymbol } from "@/apis/trading"
 import { EFFORTS, listProviders } from "@/apis/settings"
-import { ANGLE_LABELS, deleteSymbol, getStatus, isOwnOrder, listAnalyses, listEvents, listSymbols, patchConfig, patchSymbol, requestAnalysis, startTrading, stopTrading } from "@/apis/trading"
+import { ANGLE_LABELS, deleteSymbol, getStatus, getSymbolPlan, isOwnOrder, listAnalyses, listEvents, listSymbols, patchConfig, patchSymbol, requestAnalysis, startTrading, stopTrading } from "@/apis/trading"
 import { usePersisted } from "@/utils/storage"
 import { formatTime } from "@/utils/text"
 import AddSymbolDialog from "./components/AddSymbolDialog.vue"
@@ -23,6 +23,8 @@ const eventsExhausted = ref(false)
 const addVisible = ref(false)
 const toggling = ref(false)
 const latestAnalysis = ref<Analysis | null>(null)
+const livePlan = ref<Analysis | null>(null)
+const planOrders = ref<OrderRow[]>([])
 const requesting = ref(false)
 const drawerVisible = ref(false)
 const drawerAnalysisId = ref<number | null>(null)
@@ -81,6 +83,8 @@ function exitPrices(p: SnapshotPosition) {
 function symbolState(s: TradingSymbol) {
   if (s.analyzing)
     return { label: "分析中", type: "warning" } as const
+  if (s.plan_status === "closing")
+    return { label: "平仓中", type: "danger" } as const
   if (positions.value.some(p => p.symbol === s.symbol))
     return { label: "持仓中", type: "success" } as const
   if (s.plan_status === "pending")
@@ -97,9 +101,23 @@ async function refresh() {
 
 async function refreshAnalysis() {
   const symbol = selected.value
-  const latest = symbol ? (await listAnalyses({ symbol, limit: 1 }))[0] ?? null : null
+  if (!symbol)
-  if (symbol === selected.value)
+    return
-    latestAnalysis.value = latest
+  const [latest, { plan, orders }] = await Promise.all([listAnalyses({ symbol, limit: 1 }), getSymbolPlan(symbol)])
+  if (symbol !== selected.value)
+    return
+  latestAnalysis.value = latest[0] ?? null
+  livePlan.value = plan
+  planOrders.value = orders
+}
+
+/** 持仓的平仓倒计时：取该币执行中计划的最晚平仓时间 */
+function closeCountdown(symbol: string) {
+  const deadline = symbols.value.find(s => s.symbol === symbol)?.close_deadline
+  if (!deadline)
+    return "-"
+  const minutes = Math.round((new Date(deadline).getTime() - Date.now()) / 60000)
+  return minutes <= 0 ? "已到期" : `${Math.floor(minutes / 60)}h ${minutes % 60}m`
 }
 
 async function analyze() {
@@ -144,6 +162,8 @@ async function loadMoreEvents() {
 watch(selected, () => {
   events.value = []
   latestAnalysis.value = null
+  livePlan.value = null
+  planOrders.value = []
   refreshEvents()
   refreshAnalysis()
 })
@@ -348,6 +368,11 @@ onBeforeUnmount(() => clearInterval(timer))
           {{ exitPrices(row as SnapshotPosition).stop }} / {{ exitPrices(row as SnapshotPosition).target }}
         </template>
       </el-table-column>
+      <el-table-column label="平仓倒计时" width="100">
+        <template #default="{ row }">
+          {{ closeCountdown(row.symbol) }}
+        </template>
+      </el-table-column>
     </el-table>
 
     <div class="min-h-0 flex flex-1 gap-3">
@@ -388,7 +413,8 @@ onBeforeUnmount(() => clearInterval(timer))
       <el-scrollbar class="flex-1">
         <div class="flex flex-col gap-3 pr-2">
           <PlanCard
-            v-if="current" :symbol="current" :analysis="latestAnalysis" :requesting="requesting"
+            v-if="current" :symbol="current" :analysis="livePlan ?? latestAnalysis" :latest="latestAnalysis" :orders="planOrders"
+            :requesting="requesting"
             @analyze="analyze" @detail="openAnalysis"
           />