Hyperliquid交易记录持仓时间统计算法

SHI XIAOLONG

— 15 min read

持仓时间统计算法详解

📋 目录

  1. 算法概述
  2. 核心概念
  3. 计算公式
  4. 算法实现
  5. 应用场景
  6. 完整代码示例
  7. 测试用例

算法概述

什么是持仓时间统计？

持仓时间统计是计算交易者从开仓到平仓的平均持仓时长，反映交易风格和策略特点。

指标意义

平均持仓时间 交易风格 特点
< 1 小时 超短线 / 高频 快进快出，频繁交易
1-24 小时 日内交易 当日平仓，不过夜
1-7 天 短线交易 捕捉短期趋势
7-30 天 中线交易 把握中期波动
> 30 天 长线交易 长期持有，价值投资

为什么需要持仓时间统计？

  • 了解交易风格：明确自己的交易偏好
  • 优化策略：根据持仓时间调整策略参数
  • 风险管理：长短线风险特征不同
  • 资金效率：评估资金周转率

核心概念

1. 交易方向识别

Hyperliquid 交易方向分为：

  • 多头（Long）：看涨，买入开仓
  • 空头（Short）：看跌，卖出开仓

2. 开仓与平仓

操作类型 多头 空头
开仓 Open Long Open Short
平仓 Close Long Close Short
翻仓 Short > Long Long > Short

3. 部分平仓

支持部分平仓，需要按比例匹配开仓记录：

示例：

开多 BTC 10 个 @ $40,000
平多 BTC 6 个 @ $42,000  → 部分平仓 6 个
平多 BTC 4 个 @ $43,000  → 剩余 4 个全部平仓

4. FIFO 原则

先进先出（First In First Out）：

  • 平仓时优先匹配最早的开仓记录
  • 确保持仓时间计算准确

计算公式

核心公式

1. 单笔持仓时间

hold_time_seconds = close_time - open_time
hold_time_days = hold_time_seconds / 86400

2. 平均持仓时间

average_hold_time = Σ(hold_time_i) / n

其中 n 为已平仓的交易数量

3. 时间段统计

今日平均 = Σ(今日平仓的持仓时间) / 今日平仓数量
最近7天平均 = Σ(最近7天平仓的持仓时间) / 最近7天平仓数量
最近30天平均 = Σ(最近30天平仓的持仓时间) / 最近30天平仓数量
全部时间平均 = Σ(所有平仓的持仓时间) / 所有平仓数量

算法实现

改进版算法特点

代码位置: apex_fork.py:430-585

关键改进：

  1. ✅ 区分多头和空头仓位，分别配对
  2. ✅ 支持部分平仓的加权计算
  3. ✅ 使用 FIFO 原则进行配对
  4. ✅ 正确处理翻仓交易（Long > Short, Short > Long）

核心函数：calculate_hold_time_stats()

from typing import List, Dict
from datetime import datetime, timedelta
from collections import defaultdict


