Hyperliquid 净入金算法
基于该数据API获取数据

Hyperliquid 净入金算法详解
📋 目录
算法概述
什么是净入金(True Capital)?
净入金是指用户实际投入到交易账户的净资金量,排除了交易盈亏的影响,只计算纯粹的资金流入和流出。
为什么需要计算净入金?
在量化交易分析中,准确计算收益率、夏普比率、最大回撤等指标时,必须使用真实本金作为基准,而不是账户余额。
问题示例:
- 用户充值 $10,000
- 交易盈利 $5,000(账户余额 $15,000)
- 提现 $8,000(账户余额 $7,000)
- 此时累计收益率应该是多少?
❌ 错误计算:基于账户余额
净利润 = $7,000 - $10,000 = -$3,000
收益率 = -30%(错误!)
✅ 正确计算:基于净入金
净入金 = $10,000 - $8,000 = $2,000
交易盈亏 = $5,000
当前余额 = $7,000
收益率 = $5,000 / $2,000 = 250%(正确!)
核心概念
1. 资金流动类型
| 类型 | 英文 | 影响净入金 | 说明 |
|---|---|---|---|
| 充值 | Deposit | ✅ 增加 | 从外部钱包充值到交易账户 |
| 提现 | Withdraw | ✅ 减少 | 从交易账户提现到外部钱包 |
| 外部转入 | External In | ✅ 增加 | 别人通过 send 转给我的资金 |
| 外部转出 | External Out | ✅ 减少 | 我通过 send 转给别人的资金 |
| 内部转账 | Internal Transfer | ❌ 不影响 | Perp ↔ Spot 账户之间的转账 |
2. Hyperliquid 账户结构
Hyperliquid 交易所有两个子账户:
- Perp(永续合约账户):用于杠杆交易
- Spot(现货账户):用于现货交易
用户可以在这两个账户之间自由转账,但这不影响总资金量。
计算公式
核心公式
true_capital = deposits - withdrawals + external_to_spot - external_out
公式详解
净入金 = 总充值 - 总提现 + 外部转入 - 外部转出
| 符号 | 含义 | 示例 |
|---|---|---|
deposits |
用户充值总额 | +$10,000 |
withdrawals |
用户提现总额 | -$5,000 |
external_to_spot |
别人转入的金额 | +$1,000 |
external_out |
转给别人的金额 | -$500 |
true_capital |
净入金 | $5,500 |
算法实现
核心函数:calculate_true_capital()
def calculate_true_capital(self, user_address: str, ledger_records: List[Dict]) -> Dict[str, float]:
"""
计算真实本金(算法完整版)
计算公式:
true_capital = deposits - withdrawals + external_to_spot - external_out
考虑因素:
1. ✅ 充值和提现(deposit/withdraw)
2. ✅ 外部转入到 Spot(别人通过 send 转给我的)
3. ✅ 外部转出(我通过 send 转给别人的)
4. ❌ 排除内部 Perp ↔ Spot 转账(不影响总资金)
参数:
user_address: 用户地址(例如:0x7717a7a245d9f950e586822b8c9b46863ed7bd7e)
ledger_records: 账本记录列表
返回:
字典,包含:
- total_deposits: 总充值
- total_withdrawals: 总提现
- external_to_spot: 外部转入 Spot
- external_out: 外部转出
- true_capital: 真实本金
"""
# 初始化各项统计
total_deposits = 0.0
total_withdrawals = 0.0
external_to_spot = 0.0
external_out = 0.0
# 转换地址为小写(用于比较)
addr_lower = user_address.lower()
# 遍历所有账本记录
for record in ledger_records:
delta = record.get('delta', {})
delta_type = delta.get('type', '')
# ==================== 类型 A: 充值 ====================
if delta_type == 'deposit':
amount = safe_float(delta.get('usdc', 0))
total_deposits += amount
print(f" [充值] +${amount:.2f}")
# ==================== 类型 B: 提现 ====================
elif delta_type == 'withdraw':
amount = safe_float(delta.get('usdc', 0))
total_withdrawals += abs(amount) # 提现可能是负数,取绝对值
print(f" [提现] -${abs(amount):.2f}")
# ==================== 类型 C: 转账(最复杂) ====================
elif delta_type == 'send':
amount = safe_float(delta.get('amount', 0))
user = delta.get('user', '').lower()
dest = delta.get('destination', '').lower()
source_dex = delta.get('sourceDex', '')
dest_dex = delta.get('destinationDex', '')
# ---------- C1: 内部转账(Perp ↔ Spot)----------
is_internal_transfer = (user == addr_lower and dest == addr_lower)
if is_internal_transfer:
# 内部转账,不影响总资金,跳过
print(f" [内部转账] {source_dex} → {dest_dex} ${amount:.2f} (不计入)")
continue
# ---------- C2: 外部转入到 Spot ----------
if user != addr_lower and dest == addr_lower and dest_dex == 'spot':
external_to_spot += amount
print(f" [外部转入] +${amount:.2f} (from {user[:10]}...)")
# ---------- C3: 外部转出 ----------
elif user == addr_lower and dest != addr_lower:
external_out += amount
print(f" [外部转出] -${amount:.2f} (to {dest[:10]}...)")
# 计算净入金
true_capital = (
total_deposits - total_withdrawals +
external_to_spot - external_out
)
# 返回详细结果
return {
"total_deposits": total_deposits,
"total_withdrawals": total_withdrawals,
"external_to_spot": external_to_spot,
"external_out": external_out,
"true_capital": true_capital
}
辅助函数:safe_float()
def safe_float(value: Any, default: float = 0.0) -> float:
"""
安全地将值转换为浮点数
参数:
value: 任意类型的值
default: 转换失败时的默认值
返回:
float: 转换后的浮点数
"""
try:
if value is None:
return default
if isinstance(value, str):
return float(value) if value.strip() else default
return float(value)
except (ValueError, TypeError):
return default
交易类型分类
类型 A:充值(Deposit)
数据结构示例
{
"delta": {
"type": "deposit",
"usdc": "10000.5"
}
}
处理逻辑
if delta_type == 'deposit':
amount = safe_float(delta.get('usdc', 0))
total_deposits += amount # ✅ 增加净入金
业务含义
- 用户从外部钱包(如 MetaMask)向 Hyperliquid 交易账户充值
- 增加净入金
类型 B:提现(Withdraw)
数据结构示例
{
"delta": {
"type": "withdraw",
"usdc": "-5000.0"
}
}
处理逻辑
elif delta_type == 'withdraw':
amount = safe_float(delta.get('usdc', 0))
total_withdrawals += abs(amount) # ✅ 减少净入金(取绝对值)
业务含义
- 用户从 Hyperliquid 交易账户提现到外部钱包
- 减少净入金
- 注意:提现金额可能是负数,需使用
abs()取绝对值
类型 C:转账(Send)- 最复杂的情况
数据结构示例
{
"delta": {
"type": "send",
"amount": "1000.0",
"user": "0xabc...123",
"destination": "0xdef...456",
"sourceDex": "perp",
"destinationDex": "spot"
}
}
字段说明
amount: 转账金额user: 转出方地址destination: 转入方地址sourceDex: 来源账户类型(perp或spot)destinationDex: 目标账户类型(perp或spot)
C1:内部转账(Perp ↔ Spot)
判断条件
is_internal_transfer = (user == addr_lower and dest == addr_lower)
条件解释:
- 转出方是自己 且 转入方是自己
- 即:同一个用户在 Perp 和 Spot 账户之间转账
处理逻辑
if is_internal_transfer:
# ❌ 不影响总资金,跳过
continue
示例场景
用户地址: 0x7717...bd7e
操作:从 Perp 转 1000 USDC 到 Spot
数据:
user: 0x7717...bd7e
destination: 0x7717...bd7e
amount: 1000
判断:user == destination == 我的地址
结果:内部转账,不计入净入金
C2:外部转入 Spot
判断条件
if user != addr_lower and dest == addr_lower and dest_dex == 'spot':
external_to_spot += amount # ✅ 增加净入金
条件解释:
- 转出方不是自己(别人转给我)
- 转入方是自己
- 目标账户类型是
spot
处理逻辑
external_to_spot += amount # 增加净入金
示例场景
我的地址: 0x7717...bd7e
朋友地址: 0xabc...123
操作:朋友通过 send 转 500 USDC 给我的 Spot 账户
数据:
user: 0xabc...123 (朋友)
destination: 0x7717...bd7e (我)
destinationDex: spot
amount: 500
判断:user ≠ 我的地址 && destination == 我的地址 && destinationDex == 'spot'
结果:外部转入,增加净入金 +$500
C3:外部转出
判断条件
elif user == addr_lower and dest != addr_lower:
external_out += amount # ✅ 减少净入金
条件解释:
- 转出方是自己
- 转入方不是自己(我转给别人)
处理逻辑
external_out += amount # 减少净入金
示例场景
我的地址: 0x7717...bd7e
朋友地址: 0xdef...456
操作:我通过 send 转 300 USDC 给朋友
数据:
user: 0x7717...bd7e (我)
destination: 0xdef...456 (朋友)
amount: 300
判断:user == 我的地址 && destination ≠ 我的地址
结果:外部转出,减少净入金 -$300
应用场景
场景 1:计算累计收益率
代码位置:apex_fork.py:822-891
def calculate_return_metrics(
self,
current_value: float,
true_capital: float,
total_cumulative_pnl: float,
first_trade_time: int,
last_trade_time: int
) -> Dict[str, float]:
"""
计算累计收益率和年化收益率
使用净入金作为基准,确保收益率计算准确
"""
# 计算累计收益率
if true_capital > 0:
cumulative_return = (total_cumulative_pnl / true_capital) * 100
else:
cumulative_return = 0.0
return {
"cumulative_return": cumulative_return,
"annualized_return": annualized_return,
"net_profit_trading": total_cumulative_pnl,
"trading_days": trading_days
}
为什么需要净入金?
| 场景 | 基于账户余额(❌ 错误) | 基于净入金(✅ 正确) |
|---|---|---|
| 充值 $10,000 | - | - |
| 盈利 $2,000 | 账户余额 $12,000 | - |
| 提现 $5,000 | 账户余额 $7,000 | 净入金 $5,000 |
| 收益率 | $2,000 / $12,000 = 16.7% | $2,000 / $5,000 = 40% |
场景 2:计算夏普比率(Sharpe Ratio)
代码位置:apex_fork.py:952-1065
def calculate_sharpe_ratio_on_capital(
self,
fills: List[Dict],
true_capital: float,
risk_free_rate: float = 0.03
) -> Dict[str, float]:
"""
基于真实本金计算夏普比率
优势:
- ✅ 不受杠杆影响
- ✅ 真实反映风险收益比
- ✅ 与累计收益率计算逻辑一致
"""
trade_returns = []
for fill in fills:
closed_pnl = float(fill.get('closedPnl', 0))
if closed_pnl == 0:
continue
# ✅ 使用净入金计算收益率
trade_return = closed_pnl / true_capital
trade_returns.append(trade_return)
# 计算夏普比率
mean_return = sum(trade_returns) / len(trade_returns)
std_dev = calculate_std_dev(trade_returns)
sharpe_ratio = (mean_return - risk_free_rate) / std_dev
return {
"sharpe_ratio": sharpe_ratio,
"annualized_sharpe": sharpe_ratio * math.sqrt(trades_per_year)
}
为什么使用净入金?
假设:
- 投入本金: $1,000
- 使用 10 倍杠杆
- 持仓价值: $10,000
交易亏损 $500:
❌ 基于持仓价值: -$500 / $10,000 = -5%(严重低估风险)
✅ 基于净入金: -$500 / $1,000 = -50%(真实风险)
场景 3:计算最大回撤(Max Drawdown)
代码位置:apex_fork.py:1067-1205
def calculate_max_drawdown_on_capital(
self,
fills: List[Dict],
true_capital: float
) -> Dict[str, float]:
"""
基于真实本金计算最大回撤
关键改进:使用净入金而非持仓价值,不受杠杆影响
"""
trade_returns = []
for fill in fills:
closed_pnl = float(fill.get('closedPnl', 0))
if closed_pnl == 0:
continue
# ✅ 使用净入金计算收益率
trade_return = closed_pnl / true_capital
# 限制单笔收益率范围:[-0.99, 10.0]
trade_return = max(-0.99, min(trade_return, 10.0))
trade_returns.append(trade_return)
# 构建累计收益率序列
cumulative_returns = []
cumulative = 1.0
for ret in trade_returns:
cumulative *= (1 + ret)
cumulative_returns.append(cumulative)
# 计算最大回撤
peak = cumulative_returns[0]
max_drawdown = 0
for value in cumulative_returns:
if value > peak:
peak = value
drawdown = (peak - value) / peak * 100
if drawdown > max_drawdown:
max_drawdown = drawdown
return {
"max_drawdown_pct": max_drawdown,
"peak_return": (peak - 1) * 100,
"total_trades": len(trade_returns)
}
杠杆对回撤计算的影响:
| 场景 | 10倍杠杆 | 基于持仓价值(❌) | 基于净入金(✅) |
|---|---|---|---|
| 投入本金 | $100 | - | - |
| 持仓价值 | $1,000 | - | - |
| 亏损金额 | $50 | - | - |
| 回撤率 | - | -50/1000 = -5% | -50/100 = -50% |
完整代码示例
主程序示例
from typing import List, Dict, Any
from decimal import Decimal
def safe_float(value: Any, default: float = 0.0) -> float:
"""安全地将值转换为浮点数"""
try:
if value is None:
return default
if isinstance(value, str):
return float(value) if value.strip() else default
return float(value)
except (ValueError, TypeError):
return default
class TrueCapitalCalculator:
"""净入金计算器"""
def calculate_true_capital(
self,
user_address: str,
ledger_records: List[Dict]
) -> Dict[str, float]:
"""
计算真实本金
参数:
user_address: 用户地址
ledger_records: 账本记录列表
返回:
包含净入金详情的字典
"""
total_deposits = 0.0
total_withdrawals = 0.0
external_to_spot = 0.0
external_out = 0.0
addr_lower = user_address.lower()
print(f"\n分析地址: {user_address}")
print("=" * 60)
for record in ledger_records:
delta = record.get('delta', {})
delta_type = delta.get('type', '')
if delta_type == 'deposit':
amount = safe_float(delta.get('usdc', 0))
total_deposits += amount
print(f" [充值] +${amount:,.2f}")
elif delta_type == 'withdraw':
amount = safe_float(delta.get('usdc', 0))
total_withdrawals += abs(amount)
print(f" [提现] -${abs(amount):,.2f}")
elif delta_type == 'send':
amount = safe_float(delta.get('amount', 0))
user = delta.get('user', '').lower()
dest = delta.get('destination', '').lower()
source_dex = delta.get('sourceDex', '')
dest_dex = delta.get('destinationDex', '')
is_internal_transfer = (user == addr_lower and dest == addr_lower)
if is_internal_transfer:
print(f" [内部转账] {source_dex} → {dest_dex} ${amount:,.2f} (不计入)")
continue
if user != addr_lower and dest == addr_lower and dest_dex == 'spot':
external_to_spot += amount
print(f" [外部转入] +${amount:,.2f}")
elif user == addr_lower and dest != addr_lower:
external_out += amount
print(f" [外部转出] -${amount:,.2f}")
true_capital = (
total_deposits - total_withdrawals +
external_to_spot - external_out
)
print("=" * 60)
print(f"\n📊 统计结果:")
print(f" 总充值: +${total_deposits:,.2f}")
print(f" 总提现: -${total_withdrawals:,.2f}")
print(f" 外部转入: +${external_to_spot:,.2f}")
print(f" 外部转出: -${external_out:,.2f}")
print(f" ━━━━━━━━━━━━━━━━━━━━━━━━━━")
print(f" 净入金: ${true_capital:,.2f}")
return {
"total_deposits": total_deposits,
"total_withdrawals": total_withdrawals,
"external_to_spot": external_to_spot,
"external_out": external_out,
"true_capital": true_capital
}
# 使用示例
if __name__ == "__main__":
calculator = TrueCapitalCalculator()
# 模拟账本记录
ledger_records = [
{"delta": {"type": "deposit", "usdc": "10000"}},
{"delta": {"type": "withdraw", "usdc": "-3000"}},
{
"delta": {
"type": "send",
"amount": "1000",
"user": "0x7717a7a245d9f950e586822b8c9b46863ed7bd7e",
"destination": "0x7717a7a245d9f950e586822b8c9b46863ed7bd7e",
"sourceDex": "perp",
"destinationDex": "spot"
}
},
{
"delta": {
"type": "send",
"amount": "500",
"user": "0xabc123",
"destination": "0x7717a7a245d9f950e586822b8c9b46863ed7bd7e",
"destinationDex": "spot"
}
},
{
"delta": {
"type": "send",
"amount": "200",
"user": "0x7717a7a245d9f950e586822b8c9b46863ed7bd7e",
"destination": "0xdef456"
}
}
]
result = calculator.calculate_true_capital(
user_address="0x7717a7a245d9f950e586822b8c9b46863ed7bd7e",
ledger_records=ledger_records
)
运行结果示例
分析地址: 0x7717a7a245d9f950e586822b8c9b46863ed7bd7e
============================================================
[充值] +$10,000.00
[提现] -$3,000.00
[内部转账] perp → spot $1,000.00 (不计入)
[外部转入] +$500.00
[外部转出] -$200.00
============================================================
📊 统计结果:
总充值: +$10,000.00
总提现: -$3,000.00
外部转入: +$500.00
外部转出: -$200.00
━━━━━━━━━━━━━━━━━━━━━━━━━━
净入金: $7,300.00
测试用例
测试用例 1:纯充值提现
def test_case_1():
"""测试用例 1: 只有充值和提现"""
ledger_records = [
{"delta": {"type": "deposit", "usdc": "10000"}},
{"delta": {"type": "deposit", "usdc": "5000"}},
{"delta": {"type": "withdraw", "usdc": "-3000"}},
]
calculator = TrueCapitalCalculator()
result = calculator.calculate_true_capital(
user_address="0x7717a7a245d9f950e586822b8c9b46863ed7bd7e",
ledger_records=ledger_records
)
# 预期结果
expected_true_capital = 10000 + 5000 - 3000 # = 12000
assert result['true_capital'] == expected_true_capital
print("✅ 测试用例 1 通过")
测试用例 2:内部转账不影响净入金
def test_case_2():
"""测试用例 2: 内部转账(Perp ↔ Spot)"""
my_address = "0x7717a7a245d9f950e586822b8c9b46863ed7bd7e"
ledger_records = [
{"delta": {"type": "deposit", "usdc": "10000"}},
{
"delta": {
"type": "send",
"amount": "5000",
"user": my_address,
"destination": my_address,
"sourceDex": "perp",
"destinationDex": "spot"
}
},
{
"delta": {
"type": "send",
"amount": "2000",
"user": my_address,
"destination": my_address,
"sourceDex": "spot",
"destinationDex": "perp"
}
}
]
calculator = TrueCapitalCalculator()
result = calculator.calculate_true_capital(
user_address=my_address,
ledger_records=ledger_records
)
# 预期结果:内部转账不影响净入金
expected_true_capital = 10000 # 只有充值
assert result['true_capital'] == expected_true_capital
print("✅ 测试用例 2 通过")
测试用例 3:外部转入转出
def test_case_3():
"""测试用例 3: 外部转入和转出"""
my_address = "0x7717a7a245d9f950e586822b8c9b46863ed7bd7e"
friend_address = "0xabc123"
ledger_records = [
{"delta": {"type": "deposit", "usdc": "10000"}},
{
"delta": {
"type": "send",
"amount": "1000",
"user": friend_address,
"destination": my_address,
"destinationDex": "spot"
}
},
{
"delta": {
"type": "send",
"amount": "500",
"user": my_address,
"destination": friend_address
}
}
]
calculator = TrueCapitalCalculator()
result = calculator.calculate_true_capital(
user_address=my_address,
ledger_records=ledger_records
)
# 预期结果
expected_true_capital = 10000 + 1000 - 500 # = 10500
assert result['true_capital'] == expected_true_capital
print("✅ 测试用例 3 通过")
测试用例 4:综合场景
def test_case_4():
"""测试用例 4: 综合场景"""
my_address = "0x7717a7a245d9f950e586822b8c9b46863ed7bd7e"
friend_address = "0xabc123"
ledger_records = [
{"delta": {"type": "deposit", "usdc": "10000"}}, # +10000
{"delta": {"type": "withdraw", "usdc": "-2000"}}, # -2000
{
"delta": {
"type": "send",
"amount": "3000",
"user": my_address,
"destination": my_address,
"sourceDex": "perp",
"destinationDex": "spot"
}
}, # 内部转账,不计入
{
"delta": {
"type": "send",
"amount": "1500",
"user": friend_address,
"destination": my_address,
"destinationDex": "spot"
}
}, # +1500
{
"delta": {
"type": "send",
"amount": "800",
"user": my_address,
"destination": friend_address
}
}, # -800
{"delta": {"type": "deposit", "usdc": "5000"}}, # +5000
]
calculator = TrueCapitalCalculator()
result = calculator.calculate_true_capital(
user_address=my_address,
ledger_records=ledger_records
)
# 预期结果
expected_true_capital = (
10000 + 5000 # 充值
- 2000 # 提现
+ 1500 # 外部转入
- 800 # 外部转出
) # = 13700
assert result['true_capital'] == expected_true_capital
assert result['total_deposits'] == 15000
assert result['total_withdrawals'] == 2000
assert result['external_to_spot'] == 1500
assert result['external_out'] == 800
print("✅ 测试用例 4 通过")
运行所有测试
if __name__ == "__main__":
print("开始运行测试用例...\n")
test_case_1()
test_case_2()
test_case_3()
test_case_4()
print("\n🎉 所有测试用例通过!")
算法优势总结
| 优势 | 说明 |
|---|---|
| ✅ 准确性 | 排除交易盈亏,只计算资金流入流出 |
| ✅ 完整性 | 考虑充值、提现、转账等所有资金流动 |
| ✅ 智能过滤 | 自动识别并排除内部转账(Perp ↔ Spot) |
| ✅ 真实反映 | 准确反映用户的资金使用效率 |
| ✅ 不受杠杆影响 | 基于真实本金计算,不受杠杆倍数影响 |
| ✅ 多场景应用 | 支持收益率、夏普比率、最大回撤等多指标计算 |
参考资料
- Hyperliquid API 文档: https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api
- Apex Liquid Bot: https://apexliquid.bot/