盈亏因子(Profit Factor)算法详解

盈亏因子(Profit Factor)算法详解

📋 目录

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

算法概述

什么是盈亏因子(Profit Factor)?

盈亏因子是衡量交易策略盈利能力的关键指标,表示总盈利与总亏损的比率

指标意义

  • Profit Factor > 1: 策略盈利(盈利大于亏损)
  • Profit Factor = 1: 策略盈亏平衡
  • Profit Factor < 1: 策略亏损(亏损大于盈利)
  • Profit Factor = "1000+": 只有盈利没有亏损

为什么需要盈亏因子?

盈亏因子是评估交易系统最直观的指标之一:

盈亏因子 评级 说明
< 1.0 ❌ 不可接受 亏损策略,需要改进
1.0 - 1.5 ⚠️ 勉强可接受 盈利微薄,风险较高
1.5 - 2.0 ✅ 良好 稳定盈利,风险可控
2.0 - 3.0 🌟 优秀 强盈利能力
> 3.0 🏆 卓越 顶级策略表现

核心概念

1. 已实现盈亏(Realized PnL)

来自已平仓交易的盈亏,记录在成交记录(fills)中。

closed_pnl = fill.get('closedPnl', 0)

示例

  • 开多 BTC @ $40,000
  • 平多 BTC @ $42,000
  • 已实现盈亏: +$2,000 ✅

2. 未实现盈亏(Unrealized PnL)

来自当前持仓的浮动盈亏,记录在持仓数据(positions)中。

unrealized_pnl = position.get('position', {}).get('unrealizedPnl', 0)

示例

  • 开多 ETH @ $2,000(仍持仓)
  • 当前价格 @ $2,100
  • 未实现盈亏: +$100 📊

3. 总盈利与总亏损

  • 总盈利(Total Gains): 所有正盈亏的累加
  • 总亏损(Total Losses): 所有负盈亏的绝对值累加

计算公式

核心公式

profit_factor = total_gains / total_losses

公式详解

盈亏因子 = 总盈利 / 总亏损

其中:
- 总盈利 = Σ(正盈亏) = 所有 PnL > 0 的交易累加
- 总亏损 = Σ|负盈亏| = 所有 PnL < 0 的交易绝对值累加

特殊情况处理

情况 总盈利 总亏损 返回值 说明
只有盈利 > 0 = 0 "1000+" 完美策略
只有亏损 = 0 > 0 0 需要改进
无交易 = 0 = 0 0 无数据
正常情况 > 0 > 0 gains/losses 正常计算

算法实现

核心函数:calculate_profit_factor()

代码位置: apex_fork.py:234-283

from typing import List, Dict, Union, Optional
from decimal import Decimal, getcontext

# 设置高精度计算(50位精度)
getcontext().prec = 50


def calculate_profit_factor(
    fills: List[Dict],
    asset_positions: Optional[List[Dict]] = None
) -> Union[float, str]:
    """
    计算盈亏因子(基于Apex Liquid Bot算法)

    盈亏因子 = 总盈利 / 总亏损
    该指标反映了交易策略的盈利能力,大于1表示盈利,小于1表示亏损

    参数:
        fills: 成交记录列表,包含'closedPnl'字段(已实现盈亏)
        asset_positions: 可选的当前持仓列表,包含'unrealizedPnl'字段(未实现盈亏)

    返回:
        - float: 盈亏因子数值
        - "1000+": 只有盈利没有亏损时
        - 0: 无交易记录时

    算法说明:
        1. 累计所有已实现盈亏(来自fills)
        2. 累计所有未实现盈亏(来自当前持仓)
        3. 计算总盈利和总亏损的比值
    """
    # 边界条件:无数据时返回0
    if not fills and not asset_positions:
        return 0

    # 初始化累计值(使用高精度Decimal)
    total_gains = Decimal('0')
    total_losses = Decimal('0')

    # ==================== 步骤1: 处理已实现盈亏 ====================
    for fill in fills:
        closed_pnl = Decimal(str(fill.get('closedPnl', 0)))

        if closed_pnl > 0:
            # 盈利交易:累加到总盈利
            total_gains += closed_pnl
        elif closed_pnl < 0:
            # 亏损交易:取绝对值后累加到总亏损
            total_losses += abs(closed_pnl)

    # ==================== 步骤2: 处理未实现盈亏 ====================
    if asset_positions:
        for position in asset_positions:
            unrealized_pnl = Decimal(str(
                position.get('position', {}).get('unrealizedPnl', 0)
            ))

            if unrealized_pnl > 0:
                # 浮盈:累加到总盈利
                total_gains += unrealized_pnl
            elif unrealized_pnl < 0:
                # 浮亏:取绝对值后累加到总亏损
                total_losses += abs(unrealized_pnl)

    # ==================== 步骤3: 计算盈亏因子 ====================
    if total_losses == 0:
        # 特殊情况:没有亏损
        return "1000+" if total_gains > 0 else 0

    # 正常计算:总盈利 / 总亏损
    profit_factor = total_gains / total_losses

    return float(profit_factor)

