Add demo trading page, trader snapshot service and trading tables

Add trading_symbols, trading_analyses, trading_orders and trading_events.
The new trader service publishes account, position and open-order
snapshots of the Binance demo account to Redis every 3s. The /trading
page shows engine status, start/stop, balances, positions, the symbol
list with angle selection, open orders and the event timeline.

backend/src/alphapulse/api/main.py

@@ -2,7 +2,7 @@ from fastapi import Depends, FastAPI
 from fastapi.openapi.docs import get_swagger_ui_html
 from fastapi.responses import HTMLResponse
 
-from alphapulse.api import auth, chat, crawler, market, news, settings, skills
+from alphapulse.api import auth, chat, crawler, market, news, settings, skills, trading
 from alphapulse.core.auth import require_admin
 
 # 关闭默认文档路由，改为登录后才能访问
@@ -15,6 +15,7 @@ app.include_router(market.ws_router)
 app.include_router(settings.router)
 app.include_router(skills.router)
 app.include_router(chat.router)
+app.include_router(trading.router)
 
 
 @app.get("/api/health", include_in_schema=False)

backend/src/alphapulse/api/trading.py

@@ -0,0 +1,271 @@
+"""模拟盘页：引擎状态、启停、交易币种、时间线。账户和持仓来自 trader 进程写入 Redis 的快照"""
+
+import json
+from datetime import datetime, timedelta
+from typing import Annotated, Literal
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel
+from sqlalchemy import func, or_, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from alphapulse.ai import llm
+from alphapulse.api.crawler import LOCAL_TZ
+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.trading.client import ExchangeError, public_get
+from alphapulse.trading.engine import DAILY_ANALYSIS_LIMIT, OWN_CLIENT_ID, STATUS_KEY
+
+router = APIRouter(prefix="/api/trading", tags=["trading"], dependencies=[Depends(require_admin)])
+
+Angle = Literal["short", "long"]
+ANGLE_LABELS = {"short": "短线", "long": "长线"}
+SYMBOLS_CACHE_KEY = f"{KEY_PREFIX}trading:symbols"
+SYMBOLS_CACHE_SECONDS = 3600
+# 删除或切换角度前要求快照足够新，否则无法确认没有仓位和挂单
+SNAPSHOT_FRESH_SECONDS = 30
+
+
+class TodayUsage(BaseModel):
+    analyses: int
+    limit: int
+    input_tokens: int
+    output_tokens: int
+
+
+class StatusOut(BaseModel):
+    configured: bool
+    running: bool
+    running_changed_at: datetime | None
+    provider_id: int | None
+    reasoning_effort: str | None
+    # trader 进程的心跳快照；进程离线时为空
+    snapshot: dict | None
+    today: TodayUsage
+
+
+class ConfigIn(BaseModel):
+    provider_id: int | None = None
+    reasoning_effort: llm.Effort | None = None
+
+
+class SymbolOption(BaseModel):
+    symbol: str
+    base: str
+
+
+class SymbolOut(BaseModel):
+    symbol: str
+    angle: str
+    added_at: datetime
+    analysis_requested_at: datetime | None
+    analyzing: bool
+    # 当前计划的状态：pending / open，没有计划为空
+    plan_status: str | None
+
+
+class SymbolIn(BaseModel):
+    symbol: str
+    angle: Angle
+
+
+class SymbolPatch(BaseModel):
+    angle: Angle
+
+
+class EventOut(BaseModel):
+    id: int
+    created_at: datetime
+    symbol: str | None
+    analysis_id: int | None
+    kind: str
+    level: str
+    message: str
+    data: dict | None
+
+
+async def get_account(session: AsyncSession) -> TradingAccount:
+    return await session.get(TradingAccount, 1) or TradingAccount(id=1, running=False)
+
+
+async def read_snapshot() -> dict | None:
+    raw = await redis.get(STATUS_KEY)
+    return json.loads(raw) if raw else None
+
+
+@router.get("/status")
+async def get_status(session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut:
+    account = await get_account(session)
+    today = datetime.now(LOCAL_TZ).replace(hour=0, minute=0, second=0, microsecond=0)
+    count, input_tokens, output_tokens = (await session.execute(
+        select(
+            func.count(),
+            func.coalesce(func.sum(TradingAnalysis.usage["input_tokens"].as_integer()), 0),
+            func.coalesce(func.sum(TradingAnalysis.usage["output_tokens"].as_integer()), 0),
+        ).where(TradingAnalysis.started_at >= today)
+    )).one()
+    return StatusOut(
+        configured=account.api_key_encrypted is not None, running=account.running,
+        running_changed_at=account.running_changed_at, provider_id=account.provider_id,
+        reasoning_effort=account.reasoning_effort, snapshot=await read_snapshot(),
+        today=TodayUsage(analyses=count, limit=DAILY_ANALYSIS_LIMIT, input_tokens=input_tokens, output_tokens=output_tokens),
+    )
+
+
+async def set_running(session: AsyncSession, running: bool) -> StatusOut:
+    account = await get_account(session)
+    if running and account.api_key_encrypted is None:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "还没有填写模拟盘 key，请先到环境变量页填写")
+    if account.running != running:
+        account.running = running
+        account.running_changed_at = datetime.now(LOCAL_TZ)
+        session.add(account)
+        session.add(TradingEvent(
+            kind="started" if running else "stopped",
+            message="开始行情交易" if running else "停止行情交易：不再分析，撤销未成交的入场单；已有持仓的止损、止盈等程序规则继续执行",
+        ))
+        await session.commit()
+    return await get_status(session)
+
+
+@router.post("/start")
+async def start(session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut:
+    return await set_running(session, True)
+
+
+@router.post("/stop")
+async def stop(session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut:
+    return await set_running(session, False)
+
+
+@router.patch("/config")
+async def patch_config(body: ConfigIn, session: Annotated[AsyncSession, Depends(get_session)]) -> StatusOut:
+    account = await get_account(session)
+    fields = body.model_dump(exclude_unset=True)
+    if fields.get("provider_id") is not None and await session.get(AiProvider, fields["provider_id"]) is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "模型配置不存在")
+    for key, value in fields.items():
+        setattr(account, key, value)
+    session.add(account)
+    await session.commit()
+    return await get_status(session)
+
+
+async def symbol_options() -> list[dict]:
+    """可交易的币：模拟盘和实盘都在交易的 USDT 永续，只含加密货币（排除美股、商品等 TradFi）"""
+    if cached := await redis.get(SYMBOLS_CACHE_KEY):
+        return json.loads(cached)
+
+    def crypto_perpetuals(info: dict) -> dict[str, str]:
+        return {
+            s["symbol"]: s["baseAsset"] for s in info["symbols"]
+            if s["contractType"] == "PERPETUAL" and s.get("underlyingType") == "COIN"
+            and s["status"] == "TRADING" and s["quoteAsset"] == "USDT"
+        }
+
+    try:
+        demo = crypto_perpetuals(await public_get("/fapi/v1/exchangeInfo"))
+    except ExchangeError as e:
+        raise HTTPException(status.HTTP_502_BAD_GATEWAY, str(e)) from None
+    # 分析用实盘行情，执行在模拟盘，两边都要有
+    mainnet = crypto_perpetuals(await call_binance(market_client.get("um", "exchange_info")))
+    options = [{"symbol": s, "base": b} for s, b in sorted(demo.items()) if s in mainnet]
+    await redis.set(SYMBOLS_CACHE_KEY, json.dumps(options), ex=SYMBOLS_CACHE_SECONDS)
+    return options
+
+
+@router.get("/symbol-options")
+async def list_symbol_options() -> list[SymbolOption]:
+    return [SymbolOption.model_validate(o) for o in await symbol_options()]
+
+
+@router.get("/symbols")
+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(
+        select(TradingAnalysis.symbol, TradingAnalysis.plan_status)
+        .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),
+        )
+        for s in symbols
+    ]
+
+
+@router.post("/symbols", status_code=status.HTTP_201_CREATED)
+async def add_symbol(body: SymbolIn, session: Annotated[AsyncSession, Depends(get_session)]) -> None:
+    symbol = body.symbol.strip().upper()
+    if not any(o["symbol"] == symbol for o in await symbol_options()):
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"{symbol} 不是模拟盘和实盘都在交易的加密货币 USDT 永续合约")
+    session.add(TradingSymbol(symbol=symbol, angle=body.angle))
+    session.add(TradingEvent(symbol=symbol, kind="symbol_added", message=f"加入交易列表，角度：{ANGLE_LABELS[body.angle]}"))
+    try:
+        await session.commit()
+    except IntegrityError:
+        raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 已在交易列表中") from None
+
+
+async def ensure_idle(session: AsyncSession, symbol: str, action: str) -> TradingSymbol:
+    """只有空仓、没有本程序的挂单、没有进行中的计划和分析时，才能删除或切换角度"""
+    s = await session.get(TradingSymbol, symbol)
+    if s is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"{symbol} 不在交易列表中")
+    busy = await session.scalar(select(func.count()).where(
+        TradingAnalysis.symbol == symbol,
+        or_(TradingAnalysis.status == "running",
+            TradingAnalysis.plan_status.in_(("pending", "open")) & TradingAnalysis.dry_run.is_(False)),
+    ))
+    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}")
+    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"]):
+        raise HTTPException(status.HTTP_409_CONFLICT, f"{symbol} 有本程序的挂单，不能{action}")
+    return s
+
+
+@router.patch("/symbols/{symbol}")
+async def patch_symbol(symbol: str, body: SymbolPatch, session: Annotated[AsyncSession, Depends(get_session)]) -> None:
+    s = await ensure_idle(session, symbol, "切换角度")
+    if s.angle != body.angle:
+        session.add(TradingEvent(symbol=symbol, kind="angle_changed",
+                                 message=f"角度从{ANGLE_LABELS[s.angle]}改为{ANGLE_LABELS[body.angle]}"))
+        s.angle = body.angle
+        await session.commit()
+
+
+@router.delete("/symbols/{symbol}")
+async def delete_symbol(symbol: str, session: Annotated[AsyncSession, Depends(get_session)]) -> None:
+    """只移出列表；分析、订单流水和时间线保留"""
+    s = await ensure_idle(session, symbol, "删除")
+    await session.delete(s)
+    session.add(TradingEvent(symbol=symbol, kind="symbol_removed", message="移出交易列表"))
+    await session.commit()
+
+
+@router.get("/events")
+async def list_events(
+    session: Annotated[AsyncSession, Depends(get_session)],
+    symbol: str | None = None,
+    before: int | None = None,
+    limit: Annotated[int, Query(ge=1, le=200)] = 50,
+) -> list[EventOut]:
+    """按 id 倒序分页；指定 symbol 时同时返回账户级事件（启停等）"""
+    query = select(TradingEvent).order_by(TradingEvent.id.desc()).limit(limit)
+    if symbol:
+        query = query.where(or_(TradingEvent.symbol == symbol, TradingEvent.symbol.is_(None)))
+    if before:
+        query = query.where(TradingEvent.id < before)
+    return [EventOut.model_validate(e, from_attributes=True) for e in await session.scalars(query)]

