Support selectable demo exchanges and configurable analysis limits

Persist manual requests until exchange confirmation, preserve stop protection, and reconcile fills before settling plans. Tighten model stream, crawler, skill, research accounting, and frontend race handling, with regression coverage. Keep queued analyses pending until a fresh account snapshot is available.

backend/tests/conftest.py

@@ -0,0 +1,14 @@
+import pytest
+
+
+@pytest.fixture
+def trading(monkeypatch):
+    # 测试只替换 I/O 边界，不读取项目凭证，也不连接线上数据库或 Redis。
+    for key, value in {
+        "DATABASE_URL": "postgresql+asyncpg://test:[email protected]/test",
+        "REDIS_URL": "redis://127.0.0.1:1/0", "JWT_SECRET": "test-only",
+        "ADMIN_USERNAME": "test", "SECRET_ENCRYPTION_KEY": "MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA=",
+    }.items():
+        monkeypatch.setenv(key, value)
+    from alphapulse.trading import executor
+    return executor

backend/tests/test_bybit.py

@@ -0,0 +1,91 @@
+import asyncio
+import hashlib
+import hmac
+import json
+from decimal import Decimal
+
+import httpx
+import pytest
+
+from alphapulse.trading.bybit import BybitDemoClient, normalize_order, normalize_trade
+from alphapulse.trading.client import ExchangeError
+from alphapulse.trading.plan import OrderSpec
+
+
+@pytest.mark.parametrize("method,params", [("GET", {"symbol": "BTCUSDT", "cursor": "a:b+="}), ("POST", {"reduceOnly": True, "qty": "0.001"})])
+def test_signature_matches_exact_transmitted_payload(method, params):
+    def handle(request):
+        payload = request.url.query.decode() if method == "GET" else request.content.decode()
+        headers = request.headers
+        signed = headers["X-BAPI-TIMESTAMP"] + "test-key" + headers["X-BAPI-RECV-WINDOW"] + payload
+        assert headers["X-BAPI-SIGN"] == hmac.new(b"test-secret", signed.encode(), hashlib.sha256).hexdigest()
+        assert request.url.host == "api-demo.bybit.com"
+        if method == "POST":
+            assert json.loads(payload) == params
+        return httpx.Response(200, json={"retCode": 0, "result": {"confirmed": True}})
+
+    async def check():
+        async with BybitDemoClient("test-key", "test-secret") as client:
+            await client.http.aclose()
+            client.http = httpx.AsyncClient(base_url="https://api-demo.bybit.com", transport=httpx.MockTransport(handle))
+            assert await client.request(method, "/v5/test", params) == {"confirmed": True}
+    asyncio.run(check())
+
+
+def test_conditional_stop_ack_keeps_execution_unknown():
+    def handle(request):
+        params = json.loads(request.content)
+        assert (params["orderType"], params["triggerDirection"], params["triggerBy"]) == ("Market", 2, "MarkPrice")
+        assert params["reduceOnly"] and params["closeOnTrigger"]
+        assert params["positionIdx"] == 0
+        assert params["orderLinkId"] == "ap42"
+        return httpx.Response(200, json={"retCode": 0, "result": {"orderId": "7b6a537e-6f74-43d9-b775-35d0521c20de"}})
+
+    async def check():
+        async with BybitDemoClient("test-key", "test-secret") as client:
+            await client.http.aclose()
+            client.http = httpx.AsyncClient(base_url="https://api-demo.bybit.com", transport=httpx.MockTransport(handle))
+            result = await client.place_order("BTCUSDT", "ap42", OrderSpec("stop", None, "SELL", "STOP_MARKET", None, Decimal("70000"), Decimal("0.001")), True)
+            assert result["status"] == "SUBMITTING"
+    asyncio.run(check())
+
+
+def test_filled_conditional_order_preserves_execution_and_uuid():
+    order_id = "7b6a537e-6f74-43d9-b775-35d0521c20de"
+    result = normalize_order({
+        "symbol": "BTCUSDT", "orderId": order_id, "orderLinkId": "ap42", "side": "Sell", "orderType": "Market",
+        "orderStatus": "Filled", "price": "", "qty": "0.001", "cumExecQty": "0.001", "avgPrice": "70000",
+        "reduceOnly": True, "createdTime": "1792400000000", "triggerPrice": "70100",
+        "stopOrderType": "Stop", "triggerDirection": 2, "closeOnTrigger": True,
+    }, True)
+    assert (result["status"], result["executedQty"], result["avgPrice"]) == ("FILLED", "0.001", "70000")
+    assert result["actualOrderId"] == result["algoId"] == order_id
+
+
+def test_trade_uses_realized_cashflow_without_subtracting_fee_twice():
+    result = normalize_trade({"symbol": "BTCUSDT", "tradeId": "trade-uuid", "orderId": "order-uuid", "side": "Sell",
+                              "tradePrice": "81000", "qty": "0.001", "cashFlow": "1", "fee": "0.04", "currency": "USDT",
+                              "transactionTime": "1792400000000"})
+    assert (result["realizedPnl"], result["commission"], result["id"]) == ("1", "0.04", "trade-uuid")
+
+
+def test_timeout_response_is_unknown_and_not_resent():
+    calls = []
+
+    def handle(request):
+        calls.append(request)
+        return httpx.Response(200, json={"retCode": 10000, "retMsg": "Server Timeout", "result": {}})
+
+    async def check():
+        async with BybitDemoClient("test-key", "test-secret") as client:
+            await client.http.aclose()
+            client.http = httpx.AsyncClient(base_url="https://api-demo.bybit.com", transport=httpx.MockTransport(handle))
+            with pytest.raises(ExchangeError) as error:
+                await client.request("POST", "/v5/order/create", {"orderLinkId": "ap42"})
+            assert error.value.retryable
+            assert len(calls) == 1
+    asyncio.run(check())
+
+
+def test_binance_backend_timeout_is_not_definitive_rejection():
+    assert ExchangeError(408, -1007, "Timeout").retryable

