Hyperliquid 配对交易自动执行模块 — 详细设计与实施方案

Context

当前项目是一个 Hyperliquid 配对交易信号发现系统，已实现完整的实时数据采集、多周期统计分析（5m/1h/4h）、双重确认信号验证和飞书告警功能。但系统目前仅能发现信号并发送告警，缺失从信号到实际交易执行的闭环。

本方案设计一个自动交易模块，以最小侵入方式集成到现有系统中，实现信号发现 → 风控审查 → 自动下单 → 仓位管理 → 止损/平仓的完整交易闭环。

一、模块架构

1.1 设计原则

• 隔离性：交易模块作为独立层，任何交易故障不影响信号分析系统
• 可插拔性：通过钩子方法集成，基类仅新增 _on_entry_signal() 和 _on_exit_signal() 两个可选钩子
• 双网络模式：主网（mainnet，真实资金）和测试网（testnet，测试资金），两种模式都真实下单，区别仅在连接的 API 地址
• 范围：首期仅支持 HYPE/PURR 配对（对应 realtime_kline_service_hype.py）
• 渐进式策略：先实现单边模式验证系统稳定性，确认无 bug 后再增加两腿配对模式，通过 TRADING_PAIR_MODE 参数随时切换
  • single（默认）— 仅交易目标币种（如 PURR），不做对冲腿
  • pair — 目标币种 + 基准币种对冲（如 买PURR + 卖HYPE）

1.2 新增文件结构

src/
  trading/                              # 交易模块
    __init__.py                        # 统一入口 re-export
    orchestrator.py                    # TradingOrchestrator（对外统一入口）
    strategy.py                         # AdaptiveBollingerStrategy（Plan A 策略引擎）
    executor.py                         # HyperliquidExecutor（SDK 交互封装）
    position_manager.py                 # PositionManager（仓位生命周期管理）
    risk_manager.py                    # RiskManager（风控审查 + 仓位计算）
    trade_repository.py                # 交易数据持久化
    models.py                           # 数据模型（Position, Trade, OrderResult 等）
    config.py                           # TradingConfig（交易配置加载）
    protocols.py                        # Protocol 接口（Executor, TradeRepositoryProtocol）
    safety.py                           # 安全机制（KillSwitch, RateLimiter, CircuitBreaker）
database/
  init_timescaledb.sql                  # 含交易相关表结构

1.3 架构总览

┌──────────────────────────────────────────────────────────────┐
│          RealtimeKlineServiceBase (既有，最小修改)              │
│                                                               │
│  _trigger_strategy_if_ready() ──→ process_analysis() ──┐     │
│    (每次分析完成后调用，策略引擎判定入场/退场)           │     │
│  (try/except 全包裹，异常只记日志，不影响信号)           │     │
└────────────────────────────────────────────────────┼──────────┘
                                                     ▼
┌──────────────────────────────────────────────────────────────┐
│                   TradingOrchestrator                          │
│                                                               │
│  ┌─────────────────────┐                                     │
│  │AdaptiveBollingerStrategy│ Plan A: 策略驱动入场/退场        │
│  └──────────┬──────────┘                                     │
│  ┌──────────────┐  ┌────────────────┐  ┌──────────────────┐  │
│  │ RiskManager   │  │PositionManager │  │HyperliquidExecutor│  │
│  │- 开仓前审查   │  │- 内存仓位状态  │  │- Exchange SDK     │  │
│  │- 仓位大小计算 │  │- DB 持久化     │  │- market_open()    │  │
│  │- 止损检查     │  │- 交易所同步    │  │- market_close()   │  │
│  │- 回撤检查     │  │- 重启恢复      │  │- update_leverage()│  │
│  └──────────────┘  └────────────────┘  └──────────────────┘  │
│                                                               │
│  ┌──────────────┐  ┌────────────────┐  ┌──────────────────┐  │
│  │ SafetyLayer   │  │TradeRepository │  │  飞书告警通知     │  │
│  │- Kill Switch  │  │- signals 表    │  │  (复用既有)       │  │
│  │- Rate Limiter │  │- positions 表  │  │                   │  │
│  │- 熔断器       │  │- orders 表     │  │                   │  │
│  └──────────────┘  └────────────────┘  └──────────────────┘  │
└──────────────────────────────────────────────────────────────┘

二、核心类设计

2.1 数据模型 (src/trading/models.py)

class TradingNetwork(Enum):
    MAINNET = "mainnet"       # 主网：真实资金交易
    TESTNET = "testnet"       # 测试网：测试资金交易

# API 地址映射
NETWORK_URLS = {
    TradingNetwork.MAINNET: "https://api.hyperliquid.xyz",
    TradingNetwork.TESTNET: "https://api.hyperliquid-testnet.xyz",
}

