Hyperliquid Copy Trader 架构分析

项目: zhajingwen/Hyperliquid_Copy_Trader（fork 自 MaxIsOntoSomething）
语言: Python 3.12 | 协议: MIT | 部署: Docker / Shell / Batch
分析目的: 提炼可借鉴的设计模式，与当前配对交易项目做对比

1. 项目概览

Hyperliquid DEX 自动化跟单交易机器人，核心功能：

• 同时监控 最多 10 个目标钱包（受 Hyperliquid WebSocket 限制）
• 实时跟随开仓 / 平仓 / 挂单操作
• 智能仓位缩放（按余额比率自动计算）
• Telegram 远程控制与通知
• 支持模拟运行（dry_run）和测试网

2. 目录结构

Hyperliquid_Copy_Trader/
├── .env.example                  # 环境变量模板
├── Dockerfile                    # Python 3.12 容器
├── docker-compose.yml            # 单服务编排
├── requirements.txt              # 依赖清单
├── linux/                        # Linux 部署脚本
├── windows/                      # Windows 部署脚本
└── src/
    ├── main.py                   # 主入口 + 核心编排（~700行）
    ├── config/
    │   └── settings.py           # Pydantic 配置管理
    ├── copy_engine/
    │   ├── monitor.py            # WebSocket 钱包监控
    │   ├── executor.py           # 交易执行（EIP-712 签名）
    │   └── position_sizer.py     # 仓位计算器
    ├── hyperliquid/
    │   ├── client.py             # REST API 客户端（aiohttp）
    │   ├── models.py             # 数据模型（dataclass + enum）
    │   └── websocket.py          # WebSocket 客户端（websockets）
    ├── telegram_bot/
    │   ├── bot.py                # Telegram 命令处理
    │   └── notifications.py      # 通知消息服务
    └── utils/
        └── logger.py             # loguru 日志配置

3. 核心架构分析

3.1 全异步架构（asyncio）

整个系统基于 asyncio 构建，所有 I/O 操作都是异步的：

关键代码模式：

# main.py 启动编排
async def main():
    async with HyperliquidClient() as client:
        # 初始化组件...
        await asyncio.gather(
            monitor.start(),           # WebSocket 监控
            telegram_bot.start(),      # Telegram 命令循环
            hourly_report_task(),      # 定时报告
        )

vs 当前项目（threading）：

借鉴价值: 中等。当前项目有 CPU 密集的统计计算（协整检验、Z-score），纯 asyncio 不一定合适。但 I/O 层（WebSocket、HTTP、数据库写入）可以考虑迁移到 asyncio，计算密集部分用 asyncio.to_thread() 或 ProcessPoolExecutor 桥接。

3.2 Pydantic 配置校验

# src/config/settings.py
class HyperliquidConfig(BaseModel):
    api_url: str = "https://api.hyperliquid.xyz"
    ws_url: str = "wss://api.hyperliquid.xyz/ws"
    testnet_api_url: str = "https://api.hyperliquid-testnet.xyz"
    target_addresses: list[str]  # 必填，逗号分隔解析
    wallet_address: str          # 必填
    private_key: str             # 必填

class SizingConfig(BaseModel):
    portfolio_ratio: float = 0.01
    max_position_size: float = 1000.0
    max_total_exposure: float = 5000.0

class LeverageConfig(BaseModel):
    adjustment_ratio: float = 0.5  # 使用目标杠杆的 50%

class CopyRulesConfig(BaseModel):
    copy_existing_orders: bool = False
    copy_open_positions: bool = False
    max_open_trades: str = "x"     # "x" 表示无限制
    blocked_assets: list[str] = []

class Settings(BaseModel):
    hyperliquid: HyperliquidConfig
    telegram: TelegramConfig
    sizing: SizingConfig
    leverage: LeverageConfig
    copy_rules: CopyRulesConfig

    @classmethod
    def load(cls) -> "Settings":
        """从 .env 加载并校验所有配置"""
        # 启动时即校验，缺少必填项立即报错

# 模块级单例
settings = Settings.load()