backend/tests/test_failure_paths.py

@@ -0,0 +1,181 @@
+import asyncio
+import json
+import uuid
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import httpx
+import httpx2
+import pytest
+
+
+def install_model_transport(llm, monkeypatch, protocol, handler):
+    monkeypatch.setattr(llm, "decrypt", lambda value: "test-only")
+    module = llm.openai if protocol == "openai" else llm.anthropic
+    name = "AsyncOpenAI" if protocol == "openai" else "AsyncAnthropic"
+    constructor = getattr(module, name)
+    http = httpx if protocol == "openai" else httpx2
+    monkeypatch.setattr(module, name, lambda **kwargs: constructor(**kwargs, http_client=http.AsyncClient(transport=http.MockTransport(handler))))
+    return SimpleNamespace(protocol=protocol, base_url="https://model.invalid", api_key_encrypted=b"test", model="test", max_tokens=100)
+
+
+@pytest.mark.parametrize("protocol", ["openai", "anthropic"])
+@pytest.mark.parametrize("complete", [False, True])
+def test_model_requires_protocol_end_marker(trading, monkeypatch, protocol, complete):
+    from alphapulse.ai import llm
+    if protocol == "openai":
+        chunks = [{"id": "1", "object": "chat.completion.chunk", "created": 0, "model": "test", "choices": [{"index": 0, "delta": {"content": "部分回答"}, "finish_reason": None}]}]
+        if complete:
+            chunks.append(chunks[0] | {"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]})
+        body = "".join(f"data: {json.dumps(c)}\n\n" for c in chunks) + "data: [DONE]\n\n"
+    else:
+        events = [
+            {"type": "message_start", "message": {"id": "1", "type": "message", "role": "assistant", "model": "test", "content": [], "stop_reason": None, "stop_sequence": None, "usage": {"input_tokens": 1, "output_tokens": 0}}},
+            {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+            {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "部分回答"}},
+            {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, "usage": {"output_tokens": 1}},
+        ]
+        if complete:
+            events.append({"type": "message_stop"})
+        body = "".join(f"event: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in events)
+
+    def handle(request):
+        http = httpx if protocol == "openai" else httpx2
+        return http.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())
+
+    provider = install_model_transport(llm, monkeypatch, protocol, handle)
+    async def check():
+        output = []
+        try:
+            async for item in llm.stream(provider, "system", [], []):
+                output.append(item)
+        except llm.LlmError:
+            assert not complete
+        else:
+            assert complete
+        assert [x.text for x in output if isinstance(x, llm.TextDelta)] == ["部分回答"]
+        assert any(isinstance(x, llm.Completion) for x in output) == complete
+    asyncio.run(check())
+
+
+def test_connection_failure_uses_only_sdk_retry_budget(trading, monkeypatch):
+    from alphapulse.ai import llm
+    attempts = 0
+    def handle(request):
+        nonlocal attempts
+        attempts += 1
+        raise httpx.ConnectError("测试连接失败", request=request)
+    provider = install_model_transport(llm, monkeypatch, "openai", handle)
+    async def check():
+        with pytest.raises(llm.LlmError, match="无法连接"):
+            async for _ in llm.stream(provider, "system", [], []):
+                pass
+    asyncio.run(check())
+    assert attempts == llm.MAX_RETRIES + 1
+
+
+def test_pipeline_error_keeps_history_checkpoint_and_marks_run_failed(trading, monkeypatch):
+    from scrapy import signals
+    from scrapy.crawler import Crawler
+    from twisted.python.failure import Failure
+    from alphapulse.crawler.spiders import base
+
+    class Spider(base.NewsSpider):
+        name = "test"
+
+    now = datetime.now(UTC)
+    target, checkpoint = now - timedelta(days=10), now - timedelta(days=1)
+    source = SimpleNamespace(history_target=target, history_position=checkpoint, consecutive_failures=0)
+    writes = []
+    class Session:
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *args):
+            pass
+        async def execute(self, statement):
+            writes.append(statement.compile().params)
+        async def get(self, *args, **kwargs):
+            return source
+        async def commit(self):
+            pass
+    monkeypatch.setattr(base, "SessionLocal", Session)
+    async def check():
+        crawler = Crawler(Spider)
+        spider = Spider.from_crawler(crawler)
+        run = base.RunState(uuid.uuid4(), "history", now, target, history_target=target, history_position=target)
+        spider.run = run
+        crawler.signals.send_catch_log(signals.item_error, item={}, response=None, spider=spider, failure=Failure(RuntimeError("写库失败")))
+        spider._close_run()
+        await asyncio.gather(*spider._tasks)
+        assert spider.history_checkpoint is None and source.history_position == checkpoint
+        assert source.consecutive_failures == 1
+        assert writes[0]["status"] == "failed" and "写库失败" in writes[0]["error"]
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("failure_at", ["usage", "second_tool"])
+def test_chat_completes_tool_results_on_disconnect_or_exception(trading, monkeypatch, failure_at):
+    from alphapulse.api import chat
+    provider = SimpleNamespace(id=1, context_window=10000, max_tokens=100, name="test", reasoning_effort=None)
+    class Session:
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *args):
+            pass
+        async def scalar(self, *args):
+            return provider
+        def expunge(self, *args):
+            pass
+    async def current(*args):
+        return SimpleNamespace(id=1)
+    async def system(*args):
+        return "system"
+    async def history(*args):
+        return []
+    saved = []
+    async def save(*messages):
+        saved.extend(messages)
+    calls = [{"id": "a", "name": "first", "arguments": {}}, {"id": "b", "name": "second", "arguments": {}}]
+    async def stream(*args):
+        yield chat.llm.Completion(tool_calls=calls)
+    async def tool(name, args):
+        if name == "second":
+            raise RuntimeError("工具失败")
+        return chat.ToolOutcome("实际结果")
+    monkeypatch.setattr(chat, "SessionLocal", Session)
+    monkeypatch.setattr(chat, "active_session", current)
+    monkeypatch.setattr(chat, "build_system_prompt", system)
+    monkeypatch.setattr(chat, "session_messages", history)
+    monkeypatch.setattr(chat, "save_messages", save)
+    monkeypatch.setattr(chat.llm, "stream", stream)
+    monkeypatch.setattr(chat, "run_tool", tool)
+    async def check():
+        generator = chat.send_message(chat.SendIn(text="test"))
+        events = []
+        async for event in generator:
+            events.append(event.event)
+            if failure_at == "usage" and event.event == "usage":
+                await generator.aclose()
+                break
+        results = [m.content for m in saved if m.role == "tool"]
+        assert [r["tool_call_id"] for r in results] == ["a", "b"]
+        if failure_at == "second_tool":
+            assert results[0]["result"] == "实际结果"
+            assert "error" in json.loads(results[1]["result"])
+            assert events[-1] == "error"
+    asyncio.run(check())
+
+
+def test_skill_rename_rejects_whitespace_without_writing(trading):
+    from fastapi import HTTPException
+    from alphapulse.api import skills
+    node = SimpleNamespace(name="原名称")
+    class Session:
+        async def get(self, *args):
+            return node
+    async def check():
+        with pytest.raises(HTTPException) as error:
+            await skills.patch(1, skills.NodePatch(name=" \t "), Session())
+        assert error.value.status_code == 400
+        assert node.name == "原名称"
+    asyncio.run(check())

