获取 Hyperliquid 用户的出入金记录

SHI XIAOLONG

— 10 min read
Hyperliquid 净入金算法
Hyperliquid 净入金算法详解 📋 目录 1. 算法概述 2. 核心概念 3. 计算公式 4. 算法实现 5. 交易类型分类 6. 应用场景 7. 完整代码示例 8. 测试用例 算法概述 什么是净入金（True Capital）？ 净入金是指用户实际投入到交易账户的净资金量，排除了交易盈亏的影响，只计算纯粹的资金流入和流出。 为什么需要计算净入金？ 在量化交易分析中，准确计算收益率、夏普比率、最大回撤等指标时，必须使用真实本金作为基准，而不是账户余额。 问题示例： * 用户充值 $10,000 * 交易盈利 $5,000（账户余额 $15,000） * 提现 $8,000（账户余额 $7,000） * 此时累计收益率应该是多少？
My MindSHI XIAOLONG

user_non_funding_ledger_updates() API 详细说明

📋 接口概述

user_non_funding_ledger_updates() 用于获取 Hyperliquid 用户的出入金记录（非资金费用的账本变动）。

包含的操作类型：

  • 转账（send）：用户之间的资金转移
  • 子账户转账（subAccountTransfer）：主账户与子账户间的资金划转
  • 充值（deposit）：从链上充值到交易所
  • 提现（withdrawal）：从交易所提现到链上
  • 其他账本变动：清算、奖励等

🔧 方法签名

def user_non_funding_ledger_updates(
    user: str,
    startTime: int,
    endTime: Optional[int] = None
) -> List[Dict]

参数说明

参数 类型 必需 说明
user str ✅ 用户地址（42字符的十六进制格式，如 0x...）
startTime int ✅ 起始时间（毫秒时间戳，包含）
endTime int ❌ 结束时间（毫秒时间戳，包含），默认为当前时间

返回值

List[Dict] - 账本变动记录列表

📊 返回数据结构

1. 转账记录（send）

{
    "time": 1769021429147,
    "hash": "0x4e573e4df0f08feb4fd00433c936a802097700338bf3aebdf21fe9a0aff469d5",
    "delta": {
        "type": "send",
        "user": "0x162cc7c861ebd0c06b3d72319201150482518185",
        "destination": "0xe3b6e3443c8f2080704e7421bad9340f13950acb",
        "sourceDex": "",
        "destinationDex": "",
        "token": "USDC",
        "amount": "4000000.0",
        "usdcValue": "4000000.0",
        "fee": "0.0",
        "nativeTokenFee": "0.0",
        "nonce": 1769021417642,
        "feeToken": ""
    }
}

字段说明：

字段 类型 说明
type str 固定为 "send"
user str 发送方地址
destination str 接收方地址
token str 代币类型（如 "USDC"）
amount str 转账金额
usdcValue str USDC 价值
fee str 手续费
nativeTokenFee str 原生代币手续费

2. 子账户转账（subAccountTransfer）

{
    "time": 1769443113340,
    "hash": "0x7ed1bb1ed4973c23804b043417e053020e2a00046f9a5af5229a6671939b160e",
    "delta": {
        "type": "subAccountTransfer",
        "usdc": "10.0",
        "user": "0xb3a38662575bdf1541013ce987934dba919851ea",
        "destination": "0x162cc7c861ebd0c06b3d72319201150482518185"
    }
}

字段说明：

字段 类型 说明
type str 固定为 "subAccountTransfer"
usdc str 转账金额（USDC）
user str 发送方地址
destination str 接收方地址

通用字段

字段 类型 说明
time int 操作时间（毫秒时间戳）
hash str 交易哈希
delta Dict 变动详情（核心数据）

💡 核心功能解读

1️⃣ 识别资金流向

对于查询地址，需要判断是收入还是支出：

# 转账类型（send）
if delta['destination'].lower() == target_address.lower():
    # 该地址是接收方 → 收入
    flow = 'incoming'
elif delta['user'].lower() == target_address.lower():
    # 该地址是发送方 → 支出
    flow = 'outgoing'

# 子账户转账类型（subAccountTransfer）
if delta['destination'].lower() == target_address.lower():
    # 该地址是接收方 → 收入
    flow = 'incoming'
elif delta['user'].lower() == target_address.lower():
    # 该地址是发送方 → 支出
    flow = 'outgoing'

2️⃣ 计算资金流

转账（send）：

amount = float(delta['amount'])

子账户转账（subAccountTransfer）：

amount = float(delta['usdc'])

3️⃣ 净流入计算

net_flow = total_incoming - total_outgoing
  • 正数：净流入（充值多于提现）
  • 负数：净流出（提现多于充值）

