本文主要记录本人学习量化交易轨迹，如有好的学习路径请留言，在此感谢
Author:kim 2023-03-18
本文使用Python来实现期权量化交易策略。Python有很多量化交易框架，比如vnpy、rqalpha等，可以方便地进行策略开发、回测和实盘交易。下面是一个使用vnpy框架实现的期权义务仓策略示例代码，其中包括风险管理、止盈止损、错误处理和日志记录等功能。该策略只做义务仓，不做权利仓，至于原因嘛，权利仓双向收费，义务仓单向收费。
一、期权策略类（OptionStrategy）
from vnpy.app.option_master import OptionMasterApp
from vnpy.trader.constant import Direction, Offset, OptionType
from vnpy.trader.object import TradeData, OrderData
from vnpy.trader.utility import round_to
import numpy as np
import pandas as pd
import scipy.stats as si
class OptionStrategy:
"""
可以使用Python来实现期权量化交易策略。Python有很多量化交易框架，比如vnpy、rqalpha等，可以方便地进行策略开发、回测和实盘交易。下面是一个使用vnpy框架实现的期权义务仓策略示例代码，其中包括风险管理、止盈止损、错误处理和日志记录等功能。该策略只做义务仓，不做权利仓，可以根据实际需求进行修改。
"""
option_master: OptionMasterApp
option_vt_symbol: str
underlying_vt_symbol: str
underlying_size: float
option_size: float
max_pos: int
min_volume: float
max_volume: float
min_price: float
max_price: float
stop_loss: float
take_profit: float
trailing_stop: float
trailing_percent: float
trailing_buffer: float
trailing_high: float
trailing_low: float
trailing_order: OrderData
last_price: float
last_volume: float
last_direction: Direction
last_offset: Offset
parameters = [
"option_vt_symbol",
"underlying_vt_symbol",
"underlying_size",
"option_size",
"max_pos",
"min_volume",
"max_volume",
"min_price",
"max_price",
"stop_loss",
"take_profit",
"trailing_stop",
"trailing_percent",
"trailing_buffer"
]
variables = [
"last_price",
"last_volume",
"last_direction",
"last_offset"
]
def __init__(
self,
option_master: OptionMasterApp,
setting: dict
):
""""""
self.option_master = option_master
def on_tick(self, tick):
"""
当行情更新时，会调用该函数进行处理。如果当前行情不是标的物行情，则直接返回。如果当前开启了追踪止损，则会调用update_trailing_stop()函数进行处理。如果当前有追踪订单，则会调用update_trailing_order()函数进行处理。
"""
if tick.vt_symbol != self.underlying_vt_symbol:
return
self.last_price = tick.last_price
if self.trailing_stop:
self.update_trailing_stop()
if self.trailing_order:
self.update_trailing_order()
def on_trade(self, trade: TradeData):
"""
当成交回报更新时，会调用该函数进行处理。如果当前成交回报不是期权成交回报，则直接返回。如果当前开启了追踪止损，则会调用update_trailing_stop()函数进行处理。如果当前有追踪订单，则会调用update_trailing_order()函数进行处理
"""
if trade.vt_symbol != self.option_vt_symbol:
return
self.last_volume = trade.volume
self.last_direction = trade.direction
self.last_offset = trade.offset
if self.trailing_stop:
self.update_trailing_stop()
if self.trailing_order:
self.update_trailing_order()
def on_order(self, order: OrderData):
"""
当委托回报更新时，会调用该函数进行处理。如果当前委托回报不是期权委托回报，则直接返回。如果当前委托是开仓委托，则会增加当前持仓。如果当前委托是平仓委托，则会减少当前持仓。如果当前有追踪订单，则会调用update_trailing_order()函数进行处理。
"""
if order.vt_symbol != self.option_vt_symbol:
return
if order.offset == Offset.OPEN:
self.max_pos += order.volume
else:
self.max_pos -= order.volume
if self.trailing_order:
self.update_trailing_order()
def update_trailing_stop(self):
"""
根据追踪止损策略更新追踪止损价。如果当前持仓方向为多头，则计算出新的止损价和止损价下限，并更新到实例变量中。如果当前持仓方向为空头，则计算出新的止损价和止损价上限，并更新到实例变量中。如果当前价格已经低于止损价，则发送追踪订单。
"""
if self.last_direction == Direction.LONG:
stop_price = self.last_price * (1 - self.trailing_stop)
if stop_price > self.trailing_high:
self.trailing_high = stop_price
self.trailing_low = stop_price * (1 - self.trailing_buffer)
else:
stop_price = self.last_price * (1 + self.trailing_stop)
if stop_price < self.trailing_low:
self.trailing_low = stop_price
self.trailing_high = stop_price * (1 + self.trailing_buffer)
if self.last_direction == Direction.LONG:
if self.last_price <= self.trailing_high:
self.send_trailing_order(stop_price)
else:
if self.last_price >= self.trailing_low:
self.send_trailing_order(stop_price)
def send_trailing_order(self, stop_price: float):
"""
发送追踪订单。如果当前没有追踪订单，则会根据当前持仓方向发送一个新的订单。如果当前有追踪订单，则会根据当前持仓方向和止损价更新订单。如果当前持仓方向为多头，则会发送一个平空订单，否则会发送一个平多订单。
"""
if self.trailing_order:
if self.last_direction == Direction.LONG:
if stop_price > self.trailing_order.price:
self.option_master.cancel_order(self.trailing_order)
self.trailing_order = self.send_order(
Direction.SHORT,
Offset.CLOSE,
stop_price,
self.last_volume
)
else:
if stop_price < self.trailing_order.price:
self.option_master.cancel_order(self.trailing_order)
self.trailing_order
def send_order(
self,
direction: Direction,
offset: Offset,
price: float,
volume: float
):
"""
根据传入的参数发送一个期权订单。如果当前持仓方向为多头，则检查是否超过最大持仓量，如果超过则不发送订单。如果当前持仓方向为空头，则检查是否超过最大持仓量，如果超过则不发送订单。如果价格超出最大价格或最小价格范围，则不发送订单。如果数量超出最小数量或最大数量范围，则不发送订单。如果订单发送成功，则将追踪订单实例变量设置为None，并返回订单实例。如果订单发送失败，则返回None。
"""
if direction == Direction.LONG:
if self.max_pos - volume < -self.max_volume:
return None
else:
if self.max_pos + volume > self.max_volume:
return None
if price < self.min_price or price > self.max_price:
return None
if volume < self.min_volume or volume > self.max_volume:
return None
order = self.option_master.send_order(
self.option_vt_symbol,
direction,
offset,
price,
volume
)
if order:
self.trailing_order = None
return order
def update_trailing_order(self):
"""
根据追踪止损策略更新追踪订单。如果当前持仓方向为多头，则计算出新的止损价和止损价下限，并更新到实例变量中。如果当前持仓方向为空头，则计算出新的止损价和止损价上限，并更新到实例变量中。如果当前价格已经低于止损价，则发送追踪订单。
"""
if self.last_direction == Direction.LONG:
stop_price = self.last_price * (1 - self.trailing_stop)
if stop_price > self.trailing_high:
self.trailing_high = stop_price
self.trailing_low = stop_price * (1 - self.trailing_buffer)
else:
stop_price = self.last_price * (1 + self.trailing_stop)
if stop_price < self.trailing_low:
self.trailing_low = stop_price
self.trailing_high = stop_price * (1 + self.trailing_buffer)
if self.last_direction == Direction.LONG:
if self.last_price <= self.trailing_high:
self.send_trailing_order(stop_price)
else:
if self.last_price >= self.trailing_low:
self.send_trailing_order(stop_price)
def send_trailing_order(self, stop_price: float):
"""
发送追踪订单。如果当前没有追踪订单，则会根据当前持仓方向发送一个新的订单。如果当前有追踪订单，则会根据当前持仓方向和止损价更新订单。如果当前持仓方向为多头，则会发送一个平空订单，否则会发送一个平多订单。
"""
if self.trailing_order:
if self.last_direction == Direction.LONG:
if stop_price > self.trailing_order.price:
self.option_master.cancel_order(self.trailing_order)
self.trailing_order = self.send_order(
Direction.SHORT,
Offset.CLOSE,
stop_price,
self.last_volume
)
else:
if stop_price < self.trailing_order.price:
self.option_master.cancel_order(self.trailing_order)
self.trailing_order = self.send_order(
Direction.LONG,
Offset.CLOSE,
stop_price,
self.last_volume
)
else:
if self.last_direction == Direction.LONG:
self.trailing_order = self.send_order(
Direction.SHORT,
Offset.CLOSE,
stop_price,
self.last_volume
)
else:
self.trailing_order = self.send_order(
Direction.LONG,
Offset.CLOSE,
stop_price,
self.last_volume
)基于期权价格和Greeks指标的 期权交易策略.
def calc_option_price(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type):
"""
根据 Black-Scholes model计算出期权价格.
"""
d1 = (np.log(spot_price / strike_price) + (risk_free_rate + 0.5 * volatility ** 2) * time_to_maturity) / (volatility * np.sqrt(time_to_maturity))
d2 = d1 - volatility * np.sqrt(time_to_maturity)
if option_type == "call":
option_price = spot_price * si.norm.cdf(d1) - strike_price * np.exp(-risk_free_rate * time_to_maturity) * si.norm.cdf(d2)
elif option_type == "put":
option_price = strike_price * np.exp(-risk_free_rate * time_to_maturity) * si.norm.cdf(-d2) - spot_price * si.norm.cdf(-d1)
else:
raise ValueError("Invalid option type")
return option_price
def calc_option_greeks(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type):
"""
使用Black-Scholes model模型计算出Greaks指标.
"""
d1 = (np.log(spot_price / strike_price) + (risk_free_rate + 0.5 * volatility ** 2) * time_to_maturity) / (volatility * np.sqrt(time_to_maturity))
d2 = d1 - volatility * np.sqrt(time_to_maturity)
delta = si.norm.cdf(d1) if option_type == "call" else si.norm.cdf(d1) - 1
gamma = si.norm.pdf(d1) / (spot_price * volatility * np.sqrt(time_to_maturity))
vega = spot_price * si.norm.pdf(d1) * np.sqrt(time_to_maturity) / 100
theta = (-spot_price * si.norm.pdf(d1) * volatility / (2 * np.sqrt(time_to_maturity)) - risk_free_rate * strike_price * np.exp(-risk_free_rate * time_to_maturity) * si.norm.cdf(d2)) / 365
greeks = {"delta": delta, "gamma": gamma, "vega": vega, "theta": theta}
return greeks
def option_trading_base(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type):
"""
根据期权指标Delta、Gamma、Theta和Vega给出的信号.
"""
d1 = (np.log(spot_price / strike_price) + (risk_free_rate + 0.5 * volatility ** 2) * time_to_maturity) / (volatility * np.sqrt(time_to_maturity))
d2 = d1 - volatility * np.sqrt(time_to_maturity)
delta = si.norm.cdf(d1)
gamma = si.norm.pdf(d1) / (spot_price * volatility * np.sqrt(time_to_maturity))
vega = spot_price * si.norm.pdf(d1) * np.sqrt(time_to_maturity)
theta = -(spot_price * si.norm.pdf(d1) * volatility) / (2 * np.sqrt(time_to_maturity)) - risk_free_rate * strike_price * np.exp(-risk_free_rate * time_to_maturity) * si.norm.cdf(d2)
if option_type == "call":
if delta > 0 and gamma > 0 and vega > 0 and theta < 0:
signal = "认购卖出sell"
elif delta < 0 and gamma < 0 and vega < 0 and theta > 0:
signal = "认购买入buy"
else:
signal = "持有标的hold"
elif option_type == "put":
if delta < 0 and gamma > 0 and vega > 0 and theta < 0:
signal = "认沽卖出sell"
elif delta > 0 and gamma < 0 and vega < 0 and theta > 0:
signal = "认沽买入buy"
else:
signal = "持有标的hold"
else:
raise ValueError("Invalid option type")
return signal
def option_trading_strategy(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type, threshold):
"""
基于期权价格和Greeks指标的 期权交易策略.
"""
# 计算期权价格和Greeks指数
option_price = calc_option_price(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type)
delta = calc_option_greeks(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type)["delta"]
gamma = calc_option_greeks(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type)["gamma"]
vega = calc_option_greeks(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type)["vega"]
theta = calc_option_greeks(spot_price, strike_price, time_to_maturity, risk_free_rate, volatility, option_type)["theta"]
# 给出交易信号
if option_type == "call":
if option_price > threshold and delta > 0 and gamma > 0 and vega > 0 and theta < 0:
signal = "认购买入"
elif option_price < threshold and delta < 0 and gamma < 0 and vega < 0 and theta > 0:
signal = "认购卖出"
else:
signal = "持有标的"
elif option_type == "put":
if option_price > threshold and delta < 0 and gamma > 0 and vega > 0 and theta < 0:
signal = "认沽买入"
elif option_price < threshold and delta > 0 and gamma < 0 and vega < 0 and theta > 0:
signal = "认沽卖出"
else:
signal = "持有标的"
else:
raise ValueError("无效的期权类型")
return signal
弄懂上面代码需要了解以下几个概念
Greeks：在期权交易中，Greeks是一组指标，用于衡量期权价格对不同因素的敏感度。这些因素包括标的资产价格、波动率、时间和利率。Greeks指标包括Delta、Gamma、Theta和Vega。Delta衡量期权价格对标的资产价格的敏感度，Gamma衡量Delta的变化率，Theta衡量时间对期权价格的影响，Vega衡量波动率对期权价格的影响。
波动率是指资产价格的波动程度，是衡量资产风险的重要指标。波动率越高，资产价格波动的幅度就越大，风险也就越高。在交易中，可以使用波动率来计算仓位大小，以控制风险和最大化利润。具体计算方法可以参考以下公式：
$ Position Size = \frac{Account Value \times Risk per Trade}{Stop Loss Distance \times Volatility} $其中，Account Value 表示账户价值，Risk per Trade 表示每笔交易的风险，Stop Loss Distance 表示止损距离，Volatility 表示波动率。
风险价值是指在一定置信水平下，资产价格可能出现的最大损失，是衡量资产风险的另一个重要指标。风险价值越高，资产价格可能出现的最大损失就越大，风险也就越高。在交易中，可以使用风险价值来计算仓位大小，以控制风险和最大化利润。具体计算方法可以参考以下公式：
$ Position Size = \frac{Account Value \times Risk per Trade}{Stop Loss Distance \times Value at Risk} $其中，Account Value 表示账户价值，Risk per Trade 表示每笔交易的风险，Stop Loss Distance 表示止损距离，Value at Risk 表示风险价值。
卡方分布是一种概率分布，用于描述随机变量的分布情况。在交易中，可以使用卡方分布来计算仓位大小，以控制风险和最大化利润。具体计算方法可以参考以下公式：
$ Position Size = \frac{Account Value \times Risk per二、控制仓位风险管理策略篇
主要从波动率、置信水平、固定比率、固定点数来动态控制仓位
import numpy as np
import scipy.stats as stats
import talib
def calc_position_size(account_value, risk_per_trade, stop_loss_distance, volatility):
"""
本函数实现基于波动率计算控制仓位大小从而实现风控.
"""
position_size = account_value * risk_per_trade / (stop_loss_distance * volatility)
return position_size
def calc_position_size(account_value, risk_per_trade, stop_loss_distance, value_at_risk):
"""
本函数实现基于风险价值计算控制仓位大小从而实现风控.
"""
position_size = account_value * risk_per_trade / (stop_loss_distance * value_at_risk)
return position_size
def calc_position_size(account_value, risk_per_trade, stop_loss_distance, confidence_level):
"""
本函数实现基于置信区间计算控制仓位大小从而实现风控
"""
df = 1
scale = stop_loss_distance / np.sqrt(df)
x = stats.chi2.ppf(confidence_level, df)
value_at_risk = scale * np.sqrt(x)
position_size = account_value * risk_per_trade / (stop_loss_distance * value_at_risk)
return position_size
def calc_position_size(account_value, risk_per_trade, stop_loss_distance, fixed_ratio):
"""
本函数实现基于固定比率计算控制仓位大小从而实现风控.
Account Value 表示账户价值，Risk per Trade 表示每笔交易的风险，Stop Loss Distance 表示止损距离，Fixed Ratio 表示固定的比率。
"""
position_size = account_value * risk_per_trade / stop_loss_distance
position_size *= fixed_ratio
return position_size
def calc_position_size(account_value, risk_per_trade, stop_loss_distance, fixed_point):
"""
本函数实现基于固定点数计算控制仓位大小从而实现风控.
"""
position_size = fixed_point / stop_loss_distance
return position_size三、止损策略篇
以下函数calc_ma_stop_loss 函数实现了基于移动平均线的止损计算方法，calc_bollinger_band_stop_loss 函数实现了基于布林带的止损计算方法，calc_atr_stop_loss 函数实现了基于 ATR 的止损计算方法。在使用时，可以根据具体的需求选择相应的方法进行计算。
def calc_ma_stop_loss(prices, window):
"""
calc_ma_stop_loss 函数实现了基于移动平均线的止损计算方法.
"""
ma = talib.SMA(prices, window)
stop_loss = ma[-1]
return stop_loss
def calc_bollinger_band_stop_loss(prices, window, k):
"""
calc_bollinger_band_stop_loss 函数实现了基于布林带的止损计算方法
"""
upper, middle, lower = talib.BBANDS(prices, window, k, k)
stop_loss = lower[-1]
return stop_loss
def calc_atr_stop_loss(highs, lows, closes, window, k):
"""
calc_atr_stop_loss 函数实现了基于 ATR 的止损计算方法.
"""
atr = talib.ATR(highs, lows, closes, window)
stop_loss = closes[-1] - k * atr[-1]
return stop_loss四、止盈策略篇
常见的止盈策略包括以下几种：
1、固定止盈：在达到一定的盈利目标后，直接平仓离场，以保护利润。以下代码是基于传统技术实现的止盈策略
def calc_ma_take_profit(prices, window):
"""
基于移动均线计算的止盈方法.
"""
ma = talib.SMA(prices, window)
take_profit = ma[-1]
return take_profit
def calc_bollinger_band_take_profit(prices, window, k):
"""
基于布林带计算的止盈方法.
"""
upper, middle, lower = talib.BBANDS(prices, window, k, k)
take_profit = upper[-1]
return take_profit
def calc_atr_take_profit(highs, lows, closes, window, k):
"""
基于 ATR 的止盈计算方法
"""
atr = talib.ATR(highs, lows, closes, window)
take_profit = closes[-1] + k * atr[-1]
return take_profit2、移动止盈：根据市场行情和交易策略动态调整止盈价，以最大化利润
def calc_ma_dynamic_take_profit(prices, window, k):
"""
基于移动均线计算的动态止盈方法.
"""
ma = talib.SMA(prices, window)
diff = prices[-1] - ma[-1]
take_profit = prices[-1] + k * diff
return take_profit其中，k 参数表示动态调整的比例系数，可以根据具体的需求进行调整。在使用时，可以根据市场行情和交易策略动态调整止盈目标，以最大化利润。
五、策略回测篇
还是基于vnpy的回测
from vnpy.app.cta_strategy.backtesting import BacktestingEngine, OptimizationSetting
from vnpy.app.cta_strategy.strategies.ma_strategy import OptionStrategy
# 创建回测引擎
engine = BacktestingEngine()
# 设置回测参数
engine.set_parameters(
#上证50
vt_symbol="510050",
interval="1m",
start=datetime(2010, 1, 1),
end=datetime(2020, 12, 31),
rate=0.3 / 10000,
slippage=0.2,
size=300,
pricetick=0.2,
capital=1_000_000,
)
# 添加策略
engine.add_strategy(OptionStrategy, {})
#engine.add_strategy(MAStrategy, {})
# 运行回测
engine.run_backtesting()
# 输出回测结果
engine.output_performance()
# 绘制回测图表
engine.show_chart()