這是提交給 赫爾墨斯代理挑戰(Hermes Agent Challenge): 使用赫爾墨斯代理建構(Build With Hermes Agent)
我構建的內容
IYOP Trading Bot 是一款由人工智慧驅動的多鏈 DEX 交易機器人,可在 Solana(Jupiter)、Ethereum(Uniswap V3)及 BSC(PancakeSwap V2)上執行交易。它採用 MiMo v2 進行市場分析,從多種技術指標產生訊號,並透過部位規模設定與回撤保護來管理風險。
我正在解決的問題是:去中心化交易碎片化。每個鏈都有自己的 DEX、自己的特性、自己的流動性池。手動監控它們是不可能的。跨鏈執行交易需要不同的 SDK、不同的 Gas 管理、不同的錢包處理。
我需要一個統一的層,能夠:
- 同時掃描多個 DEX
- 從技術指標和AI分析生成訊號
- 以適當的風險管理執行交易
- 跨鏈追蹤持倉
- 自動處理止損和止盈
而且我需要Hermes Agent(Hermes Agent)來讓開發過程足夠快速以實際發佈。
示範
即時示範: iyop-trading-terminal.vercel.app
來源: github.com/iyop666/iyop-trading-bot
該機器人以 FastAPI 後端搭配網頁儀表板運行。主要功能:
- 多鏈 DEX 掃描(Jupiter、Uniswap V3、PancakeSwap)
- 透過 IYOP v2 API 的 AI 驅動市場分析
- 帶有可信度評分的信號生成
- 使用倉位規模和回撤限制的風險管理
- 透過停損/停利執行交易
- 即時 WebSocket 價格推送
程式碼
以下是 Hermes Agent 協助建構的核心模組:
AI 市場分析器
該分析器連接到 IYOP v2,以進行快速市場分析、訊號產生和情緒分析:
import httpx
from dataclasses import dataclass
from enum import Enum
MIMO_API_URL = "http://127.0.0.1:19911/v1/chat/completions"
MIMO_MODEL = "gitlawb/mimo-v2-flash"
class Signal(Enum):
BUY = "buy"
SELL = "sell"
HOLD = "hold"
@dataclass
class MarketAnalysis:
signal: Signal
confidence: float
reasoning: str
risk_level: str
async def analyze_market(pair: str, timeframe: str) -> MarketAnalysis:
"""Send market data to MiMo v2 for analysis."""
async with httpx.AsyncClient() as client:
response = await client.post(
MIMO_API_URL,
json={
"model": MIMO_MODEL,
"messages": [
{"role": "system", "content": "Analyze this trading pair..."},
{"role": "user", "content": f"Pair: {pair}, Timeframe: {timeframe}"}
],
"temperature": 0.1
},
timeout=30
)
result = response.json()
return parse_analysis(result)
訊號生成
結合多種技術指標與加權評分:
from dataclasses import dataclass
from enum import Enum
class SignalType(Enum):
STRONG_BUY = "strong_buy"
BUY = "buy"
WEAK_BUY = "weak_buy"
HOLD = "hold"
WEAK_SELL = "weak_sell"
SELL = "sell"
STRONG_SELL = "strong_sell"
@dataclass
class TradeSignal:
signal_type: SignalType
confidence: float # 0.0 to 1.0
pair: str
price: float
indicators: dict
reasoning: str
def generate_signal(indicators: dict, ai_analysis: dict) -> TradeSignal:
"""Combine technical indicators with AI analysis."""
weights = {
"rsi": 0.25,
"macd": 0.20,
"bollinger": 0.15,
"volume": 0.15,
"ai_analysis": 0.25
}
score = 0.0
for indicator, weight in weights.items():
score += normalize(indicators.get(indicator, 0)) * weight
signal_type = classify_signal(score)
return TradeSignal(
signal_type=signal_type,
confidence=abs(score),
pair=indicators["pair"],
price=indicators["price"],
indicators=indicators,
reasoning=build_reasoning(indicators, ai_analysis)
)
多鏈DEX整合
統一的 Jupiter、Uniswap V3 和 PancakeSwap 介面:
from enum import Enum
import httpx
class DEX(Enum):
JUPITER = "jupiter" # Solana
UNISWAP = "uniswap" # Ethereum
PANCAKESWAP = "pancakeswap" # BSC
class Chain(Enum):
SOLANA = "solana"
ETHEREUM = "ethereum"
BSC = "bsc"
@dataclass
class SwapResult:
success: bool
tx_hash: str
input_amount: float
output_amount: float
chain: Chain
dex: DEX
gas_used: float
async def execute_swap(
dex: DEX,
chain: Chain,
token_in: str,
token_out: str,
amount: float,
slippage: float = 0.5
) -> SwapResult:
"""Execute a swap on the specified DEX."""
if dex == DEX.JUPITER:
return await jupiter_swap(token_in, token_out, amount, slippage)
elif dex == DEX.UNISWAP:
return await uniswap_swap(token_in, token_out, amount, slippage)
elif dex == DEX.PANCAKESWAP:
return await pancakeswap_swap(token_in, token_out, amount, slippage)
風險管理
倉位規模與回撤保護:
@dataclass
class RiskManager:
max_position_pct: float = 0.1 # 10% max per position
max_daily_loss: float = 0.05 # 5% max daily loss
max_drawdown: float = 0.15 # 15% max drawdown
current_drawdown: float = 0.0
daily_pnl: float = 0.0
def can_open_position(self, portfolio_value: float, position_size: float) -> bool:
"""Check if a new position is within risk limits."""
# Check position size limit
if position_size / portfolio_value > self.max_position_pct:
return False
# Check daily loss limit
if self.daily_pnl < -(portfolio_value * self.max_daily_loss):
return False
# Check drawdown limit
if self.current_drawdown > self.max_drawdown:
return False
return True
def calculate_position_size(
self,
portfolio_value: float,
entry_price: float,
stop_loss: float
) -> float:
"""Calculate optimal position size based on risk parameters."""
risk_per_trade = portfolio_value * self.max_position_pct
price_risk = abs(entry_price - stop_loss) / entry_price
return risk_per_trade / price_risk if price_risk > 0 else 0
交易執行引擎
處理包含止損和止盈的訂單生命週期:
class Side(str, Enum):
BUY = "buy"
SELL = "sell"
class OrderStatus(str, Enum):
PENDING = "pending"
FILLED = "filled"
CANCELLED = "cancelled"
SL_HIT = "stop_loss_hit"
TP_HIT = "take_profit_hit"
@dataclass
class Position:
id: str
pair: str
side: Side
entry_price: float
current_price: float
size: float
stop_loss: float
take_profit: float
pnl: float
chain: Chain
dex: DEX
class Trader:
def __init__(self, risk_manager: RiskManager):
self.positions: dict[str, Position] = {}
self.risk_manager = risk_manager
self.trade_history: list[dict] = []
async def open_position(
self,
pair: str,
side: Side,
size: float,
stop_loss: float,
take_profit: float,
chain: Chain,
dex: DEX
) -> Position:
"""Open a new position with risk checks."""
if not self.risk_manager.can_open_position(self.portfolio_value, size):
raise RiskLimitExceeded("Position exceeds risk limits")
# Execute the swap
result = await execute_swap(dex, chain, pair, size)
position = Position(
id=str(uuid.uuid4()),
pair=pair,
side=side,
entry_price=result.price,
current_price=result.price,
size=size,
stop_loss=stop_loss,
take_profit=take_profit,
pnl=0.0,
chain=chain,
dex=dex
)
self.positions[position.id] = position
return position
我的技術棧
- 後端: Python, FastAPI, Pydantic
- AI引擎: MiMo v2(本地API)
- DEX整合: Jupiter(Solana)、Uniswap V3(Ethereum)、PancakeSwap V2(BSC)
- 資料: WebSocket價格饋送、httpx用於非同步HTTP
- 前端: HTML 儀表板
- AI 助手(AI Assistant): 赫耳墨斯代理(Hermes Agent)(用於開發、偵錯與部署)
我是如何使用赫耳墨斯代理(Hermes Agent)的
1. 架構設計
當我開始建立交易機器人時,我向赫耳墨斯代理(Hermes Agent)描述了需求:
"我需要一個支援多鏈的去中心化交易所交易機器人,具備AI分析、訊號生成、風險管理及交易執行功能。它應支援Jupiter、Uniswap V3及PancakeSwap。"
Hermes設計了模組結構:
-
backend/ai/用於AI分析與訊號生成 -
backend/core/用於去中心化交易所整合、風險管理與交易執行 - 分析與執行的關注點分離
那種架構後來為我省下了數小時的重構工作。
2. DEX 整合程式碼
每個去中心化交易所都有不同的 API、不同的交易格式、不同的 Gas 估算。我沒有逐一閱讀說明文件,而是告訴赫耳墨斯(Hermes):
「為 Jupiter、Uniswap V3 和 PancakeSwap 編寫非同步交換函式。使用 httpx,處理錯誤,回傳交易雜湊值。」
Hermes(Hermes) 為所有三個去中心化交易所(DEXs)生成了整合程式碼。我審查了錯誤處理機制,調整了滑點參數,並進行了測試。原本需要花費一整天閱讀文件和編寫樣板程式碼的工作,僅用30分鐘就完成了。
3. 風險管理邏輯
風險管理員需要處理倉位規模、回撤限制和每日虧損上限。我描述了規則:
「每檔部位上限10%,每日最大虧損5%,最大回撤15%。依據止損距離計算部位規模。」
Hermes 在實作邏輯時考慮了適當的邊界情況。當我以極端值(100%虧損情境)測試時,系統正確地拒絕了這些交易。
4. 訊號權重系統
要將多個指標結合成單一訊號,需要一套權重系統。我問 Hermes:
「建立一個訊號產生器,對RSI(25%)、MACD(20%)、布林帶(15%)、成交量(15%)和AI分析(25%)進行加權。將每個標準化到-1到+1的範圍。」
赫爾墨斯(Hermes)構建了具有可配置權重的加權評分系統。測試結果顯示AI分析最具預測性後,我後來將其權重調整為30%。
當機器人沒有正確執行交易時,我請 Hermes 追蹤問題:
「交易沒有執行。檢查 swap 函式、風險管理器及交易執行器,找出瓶頸。」
Hermes 閱讀了程式碼,發現風險管理器之所以拒絕交易,是因為投資組合價值在每次交易後未更新。一行修復即可解決,但手動查找這個問題需要花費一小時除錯。
為何 Hermes Agent 至關重要
建立交易機器人需要:
- 快速迭代(市場在變動,程式碼必須跟上)
- 跨模組感知(風險管理員影響交易者,交易者影響投資組合)
- 非同步程式設計模式(所有 DEX 呼叫都是非同步的)
- 錯誤處理邊界情況(網路失敗、燃料不足、滑點)
Hermes 處理了所有這些,因為它在整個程式碼庫中保持了上下文。當我更換風險管理員時,它自動建議更新給交易者。當我新增一個新的 DEX 時,它更新了訊號產生器以包含新的鏈。
這種跨模組的感知能力就是原型與可運作機器人之間的差異。
專案統計:
- 約 2,000 行 Python 程式碼。
- 3 種 DEX 整合(Jupiter、Uniswap V3、PancakeSwap)
- 5 種交易策略(動能、均值回歸、AI 驅動、剝頭皮、混合)
- 具備 3 層安全層的風險管理
- 即時 WebSocket 價格推送
- 透過 MiMo v2 進行的 AI 驅動市場分析
連結:












