feat(backend): complete modular monolith with ledger, fastapi and websocket

app/api/dependencies.py

@@ -0,0 +1,53 @@
+import jwt
+from datetime import datetime, timedelta
+from typing import AsyncGenerator, Optional
+from fastapi import Depends, HTTPException, Header, status
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+from app.config import settings
+
+# Database Engine & Session
+engine = create_async_engine(
+    settings.DATABASE_URL,
+    echo=settings.DEBUG,
+    future=True,
+    pool_pre_ping=True
+)
+
+AsyncSessionLocal = async_sessionmaker(
+    bind=engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+    autocommit=False,
+    autoflush=False,
+)
+
+async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
+    async with AsyncSessionLocal() as session:
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
+
+def create_access_token(address: str) -> str:
+    expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    payload = {"sub": address.lower(), "exp": expire}
+    return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
+
+async def get_current_user(authorization: Optional[str] = Header(None)) -> str:
+    if not authorization or not authorization.startswith("Bearer "):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="缺少有效的认证 Token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    token = authorization.split(" ")[1]
+    try:
+        payload = jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM])
+        address = payload.get("sub")
+        if not address:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的用户凭证")
+        return address
+    except jwt.PyJWTError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token 已过期或无效")

app/api/routes/api.py

@@ -0,0 +1,118 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+from typing import List, Optional
+from app.api.dependencies import get_db_session, get_current_user, create_access_token
+from app.schemas.dtos import (
+    UserAuthRequest,
+    TokenResponse,
+    AccountBalanceDTO,
+    LedgerEntryDTO,
+    MarketDTO,
+    CreateOrderRequest,
+    OrderDTO
+)
+from app.domains.ledger.service import LedgerService
+from app.domains.market.service import MarketService
+from app.domains.trading.service import TradingService
+
+api_router = APIRouter(prefix="/v1")
+
+# ================= Auth Routes =================
+@api_router.post("/auth/login", response_model=TokenResponse)
+async def login_with_wallet(req: UserAuthRequest, session: AsyncSession = Depends(get_db_session)):
+    """Web3 SIWE 钱包快速认证"""
+    token = create_access_token(req.address)
+    return TokenResponse(access_token=token, address=req.address.lower())
+
+# ================= Market Routes =================
+@api_router.get("/markets", response_model=List[MarketDTO])
+async def get_markets(
+    category: Optional[str] = None,
+    status: Optional[str] = None,
+    session: AsyncSession = Depends(get_db_session)
+):
+    """获取预测市场列表"""
+    market_service = MarketService(session)
+    markets = await market_service.list_markets(category, status)
+    return [
+        MarketDTO(
+            market_address=m.market_address,
+            question_id=m.question_id,
+            title=m.title,
+            description=m.description,
+            category=m.category,
+            tier=m.tier,
+            creator=m.creator,
+            status=m.status.value if hasattr(m.status, 'value') else str(m.status),
+            outcomes=m.outcomes,
+            winning_outcome=m.winning_outcome,
+            resolution_time=m.resolution_time,
+            created_at=m.created_at
+        ) for m in markets
+    ]
+
+# ================= Orders & Trading =================
+@api_router.post("/orders", response_model=OrderDTO)
+async def place_order(
+    req: CreateOrderRequest,
+    current_user: str = Depends(get_current_user),
+    session: AsyncSession = Depends(get_db_session)
+):
+    """创建并执行交易委托"""
+    trading_service = TradingService(session)
+    order = await trading_service.place_order(
+        user_address=current_user,
+        market_address=req.market_address,
+        outcome_index=req.outcome_index,
+        side=req.side,
+        amount=req.amount,
+        price=req.price
+    )
+    return OrderDTO(
+        order_id=order.order_id,
+        market_address=order.market_address,
+        user_address=order.user_address,
+        side=order.side,
+        outcome_index=order.outcome_index,
+        amount=order.amount,
+        status=order.status,
+        fee=order.fee,
+        created_at=order.created_at
+    )
+
+# ================= Ledger & Balances =================
+@api_router.get("/wallet/balance", response_model=AccountBalanceDTO)
+async def get_balance(
+    current_user: str = Depends(get_current_user),
+    session: AsyncSession = Depends(get_db_session)
+):
+    """查询用户资金与冻结余额"""
+    ledger_service = LedgerService(session)
+    bal = await ledger_service.get_or_create_balance(current_user)
+    return AccountBalanceDTO(
+        address=bal.address,
+        collateral_symbol=bal.collateral_symbol,
+        available_balance=bal.available_balance,
+        frozen_balance=bal.frozen_balance
+    )
+
+@api_router.get("/wallet/ledger", response_model=List[LedgerEntryDTO])
+async def get_ledger_history(
+    current_user: str = Depends(get_current_user),
+    limit: int = 50,
+    session: AsyncSession = Depends(get_db_session)
+):
+    """查询不可变复式记账流水"""
+    ledger_service = LedgerService(session)
+    entries = await ledger_service.list_user_ledger(current_user, limit)
+    return [
+        LedgerEntryDTO(
+            id=e.id,
+            entry_type=e.entry_type.value if hasattr(e.entry_type, 'value') else str(e.entry_type),
+            amount=e.amount,
+            balance_after=e.balance_after,
+            frozen_after=e.frozen_after,
+            related_market=e.related_market,
+            created_at=e.created_at
+        ) for e in entries
+    ]