算法流程图

开始
  ↓
检查数据是否为空?
  ↓ 是
返回 0
  ↓ 否
初始化 total_gains = 0, total_losses = 0
  ↓
遍历成交记录(fills)
  ↓
对每条记录:
  - 获取 closedPnl
  - 如果 > 0 → total_gains += closedPnl
  - 如果 < 0 → total_losses += |closedPnl|
  ↓
是否有持仓数据?
  ↓ 是
遍历持仓(positions)
  ↓
对每个持仓:
  - 获取 unrealizedPnl
  - 如果 > 0 → total_gains += unrealizedPnl
  - 如果 < 0 → total_losses += |unrealizedPnl|
  ↓
total_losses == 0?
  ↓ 是
返回 "1000+"(如果 total_gains > 0)或 0
  ↓ 否
返回 total_gains / total_losses
  ↓
结束

应用场景

场景 1:评估交易策略表现

# 获取用户交易数据
fills = api_client.get_user_fills(user_address)
positions = api_client.get_user_asset_positions(user_address)

# 计算盈亏因子
profit_factor = calculate_profit_factor(fills, positions)

# 评估策略
if profit_factor == "1000+":
    print("🏆 完美策略:只盈不亏")
elif profit_factor > 3.0:
    print("🌟 卓越策略:强盈利能力")
elif profit_factor > 2.0:
    print("✅ 优秀策略:稳定盈利")
elif profit_factor > 1.5:
    print("✅ 良好策略:可持续盈利")
elif profit_factor > 1.0:
    print("⚠️ 勉强可接受:需要优化")
else:
    print("❌ 亏损策略:需要改进")

场景 2:策略对比

# 对比两个策略
strategy_a_pf = calculate_profit_factor(fills_a, positions_a)
strategy_b_pf = calculate_profit_factor(fills_b, positions_b)

print(f"策略 A 盈亏因子: {strategy_a_pf}")
print(f"策略 B 盈亏因子: {strategy_b_pf}")

if strategy_a_pf > strategy_b_pf:
    improvement = (strategy_a_pf - strategy_b_pf) / strategy_b_pf * 100
    print(f"策略 A 优于策略 B,提升 {improvement:.2f}%")

场景 3:风险管理

# 根据盈亏因子调整仓位
def adjust_position_size(profit_factor: float, base_size: float) -> float:
    """根据盈亏因子调整仓位大小"""
    if profit_factor > 3.0:
        return base_size * 1.5  # 策略表现优秀,增加仓位
    elif profit_factor > 2.0:
        return base_size * 1.2
    elif profit_factor > 1.5:
        return base_size
    elif profit_factor > 1.0:
        return base_size * 0.7  # 策略表现一般,减小仓位
    else:
        return base_size * 0.3  # 策略亏损,大幅减仓

完整代码示例

主程序示例

from typing import List, Dict, Union, Optional
from decimal import Decimal, getcontext

getcontext().prec = 50


