diff --git a/app/__pycache__/__init__.cpython-310.pyc b/app/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..0fd2f3d Binary files /dev/null and b/app/__pycache__/__init__.cpython-310.pyc differ diff --git a/app/api/dependencies.py b/app/api/dependencies.py new file mode 100644 index 0000000..13040ef --- /dev/null +++ b/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 已过期或无效") diff --git a/app/api/routes/api.py b/app/api/routes/api.py new file mode 100644 index 0000000..fa938d1 --- /dev/null +++ b/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 + ] diff --git a/app/config.py b/app/config.py index 0cb44c3..fe95726 100644 --- a/app/config.py +++ b/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() diff --git a/app/domains/ledger/__pycache__/__init__.cpython-310.pyc b/app/domains/ledger/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..140a8fc Binary files /dev/null and b/app/domains/ledger/__pycache__/__init__.cpython-310.pyc differ diff --git a/app/domains/ledger/__pycache__/service.cpython-310.pyc b/app/domains/ledger/__pycache__/service.cpython-310.pyc new file mode 100644 index 0000000..101f849 Binary files /dev/null and b/app/domains/ledger/__pycache__/service.cpython-310.pyc differ diff --git a/app/domains/ledger/service.py b/app/domains/ledger/service.py new file mode 100644 index 0000000..d9b96ba --- /dev/null +++ b/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()) diff --git a/app/domains/trading/service.py b/app/domains/trading/service.py new file mode 100644 index 0000000..b635f92 --- /dev/null +++ b/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 diff --git a/app/main.py b/app/main.py index 9a0bced..29491a0 100644 --- a/app/main.py +++ b/app/main.py @@ -1,22 +1,25 @@ -from fastapi import FastAPI -from fastapi.middleware.cors import CORSMiddleware +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): - # 启动时：初始化 DB 连接池、Redis 事件总线与外部适配器 - print(f"Starting {settings.PROJECT_NAME} in [{settings.ENVIRONMENT}] mode...") + logger.info(f"Starting {settings.PROJECT_NAME} v{settings.VERSION} [{settings.ENVIRONMENT}]") yield - # 关闭时：优雅断开连接 - print(f"Shutting down {settings.PROJECT_NAME}...") + logger.info("Shutting down WTFX Backend service...") app = FastAPI( title=settings.PROJECT_NAME, version=settings.VERSION, lifespan=lifespan, - docs_url="/docs" if settings.ENVIRONMENT != "production" else None, - redoc_url="/redoc" if settings.ENVIRONMENT != "production" else None + docs_url="/docs" if settings.DEBUG or 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", - "service": "wtfx-backend", - "version": settings.VERSION, - "environment": settings.ENVIRONMENT - } + return {"status": "ok", "version": settings.VERSION, "env": settings.ENVIRONMENT} diff --git a/app/models/__pycache__/__init__.cpython-310.pyc b/app/models/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..37cf7a8 Binary files /dev/null and b/app/models/__pycache__/__init__.cpython-310.pyc differ diff --git a/app/models/__pycache__/entities.cpython-310.pyc b/app/models/__pycache__/entities.cpython-310.pyc new file mode 100644 index 0000000..e8bd519 Binary files /dev/null and b/app/models/__pycache__/entities.cpython-310.pyc differ diff --git a/app/models/entities.py b/app/models/entities.py new file mode 100644 index 0000000..be8f059 --- /dev/null +++ b/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) diff --git a/app/schemas/dtos.py b/app/schemas/dtos.py new file mode 100644 index 0000000..f1652d2 --- /dev/null +++ b/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 diff --git a/app/ws/manager.py b/app/ws/manager.py new file mode 100644 index 0000000..d5a972b --- /dev/null +++ b/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() diff --git a/tests/__pycache__/test_ledger.cpython-310-pytest-8.4.1.pyc b/tests/__pycache__/test_ledger.cpython-310-pytest-8.4.1.pyc new file mode 100644 index 0000000..91b9732 Binary files /dev/null and b/tests/__pycache__/test_ledger.cpython-310-pytest-8.4.1.pyc differ diff --git a/tests/test_ledger.py b/tests/test_ledger.py new file mode 100644 index 0000000..037756e --- /dev/null +++ b/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