如何在 JoinQuant 中实现备兑看涨期权（Covered Call）策略？

这是一个基于聚宽（JoinQuant）API 实现的 备兑看涨期权（Covered Call） 策略。

策略原理

备兑看涨期权 (Covered Call) 是一种增强收益的策略，适用于预期标的资产价格温和上涨或横盘震荡的市场环境。

1. 持有标的：买入并持有现货（如 50ETF）。
2. 卖出认购：卖出（Short）同等数量的认购期权（Call Option）。
3. 收益来源：
   • 标的资产的上涨收益（但在行权价处封顶）。
   • 卖出期权获得的权利金（Premium）。

策略逻辑

1. 标的：华夏上证50ETF (510050.XSHG)。
2. 交易频率：按日回测，每日收盘前检查。
3. 开仓逻辑：
   • 如果未持有 50ETF，买入 10000 股（对应一张期权合约的单位）。
   • 如果未持有卖方期权头寸，选择下个月到期的、行权价略高于当前价格（虚值 OTM）的认购期权进行卖出开仓。
4. 换仓/平仓逻辑：
   • 如果持有的期权临近到期日（例如剩余天数小于 7 天），平仓该期权，并卖出新的下月期权（移仓换月）。

策略代码

# -*- coding: utf-8 -*-
import numpy as np
import pandas as pd
import datetime

