name: DeFi交易系统 description: 设计用于永续期货、流动性提供和自动化策略的DeFi交易系统，附带风险管理和MEV保护。 license: MIT

DeFi交易系统

这个技能提供构建去中心化金融交易系统的指导，重点在于永续期货、自动化做市商和风险管理。

核心能力

永续期货: 资金费率、杠杆、清算机制
自动化做市商: 流动性提供、非永久性损失
风险管理: 头寸规模、止损、投资组合对冲
MEV保护: 三明治攻击、抢先交易缓解

DeFi交易基础

永续期货机制

传统期货:                 永续期货:
┌─────────────────┐              ┌─────────────────┐
│   到期日期     │              │   无到期       │
│   结算        │              │   资金费率     │
│   滚动成本     │              │   连续        │
└─────────────────┘              └─────────────────┘

资金费率 = (标记价格 - 指数价格) / 指数价格 × 间隔

如果资金费率 > 0: 多头支付空头
如果资金费率 < 0: 空头支付多头

关键概念

术语	定义
标记价格	用于清算的公允价值
指数价格	来自交易所的现货价格
资金费率	多头和空头之间的定期支付
维持保证金	避免清算的最低权益
清算价格	头寸被强制关闭的价格

头寸管理

头寸规模

from dataclasses import dataclass
from decimal import Decimal

@dataclass
class Position:
    symbol: str
    side: str  # 'long' or 'short'
    size: Decimal
    entry_price: Decimal
    leverage: int
    margin: Decimal

    @property
    def notional_value(self) -> Decimal:
        return self.size * self.entry_price

    @property
    def liquidation_price(self) -> Decimal:
        """计算清算价格"""
        maintenance_margin_rate = Decimal('0.005')  # 0.5%

        if self.side == 'long':
            # 清算价格 = 入场价 × (1 - 初始保证金 + 维持保证金)
            return self.entry_price * (
                1 - (1 / self.leverage) + maintenance_margin_rate
            )
        else:
            return self.entry_price * (
                1 + (1 / self.leverage) - maintenance_margin_rate
            )

    def unrealized_pnl(self, current_price: Decimal) -> Decimal:
        """计算未实现盈亏"""
        if self.side == 'long':
            return self.size * (current_price - self.entry_price)
        else:
            return self.size * (self.entry_price - current_price)

    def roi_percent(self, current_price: Decimal) -> Decimal:
        """投资回报率百分比"""
        pnl = self.unrealized_pnl(current_price)
        return (pnl / self.margin) * 100


class PositionSizer:
    """基于风险参数计算头寸规模"""

    def __init__(self, account_balance: Decimal):
        self.balance = account_balance
        self.max_risk_per_trade = Decimal('0.02')  # 2% 的账户
        self.max_leverage = 10

    def calculate_size(
        self,
        entry_price: Decimal,
        stop_loss_price: Decimal,
        leverage: int
    ) -> dict:
        """为给定风险参数计算头寸规模"""
        # 限制杠杆
        leverage = min(leverage, self.max_leverage)

        # 风险金额
        risk_amount = self.balance * self.max_risk_per_trade

        # 价格到止损的距离
        price_distance = abs(entry_price - stop_loss_price)
        price_distance_pct = price_distance / entry_price

        # 基于风险的头寸规模
        # 规模 × 价格距离 = 风险金额
        size = risk_amount / price_distance

        # 检查保证金要求
        required_margin = (size * entry_price) / leverage

        if required_margin > self.balance:
            # 减小规模以适合可用保证金
            size = (self.balance * leverage) / entry_price

        return {
            'size': size,
            'leverage': leverage,
            'margin_required': (size * entry_price) / leverage,
            'risk_amount': size * price_distance,
            'risk_percent': (size * price_distance / self.balance) * 100
        }

清算预防

class LiquidationMonitor:
    """监控头寸的清算风险"""

    def __init__(self, warning_threshold: float = 0.8):
        self.warning_threshold = warning_threshold
        self.positions: dict[str, Position] = {}

    def check_health(self, position: Position, current_price: Decimal) -> dict:
        """计算头寸健康指标"""
        liq_price = position.liquidation_price

        if position.side == 'long':
            distance_to_liq = (current_price - liq_price) / current_price
        else:
            distance_to_liq = (liq_price - current_price) / current_price

        # 健康比率: 1.0 = 最佳健康, 0.0 = 清算
        health_ratio = distance_to_liq / (1 / position.leverage)

        return {
            'liquidation_price': liq_price,
            'distance_percent': float(distance_to_liq) * 100,
            'health_ratio': float(health_ratio),
            'at_risk': health_ratio < self.warning_threshold,
            'margin_ratio': self._calculate_margin_ratio(position, current_price)
        }

    def _calculate_margin_ratio(
        self,
        position: Position,
        current_price: Decimal
    ) -> float:
        """计算当前保证金比率"""
        pnl = position.unrealized_pnl(current_price)
        equity = position.margin + pnl
        return float(equity / position.notional_value)

资金费率策略

资金费率套利

from typing import List
import asyncio

@dataclass
class FundingRateInfo:
    exchange: str
    symbol: str
    funding_rate: Decimal
    next_funding_time: int
    mark_price: Decimal
    index_price: Decimal

class FundingArbitrage:
    """捕获资金费率差异"""

    def __init__(self, exchanges: List[str]):
        self.exchanges = exchanges
        self.min_rate_threshold = Decimal('0.0001')  # 0.01%

    async def find_opportunities(self, symbol: str) -> List[dict]:
        """寻找资金费率套利机会"""
        rates = await self._fetch_all_rates(symbol)

        opportunities = []

        # 按资金费率排序
        rates.sort(key=lambda x: x.funding_rate)

        # 找到极值
        lowest = rates[0]
        highest = rates[-1]

        spread = highest.funding_rate - lowest.funding_rate

        if spread > self.min_rate_threshold:
            opportunities.append({
                'type': '跨交易所',
                '多头交易所': lowest.exchange,
                '空头交易所': highest.exchange,
                '预期回报': spread,
                '年化回报': spread * 3 * 365  # 8小时资金费率
            })

        # 现货-永续基差交易
        for rate in rates:
            if abs(rate.funding_rate) > self.min_rate_threshold:
                opportunities.append({
                    'type': '基差交易',
                    'exchange': rate.exchange,
                    'direction': '空头永续多头现货' if rate.funding_rate > 0
                                 else '多头永续空头现货',
                    'expected_return': abs(rate.funding_rate),
                    'mark_index_spread': rate.mark_price - rate.index_price
                })

        return opportunities

    def calculate_basis_trade_pnl(
        self,
        funding_received: Decimal,
        entry_costs: Decimal,
        price_change_pnl: Decimal
    ) -> dict:
        """计算基差交易的盈亏"""
        gross_pnl = funding_received + price_change_pnl
        net_pnl = gross_pnl - entry_costs

        return {
            '资金盈亏': funding_received,
            '价格盈亏': price_change_pnl,  # 如果对冲应接近零
            '成本': entry_costs,
            '净盈亏': net_pnl
        }

AMM与流动性提供

非永久性损失计算器

import math

def calculate_impermanent_loss(price_ratio: float) -> float:
    """
    计算恒定乘积AMM的非永久性损失

    price_ratio: 新价格 / 原始价格

    返回: 非永久性损失为负百分比（例如，-0.05表示5%损失）
    """
    # IL = 2 * sqrt(price_ratio) / (1 + price_ratio) - 1
    return 2 * math.sqrt(price_ratio) / (1 + price_ratio) - 1


def lp_value_vs_holding(
    initial_token_a: float,
    initial_token_b: float,
    initial_price: float,
    final_price: float
) -> dict:
    """比较LP头寸价值与仅持有"""
    # 初始价值
    initial_value = initial_token_a * initial_price + initial_token_b

    # 如果仅持有
    hold_value = initial_token_a * final_price + initial_token_b

    # LP头寸（恒定乘积: x * y = k）
    k = initial_token_a * initial_token_b
    # 在新价格下: token_a_new * price = token_b_new（等值）
    # token_a_new * token_b_new = k
    token_a_new = math.sqrt(k / final_price)
    token_b_new = math.sqrt(k * final_price)

    lp_value = token_a_new * final_price + token_b_new

    return {
        'initial_value': initial_value,
        'hold_value': hold_value,
        'lp_value': lp_value,
        'impermanent_loss': lp_value - hold_value,
        'il_percent': (lp_value - hold_value) / hold_value * 100,
        'new_token_a': token_a_new,
        'new_token_b': token_b_new
    }

集中流动性（Uniswap V3风格）

@dataclass
class LPPosition:
    """集中流动性头寸"""
    lower_tick: int
    upper_tick: int
    liquidity: Decimal
    token_a_deposited: Decimal
    token_b_deposited: Decimal

    @property
    def price_range(self) -> tuple[float, float]:
        """将价格刻度转换为价格"""
        return (
            1.0001 ** self.lower_tick,
            1.0001 ** self.upper_tick
        )

    def in_range(self, current_price: float) -> bool:
        """检查当前价格是否在头寸范围内"""
        lower, upper = self.price_range
        return lower <= current_price <= upper


