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

ARCHITECTURE.md

@@ -0,0 +1,112 @@
+# WTFX Backend Architecture Rules (后端架构铁律)
+
+> **核心定位**：WTFX 采用 **Python + FastAPI + PostgreSQL + Redis** 的**模块化单体 (Modular Monolith)** 架构。
+> **适用对象**：所有开发者与 AI Coding Agent。在编写或修改任何后端代码之前，**必须严格遵守本规范**。
+
+---
+
+## 1. 架构核心铁律 (Non-Negotiable Rules)
+
+1. **【分层单向依赖】**：
+   - 依赖调用链严格为：`API Route -> Domain Service -> Repository -> Database/Infrastructure`。
+   - **严禁反向或跨层调用**（如 Route 直连 DB，或 Repo 调用 Service）。
+2. **【禁止 Route 直连数据库】**：
+   - API 路由层（`app/api/`）只允许做：入参校验 (Pydantic Schema)、鉴权依赖注入、调用 Domain Service、返回标准响应。
+   - **严禁在 Route 中执行 SQL 或 ORM 查询**。
+3. **【Domain 纯净性】**：
+   - 业务领域层（`app/domains/`）代表纯粹的业务大脑，**严禁导入 FastAPI 或任何 HTTP/WebSocket 相关的 Web 框架依赖**。
+4. **【资金与状态变更强制走 Ledger】**：
+   - 严禁直接执行 `user.balance += amount` 等无据操作。
+   - 所有账户资金变动（充值、下注扣款、保证金冻结/解冻、手续费、获胜兑付）**必须通过 LedgerService 生成不可变的复式记账流水 (`LedgerEntry`)**。
+5. **【外部第三方服务强制使用 Adapter 隔离】**：
+   - 链上 RPC、Hyperliquid、价格预言机等外部系统交互必须统一封装在 `app/infrastructure/` 适配器中。
+   - Domain 业务层只依赖抽象接口（如 `TradingProvider` / `ChainClient`），不感知底层具体 API 细节。
+6. **【事件总线驱动，解耦 WebSocket】**：
+   - 业务状态变更（如订单成交、市场毕业）只向 `EventBus` 发送领域事件。
+   - WebSocket 模块（`app/ws/`）独立订阅事件并广播给在线客户端，业务逻辑层严禁直接依赖 WS 连接对象。
+7. **【Model 与 Schema 严格分离】**：
+   - `models/` 映射数据库实体，`schemas/` 定义 API 输入输出结构体。
+   - **严禁将数据库 Model 直接作为 API Response 返回**，防止敏感字段（密码哈希、内部风控分等）意外泄露。
+
+---
+
+## 2. 目录职责分工
+
+```
+wtf-backend/
+├── app/
+│   ├── main.py                     # 应用入口与生命周期管理 (Lifespan)
+│   ├── config.py                   # Pydantic Settings 环境变量配置
+│   ├── logging.py                  # 结构化日志配置
+│   │
+│   ├── api/                        # HTTP 接口接入层 (只做请求解析与转发)
+│   │   ├── dependencies.py         # 共享依赖注入 (DB Session, 当前用户, Auth)
+│   │   └── routes/                 # 各业务模块的路由入口
+│   │       ├── auth.py             # Web3 SIWE 钱包登录与 JWT
+│   │       ├── markets.py          # 预测市场查询与建盘
+│   │       ├── orders.py           # 交易下单与撤单
+│   │       ├── positions.py        # 用户持仓查询
+│   │       └── wallet.py           # 充值、提现与流水
+│   │
+│   ├── domains/                    # 业务领域大脑 (纯业务逻辑，无 Web 依赖)
+│   │   ├── market/                 # 市场状态机与联合曲线规则
+│   │   ├── trading/                # 交易撮合与风控检查
+│   │   ├── ledger/                 # 【核心】不可变流水账本
+│   │   ├── wallet/                 # 钱包余额状态管理
+│   │   ├── settlement/             # 预测到期结算与兑付
+│   │   └── user/                   # 用户信息与邀请返佣
+│   │
+│   ├── infrastructure/             # 外部世界适配层 (数据库、缓存、链上、三方 API)
+│   │   ├── database/               # PostgreSQL 连接池与异步 Session 工厂
+│   │   ├── redis/                  # Redis 缓存与 Streams 事件总线
+│   │   ├── blockchain/             # EVM RPC 监听与合约调用客户端
+│   │   └── external/               # Hyperliquid / CLOB 等外部交易适配器
+│   │
+│   ├── models/                     # SQLAlchemy 2.0 异步数据库实体
+│   │   ├── market.py
+│   │   ├── order.py
+│   │   ├── position.py
+│   │   ├── ledger.py
+│   │   └── user.py
+│   │
+│   ├── schemas/                    # Pydantic v2 请求与响应数据传输对象 (DTO)
+│   │   ├── market.py
+│   │   ├── order.py
+│   │   ├── ledger.py
+│   │   └── common.py
+│   │
+│   └── ws/                         # WebSocket 实时网关 (行情推流与用户私有信道)
+│       ├── manager.py              # 连接池与订阅频道管理
+│       └── handlers.py             # WS 消息解析与心跳
+│
+├── migrations/                     # Alembic 数据库迁移脚本
+├── tests/                          # 单元测试与集成测试
+├── Dockerfile                      # 生产镜像构建文件
+├── docker-compose.yml              # 本地开发与服务器运行编排
+└── requirements.txt                # 锁定依赖项
+```
+
+---
+
+## 3. 核心业务流程标准示例 (How to Write Code)
+
+### 正确的下单流程示例 (Trading Flow):
+```
+[Client] ---> POST /api/v1/orders
+                 │
+                 ▼
+          [orders.py (Route)]
+                 │ (1. 解析 CreateOrderRequest Schema)
+                 │ (2. 依赖注入 TradingService)
+                 ▼
+       [TradingService (Domain)]
+                 │ (1. 风险检查 RiskCheck)
+                 │ (2. 冻结资金: 调用 LedgerService.freeze_balance)
+                 │ (3. 创建订单: 调用 OrderRepository.create)
+                 │ (4. 抛出事件: EventBus.publish("OrderCreated"))
+                 ▼
+          [PostgreSQL] (Commit Transaction)
+                 │
+                 ▼
+          [WS Gateway] (异步监听到 OrderCreated -> 广播行情)
+```

Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# 安装基础编译依赖
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    libpq-dev \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# 安装 Python 依赖
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 复制代码
+COPY . .
+
+EXPOSE 8080
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "4"]

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

