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

一、冰山订单算法详解

冰山订单算法最早出现在上世纪90年代中后期，伴随着电子交易平台的兴起。随着机构投资者和对冲基金对市场执行效率的要求逐步提高，传统的大额挂单模式暴露出信息泄露、交易滑点大、冲击成本高等问题。

市场对一种可以有效隐藏大宗交易意图的自动化执行工具产生了强烈需求，冰山算法正是在此背景下被开发并广泛应用。

冰山算法的命名灵感来自自然界中的冰山现象：水面上只露出一小部分，其主体隐藏在水下。它的基本思想是将一个大额订单拆分为多个可见的小订单，每次仅挂出其中一部分，待当前批次完全成交后，再挂出下一批，直至全部成交完毕。

整个过程中系统会持续监控订单状态，并根据剩余数量动态调整挂单数量。该策略不依赖于预测模型或历史成交量分布，属于成交驱动型逻辑。其主要优势包括：

（1）降低市场滑点

（2）隐藏实际交易规模

（3）控制成交节奏

冰山算法适用于流动性较高的市场环境，在逐步完成交易的同时，降低市场对单一订单来源的识别概率。 此外，冰山算法借助自动化系统持续刷新挂单，无需人工干预，提升了执行效率与一致性。

不过，在市场波动剧烈或成交稀疏的时段，冰山算法的被动执行特性可能导致部分订单长时间未能成交，甚至错失有利价格窗口。 尤其在当前高频交易与智能算法广泛应用的背景下，一些对手方程序可以通过分析订单簿中的刷新节奏、成交分布等信号，反推出潜在的冰山订单。 这种“被识别”的风险可能削弱算法的原始隐蔽性设计，甚至引发市场的逆向反应。

此外，冰山策略的整体表现高度依赖于系统稳定性与数据响应速度，一旦执行中断或交易系统延迟，可能导致订单管理混乱，影响最终成交效果。

冰山订单的量化模型：为了更准确地探测冰山订单，可以使用历史数据建立概率模型。例如，通过分析买卖价差、买单量对卖单量的比值等数据，可以计算出冰山订单存在的概率。这个模型可以帮助交易者在实时交易中更好地判断冰山订单的存在。

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

使用 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()