def calculate_hold_time_stats(fills: List[Dict]) -> Dict[str, float]:
    """
    计算平均持仓时间统计（改进版：区分多空方向，支持部分平仓）

    参数：
        fills: 成交记录列表，包含'time'、'dir'、'coin'和'sz'字段

    返回：
        字典，包含不同时间段的平均持仓时间（天数）：
        - todayCount: 今日平均持仓时间
        - last7DaysAverage: 最近7天平均持仓时间
        - last30DaysAverage: 最近30天平均持仓时间
        - allTimeAverage: 全部时间平均持仓时间

    算法改进：
        1. 区分多头(Long)和空头(Short)仓位，分别配对
        2. 支持部分平仓的加权计算
        3. 使用FIFO原则进行配对
        4. 正确处理翻仓交易(Long > Short, Short > Long)

    算法说明：
        1. 为每个币种维护独立的多头和空头开仓队列
        2. 平仓时从对应方向的队列中按FIFO原则匹配
        3. 支持部分平仓：按数量比例匹配开仓记录
        4. 计算每对交易的持仓时长并按时间段统计
    """
    # 边界条件检查
    if not fills:
        return {
            "todayCount": 0,
            "last7DaysAverage": 0,
            "last30DaysAverage": 0,
            "allTimeAverage": 0
        }

    # 时间基准点
    now = datetime.now()
    today_start = datetime(now.year, now.month, now.day)
    week_ago = today_start - timedelta(days=7)
    month_ago = today_start - timedelta(days=30)

    # ==================== 数据结构初始化 ====================

    # 为每个币种维护多头和空头的开仓队列
    # 队列中存储 [开仓时间, 剩余数量]
    long_open_positions = defaultdict(list)
    short_open_positions = defaultdict(list)

    # 存储所有已配对的持仓记录 (开仓时间, 平仓时间, 持仓数量)
    completed_positions = []

    # 按时间排序
    sorted_fills = sorted(fills, key=lambda x: x.get('time', 0))

    # ==================== 主循环：遍历所有成交记录 ====================

    for fill in sorted_fills:
        coin = fill.get('coin', '')
        direction = fill.get('dir', '').strip()
        timestamp = fill.get('time', 0)
        size = abs(float(fill.get('sz', 0)))

        if not coin or not timestamp or size == 0:
            continue

        # 标准化方向字符串（不区分大小写）
        dir_lower = direction.lower()

        # ==================== 处理开多仓交易 ====================
        if 'open long' in dir_lower and 'short' not in dir_lower:
            long_open_positions[coin].append([timestamp, size])

        # ==================== 处理开空仓交易 ====================
        elif 'open short' in dir_lower and 'long' not in dir_lower:
            short_open_positions[coin].append([timestamp, size])

        # ==================== 处理平多仓交易（支持部分平仓）====================
        elif 'close long' in dir_lower and 'short' not in dir_lower:
            remaining_size = size

            # 循环匹配开仓记录，直到全部平仓
            while remaining_size > 1e-9 and long_open_positions[coin]:
                open_time, open_size = long_open_positions[coin][0]

                if open_size <= remaining_size:
                    # 完全平掉这笔开仓
                    completed_positions.append((open_time, timestamp, open_size))
                    remaining_size -= open_size
                    long_open_positions[coin].pop(0)
                else:
                    # 部分平仓
                    completed_positions.append((open_time, timestamp, remaining_size))
                    long_open_positions[coin][0][1] -= remaining_size
                    remaining_size = 0

        # ==================== 处理平空仓交易（支持部分平仓）====================
        elif 'close short' in dir_lower and 'long' not in dir_lower:
            remaining_size = size

            # 循环匹配开仓记录，直到全部平仓
            while remaining_size > 1e-9 and short_open_positions[coin]:
                open_time, open_size = short_open_positions[coin][0]

                if open_size <= remaining_size:
                    # 完全平掉这笔开仓
                    completed_positions.append((open_time, timestamp, open_size))
                    remaining_size -= open_size
                    short_open_positions[coin].pop(0)
                else:
                    # 部分平仓
                    completed_positions.append((open_time, timestamp, remaining_size))
                    short_open_positions[coin][0][1] -= remaining_size
                    remaining_size = 0

        # ==================== 处理翻仓交易：从空翻多 (Short > Long) ====================
        elif 'short > long' in dir_lower or 'short>long' in dir_lower:
            # 先平掉所有空头仓位
            while short_open_positions[coin]:
                open_time, open_size = short_open_positions[coin].pop(0)
                completed_positions.append((open_time, timestamp, open_size))
            # 然后作为开多仓处理
            long_open_positions[coin].append([timestamp, size])

        # ==================== 处理翻仓交易：从多翻空 (Long > Short) ====================
        elif 'long > short' in dir_lower or 'long>short' in dir_lower:
            # 先平掉所有多头仓位
            while long_open_positions[coin]:
                open_time, open_size = long_open_positions[coin].pop(0)
                completed_positions.append((open_time, timestamp, open_size))
            # 然后作为开空仓处理
            short_open_positions[coin].append([timestamp, size])

    # ==================== 计算所有配对交易的持仓时间 ====================

    today_hold_times = []
    week_hold_times = []
    month_hold_times = []
    all_hold_times = []

    for open_time, close_time, position_size in completed_positions:
        open_dt = datetime.fromtimestamp(open_time / 1000)
        close_dt = datetime.fromtimestamp(close_time / 1000)

        # 计算持仓时间（天数）
        hold_time_days = (close_dt - open_dt).total_seconds() / 86400
        all_hold_times.append(hold_time_days)

        # 按时间段分类
        if close_dt >= today_start:
            today_hold_times.append(hold_time_days)

        if close_dt >= week_ago:
            week_hold_times.append(hold_time_days)

        if close_dt >= month_ago:
            month_hold_times.append(hold_time_days)

    # ==================== 返回统计结果 ====================

    return {
        "todayCount": sum(today_hold_times) / len(today_hold_times) if today_hold_times else 0,
        "last7DaysAverage": sum(week_hold_times) / len(week_hold_times) if week_hold_times else 0,
        "last30DaysAverage": sum(month_hold_times) / len(month_hold_times) if month_hold_times else 0,
        "allTimeAverage": sum(all_hold_times) / len(all_hold_times) if all_hold_times else 0
    }