class PositionStatus(Enum):
    PENDING / OPENING / OPEN / CLOSING / CLOSED / ERROR

class SignalType(Enum):
    ENTRY / EXIT

@dataclass
class PairTradeSignal:
    """从信号系统传入的交易信号"""
    signal_id: str              # UUID
    signal_type: SignalType
    symbol: str                 # "PURR/USDC:USDC"
    base_symbol: str            # "HYPE/USDC:USDC"
    direction: str              # 'long' / 'short'
    zscore_5m / zscore_1h / zscore_4h: float
    avg_zscore_4h: Optional[float]
    cointegration_count: int
    signal_strength: str
    latest_alt_price: Optional[float]
    timestamp: datetime
    raw_data: Optional[Dict]    # multi_period_result 原始数据

@dataclass
class OrderResult:
    """单笔订单执行结果"""
    success: bool
    order_id: Optional[int]     # Hyperliquid oid
    coin / side / size / price / status / error_message
    raw_response: Optional[Dict]

@dataclass
class PairOrderResult:
    """配对订单结果（两腿）"""
    signal_id: str
    leg_a: OrderResult          # 目标币种
    leg_b: OrderResult          # 基准币种
    is_dry_run: bool

@dataclass
class PairPosition:
    """配对仓位状态"""
    position_id: str            # UUID
    symbol / base_symbol / direction / status
    alt_side / alt_size / alt_entry_price / alt_current_price
    base_side / base_size / base_entry_price / base_current_price
    entry_zscore_4h / entry_avg_zscore_4h / entry_signal_strength
    open_time / close_time
    unrealized_pnl / realized_pnl
    entry_signal_id / exit_signal_id

2.2 交易执行器 (src/trading/executor.py)

class HyperliquidExecutor:
    """封装 Hyperliquid SDK Exchange 类"""

    def initialize(self) -> bool
        # eth_account.Account.from_key(private_key)
        # 根据 TradingNetwork 选择 API URL：
        #   MAINNET → "https://api.hyperliquid.xyz"
        #   TESTNET → "https://api.hyperliquid-testnet.xyz"
        # Exchange(wallet, base_url) + Info(base_url)

    def market_open(signal, alt_size, base_size=None) -> TradeResult
        # single 模式: 只执行目标币种
        # pair 模式: 执行两腿，一腿失败则回滚

    def market_close(position) -> TradeResult
        # single 模式: 只平目标币种
        # pair 模式: 平两腿

    def get_account_state() -> Dict
    def get_positions() -> List[Dict]
    def get_mid_price(coin) -> float

下单方向映射：

single 模式（首期）：仅执行目标币种一笔订单，逻辑简单，用于验证系统稳定性。
pair 模式（后续）：在 single 基础上增加基准币种对冲腿，需处理一腿失败回滚。

币种名格式转换："PURR/USDC:USDC" → "PURR" (SDK 格式)

2.3 仓位管理器 (src/trading/position_manager.py)

class PositionManager:
    """配对仓位生命周期管理"""

    def open_position(signal) -> Optional[PairPosition]
        # 防重复 → 计算仓位 → 执行开仓 → 持久化

    def close_position(symbol, signal) -> Optional[PairPosition]
        # 查找仓位 → 执行平仓 → 计算盈亏 → 持久化

    def sync_with_exchange()
        # 定期与交易所仓位对比，处理不一致

    def _recover_positions_from_db()
        # 重启恢复未关闭仓位

2.4 风险管理器 (src/trading/risk_manager.py)

class RiskManager:
    """风控审查 + 仓位计算"""

    def pre_trade_check(signal) -> Tuple[bool, str]
        # 9 项检查：总开关、kill switch、熔断器、重复仓位、
        # 最大对数、总敞口、每日亏损、最大回撤、余额

    def calculate_position_size(signal) -> Tuple[float, float]
        # base_usd × 信号强度缩放 → 币种数量
        # extreme: ×1.5, strong: ×1.0, medium: ×0.7

    def check_stop_loss(position) -> bool
        # 配对 PnL < -stop_loss_pct → True

    def check_max_drawdown() -> bool
    def update_daily_pnl(realized_pnl)
    def reset_daily_stats()

2.5 安全机制 (src/trading/safety.py)

class KillSwitch:
    """文件系统 Kill Switch — touch 文件即停止交易"""

class RateLimiter:
    """滑动窗口下单频率限制"""

class CircuitBreaker:
    """连续失败熔断器 — 冷却后自动恢复"""

2.6 交易编排器 (src/trading/orchestrator.py)

