124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
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())
|