冰山订单算法（iceberg-order）（难度：初级）

一、冰山订单算法详解

该算法最早出现在上世纪90年代中后期，伴随着电子交易平台的兴起。随着机构投资者和对冲基金对市场执行效率的要求逐步提高，传统的大额挂单模式暴露出信息泄露、交易滑点大、冲击成本高等问题。市场对一种可以有效隐藏大宗交易意图的自动化执行工具产生了强烈需求，冰山算法正是在此背景下被开发并广泛应用。

冰山算法的命名灵感来自自然界中的冰山现象：水面上只露出一小部分，其主体隐藏在水下。它的基本思想是将一个大额订单拆分为多个可见的小订单，每次仅挂出其中一部分，待当前批次完全成交后，再挂出下一批，直至全部成交完毕。整个过程中系统会持续监控订单状态，并根据剩余数量动态调整挂单数量。该策略不依赖于预测模型或历史成交量分布，属于成交驱动型逻辑。其主要优势包括：

降低市场滑点
隐藏实际交易规模
控制成交节奏

冰山算法适用于流动性较高的市场环境，在逐步完成交易的同时，降低市场对单一订单来源的识别概率。此外，冰山算法借助自动化系统持续刷新挂单，无需人工干预，提升了执行效率与一致性。不过，在市场波动剧烈或成交稀疏的时段，冰山算法的被动执行特性可能导致部分订单长时间未能成交，甚至错失有利价格窗口。尤其在当前高频交易与智能算法广泛应用的背景下，一些对手方程序可以通过分析订单簿中的刷新节奏、成交分布等信号，反推出潜在的冰山订单。这种“被识别”的风险可能削弱算法的原始隐蔽性设计，甚至引发市场的逆向反应。此外，冰山策略的整体表现高度依赖于系统稳定性与数据响应速度，一旦执行中断或交易系统延迟，可能导致订单管理混乱，影响最终成交效果。

二、天勤介绍

天勤平台概述

天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特点：

丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。

策略开发流程

环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制

三、冰山订单算法代码实现

本节将介绍一个冰山订单策略的实现方式，使用 Python 的 Pandas 库及天勤量化交易接口（TqSdk）。本策略适用于将大额订单在市场中分批执行，通过控制每次挂出的订单数量，实现隐藏真实交易意图、降低滑点和市场冲击的目标。

与通常的“定时挂单”不同，冰山策略的关键在于“成交后刷新”，即系统侦测到当前挂单已成交时，再立即挂出下一批小额订单，直至总目标手数成交完毕。本策略的实现基于天勤的目标持仓工具TargetPosTask，它会根据用户设定的参数，如最小挂单手数、最大挂单手数、目标总量、交易方向等，动态控制每一笔挂单，并且每批订单的手数在指定区间内随机生成。

策略设计思路与结构

冰山策略的实现流程可分为以下三步：

1. 参数设定与初始化

首先导入所需模块并建立交易连接。
用户需设定交易合约代码、目标总手数、每笔订单的最小与最大手数、交易方向（买入或卖出），以及下单方式（对价、挂价或函数形式的指定价）。
创建目标持仓工具TargetPosTask以管理交易执行。

2. 挂单刷新与成交确认

策略进入主循环后，持续等待行情更新
通过position.pos的变化，系统更新执行进度并打印成交信息。每次订单成交后，策略将自动判断是否达到目标手数，若未完成则继续下一轮挂单。

3. 策略终止与输出

当累计成交手数达到设定目标后，策略结束。

天勤实现算法代码

#!/usr/bin/env python
# coding=utf-8
__author__ = "Chaos"

from tqsdk import TqApi, TqAuth, TqKq, TargetPosTask

# === 用户参数 ===
SYMBOL = "SHFE.ag2506"      # 交易合约
TOTAL_VOLUME = 100          # 目标总手数
MIN_VOLUME = 1              # 每笔最小委托手数
MAX_VOLUME = 10             # 每笔最大委托手数
DIRECTION = "BUY"           # "BUY"为买入，"SELL"为卖出
ORDER_TYPE = "ACTIVE"       # 对价 "ACTIVE" / 挂价 "PASSIVE" / 指定价 lambda direction: 价格

# === 初始化 ===
api = TqApi(account=TqKq(), auth=TqAuth("快期账号", "快期密码"))
quote = api.get_quote(SYMBOL)

# 创建目标持仓任务
target_pos = TargetPosTask(
    api, SYMBOL,
    price=ORDER_TYPE,
    min_volume=MIN_VOLUME,
    max_volume=MAX_VOLUME
)

# 获取下单方式描述
order_type_str = (f"指定价 {ORDER_TYPE(DIRECTION)}" if callable(ORDER_TYPE) 
                 else str(ORDER_TYPE))

print(f"冰山算法启动，合约: {SYMBOL}，目标: {TOTAL_VOLUME}手，"
      f"每批: {MIN_VOLUME}-{MAX_VOLUME}手，方向: {DIRECTION}，下单方式: {order_type_str}")

try:
    # 获取初始持仓并设置目标
    pos = api.get_position(SYMBOL)
    start_net_pos = pos.pos_long - pos.pos_short
    target_volume = start_net_pos + (TOTAL_VOLUME if DIRECTION == "BUY" else -TOTAL_VOLUME)
    target_pos.set_target_volume(target_volume)
    
    last_progress = 0  # 记录上次进度

    while True:
        api.wait_update()
        pos = api.get_position(SYMBOL)
        net_pos = pos.pos_long - pos.pos_short
        progress = abs(net_pos - start_net_pos)
        
        # 当进度发生变化时打印
        if progress != last_progress:
            print(f"当前进度: {progress}/{TOTAL_VOLUME}")
            last_progress = progress
        
        # 检查是否完成
        if (DIRECTION == "BUY" and net_pos >= target_volume) or \
           (DIRECTION == "SELL" and net_pos <= target_volume):
            print(f"冰山算法完成")
            break

except Exception as e:
    print(f"算法执行异常: {e}")
finally:
    api.close()