Add per-symbol close position and clear orders actions to demo trading

The API queues the request in Redis and the executor applies it on its
next pass, so it never races reconciliation. Both actions cancel every
open order on the symbol, including manual ones, and market-close any
position. Plans end as manual_close or manual_cancel without queueing
a new analysis. Closing also stops trading so the engine does not
reopen right away.

backend/src/alphapulse/api/trading.py

@@ -19,6 +19,7 @@ from alphapulse.core.db import get_session
 from alphapulse.core.redis import KEY_PREFIX, redis
 from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingFill, TradingOrder, TradingSymbol
 from alphapulse.trading.client import OWN_CLIENT_ID, ExchangeError, public_get
+from alphapulse.trading.executor import MANUAL_ACTION_KEY
 from alphapulse.trading.positions import build_positions
 from alphapulse.trading.plan import DAILY_ANALYSIS_LIMIT
 from alphapulse.trading.snapshot import read_snapshot
@@ -29,6 +30,8 @@ SYMBOLS_CACHE_KEY = f"{KEY_PREFIX}trading:symbols"
 SYMBOLS_CACHE_SECONDS = 3600
 # 删除前要求快照足够新，否则无法确认没有仓位和挂单
 SNAPSHOT_FRESH_SECONDS = 30
+# 平仓、清除挂单的请求等执行器取走的时限：引擎没及时处理就作废，不在之后意外执行
+MANUAL_ACTION_TTL_SECONDS = 30
 # 执行器正在管理的计划状态
 LIVE_PLAN_STATUSES = ("pending", "open", "closing")
 
@@ -312,6 +315,15 @@ async def add_symbol(body: SymbolIn, session: Annotated[AsyncSession, Depends(ge
         raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 已在交易列表中") from None
 
 
+async def fresh_snapshot(symbol: str, action: str) -> dict:
+    """引擎在线且刚和币安同步过的快照"""
+    snapshot = await read_snapshot()
+    synced_at = snapshot and snapshot["synced_at"] and datetime.fromisoformat(snapshot["synced_at"])
+    if not synced_at or datetime.now(LOCAL_TZ) - synced_at > timedelta(seconds=SNAPSHOT_FRESH_SECONDS):
+        raise HTTPException(status.HTTP_409_CONFLICT, f"无法确认 {symbol} 的仓位和挂单（引擎离线或连接币安失败），暂时不能{action}")
+    return snapshot
+
+
 async def ensure_idle(session: AsyncSession, symbol: str, action: str) -> TradingSymbol:
     """只有空仓、没有本程序的挂单、没有进行中的计划和分析时，才能删除"""
     s = await session.get(TradingSymbol, symbol)
@@ -323,10 +335,7 @@ async def ensure_idle(session: AsyncSession, symbol: str, action: str) -> Tradin
     ))
     if busy:
         raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有进行中的分析或计划，不能{action}")
-    snapshot = await read_snapshot()
-    synced_at = snapshot and snapshot["synced_at"] and datetime.fromisoformat(snapshot["synced_at"])
-    if not synced_at or datetime.now(LOCAL_TZ) - synced_at > timedelta(seconds=SNAPSHOT_FRESH_SECONDS):
-        raise HTTPException(status.HTTP_409_CONFLICT, f"无法确认 {symbol} 的仓位和挂单（引擎离线或连接币安失败），暂时不能{action}")
+    snapshot = await fresh_snapshot(symbol, action)
     if any(p["symbol"] == symbol for p in snapshot["positions"]):
         raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有持仓，不能{action}")
     if any(o["symbol"] == symbol and OWN_CLIENT_ID.match(o["client_id"]) for o in snapshot["orders"] + snapshot["algo_orders"]):
@@ -343,6 +352,19 @@ async def delete_symbol(symbol: str, session: Annotated[AsyncSession, Depends(ge
     await session.commit()
 
 
+@router.post("/symbols/{symbol}/manual/{action}", status_code=status.HTTP_202_ACCEPTED)
+async def manual_action(symbol: str, action: Literal["close", "cancel"], session: Annotated[AsyncSession, Depends(get_session)]) -> None:
+    """一键平仓（同时停止行情交易）、清除挂单：交给执行器执行，结果写入时间线"""
+    label = "平仓" if action == "close" else "清除挂单"
+    if await session.get(TradingSymbol, symbol) is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"{symbol} 不在交易列表中")
+    await fresh_snapshot(symbol, label)
+    if action == "close":
+        # 不停止的话，平仓后会重新分析，可能马上又开仓
+        await set_running(session, False)
+    await redis.set(MANUAL_ACTION_KEY.format(symbol), action, ex=MANUAL_ACTION_TTL_SECONDS)
+
+
 @router.get("/events")
 async def list_events(
     session: Annotated[AsyncSession, Depends(get_session)],

backend/src/alphapulse/trading/executor.py

@@ -20,6 +20,7 @@ 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.core.redis import KEY_PREFIX, redis
 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
@@ -46,7 +47,11 @@ ORDER_STATUSES = {"NEW": "new", "PARTIALLY_FILLED": "partially_filled", "FILLED"
                   "EXPIRED": "expired", "EXPIRED_IN_MATCH": "expired", "REJECTED": "rejected"}
 PURPOSE_LABELS = {"entry": "入场", "tp": "止盈", "stop": "止损", "close": "市价平仓"}
 CLOSE_REASON_LABELS = {"stop": "止损", "take_profit": "止盈", "time": "到达持仓时限", "exit": "模型决定结束",
-                       "external": "外部平仓（如在网页手动平仓）", "stop_failed": "止损挂单失败", "close": "市价平仓"}
+                       "external": "外部平仓（如在网页手动平仓）", "stop_failed": "止损挂单失败", "close": "市价平仓",
+                       "manual_close": "一键平仓", "manual_cancel": "清除挂单"}
+# 页面上的「平仓」「清除挂单」：API 写入请求，执行器下一轮取走执行，避免和对账同时改计划
+MANUAL_ACTION_KEY = f"{KEY_PREFIX}trader:manual:{{}}"
+MANUAL_ACTIONS = {"close": "一键平仓", "cancel": "清除挂单"}
 CANCEL_REASON_LABELS = {"expired": "限价单到期未成交", "stopped": "已停止交易", "entry_failed": "入场单被拒"}
 
 
@@ -333,7 +338,8 @@ class Executor:
 
     # ---------- 对账 ----------
 
-    async def reconcile(self, session: AsyncSession, client: DemoClient, symbol: TradingSymbol, running: bool, position: Position | None) -> None:
+    async def reconcile(self, session: AsyncSession, client: DemoClient, symbol: TradingSymbol, running: bool, position: Position | None,
+                        manual_action: str | 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})}
@@ -359,6 +365,9 @@ class Executor:
         )
         plan = await self.adopt(session, client, symbol, plan, position)
         await session.commit()