算法流程图

开始
  ↓
检查数据是否为空？
  ↓ 是
返回零值结果
  ↓ 否
初始化数据结构
  ├─ long_open_positions: 多头开仓队列（按币种）
  ├─ short_open_positions: 空头开仓队列（按币种）
  └─ completed_positions: 已配对的持仓记录
  ↓
按时间排序成交记录
  ↓
遍历所有成交记录
  ↓
对每条记录：
  获取 coin, direction, timestamp, size
  ↓
  判断交易类型：
  ├─ "Open Long" → 添加到 long_open_positions[coin]
  ├─ "Open Short" → 添加到 short_open_positions[coin]
  ├─ "Close Long" → 匹配 long_open_positions[coin]（FIFO）
  │   ├─ 完全平仓 → 记录持仓，移除开仓记录
  │   └─ 部分平仓 → 记录持仓，更新剩余数量
  ├─ "Close Short" → 匹配 short_open_positions[coin]（FIFO）
  │   ├─ 完全平仓 → 记录持仓，移除开仓记录
  │   └─ 部分平仓 → 记录持仓，更新剩余数量
  ├─ "Short > Long" → 平掉所有空头，开新多头
  └─ "Long > Short" → 平掉所有多头，开新空头
  ↓
所有记录处理完毕
  ↓
计算持仓时间
  对每个 completed_position:
    - 计算 hold_time_days = (close_time - open_time) / 86400000
    - 按平仓时间分类到不同时间段
  ↓
计算各时间段平均持仓时间
  ├─ todayCount: 今日平均
  ├─ last7DaysAverage: 最近7天平均
  ├─ last30DaysAverage: 最近30天平均
  └─ allTimeAverage: 全部时间平均
  ↓
返回统计结果
  ↓
结束

应用场景

场景 1：识别交易风格

# 计算持仓时间
hold_stats = calculate_hold_time_stats(fills)

avg_hold_days = hold_stats['allTimeAverage']

print(f"平均持仓时间: {avg_hold_days:.2f} 天")

# 识别交易风格
if avg_hold_days < 1/24:  # < 1小时
    print("交易风格: 超短线 / 高频交易")
elif avg_hold_days < 1:  # < 1天
    print("交易风格: 日内交易")
elif avg_hold_days < 7:  # < 1周
    print("交易风格: 短线交易")
elif avg_hold_days < 30:  # < 1月
    print("交易风格: 中线交易")
else:
    print("交易风格: 长线交易")

场景 2：趋势分析

# 对比不同时期的持仓时间
hold_stats = calculate_hold_time_stats(fills)

