ah_premium_strategy.py

# A股与H股溢价套利策略
# 作者：Claude
# 基于聚宽内置AH股溢价数据

"""
策略名称：A股与H股溢价套利策略
策略类型：跨市场套利

策略核心逻辑：
1. 当A股相对H股存在折价时（h_a_comp较低），买入A股标的
2. 持有A股直至A股相对H股溢价过高（h_a_comp > 阈值）时平仓

数据来源：
使用聚宽内置的 finance.STK_AH_PRICE_COMP 表
- a_code: A股代码
- h_a_comp: 比值指标

h_a_comp 解读（根据实际数据分析）：
- h_a_comp 较低: A股相对H股便宜（买入机会）
- h_a_comp 较高: A股相对H股贵（卖出信号）
- 策略选择h_a_comp最低的股票，即A股相对H股折价最大的标的

风险提示：
1. AH股价差可能长期存在，不一定会收敛
2. 汇率波动会影响实际收益
3. 港股通交易有额度限制和T+0规则差异
"""

from jqdata import finance
import pandas as pd


def initialize(context):
    """
    策略初始化函数
    """
    # 设定中证500指数作为基准
    set_benchmark('000905.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 设置滑点 - 0.2%
    set_slippage(FixedSlippage(0.002))
    # 设置交易成本
    set_order_cost(
        OrderCost(
            open_tax=0,
            close_tax=0.001,      # 印花税千分之一
            open_commission=0.0003,  # 佣金万分之三
            close_commission=0.0003,
            close_today_commission=0,
            min_commission=5
        ),
        type='stock'
    )
    # 过滤日志
    log.set_level('system', 'error')

    # === 策略参数 ===
    # 买入阈值：只买入h_a_comp < 此值的股票（A股相对H股折价）
    g.buy_threshold = 0.95  # A股相对H股折价5%以上才买入

    # 卖出阈值：h_a_comp > 此值时卖出A股（A股相对H股溢价了）
    g.sell_threshold = 1.00  # A股相对H股溢价0%时卖出（即A股比H股贵就卖）

    # 最大持仓数量
    g.max_holdings = 5

    # 单只股票最大仓位比例
    g.max_position_ratio = 0.2  # 20%

    # === 定时任务 ===
    # 每日14:30执行交易检查
    run_daily(trade, time='14:30')
    # 每日收盘后记录持仓
    run_daily(log_positions, time='15:10')


def get_ah_premium_data(context):
    """
    获取AH股溢价数据（使用聚宽内置API）

    返回：
    DataFrame: 包含 a_code, h_a_comp 等字段，按h_a_comp升序排列
    - h_a_comp 较低: A股相对便宜（买入机会）
    - h_a_comp 较高: A股相对较贵
    """
    try:
        # 查询AH股价比较数据，按h_a_comp升序（低的在前，即A股便宜的在前）
        df = finance.run_query(
            query(
                finance.STK_AH_PRICE_COMP.a_code,
                finance.STK_AH_PRICE_COMP.h_a_comp
            )
            .filter(finance.STK_AH_PRICE_COMP.day == context.previous_date)
            .order_by(finance.STK_AH_PRICE_COMP.h_a_comp)  # 升序：A股便宜的在前
        )
        return df
    except Exception as e:
        print(f"获取AH股溢价数据失败: {e}")
        return pd.DataFrame()


def filter_paused_stock(stock_list):
    """过滤停牌股票"""
    current_data = get_current_data()
    return [stock for stock in stock_list if not current_data[stock].paused]


def filter_st_stock(stock_list):
    """过滤ST股票"""
    current_data = get_current_data()
    return [stock for stock in stock_list if not current_data[stock].is_st]


def delisted_filter(stock_list):
    """过滤退市股票"""
    current_data = get_current_data()
    return [stock for stock in stock_list if '退' not in current_data[stock].name]


def filter_tradable_stocks(stock_list):
    """
    过滤可交易股票：排除停牌、ST、退市、涨停
    参考 get_ah_premium_data.py 的实现，并增加涨停过滤
    """
    # 过滤停牌
    result = filter_paused_stock(stock_list)
    # 过滤ST
    result = filter_st_stock(result)
    # 过滤退市
    result = delisted_filter(result)
    # 过滤涨停（无法买入）
    result = filter_limit_up_stock(result)

    return result


def filter_limit_up_stock(stock_list):
    """过滤涨停股票（无法买入）"""
    current_data = get_current_data()
    tradable = []
    for stock in stock_list:
        data = current_data[stock]
        # 涨停时 last_price >= high_limit
        if data.last_price < data.high_limit * 0.998:  # 允许0.2%误差
            tradable.append(stock)
    return tradable


def trade(context):
    """
    交易执行函数

    策略逻辑（修正后）：
    1. 买入：选择h_a_comp最低的股票（A股相对H股最便宜的）
    2. 卖出：h_a_comp > 阈值时卖出（A股相对H股变贵了）
    """
    print("=" * 60)
    print("每日AH股溢价套利检查")
    print("=" * 60)

    # 获取AH股溢价数据（已按h_a_comp升序排列）
    df = get_ah_premium_data(context)

    if len(df) == 0:
        print("无法获取溢价数据，跳过本次交易")
        return

    # 打印溢价率排名（h_a_comp最低的10只，即A股最便宜的）
    print("\n当前A股相对H股折价排名（h_a_comp最低，前10）：")
    print("-" * 60)
    for idx, row in df.head(10).iterrows():
        h_a_comp = row['h_a_comp']
        # h_a_comp越低，A股相对越便宜
        discount_pct = (1 - h_a_comp) * 100  # A股相对H股的折价百分比
        print(f"  {row['a_code']}: 比值 {h_a_comp:.3f} | "
              f"A股相对H股折价 {discount_pct:.2f}%")

    # === 卖出逻辑：h_a_comp > 阈值（A股相对H股变贵了）===
    positions_to_sell = []

    for stock in list(context.portfolio.positions.keys()):
        # 查找该股票的溢价数据
        stock_data = df[df['a_code'] == stock]

        if len(stock_data) == 0:
            # 不在AH股列表中，保持持仓
            continue

        h_a_comp = stock_data['h_a_comp'].iloc[0]

        # h_a_comp > 阈值 表示A股相对H股变贵了，应该卖出
        if h_a_comp > g.sell_threshold:
            positions_to_sell.append(stock)
            discount_pct = (1 - h_a_comp) * 100
            print(f"\n>>> 卖出信号: {stock} 比值 {h_a_comp:.3f}, "
                  f"A股折价 {discount_pct:.2f}% (接近平价)")

    # 执行卖出
    for stock in positions_to_sell:
        order_target_value(stock, 0)
        print(f"卖出 {stock}（A股相对H股不再便宜）")

    # === 买入逻辑：选择h_a_comp最低的股票（A股相对最便宜）===
    # 过滤可交易股票
    tradable_stocks = filter_tradable_stocks(df['a_code'].tolist())

    # 筛选可交易的股票
    buy_candidates = df[df['a_code'].isin(tradable_stocks)].copy()

    # 买入阈值过滤：只买入h_a_comp < buy_threshold的股票（A股折价足够大）
    buy_candidates = buy_candidates[buy_candidates['h_a_comp'] < g.buy_threshold]

    # 排除已持仓的股票
    current_holdings = list(context.portfolio.positions.keys())
    buy_candidates = buy_candidates[~buy_candidates['a_code'].isin(current_holdings)]

    # 数据已经按h_a_comp升序排列，前面的就是A股最便宜的
    # 计算当前持仓数量
    current_holding_count = len([s for s in current_holdings if s in df['a_code'].values])

    # 计算可买入数量
    slots_available = g.max_holdings - current_holding_count

    if slots_available <= 0:
        print(f"\n当前持仓数量已达上限 ({g.max_holdings})，不再买入")
    elif len(buy_candidates) > 0:
        print(f"\n>>> 买入候选股票（h_a_comp最低，A股相对最便宜）：")
        for idx, row in buy_candidates.head(slots_available).iterrows():
            discount_pct = (1 - row['h_a_comp']) * 100
            print(f"  {row['a_code']}: 比值 {row['h_a_comp']:.3f}, A股折价 {discount_pct:.2f}%")

        # 计算买入金额（等权分配）
        buy_list = buy_candidates.head(slots_available)['a_code'].tolist()
        cash_per_stock = context.portfolio.available_cash / len(buy_list)

        # 限制单只股票仓位
        max_value = context.portfolio.total_value * g.max_position_ratio
        cash_per_stock = min(cash_per_stock, max_value)

        for stock in buy_list:
            # 获取当前价格，检查是否足够买入100股
            current_price = get_bars(stock, 1, '1m', ['close'], include_now=True,
                                     end_dt=context.current_dt, df=True).iloc[-1]['close']

            # 检查资金是否足够购买100股
            if cash_per_stock < current_price * 100:
                print(f"跳过 {stock}：资金 {cash_per_stock:.2f} 不足买入100股（需要 {current_price * 100:.2f}）")
                continue

            order_target_value(stock, cash_per_stock)
            stock_info = buy_candidates[buy_candidates['a_code'] == stock].iloc[0]
            discount_pct = (1 - stock_info['h_a_comp']) * 100
            print(f"买入 {stock}，金额 {cash_per_stock:.2f}，A股折价 {discount_pct:.2f}%")
    else:
        print(f"\n没有符合买入条件的股票")

    print("=" * 60)


def log_positions(context):
    """
    记录当前持仓情况
    """
    print("\n当前持仓：")
    print("-" * 40)

    if len(context.portfolio.positions) == 0:
        print("  空仓")
    else:
        # 获取溢价数据
        df = get_ah_premium_data(context)

        for stock in context.portfolio.positions:
            pos = context.portfolio.positions[stock]
            stock_data = df[df['a_code'] == stock]

            if len(stock_data) > 0:
                h_a_comp = stock_data['h_a_comp'].iloc[0]
                discount_pct = (1 - h_a_comp) * 100
                print(f"  {stock}: 数量 {pos.total_amount}, "
                      f"市值 {pos.value:.2f}, A股折价 {discount_pct:.2f}%")
            else:
                print(f"  {stock}: 数量 {pos.total_amount}, 市值 {pos.value:.2f}")

    print(f"\n账户总资产: {context.portfolio.total_value:.2f}")
    print(f"可用资金: {context.portfolio.available_cash:.2f}")
    print(f"持仓市值: {context.portfolio.positions_value:.2f}")
    print("-" * 40)