class ProfitFactorCalculator:
    """盈亏因子计算器"""

    def calculate_profit_factor(
        self,
        fills: List[Dict],
        asset_positions: Optional[List[Dict]] = None
    ) -> Union[float, str]:
        """计算盈亏因子"""
        if not fills and not asset_positions:
            return 0

        total_gains = Decimal('0')
        total_losses = Decimal('0')

        # 处理已实现盈亏
        print("\n📊 已实现盈亏分析:")
        print("=" * 60)

        for i, fill in enumerate(fills, 1):
            closed_pnl = Decimal(str(fill.get('closedPnl', 0)))

            if closed_pnl > 0:
                total_gains += closed_pnl
                print(f"  交易 {i}: +${float(closed_pnl):,.2f} ✅")
            elif closed_pnl < 0:
                total_losses += abs(closed_pnl)
                print(f"  交易 {i}: -${float(abs(closed_pnl)):,.2f} ❌")

        # 处理未实现盈亏
        if asset_positions:
            print("\n📈 未实现盈亏分析:")
            print("=" * 60)

            for position in asset_positions:
                pos_data = position.get('position', {})
                coin = position.get('coin', 'Unknown')
                unrealized_pnl = Decimal(str(pos_data.get('unrealizedPnl', 0)))

                if unrealized_pnl > 0:
                    total_gains += unrealized_pnl
                    print(f"  {coin}: +${float(unrealized_pnl):,.2f} ✅ (浮盈)")
                elif unrealized_pnl < 0:
                    total_losses += abs(unrealized_pnl)
                    print(f"  {coin}: -${float(abs(unrealized_pnl)):,.2f} ❌ (浮亏)")

        # 计算结果
        print("\n" + "=" * 60)
        print(f"总盈利: ${float(total_gains):,.2f}")
        print(f"总亏损: ${float(total_losses):,.2f}")
        print("=" * 60)

        if total_losses == 0:
            result = "1000+" if total_gains > 0 else 0
            print(f"盈亏因子: {result} 🏆")
            return result

        profit_factor = total_gains / total_losses
        pf_value = float(profit_factor)

        # 评级
        if pf_value > 3.0:
            rating = "🏆 卓越"
        elif pf_value > 2.0:
            rating = "🌟 优秀"
        elif pf_value > 1.5:
            rating = "✅ 良好"
        elif pf_value > 1.0:
            rating = "⚠️ 勉强可接受"
        else:
            rating = "❌ 需要改进"

        print(f"盈亏因子: {pf_value:.4f} {rating}")

        return pf_value

    def get_detailed_analysis(
        self,
        fills: List[Dict],
        asset_positions: Optional[List[Dict]] = None
    ) -> Dict:
        """获取详细分析报告"""
        profit_factor = self.calculate_profit_factor(fills, asset_positions)

        # 统计交易次数
        winning_trades = sum(1 for f in fills if float(f.get('closedPnl', 0)) > 0)
        losing_trades = sum(1 for f in fills if float(f.get('closedPnl', 0)) < 0)
        total_trades = len(fills)

        # 计算总盈亏
        total_gains = sum(
            float(f.get('closedPnl', 0))
            for f in fills
            if float(f.get('closedPnl', 0)) > 0
        )
        total_losses = sum(
            abs(float(f.get('closedPnl', 0)))
            for f in fills
            if float(f.get('closedPnl', 0)) < 0
        )

        # 添加未实现盈亏
        if asset_positions:
            for position in asset_positions:
                unrealized = float(position.get('position', {}).get('unrealizedPnl', 0))
                if unrealized > 0:
                    total_gains += unrealized
                else:
                    total_losses += abs(unrealized)

        return {
            "profit_factor": profit_factor,
            "total_gains": total_gains,
            "total_losses": total_losses,
            "net_profit": total_gains - total_losses,
            "winning_trades": winning_trades,
            "losing_trades": losing_trades,
            "total_trades": total_trades,
            "win_rate": (winning_trades / total_trades * 100) if total_trades > 0 else 0
        }


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

    # 模拟交易数据
    fills = [
        {"closedPnl": "500"},    # 盈利 $500
        {"closedPnl": "-200"},   # 亏损 $200
        {"closedPnl": "300"},    # 盈利 $300
        {"closedPnl": "-100"},   # 亏损 $100
        {"closedPnl": "800"},    # 盈利 $800
        {"closedPnl": "-150"},   # 亏损 $150
    ]

    # 模拟持仓数据
    positions = [
        {
            "coin": "BTC",
            "position": {"unrealizedPnl": "200"}  # 浮盈 $200
        },
        {
            "coin": "ETH",
            "position": {"unrealizedPnl": "-50"}  # 浮亏 $50
        }
    ]

    # 计算盈亏因子
    profit_factor = calculator.calculate_profit_factor(fills, positions)

    # 获取详细分析
    print("\n" + "=" * 60)
    print("详细分析报告")
    print("=" * 60)

    analysis = calculator.get_detailed_analysis(fills, positions)

    for key, value in analysis.items():
        if isinstance(value, float):
            if "rate" in key:
                print(f"{key}: {value:.2f}%")
            elif "factor" in key:
                print(f"{key}: {value}")
            else:
                print(f"{key}: ${value:,.2f}")
        else:
            print(f"{key}: {value}")

运行结果示例

📊 已实现盈亏分析:
============================================================
  交易 1: +$500.00 ✅
  交易 2: -$200.00 ❌
  交易 3: +$300.00 ✅
  交易 4: -$100.00 ❌
  交易 5: +$800.00 ✅
  交易 6: -$150.00 ❌

📈 未实现盈亏分析:
============================================================
  BTC: +$200.00 ✅ (浮盈)
  ETH: -$50.00 ❌ (浮亏)

============================================================
总盈利: $1,800.00
总亏损: $500.00
============================================================
盈亏因子: 3.6000 🏆 卓越

============================================================
详细分析报告
============================================================
profit_factor: 3.6
total_gains: $1,800.00
total_losses: $500.00
net_profit: $1,300.00
winning_trades: 3
losing_trades: 3
total_trades: 6
win_rate: 50.00%

测试用例

测试用例 1:只有盈利