print("\n持仓时间趋势分析:")
print(f"今日平均: {hold_stats['todayCount']:.2f} 天")
print(f"最近7天平均: {hold_stats['last7DaysAverage']:.2f} 天")
print(f"最近30天平均: {hold_stats['last30DaysAverage']:.2f} 天")
print(f"全部时间平均: {hold_stats['allTimeAverage']:.2f} 天")

# 判断趋势
recent = hold_stats['last7DaysAverage']
overall = hold_stats['allTimeAverage']

if recent > overall * 1.2:
    print("\n趋势: 持仓时间延长 📈（可能转向中长线）")
elif recent < overall * 0.8:
    print("\n趋势: 持仓时间缩短 📉（可能转向短线）")
else:
    print("\n趋势: 持仓时间稳定 ⚖️")

场景 3：策略优化建议

def get_strategy_advice(hold_stats: Dict, win_rate: float, profit_factor: float) -> str:
    """根据持仓时间和其他指标给出建议"""
    avg_hold = hold_stats['allTimeAverage']

    if avg_hold < 1:
        # 日内交易
        if win_rate < 55:
            return "⚠️ 日内交易胜率偏低，建议提高交易准确性或延长持仓"
        elif profit_factor < 1.5:
            return "⚠️ 日内交易盈亏比不足，建议优化止损止盈策略"
        else:
            return "✅ 日内交易表现良好"

    elif avg_hold < 7:
        # 短线交易
        if profit_factor < 2.0:
            return "⚠️ 短线交易建议盈亏比 > 2.0"
        else:
            return "✅ 短线交易表现合理"

    else:
        # 中长线交易
        if profit_factor < 2.5:
            return "⚠️ 中长线交易建议盈亏比 > 2.5"
        else:
            return "✅ 中长线交易表现优秀"


# 使用示例
advice = get_strategy_advice(hold_stats, win_rate=60, profit_factor=2.2)
print(advice)

完整代码示例

主程序示例

from typing import List, Dict
from datetime import datetime, timedelta
from collections import defaultdict