backend/src/alphapulse/models/__init__.py

@@ -2,7 +2,7 @@ from alphapulse.core.db import Base
 from alphapulse.models.ai import AiProvider, ChatMessage, ChatSession, ContextVersion, SkillNode, SkillVersion
 from alphapulse.models.market import Kline, KlineSyncState, MarketSymbol
 from alphapulse.models.news import CrawlLog, CrawlRun, NewsItem, NewsRevision, RawBody, RawResponse, Source
-from alphapulse.models.trading import TradingAccount
+from alphapulse.models.trading import TradingAccount, TradingAnalysis, TradingEvent, TradingOrder, TradingSymbol
 
 __all__ = [
     "AiProvider",
@@ -23,4 +23,8 @@ __all__ = [
     "SkillVersion",
     "Source",
     "TradingAccount",
+    "TradingAnalysis",
+    "TradingEvent",
+    "TradingOrder",
+    "TradingSymbol",
 ]

backend/src/alphapulse/models/trading.py

@@ -1,6 +1,8 @@
 from datetime import datetime
+from decimal import Decimal
 
-from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, LargeBinary, String, func
+from sqlalchemy import BigInteger, Boolean, CheckConstraint, DateTime, ForeignKey, Index, Integer, LargeBinary, Numeric, String, Text, func
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
 from alphapulse.core.db import Base
@@ -22,3 +24,104 @@ class TradingAccount(Base):
     provider_id: Mapped[int | None] = mapped_column(ForeignKey("ai_providers.id", ondelete="SET NULL"))
     reasoning_effort: Mapped[str | None] = mapped_column(String(8))
     updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class TradingSymbol(Base):
+    __tablename__ = "trading_symbols"
+
+    symbol: Mapped[str] = mapped_column(String(32), primary_key=True)
+    # short：24h 内平仓；long：12–48h 平仓
+    angle: Mapped[str] = mapped_column(String(8))
+    added_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    # 待执行的分析请求：多个触发在执行前合并为一次
+    analysis_requested_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    analysis_reasons: Mapped[list[str]] = mapped_column(JSONB, default=list)
+
+
+class TradingAnalysis(Base):
+    """一次模型分析；action=plan 的行同时是一个交易计划，plan_* 等字段记录其生命周期"""
+
+    __tablename__ = "trading_analyses"
+    __table_args__ = (Index("ix_trading_analyses_symbol_started", "symbol", "started_at"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    symbol: Mapped[str] = mapped_column(String(32))
+    angle: Mapped[str] = mapped_column(String(8))
+    reasons: Mapped[list[str]] = mapped_column(JSONB)
+    # 停止状态下的分析只记录计划、不下单
+    dry_run: Mapped[bool] = mapped_column(Boolean)
+    # running / done / failed
+    status: Mapped[str] = mapped_column(String(16), default="running")
+    # plan / keep / exit
+    action: Mapped[str | None] = mapped_column(String(8))
+    # submit_trade_plan 的参数原样保存
+    decision: Mapped[dict | None] = mapped_column(JSONB)
+    # 程序算出的各档数量、风险金额、盈亏比等
+    sizing: Mapped[dict | None] = mapped_column(JSONB)
+    text: Mapped[str | None] = mapped_column(Text)
+    transcript: Mapped[list | None] = mapped_column(JSONB)
+    provider_id: Mapped[int | None] = mapped_column(ForeignKey("ai_providers.id", ondelete="SET NULL"))
+    model: Mapped[str | None] = mapped_column(String(128))
+    reasoning_effort: Mapped[str | None] = mapped_column(String(8))
+    usage: Mapped[dict | None] = mapped_column(JSONB)
+    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（被新计划替换）
+    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))
+    close_deadline: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    close_reason: Mapped[str | None] = mapped_column(String(32))
+    # 平仓后的已实现盈亏、手续费
+    result: Mapped[dict | None] = mapped_column(JSONB)
+
+
+class TradingOrder(Base):
+    """订单流水：先写行再发单，clientOrderId / clientAlgoId 为 ap{id}，超时后按它查询而不是换 id 重发"""
+
+    __tablename__ = "trading_orders"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    analysis_id: Mapped[int | None] = mapped_column(ForeignKey("trading_analyses.id"), index=True)
+    symbol: Mapped[str] = mapped_column(String(32), index=True)
+    # entry / tp / stop / close
+    purpose: Mapped[str] = mapped_column(String(8))
+    # 入场或止盈的档位序号，从 0 开始
+    leg: Mapped[int | None] = mapped_column(Integer)
+    # 条件单走 /fapi/v1/algoOrder
+    is_algo: Mapped[bool] = mapped_column(Boolean)
+    side: Mapped[str] = mapped_column(String(4))
+    type: Mapped[str] = mapped_column(String(24))
+    price: Mapped[Decimal | None] = mapped_column(Numeric)
+    trigger_price: Mapped[Decimal | None] = mapped_column(Numeric)
+    quantity: Mapped[Decimal] = mapped_column(Numeric)
+    reduce_only: Mapped[bool] = mapped_column(Boolean)
+    # submitting / new / partially_filled / filled / canceled / expired / rejected / finished / failed
+    status: Mapped[str] = mapped_column(String(16), default="submitting")
+    # orderId 或 algoId
+    exchange_id: Mapped[str | None] = mapped_column(String(32))
+    filled_qty: Mapped[Decimal] = mapped_column(Numeric, default=0)
+    avg_price: Mapped[Decimal | None] = mapped_column(Numeric)
+    error: Mapped[str | None] = mapped_column(Text)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class TradingEvent(Base):
+    """只追加的时间线"""
+
+    __tablename__ = "trading_events"
+    __table_args__ = (Index("ix_trading_events_symbol_id", "symbol", "id"),)
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    # 账户级事件（启停等）为空
+    symbol: Mapped[str | None] = mapped_column(String(32))
+    analysis_id: Mapped[int | None] = mapped_column(ForeignKey("trading_analyses.id"))
+    kind: Mapped[str] = mapped_column(String(32))
+    # info / warn / error
+    level: Mapped[str] = mapped_column(String(8), default="info")
+    message: Mapped[str] = mapped_column(Text)
+    data: Mapped[dict | None] = mapped_column(JSONB)

backend/src/alphapulse/trading/client.py

@@ -36,6 +36,18 @@ class ExchangeError(Exception):
         return f"币安返回 {self.status}（{self.code}）：{self.message}"
 
 
+async def public_get(path: str, params: dict[str, Any] | None = None) -> Any:
+    """模拟盘公开接口（exchangeInfo 等），不需要 key"""
+    async with httpx.AsyncClient(base_url=BASE_URL, timeout=TIMEOUT_SECONDS) as http:
+        try:
+            resp = await http.get(path, params=params)
+        except httpx.TransportError as e:
+            raise ExchangeError(None, None, f"无法连接币安模拟盘：{e!r}") from e
+    if resp.status_code >= 400:
+        raise ExchangeError(resp.status_code, None, resp.text[:200])
+    return resp.json()
+
+
 class DemoClient:
     def __init__(self, api_key: str, api_secret: str) -> None:
         self.secret = api_secret.encode()

backend/src/alphapulse/trading/engine.py

@@ -0,0 +1,159 @@
+"""模拟盘交易进程：定时把账户、持仓、挂单快照写入 Redis，供页面读取。
+
+用法：python -m alphapulse.trading.engine
+"""
+
+import asyncio
+import json
+import logging
+import re
+import time
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+
+from alphapulse.core.crypto import decrypt
+from alphapulse.core.db import SessionLocal
+from alphapulse.core.redis import KEY_PREFIX, redis
+from alphapulse.models import TradingAccount, TradingSymbol
+from alphapulse.trading.client import DemoClient, ExchangeError
+
+logger = logging.getLogger("alphapulse.trading.engine")
+
+# 进程挂掉后快照自动过期，页面据此显示「引擎离线」
+STATUS_KEY = f"{KEY_PREFIX}trader:status"
+STATUS_TTL_SECONDS = 15
+SNAPSHOT_SECONDS = 3
+# 杠杆和保证金模式很少变，按币缓存
+SYMBOL_CONFIG_SECONDS = 60
+# 所有币合计每天（北京时间）最多自动分析的次数
+DAILY_ANALYSIS_LIMIT = 60
+# 本程序下的单 clientOrderId / clientAlgoId 为 ap{流水 id}；其他（含用户手动下的）不动
+OWN_CLIENT_ID = re.compile(r"^ap\d+$")
+
+
+class Snapshot:
+    def __init__(self) -> None:
+        self.client: DemoClient | None = None
+        self.key_ciphertext: bytes | None = None
+        # 最近一次成功同步的数据；同步失败时保留并标注错误，避免页面闪空
+        self.data: dict | None = None
+        self.synced_at: datetime | None = None
+        self.symbol_config: dict[str, tuple[float, dict]] = {}
+
+    async def ensure_client(self, account: TradingAccount | None) -> DemoClient | None:
+        ciphertext = account.api_key_encrypted if account else None
+        if ciphertext != self.key_ciphertext:
+            # key 被修改或删除：旧连接作废，旧数据属于另一个账户，一并清掉
+            if self.client is not None:
+                await self.client.http.aclose()
+            self.client = None
+            self.data = self.synced_at = None
+            self.symbol_config.clear()
+            self.key_ciphertext = ciphertext
+            if account is not None and ciphertext is not None:
+                self.client = DemoClient(decrypt(ciphertext), decrypt(account.api_secret_encrypted))
+        return self.client
+
+    async def leverage_and_margin(self, client: DemoClient, symbol: str) -> dict:
+        cached = self.symbol_config.get(symbol)
+        if cached is None or time.monotonic() - cached[0] > SYMBOL_CONFIG_SECONDS:
+            config = (await client.request("GET", "/fapi/v1/symbolConfig", {"symbol": symbol}))[0]
+            cached = (time.monotonic(), {"leverage": config["leverage"], "margin_type": config["marginType"]})
+            self.symbol_config[symbol] = cached
+        return cached[1]
+
+    async def fetch(self, client: DemoClient, symbols: list[str]) -> dict:
+        account = await client.request("GET", "/fapi/v3/account")
+        positions = [p for p in await client.request("GET", "/fapi/v3/positionRisk") if float(p["positionAmt"]) != 0]
+        # 挂单只查交易列表里的币和有持仓的币：不带 symbol 查全部的权重是 40，3 秒一次太贵
+        watched = sorted(set(symbols) | {p["symbol"] for p in positions})
+        orders, algo_orders = [], []
+        for symbol in watched:
+            orders += await client.request("GET", "/fapi/v1/openOrders", {"symbol": symbol})
+            algo_orders += await client.request("GET", "/fapi/v1/openAlgoOrders", {"symbol": symbol})
+        return {
+            "account": {
+                "wallet_balance": account["totalWalletBalance"],
+                "margin_balance": account["totalMarginBalance"],
+                "available_balance": account["availableBalance"],
+                "unrealized_pnl": account["totalUnrealizedProfit"],
+            },
+            "positions": [
+                {
+                    "symbol": p["symbol"],
+                    "side": "LONG" if float(p["positionAmt"]) > 0 else "SHORT",
+                    "amount": p["positionAmt"].lstrip("-"),
+                    "entry_price": p["entryPrice"],
+                    "break_even_price": p["breakEvenPrice"],
+                    "mark_price": p["markPrice"],
+                    "unrealized_pnl": p["unRealizedProfit"],
+                    "liquidation_price": p["liquidationPrice"],
+                    "notional": p["notional"].lstrip("-"),
+                    "margin": p["isolatedMargin"] if float(p["isolatedMargin"]) else p["initialMargin"],
+                    **await self.leverage_and_margin(client, p["symbol"]),
+                }
+                for p in positions
+            ],
+            "orders": [
+                {
+                    "symbol": o["symbol"], "client_id": o["clientOrderId"], "side": o["side"], "type": o["type"],
+                    "price": o["price"], "quantity": o["origQty"], "filled": o["executedQty"], "reduce_only": o["reduceOnly"],
+                    "time": o["time"],
+                }
+                for o in orders
+            ],
+            "algo_orders": [
+                {
+                    "symbol": o["symbol"], "client_id": o["clientAlgoId"], "side": o["side"], "type": o["orderType"],
+                    "trigger_price": o["triggerPrice"], "quantity": o["quantity"], "close_position": o["closePosition"],
+                    "reduce_only": o["reduceOnly"], "time": o["createTime"],
+                }
+                for o in algo_orders
+            ],
+        }
+
+    async def publish(self) -> None:
+        async with SessionLocal() as session:
+            account = await session.get(TradingAccount, 1)
+            symbols = list((await session.scalars(select(TradingSymbol.symbol))).all())
+        error = None
+        client = await self.ensure_client(account)
+        if client is None:
+            error = "还没有填写模拟盘 key"
+        else:
+            try:
+                self.data = await self.fetch(client, symbols)
+                self.synced_at = datetime.now(UTC)
+            except ExchangeError as e:
+                logger.warning("同步模拟盘失败：%s", e)
+                error = str(e)
+        status = {
+            "updated_at": datetime.now(UTC).isoformat(),
+            "synced_at": self.synced_at.isoformat() if self.synced_at else None,
+            "error": error,
+            **(self.data or {"account": None, "positions": [], "orders": [], "algo_orders": []}),
+        }
+        await redis.set(STATUS_KEY, json.dumps(status), ex=STATUS_TTL_SECONDS)
+
+
+async def snapshot_loop() -> None:
+    snapshot = Snapshot()
+    while True:
+        try:
+            await snapshot.publish()
+        except Exception:
+            logger.exception("发布快照失败，稍后重试")
+        await asyncio.sleep(SNAPSHOT_SECONDS)
+
+
+async def main() -> None:
+    # 分析器和执行器在后续里程碑加入同一个 TaskGroup
+    async with asyncio.TaskGroup() as tg:
+        tg.create_task(snapshot_loop())
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    logging.getLogger("httpx").setLevel(logging.WARNING)
+    asyncio.run(main())