3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
这是一个非常经典的量化交易场景。在QMT中实现Delta中性对冲策略,核心步骤如下:
组合Delta = Σ (单期权Delta × 持仓数量 × 合约乘数)。目标ETF数量 = -1 × 组合Delta),并与当前ETF持仓比较,进行买卖操作。以下是一个完整的Python策略示例。该策略会自动扫描账户中的期权持仓,计算总Delta敞口,并交易对应的ETF(如510050或510300)进行对冲。
# -*- coding: gbk -*-
import numpy as np
from scipy.stats import norm
import datetime
import time
def init(ContextInfo):
# 设置资金账号 (请修改为您的实际账号)
ContextInfo.accID = 'YOUR_ACCOUNT_ID'
ContextInfo.set_account(ContextInfo.accID)
# 设置对冲标的 (例如 510050.SH 华夏上证50ETF)
ContextInfo.etf_code = '510050.SH'
# 设置对冲阈值 (Delta敞口超过多少股时才进行交易,避免频繁交易手续费)
ContextInfo.hedge_threshold = 2000
# 设定无风险利率 (可调用API获取或设定固定值,如 3%)
ContextInfo.risk_free_rate = 0.03
print("Delta中性对冲策略初始化完成")
def handlebar(ContextInfo):
# 仅在最后一根K线(实时行情)运行
if not ContextInfo.is_last_bar():
return
# 1. 获取账户持仓信息
positions = get_trade_detail_data(ContextInfo.accID, 'STOCK', 'POSITION')
total_delta_exposure = 0.0
etf_position = 0
# 2. 遍历持仓,计算期权组合的 Delta
for pos in positions:
instrument_id = pos.m_strInstrumentID
# 如果是ETF本身,记录当前持仓量
if instrument_id == ContextInfo.etf_code:
etf_position = pos.m_nVolume
continue
# 判断是否为期权 (根据代码后缀或长度判断,这里假设是 .SHO 后缀)
if not instrument_id.endswith('.SHO'):
continue
# 获取期权详细信息
detail = ContextInfo.get_option_detail_data(instrument_id)
if not detail:
continue
# 提取BSM模型所需参数
S = get_underlying_price(ContextInfo, detail['OptUndlCode']) # 标的价格
K = detail['OptExercisePrice'] # 行权价
# 计算剩余时间 T (年化)
expire_date_str = str(detail['ExpireDate']) # 格式 YYYYMMDD
T = calculate_time_to_maturity(expire_date_str)
if T <= 0:
continue # 已过期
r = ContextInfo.risk_free_rate
# 获取隐含波动率 IV
# QMT提供了直接获取IV的接口,如果获取失败则给一个默认值或自行计算
sigma = ContextInfo.get_option_iv(instrument_id)
if sigma <= 0.001:
sigma = 0.2 # 默认波动率防止报错
opt_type = detail['optType'] # 'CALL' or 'PUT'
# 计算单张期权的 Delta
delta = calculate_bsm_delta(S, K, T, r, sigma, opt_type)
# 获取合约乘数 (通常ETF期权为10000)
multiplier = ContextInfo.get_contract_multiplier(instrument_id)
# 计算该持仓的 Delta 敞口 (Delta * 持仓量 * 乘数)
# 注意:如果是义务仓(卖方),持仓量通常显示为正,但Delta影响是反的?
# QMT中 m_nVolume 是正数,需要结合 m_nDirection 或 m_eSideFlag 判断方向
# 简单处理:买入持仓(权利方) Delta为正,卖出持仓(义务方) Delta为负
# 这里假设 pos.m_nVolume 为净持仓,正数为多头,负数为空头(如果API返回负数)
# 如果API返回的Volume永远为正,需结合 m_eSideFlag (0权利, 1义务)
position_sign = 1
if hasattr(pos, 'm_eSideFlag'):
# '1' 表示义务仓 (Short)
if str(pos.m_eSideFlag) == '1':
position_sign = -1
position_delta = delta * pos.m_nVolume * position_sign * multiplier
total_delta_exposure += position_delta
# print(f"合约: {instrument_id}, 类型: {opt_type}, Delta: {delta:.4f}, 敞口: {position_delta:.2f}")
print(f"当前期权组合总Delta敞口 (折合股数): {total_delta_exposure:.2f}")
print(f"当前ETF持仓: {etf_position}")
# 3. 计算对冲所需的 ETF 数量
# 为了保持中性,ETF Delta + Option Delta = 0
# ETF Delta = 1 * 股数
# 所以:目标ETF股数 = -1 * Option Delta
target_etf_position = -1 * total_delta_exposure
# 计算需要交易的数量
diff = target_etf_position - etf_position
# 4. 执行交易 (设置阈值防止噪音交易)
if abs(diff) > ContextInfo.hedge_threshold:
trade_vol = int(diff / 100) * 100 # 调整为100的整数倍(手)
if trade_vol != 0:
print(f"触发对冲交易,目标调整量: {trade_vol}")
order_shares(ContextInfo.etf_code, trade_vol, ContextInfo, ContextInfo.accID)
else:
print("Delta敞口在阈值内,无需对冲")
def get_underlying_price(ContextInfo, undl_code):
"""获取标的最新价格"""
data = ContextInfo.get_market_data_ex(
['close'], [undl_code], period='1d', count=1, subscribe=True
)
if undl_code in data and not data[undl_code].empty:
return data[undl_code].iloc[-1]['close']
return 0
def calculate_time_to_maturity(expire_date_str):
"""计算剩余到期时间(年化)"""
try:
exp_date = datetime.datetime.strptime(expire_date_str, "%Y%m%d")
now = datetime.datetime.now()
days = (exp_date - now).days
if days < 0:
return 0
# 加上当天剩余时间的小数部分会更精确,这里简化为天数/365
return (days + 0.5) / 365.0
except:
return 0
def calculate_bsm_delta(S, K, T, r, sigma, opt_type):
"""
计算BSM模型的Delta
S: 标的价格
K: 行权价
T: 剩余期限(年)
r: 无风险利率
sigma: 波动率
opt_type: 'CALL' or 'PUT'
"""
if T <= 0 or sigma <= 0 or S <= 0:
return 0
d1 = (np.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * np.sqrt(T))
if opt_type == 'CALL':
delta = norm.cdf(d1)
else: # PUT
delta = norm.cdf(d1) - 1
return delta
BSM Delta 计算 (calculate_bsm_delta):
bsm_price 和 bsm_iv,但没有直接提供 bsm_delta。因此,我们引入 scipy.stats.norm 库,根据标准的 Black-Scholes 公式手动计算 $d_1$ 和 $N(d_1)$。数据获取:
get_option_detail_data: 获取行权价、到期日、标的代码等静态数据。get_option_iv: QMT 提供的非常方便的接口,直接获取实时的隐含波动率,省去了自己反推波动率的麻烦。get_contract_multiplier: 获取合约乘数(通常ETF期权是10000),这对于计算总敞口至关重要。持仓方向处理:
get_trade_detail_data 返回的 POSITION 对象中,通常包含 m_eSideFlag(持仓方向)。position_sign 处理了这一逻辑。对冲逻辑:
目标ETF持仓 = -1 * 期权总Delta。hedge_threshold (如2000股),只有当偏差超过这个数值时才下单。这可以防止因为行情微小波动导致的频繁开仓和平仓,节省手续费。get_market_data_ex 可能取不到最新价格。scipy 和 numpy (QMT自带环境通常已包含)。Q1: 为什么需要乘以合约乘数?
A1: 期权的Delta通常表示一份期权对应1股标的的变动率,但一张ETF期权合约通常代表10,000份ETF。因此,一张Delta为0.5的Call期权,实际上相当于持有 0.5 * 10,000 = 5,000 股ETF的风险敞口。
Q2: 如何处理Gamma风险?
A2: Delta对冲只能抵消一阶风险。随着标的价格变化,Delta本身也会变化(这就是Gamma)。为了处理Gamma风险,通常需要更频繁地进行Delta对冲(动态对冲),或者买入/卖出其他期权来构建Gamma中性组合。
Q3: get_option_iv 获取不到数据怎么办?
A3: 如果期权流动性差,API可能返回0。代码中加入了 if sigma <= 0.001: sigma = 0.2 的保护逻辑。更严谨的做法是使用标的历史波动率作为替代,或者使用 ContextInfo.bsm_iv 根据期权最新成交价反推。