vs 当前项目（dataclass + env）：

借鉴价值: 高。启动时配置校验可以避免 "跑了 10 分钟后因为缺配置而崩溃" 的问题。建议在当前项目中引入 Pydantic 或至少在启动时做一次完整的配置检查。

3.3 Loguru 结构化日志

# src/utils/logger.py
from loguru import logger
import sys

logger.remove()  # 移除默认 handler
logger.add(
    sys.stderr,
    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
           "<level>{level: <8}</level> | "
           "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | "
           "<level>{message}</level>",
    level="DEBUG",
)
logger.add(
    "logs/copy_trader.log",
    rotation="10 MB",
    retention="7 days",
    compression="zip",
    level="INFO",
)

vs 当前项目（标准 logging）：

借鉴价值: 中等。Loguru 确实更易用，但当前项目已有完善的日志体系（含第三方库抑制、双级别输出等），迁移收益不大。如果从零开始则推荐 Loguru。

3.4 Telegram 远程控制

# src/telegram_bot/bot.py
class TelegramBot:
    def __init__(self, token, chat_id, get_status_fn, get_positions_fn, ...):
        self.app = ApplicationBuilder().token(token).build()
        # 命令注册
        self.app.add_handler(CommandHandler("status", self.status_command))
        self.app.add_handler(CommandHandler("positions", self.positions_command))
        self.app.add_handler(CommandHandler("pause", self.pause_command))
        self.app.add_handler(CommandHandler("resume", self.resume_command))
        self.app.add_handler(CommandHandler("stop", self.stop_command))
        # 回调按钮（停止时的确认：平仓 / 保留 / 取消）
        self.app.add_handler(CallbackQueryHandler(self.stop_callback))

    async def _is_authorized(self, update):
        """仅允许配置的 chat_id 操作"""
        return str(update.effective_chat.id) == self.chat_id

支持的命令集：

vs 当前项目（飞书单向通知）：

借鉴价值: 高。远程控制能力是生产系统的关键需求。建议：

1. 飞书也支持机器人命令，可以实现类似的双向交互
2. 或者增加一个轻量级 HTTP API（FastAPI）用于远程控制
3. 至少实现：状态查询、暂停/恢复、紧急停止

3.5 模拟交易模式（Paper Trading / Dry Run）

# src/copy_engine/executor.py
class TradeExecutor:
    def __init__(self, ..., dry_run: bool = False):
        self.dry_run = dry_run
        self._dry_run_counter = 0

    async def execute_market_order(self, symbol, side, size, ...):
        if self.dry_run:
            self._dry_run_counter += 1
            order_id = f"DRY-{self._dry_run_counter}"
            logger.info(f"[DRY RUN] Market {side} {size} {symbol} @ ~{price}")
            return {"status": "ok", "order_id": order_id, "dry_run": True}
        # 实际执行...

    async def close_position(self, symbol, ...):
        if self.dry_run:
            logger.info(f"[DRY RUN] Close position {symbol}")
            return {"status": "ok", "dry_run": True}
        # 实际执行...

同时在 main.py 中，dry_run 模式下也会追踪模拟盈亏：

# main.py on_position_close 回调
async def on_position_close(target_addr, symbol, last_position):
    if executor.dry_run:
        # 计算模拟盈亏
        pnl = (close_price - entry_price) * size * direction_multiplier
        tracked_pnl += pnl
        logger.info(f"[DRY RUN] Simulated PnL: ${pnl:.2f}")

vs 当前项目: 当前项目 没有模拟交易模式，所有策略测试必须实盘。

借鉴价值: 高。模拟模式的关键价值：

1. 策略验证: 不花真金白银验证信号质量
2. 系统测试: 验证全链路（信号 → 下单 → 持仓管理）不出 bug
3. 新手友好: 降低使用门槛
4. 实现成本低: 在 executor 层加 if/else 即可，不影响上层逻辑

3.6 闭包工厂模式隔离多钱包事件