backend/tests/test_research_accounting.py

@@ -0,0 +1,64 @@
+import asyncio
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from types import SimpleNamespace
+
+import pytest
+
+
+def test_forward_endpoint_excludes_labels_at_or_after_entry(trading):
+    from alphapulse.api.trading import list_news_signals
+    from alphapulse.trading.news_signals import is_forward_sample
+    published = datetime(2026, 10, 20, 12, 0, 59, tzinfo=UTC)
+    entry = published.replace(second=0) + timedelta(minutes=5)
+    assert is_forward_sample(published, entry - timedelta(microseconds=1))
+    assert not is_forward_sample(published, entry)
+    prices = {symbol: {"entry": "100", "1h": "101", "4h": "102"} for symbol in ("BTCUSDT", "ETHUSDT")}
+    rows = [(SimpleNamespace(news_id=i, labeled_at=labeled, category="event", direction="bullish", prices=prices), published, "test", "新闻")
+            for i, labeled in enumerate((entry - timedelta(seconds=1), entry, entry + timedelta(hours=5)), 1)]
+    class Session:
+        async def execute(self, statement):
+            return SimpleNamespace(all=lambda: rows)
+    result = asyncio.run(list_news_signals(Session()))
+    assert result.labeled == result.filled == 1
+    assert [item.news_id for item in result.recent] == [1]
+    assert all(s.n == 1 for s in result.stats)
+
+
+@pytest.mark.parametrize("scenario", ["complete", "missing_close", "partial_entry"])
+def test_plan_pnl_includes_early_entry_fee_and_manual_close_only(trading, monkeypatch, scenario):
+    from alphapulse.models import TradingFill
+    start = datetime(2026, 10, 20, 10, tzinfo=UTC)
+    fills = [TradingFill(symbol="BTCUSDT", order_id=order_id, side=side, qty=Decimal(1), price=Decimal(price),
+                         realized_pnl=Decimal(pnl), commission=Decimal("0.1"), commission_asset="USDT", time=start + timedelta(seconds=seconds))
+             for seconds, order_id, side, price, pnl in [(1, "entry", "BUY", "100", "0"), (120, "manual", "SELL", "102", "2"),
+                                                         (130, "other", "BUY", "100", "0"), (140, "other-close", "SELL", "200", "100")]
+             if not (scenario == "missing_close" and order_id == "manual")]
+    if scenario == "partial_entry":
+        # 同一张限价单在前一笔仓位归零的同一毫秒继续成交，后一笔平仓记录尚未同步。
+        fills.insert(2, TradingFill(symbol="BTCUSDT", order_id="entry", side="BUY", qty=Decimal(1), price=Decimal(100),
+                                    realized_pnl=Decimal(0), commission=Decimal("0.1"), commission_asset="USDT", time=start + timedelta(seconds=120)))
+    class Session:
+        async def scalars(self, query):
+            return fills
+    class Client:
+        exchange = "bybit"
+        async def trades(self, symbol, since_ms):
+            assert since_ms <= int(start.timestamp() * 1000)
+            return []
+    async def save(*args):
+        pass
+    monkeypatch.setattr(trading, "save_fills", save)
+    async def check():
+        engine = trading.Executor()
+        p = SimpleNamespace(symbol="BTCUSDT", first_fill_at=start + timedelta(seconds=90))
+        entry = SimpleNamespace(purpose="entry", filled_qty=Decimal(2 if scenario == "partial_entry" else 1), exchange_id="entry", created_at=start)
+        try:
+            if scenario != "complete":
+                with pytest.raises(trading.ExchangeError, match="完整的开平仓成交"):
+                    await engine.realized(Session(), Client(), p, [entry])
+            else:
+                assert await engine.realized(Session(), Client(), p, [entry]) == {"realized_pnl": "2", "commission": "0.2", "net": "1.8"}
+        finally:
+            await engine.market.http.aclose()
+    asyncio.run(check())