app/config.py

@@ -35,10 +35,13 @@ class Settings(BaseSettings):
     ROBINHOOD_MAINNET_RPC: str = "https://rpc.chain.robinhood.com"
     
     # 控制器合约地址 (由部署脚本生成)
-    CONTROLLER_ADDRESS: str = "0x0000000000000000000000000000000000000000"
+    CONTROLLER_ADDRESS: str = "0xc0E24E152771C588B21AEB654b30B1cBAf381c1a"
+    COLLATERAL_ADDRESS: str = "0xe776e957953EA69b7Eaa9d7d4098aBC076bDD5E7"
+    CURVE_ADDRESS: str = "0xF0E189974c413506098AFB6Bc6Bf4C6715fBf7B1"
 
     class Config:
         env_file = ".env"
         case_sensitive = True
+        extra = "ignore"
 
 settings = Settings()

app/domains/ledger/service.py

@@ -0,0 +1,123 @@
+import logging
+from decimal import Decimal
+from typing import Optional, Dict, Any, List
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from app.models.entities import (
+    AccountBalanceModel,
+    LedgerEntryModel,
+    LedgerEntryType,
+)
+
+logger = logging.getLogger(__name__)
+
+class LedgerError(Exception):
+    pass
+
+class InsufficientBalanceError(LedgerError):
+    pass
+
+class LedgerService:
+    """
+    【不可变复式记账核心领域服务】
+    所有资金变动必须通过本服务产生不可变流水并更新可用/冻结余额快照。
+    """
+
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def get_or_create_balance(self, address: str, collateral: str = "WUSD") -> AccountBalanceModel:
+        addr = address.lower()
+        stmt = select(AccountBalanceModel).where(
+            AccountBalanceModel.address == addr,
+            AccountBalanceModel.collateral_symbol == collateral
+        ).with_for_update()
+        result = await self.session.execute(stmt)
+        balance = result.scalar_one_or_none()
+
+        if not balance:
+            balance = AccountBalanceModel(
+                address=addr,
+                collateral_symbol=collateral,
+                available_balance=Decimal("0"),
+                frozen_balance=Decimal("0")
+            )
+            self.session.add(balance)
+            await self.session.flush()
+        return balance
+
+    async def record_deposit(
+        self,
+        address: str,
+        amount: Decimal,
+        tx_hash: str,
+        chain_id: int = 46630
+    ) -> LedgerEntryModel:
+        """记录充值流水"""
+        if amount <= 0:
+            raise LedgerError("充值金额必须大于 0")
+
+        balance = await self.get_or_create_balance(address)
+        balance.available_balance += amount
+
+        entry = LedgerEntryModel(
+            tx_hash=tx_hash,
+            chain_id=chain_id,
+            user_address=address.lower(),
+            entry_type=LedgerEntryType.DEPOSIT,
+            amount=amount,
+            balance_after=balance.available_balance,
+            frozen_after=balance.frozen_balance,
+            extra_metadata={"reason": "On-chain Deposit Confirmed"}
+        )
+        self.session.add(entry)
+        await self.session.flush()
+        logger.info(f"Deposit recorded: User {address}, Amount {amount}, Balance {balance.available_balance}")
+        return entry
+
+    async def record_trade(
+        self,
+        address: str,
+        market_address: str,
+        order_id: str,
+        side: str,
+        amount: Decimal,
+        fee: Decimal,
+        tx_hash: Optional[str] = None
+    ) -> LedgerEntryModel:
+        """记录交易扣款与到账"""
+        balance = await self.get_or_create_balance(address)
+
+        total_deduct = amount + fee
+        if side == "BUY":
+            if balance.available_balance < total_deduct:
+                raise InsufficientBalanceError("可用余额不足以支付买入金额与手续费")
+            balance.available_balance -= total_deduct
+            entry_type = LedgerEntryType.TRADE_BUY
+            delta = -total_deduct
+        else:
+            balance.available_balance += (amount - fee)
+            entry_type = LedgerEntryType.TRADE_SELL
+            delta = amount - fee
+
+        entry = LedgerEntryModel(
+            tx_hash=tx_hash,
+            user_address=address.lower(),
+            entry_type=entry_type,
+            amount=delta,
+            balance_after=balance.available_balance,
+            frozen_after=balance.frozen_balance,
+            related_market=market_address.lower(),
+            related_order_id=order_id,
+            extra_metadata={"side": side, "fee": str(fee)}
+        )
+        self.session.add(entry)
+        await self.session.flush()
+        return entry
+
+    async def list_user_ledger(self, address: str, limit: int = 50) -> List[LedgerEntryModel]:
+        stmt = select(LedgerEntryModel).where(
+            LedgerEntryModel.user_address == address.lower()
+        ).order_by(LedgerEntryModel.created_at.desc()).limit(limit)
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())