# main.py — 为每个目标钱包创建独立的回调函数
def make_callbacks(target_address):
    """闭包工厂：为每个目标钱包生成隔离的事件回调"""

    async def on_position_close(symbol, last_position):
        # target_address 被闭包捕获
        logger.info(f"[{target_address[:8]}] Position closed: {symbol}")
        await executor.close_position(symbol, ...)

    async def on_order_fill(symbol, fill_data):
        ratio = target_ratios[target_address]
        # 使用该目标的专属比率计算仓位
        size = fill_data.size * ratio
        await executor.execute_market_order(symbol, side, size, ...)

    return on_position_close, on_order_fill, ...

# 启动时为每个目标注册独立回调
for addr in settings.hyperliquid.target_addresses:
    callbacks = make_callbacks(addr)
    monitor.set_callbacks(addr, *callbacks)

设计亮点：

• 每个目标钱包的事件互不干扰
• 闭包捕获 target_address 和对应的 ratio，无需全局映射查找
• 回调签名统一，WalletMonitor 不需要知道多钱包的存在

vs 当前项目: 当前项目是单一交易对分析，不存在多钱包路由问题。但如果未来扩展到多策略/多交易对，这种模式值得借鉴。

3.7 资产级别杠杆上限

# main.py
def calculate_adjusted_leverage(symbol: str, target_leverage: int) -> int:
    """按资产类别设置杠杆上限"""
    MAX_LEVERAGE = {
        "BTC": 50, "ETH": 50,
        "SOL": 20, "MATIC": 20, "ARB": 20, "AVAX": 20,
        # ... 更多资产
    }
    default_max = 10  # 其他资产默认上限

    asset = symbol.replace("-USD", "").replace("PERP", "")
    max_lev = MAX_LEVERAGE.get(asset, default_max)

    # 应用调整比率（如 0.5 = 使用目标杠杆的 50%）
    adjusted = int(target_leverage * settings.leverage.adjustment_ratio)
    return min(adjusted, max_lev)

设计思路：

• 主流资产（BTC/ETH）流动性好 → 允许高杠杆
• 中等流动性资产 → 20x 上限
• 未知/小市值资产 → 10x 上限（保守）
• 额外的全局调整比率（如 0.5）进一步降低实际使用杠杆

借鉴价值: 中等。当前项目的杠杆管理是固定值，如果未来支持多币种，按流动性分级管理杠杆是合理的风控措施。

3.8 EIP-712 签名（原生实现）

# src/copy_engine/executor.py
from eth_account import Account
from eth_account.messages import encode_typed_data

class TradeExecutor:
    DOMAIN_MAINNET = {"name": "Exchange", "version": "1", "chainId": 1337, ...}
    DOMAIN_TESTNET = {"name": "Exchange", "version": "1", "chainId": 421614, ...}

    def _sign_action(self, action: dict, nonce: int) -> dict:
        """EIP-712 结构化数据签名"""
        typed_data = {
            "domain": self.domain,
            "types": {
                "EIP712Domain": [...],
                "HyperliquidTransaction:Order": [...]
            },
            "primaryType": "HyperliquidTransaction:Order",
            "message": {
                "action": action,
                "nonce": nonce,
            }
        }
        signed = Account.sign_typed_data(
            self.private_key,
            full_message=typed_data
        )
        return {"r": hex(signed.r), "s": hex(signed.s), "v": signed.v}

vs 使用官方 SDK：

借鉴价值: 低-中。当前项目已使用官方 SDK，除非遇到 SDK 的具体限制，否则没必要切换。但了解底层签名机制有助于排查问题。

