DeFi

引言 AI Agent与Web3的融合代表了两个最前沿技术的交汇点。当AI Agent能够自主地与区块链交互，我们将迎来全新的应用范式。本文将探讨如何构建能够理解、操作和优化链上系统的智能体。 链上AI推理 去中心化AI推理网络 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 from typing import List, Dict import hashlib import json class OnChainInferenceNetwork: """链上AI推理网络""" def __init__(self, blockchain_rpc: str): self.web3 = Web3(Web3.HTTPProvider(blockchain_rpc)) self.private_key = os.getenv("PRIVATE_KEY") self.account = self.web3.eth.account.from_key(self.private_key) # 加载推理合约ABI self.inference_contract = self.web3.eth.contract( address="0x...", # AI推理合约地址 abi=[...] # 合约ABI ) def submit_inference_task( self, model_id: str, input_data: Dict, reward: int ) -> str: """提交推理任务""" # 准备任务数据 task_data = { "modelId": model_id, "input": input_data, "reward": reward, "timeout": 3600, # 1小时超时 "timestamp": int(time.time()) } # 计算任务哈希 task_hash = self._compute_task_hash(task_data) # 提交到链上 tx_hash = self.inference_contract.functions.submitTask( task_hash, json.dumps(input_data), reward, task_data["timeout"] ).transact({'from': self.account.address}) self.web3.eth.wait_for_transaction_receipt(tx_hash) return task_hash def submit_inference_result( self, task_hash: str, output_data: Dict ) -> str: """提交推理结果""" # 准备结果数据 result_hash = self._compute_result_hash({ "taskHash": task_hash, "output": output_data, "submitter": self.account.address }) # 提交结果 tx_hash = self.inference_contract.functions.submitResult( task_hash, json.dumps(output_data), result_hash ).transact({'from': self.account.address}) self.web3.eth.wait_for_transaction_receipt(tx_hash) return tx_hash def claim_reward(self, task_hash: str) -> str: """领取奖励""" # 检查任务是否完成 task = self.inference_contract.functions.tasks(task_hash).call() if not task["completed"]: raise Exception("Task not completed yet") # 领取奖励 tx_hash = self.inference_contract.functions.claimReward( task_hash ).transact({'from': self.account.address}) self.web3.eth.wait_for_transaction_receipt(tx_hash) return tx_hash def verify_result( self, task_hash: str, output_data: Dict ) -> bool: """验证结果""" # 从链上获取任务 task = self.inference_contract.functions.tasks(task_hash).call() # 计算期望的输出哈希 expected_hash = self._compute_output_hash( task["input"], task["modelId"] ) # 验证结果哈希 result_hash = self._compute_result_hash({ "taskHash": task_hash, "output": output_data }) return result_hash == expected_hash def _compute_task_hash(self, task_data: Dict) -> str: """计算任务哈希""" data_string = json.dumps(task_data, sort_keys=True) return hashlib.sha256(data_string.encode()).hexdigest() def _compute_result_hash(self, result_data: Dict) -> str: """计算结果哈希""" data_string = json.dumps(result_data, sort_keys=True) return hashlib.sha256(data_string.encode()).hexdigest() def _compute_output_hash(self, input_data: Dict, model_id: str) -> str: """计算输出哈希（模拟AI推理）""" # 实际应用中，这里应该运行AI模型 # 这里简化处理 output = self._run_model(input_data, model_id) return hashlib.sha256(json.dumps(output).encode()).hexdigest() def _run_model(self, input_data: Dict, model_id: str) -> Dict: """运行AI模型""" # 实际应用中，这里应该调用真实的AI模型 # 可以使用OpenAI API、本地模型等 if model_id == "text-classifier": return self._classify_text(input_data["text"]) elif model_id == "image-analyzer": return self._analyze_image(input_data["imageUrl"]) elif model_id == "sentiment-analyzer": return self._analyze_sentiment(input_data["text"]) else: raise Exception(f"Unknown model: {model_id}") def _classify_text(self, text: str) -> Dict: """文本分类""" # 简化实现，实际应该调用真实模型 categories = { "technology": 0.8, "finance": 0.6, "sports": 0.1 } predicted_category = max(categories, key=categories.get) return { "category": predicted_category, "confidence": categories[predicted_category] } def _analyze_image(self, image_url: str) -> Dict: """图像分析""" return { "objects": ["person", "car", "building"], "scene": "street", "confidence": 0.95 } def _analyze_sentiment(self, text: str) -> Dict: """情感分析""" # 简化实现 positive_words = ["good", "great", "excellent", "happy"] negative_words = ["bad", "terrible", "awful", "sad"] words = text.lower().split() positive_count = sum(1 for word in words if word in positive_words) negative_count = sum(1 for word in words if word in negative_words) if positive_count > negative_count: sentiment = "positive" elif negative_count > positive_count: sentiment = "negative" else: sentiment = "neutral" return { "sentiment": sentiment, "score": (positive_count - negative_count) / len(words) } 智能合约与AI协作 AI辅助的智能合约审计 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 from transformers import AutoTokenizer, AutoModelForCausalLM from typing import List, Dict class AIContractAuditor: """AI智能合约审计助手""" def __init__(self, model_name="microsoft/CodeGPT-small"): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForCausalLM.from_pretrained(model_name) def audit_contract( self, contract_source: str ) -> Dict: """审计智能合约""" # 1. 代码结构分析 structure_analysis = self._analyze_structure(contract_source) # 2. 漏洞检测 vulnerabilities = self._detect_vulnerabilities(contract_source) # 3. 最佳实践检查 best_practices = self._check_best_practices(contract_source) # 4. Gas优化建议 gas_optimization = self._suggest_gas_optimization(contract_source) # 5. 生成审计报告 report = self._generate_report({ "structure": structure_analysis, "vulnerabilities": vulnerabilities, "best_practices": best_practices, "gas_optimization": gas_optimization }) return report def _analyze_structure(self, source: str) -> Dict: """分析合约结构""" prompt = f""" Analyze the following smart contract code structure: {source} Provide: 1. Contract architecture 2. Key functions and their roles 3. State variables and their purposes 4. Access control mechanisms 5. External dependencies """ response = self._generate(prompt) return { "architecture": self._parse_response(response, "architecture"), "functions": self._parse_response(response, "functions"), "state_variables": self._parse_response(response, "state_variables"), "access_control": self._parse_response(response, "access_control"), "dependencies": self._parse_response(response, "dependencies") } def _detect_vulnerabilities(self, source: str) -> List[Dict]: """检测漏洞""" known_vulnerabilities = { "reentrancy": { "patterns": [ r"\.call\{.*value:\s*msg\.value", r"\.send\{.*value:\s*msg\.value" ], "severity": "critical", "description": "Reentrancy vulnerability detected" }, "overflow": { "patterns": [ r"uint256.*=.*\+.*(?!\.add\()", r"uint256.*=.*-.*(?!\.sub\()" ], "severity": "high", "description": "Potential integer overflow/underflow" }, "access_control": { "patterns": [ r"function\s+\w+\s*\(\s*\)\s*public(?!\s*onlyOwner|onlyRole)", r"tx\.origin" ], "severity": "high", "description": "Weak access control" }, "unchecked_call": { "patterns": [ r"\.call\s*\(", r"\.send\s*\(" ], "severity": "medium", "description": "Unchecked external call" } } detected = [] for vuln_type, vuln_info in known_vulnerabilities.items(): for pattern in vuln_info["patterns"]: matches = re.finditer(pattern, source) for match in matches: detected.append({ "type": vuln_type, "severity": vuln_info["severity"], "description": vuln_info["description"], "location": match.span(), "code_snippet": source[match.start()-20:match.end()+20] }) return detected def _check_best_practices(self, source: str) -> List[Dict]: """检查最佳实践""" checks = { "uses_safe_math": r"SafeMath|\.add\(|\.sub\(" in source, "has_reentrancy_guard": r"ReentrancyGuard|nonReentrant" in source, "uses_openzeppelin": r"@openzeppelin" in source, "has_events": r"event\s+\w+" in source, "uses_checks_effects_interactions": r"Checks-Effects-Interactions" in source, "has_pause": r"whenNotPaused|Pausable" in source, "has_timelock": r"TimelockController|releaseTimeLock" in source } results = [] for check_name, check_result in checks.items(): results.append({ "check": check_name, "passed": check_result, "description": self._get_check_description(check_name) }) return results def _suggest_gas_optimization(self, source: str) -> List[str]: """建议Gas优化""" optimizations = [] # 检查循环 if "for (" in source: optimizations.append("Consider using unchecked blocks for loop iterations") # 检查storage操作 if re.search(r"uint256\s+public\s+\w+", source): optimizations.append("Consider packing struct variables to save storage") # 检查重复计算 if re.search(r"keccak256\(", source): optimizations.append("Cache keccak256 results in local variables") # 检查memory vs storage if re.search(r".*\.\w+\s*=\s*\w+\[.*\]\s*\+\s*1", source): optimizations.append("Consider using calldata instead of memory for arrays") return optimizations def _generate_report(self, audit_data: Dict) -> Dict: """生成审计报告""" # 计算风险评分 risk_score = self._calculate_risk_score(audit_data) # 生成总结 summary = self._generate_summary(audit_data, risk_score) # 生成修复建议 recommendations = self._generate_recommendations(audit_data) return { "risk_score": risk_score, "summary": summary, "vulnerabilities": audit_data["vulnerabilities"], "best_practices": audit_data["best_practices"], "optimizations": audit_data["gas_optimization"], "recommendations": recommendations } def _calculate_risk_score(self, audit_data: Dict) -> int: """计算风险评分（0-100）""" score = 100 for vuln in audit_data["vulnerabilities"]: if vuln["severity"] == "critical": score -= 30 elif vuln["severity"] == "high": score -= 15 elif vuln["severity"] == "medium": score -= 5 elif vuln["severity"] == "low": score -= 2 return max(score, 0) def _generate_summary(self, audit_data: Dict, risk_score: int) -> str: """生成审计总结""" vuln_count = len(audit_data["vulnerabilities"]) critical_count = sum(1 for v in audit_data["vulnerabilities"] if v["severity"] == "critical") summary = f""" Smart Contract Audit Summary ========================== Risk Score: {risk_score}/100 Total Vulnerabilities: {vuln_count} Critical Issues: {critical_count} """ if risk_score >= 80: summary += "Overall Assessment: LOW RISK" elif risk_score >= 50: summary += "Overall Assessment: MEDIUM RISK" else: summary += "Overall Assessment: HIGH RISK" return summary def _generate_recommendations(self, audit_data: Dict) -> List[str]: """生成修复建议""" recommendations = [] for vuln in audit_data["vulnerabilities"]: if vuln["type"] == "reentrancy": recommendations.append( "Use ReentrancyGuard or implement Checks-Effects-Interactions pattern" ) elif vuln["type"] == "overflow": recommendations.append( "Use Solidity 0.8.0+ or SafeMath library for arithmetic operations" ) elif vuln["type"] == "access_control": recommendations.append( "Implement proper access control using onlyOwner or role-based access" ) return recommendations def _generate(self, prompt: str) -> str: """生成文本""" inputs = self.tokenizer(prompt, return_tensors="pt") with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=500, temperature=0.3, do_sample=True ) response = self.tokenizer.decode(outputs[0], skip_special_tokens=True) return response 自主交易Agent DeFi交易Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 from typing import List, Dict, Optional from datetime import datetime import ccxt import pandas as pd import numpy as np class DeFiTradingAgent: """DeFi交易Agent""" def __init__( self, initial_capital: float, exchanges: List[str], llm_model: str = "gpt-4" ): self.capital = initial_capital self.portfolio = {} # {token: amount} self.exchanges = {} self.trade_history = [] # 初始化交易所连接 for exchange_name in exchanges: if exchange_name == "uniswap": exchange = ccxt.uniswap({ "enableRateLimit": True }) elif exchange_name == "pancakeswap": exchange = ccxt.pancakeswap({ "enableRateLimit": True }) else: exchange = ccxt.binance({ "enableRateLimit": True }) self.exchanges[exchange_name] = exchange # 初始化LLM self.llm = self._init_llm(llm_model) def _init_llm(self, model_name: str): """初始化LLM""" # 实际应用中，这里应该连接到真实的LLM API # 或运行本地模型 from transformers import AutoTokenizer, AutoModelForCausalLM model = AutoModelForCausalLM.from_pretrained(model_name) tokenizer = AutoTokenizer.from_pretrained(model_name) return { "model": model, "tokenizer": tokenizer } def analyze_market( self, tokens: List[str], timeframe: str = "1h" ) -> Dict: """分析市场""" # 获取市场数据 market_data = self._fetch_market_data(tokens, timeframe) # 技术分析 ta_analysis = self._technical_analysis(market_data) # 使用LLM生成市场洞察 market_insight = self._generate_market_insight( market_data, ta_analysis ) return { "market_data": market_data, "technical_analysis": ta_analysis, "insight": market_insight } def _fetch_market_data( self, tokens: List[str], timeframe: str ) -> Dict: """获取市场数据""" data = {} for token in tokens: # 从各个交易所获取数据 for exchange_name, exchange in self.exchanges.items(): try: ohlcv = exchange.fetch_ohlcv( f"{token}/USDT", timeframe, limit=100 ) if token not in data: data[token] = [] # 转换为DataFrame df = pd.DataFrame( ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'] ) df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') data[token].append(df) break # 使用第一个成功的数据源 except Exception as e: print(f"Error fetching data for {token}: {e}") return data def _technical_analysis(self, market_data: Dict) -> Dict: """技术分析""" analysis = {} for token, dfs in market_data.items(): if not dfs: continue df = dfs[0] # 使用第一个数据源 # 计算技术指标 df['sma_20'] = df['close'].rolling(window=20).mean() df['sma_50'] = df['close'].rolling(window=50).mean() df['rsi'] = self._calculate_rsi(df['close'], 14) df['macd'] = self._calculate_macd(df['close']) # 趋势分析 latest_close = df['close'].iloc[-1] sma_20 = df['sma_20'].iloc[-1] sma_50 = df['sma_50'].iloc[-1] trend = "bullish" if latest_close > sma_20 > sma_50 else "bearish" analysis[token] = { "current_price": latest_close, "sma_20": sma_20, "sma_50": sma_50, "rsi": df['rsi'].iloc[-1], "trend": trend, "support_levels": self._find_support_levels(df), "resistance_levels": self._find_resistance_levels(df) } return analysis def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> pd.Series: """计算RSI""" delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi def _calculate_macd(self, prices: pd.Series) -> Dict: """计算MACD""" exp1 = prices.ewm(span=12, adjust=False).mean() exp2 = prices.ewm(span=26, adjust=False).mean() macd = exp1 - exp2 signal = macd.ewm(span=9, adjust=False).mean() histogram = macd - signal return { "macd": macd.iloc[-1], "signal": signal.iloc[-1], "histogram": histogram.iloc[-1] } def _find_support_levels(self, df: pd.DataFrame) -> List[float]: """寻找支撑位""" # 简化实现：使用局部最小值 from scipy.signal import argrelextrema prices = df['close'].values local_min = argrelextrema(prices, np.less, order=20) support_levels = sorted(prices[local_min]) return support_levels[-5:] # 返回最近的5个支撑位 def _find_resistance_levels(self, df: pd.DataFrame) -> List[float]: """寻找阻力位""" from scipy.signal import argrelextrema prices = df['close'].values local_max = argrelextrema(prices, np.greater, order=20) resistance_levels = sorted(prices[local_max], reverse=True) return resistance_levels[-5:] # 返回最近的5个阻力位 def _generate_market_insight( self, market_data: Dict, ta_analysis: Dict ) -> str: """生成市场洞察""" # 准备prompt prompt = f""" Analyze the following cryptocurrency market data and provide trading insights: Technical Analysis: {json.dumps(ta_analysis, indent=2)} Based on this analysis, provide: 1. Market trend analysis 2. Key support and resistance levels 3. Trading recommendations 4. Risk factors to consider 5. Optimal entry and exit points Be specific and actionable. """ # 调用LLM response = self._generate(prompt) return response def execute_trade( self, exchange: str, symbol: str, side: str, amount: float, price: Optional[float] = None ) -> Dict: """执行交易""" exchange_obj = self.exchanges[exchange] try: if side == "buy": # 限价买单 if price: order = exchange_obj.create_limit_buy_order( symbol, amount, price ) else: # 市价买单 order = exchange_obj.create_market_buy_order( symbol, amount ) else: # 卖单 if price: order = exchange_obj.create_limit_sell_order( symbol, amount, price ) else: order = exchange_obj.create_market_sell_order( symbol, amount ) # 记录交易 trade_record = { "exchange": exchange, "symbol": symbol, "side": side, "amount": amount, "price": price, "timestamp": datetime.now().isoformat(), "status": "executed" } self.trade_history.append(trade_record) return trade_record except Exception as e: print(f"Trade execution failed: {e}") return { "status": "failed", "error": str(e) } def run_strategy( self, strategy_config: Dict ) -> List[Dict]: """运行交易策略""" # 1. 分析市场 market_analysis = self.analyze_market( strategy_config["tokens"], strategy_config.get("timeframe", "1h") ) # 2. 生成交易信号 signals = self._generate_trading_signals( market_analysis, strategy_config ) # 3. 执行交易 executed_trades = [] for signal in signals: if signal["action"] == "hold": continue trade = self.execute_trade( exchange=signal["exchange"], symbol=signal["symbol"], side=signal["side"], amount=signal["amount"], price=signal.get("price") ) if trade.get("status") == "executed": executed_trades.append(trade) return executed_trades def _generate_trading_signals( self, market_analysis: Dict, strategy_config: Dict ) -> List[Dict]: """生成交易信号""" signals = [] ta_analysis = market_analysis["technical_analysis"] for token, analysis in ta_analysis.items(): # 简单的移动平均策略 if (analysis["trend"] == "bullish" and analysis["rsi"] < 70 and analysis["current_price"] > analysis["sma_20"]): signals.append({ "action": "buy", "exchange": "uniswap", "symbol": f"{token}/USDT", "side": "buy", "amount": strategy_config.get("trade_size", 100), "reason": "Bullish trend with RSI below overbought" }) elif (analysis["trend"] == "bearish" and analysis["rsi"] > 30): signals.append({ "action": "sell", "exchange": "uniswap", "symbol": f"{token}/USDT", "side": "sell", "amount": strategy_config.get("trade_size", 100), "reason": "Bearish trend detected" }) return signals DAO治理Agent 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 from typing import List, Dict, Optional class DAOGovernanceAgent: """DAO治理Agent""" def __init__( self, dao_address: str, llm_model: str = "gpt-4" ): self.dao_address = dao_address self.llm = self._init_llm(llm_model) # 治理历史 self.governance_history = [] # 提案分析 self.proposals_db = {} def analyze_proposal( self, proposal_data: Dict ) -> Dict: """分析提案""" # 1. 提取关键信息 key_info = self._extract_proposal_info(proposal_data) # 2. 风险评估 risk_assessment = self._assess_risk(proposal_data) # 3. 财务影响分析 financial_impact = self._analyze_financial_impact(proposal_data) # 4. 生成投票建议 voting_recommendation = self._generate_voting_recommendation({ "key_info": key_info, "risk_assessment": risk_assessment, "financial_impact": financial_impact }) return { "proposal_id": proposal_data["id"], "key_info": key_info, "risk_assessment": risk_assessment, "financial_impact": financial_impact, "recommendation": voting_recommendation } def _extract_proposal_info(self, proposal: Dict) -> Dict: """提取提案关键信息""" prompt = f""" Extract key information from this DAO proposal: Title: {proposal.get('title', '')} Description: {proposal.get('description', '')} Please extract: 1. Proposal type (e.g., parameter change, spending, governance change) 2. Key changes proposed 3. Affected stakeholders 4. Implementation timeline 5. Required resources """ response = self._generate(prompt) # 解析LLM响应 key_info = { "type": self._parse_field(response, "Proposal type"), "changes": self._parse_field(response, "Key changes"), "stakeholders": self._parse_field(response, "Stakeholders"), "timeline": self._parse_field(response, "Timeline"), "resources": self._parse_field(response, "Resources") } return key_info def _assess_risk(self, proposal: Dict) -> Dict: """评估风险""" risk_factors = [] # 检查提案类型 proposal_type = self._extract_proposal_type(proposal) if proposal_type == "spending": # 检查金额 amount = self._extract_amount(proposal) if amount > 1000000: risk_factors.append({ "type": "financial", "severity": "high", "description": "Large expenditure proposed" }) elif proposal_type == "parameter_change": # 检查参数范围 params = self._extract_parameters(proposal) if self._is_risk_parameter_change(params): risk_factors.append({ "type": "governance", "severity": "medium", "description": "Parameter changes may affect protocol stability" }) return { "risk_score": self._calculate_risk_score(risk_factors), "risk_factors": risk_factors, "mitigation_strategies": self._suggest_mitigation(risk_factors) } def _generate_voting_recommendation(self, analysis: Dict) -> Dict: """生成投票建议""" prompt = f""" Based on the following DAO proposal analysis: Key Information: {json.dumps(analysis['key_info'], indent=2)} Risk Assessment: {json.dumps(analysis['risk_assessment'], indent=2)} Financial Impact: {json.dumps(analysis['financial_impact'], indent=2)} Provide a voting recommendation: 1. Vote: For/Against/Abstain 2. Confidence: High/Medium/Low 3. Reasoning: Detailed explanation 4. Conditions: Any conditions for changing the vote Consider: - Long-term sustainability - Community impact - Financial health - Innovation vs stability """ response = self._generate(prompt) # 解析建议 recommendation = { "vote": self._parse_field(response, "Vote"), "confidence": self._parse_field(response, "Confidence"), "reasoning": self._parse_field(response, "Reasoning"), "conditions": self._parse_field(response, "Conditions") } return recommendation def automate_governance(self) -> None: """自动化治理决策""" # 获取待处理提案 pending_proposals = self._fetch_pending_proposals() for proposal in pending_proposals: # 分析提案 analysis = self.analyze_proposal(proposal) # 根据建议自动投票 if analysis["recommendation"]["vote"].lower() == "for": self._cast_vote( proposal["id"], "for", analysis["recommendation"]["reasoning"] ) def _fetch_pending_proposals(self) -> List[Dict]: """获取待处理提案""" # 实际应用中，这里应该从链上或DAO的API获取 # 简化实现 proposals = [] # 示例提案 proposals.append({ "id": "proposal-123", "title": "Grant Program Funding", "description": "Allocate $500,000 for grants", "status": "pending", "voting_deadline": datetime.now() + timedelta(days=7) }) return proposals def _cast_vote(self, proposal_id: str, vote: str, reason: str) -> str: """投票""" # 实际应用中，这里应该调用链上治理合约 print(f"Voting {vote} on proposal {proposal_id}") print(f"Reason: {reason}") return f"voted-{vote}-{proposal_id}" 总结 AI Agent与Web3的融合将开启全新的应用范式： ...