class HoldTimeCalculator:
    """持仓时间统计计算器"""

    def calculate_hold_time_stats(self, fills: List[Dict]) -> Dict[str, float]:
        """计算持仓时间统计"""

        if not fills:
            return self._zero_result()

        # 时间基准
        now = datetime.now()
        today_start = datetime(now.year, now.month, now.day)
        week_ago = today_start - timedelta(days=7)
        month_ago = today_start - timedelta(days=30)

        # 数据结构
        long_open_positions = defaultdict(list)
        short_open_positions = defaultdict(list)
        completed_positions = []

        # 按时间排序
        sorted_fills = sorted(fills, key=lambda x: x.get('time', 0))

        print("\n📊 持仓时间配对分析:")
        print("=" * 90)
        print(f"{'序号':<6} {'币种':<8} {'方向':<15} {'操作':<15} {'数量':<10} {'状态'}")
        print("=" * 90)

        # 遍历处理
        for i, fill in enumerate(sorted_fills, 1):
            coin = fill.get('coin', '')
            direction = fill.get('dir', '').strip()
            timestamp = fill.get('time', 0)
            size = abs(float(fill.get('sz', 0)))

            if not coin or not timestamp or size == 0:
                continue

            dir_lower = direction.lower()
            operation = ""
            status = ""

            # 开多仓
            if 'open long' in dir_lower and 'short' not in dir_lower:
                long_open_positions[coin].append([timestamp, size])
                operation = "开多仓"
                status = f"✅ 队列+1"

            # 开空仓
            elif 'open short' in dir_lower and 'long' not in dir_lower:
                short_open_positions[coin].append([timestamp, size])
                operation = "开空仓"
                status = f"✅ 队列+1"

            # 平多仓
            elif 'close long' in dir_lower and 'short' not in dir_lower:
                remaining_size = size
                matched_count = 0

                while remaining_size > 1e-9 and long_open_positions[coin]:
                    open_time, open_size = long_open_positions[coin][0]

                    matched_size = min(open_size, remaining_size)
                    completed_positions.append((open_time, timestamp, matched_size))

                    if open_size <= remaining_size:
                        remaining_size -= open_size
                        long_open_positions[coin].pop(0)
                    else:
                        long_open_positions[coin][0][1] -= remaining_size
                        remaining_size = 0

                    matched_count += 1

                operation = "平多仓"
                status = f"✅ 配对 {matched_count} 笔"

            # 平空仓
            elif 'close short' in dir_lower and 'long' not in dir_lower:
                remaining_size = size
                matched_count = 0

                while remaining_size > 1e-9 and short_open_positions[coin]:
                    open_time, open_size = short_open_positions[coin][0]

                    matched_size = min(open_size, remaining_size)
                    completed_positions.append((open_time, timestamp, matched_size))

                    if open_size <= remaining_size:
                        remaining_size -= open_size
                        short_open_positions[coin].pop(0)
                    else:
                        short_open_positions[coin][0][1] -= remaining_size
                        remaining_size = 0

                    matched_count += 1

                operation = "平空仓"
                status = f"✅ 配对 {matched_count} 笔"

            # 从空翻多
            elif 'short > long' in dir_lower or 'short>long' in dir_lower:
                closed_count = len(short_open_positions[coin])
                while short_open_positions[coin]:
                    open_time, open_size = short_open_positions[coin].pop(0)
                    completed_positions.append((open_time, timestamp, open_size))
                long_open_positions[coin].append([timestamp, size])
                operation = "从空翻多"
                status = f"✅ 平{closed_count}空 开1多"

            # 从多翻空
            elif 'long > short' in dir_lower or 'long>short' in dir_lower:
                closed_count = len(long_open_positions[coin])
                while long_open_positions[coin]:
                    open_time, open_size = long_open_positions[coin].pop(0)
                    completed_positions.append((open_time, timestamp, open_size))
                short_open_positions[coin].append([timestamp, size])
                operation = "从多翻空"
                status = f"✅ 平{closed_count}多 开1空"

            # 打印详情
            print(f"{i:<6} {coin:<8} {direction:<15} {operation:<15} {size:<10.4f} {status}")

        print("=" * 90)

        # 计算持仓时间
        today_hold_times = []
        week_hold_times = []
        month_hold_times = []
        all_hold_times = []

        print("\n📈 持仓时间详细分析:")
        print("=" * 90)
        print(f"{'序号':<6} {'开仓时间':<20} {'平仓时间':<20} {'持仓天数':<12} {'分类'}")
        print("=" * 90)

        for idx, (open_time, close_time, position_size) in enumerate(completed_positions, 1):
            open_dt = datetime.fromtimestamp(open_time / 1000)
            close_dt = datetime.fromtimestamp(close_time / 1000)

            hold_time_days = (close_dt - open_dt).total_seconds() / 86400
            all_hold_times.append(hold_time_days)

            # 分类
            categories = []
            if close_dt >= today_start:
                today_hold_times.append(hold_time_days)
                categories.append("今日")

            if close_dt >= week_ago:
                week_hold_times.append(hold_time_days)
                if "今日" not in categories:
                    categories.append("本周")

            if close_dt >= month_ago:
                month_hold_times.append(hold_time_days)
                if not categories:
                    categories.append("本月")

            if not categories:
                categories.append("历史")

            category_str = "/".join(categories)

            # 格式化时间
            open_str = open_dt.strftime('%Y-%m-%d %H:%M:%S')
            close_str = close_dt.strftime('%Y-%m-%d %H:%M:%S')

            # 打印（只显示前20笔）
            if idx <= 20:
                print(f"{idx:<6} {open_str:<20} {close_str:<20} {hold_time_days:<11.2f}天 {category_str}")

        if len(completed_positions) > 20:
            print(f"... 省略 {len(completed_positions) - 20} 笔 ...")

        print("=" * 90)

        # 计算统计结果
        result = {
            "todayCount": sum(today_hold_times) / len(today_hold_times) if today_hold_times else 0,
            "last7DaysAverage": sum(week_hold_times) / len(week_hold_times) if week_hold_times else 0,
            "last30DaysAverage": sum(month_hold_times) / len(month_hold_times) if month_hold_times else 0,
            "allTimeAverage": sum(all_hold_times) / len(all_hold_times) if all_hold_times else 0
        }

        # 打印统计结果
        print(f"\n🎯 持仓时间统计:")
        print("=" * 90)
        print(f"今日平均持仓: {result['todayCount']:.2f} 天 ({len(today_hold_times)} 笔)")
        print(f"最近7天平均: {result['last7DaysAverage']:.2f} 天 ({len(week_hold_times)} 笔)")
        print(f"最近30天平均: {result['last30DaysAverage']:.2f} 天 ({len(month_hold_times)} 笔)")
        print(f"全部时间平均: {result['allTimeAverage']:.2f} 天 ({len(all_hold_times)} 笔)")

        # 交易风格判断
        avg_hold = result['allTimeAverage']
        if avg_hold < 1/24:
            style = "超短线 / 高频交易"
        elif avg_hold < 1:
            style = "日内交易"
        elif avg_hold < 7:
            style = "短线交易"
        elif avg_hold < 30:
            style = "中线交易"
        else:
            style = "长线交易"

        print(f"\n交易风格: {style}")

        return result

    def _zero_result(self) -> Dict[str, float]:
        """返回零值结果"""
        return {
            "todayCount": 0,
            "last7DaysAverage": 0,
            "last30DaysAverage": 0,
            "allTimeAverage": 0
        }