def test_only_profits():
    """测试用例 1: 只有盈利交易"""
    calculator = ProfitFactorCalculator()

    fills = [
        {"closedPnl": "100"},
        {"closedPnl": "200"},
        {"closedPnl": "300"},
    ]

    result = calculator.calculate_profit_factor(fills)

    assert result == "1000+", "只有盈利应返回 '1000+'"
    print("✅ 测试用例 1 通过:只有盈利")

测试用例 2:只有亏损

def test_only_losses():
    """测试用例 2: 只有亏损交易"""
    calculator = ProfitFactorCalculator()

    fills = [
        {"closedPnl": "-100"},
        {"closedPnl": "-200"},
        {"closedPnl": "-300"},
    ]

    result = calculator.calculate_profit_factor(fills)

    assert result == 0, "只有亏损应返回 0"
    print("✅ 测试用例 2 通过:只有亏损")

测试用例 3:盈亏混合

def test_mixed_trades():
    """测试用例 3: 盈亏混合交易"""
    calculator = ProfitFactorCalculator()

    fills = [
        {"closedPnl": "1000"},  # 盈利
        {"closedPnl": "-500"},  # 亏损
        {"closedPnl": "500"},   # 盈利
        {"closedPnl": "-200"},  # 亏损
    ]

    # 预期:总盈利 = 1500, 总亏损 = 700
    # 盈亏因子 = 1500 / 700 ≈ 2.14

    result = calculator.calculate_profit_factor(fills)

    expected = 1500.0 / 700.0
    assert abs(result - expected) < 0.01, f"预期 {expected},实际 {result}"
    print(f"✅ 测试用例 3 通过:盈亏因子 = {result:.4f}")

测试用例 4:包含未实现盈亏

def test_with_unrealized_pnl():
    """测试用例 4: 包含未实现盈亏"""
    calculator = ProfitFactorCalculator()

    fills = [
        {"closedPnl": "500"},   # 已实现盈利
        {"closedPnl": "-200"},  # 已实现亏损
    ]

    positions = [
        {"position": {"unrealizedPnl": "300"}},   # 未实现盈利
        {"position": {"unrealizedPnl": "-100"}},  # 未实现亏损
    ]

    # 预期:
    # 总盈利 = 500 + 300 = 800
    # 总亏损 = 200 + 100 = 300
    # 盈亏因子 = 800 / 300 ≈ 2.67

    result = calculator.calculate_profit_factor(fills, positions)

    expected = 800.0 / 300.0
    assert abs(result - expected) < 0.01, f"预期 {expected},实际 {result}"
    print(f"✅ 测试用例 4 通过:盈亏因子 = {result:.4f}")

测试用例 5:无交易数据

def test_no_data():
    """测试用例 5: 无交易数据"""
    calculator = ProfitFactorCalculator()

    fills = []
    positions = []

    result = calculator.calculate_profit_factor(fills, positions)

    assert result == 0, "无数据应返回 0"
    print("✅ 测试用例 5 通过:无交易数据")

测试用例 6:高精度计算

def test_high_precision():
    """测试用例 6: 高精度计算"""
    calculator = ProfitFactorCalculator()

    fills = [
        {"closedPnl": "123.456789"},
        {"closedPnl": "-45.123456"},
        {"closedPnl": "67.891234"},
    ]

    # 预期:
    # 总盈利 = 123.456789 + 67.891234 = 191.348023
    # 总亏损 = 45.123456
    # 盈亏因子 = 191.348023 / 45.123456 ≈ 4.2408

    result = calculator.calculate_profit_factor(fills)

    expected = 191.348023 / 45.123456
    assert abs(result - expected) < 0.0001, f"预期 {expected},实际 {result}"
    print(f"✅ 测试用例 6 通过:高精度盈亏因子 = {result:.6f}")

运行所有测试

if __name__ == "__main__":
    print("开始运行盈亏因子测试用例...\n")

    test_only_profits()
    test_only_losses()
    test_mixed_trades()
    test_with_unrealized_pnl()
    test_no_data()
    test_high_precision()

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

算法优势

优势 说明
直观易懂 简单的比率,易于理解和沟通
全面评估 同时考虑已实现和未实现盈亏
高精度计算 使用 Decimal 确保 50 位精度
特殊处理 正确处理无亏损、无盈利等边界情况
实时反映 包含当前持仓,反映最新表现

与其他指标对比

指标 优势 劣势 适用场景
盈亏因子 直观、易懂 不考虑交易频率 快速评估策略盈利能力
胜率 反映成功率 不考虑盈亏金额 评估交易准确性
夏普比率 考虑风险 计算复杂 风险调整后的收益评估
最大回撤 反映风险 不反映盈利 风险控制评估

参考资料


版本历史

  • v1.0 (2026-02-03): 初始版本,完整实现盈亏因子算法
  • 支持已实现和未实现盈亏
  • 高精度计算(50位精度)
  • 完善的边界条件处理

文档生成时间: 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