引言 DeFi（去中心化金融）是WEB3最重要的应用场景之一。从AMM到借贷协议，DeFi正在重塑传统金融。本文将深入探讨DeFi协议的核心机制和开发实践。 AMM（自动做市商） 恒定乘积AMM（Uniswap V2） 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; contract UniswapV2Pair is ReentrancyGuard { string public constant name = "Uniswap V2 Pair"; string public constant symbol = "UNI-V2"; uint256 public constant MINIMUM_LIQUIDITY = 1000; address public token0; address public token1; uint256 public reserve0; uint256 public reserve1; uint256 public totalSupply; mapping(address => uint256) public balanceOf; event Mint(address indexed sender, uint256 amount0, uint256 amount1); event Burn(address indexed sender, uint256 amount0, uint256 amount1); event Swap( address indexed sender, uint256 amount0In, uint256 amount1In, uint256 amount0Out, uint256 amount1Out, address indexed to ); event Sync(uint256 reserve0, uint256 reserve1); constructor() { factory = msg.sender; } function initialize(address _token0, address _token1) external { require(msg.sender == factory, "Forbidden"); require(_token0 < _token1, "Invalid tokens"); token0 = _token0; token1 = _token1; } // 添加流动性 function mint(address to) external nonReentrant returns (uint256 liquidity) { (uint256 reserve0_, uint256 reserve1_) = getReserves(); uint256 balance0 = IERC20(token0).balanceOf(address(this)); uint256 balance1 = IERC20(token1).balanceOf(address(this)); uint256 amount0 = balance0 - reserve0_; uint256 amount1 = balance1 - reserve1_; uint256 _totalSupply = totalSupply; if (_totalSupply == 0) { // 首次添加流动性 liquidity = Math.sqrt(amount0 * amount1) - MINIMUM_LIQUIDITY; _mint(address(0), MINIMUM_LIQUIDITY); // 永久锁定最小流动性 } else { liquidity = Math.min( (amount0 * _totalSupply) / reserve0_, (amount1 * _totalSupply) / reserve1_ ); } require(liquidity > 0, "Insufficient liquidity minted"); _mint(to, liquidity); _update(balance0, balance1); emit Mint(to, amount0, amount1); } // 移除流动性 function burn(address to) external nonReentrant returns (uint256 amount0, uint256 amount1) { uint256 liquidity = balanceOf[address(this)]; require(liquidity > 0, "No liquidity"); (uint256 reserve0_, uint256 reserve1_) = getReserves(); uint256 _totalSupply = totalSupply; amount0 = (liquidity * reserve0_) / _totalSupply; amount1 = (liquidity * reserve1_) / _totalSupply; _burn(address(this), liquidity); _transfer( token0, address(this), to, amount0 ); _transfer( token1, address(this), to, amount1 ); (uint256 balance0, uint256 balance1) = getBalances(); _update(balance0, balance1); emit Burn(to, amount0, amount1); } // 交换（核心功能） function swap( uint256 amount0Out, uint256 amount1Out, address to, bytes calldata data ) external nonReentrant { require( amount0Out > 0 || amount1Out > 0, "Insufficient output amount" ); (uint256 reserve0_, uint256 reserve1_) = getReserves(); if (amount0Out > 0) { uint256 amount0In = getInputAmount( amount0Out, reserve0_, reserve1_ ); require( amount0In <= reserve0_ - amount0Out, "Insufficient liquidity" ); uint256 balance0Before = IERC20(token0).balanceOf(address(this)); _transfer(token0, msg.sender, address(this), amount0In); uint256 balance0After = IERC20(token0).balanceOf(address(this)); amount0In = balance0After - balance0Before; _transfer(token1, address(this), to, amount0Out); } if (amount1Out > 0) { uint256 amount1In = getInputAmount( amount1Out, reserve1_, reserve0_ ); require( amount1In <= reserve1_ - amount1Out, "Insufficient liquidity" ); uint256 balance1Before = IERC20(token1).balanceOf(address(this)); _transfer(token1, msg.sender, address(this), amount1In); uint256 balance1After = IERC20(token1).balanceOf(address(this)); amount1In = balance1After - balance1Before; _transfer(token0, address(this), to, amount1Out); } (uint256 balance0, uint256 balance1) = getBalances(); _update(balance0, balance1); emit Swap( msg.sender, amount0In, amount1In, amount0Out, amount1Out, to ); } // 计算输入量（恒定乘积公式） function getInputAmount( uint256 outputAmount, uint256 inputReserve, uint256 outputReserve ) public pure returns (uint256 inputAmount) { require(inputReserve > 0 && outputReserve > 0, "Invalid reserves"); require(outputAmount < outputReserve, "Output amount too high"); uint256 numerator = inputReserve * outputAmount * 1000; uint256 denominator = (outputReserve - outputAmount) * 997; return (numerator / denominator) + 1; } // 滑点计算 function getAmountOut( uint256 amountIn, uint256 reserveIn, uint256 reserveOut ) public pure returns (uint256 amountOut) { require(amountIn > 0, "Insufficient input amount"); require(reserveIn > 0 && reserveOut > 0, "Invalid reserves"); uint256 amountInWithFee = amountIn * 997; uint256 numerator = amountInWithFee * reserveOut; uint256 denominator = reserveIn * 1000 + amountInWithFee; return numerator / denominator; } function getReserves() public view returns (uint256, uint256) { return (reserve0, reserve1); } function _update(uint256 balance0, uint256 balance1) private { reserve0 = balance0; reserve1 = balance1; emit Sync(balance0, balance1); } function _mint(address to, uint256 amount) private { totalSupply += amount; balanceOf[to] += amount; } function _burn(address from, uint256 amount) private { require(balanceOf[from] >= amount, "Insufficient balance"); balanceOf[from] -= amount; totalSupply -= amount; } function _transfer( address token, address from, address to, uint256 amount ) private { IERC20(token).transferFrom(from, to, amount); } function getBalances() public view returns (uint256, uint256) { return ( IERC20(token0).balanceOf(address(this)), IERC20(token1).balanceOf(address(this)) ); } } 集中流动性（Uniswap V3） 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 // Uniswap V3核心概念 contract UniswapV3Pool { struct Position { uint96 nonce; address operator; address token0; address token1; int24 tickLower; int24 tickUpper; uint128 liquidity; uint256 feeGrowthInside0LastX128; uint256 feeGrowthInside1LastX128; uint128 tokensOwed0; uint128 tokensOwed1; } // Tick（价格）概念 // 价格 = 1.0001^tick // 例如：tick = 1000 => price = 1.0001^1000 ≈ 1.105 int24 internal constant MIN_TICK = -887272; int24 internal constant MAX_TICK = 887272; function getRatioFromTick(int24 tick) public pure returns (uint256) { uint256 ratio = 1.0001e18; int24 absTick = tick < 0 ? -tick : tick; for (int i = 0; i < absTick; i++) { if (tick < 0) { ratio = (ratio * 1e18) / 1000100000000000000; // /1.0001 } else { ratio = (ratio * 1000100000000000000) / 1e18; // *1.0001 } } return ratio; } function getTickFromRatio(uint256 ratio) public pure returns (int24 tick) { // 二分查找 int24 low = MIN_TICK; int24 high = MAX_TICK; while (low < high) { int24 mid = (low + high + 1) / 2; uint256 midRatio = getRatioFromTick(mid); if (ratio < midRatio) { high = mid - 1; } else { low = mid; } } return low; } // 流动性计算 function getLiquidityForAmounts( uint160 sqrtRatioAX96, uint160 sqrtRatioBX96, uint256 amount0, uint256 amount1 ) public pure returns (uint128 liquidity) { if (sqrtRatioAX96 > sqrtRatioBX96) (sqrtRatioAX96, sqrtRatioBX96) = (sqrtRatioBX96, sqrtRatioAX96); uint256 intermediate = sqrtRatioAX96 * sqrtRatioBX96 / 96; uint256 amount0Intermediate = (amount0 * intermediate) / sqrtRatioBX96; if (amount0Intermediate <= amount1) { liquidity = uint128(amount0Intermediate); } else { liquidity = uint128((amount1 * sqrtRatioAX96 * sqrtRatioBX96) / 96); } } function getPositionAmounts( uint160 sqrtPriceX96, int24 tickLower, int24 tickUpper, uint128 liquidity ) public pure returns (uint256 amount0, uint256 amount1) { uint160 sqrtRatioAX96 = getSqrtRatioAtTick(tickLower); uint160 sqrtRatioBX96 = getSqrtRatioAtTick(tickUpper); if (sqrtPriceX96 <= sqrtRatioAX96) { amount0 = getAmount0ForLiquidity( sqrtRatioAX96, sqrtRatioBX96, liquidity ); } else if (sqrtPriceX96 < sqrtRatioBX96) { amount0 = getAmount0ForLiquidity( sqrtPriceX96, sqrtRatioBX96, liquidity ); amount1 = getAmount1ForLiquidity( sqrtRatioAX96, sqrtPriceX96, liquidity ); } else { amount1 = getAmount1ForLiquidity( sqrtRatioAX96, sqrtRatioBX96, liquidity ); } } } 借贷协议 Compound风格借贷 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; contract LendingPool is ReentrancyGuard { struct Reserve { uint256 totalSupply; uint256 totalBorrowed; uint256 borrowRate; uint256 supplyRate; uint256 lastUpdate; uint256 index; } struct UserState { uint256 supplied; uint256 borrowed; uint256 borrowIndex; uint256 supplyIndex; uint256 collateralFactor; } mapping(address => Reserve) public reserves; mapping(address => UserState) public users; mapping(address => address[]) public userAssets; address public constant WETH = 0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2; uint256 public constant COLLATERAL_FACTOR = 750; // 75% event Supply(address indexed user, address indexed asset, uint256 amount); event Borrow(address indexed user, address indexed asset, uint256 amount); event Repay(address indexed user, address indexed asset, uint256 amount); event Withdraw(address indexed user, address indexed asset, uint256 amount); event Liquidate(address indexed user, address indexed borrower, address indexed asset, uint256 amount); // 供应资产 function supply(address asset, uint256 amount) external nonReentrant { require(amount > 0, "Amount must be greater than 0"); IERC20(asset).transferFrom(msg.sender, address(this), amount); // 更新累计指数 _accrueInterest(asset); // 更新用户状态 UserState storage user = users[msg.sender]; Reserve storage reserve = reserves[asset]; uint256 userSupply = (amount * reserve.index) / 1e18; user.supplied += userSupply; user.supplyIndex = reserve.index; reserve.totalSupply += userSupply; // 记录用户资产 if (!_hasAsset(msg.sender, asset)) { userAssets[msg.sender].push(asset); } emit Supply(msg.sender, asset, amount); } // 借款 function borrow( address asset, uint256 amount ) external nonReentrant { require(amount > 0, "Amount must be greater than 0"); // 更新累计指数 _accrueInterest(asset); UserState storage user = users[msg.sender]; Reserve storage reserve = reserves[asset]; // 检查抵押品 uint256 maxBorrow = _getMaxBorrow(msg.sender); uint256 currentBorrow = _getUserBorrow(msg.sender); require( currentBorrow + amount <= maxBorrow, "Insufficient collateral" ); // 更新借款 uint256 borrowAmount = (amount * 1e18) / reserve.index; user.borrowed += borrowAmount; user.borrowIndex = reserve.index; reserve.totalBorrowed += borrowAmount; // 转出资产 IERC20(asset).transfer(msg.sender, amount); emit Borrow(msg.sender, asset, amount); } // 还款 function repay(address asset, uint256 amount) external nonReentrant { require(amount > 0, "Amount must be greater than 0"); // 更新累计指数 _accrueInterest(asset); UserState storage user = users[msg.sender]; Reserve storage reserve = reserves[asset]; // 计算实际债务 uint256 debt = _getUserBorrow(msg.sender); if (amount >= debt) { amount = debt; // 如果还清，可以提取抵押品 } IERC20(asset).transferFrom(msg.sender, address(this), amount); uint256 repayAmount = (amount * 1e18) / reserve.index; user.borrowed -= repayAmount; reserve.totalBorrowed -= (amount * 1e18) / reserve.index; emit Repay(msg.sender, asset, amount); } // 提取供应的资产 function withdraw(address asset, uint256 amount) external nonReentrant { require(amount > 0, "Amount must be greater than 0"); // 更新累计指数 _accrueInterest(asset); UserState storage user = users[msg.sender]; Reserve storage reserve = reserves[asset]; // 检查是否有未还贷款 uint256 currentBorrow = _getUserBorrow(msg.sender); uint256 maxBorrow = _getMaxBorrow(msg.sender); require( currentBorrow <= maxBorrow, "Cannot withdraw: insufficient collateral" ); // 计算可提取金额 uint256 userSupply = _getUserSupply(msg.sender, asset); require(userSupply >= amount, "Insufficient balance"); uint256 withdrawAmount = (amount * reserve.index) / 1e18; user.supplied -= withdrawAmount; reserve.totalSupply -= withdrawAmount; IERC20(asset).transfer(msg.sender, amount); emit Withdraw(msg.sender, asset, amount); } // 清算 function liquidate( address borrower, address asset, uint256 amount ) external nonReentrant { // 更新累计指数 _accrueInterest(asset); UserState storage user = users[borrower]; // 检查是否需要清算 uint256 currentBorrow = _getUserBorrow(borrower); uint256 maxBorrow = _getMaxBorrow(borrower); require( currentBorrow > maxBorrow, "Not eligible for liquidation" ); // 扣除抵押品 // 这里简化处理，实际需要拍卖机制 emit Liquidate(msg.sender, borrower, asset, amount); } // 计算累计利息 function _accrueInterest(address asset) internal { Reserve storage reserve = reserves[asset]; uint256 timeElapsed = block.timestamp - reserve.lastUpdate; if (timeElapsed == 0) return; // 简化的利息计算 uint256 interest = (reserve.totalBorrowed * reserve.borrowRate * timeElapsed) / (365 days * 1e18); uint256 supplyInterest = (reserve.totalSupply * reserve.supplyRate * timeElapsed) / (365 days * 1e18); reserve.totalBorrowed += interest; reserve.totalSupply += supplyInterest; reserve.lastUpdate = block.timestamp; } function _getUserSupply( address user, address asset ) internal view returns (uint256) { Reserve storage reserve = reserves[asset]; UserState storage userState = users[user]; if (userState.supplyIndex == 0) { return 0; } return (userState.supplied * reserve.index) / userState.supplyIndex; } function _getUserBorrow(address user) internal view returns (uint256) { uint256 totalBorrow = 0; for (uint256 i = 0; i < userAssets[user].length; i++) { address asset = userAssets[user][i]; UserState storage userState = users[user]; Reserve storage reserve = reserves[asset]; if (userState.borrowed > 0) { uint256 borrow = (userState.borrowed * reserve.index) / userState.borrowIndex; totalBorrow += borrow; } } return totalBorrow; } function _getMaxBorrow(address user) internal view returns (uint256) { uint256 totalCollateral = 0; for (uint256 i = 0; i < userAssets[user].length; i++) { address asset = userAssets[user][i]; uint256 supply = _getUserSupply(user, asset); // 假设所有资产都有同样的抵押因子 totalCollateral += (supply * COLLATERAL_FACTOR) / 1000; } return totalCollateral; } function _hasAsset(address user, address asset) internal view returns (bool) { for (uint256 i = 0; i < userAssets[user].length; i++) { if (userAssets[user][i] == asset) { return true; } } return false; } } 收益聚合器（Yearn风格） 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract YieldAggregator is Ownable, ReentrancyGuard { struct Strategy { address strategy; uint256 allocation; // 分配比例（基点） uint256 performanceFee; bool active; } mapping(address => Strategy) public strategies; address[] public strategyList; uint256 public constant MAX_ALLOCATION = 10000; // 100% uint256 public constant PERFORMANCE_FEE = 1000; // 10% uint256 public totalShares; mapping(address => uint256) public shares; mapping(address => uint256) public userPrincipal; event Deposit(address indexed user, uint256 amount); event Withdraw(address indexed user, uint256 amount); event Harvest(address indexed strategy, uint256 amount); event Rebalance(address[] strategies, uint256[] allocations); // 存款 function deposit(uint256 amount) external payable nonReentrant { require(amount > 0, "Amount must be greater than 0"); // 转入资产 if (msg.value > 0) { require(amount == msg.value, "ETH amount mismatch"); } else { IERC20(WETH).transferFrom(msg.sender, address(this), amount); IWETH(WETH).deposit{value: amount}(); } // 计算份额 uint256 shares; if (totalShares == 0) { shares = amount; } else { shares = (amount * totalShares) / totalAssets(); } shares[msg.sender] += shares; totalShares += shares; userPrincipal[msg.sender] += amount; // 分配到各个策略 _rebalance(); emit Deposit(msg.sender, amount); } // 提款 function withdraw(uint256 shares) external nonReentrant { require(shares > 0, "Shares must be greater than 0"); require(shares[msg.sender] >= shares, "Insufficient shares"); // 计算可提取金额 uint256 assets = (totalAssets() * shares) / totalShares; // 从策略中提取 _withdrawFromStrategies(assets); // 转出资产 if (address(this).balance >= assets) { payable(msg.sender).transfer(assets); } else { IWETH(WETH).withdraw(assets); payable(msg.sender).transfer(assets); } // 更新份额 shares[msg.sender] -= shares; totalShares -= shares; uint256 principal = (userPrincipal[msg.sender] * shares) / (shares + shares[msg.sender]); userPrincipal[msg.sender] -= principal; emit Withdraw(msg.sender, assets); } // 收获收益 function harvest(address strategy) external onlyOwner { Strategy storage s = strategies[strategy]; require(s.active, "Strategy not active"); // 调用策略的harvest函数 uint256 beforeBalance = address(this).balance; IStrategy(strategy).harvest(); uint256 afterBalance = address(this).balance; uint256 profit = afterBalance - beforeBalance; if (profit > 0) { // 提取性能费 uint256 fee = (profit * PERFORMANCE_FEE) / 10000; uint256 performanceFee = (fee * s.performanceFee) / 10000; payable(owner()).transfer(performanceFee); emit Harvest(strategy, profit); } } // 重新平衡 function _rebalance() internal { uint256 totalAssets = address(this).balance; for (uint256 i = 0; i < strategyList.length; i++) { Strategy storage s = strategies[strategyList[i]]; if (s.active && s.allocation > 0) { uint256 amount = (totalAssets * s.allocation) / MAX_ALLOCATION; IStrategy(s.strategy).invest{value: amount}(); } } } function _withdrawFromStrategies(uint256 amount) internal { uint256 withdrawn; for (uint256 i = 0; i < strategyList.length; i++) { Strategy storage s = strategies[strategyList[i]]; if (s.active && withdrawn < amount) { uint256 toWithdraw = amount - withdrawn; uint256 available = IStrategy(s.strategy).withdraw(toWithdraw); withdrawn += available; if (withdrawn >= amount) { break; } } } } function totalAssets() public view returns (uint256) { uint256 total = address(this).balance; for (uint256 i = 0; i < strategyList.length; i++) { Strategy storage s = strategies[strategyList[i]]; if (s.active) { total += IStrategy(s.strategy).estimatedTotalAssets(); } } return total; } // 添加策略 function addStrategy( address _strategy, uint256 _allocation ) external onlyOwner { require(_strategy != address(0), "Invalid strategy"); strategies[_strategy] = Strategy({ strategy: _strategy, allocation: _allocation, performanceFee: 5000, // 50% active: true }); strategyList.push(_strategy); _validateAllocations(); } // 更新分配比例 function updateAllocation( address _strategy, uint256 _allocation ) external onlyOwner { strategies[_strategy].allocation = _allocation; _validateAllocations(); _rebalance(); } function _validateAllocations() internal view { uint256 total; for (uint256 i = 0; i < strategyList.length; i++) { Strategy storage s = strategies[strategyList[i]]; if (s.active) { total += s.allocation; } } require(total <= MAX_ALLOCATION, "Total allocation exceeds 100%"); } } interface IStrategy { function invest(uint256 amount) external; function withdraw(uint256 amount) external returns (uint256); function harvest() external; function estimatedTotalAssets() external view returns (uint256); } interface IWETH { function deposit() external payable; function withdraw(uint256 wad) external; } 总结 DeFi协议开发需要深入理解： ...

引言 智能合约安全是WEB3生态的生命线。一次漏洞可能导致数千万甚至数亿美元的损失。本文将系统性地探讨智能合约安全审计的完整方法论，从常见漏洞到审计工具，再到最佳实践。 常见漏洞类型 重入攻击（Reentrancy） 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 // ❌ 有重入漏洞的合约 contract VulnerableBank { mapping(address => uint256) public balances; function deposit() public payable { balances[msg.sender] += msg.value; } function withdraw(uint256 amount) public { require(balances[msg.sender] >= amount, "Insufficient balance"); // 滑洞：在更新状态前进行外部调用 (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); balances[msg.sender] -= amount; } } // ✅ 修复后的合约（使用Checks-Effects-Interactions模式） contract SecureBank { mapping(address => uint256) public balances; function deposit() public payable { balances[msg.sender] += msg.value; } function withdraw(uint256 amount) public { require(balances[msg.sender] >= amount, "Insufficient balance"); // 先更新状态 balances[msg.sender] -= amount; // 再进行外部调用 (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } // ✅ 使用ReentrancyGuard import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; contract GuardedBank is ReentrancyGuard { mapping(address => uint256) public balances; function withdraw(uint256 amount) external nonReentrant { require(balances[msg.sender] >= amount, "Insufficient balance"); balances[msg.sender] -= amount; (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } 整数溢出/下溢 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 // ❌ Solidity 0.8.0之前的溢出漏洞 contract OldVulnerable { uint256 public value; function unsafeAdd(uint256 a, uint256 b) public { // 可能溢出 value = a + b; } function unsafeSubtract(uint256 a, uint256 b) public { // 可能下溢 value = a - b; } } // ✅ Solidity 0.8.0+自动检查溢出 contract ModernSafe { uint256 public value; function safeAdd(uint256 a, uint256 b) public { // Solidity 0.8.0+自动检查溢出 value = a + b; } function safeSubtract(uint256 a, uint256 b) public { // 自动检查下溢 value = a - b; } // 使用SafeMath库（0.8.0之前） // using SafeMath for uint256; } // ✅ 使用OpenZeppelin的SafeMath（旧版Solidity） import "@openzeppelin/contracts/utils/math/SafeMath.sol"; contract SafeMathContract { using SafeMath for uint256; function add(uint256 a, uint256 b) public pure returns (uint256) { return a.add(b); // 自动检查溢出 } function sub(uint256 a, uint256 b) public pure returns (uint256) { return a.sub(b); // 自动检查下溢 } } 访问控制漏洞 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 // ❌ 缺少访问控制 contract NoAccessControl { uint256 public importantValue; function setImportantValue(uint256 _value) public { // 任何人都可以调用 importantValue = _value; } function destroy() public { // 任何人都可以销毁合约 selfdestruct(payable(msg.sender)); } } // ✅ 正确的访问控制 import "@openzeppelin/contracts/access/Ownable.sol"; import "@openzeppelin/contracts/access/AccessControl.sol"; contract ProperAccessControl is Ownable, AccessControl { uint256 public importantValue; bytes32 public constant ADMIN_ROLE = keccak256("ADMIN_ROLE"); bytes32 public constant MANAGER_ROLE = keccak256("MANAGER_ROLE"); constructor() Ownable(msg.sender) { _grantRole(DEFAULT_ADMIN_ROLE, msg.sender); _grantRole(ADMIN_ROLE, msg.sender); _setRoleAdmin(MANAGER_ROLE, ADMIN_ROLE); } // 只有所有者可以调用 function setImportantValue(uint256 _value) external onlyOwner { importantValue = _value; } // 只有管理员可以调用 function adminFunction() external onlyRole(ADMIN_ROLE) { // 管理员专属功能 } // 管理员或经理可以调用 function managerFunction() external onlyRole(ADMIN_ROLE) onlyRole(MANAGER_ROLE) { // 功能实现 } // 基于时间的访问控制 modifier onlyBefore(uint256 deadline) { require(block.timestamp < deadline, "Deadline passed"); _; } function timedFunction() external onlyBefore(1735689600) { // 只能在指定时间前调用 } // 多重签名 mapping(bytes32 => bool) public signatures; uint256 public requiredSignatures = 2; function multiSigFunction(bytes32 data) external { bytes32 signature = keccak256(abi.encodePacked(data, msg.sender)); signatures[signature] = true; uint256 count; bytes32 hash; for (uint256 i = 0; i < 255; i++) { hash = keccak256(abi.encodePacked(data, i)); if (signatures[hash]) { count++; } } require(count >= requiredSignatures, "Not enough signatures"); } } 前端运行（Front-Running） 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 // ❌ 容易被抢跑的合约 contract FrontRunnable { mapping(uint256 => uint256) public bids; uint256 public auctionEnd; function bid(uint256 amount) external payable { require(block.timestamp < auctionEnd, "Auction ended"); // 滑洞：未隐藏出价，容易被抢跑 bids[msg.sender] = msg.value; if (msg.value > bids[highestBidder]) { highestBidder = msg.sender; } } } // ✅ 使用commit-reveal方案 import "@openzeppelin/contracts/utils/ReentrancyGuard.sol"; class CommitRevealAuction is ReentrancyGuard { struct Commitment { bytes32 hash; uint256 amount; bool revealed; } mapping(address => Commitment) public commitments; uint256 public commitDeadline; uint256 public revealDeadline; uint256 public highestBid; address public highestBidder; function commit(bytes32 hash) external payable { require(block.timestamp < commitDeadline, "Commit period ended"); require(msg.value > 0, "Must commit with ETH"); commitments[msg.sender] = Commitment({ hash: hash, amount: msg.value, revealed: false }); } function reveal(uint256 value, bytes32 salt) external nonReentrant { require( block.timestamp >= commitDeadline && block.timestamp < revealDeadline, "Not in reveal period" ); bytes32 hash = keccak256(abi.encodePacked(value, salt)); require(commitments[msg.sender].hash == hash, "Invalid reveal"); commitments[msg.sender].revealed = true; if (value > highestBid) { // 退还之前的最高出价 if (highestBidder != address(0)) { payable(highestBidder).transfer(highestBid); } highestBid = value; highestBidder = msg.sender; } } } // ✅ 使用暗池（暗拍卖） import "@openzeppelin/contracts/utils/cryptography/ECDSA.sol"; contract SealedBidAuction { struct Bid { bytes32 blindedBid; uint256 deposit; } mapping(address => Bid) public bids; mapping(address => uint256) public refunds; uint256 public auctionEnd; address public highestBidder; uint256 public highestBid; bool public ended; function bid(bytes32 blindedBid) external payable { require(block.timestamp < auctionEnd, "Auction ended"); require(msg.value >= highestBid / 10, "Deposit too low"); bids[msg.sender] = Bid({ blindedBid: blindedBid, deposit: msg.value }); } function reveal( uint256[] calldata values, bytes32[] calldata secrets ) external { require( block.timestamp >= auctionEnd && !ended, "Cannot reveal" ); for (uint256 i = 0; i < values.length; i++) { address bidder = msg.sender; Bid storage bid = bids[bidder]; bytes32 hash = keccak256(abi.encodePacked(values[i], secrets[i])); if (hash != bid.blindedBid) { refunds[bidder] += bid.deposit; continue; } if (bid.deposit < values[i]) { refunds[bidder] += bid.deposit; continue; } if (values[i] > highestBid) { if (highestBidder != address(0)) { refunds[highestBidder] += highestBid; } highestBidder = bidder; highestBid = values[i]; } refunds[bidder] += bid.deposit - values[i]; } } function withdrawRefund() external { uint256 refund = refunds[msg.sender]; refunds[msg.sender] = 0; payable(msg.sender).transfer(refund); } } 静态分析工具 Slither 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 # Slither安装 # pip install slither-analyzer # 基础扫描 slither contract.sol # 生成报告 slither contract.sol --json output.json slither contract.sol --markdown output.md # 自定义打印机 from slither import Slither from slither.detectors import ReentrancyDetector from slither.printers import CustomPrinter slither = Slither('contract.sol') # 检测重入漏洞 for detector in slither.detectors: if isinstance(detector, ReentrancyDetector): for finding in detector.detect(): print(f"Reentrancy found: {finding}") # 自定义检测器 from slither.detectors.abstract_detector import AbstractDetector, DetectorClassification class MyCustomDetector(AbstractDetector): ARGUMENT = 'my-custom-detector' HELP = 'Custom detector description' IMPACT = DetectorClassification.HIGH CONFIDENCE = DetectorClassification.HIGH WIKI = 'https://github.com/my-detector/wiki' def detect(self): results = [] for contract in self.contracts: for function in contract.functions: # 自定义检测逻辑 if self.has_vulnerability(function): results.append({ 'contract': contract.name, 'function': function.name, 'line': function.source_mapping['start']['line'] }) return results Mythril 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 # Mythril安装 # pip install mythril # 命令行使用 myth analyze contract.sol # Python API from mythril.platform import mythril_platform from mythril.analysis import symbolic_executor def analyze_contract(contract_path: str): platform = mythril_platform.get_platform() # 加载合约 platform.set_execution_timeout(30) platform.load_bytecode(contract_path) # 执行符号执行 executor = symbolic_executor.SymbolicExecutor() issues = executor.execute(platform.bytecode) # 分析结果 for issue in issues: print(f"[{issue.severity}] {issue.title}") print(f" Description: {issue.description}") print(f" SWC ID: {issue.swc_id}") print() # 自定义分析规则 from mythril.analysis.issue import Severity from mythril.analysis.reporter import Issue class MyCustomAnalyzer: def __init__(self): self.issues = [] def check_access_control(self, bytecode): # 检查访问控制问题 if not self.has_access_control(bytecode): self.issues.append(Issue( severity=Severity.HIGH, title="Missing Access Control", description="Critical functions lack access control", swc_id="SWC-105" )) def has_access_control(self, bytecode): # 实现检查逻辑 return True Echidna 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 # Echidna安装 # git clone https://github.com/crytic/echidna.git # cd echidna # cabal install # Echidna配置文件 """ echidna-test: # 测试用例 testMode: assertion # 最大时间（秒） testLimit: 50000 # 最大序列长度 seqLen: 20 # 合约覆盖率 coverage: true # 指定部署账户 deployer: "0x00a329c0648769a73afac7f9381e08fb43dbea70" """ # Solidity测试合约 // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "echidna-test.sol"; contract VulnerableContract { uint256 public publicVar = 100; // 不变式：publicVar应该始终 <= 100 function invariant_publicVar_not_greater_than_100() public view { assert(publicVar <= 100); } // 有漏洞的函数 function setPublicVar(uint256 _value) public { publicVar = _value; // Echidna会发现这里违反了不变式 } // 正确的函数 function safeSetPublicVar(uint256 _value) public { require(_value <= 100, "Value too large"); publicVar = _value; } } // 运行Echidna // echidna-test contract.sol --test-mode assertion 形式化验证 SMT求解器验证 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 from z3 import * def verify_swap_function(): """使用Z3验证swap函数的正确性""" # 定义变量 x = Real('x') # 用户输入的token A数量 y = Real('y') # 用户输入的token B数量 reserve_x = Real('reserve_x') # 池中token A储备 reserve_y = Real('reserve_y') # 池中token B储备 # 不变量 k = Real('k') invariant = (reserve_x * reserve_y == k) # 前置条件 preconditions = And( x > 0, y > 0, reserve_x > 0, reserve_y > 0, # 满足恒定乘积公式 (reserve_x + x) * (reserve_y - y) == reserve_x * reserve_y, # y不能超过储备量 y < reserve_y ) # Swap后的状态 new_reserve_x = reserve_x + x new_reserve_y = reserve_y - y # 后置条件 postconditions = And( # 储备量应该增加/减少 new_reserve_x == reserve_x + x, new_reserve_y == reserve_y - y, # 仍然满足恒定乘积 new_reserve_x * new_reserve_y == reserve_x * reserve_y, # 储备量非负 new_reserve_y >= 0 ) # 求解器验证 s = Solver() s.add(invariant) s.add(preconditions) s.add(Not(postconditions)) # 如果无解，说明后置条件总是满足 if s.check() == unsat: print("✓ Swap函数是正确的") return True else: print("✗ 发现反例:") model = s.model() print(f" x = {model[x]}") print(f" y = {model[y]}") print(f" reserve_x = {model[reserve_x]}") print(f" reserve_y = {model[reserve_y]}") return False # 验证AMM池 def verify_amm_invariant(): """验证AMM恒定乘积不变式""" # 初始状态 x0 = Real('x0') y0 = Real('y0') k = x0 * y0 # 交易后状态 dx = Real('dx') dy = Real('dy') x1 = x0 + dx y1 = y0 + dy # 验证恒定乘积 s = Solver() # 约束条件 s.add(x0 > 0, y0 > 0) s.add(k == x0 * y0) s.add(x0 * y0 == x1 * y1) # 检查是否可满足 if s.check() == sat: model = s.model() print(f"有效交易: dx = {model[dx]}, dy = {model[dy]}") return True else: print("违反恒定乘积") return False Certora规范 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 // Certora规范语言 // 使用SMT求解器验证智能合约 methods { function swap(uint256 amount0In, uint256 amount1In, address to, bytes calldata data) external; function getReserves() external view returns (uint112 reserve0, uint112 reserve1); } // 确保swap遵循恒定乘积公式 RULE invariant CONSTANT_PRODUCT calldataarg uint256 amount0In; calldataarg uint256 amount1In; address to; bytes data; { env e; require e.msg.value == 0; uint256 reserve0Before; uint256 reserve1Before; reserve0Before, reserve1Before = getReserves(); swap(e, amount0In, amount1In, to, data); uint256 reserve0After; uint256 reserve1After; reserve0After, reserve1After = getReserves(); // 恒定乘积公式：reserve0 * reserve1 应该保持不变 assert(reserve0Before * reserve1Before == reserve0After * reserve1After, "Invariant violated: constant product formula"); } // 确保swap不会导致储备量变为0 RULE invariant NO_ZERO_RESERVES calldataarg uint256 amount0In; calldataarg uint256 amount1In; address to; bytes data; { env e; swap(e, amount0In, amount1In, to, data); uint256 reserve0; uint256 reserve1; reserve0, reserve1 = getReserves(); assert(reserve0 > 0 && reserve1 > 0, "Reserves cannot be zero"); } // 验证转账函数 FUNCTION transfer(address to, uint256 amount) creates evm(uint256 balance) = balanceOf(to), evm(uint256 balance) = balanceOf(msg.sender) updates balanceOf(to) = toBalance => toBalance >= balance, balanceOf(msg.sender) = fromBalance => fromBalance <= balance; // 确保总供应量不变 RULE invariant TOTAL_SUPPLY { env e; uint256 totalBefore = totalSupply(); // 执行任意操作 havoc(e); uint256 totalAfter = totalSupply(); assert(totalBefore == totalAfter, "Total supply changed"); } 审计流程 完整审计清单 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 # 智能合约审计清单 ## 1. 代码质量检查 - [ ] 遵循Solidity最佳实践 - [ ] 使用最新编译器版本 - [ ] 启用优化器 - [ ] 遵循Checks-Effects-Interactions模式 - [ ] 避免使用tx.origin进行身份验证 - [ ] 使用SafeMath（旧版本） - [ ] 正确处理浮点数（使用定点数） ## 2. 访问控制审查 - [ ] 关键函数有适当的访问控制 - [ ] onlyOwner修饰符正确使用 - [ ] 角色权限合理配置 - [ ] 多重签名机制 - [ ] 时间锁机制 ## 3. 状态管理 - [ ] 外部调用在状态更新之后 - [ ] 重入保护 - [ ] 正确的事件记录 - [ ] 临界区管理 ## 4. 数值处理 - [ ] 整数溢出保护 - [ ] 除法检查（除数不为0） - [ ] 舍入误差处理 - [ ] 浮点数使用正确 ## 5. 逻辑漏洞 - [ ] 业务逻辑完整性 - [ ] 边界条件处理 - [ ] 异常情况处理 - [ ] 竞态条件检查 ## 6. DeFi特定检查 - [ ] 滑点保护 - [ ] MEV防护 - [ ] 抢跑保护 - [ ] 清算机制 - [ ] 价格操纵防护 - [ ] oracle使用正确 ## 7. Gas优化 - [ ] 循环优化 - [ ] 存储优化 - [ ] 批量操作 - [ ] 事件记录优化 ## 8. 升级机制 - [ ] 代理模式正确实现 - [ ] 存储布局兼容性 - [ ] 升级流程安全 - [ ] 紧急暂停机制 ## 9. 测试覆盖 - [ ] 单元测试覆盖率 > 90% - [ ] 集成测试完整 - [ ] 模糊测试 - [ ] 形式化验证 ## 10. 文档 - [ ] NatSpec注释完整 - [ ] 架构文档 - [ ] 用户文档 - [ ] API文档 分阶段审计 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 interface AuditPhases { phase1: { name: "初步扫描", duration: "2-3天", tools: ["Slither", "Mythril", "MythX"], output: "漏洞清单" }, phase2: { name: "人工代码审查", duration: "1-2周", methods: ["行内审查", "架构分析", "威胁建模"], output: "审计报告初稿" }, phase3: { name: "测试和验证", duration: "1周", methods: ["单元测试", "集成测试", "形式化验证"], output: "测试报告" }, phase4: { name: "修复验证", duration: "3-5天", process: "修复后重新测试", output: "最终审计报告" } } DeFi安全最佳实践 Oracle使用 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 // ✅ 使用Chainlink Price Feed import "@chainlink/contracts/src/v0.8/interfaces/AggregatorV3Interface.sol"; contract OracleExample { AggregatorV3Interface internal priceFeed; constructor() { // ETH/USD Price Feed priceFeed = AggregatorV3Interface( 0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419 ); } function getLatestPrice() public view returns (int256) { ( uint80 roundId, int256 price, uint256 startedAt, uint256 timeStamp, uint80 answeredInRound ) = priceFeed.latestRoundData(); // 检查价格是否新鲜 require( timeStamp + 3 hours > block.timestamp, "Price too stale" ); // 检查数据是否为空 require(price > 0, "Invalid price"); return price; } function getPrice(uint256 amount) external view returns (uint256) { int256 price = getLatestPrice(); // price有8位小数 return (amount * uint256(price)) / 1e8; } } // ✅ TWAP（时间加权平均价格） import "@uniswap/v3-periphery/contracts/interfaces/ISwapRouter.sol"; contract TWAPExample { struct Observation { uint256 timestamp; uint256 price0Cumulative; uint256 price1Cumulative; } function getTWAP( address pool, uint32 secondsAgo ) external view returns (uint256 price) { // 获取当前观察值 ( uint256 price0Cumulative, uint256 price1Cumulative, uint32 blockTimestamp ) = IUniswapV3Pool(pool).observe(secondsAgo); // 计算TWAP uint256 timeElapsed = blockTimestamp - (blockTimestamp - secondsAgo); require(timeElapsed > 0, "Not enough data"); price = (price0Cumulative - price1Cumulative) / timeElapsed; return price; } } 滑点保护 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 // ✅ 滑点保护实现 import "@uniswap/v2-periphery/contracts/interfaces/IUniswapV2Router02.sol"; contract SlippageProtection { IUniswapV2Router02 public router = IUniswapV2Router02(0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D); function swapWithSlippage( address tokenIn, address tokenOut, uint256 amountIn, uint256 minAmountOut ) external returns (uint256 amountOut) { // 授权 IERC20(tokenIn).approve(address(router), amountIn); // 定义路径 address[] memory path = new address[](2); path[0] = tokenIn; path[1] = tokenOut; // 执行交换 uint256[] memory amounts = router.swapExactTokensForTokens( amountIn, minAmountOut, // 最小输出量（滑点保护） path, block.timestamp ); amountOut = amounts[1]; require( amountOut >= minAmountOut, "Slippage exceeded" ); } function calculateMinAmountOut( uint256 amountIn, uint256 slippageBps ) external pure returns (uint256) { // slippageBps: 基点，100 = 1% uint256 slippage = (amountIn * slippageBps) / 10000; return amountIn - slippage; } } // ✅ 动态滑点 contract DynamicSlippage { uint256 public baseSlippage = 30; // 0.3% uint256 public maxSlippage = 300; // 3% function getDynamicSlippage(uint256 volatility) public view returns (uint256) { // 根据波动率调整滑点 uint256 slippage = baseSlippage + (volatility * 10); // 不超过最大滑点 if (slippage > maxSlippage) { slippage = maxSlippage; } return slippage; } function swapWithDynamicSlippage( uint256 amountIn, uint256 volatility ) external returns (uint256) { uint256 slippageBps = getDynamicSlippage(volatility); uint256 minAmountOut = calculateMinAmountOut(amountIn, slippageBps); // 执行交换... } } 总结 智能合约安全审计是WEB3开发不可或缺的环节。系统性的审计流程、专业的分析工具和严格的安全实践是保护资产安全的关键。 ...

引言 Web3和区块链技术正在重塑互联网的形态。本文将深入探讨智能合约开发、DeFi协议、NFT等核心主题，帮助开发者进入Web3世界。 一、Solidity智能合约 1.1 基础合约结构 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 // SPDX-License-Identifier: MIT pragma solidity ^0.8.20; import "@openzeppelin/contracts/token/ERC20/ERC20.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract MyToken is ERC20, Ownable { uint256 public constant MAX_SUPPLY = 1_000_000_000 * 10**18; constructor() ERC20("MyToken", "MTK") { _mint(msg.sender, MAX_SUPPLY); } function mint(address to, uint256 amount) public onlyOwner { _mint(to, amount); } function burn(uint256 amount) public { _burn(msg.sender, amount); } } 1.2 安全最佳实践 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 // ========== 重入攻击防护 ========== contract ReentrancyGuard { bool private locked; modifier noReentrant() { require(!locked, "Reentrant call"); locked = true; _; locked = false; } function withdraw() external noReentrant { // ... } } // ========== 访问控制 ========== contract AccessControl { mapping(address => bool) public admins; modifier onlyAdmin() { require(admins[msg.sender], "Not admin"); _; } function addAdmin(address admin) external onlyAdmin { admins[admin] = true; } } // ========== 安全数学运算 ========== library SafeMath { function add(uint256 a, uint256 b) internal pure returns (uint256) { require(a + b >= a, "Overflow"); return a + b; } function sub(uint256 a, uint256 b) internal pure returns (uint256) { require(a >= b, "Underflow"); return a - b; } } 二、DeFi协议开发 2.1 AMM交换池 1 2 3 4 5 6 7 8 9 10 11 12 contract AMMPool { uint256 public reserve0; uint256 public reserve1; function addLiquidity(uint256 amount0, uint256 amount1) external { // ... } function swap(uint256 amount0In, uint256 amount1In) external { // ... } } 三、NFT开发 1 2 3 4 5 6 7 8 9 10 11 12 import "@openzeppelin/contracts/token/ERC721/extensions/ERC721URIStorage.sol"; contract MyNFT is ERC721URIStorage { uint256 private _tokenIdCounter; function mint(address to, string memory uri) public returns (uint256) { uint256 tokenId = _tokenIdCounter++; _safeMint(to, tokenId); _setTokenURI(tokenId, uri); return tokenId; } } 四、前端集成 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 import { ethers } from 'ethers'; // 连接钱包 async function connectWallet() { const provider = new ethers.BrowserProvider(window.ethereum); await provider.send("eth_requestAccounts", []); const signer = await provider.getSigner(); return signer; } // 调用合约 async function mintNFT(signer, contractAddress, uri) { const contract = new ethers.Contract( contractAddress, ['function mint(address to, string memory uri) returns (uint256)'], signer ); const tx = await contract.mint(await signer.getAddress(), uri); await tx.wait(); } 总结 Web3开发需要掌握智能合约、区块链原理和前端集成。持续关注安全最佳实践至关重要。 ...