4. 数据流架构

                     ┌─────────────────────────────────────────┐
                     │           Hyperliquid DEX                │
                     │  ┌────────────┐    ┌─────────────────┐  │
                     │  │ REST API   │    │ WebSocket        │  │
                     │  │ /info      │    │ userEvents 订阅  │  │
                     │  └─────┬──────┘    └────────┬────────┘  │
                     └────────┼────────────────────┼───────────┘
                              │                    │
                              ▼                    ▼
                     ┌────────────────┐  ┌──────────────────────┐
                     │ Hyperliquid    │  │ HyperliquidWebSocket  │
                     │ Client (REST)  │  │ - subscribe_user()    │
                     │ - user_state   │  │ - auto reconnect      │
                     │ - async ctx    │  │ - callback dispatch    │
                     └────────────────┘  └──────────┬───────────┘
                                                    │
                              ┌──────────────────────┘
                              ▼
                     ┌──────────────────────────────────────┐
                     │ WalletMonitor                         │
                     │ - 管理多目标钱包状态                    │
                     │ - 分发 3 类事件:                       │
                     │   fills → on_order_fill               │
                     │   positions → on_position_close/update│
                     │   orders → on_new_order               │
                     │ - 资产黑名单过滤                       │
                     └──────────────────┬───────────────────┘
                                        │ 闭包回调
                     ┌──────────────────┘
                     ▼
         ┌──────────────────────────────────────────────┐
         │ main.py 核心编排                               │
         │                                              │
         │  on_order_fill:                              │
         │    检测开仓/加仓/平仓/方向翻转                  │
         │    → PositionSizer.calculate() 计算仓位       │
         │    → calculate_adjusted_leverage() 调整杠杆   │
         │    → TradeExecutor.execute_*() 执行交易       │
         │                                              │
         │  on_position_close:                          │
         │    → TradeExecutor.close_position()           │
         │    → 计算模拟盈亏（dry_run 模式）              │
         │                                              │
         │  on_new_order:                               │
         │    → TradeExecutor.execute_limit_order()      │
         └──────────┬──────────────────┬────────────────┘
                    │                  │
                    ▼                  ▼
          ┌─────────────────┐ ┌──────────────────────┐
          │ TradeExecutor   │ │ TelegramBot          │
          │ - EIP-712 签名   │ │ - 命令: /status ...  │
          │ - 市价/限价单    │ │ - 控制: /pause ...   │
          │ - 杠杆更新      │ │ - NotificationService│
          │ - dry_run 模式  │ │   - 交易通知          │
          │                 │ │   - 小时报告          │
          └─────────────────┘ └──────────────────────┘

5. 全局状态管理

项目使用模块级全局变量管理运行时状态：

# main.py 全局变量
is_paused = False               # 暂停标志（Telegram 控制）
trades_copied_count = 0         # 已跟单计数
account_balance = 0.0           # 账户余额
tracked_positions = {}          # "target:symbol" → 持仓详情
tracked_pnl = 0.0              # 累计盈亏（含模拟）
target_ratios = {}             # addr → balance_ratio

问题：

• 全局可变状态，难以测试
• main.py 承担了过多职责（700+ 行）
• 无持久化，重启后状态丢失

6. 依赖分析

7. 与当前项目的对比矩阵

7.1 架构维度对比

7.2 当前项目独有优势（不需要改动）

1. TimescaleDB 持久化 + 崩溃恢复 — 重启后可以恢复运行，Copy Trader 完全无状态
2. 双腿配对交易 — 统计套利模型，买目标 + 卖基准的对冲结构
3. 三层安全防护 — KillSwitch（文件触发紧急停止）+ RateLimiter（滑动窗口限流）+ CircuitBreaker（熔断器）
4. 多周期协整分析 — 5m/1h/4h 三周期 Z-score + 双重确认机制（DoubleCheckState）
5. 日损限额 + 最大回撤监控 — 企业级风控，Copy Trader 只有仓位上限
6. 模板方法模式 — RealtimeKlineServiceBase 实现 90% 共享逻辑，子类只需 4 个抽象方法
7. 连接池管理 — TimescaleDB 连接池（5-60），Copy Trader 无数据库
8. 批量写入优化 — COPY 方法 >40K records/sec

8. 可借鉴的改进建议

按优先级排序：

P0 - 高优先级（低成本高收益）

8.1 模拟交易模式

现状: 无模拟模式，策略测试只能实盘
改进: 在 executor 层加 dry_run 标志