class ConcentratedLPCalculator:
    """计算集中流动性头寸的指标"""

    def calculate_liquidity(
        self,
        amount_a: Decimal,
        amount_b: Decimal,
        price_lower: float,
        price_upper: float,
        current_price: float
    ) -> Decimal:
        """计算给定存款的流动性价值"""
        sqrt_price = math.sqrt(current_price)
        sqrt_lower = math.sqrt(price_lower)
        sqrt_upper = math.sqrt(price_upper)

        if current_price <= price_lower:
            # 全部在代币A
            liquidity = float(amount_a) * (sqrt_lower * sqrt_upper) / (sqrt_upper - sqrt_lower)
        elif current_price >= price_upper:
            # 全部在代币B
            liquidity = float(amount_b) / (sqrt_upper - sqrt_lower)
        else:
            # 混合
            liquidity_a = float(amount_a) * (sqrt_price * sqrt_upper) / (sqrt_upper - sqrt_price)
            liquidity_b = float(amount_b) / (sqrt_price - sqrt_lower)
            liquidity = min(liquidity_a, liquidity_b)

        return Decimal(str(liquidity))

    def calculate_position_value(
        self,
        position: LPPosition,
        current_price: float
    ) -> dict:
        """计算LP头寸的当前价值"""
        sqrt_price = math.sqrt(current_price)
        lower, upper = position.price_range
        sqrt_lower = math.sqrt(lower)
        sqrt_upper = math.sqrt(upper)

        liquidity = float(position.liquidity)

        if current_price <= lower:
            amount_a = liquidity * (sqrt_upper - sqrt_lower) / (sqrt_lower * sqrt_upper)
            amount_b = 0
        elif current_price >= upper:
            amount_a = 0
            amount_b = liquidity * (sqrt_upper - sqrt_lower)
        else:
            amount_a = liquidity * (sqrt_upper - sqrt_price) / (sqrt_price * sqrt_upper)
            amount_b = liquidity * (sqrt_price - sqrt_lower)

        return {
            'token_a': Decimal(str(amount_a)),
            'token_b': Decimal(str(amount_b)),
            'total_value_in_b': Decimal(str(amount_a * current_price + amount_b)),
            'in_range': position.in_range(current_price)
        }

MEV保护

理解MEV攻击

三明治攻击:

用户想交换:     攻击者抢先交易:   用户的交换:        攻击者后执行:
代币A → 代币B       购买代币B           对用户更差的价格    出售代币B
                        （抬高价格）                            （获利）

时间线:
─────────────────────────────────────────────────────────────────────▶
         攻击者交易1        用户交易            攻击者交易2
         （抢先）           （受害者）            （后执行）

保护策略

from web3 import Web3

class MEVProtection:
    """最小化MEV风险的策略"""

    def __init__(self, web3: Web3):
        self.w3 = web3

    def calculate_max_slippage(
        self,
        expected_output: Decimal,
        tolerance_percent: float = 0.5
    ) -> Decimal:
        """计算带滑点保护的最小输出"""
        return expected_output * (1 - Decimal(str(tolerance_percent / 100)))

    def should_use_private_mempool(
        self,
        trade_size_usd: float,
        gas_price_gwei: float
    ) -> bool:
        """判断交易是否应使用私有内存池"""
        # 更大交易对MEV机器人更有吸引力
        # 更高燃气价格表示竞争性MEV环境

        mev_risk_score = (trade_size_usd / 10000) * (gas_price_gwei / 50)
        return mev_risk_score > 1.0

    def chunk_large_order(
        self,
        total_size: Decimal,
        max_chunk_size: Decimal,
        min_time_between_chunks: int = 30  # 秒
    ) -> List[dict]:
        """将大订单拆分为小块"""
        chunks = []
        remaining = total_size

        while remaining > 0:
            chunk_size = min(remaining, max_chunk_size)
            chunks.append({
                'size': chunk_size,
                'delay': len(chunks) * min_time_between_chunks
            })
            remaining -= chunk_size

        return chunks

    def calculate_twap_schedule(
        self,
        total_size: Decimal,
        duration_minutes: int,
        num_orders: int
    ) -> List[dict]:
        """创建TWAP（时间加权平均价格）计划"""
        interval = duration_minutes / num_orders
        order_size = total_size / num_orders

        return [
            {
                'size': order_size,
                'execute_at_minute': i * interval
            }
            for i in range(num_orders)
        ]

Flashbots集成

from eth_account import Account
import requests