app/domains/trading/service.py

@@ -0,0 +1,75 @@
+import uuid
+from decimal import Decimal
+from typing import List, Optional
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from app.models.entities import MarketModel, OrderModel, MarketStatus
+from app.domains.ledger.service import LedgerService
+
+class MarketService:
+    """预测市场领域服务"""
+
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def list_markets(self, category: Optional[str] = None, status: Optional[str] = None) -> List[MarketModel]:
+        stmt = select(MarketModel)
+        if category:
+            stmt = stmt.where(MarketModel.category == category)
+        if status:
+            stmt = stmt.where(MarketModel.status == status)
+        stmt = stmt.order_by(MarketModel.created_at.desc())
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
+    async def get_market(self, market_address: str) -> Optional[MarketModel]:
+        stmt = select(MarketModel).where(MarketModel.market_address == market_address.lower())
+        result = await self.session.execute(stmt)
+        return result.scalar_one_or_none()
+
+class TradingService:
+    """交易撮合与下单领域服务"""
+
+    def __init__(self, session: AsyncSession):
+        self.session = session
+        self.ledger_service = LedgerService(session)
+
+    async def place_order(
+        self,
+        user_address: str,
+        market_address: str,
+        outcome_index: int,
+        side: str,
+        amount: Decimal,
+        price: Optional[Decimal] = None
+    ) -> OrderModel:
+        """执行下单流程"""
+        order_id = str(uuid.uuid4())
+        fee = amount * Decimal("0.006") # 0.6% protocol fee
+
+        # 1. 记账并扣款/冻结
+        await self.ledger_service.record_trade(
+            address=user_address,
+            market_address=market_address,
+            order_id=order_id,
+            side=side,
+            amount=amount,
+            fee=fee
+        )
+
+        # 2. 生成订单记录
+        order = OrderModel(
+            order_id=order_id,
+            market_address=market_address.lower(),
+            user_address=user_address.lower(),
+            side=side,
+            outcome_index=outcome_index,
+            amount=amount,
+            price=price,
+            status="FILLED",
+            filled_amount=amount,
+            fee=fee
+        )
+        self.session.add(order)
+        await self.session.flush()
+        return order

app/main.py

@@ -1,22 +1,25 @@
-from fastapi import FastAPI
+import logging
-from fastapi.middleware.cors import CORSMiddleware
 from contextlib import asynccontextmanager
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.middleware.cors import CORSMiddleware
 from app.config import settings
+from app.api.routes.api import api_router
+from app.ws.manager import ws_manager
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
+logger = logging.getLogger("WTFX-Backend")
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # 启动时：初始化 DB 连接池、Redis 事件总线与外部适配器
+    logger.info(f"Starting {settings.PROJECT_NAME} v{settings.VERSION} [{settings.ENVIRONMENT}]")
-    print(f"Starting {settings.PROJECT_NAME} in [{settings.ENVIRONMENT}] mode...")
     yield