@@ -0,0 +1,47 @@
+from pydantic_settings import BaseSettings
+from typing import List
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = "WTFX Backend API"
+    VERSION: str = "2.0.0"
+    DEBUG: bool = False
+    
+    # 环境：development, staging, production
+    ENVIRONMENT: str = "development"
+    
+    # 数据库配置 (PostgreSQL)
+    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/wtfx"
+    
+    # Redis 配置
+    REDIS_URL: str = "redis://localhost:6379/0"
+    
+    # JWT 鉴权
+    JWT_SECRET: str = "wtfx_secret_jwt_key_change_in_production"
+    JWT_ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 7 # 7 days
+    
+    # 跨域设置
+    CORS_ORIGINS: List[str] = [
+        "http://localhost:3000",
+        "http://localhost:3001",
+        "https://wtfx.app",
+        "https://test.wtfx.app",
+        "https://admin.wtfx.app",
+        "https://admin.test.wtfx.app"
+    ]
+    
+    # 链上 RPC 配置 (Robinhood Chain)
+    ROBINHOOD_TESTNET_RPC: str = "https://rpc.testnet.chain.robinhood.com"
+    ROBINHOOD_MAINNET_RPC: str = "https://rpc.chain.robinhood.com"
+    
+    # 控制器合约地址 (由部署脚本生成)
+    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

@@ -0,0 +1,53 @@
+import logging
+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):
+    logger.info(f"Starting {settings.PROJECT_NAME} v{settings.VERSION} [{settings.ENVIRONMENT}]")
+    yield
+    logger.info("Shutting down WTFX Backend service...")
+
+app = FastAPI(
+    title=settings.PROJECT_NAME,
+    version=settings.VERSION,
+    lifespan=lifespan,
+    docs_url="/docs" if settings.DEBUG or settings.ENVIRONMENT != "production" else None,
+)
+
+# CORS 配置
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.CORS_ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 注册 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():
+    return {"status": "ok", "version": settings.VERSION, "env": 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()

docker-compose.yml

@@ -0,0 +1,42 @@
+version: '3.8'
+
+services:
+  backend:
+    build: .
+    container_name: wtfx-backend
+    restart: unless-stopped
+    ports:
+      - "8080:8080"
+    environment:
+      - ENVIRONMENT=production
+      - DATABASE_URL=postgresql+asyncpg://wtfx_user:wtfx_pass@postgres:5432/wtfx_db
+      - REDIS_URL=redis://redis:6379/0
+    depends_on:
+      - postgres
+      - redis
+
+  postgres:
+    image: postgres:16-alpine
+    container_name: wtfx-postgres
+    restart: unless-stopped
+    environment:
+      POSTGRES_USER: wtfx_user
+      POSTGRES_PASSWORD: wtfx_pass
+      POSTGRES_DB: wtfx_db
+    volumes:
+      - pgdata:/var/lib/postgresql/data
+    ports:
+      - "5432:5432"
+
+  redis:
+    image: redis:7-alpine
+    container_name: wtfx-redis
+    restart: unless-stopped
+    volumes:
+      - redisdata:/data
+    ports:
+      - "6379:6379"
+
+volumes:
+  pgdata:
+  redisdata:

requirements.txt

@@ -0,0 +1,15 @@
+fastapi>=0.111.0
+uvicorn[standard]>=0.30.0
+pydantic>=2.7.0
+pydantic-settings>=2.3.0
+sqlalchemy[asyncio]>=2.0.30
+asyncpg>=0.29.0
+alembic>=1.13.1
+redis>=5.0.4
+web3>=6.19.0
+eth-account>=0.11.2
+pyjwt>=2.8.0
+python-multipart>=0.0.9
+httpx>=0.27.0
+pytest>=8.2.0
+pytest-asyncio>=0.23.7

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