名称: defi-trading-systems 描述: 设计用于永续合约、流动性提供和自动化策略的DeFi交易系统，包括风险管理和MEV保护。 许可证: MIT
DeFi交易系统
本技能提供构建去中心化金融交易系统的指导，专注于永续合约、自动做市商和风险管理。
核心能力
- 永续合约: 资金费率、杠杆、清算机制
- 自动做市商: 流动性提供、非永久性损失
- 风险管理: 头寸规模、止损、投资组合对冲
- MEV保护: 三明治攻击、抢先交易缓解
DeFi交易基础
永续合约机制
传统期货: 永续合约:
┌─────────────────┐ ┌─────────────────┐
│ 到期日 │ │ 无到期日 │
│ 结算 │ │ 资金费率 │
│ 滚动成本 │ │ 连续 │
└─────────────────┘ └─────────────────┘
资金费率 = (标记价格 - 指数价格) / 指数价格 × 间隔
如果资金费率 > 0: 多头支付空头
如果资金费率 < 0: 空头支付多头
关键概念
| 术语 | 定义 |
|---|---|
| 标记价格 | 用于清算的公允价值 |
| 指数价格 | 交易所的现货价格 |
| 资金费率 | 多头和空头之间的周期性支付 |
| 维持保证金 | 避免清算的最低权益 |
| 清算价格 | 头寸被强制平仓的价格 |
头寸管理
头寸规模
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class Position:
symbol: str
side: str # 'long' 或 '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': 'cross_exchange',
'long_exchange': lowest.exchange,
'short_exchange': highest.exchange,
'expected_return': spread,
'annualized_return': spread * 3 * 365 # 8小时资金费率
})
# 现货-永续基差交易
for rate in rates:
if abs(rate.funding_rate) > self.min_rate_threshold:
opportunities.append({
'type': 'basis_trade',
'exchange': rate.exchange,
'direction': 'short_perp_long_spot' if rate.funding_rate > 0
else 'long_perp_short_spot',
'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_pnl': funding_received,
'price_pnl': price_change_pnl, # 如果对冲，应接近0
'costs': entry_costs,
'net_pnl': net_pnl
}
AMM和流动性提供
非永久性损失计算器
import math
def calculate_impermanent_loss(price_ratio: float) -> float:
"""
计算恒定乘积AMM的非永久性损失
price_ratio: 新价格 / 原价格
返回: IL作为负百分比（例如，-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]:
"""将tick转换为价格"""
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:
# 全部在token A中
liquidity = float(amount_a) * (sqrt_lower * sqrt_upper) / (sqrt_upper - sqrt_lower)
elif current_price >= price_upper:
# 全部在token 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攻击
三明治攻击:
用户想交换: 攻击者抢先交易: 用户的交换: 攻击者后置交易:
Token A → Token B 买入Token B 用户价格较差 卖出Token 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机器人更具吸引力
# 更高gas价格表示竞争激烈的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的预期损失"""
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))
]
}
最佳实践
交易系统检查清单
-
头寸管理
- [ ] 最大头寸规模限制
- [ ] 按资产波动率的杠杆限制
- [ ] 自动止损订单
- [ ] 清算价格监控
-
风险管理
- [ ] 投资组合级别风险限制
- [ ] 相关性监控
- [ ] 回撤限制与自动降杠杆
- [ ] 定期VaR计算
-
执行
- [ ] MEV保护（私有内存池、分块）
- [ ] 所有交易的滑点限制
- [ ] gas价格限制
- [ ] 失败交易处理
-
运营
- [ ] 热钱包限制
- [ ] 大额提款的多重签名
- [ ] 定期对账
- [ ] 事件响应计划
参考
references/perpetual-mechanics.md- 永续合约数学深入探讨references/amm-formulas.md- AMM恒定乘积和集中流动性references/mev-protection.md- 详细MEV缓解策略