diff --git a/tests/trader/test_d4a_integration.py b/tests/trader/test_d4a_integration.py
new file mode 100644
index 0000000..52cfdcb
--- /dev/null
+++ b/tests/trader/test_d4a_integration.py
@@ -0,0 +1,152 @@
+"""D-4a 集成测试：sanguo → bridge 真实 HTTP round-trip（mock bridge server）。
+
+不依赖 Windows——本地起 mock HTTP server 模拟 bridge（health/order/account/positions + 鉴权），
+验证 BridgeClient 真发 HTTP（urlopen + X-Bridge-Token + JSON 解析）+ _shadow_trades_to_bridge
+完整链路在真实 HTTP 环境跑通。
+
+验证范围（D-4a sanguo 半边）：
+- bridge_client HTTP 层真发请求、带鉴权 header、解析响应
+- _shadow_trades_to_bridge → bridge_client → 真 HTTP → mock bridge 收到下单 → paper_shadow_orders
+
+未覆盖（D-4a 另一半，物理依赖 Windows）：
+- bridge → xtquant → miniQMT 真报单（miniQMT 只 Windows，需 Windows bridge 实跑）
+"""
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from types import SimpleNamespace
+from urllib.parse import urlparse
+
+from sanguo_trader.bridge_client import BridgeClient
+from sanguo_trader.live_orchestrator import _shadow_trades_to_bridge
+from sanguo_trader.persistence import init_db, is_trade_shadowed, save_trade
+
+_TOKEN = "integration-test-token"
+
+
+class _MockBridgeHandler(BaseHTTPRequestHandler):
+    """模拟 Windows bridge 的 4 接口 + token 鉴权（契约同 sanguo_qmt_bridge/bridge.py）。"""
+
+    def log_message(self, *args):  # 静默测试日志
+        pass
+
+    def _has_token(self) -> bool:
+        return self.headers.get("X-Bridge-Token") == _TOKEN
+
+    def _respond(self, code: int, payload: dict) -> None:
+        body = json.dumps(payload).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        path = urlparse(self.path).path
+        if path == "/health":
+            self._respond(200, {"status": "ok", "miniqmt_connected": True})
+        elif path in ("/account", "/positions"):
+            if not self._has_token():
+                self._respond(401, {"detail": "token 无效"})
+                return
+            self._respond(200, {"ok": True, "cash": 1e7, "frozen": 0.0,
+                                "market_value": 0.0, "total": 1e7, "positions": []})
+        else:
+            self._respond(404, {"detail": "not found"})
+
+    def do_POST(self):
+        path = urlparse(self.path).path
+        if path != "/order":
+            self._respond(404, {"detail": "not found"})
+            return
+        if not self._has_token():
+            self._respond(401, {"detail": "token 无效"})
+            return
+        length = int(self.headers.get("Content-Length", 0))
+        req = json.loads(self.rfile.read(length))
+        self.server.orders.append(req)  # 记录，供测试断言
+        self._respond(200, {"ok": True, "order_id": 99999})
+
+
+class _MockBridge:
+    """mock bridge HTTP server（后台线程）。"""
+
+    def __init__(self):
+        self.server = HTTPServer(("127.0.0.1", 0), _MockBridgeHandler)
+        self.server.orders = []
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+
+    def start(self) -> "_MockBridge":
+        self.thread.start()
+        return self
+
+    def stop(self) -> None:
+        self.server.shutdown()
+
+    @property
+    def url(self) -> str:
+        return f"http://127.0.0.1:{self.port}"
+
+    @property
+    def orders(self) -> list:
+        """mock bridge 收到的下单请求（handler 写入 server.orders）。"""
+        return self.server.orders
+
+
+class TestBridgeClientRealHttp:
+    """BridgeClient 真发 HTTP round-trip（验证 HTTP 层 + 鉴权 + 解析）。"""
+
+    def test_get_account_with_token(self):
+        b = _MockBridge().start()
+        try:
+            r = BridgeClient(b.url, _TOKEN).get_account()
+            assert r is not None and r["ok"] is True and r["cash"] == 1e7
+        finally:
+            b.stop()
+
+    def test_order_real_http_payload(self):
+        b = _MockBridge().start()
+        try:
+            r = BridgeClient(b.url, _TOKEN).place_order("sh600000", "buy", 10.5, 100)
+            assert r == {"ok": True, "order_id": 99999}
+            assert b.orders == [{"code": "sh600000", "action": "buy", "price": 10.5,
+                                 "volume": 100, "price_type": "limit", "reason": ""}]
+        finally:
+            b.stop()
+
+    def test_wrong_token_returns_none_on_401(self):
+        b = _MockBridge().start()
+        try:
+            # 错 token → mock 返回 401 → urlopen 抛 HTTPError → bridge_client 返回 None
+            r = BridgeClient(b.url, "wrong-token").get_account()
+            assert r is None
+        finally:
+            b.stop()
+
+
+def _cfg(url: str) -> SimpleNamespace:
+    return SimpleNamespace(live={"enabled": True, "bridge_url": url, "shadow": True})
+
+
+class TestShadowTradesRealHttp:
+    """_shadow_trades_to_bridge 真发 HTTP 到 mock bridge（D-4a sanguo 半边端到端）。"""
+
+    def test_full_shadow_roundtrip(self, tmp_path, monkeypatch):
+        db = str(tmp_path / "t.db")
+        init_db(db)
+        trade_id = save_trade(db, 1, {"symbol": "600000", "direction": "buy", "price": 10.5,
+                                      "volume": 100, "bar_date": "2026-07-11",
+                                      "strategy_id": "s1"})
+        monkeypatch.setenv("BRIDGE_TOKEN", _TOKEN)
+        b = _MockBridge().start()
+        try:
+            _shadow_trades_to_bridge(db, 1, "2026-07-11", _cfg(b.url))
+            # mock bridge 真收到下单（symbol 已转 sh600000）
+            assert len(b.orders) == 1
+            assert b.orders[0]["code"] == "sh600000"
+            assert b.orders[0]["action"] == "buy"
+            # paper_shadow_orders 真记录
+            assert is_trade_shadowed(db, 1, trade_id)
+        finally:
+            b.stop()