+        if manual_action is not None:
+            await self.manual(session, client, name, manual_action, plan, position, open_orders, open_algos)
+            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)]
@@ -428,6 +437,31 @@ class Executor:
             await self.check_reevaluate(session, name, position.mark if position else None)
         await session.commit()
 
+    async def manual(self, session: AsyncSession, client: DemoClient, symbol: str, action: str, plan: TradingAnalysis | None,
+                     position: Position | None, open_orders: dict, open_algos: dict) -> None:
+        """一键平仓、清除挂单：撤掉该币全部挂单（含止损止盈和手动挂的），有持仓就市价平仓——程序不留没有止损的仓位。
+        本程序的单通过结束计划撤销，否则下一轮对账会按计划补挂回来。撤销计划不像到期那样排队重新分析，免得马上又挂单"""
+        reason = f"manual_{action}"
+        label = MANUAL_ACTIONS[action]
+        for client_id in [*open_orders, *open_algos]:
+            if not OWN_CLIENT_ID.match(client_id):
+                await self.cancel_client_id(client, symbol, client_id, client_id in open_algos)
+        detail = f"，市价平仓 {fmt(position.qty)}" if position else ""
+        session.add(TradingEvent(symbol=symbol, analysis_id=plan.id if plan else None, kind=reason, level="warn",
+                                 message=f"{label}：撤销全部挂单{detail}"))
+        if plan is not None and plan.plan_status in ("pending", "open") and position is not None:
+            # 平仓中的计划：撤入场和止盈、市价平仓，持仓归零后结束并撤掉止损
+            plan.plan_status, plan.close_reason = "closing", reason
+        elif plan is not None and plan.plan_status == "pending":
+            plan.plan_status, plan.close_reason, plan.closed_at = "canceled", reason, utcnow()
+        elif plan is None and position is not None:
+            closing = await session.scalar(select(func.count()).where(
+                TradingOrder.symbol == symbol, TradingOrder.purpose == "close", TradingOrder.status.in_(ACTIVE_ORDER_STATUSES),
+            ))
+            if not closing:
+                close_side = "SELL" if position.side == "long" else "BUY"
+                await self.place(session, client, symbol, None, "close", OrderSpec("close", None, close_side, "MARKET", None, None, position.qty))
+
     async def check_reevaluate(self, session: AsyncSession, symbol: str, mark: Decimal | None) -> None:
         """最近一次交易中分析给出的 reevaluate 条件：满足任一个就排队分析"""
         latest = await session.scalar(
@@ -564,7 +598,8 @@ class Executor:
                     logger.warning("同步成交记录失败，%ds 后重试", FILLS_SYNC_SECONDS, exc_info=True)
             for symbol in symbols:
                 try:
-                    await self.reconcile(session, client, symbol, account.running, positions.get(symbol.symbol))
+                    manual_action = await redis.getdel(MANUAL_ACTION_KEY.format(symbol.symbol))
+                    await self.reconcile(session, client, symbol, account.running, positions.get(symbol.symbol), manual_action)
                 except ExchangeError as e:
                     await session.rollback()
                     await self.report(session, symbol.symbol, f"对账失败：{e}")