class TradingOrchestrator:
    """对外统一接口"""

    def start()   # 初始化 executor → 恢复仓位 → 启动监控线程 → 初始化策略引擎
    def stop()    # 优雅关闭

    def process_analysis(symbol, z4h, multi_period_result, timestamp, ...)
        # 策略驱动入口：strategy.process_tick() → entry/exit 信号 → on_entry/on_exit

    def on_entry_signal(symbol, multi_period_result, ..., direction, signal_strength, adaptive_z)
        # 构造信号 → safety → risk → open_position → 通知

    def on_exit_signal(reversion_info)
        # 构造信号 → close_position → 更新盈亏 → 通知

    def _stop_loss_monitor()   # 后台线程，定期扫描止损
    def _position_sync()       # 后台线程，定期同步仓位

2.7 策略引擎 (src/trading/strategy.py)

class AdaptiveBollingerStrategy:
    """Plan A: Adaptive Bollinger Z-Score 策略引擎"""

    # 核心逻辑
    # - z4h 滚动缓冲区 (144 个 5min bar)
    # - adaptive_z = (z4h - ema) / rolling_std
    # - 入场: 首次穿越 abs(adaptive_z) >= threshold
    # - 退场: adaptive_z 回归至 entry_adaptive_z × reversion_factor

    def process_tick(symbol, z4h, timestamp) -> Tuple[EntrySignal | None, ExitSignal | None]
    def prime_buffer(z4h_values)   # 从 DB 灌入历史数据初始化
    def sync_position(...)        # 恢复仓位时同步到策略

三、与现有系统的集成

对 realtime_kline_service_base.py 的集成方式：

修改 1: __init__ 中初始化交易模块

# __init__ 末尾新增
self._trading_orchestrator = None
self._init_trading_module()

修改 2: _trigger_strategy_if_ready 策略驱动入口

分析完成后，若策略引擎就绪则调用 process_analysis()：

# _analyze_and_alert 流程中
strategy_acted = self._trigger_strategy_if_ready(
    symbol, timeframe, multi_period_result, price_data_cache, kline_time, start_time
)

_trigger_strategy_if_ready 内部逻辑：

• 检查 _trading_orchestrator 和 strategy.is_ready
• 调用 process_analysis(symbol, z4h, multi_period_result, timestamp, latest_alt_price, avg_zscore_4h)
• 策略引擎 process_tick() 返回 entry/exit 信号，由 Orchestrator 执行开/平仓

修改 3: stop() 中停止交易模块

四、数据库设计

新增 4 张表，遵循既有 TimescaleDB 模式（database/migrations/002_add_trading_tables.sql）：

trading_signals — 交易信号记录

pair_positions — 配对仓位

索引：(symbol, status), (status) WHERE status IN ('open','opening','closing')

trade_orders — 订单明细

daily_trading_stats — 每日交易统计

日期主键，汇总每日信号数/执行数/盈亏/回撤等。

五、配置参数

config.py 新增（环境变量驱动）

# ===== 自动交易配置 =====
TRADING_ENABLED = os.getenv('TRADING_ENABLED', 'false') == 'true'  # 总开关
TRADING_NETWORK = os.getenv('TRADING_NETWORK', 'testnet')  # 'mainnet' / 'testnet'
TRADING_PRIVATE_KEY = os.getenv('HYPERLIQUID_PRIVATE_KEY', '')

# 仓位
TRADING_BASE_POSITION_USD = 100.0    # 单腿基础仓位(USD)
TRADING_MAX_POSITION_USD = 500.0     # 单腿最大仓位

# 杠杆
TRADING_LEVERAGE = 3
TRADING_LEVERAGE_MODE = 'cross'

# 风控
TRADING_MAX_OPEN_PAIRS = 3           # 最大同时持仓对数
TRADING_MAX_EXPOSURE_USD = 3000.0    # 总最大敞口
TRADING_STOP_LOSS_PCT = 0.05         # 止损 5%
TRADING_MAX_DRAWDOWN_PCT = 0.10      # 最大回撤 10%
TRADING_MAX_DAILY_LOSS_USD = 500.0   # 每日最大亏损

# 安全
TRADING_RATE_LIMIT_PER_MINUTE = 30
TRADING_CIRCUIT_BREAKER_THRESHOLD = 5
TRADING_CIRCUIT_BREAKER_COOLDOWN = 300
TRADING_KILL_SWITCH_FILE = '/tmp/trading_kill_switch'

# 下单模式
TRADING_PAIR_MODE = 'single'   # 'pair'(配对两腿) / 'single'(仅目标币种)

# Plan A 策略参数
TRADING_STRATEGY_EMA_SPAN = 72
TRADING_STRATEGY_ADAPTIVE_THRESHOLD = 2.5
TRADING_STRATEGY_REVERSION_FACTOR = 0.18
TRADING_STRATEGY_COOLDOWN_MINUTES = 30

.env.example 新增