def initialize(context):
    """
    初始化函数
    """
    # 设定基准为 50ETF
    set_benchmark('510050.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # --- 策略参数设置 ---
    # 标的资产：50ETF
    g.underlying = '510050.XSHG'
    # 每次操作的现货数量（50ETF期权一张合约对应10000份）
    g.share_amount = 10000
    # 移仓阈值：距离到期日少于多少天时进行移仓
    g.days_to_expire_threshold = 7
    # 虚值程度：选择行权价高于当前价格百分之多少的期权 (例如 1.02 代表 2% OTM)
    g.otm_ratio = 1.02
    
    # 每天收盘前 14:50 运行策略
    run_daily(trade_logic, '14:50')

def trade_logic(context):
    """
    每日交易主逻辑
    """
    # 1. 确保持有标的资产 (Covered)
    check_and_buy_underlying(context)
    
    # 2. 管理期权头寸 (Call)
    manage_option_position(context)

def check_and_buy_underlying(context):
    """
    检查并买入标的资产
    """
    # 获取当前持仓
    long_positions = context.portfolio.long_positions
    
    # 如果没有持有 50ETF 或者持仓数量不足，则补足到 g.share_amount
    if g.underlying not in long_positions or long_positions[g.underlying].total_amount < g.share_amount:
        log.info("标的资产不足，进行买入: %s" % g.underlying)
        order_target(g.underlying, g.share_amount)

def manage_option_position(context):
    """
    管理期权空头头寸：开仓、平仓、移仓
    """
    # 获取当前持有的空头期权（Short Call）
    short_positions = context.portfolio.short_positions
    current_short_option = None
    
    # 遍历空头持仓，找到我们卖出的期权
    for security in short_positions:
        # 简单判断：如果是期权（通常代码较长或通过 get_security_info 判断）
        # 这里假设账户里只有我们策略卖出的那个期权
        info = get_security_info(security)
        if info.type == 'option' or info.type == 'stock_option': # 兼容不同环境的类型名称
            current_short_option = security
            break
    
    # 获取标的当前价格
    underlying_price = get_current_data()[g.underlying].last_price
    
    # --- 情况 A: 没有期权持仓，开新仓 ---
    if current_short_option is None:
        log.info("当前无期权持仓，准备卖出开仓")
        open_new_option(context, underlying_price)
        return

    # --- 情况 B: 有期权持仓，检查是否需要移仓 ---
    # 获取期权到期日
    info = get_security_info(current_short_option)
    end_date = info.end_date
    
    # 计算距离到期日的天数
    days_to_expire = (end_date - context.current_dt.date()).days
    
    # 如果临近到期，平仓旧的，开新的
    if days_to_expire <= g.days_to_expire_threshold:
        log.info("期权 %s 临近到期（剩余%d天），进行平仓换月" % (current_short_option, days_to_expire))
        
        # 平仓：买入平仓 (Close Short Position)
        order_target(current_short_option, 0, side='short')
        
        # 开新仓
        open_new_option(context, underlying_price)
    else:
        log.info("持有期权 %s，距离到期还有 %d 天，继续持有" % (current_short_option, days_to_expire))

def open_new_option(context, current_price):
    """
    选择并卖出新的期权合约
    """
    # 1. 获取所有期权列表
    # 注意：get_all_securities(['options']) 获取的是所有期权，需要筛选
    try:
        # 获取当天上市的所有期权
        all_options = get_all_securities(types=['option'], date=context.current_dt).index.tolist()
    except:
        # 兼容旧版API或不同数据源配置
        # 如果 types=['option'] 报错，尝试获取所有标的再筛选，或者直接查询数据库
        # 这里使用一个假设的获取方式，实际建议使用 query(opt.OPT_CONTRACT_INFO) 如果可用
        # 为保证回测运行，这里简化处理，假设能取到列表。
        # 在聚宽回测环境中，通常需要自行筛选 50ETF 的期权
        log.warn("获取期权列表失败，请检查API权限或数据源")
        return

    # 2. 筛选 50ETF 的认购期权 (Call)
    target_options = []
    for code in all_options:
        info = get_security_info(code)
        name = info.display_name
        
        # 筛选逻辑：
        # 1. 标的必须是 50ETF (通常名字里包含 50ETF)
        # 2. 必须是认购期权 (名字包含 '购')
        # 3. 到期日必须在未来 (排除已过期的)
        if '50ETF' in name and '购' in name and info.end_date > context.current_dt.date():
            target_options.append(code)
    
    if not target_options:
        log.info("未找到合适的期权合约")
        return

    # 3. 选择到期日：选择下个月到期的（避免选到当月即将到期的）
    # 简单逻辑：按到期日排序，选择到期日距离现在大于 20 天的最近一个
    target_options_info = []
    for code in target_options:
        info = get_security_info(code)
        days = (info.end_date - context.current_dt.date()).days
        if days > 20: 
            target_options_info.append({'code': code, 'end_date': info.end_date, 'name': info.display_name})
    
    if not target_options_info:
        log.info("未找到符合日期要求的期权合约")
        return
        
    # 按到期日排序，取最近的一批（即下月合约）
    target_options_info.sort(key=lambda x: x['end_date'])
    nearest_date = target_options_info[0]['end_date']
    # 筛选出所有该日期到期的合约
    candidates = [x for x in target_options_info if x['end_date'] == nearest_date]
    
    # 4. 选择行权价：选择虚值期权 (行权价 > 当前价格 * 1.02)
    # 需要从名称中解析行权价，例如 "50ETF购9月3.000" -> 3.000
    final_target = None
    min_diff = 9999.0
    
    target_strike = current_price * g.otm_ratio
    
    for item in candidates:
        name = item['name']
        try:
            # 解析行权价：通常在名称的最后部分，或者通过 get_option_contract_info 获取
            # 这里尝试简单的字符串解析，假设行权价在最后
            # 示例名称: "50ETF购12月2.500"
            import re
            # 匹配末尾的浮点数
            match = re.search(r'(\d+\.\d+)$', name)
            if match:
                strike_price = float(match.group(1))
                
                # 我们需要找行权价 >= 目标行权价，且最接近的那个
                if strike_price >= target_strike:
                    diff = strike_price - target_strike
                    if diff < min_diff:
                        min_diff = diff
                        final_target = item['code']
        except Exception as e:
            continue

    # 5. 执行交易
    if final_target:
        log.info("选定目标期权: %s, 标的价格: %.3f, 目标行权价: >%.3f" % (final_target, current_price, target_strike))
        # 卖出开仓 1 张 (对应 10000 股标的)
        # 注意：side='short' 表示开空单
        order(final_target, 1, side='short')
    else:
        log.info("未找到合适的行权价")

关键点说明

1. 数据获取 (get_all_securities)：
   • 代码中使用了 get_all_securities(types=['option']) 来获取期权列表。在实盘或特定回测环境中，可能需要根据具体的数据库表（如 opt.OPT_CONTRACT_INFO）来查询更详细的行权价和到期日信息，这里为了通用性使用了名称解析法。
2. 行权价解析：
   • 代码通过正则表达式 re.search(r'(\d+\.\d+)$', name) 从期权名称（如“50ETF购9月3.000”）中提取行权价。这是处理 50ETF 期权最直观的方法。
3. 交易方向：
   • 备兑策略的核心是卖出期权。在 order 函数中，必须指定 side='short' 来表示卖出开仓（做空期权）。
4. 数量匹配：
   • 上证 50ETF 期权一张合约对应 10000 份 ETF。策略中设置 g.share_amount = 10000 并在卖出期权时卖出 1 张，以保持 1:1 的完全备兑关系，避免裸卖空的风险。
5. 移仓换月：
   • 策略设定了 g.days_to_expire_threshold = 7，即在到期前 7 天平仓旧合约并开新合约，避免进入交割周的剧烈波动和被行权的风险。

风险提示

• 上行收益封顶：如果 50ETF 暴涨，超过了行权价，你的收益会被锁定在行权价加上权利金的位置，无法享受后续的暴涨收益。
• 下行风险：虽然权利金提供了一定的安全垫，但如果 50ETF 大幅下跌，持有的现货亏损可能超过权利金收入，导致总资产亏损。

Q&A

Q: 为什么选择卖出虚值（OTM）期权而不是平值（ATM）期权？
A: 卖出虚值期权（行权价 > 当前价）允许标的资产在一定范围内上涨，策略可以同时获得部分股价上涨收益和权利金。卖出平值期权虽然权利金更多，但一旦股价上涨就会被行权，完全放弃了股价上涨的潜在收益。

Q: 如果期权被行权了怎么办？
A: 在回测中，如果持有到期且为实值，系统会自动行权（现金交割或实物交割取决于回测引擎设置）。在实盘中，备兑策略通常会在到期前平仓（移仓）以避免被行权。本策略代码包含到期前 7 天自动移仓的逻辑，因此大概率不会被行权。

Q: 如何调整策略的杠杆？
A: 备兑策略本身不带杠杆（10000股现货对应1张期权）。如果你卖出超过现货数量的期权（例如持有10000股却卖出2张期权），则构成了“比例备兑”或部分“裸卖空”，风险会显著增加，不建议初学者尝试。