# 使用示例
if __name__ == "__main__":
    calculator = HoldTimeCalculator()

    # 模拟交易数据
    import time
    current_time = int(time.time() * 1000)

    fills = [
        # 第1组：BTC 多头交易
        {"coin": "BTC", "dir": "Open Long", "time": current_time - 10 * 86400000, "sz": "1.0"},
        {"coin": "BTC", "dir": "Close Long", "time": current_time - 8 * 86400000, "sz": "0.5"},  # 部分平仓
        {"coin": "BTC", "dir": "Close Long", "time": current_time - 7 * 86400000, "sz": "0.5"},  # 剩余平仓

        # 第2组：ETH 空头交易
        {"coin": "ETH", "dir": "Open Short", "time": current_time - 6 * 86400000, "sz": "10.0"},
        {"coin": "ETH", "dir": "Close Short", "time": current_time - 4 * 86400000, "sz": "10.0"},

        # 第3组：BTC 翻仓交易
        {"coin": "BTC", "dir": "Open Short", "time": current_time - 3 * 86400000, "sz": "2.0"},
        {"coin": "BTC", "dir": "Short > Long", "time": current_time - 1 * 86400000, "sz": "1.5"},  # 翻仓

        # 第4组：SOL 日内交易
        {"coin": "SOL", "dir": "Open Long", "time": current_time - 12 * 3600000, "sz": "100.0"},
        {"coin": "SOL", "dir": "Close Long", "time": current_time - 6 * 3600000, "sz": "100.0"},
    ]

    # 计算持仓时间
    result = calculator.calculate_hold_time_stats(fills)

测试用例

测试用例 1：简单开平仓

def test_simple_open_close():
    """测试用例 1: 简单开平仓"""
    calculator = HoldTimeCalculator()

    fills = [
        {"coin": "BTC", "dir": "Open Long", "time": 1000000, "sz": "1.0"},
        {"coin": "BTC", "dir": "Close Long", "time": 1000000 + 86400000, "sz": "1.0"},  # 1天后
    ]

    result = calculator.calculate_hold_time_stats(fills)

    assert abs(result['allTimeAverage'] - 1.0) < 0.01  # 应该是1天
    print("✅ 测试用例 1 通过：简单开平仓")

测试用例 2：部分平仓