-    # 关闭时：优雅断开连接
+    logger.info("Shutting down WTFX Backend service...")
-    print(f"Shutting down {settings.PROJECT_NAME}...")
 
 app = FastAPI(
     title=settings.PROJECT_NAME,
     version=settings.VERSION,
     lifespan=lifespan,
-    docs_url="/docs" if settings.ENVIRONMENT != "production" else None,
+    docs_url="/docs" if settings.DEBUG or settings.ENVIRONMENT != "production" else None,
-    redoc_url="/redoc" if settings.ENVIRONMENT != "production" else None
 )
 
 # CORS 配置
@@ -28,12 +31,23 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-@app.get("/health", tags=["System"])
+# 注册 API 路由
+app.include_router(api_router, prefix="/api")
+
+# WebSocket 实时订阅端点
+@app.websocket("/ws/market/{market_address}")
+async def websocket_market_endpoint(websocket: WebSocket, market_address: str):
+    channel = f"market:{market_address.lower()}"
+    await ws_manager.connect(websocket, channel)
+    try:
+        while True:
+            data = await websocket.receive_text()
+            # 客户端心跳 ping/pong
+            if data == "ping":
+                await websocket.send_text("pong")
+    except WebSocketDisconnect:
+        ws_manager.disconnect(websocket, channel)
+
+@app.get("/health")
 async def health_check():
-    """生产探活与健康检查端点 (Blue-Green Probe)"""
+    return {"status": "ok", "version": settings.VERSION, "env": settings.ENVIRONMENT}
-    return {
-        "status": "ok",
-        "service": "wtfx-backend",
-        "version": settings.VERSION,
-        "environment": settings.ENVIRONMENT
-    }

app/models/entities.py

@@ -0,0 +1,120 @@
+import enum
+from datetime import datetime
+from decimal import Decimal
+from sqlalchemy import (
+    Column,
+    String,
+    BigInteger,
+    Integer,
+    Numeric,
+    DateTime,
+    Enum,
+    Index,
+    JSON,
+    ForeignKey,
+    Boolean,
+    Text,
+)
+from sqlalchemy.orm import declarative_base
+
+Base = declarative_base()
+
+class LedgerEntryType(str, enum.Enum):
+    """复式记账操作类型"""
+    DEPOSIT = "DEPOSIT"                         # 链上充值
+    WITHDRAW = "WITHDRAW"                       # 提现
+    TRADE_BUY = "TRADE_BUY"                     # 现货/曲线买入扣保证金
+    TRADE_SELL = "TRADE_SELL"                   # 现货/曲线卖出增保证金
+    FEE_PROTOCOL = "FEE_PROTOCOL"               # 协议手续费扣除
+    FEE_CREATOR_REBATE = "FEE_CREATOR_REBATE"   # 建盘者返佣
+    ORDER_FREEZE = "ORDER_FREEZE"               # 挂单冻结
+    ORDER_UNFREEZE = "ORDER_UNFREEZE"           # 撤单解冻
+    SETTLEMENT_PAYOUT = "SETTLEMENT_PAYOUT"     # 最终获胜兑付
+
+class MarketStatus(str, enum.Enum):
+    ACTIVE = "ACTIVE"
+    RESOLVED = "RESOLVED"
+    FINALISED = "FINALISED"
+    PAUSED = "PAUSED"
+
+class UserModel(Base):
+    """用户基础表"""
+    __tablename__ = "users"
+
+    address = Column(String(42), primary_key=True, index=True) # Checksum EVM Address
+    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
+    last_login_at = Column(DateTime, default=datetime.utcnow, nullable=False)
+    is_active = Column(Boolean, default=True, nullable=False)
+
+class AccountBalanceModel(Base):
+    """账户资金总览表 (由 Ledger 聚合维护)"""
+    __tablename__ = "account_balances"
+
+    address = Column(String(42), primary_key=True, index=True)
+    collateral_symbol = Column(String(16), primary_key=True, default="WUSD")
+    available_balance = Column(Numeric(precision=36, scale=18), default=Decimal("0"), nullable=False)
+    frozen_balance = Column(Numeric(precision=36, scale=18), default=Decimal("0"), nullable=False)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
+
+class LedgerEntryModel(Base):
+    """
+    不可变复式记账明细表 (Immutable Ledger)
+    资金流转与状态变更的唯一法定事实源
+    """
+    __tablename__ = "ledger_entries"
+
+    id = Column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
+    tx_hash = Column(String(66), nullable=True, index=True)
+    chain_id = Column(BigInteger, default=46630, nullable=False)
+    user_address = Column(String(42), nullable=False, index=True)
+    entry_type = Column(Enum(LedgerEntryType), nullable=False, index=True)
+    amount = Column(Numeric(precision=36, scale=18), nullable=False) # 变动金额 (+ / -)
+    balance_after = Column(Numeric(precision=36, scale=18), nullable=False) # 变动后可用余额
+    frozen_after = Column(Numeric(precision=36, scale=18), nullable=False) # 变动后冻结金额
+    related_market = Column(String(42), nullable=True, index=True)
+    related_order_id = Column(String(64), nullable=True, index=True)
+    extra_metadata = Column(JSON, default=dict)
+    created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True)
+
+    __table_args__ = (
+        Index("idx_ledger_user_created", "user_address", "created_at"),
+    )
+
+class MarketModel(Base):
+    """预测市场主表"""
+    __tablename__ = "markets"
+
+    market_address = Column(String(42), primary_key=True)
+    question_id = Column(String(66), unique=True, nullable=False, index=True)
+    title = Column(String(512), nullable=False)
+    description = Column(Text, nullable=True)
+    category = Column(String(64), default="Crypto", nullable=False, index=True)
+    tier = Column(BigInteger, default=1, nullable=False)
+    creator = Column(String(42), nullable=False, index=True)
+    curve_address = Column(String(42), nullable=False)
+    collateral_address = Column(String(42), nullable=False)
+    fee_rate = Column(Numeric(precision=36, scale=18), default=Decimal("0.006"), nullable=False)
+    resolution_time = Column(DateTime, nullable=False)
+    outcomes = Column(JSON, nullable=False) # e.g. ["YES", "NO"]
+    num_outcomes = Column(BigInteger, default=2, nullable=False)
+    status = Column(Enum(MarketStatus), default=MarketStatus.ACTIVE, nullable=False, index=True)
+    winning_outcome = Column(BigInteger, nullable=True)
+    created_at_block = Column(BigInteger, nullable=False)
+    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
+
+class OrderModel(Base):
+    """链下/链上委托订单明细"""
+    __tablename__ = "orders"
+
+    order_id = Column(String(64), primary_key=True)
+    market_address = Column(String(42), ForeignKey("markets.market_address"), nullable=False, index=True)
+    user_address = Column(String(42), nullable=False, index=True)
+    side = Column(String(8), nullable=False) # "BUY" | "SELL"
+    outcome_index = Column(BigInteger, nullable=False)
+    amount = Column(Numeric(precision=36, scale=18), nullable=False) # WUSD 或 Token 数量
+    price = Column(Numeric(precision=36, scale=18), nullable=True) # 限价 (若为市价则为空)
+    status = Column(String(16), default="FILLED", nullable=False) # PENDING, FILLED, CANCELLED
+    filled_amount = Column(Numeric(precision=36, scale=18), default=Decimal("0"), nullable=False)
+    fee = Column(Numeric(precision=36, scale=18), default=Decimal("0"), nullable=False)
+    tx_hash = Column(String(66), nullable=True)
+    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