class FlashbotsSubmitter:
    """通过Flashbots提交交易以避免公共内存池"""

    def __init__(self, flashbots_url: str, signing_key: str):
        self.url = flashbots_url
        self.signer = Account.from_key(signing_key)

    def submit_bundle(
        self,
        signed_transactions: List[str],
        target_block: int
    ) -> dict:
        """向Flashbots提交交易包"""
        bundle = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "eth_sendBundle",
            "params": [{
                "txs": signed_transactions,
                "blockNumber": hex(target_block)
            }]
        }

        # 签署包
        message = Web3.keccak(text=str(bundle))
        signature = self.signer.sign_message(message)

        headers = {
            "X-Flashbots-Signature": f"{self.signer.address}:{signature.signature.hex()}"
        }

        response = requests.post(self.url, json=bundle, headers=headers)
        return response.json()

风险管理

投资组合风险指标

import numpy as np

class RiskMetrics:
    """计算投资组合风险指标"""

    def calculate_var(
        self,
        returns: List[float],
        confidence_level: float = 0.95
    ) -> float:
        """风险价值 - 在置信水平下的最大预期损失"""
        return np.percentile(returns, (1 - confidence_level) * 100)

    def calculate_cvar(
        self,
        returns: List[float],
        confidence_level: float = 0.95
    ) -> float:
        """条件风险价值 - 超过风险价值的预期损失"""
        var = self.calculate_var(returns, confidence_level)
        return np.mean([r for r in returns if r <= var])

    def calculate_sharpe_ratio(
        self,
        returns: List[float],
        risk_free_rate: float = 0.0
    ) -> float:
        """风险调整回报指标"""
        excess_returns = np.array(returns) - risk_free_rate
        return np.mean(excess_returns) / np.std(excess_returns) if np.std(excess_returns) > 0 else 0

    def calculate_max_drawdown(self, equity_curve: List[float]) -> dict:
        """最大峰谷下跌"""
        peak = equity_curve[0]
        max_dd = 0
        max_dd_start = 0
        max_dd_end = 0

        current_dd_start = 0

        for i, value in enumerate(equity_curve):
            if value > peak:
                peak = value
                current_dd_start = i

            dd = (peak - value) / peak
            if dd > max_dd:
                max_dd = dd
                max_dd_start = current_dd_start
                max_dd_end = i

        return {
            'max_drawdown': max_dd,
            'start_index': max_dd_start,
            'end_index': max_dd_end
        }


class PortfolioRiskManager:
    """管理整体投资组合风险"""

    def __init__(self, max_portfolio_risk: float = 0.1):
        self.max_risk = max_portfolio_risk  # 10% 最大投资组合风险
        self.positions: List[Position] = []

    def can_add_position(
        self,
        new_position: Position,
        current_price: Decimal
    ) -> tuple[bool, str]:
        """检查新头寸是否适合风险限制"""
        # 计算当前投资组合风险
        current_risk = self._calculate_portfolio_risk(current_price)

        # 计算新头寸的风险
        new_position_risk = self._calculate_position_risk(new_position)

        # 检查相关性（简化）
        total_risk = current_risk + new_position_risk  # 假设相关性

        if total_risk > self.max_risk:
            return False, f"会超出最大风险: {total_risk:.2%} > {self.max_risk:.2%}"

        return True, "头寸在风险限制内"

    def get_risk_summary(self, current_prices: dict) -> dict:
        """获取投资组合风险摘要"""
        return {
            'total_exposure': sum(p.notional_value for p in self.positions),
            'net_exposure': self._calculate_net_exposure(),
            'var_95': self._calculate_portfolio_var(0.95),
            'positions_at_risk': [
                p.symbol for p in self.positions
                if self._is_position_at_risk(p, current_prices.get(p.symbol))
            ]
        }

最佳实践

交易系统检查清单

头寸管理
- [ ] 最大头寸规模限制
- [ ] 按资产波动性的杠杆限制
- [ ] 自动止损订单
- [ ] 清算价格监控
风险管理
- [ ] 投资组合级别风险限制
- [ ] 相关性监控
- [ ] 最大回撤限制与自动减杠杆
- [ ] 定期风险价值计算
执行
- [ ] MEV保护（私有内存池、分块）
- [ ] 所有交易上的滑点限制
- [ ] 燃气价格限制
- [ ] 失败交易处理
操作
- [ ] 热钱包限制
- [ ] 大额提款的多重签名
- [ ] 定期对账
- [ ] 事件响应计划

参考

references/perpetual-mechanics.md - 永续机制的深入探讨
references/amm-formulas.md - AMM恒定乘积和集中流动性公式
references/mev-protection.md - 详细的MEV缓解策略