def test_partial_close():
    """测试用例 2: 部分平仓"""
    calculator = HoldTimeCalculator()

    fills = [
        {"coin": "BTC", "dir": "Open Long", "time": 1000000, "sz": "2.0"},
        {"coin": "BTC", "dir": "Close Long", "time": 1000000 + 86400000, "sz": "1.0"},  # 1天后平一半
        {"coin": "BTC", "dir": "Close Long", "time": 1000000 + 2 * 86400000, "sz": "1.0"},  # 2天后平剩余
    ]

    result = calculator.calculate_hold_time_stats(fills)

    # 平均持仓 = (1天 + 2天) / 2 = 1.5天
    assert abs(result['allTimeAverage'] - 1.5) < 0.01
    print("✅ 测试用例 2 通过：部分平仓")

测试用例 3：多空分离

def test_long_short_separation():
    """测试用例 3: 多空分离配对"""
    calculator = HoldTimeCalculator()

    fills = [
        {"coin": "BTC", "dir": "Open Long", "time": 1000000, "sz": "1.0"},
        {"coin": "BTC", "dir": "Open Short", "time": 2000000, "sz": "2.0"},
        {"coin": "BTC", "dir": "Close Long", "time": 3000000, "sz": "1.0"},  # 应匹配第1笔
        {"coin": "BTC", "dir": "Close Short", "time": 4000000, "sz": "2.0"},  # 应匹配第2笔
    ]

    result = calculator.calculate_hold_time_stats(fills)

    # 两笔交易的平均持仓时间
    assert result['allTimeAverage'] > 0
    print("✅ 测试用例 3 通过：多空分离配对")

测试用例 4：翻仓交易

def test_flip_positions():
    """测试用例 4: 翻仓交易"""
    calculator = HoldTimeCalculator()

    fills = [
        {"coin": "BTC", "dir": "Open Short", "time": 1000000, "sz": "1.0"},
        {"coin": "BTC", "dir": "Short > Long", "time": 2000000, "sz": "2.0"},  # 翻仓
        {"coin": "BTC", "dir": "Close Long", "time": 3000000, "sz": "2.0"},
    ]

    result = calculator.calculate_hold_time_stats(fills)

    # 应该有2笔完成的持仓记录
    assert result['allTimeAverage'] > 0
    print("✅ 测试用例 4 通过：翻仓交易")

测试用例 5：FIFO 原则

def test_fifo_principle():
    """测试用例 5: FIFO 先进先出原则"""
    calculator = HoldTimeCalculator()

    fills = [
        {"coin": "BTC", "dir": "Open Long", "time": 1000000, "sz": "1.0"},  # 第1笔开仓
        {"coin": "BTC", "dir": "Open Long", "time": 2000000, "sz": "1.0"},  # 第2笔开仓
        {"coin": "BTC", "dir": "Close Long", "time": 3000000, "sz": "1.0"},  # 应先匹配第1笔
    ]

    # 由于 FIFO，应该匹配第1笔（持仓2天）
    # 而不是第2笔（持仓1天）
    # 但这个测试需要查看内部逻辑，这里只验证有结果
    result = calculator.calculate_hold_time_stats(fills)

    assert result['allTimeAverage'] > 0
    print("✅ 测试用例 5 通过：FIFO 原则")

运行所有测试

if __name__ == "__main__":
    print("开始运行持仓时间统计测试用例...\n")

    test_simple_open_close()
    test_partial_close()
    test_long_short_separation()
    test_flip_positions()
    test_fifo_principle()

    print("\n🎉 所有测试用例通过！")

算法优势

优势 说明
✅ 多空分离 正确区分多头和空头仓位
✅ 部分平仓 支持部分平仓的准确计算
✅ FIFO 原则 先进先出，确保配对准确
✅ 翻仓处理 正确处理翻仓交易
✅ 时间段统计 提供多个时间维度的分析

参考资料

文档生成时间: 2026-02-03
作者: Apex Calculator Team

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