TRADING_ENABLED=false
TRADING_NETWORK=testnet                     # mainnet / testnet
HYPERLIQUID_PRIVATE_KEY=                    # ETH 私钥
TRADING_PAIR_MODE=pair                      # pair(配对两腿) / single(仅目标币种)
TRADING_BASE_POSITION_USD=100
TRADING_LEVERAGE=3
TRADING_MAX_OPEN_PAIRS=3
TRADING_STOP_LOSS_PCT=0.05

六、错误处理与安全

层次化错误处理

下单失败处理

single 模式（阶段一~四）：一笔订单，失败直接标记 rejected，逻辑简单。

pair 模式（阶段五）：

测试网模式 (Testnet)

• 连接 https://api.hyperliquid-testnet.xyz，使用测试资金真实下单
• 与主网完全相同的代码路径，仅 API 地址不同
• 飞书告警标题带 [测试网] 前缀以区分
• DB 记录标记 network='testnet'
• 开发和验证阶段使用测试网，确认无误后切换到主网

七、分阶段实施计划

阶段一：基础框架 + 单边模式 + 测试网

目标：建立完整模块骨架，以单边模式在测试网验证信号→下单→平仓全流程。

任务：

1. 创建 src/trading/models.py — 数据模型（TradingNetwork、Position、OrderResult 等）
2. 创建 src/trading/config.py — TradingConfig 配置加载（mainnet/testnet URL 映射）
3. 创建 src/trading/safety.py — KillSwitch, RateLimiter, CircuitBreaker
4. 创建 src/trading/executor.py — 单边执行引擎（wallet 初始化、Exchange、market_open/close 单笔订单）
5. 创建 database/migrations/002_add_trading_tables.sql — 建表 SQL
6. 创建 src/utils/database/trade_repository.py — 基础 CRUD
7. 创建 src/trading/position_manager.py — 开仓/平仓逻辑（单边）
8. 创建 src/trading/risk_manager.py — pre_trade_check + calculate_position_size
9. 创建 src/trading/__init__.py — TradingOrchestrator
10. 修改 src/services/realtime_kline_service_base.py — 5 处钩子
11. 更新 src/utils/core/config.py — 新增 TRADING_* 常量
12. 更新 .env.example

验收：

• 测试网连接成功，能查询账户状态
• 入场信号触发 → 测试网单边下单成功（仅 PURR）→ 仓位记录到 DB → 飞书通知 [测试网]
• 平仓信号触发 → 测试网平仓成功
• 交易模块异常不影响信号服务

阶段二：风控完善 + 止损

任务：

1. 止损监控线程（定期扫描活跃仓位 PnL）
2. 最大回撤检查
3. 每日亏损累计与每日重置
4. 账户余额检查
5. 信号强度缩放仓位大小
6. 熔断器自动恢复与告警

验收：止损触发自动平仓，回撤超限拒绝开仓

阶段三：恢复 + 可观测性

任务：

1. _recover_positions_from_db() 重启恢复
2. daily_trading_stats 自动更新
3. 交易统计加入监控日志
4. 飞书每日交易报告
5. stop() 优雅关闭

验收：重启后仓位正确恢复，每日统计完整

阶段四：主网验证（单边模式）

任务：

1. 单元测试（Mock SDK）：risk_manager, position_manager, safety
2. 测试网全流程集成测试
3. 切换到主网，极小仓位验证 (TRADING_BASE_POSITION_USD=10)
4. 主网运行 48h 监控
5. 确认无异常后逐步放大仓位

验收：主网单边模式稳定运行，无 bug

阶段五：两腿配对模式 (pair)

前提：阶段四完成，单边模式已在主网稳定运行一段时间。

任务：

1. executor.py 扩展 pair 路径 — 两腿下单、一腿失败回滚逻辑
2. position_manager.py 扩展 — 配对仓位管理（两腿盈亏计算）
3. position_manager.sync_with_exchange() — 两腿仓位同步
4. 测试网验证 pair 模式全流程
5. 主网小仓位验证 pair 模式

验收：配对下单成功，一腿失败正确回滚，TRADING_PAIR_MODE 切换正常

八、关键文件

九、验证路径

测试网+单边 → 测试网+单边+风控 → 主网+单边(小仓) → 主网+单边(正常仓) → 测试网+两腿 → 主网+两腿

1. 阶段一：TRADING_NETWORK=testnet, TRADING_PAIR_MODE=single → 信号触发 → 测试网 PURR 单边下单 → DB 记录 → 飞书 [测试网] 通知
2. 阶段二：测试网止损触发自动平仓 → 回撤超限拒绝开仓
3. 阶段三：重启服务 → 仓位自动恢复 → 每日统计完整
4. 阶段四：TRADING_NETWORK=mainnet + $10 仓位 → 运行 48h → DB/飞书/交易所一致 → 逐步放量
5. 阶段五：TRADING_PAIR_MODE=pair → 测试网验证两腿 → 主网小仓位验证 → 逐步放量