app/schemas/dtos.py

@@ -0,0 +1,69 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Any
+from decimal import Decimal
+from datetime import datetime
+
+class UserAuthRequest(BaseModel):
+    address: str = Field(..., description="EVM 钱包地址")
+    signature: str = Field(..., description="签名")
+    message: str = Field(..., description="SIWE 原始签名消息")
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    address: str
+
+class AccountBalanceDTO(BaseModel):
+    address: str
+    collateral_symbol: str
+    available_balance: Decimal
+    frozen_balance: Decimal
+
+class LedgerEntryDTO(BaseModel):
+    id: int
+    entry_type: str
+    amount: Decimal
+    balance_after: Decimal
+    frozen_after: Decimal
+    related_market: Optional[str] = None
+    created_at: datetime
+
+class CreateMarketRequest(BaseModel):
+    title: str
+    description: Optional[str] = None
+    category: str = "Crypto"
+    tier: int = 1
+    resolution_time: datetime
+    outcomes: List[str] = Field(..., min_length=2, max_length=8)
+
+class MarketDTO(BaseModel):
+    market_address: str
+    question_id: str
+    title: str
+    description: Optional[str] = None
+    category: str
+    tier: int
+    creator: str
+    status: str
+    outcomes: List[str]
+    winning_outcome: Optional[int] = None
+    resolution_time: datetime
+    created_at: datetime
+
+class CreateOrderRequest(BaseModel):
+    market_address: str
+    outcome_index: int = Field(..., ge=0, le=7)
+    side: str = Field(..., pattern="^(BUY|SELL)$")
+    amount: Decimal = Field(..., gt=0)
+    price: Optional[Decimal] = None
+
+class OrderDTO(BaseModel):
+    order_id: str
+    market_address: str
+    user_address: str
+    side: str
+    outcome_index: int
+    amount: Decimal
+    status: str
+    fee: Decimal
+    created_at: datetime