📈 使用场景

1. 资金流统计

计算用户在特定时间段内的总流入/流出：

from hyperliquid.info import Info
import time

info = Info(skip_ws=True)
address = "0x162cc7c861ebd0c06b3d72319201150482518185"

# 获取最近 30 天数据
current_time = int(time.time() * 1000)
start_time = current_time - (30 * 24 * 60 * 60 * 1000)

ledger_data = info.user_non_funding_ledger_updates(address, start_time)

# 区分收入和支出
incoming = []
outgoing = []

for record in ledger_data:
    delta = record['delta']
    record_type = delta['type']

    if record_type == 'send':
        if delta['destination'].lower() == address.lower():
            incoming.append(float(delta['amount']))
        elif delta['user'].lower() == address.lower():
            outgoing.append(float(delta['amount']))

    elif record_type == 'subAccountTransfer':
        if delta['destination'].lower() == address.lower():
            incoming.append(float(delta['usdc']))
        elif delta['user'].lower() == address.lower():
            outgoing.append(float(delta['usdc']))

total_in = sum(incoming)
total_out = sum(outgoing)
net_flow = total_in - total_out

print(f"总流入: {total_in:,.2f} USDC")
print(f"总流出: {total_out:,.2f} USDC")
print(f"净流入: {net_flow:,.2f} USDC")

2. 识别大额转账

筛选超过特定金额的转账：

threshold = 100000  # 10万 USDC

large_transfers = []

for record in ledger_data:
    delta = record['delta']

    if delta['type'] == 'send':
        amount = float(delta['amount'])
        if amount >= threshold:
            large_transfers.append({
                'time': record['time'],
                'from': delta['user'],
                'to': delta['destination'],
                'amount': amount,
                'token': delta['token']
            })

for t in large_transfers:
    print(f"{t['time']}: {t['amount']:,.2f} {t['token']} from {t['from']} to {t['to']}")

3. 时间线分析

按日统计资金流动：

from datetime import datetime
from collections import defaultdict

daily_stats = defaultdict(lambda: {'in': 0.0, 'out': 0.0})

for record in ledger_data:
    ts = record['time']
    date = datetime.fromtimestamp(ts / 1000).strftime('%Y-%m-%d')
    delta = record['delta']

    if delta['type'] == 'send':
        amount = float(delta.get('amount', 0))
        if delta['destination'].lower() == address.lower():
            daily_stats[date]['in'] += amount
        elif delta['user'].lower() == address.lower():
            daily_stats[date]['out'] += amount

# 显示每日净流入
for date in sorted(daily_stats.keys()):
    stats = daily_stats[date]
    net = stats['in'] - stats['out']
    print(f"{date}: 流入 {stats['in']:,.2f}  流出 {stats['out']:,.2f}  净额 {net:,.2f}")

⚠️ 注意事项

1. API 数据限制与分页

重要发现：此接口存在隐式数据上限（约 448 条记录）！

问题描述

  • 接口不支持显式分页参数（无 limit、offset、cursor 等）
  • 实测发现约 448 条记录的硬性限制
  • 无分页元数据（hasMore、nextCursor 等）
  • 超过上限的数据会被静默截断

影响评估

账户类型 数据量 风险等级 影响
普通用户 <100 条 🟢 低 单次查询完整
活跃交易者 100-448 条 🟡 中 接近 API 上限
专业机构 >448 条 🔴 高 数据截断严重
长期账户（1年+） 任意 🔴 高 早期数据丢失

解决方案：自适应分页

项目已实现自适应分页机制（参考 get_user_fills 实现）：

# 使用客户端的分页方法（推荐）
from address_analyzer.api_client import HyperliquidAPIClient

client = HyperliquidAPIClient(store=store)

# 自动分页，获取完整数据
ledger = await client.get_user_ledger(
    address,
    start_time=0,  # 从最早开始
    enable_pagination=True  # 默认启用
)

# 禁用分页（快速降级）
ledger = await client.get_user_ledger(
    address,
    enable_pagination=False  # 仅获取最多 448 条
)

分页机制说明

  • 阈值：2000 条（与 get_user_fills 一致，实际很少触发）
  • 触发条件：返回记录数 >= 2000
  • 分页方式：基于 last_time + 1 ms 连续查询
  • 终止条件：返回量 < 2000 或无新数据
  • 去重保障：基于 (time, hash, delta.type) 三元组去重
  • 性能优化：
    • 页间延迟 500ms 防止限流
    • 自动缓存避免重复查询
    • 99%+ 地址仅需 1 次 API 调用

性能影响

