custom_fut_mapping.py

import datetime
import tushare as ts
import pandas as pd

pro = ts.pro_api()


def fut_mapping(fut_code, exchange):
    df = pro.fut_basic(exchange=exchange, fut_type="1", fut_code=fut_code)
    df["list_date"] = pd.to_datetime(df["list_date"])
    df["delist_date"] = pd.to_datetime(df["delist_date"])

    # 生成换仓日期列表（每月28号）
    all_dates = pd.date_range(
        start=df["list_date"].min(), end=df["delist_date"].max(), freq="MS"
    ) + pd.DateOffset(days=27)
    roll_dates = [d for d in all_dates if d <= df["delist_date"].max()]

    result = []

    start_date = None
    current_future = None
    for roll_date in roll_dates:
        # 获取下一个合约（到期日最近且 40 天不到期的后续合约）
        next_futures = df[
            (df["delist_date"] > roll_date + datetime.timedelta(days=40))
            & (df["list_date"] <= roll_date)
        ].sort_values("delist_date")

        if not next_futures.empty:
            next_future = next_futures.iloc[0]
            if (
                current_future is None
                or next_future["ts_code"] != current_future["ts_code"]
            ):
                if start_date is None:
                    start_date = next_future["list_date"].strftime("%Y%m%d")

                current_future = next_future
                result.append(
                    {
                        "ts_code": fut_code + "C",
                        "trade_date": roll_date.strftime("%Y%m%d"),
                        "mapping_ts_code": next_future["ts_code"],
                    }
                )

    if len(result) == 0:
        return
    mapping_data = pd.DataFrame(result).drop_duplicates().sort_values(by="trade_date")

    trade_cal = pro.trade_cal(
        exchange=exchange,
        start_date=start_date,
        end_date=datetime.datetime.now().strftime("%Y%m%d"),
    )
    trade_cal = trade_cal[trade_cal["is_open"] == 1].sort_values(by="cal_date")
    trade_cal.rename(columns={"cal_date": "trade_date"}, inplace=True)

    trade_cal["trade_date"] = trade_cal["trade_date"].astype(int)
    mapping_data["trade_date"] = mapping_data["trade_date"].astype(int)

    # pd.merge_asol 会负责当换仓日期不是交易日，会順移到下个交易日
    mapping_data = pd.merge_asof(
        trade_cal["trade_date"],
        mapping_data,
        on=["trade_date"],
        direction="forward",
    )

    return mapping_data


mapping_data = fut_mapping("LH", exchange="DCE")
print(mapping_data)