app/ws/manager.py

@@ -0,0 +1,38 @@
+import logging
+from typing import Dict, Set
+from fastapi import WebSocket, WebSocketDisconnect
+
+logger = logging.getLogger(__name__)
+
+class ConnectionManager:
+    """WebSocket 实时连接与订阅频道管理器"""
+
+    def __init__(self):
+        # 活跃连接池: market_address -> Set[WebSocket]
+        self.active_connections: Dict[str, Set[WebSocket]] = {}
+
+    async def connect(self, websocket: WebSocket, channel: str):
+        await websocket.accept()
+        if channel not in self.active_connections:
+            self.active_connections[channel] = set()
+        self.active_connections[channel].add(websocket)
+        logger.info(f"Client connected to channel {channel}. Total: {len(self.active_connections[channel])}")
+
+    def disconnect(self, websocket: WebSocket, channel: str):
+        if channel in self.active_connections:
+            self.active_connections[channel].discard(websocket)
+            if not self.active_connections[channel]:
+                del self.active_connections[channel]
+        logger.info(f"Client disconnected from channel {channel}")
+
+    async def broadcast_to_channel(self, channel: str, message: dict):
+        """向特定频道所有订阅者推送 JSON 消息"""
+        if channel in self.active_connections:
+            for connection in list(self.active_connections[channel]):
+                try:
+                    await connection.send_json(message)
+                except Exception as e:
+                    logger.error(f"Error broadcasting message: {e}")
+                    self.disconnect(connection, channel)
+
+ws_manager = ConnectionManager()

tests/test_ledger.py

@@ -0,0 +1,76 @@
+import pytest
+import pytest_asyncio
+from decimal import Decimal
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+from app.models.entities import Base, LedgerEntryType
+from app.domains.ledger.service import LedgerService, InsufficientBalanceError
+
+TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
+
+@pytest_asyncio.fixture
+async def test_session():
+    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    session_maker = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
+    async with session_maker() as session:
+        yield session
+
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+    await engine.dispose()
+
+@pytest.mark.asyncio
+async def test_ledger_deposit_flow(test_session: AsyncSession):
+    ledger = LedgerService(test_session)
+    user = "0x6cddF384792C77219fc25C454Cfe264757842830"
+
+    # 1. 初始余额为 0
+    bal = await ledger.get_or_create_balance(user)
+    assert bal.available_balance == Decimal("0")
+
+    # 2. 存入 1000 WUSD
+    entry = await ledger.record_deposit(user, Decimal("1000"), "0xtx1")
+    assert entry.entry_type == LedgerEntryType.DEPOSIT
+    assert entry.amount == Decimal("1000")
+    assert entry.balance_after == Decimal("1000")
+
+    bal = await ledger.get_or_create_balance(user)
+    assert bal.available_balance == Decimal("1000")
+
+@pytest.mark.asyncio
+async def test_ledger_trade_and_insufficient_balance(test_session: AsyncSession):
+    ledger = LedgerService(test_session)
+    user = "0x6cddF384792C77219fc25C454Cfe264757842830"
+
+    # 充值 500 WUSD
+    await ledger.record_deposit(user, Decimal("500"), "0xtx1")
+
+    # 尝试买入 600 WUSD (超额报错)
+    with pytest.raises(InsufficientBalanceError):
+        await ledger.record_trade(
+            address=user,
+            market_address="0xmarket1",
+            order_id="order_1",
+            side="BUY",
+            amount=Decimal("600"),
+            fee=Decimal("3.6")
+        )
+
+    # 成功买入 100 WUSD (扣除 100 + 0.6 手续费)
+    entry = await ledger.record_trade(
+        address=user,
+        market_address="0xmarket1",
+        order_id="order_2",
+        side="BUY",
+        amount=Decimal("100"),
+        fee=Decimal("0.6")
+    )
+    assert entry.entry_type == LedgerEntryType.TRADE_BUY
+    assert entry.amount == Decimal("-100.6")
+    assert entry.balance_after == Decimal("399.4")
+
+    # 验证不可变流水历史条数
+    history = await ledger.list_user_ledger(user)
+    assert len(history) == 2