# 实现方案
class HyperliquidExecutor:
    def __init__(self, ..., dry_run: bool = False):
        self.dry_run = dry_run
        self._paper_positions = {}  # 模拟持仓追踪

    async def market_open(self, ...):
        if self.dry_run:
            self._paper_positions[symbol] = {
                "entry": price, "size": size, "side": side
            }
            logger.info(f"[PAPER] Open {side} {size} {symbol} @ {price}")
            return PaperOrderResult(...)
        # 实际执行...

工作量: ~2-3 小时 | 风险: 低 | 收益: 大幅降低测试成本

8.2 启动配置校验

现状: 运行时才发现配置缺失
改进: 启动时统一校验所有必需配置

# 方案 A: Pydantic（推荐，如果愿意加依赖）
class AppConfig(BaseModel):
    timescaledb_host: str
    timescaledb_port: int = 5432
    larkbot_id: str
    # ... 缺少必填项直接 ValidationError

# 方案 B: 手动校验（零依赖）
def validate_config():
    required = ["TIMESCALEDB_HOST", "LARKBOT_ID"]
    missing = [k for k in required if not os.getenv(k)]
    if missing:
        raise SystemExit(f"Missing required config: {missing}")

工作量: ~1-2 小时 | 风险: 低 | 收益: 避免运行时配置错误

P1 - 中优先级

8.3 双向通知（飞书交互式消息）

现状: 飞书只发不收
改进: 利用飞书机器人的消息卡片 + 回调 URL 实现远程控制

需要的命令集：

• 状态查询（当前持仓、队列健康度、盈亏）
• 暂停/恢复分析
• 紧急停止

工作量: ~1-2 天 | 风险: 中（需要飞书应用配置） | 收益: 远程运维能力

8.4 资产分级杠杆管理

现状: 杠杆固定
改进: 按资产流动性设置杠杆上限

LEVERAGE_TIERS = {
    "tier1": {"assets": ["BTC", "ETH"], "max": 50},
    "tier2": {"assets": ["SOL", "ARB", "AVAX"], "max": 20},
    "default": {"max": 10},
}

工作量: ~2 小时 | 风险: 低 | 收益: 更精细的风控

P2 - 低优先级（长期改进）

8.5 I/O 层 asyncio 迁移

将 WebSocket、HTTP、数据库写入迁移到 asyncio，CPU 密集计算保留线程池：

asyncio event loop
├── WebSocket 连接（asyncio）
├── 数据库写入（asyncio + asyncpg）
├── 飞书通知（asyncio + aiohttp）
└── 计算任务（asyncio.to_thread → ThreadPoolExecutor）

工作量: ~1-2 周 | 风险: 高（架构重构） | 收益: 资源利用率提升、代码简化

9. 架构问题警示（Copy Trader 的反面教材）

以下是 Copy Trader 的设计缺陷，当前项目应避免这些问题：

1. main.py 膨胀（700+ 行）— 业务逻辑、状态管理、回调全堆一个文件。当前项目的模板方法模式更优
2. 全局可变状态 — is_paused, tracked_positions 等全局变量，难测试。当前项目用类封装状态更好
3. 无持久化 — 重启丢失所有状态和统计。当前项目的 TimescaleDB 方案是正确选择
4. 无测试代码 — 声明了 pytest 依赖但 0 测试文件
5. on_position_update 未实现 — 只有 TODO 注释，加仓/减仓不会被跟随
6. 冗余依赖 — sqlalchemy、alembic、requests 声明了但未使用
7. 平仓逻辑有 bug — 未提供数量时使用硬编码 0.001 的 reduce_only 单，可能无法完全平仓
8. 无错误恢复 — WebSocket 断线重连后不恢复错过的事件

10. 总结

Copy Trader 的核心价值

行动清单

• [ ] P0 实现模拟交易模式（executor 层 dry_run）
• [ ] P0 添加启动配置校验（Pydantic 或手动检查）
• [ ] P1 飞书双向交互（消息卡片 + 远程控制）
• [ ] P1 资产分级杠杆管理
• [ ] P2 评估 I/O 层 asyncio 迁移可行性