单页查询：~2-5s（适用于 99%+ 地址）
双页查询：~5-10s（极少触发）
三页查询：~8-15s（几乎不会发生）

2. 时间戳单位

重要：startTime 和 endTime 必须是毫秒时间戳，不是秒！

# ✅ 正确
import time
current_ms = int(time.time() * 1000)
start_ms = current_ms - (30 * 24 * 60 * 60 * 1000)

# ❌ 错误
current_sec = int(time.time())  # 这是秒，不是毫秒！

3. 地址格式

所有地址必须是小写才能正确比较：

# ✅ 正确
if delta['destination'].lower() == target_address.lower():
    pass

# ❌ 错误（可能因大小写不匹配而失败）
if delta['destination'] == target_address:
    pass

4. 数值类型

所有金额字段都是字符串类型，使用前需转换：

# ✅ 正确
amount = float(delta['amount'])

# ❌ 错误 - 不能直接计算
total = delta['amount'] + another_amount  # TypeError!

5. 记录类型

不同类型的记录，字段结构不同：

  • send: 使用 amount 字段
  • subAccountTransfer: 使用 usdc 字段
  • 其他类型可能有不同字段

建议使用 .get() 安全访问：

amount = float(delta.get('amount', 0))
usdc = float(delta.get('usdc', 0))

🔄 与其他接口的关系

接口 功能 区别
user_non_funding_ledger_updates() 出入金记录 资金的实际流入/流出
user_funding_history() 资金费率历史 持仓产生的资金费收支
user_fills() 交易成交记录 开平仓操作（主动交易）
user_state() 当前账户状态 实时持仓和余额（快照）

📌 实际案例分析

根据测试地址 0x162cc7c861ebd0c06b3d72319201150482518185 的数据：

【统计周期】30 天
【记录总数】43 条

【数据分类】
• send: 40 条
• subAccountTransfer: 3 条

【资金流分析】
转账统计 (send):
  收入: 18 笔，共 6,420,000.00 USDC
  支出: 22 笔，共 9,850,000.00 USDC
  净流入: -3,430,000.00 USDC

子账户转账 (subAccountTransfer):
  收入: 2 笔，共 30.00 USDC
  支出: 1 笔，共 10.00 USDC
  净流入: 20.00 USDC

【总计】
  总流入: 6,420,030.00 USDC
  总流出: 9,850,010.00 USDC
  净流入: -3,429,980.00 USDC（净流出）

🚀 API 客户端集成

项目已实现完整的分页支持，位于 address_analyzer/api_client.py：

async def get_user_ledger(
    self,
    address: str,
    start_time: int = 0,
    end_time: Optional[int] = None,
    use_cache: bool = True,
    enable_pagination: bool = True  # 默认启用分页
) -> List[Dict]:
    """
    获取用户出入金记录（支持自动分页）

    支持自动分页以确保数据完整性。API 限制约 448 条记录，
    对于活跃账户需要分页查询以获取全部历史。

    Args:
        address: 用户地址
        start_time: 起始时间戳（毫秒）
        end_time: 结束时间戳（毫秒）
        use_cache: 是否使用缓存
        enable_pagination: 是否启用分页（默认 True）

    Returns:
        账本变动列表，包含 send 和 subAccountTransfer 类型
    """
    # 实现包括：
    # - 自动分页查询（阈值 2000 条）
    # - 基于时间戳的连续查询（last_time + 1）
    # - 三元组去重：(time, hash, delta.type)
    # - 时间排序保证
    # - 智能缓存（包含时间范围）
    # - 页间延迟防止限流

使用示例

基础用法（自动分页）

from address_analyzer.api_client import HyperliquidAPIClient
from address_analyzer.data_store import get_store

store = get_store()
await store.connect()

client = HyperliquidAPIClient(store=store)

# 获取完整历史（自动分页）
ledger = await client.get_user_ledger(address)

# 指定时间范围
ledger = await client.get_user_ledger(
    address,
    start_time=start_ms,
    end_time=end_ms
)

高级用法（控制分页）

# 禁用分页（仅获取最多 448 条）
ledger = await client.get_user_ledger(
    address,
    enable_pagination=False
)

# 禁用缓存（强制从 API 获取）
ledger = await client.get_user_ledger(
    address,
    use_cache=False
)

内部方法

def _deduplicate_ledger(self, ledger: List[Dict]) -> List[Dict]:
    """
    去重账本记录（基于三元组：时间、哈希、类型）

    分页查询可能产生重复记录，使用三元组确保唯一性
    """
    # 实现基于 (time, hash, delta.type) 去重
    # 并按时间升序排序