backend/tests/test_trading_safety.py

@@ -0,0 +1,290 @@
+import asyncio
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from types import SimpleNamespace
+
+import pytest
+
+from alphapulse.trading.bybit import normalize_order
+from alphapulse.trading.client import ExchangeError
+from alphapulse.trading.plan import SymbolRules
+
+
+class Session:
+    def __init__(self, scalar=None, groups=()):
+        self.current = scalar
+        self.groups = iter(groups)
+        self.added = []
+
+    def add(self, row):
+        if row.id is None:
+            row.id = 100 + len(self.added)
+        self.added.append(row)
+
+    async def commit(self):
+        for row in self.added:
+            if hasattr(row, "filled_qty") and row.filled_qty is None:
+                row.filled_qty = Decimal(0)
+
+    async def scalar(self, query):
+        return self.current
+
+    async def scalars(self, query):
+        class Rows(list):
+            def all(self):
+                return self
+        return Rows(next(self.groups))
+
+
+class Client:
+    exchange = "bybit"
+
+    def __init__(self):
+        self.canceled = []
+        self.created = []
+        self.position_rows = []
+        self.entry_status = "NEW"
+
+    async def place_order(self, symbol, client_id, spec, reduce_only):
+        self.created.append(spec)
+        return {"orderId": client_id, "algoId": client_id, "status": "SUBMITTING"}
+
+    async def cancel_order(self, symbol, client_id, is_algo):
+        self.canceled.append(client_id)
+
+    async def open_orders(self, symbol, is_algo=False):
+        return []
+
+    async def order(self, symbol, client_id, is_algo=False):
+        return {"orderId": client_id, "status": self.entry_status,
+                "executedQty": "1" if self.entry_status == "FILLED" else "0", "avgPrice": "100"}
+
+    async def positions(self):
+        return self.position_rows
+
+
+def plan(trading):
+    return trading.TradingAnalysis(id=1, symbol="BTCUSDT", plan_status="pending", exchange="bybit",
+                                   sizing={"direction": "long", "entry": {"quantity": "1", "price": "100", "stop": "90"}},
+                                   valid_until=datetime.now(UTC) + timedelta(hours=1))
+
+
+def order(trading, purpose, **values):
+    return trading.TradingOrder(id=2 if purpose == "entry" else 3, analysis_id=1, exchange="bybit", symbol="BTCUSDT",
+                                purpose=purpose, side="BUY" if purpose == "entry" else "SELL", leg=0 if purpose == "entry" else None,
+                                type="LIMIT" if purpose == "entry" else "STOP_MARKET", price=Decimal(100) if purpose == "entry" else None,
+                                trigger_price=Decimal(80) if purpose == "stop" else None, quantity=Decimal(1), filled_qty=Decimal(0),
+                                is_algo=purpose == "stop", status="new", **values)
+
+
+RULES = SymbolRules(Decimal("0.1"), Decimal("0.01"), Decimal(5))
+
+
+def test_stop_replacement_waits_for_exchange_confirmation(trading):
+    async def check():
+        engine, client, session = trading.Executor(), Client(), Session()
+        try:
+            p, old = plan(trading), order(trading, "stop")
+            p.plan_status = "open"
+            position = trading.Position("long", Decimal(1), Decimal(100), Decimal(100), Decimal(101))
+            assert not await engine.sync_orders(session, client, p, [old], position, False, RULES, False)
+            assert client.canceled == []
+            [replacement] = [r for r in session.added if isinstance(r, trading.TradingOrder)]
+            assert replacement.status == "submitting"
+            replacement.status = "new"
+            assert await engine.sync_orders(session, client, p, [old, replacement], position, False, RULES, False)
+            assert client.canceled == ["ap3"]
+        finally:
+            await engine.market.http.aclose()
+    asyncio.run(check())
+
+
+def test_rejected_replacement_does_not_cancel_existing_stop(trading, monkeypatch):
+    async def reject(*args):
+        raise ExchangeError(400, 110092, "止损被拒")
+    async def check():
+        engine, client, session = trading.Executor(), Client(), Session()
+        monkeypatch.setattr(client, "place_order", reject)
+        p = plan(trading)
+        try:
+            protected = await engine.sync_orders(session, client, p, [order(trading, "stop")],
+                                                  trading.Position("long", Decimal(1), Decimal(100), Decimal(100), Decimal(101)), False, RULES, False)
+            assert not protected and client.canceled == []
+            assert p.plan_status == "closing" and p.close_reason == "stop_failed"
+        finally:
+            await engine.market.http.aclose()
+    asyncio.run(check())
+
+
+def test_stopped_pending_plan_keeps_late_fill_and_stop(trading, monkeypatch):
+    async def rules(*args):
+        return RULES
+    monkeypatch.setattr(trading, "symbol_rules", rules)
+
+    async def check():
+        engine, client = trading.Executor(), Client()
+        p, entry, stop = plan(trading), order(trading, "entry"), order(trading, "stop")
+        stop.trigger_price = Decimal(90)
+        symbol = SimpleNamespace(symbol="BTCUSDT")
+        # 首轮撤单只收到 ACK，计划不能结束，也不能撤保护止损。
+        session = Session(p, [[entry, stop], [], [entry, stop]])
+        async def sync(session, client, row, *args):
+            if row.purpose == "entry" and client.entry_status == "FILLED":
+                row.status, row.filled_qty, row.avg_price = "filled", Decimal(1), Decimal(100)
+        monkeypatch.setattr(engine, "sync_order", sync)
+        try:
+            await engine.reconcile(session, client, symbol, False, None)
+            assert p.plan_status == "closing" and p.closed_at is None
+            assert "ap2" in client.canceled and "ap3" not in client.canceled
+            client.entry_status = "FILLED"
+            client.position_rows = [{"symbol": "BTCUSDT", "positionAmt": "1", "entryPrice": "100", "breakEvenPrice": "100", "markPrice": "101"}]
+            await engine.reconcile(Session(p, [[entry, stop], [], [entry, stop]]), client, symbol, False, None)
+            assert p.plan_status == "open" and p.first_fill_at is not None
+            assert all(spec.type != "MARKET" for spec in client.created)
+            assert "ap3" not in client.canceled
+        finally:
+            await engine.market.http.aclose()
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("stop_type,direction,expected", [("TakeProfit", 1, "TAKE_PROFIT_MARKET"), ("Stop", 1, "TAKE_PROFIT_MARKET"), ("Stop", 2, "STOP_MARKET")])
+def test_bybit_conditional_profit_is_not_a_protective_stop(stop_type, direction, expected):
+    result = normalize_order({"symbol": "BTCUSDT", "orderId": "external", "orderLinkId": "", "side": "Sell", "orderType": "Market",
+                              "orderStatus": "Untriggered", "price": "", "qty": "1", "cumExecQty": "0", "avgPrice": "",
+                              "reduceOnly": True, "createdTime": "1792400000000", "triggerPrice": "110", "stopOrderType": stop_type,
+                              "triggerDirection": direction}, True)
+    assert result["orderType"] == expected
+
+
+@pytest.mark.parametrize("kind,quantity,reduce_only", [("TAKE_PROFIT_MARKET", "1", True), ("STOP_MARKET", "0.1", True), ("STOP_MARKET", "1", False)])
+def test_unmanaged_requires_full_reduce_only_stop(trading, kind, quantity, reduce_only):
+    async def check():
+        engine, session, client = trading.Executor(), Session(0), Client()
+        try:
+            await engine.unmanaged(session, client, "BTCUSDT", trading.Position("long", Decimal(1), Decimal(100), Decimal(100), Decimal(101)),
+                                   {"external": {"side": "SELL", "orderType": kind, "quantity": quantity, "reduceOnly": reduce_only, "closePosition": False}})
+            assert [spec.type for spec in client.created] == ["MARKET"]
+        finally:
+            await engine.market.http.aclose()
+    asyncio.run(check())
+
+
+def test_analysis_snapshot_rejects_recent_error_and_old_success(trading):
+    from alphapulse.trading.snapshot import SnapshotError, require_fresh_snapshot
+    snapshot = {"account": {"margin_balance": "100"}, "error": None, "synced_at": datetime.now(UTC).isoformat()}
+    assert require_fresh_snapshot(snapshot) is snapshot
+    for change in ({"error": "网络错误"}, {"synced_at": (datetime.now(UTC) - timedelta(minutes=1)).isoformat()}):
+        with pytest.raises(SnapshotError):
+            require_fresh_snapshot(snapshot | change)
+
+
+@pytest.mark.parametrize("unavailable", ["missing", "error", "stale"])
+@pytest.mark.parametrize("reasons", [["例行分析"], ["手动分析"]])
+def test_analysis_request_waits_for_fresh_snapshot(trading, monkeypatch, unavailable, reasons):
+    from alphapulse.trading import engine
+
+    requested_at = datetime.now(UTC) - timedelta(seconds=1)
+    target = trading.TradingSymbol(symbol="BTCUSDT", analysis_requested_at=requested_at, analysis_reasons=list(reasons))
+    fresh = {"account": {"margin_balance": "100"}, "error": None, "synced_at": datetime.now(UTC).isoformat()}
+    snapshot = {
+        "missing": None,
+        "error": fresh | {"error": "账户同步失败"},
+        "stale": fresh | {"synced_at": (datetime.now(UTC) - timedelta(minutes=1)).isoformat()},
+    }[unavailable]
+
+    class QueueSession:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            pass
+
+        async def scalars(self, query):
+            return SimpleNamespace(all=lambda: [target] if target.analysis_requested_at is not None else [])
+
+        async def commit(self):
+            pass
+
+    async def read_snapshot():
+        return snapshot
+
+    monkeypatch.setattr(engine, "SessionLocal", QueueSession)
+    monkeypatch.setattr(engine, "read_snapshot", read_snapshot)
+
+    async def check():
+        nonlocal snapshot
+        assert await engine.claim_requests(1, set()) == []
+        assert target.analysis_requested_at == requested_at
+        assert target.analysis_reasons == reasons
+        snapshot = fresh
+        assert await engine.claim_requests(1, set()) == [("BTCUSDT", reasons)]
+        assert target.analysis_requested_at is None and target.analysis_reasons == []
+        assert await engine.claim_requests(1, set()) == []
+
+    asyncio.run(check())
+
+
+def test_manual_request_survives_failure_before_execution(trading, monkeypatch):
+    from alphapulse.api import trading as api
+
+    async def get_symbol(*args):
+        return SimpleNamespace(symbol="BTCUSDT")
+
+    async def fresh(*args):
+        return {}
+
+    async def exchange(*args):
+        return "bybit"
+
+    async def rules(*args):
+        return RULES
+
+    monkeypatch.setattr(api, "get_symbol", get_symbol)
+    monkeypatch.setattr(api, "fresh_snapshot", fresh)
+    monkeypatch.setattr(api, "active_exchange", exchange)
+    monkeypatch.setattr(trading, "symbol_rules", rules)
+
+    async def check():
+        session = Session()
+        await api.manual_action("BTCUSDT", "cancel", session)
+        [request] = session.added
+        engine, client = trading.Executor(), Client()
+        async def unavailable(*args):
+            raise ExchangeError(408, -1007, "结果未知")
+        monkeypatch.setattr(client, "open_orders", unavailable)
+        try:
+            with pytest.raises(ExchangeError) as error:
+                await engine.reconcile(session, client, SimpleNamespace(symbol="BTCUSDT"), True, request)
+            assert error.value.code == -1007
+            assert request.kind == "manual_requested" and request.data == {"action": "cancel"}
+            assert not any(e.kind == "manual_completed" for e in session.added)
+            async def empty(*args):
+                return []
+            monkeypatch.setattr(client, "open_orders", empty)
+            recovered = Session(None, [[], []])
+            await engine.reconcile(recovered, client, SimpleNamespace(symbol="BTCUSDT"), True, request)
+            [completed] = recovered.added
+            assert completed.kind == "manual_completed" and completed.data == {"request_id": request.id}
+        finally:
+            await engine.market.http.aclose()
+    asyncio.run(check())
+
+
+def test_replacement_proposal_survives_old_plan_cancellation(trading):
+    async def check():
+        engine, client = trading.Executor(), Client()
+        current, proposal = plan(trading), plan(trading)
+        proposal.id, proposal.action, proposal.plan_status = 4, "plan", "proposed"
+        proposal.decision = {"valid_hours": 1}
+        proposal.sizing["entry"]["stop_pct"] = "1"
+        symbol = SimpleNamespace(symbol="BTCUSDT")
+        try:
+            assert await engine.adopt(Session(groups=[[proposal]]), client, symbol, current, None) is current
+            assert current.plan_status == "closing" and proposal.plan_status == "proposed"
+            assert await engine.adopt(Session(), client, symbol, current, None) is current
+            assert proposal.plan_status == "proposed"
+            assert await engine.adopt(Session(groups=[[proposal]]), client, symbol, None, None) is proposal
+            assert proposal.plan_status == "pending"
+        finally:
+            await engine.market.http.aclose()
+    asyncio.run(check())