async def _get_user_ledger_single(
    self,
    address: str,
    start_time: int = 0,
    end_time: Optional[int] = None
) -> List[Dict]:
    """
    单次查询版本（保留用于快速降级）

    仅当 enable_pagination=False 时使用
    """
    # 单次 API 调用，无分页逻辑

📚 参考资料

🧪 测试

项目提供完整的测试套件（test_user_ledger.py）：

# 运行所有测试
python3 test_user_ledger.py

# 运行特定测试
python3 test_user_ledger.py 3  # 分页功能测试
python3 test_user_ledger.py 4  # 数据完整性验证

测试覆盖

  1. 基础接口测试：验证 API 返回数据结构
  2. 完整工作流测试：数据获取 → 保存 → 指标计算
  3. 分页功能测试（新增）：
    • 启用/禁用分页对比
    • 去重机制验证
    • 时间范围查询
    • 数据完整性检查
    • 缓存机制验证
  4. 数据完整性验证（新增）：
    • 完整查询 vs 分段查询对比
    • 哈希去重验证
    • 覆盖率统计

文档版本: v2.0
更新日期: 2026-02-03
测试环境: Hyperliquid Python SDK
更新内容: 添加分页支持说明和测试方法

Read more

跑步的技巧（滚动落地）

“滚动落地（rolling contact / rolling foot strike）”不是一种教条式的“脚法”，而是一种 让冲击沿着整只脚、整条后链逐级传递的落地机制。 它的核心不是“你先用哪儿着地”，而是： 你的脚落地之后，冲击是不是像轮子一样滚过去，而不是像锤子一样砸下去。 这就是滚动落地的本质。 一、什么叫“滚动落地”？ 你可以把它理解成两种完全不同的落地方式： 1. 砸地（撞击式） 脚像锤子一样拍到地上： * 要么后跟先砸 * 要么前掌先戳 * 冲击集中在一个点 * 一个结构瞬间吃掉大部分载荷 结果就是： * 后跟砸 → 膝盖难受 * 前掌戳 → 前脚掌磨烂 * 都不是长跑友好模式 这叫 撞击式着地（impact strike）。 2. 滚地（滚动式） 脚像轮胎一样“滚”过地面： * 不是某一点硬砸 * 而是外侧中足先轻触 * 再向前滚到前掌 * 最后从大脚趾蹬离

By SHI XIAOLONG

AMI的优越性

世界模型（World Models）的具体例子 如下，我按类型分类，便于理解。每类都附带实际实现、演示效果和应用场景。 1. Yann LeCun / Meta 的 JEPA 系列（最直接对应“世界模型”概念） 这些是 LeCun 主张的非生成式抽象预测世界模型代表。 * I-JEPA（Image JEPA，2023） 输入一张图像，模型把不同区域（context 和 target）编码成抽象表示，然后预测 target 的表示（不在像素级别重建）。 例子：给定一张遮挡了部分物体的图片，模型能预测“被遮挡物体的大致位置和属性”，构建对物体持久性和空间关系的理解。 这是一个“原始世界模型”，能学习物理常识（如物体不会凭空消失）。 * V-JEPA / V-JEPA 2（Video JEPA，

By SHI XIAOLONG

什么是：“世界模型（World Models）”

世界模型（World Models） 是人工智能领域的一个核心概念，尤其在 Yann LeCun 等研究者推动的下一代 AI 架构中占据中心位置。它指的是 AI 系统在内部构建的对现实世界的抽象模拟或内部表示，让机器能够像人类或动物一样“理解”物理世界、预测未来、规划行动。 简单比喻 想象你闭上眼睛也能“看到”房间里的物体会如何移动、碰撞或掉落——这就是你大脑里的世界模型。AI 的世界模型就是类似的“数字孪生”（digital twin）或“内部模拟器”：它不是简单记住数据，而是学习世界的动态、因果关系和物理直觉（如重力、物体持久性、遮挡、因果等）。 为什么需要世界模型？ 当前主流的大型语言模型（LLM） 擅长处理文本（统计模式预测），但存在根本局限： * 缺乏对物理世界的真正理解 → 容易“幻觉”、无法可靠规划。 * 样本效率低 → 人类/

By SHI XIAOLONG

K线周期可配置化设计方案

K线周期可配置化设计方案 1. 背景与目标 当前 Beta 套利策略的 K 线周期硬编码为 "1h"，分散在多个文件中。需要： 1. 将 K 线周期从 1h 改为 2h 2. 提取为环境变量 BETA_ARB_KLINE_INTERVAL，使其可在 .env 中配置 2. 影响范围分析 2.1 需要修改的文件（共 6 个） 文件 硬编码位置 修改内容 src/trading/config.py BetaArbConfig dataclass 新增 kline_interval 字段，

By SHI XIAOLONG