基于Polars的进阶横截面交易算法实现（路径依赖场景）

策略规则 #

• 每期选收益率排名Top2的标的进入组合
• 调仓规则：下一期仅当持仓标的当期收益率排名小于3时卖出该标的，并用当期排名最高的标的替换，维持组合数量为2

实现方案 #

由于策略存在路径依赖（持仓状态影响后续调仓），我们结合Polars的向量化计算能力与状态跟踪来实现，兼顾大数据集效率与逻辑准确性。

1. 数据预处理：计算收益率与排名 #

首先基于长格式价格数据，计算每期标的收益率并按日期横截面排名：

import polars as pl

# 示例长格式价格数据
price_data = pl.DataFrame({
    "date": ["2023-01-01", "2023-01-01", "2023-01-01", 
             "2023-01-02", "2023-01-02", "2023-01-02", 
             "2023-01-03", "2023-01-03", "2023-01-03"],
    "symbol": ["A", "B", "C", "A", "B", "C", "A", "B", "C"],
    "price": [10, 20, 30, 11, 18, 33, 12, 22, 31]
})

# 计算收益率+按日期排名
ranked_data = (
    price_data.sort(["symbol", "date"])
    .with_columns(pl.col("price").pct_change().over("symbol").alias("return"))
    .drop_nulls()
    .sort("date")
    .with_columns(
        pl.col("return")
        .rank(descending=True, method="ordinal")
        .over("date")
        .alias("rank")
    )
)

2. 策略执行：状态跟踪与调仓 #

我们通过按日期分组后迭代处理，跟踪持仓状态并执行调仓逻辑：

def execute_strategy(grouped_data):
    sorted_dates = sorted(grouped_data.keys())
    # 初始持仓：第一期Top2标的
    first_group = grouped_data[sorted_dates[0]]
    current_hold = first_group.filter(pl.col("rank") <= 2)["symbol"].to_list()
    
    # 存储每期持仓记录
    hold_history = {sorted_dates[0]: current_hold.copy()}
    
    for date in sorted_dates[1:]:
        current_rank_map = (
            grouped_data[date]
            .select(["symbol", "rank"])
            .to_dict(as_series=False)
        )
        rank_lookup = dict(zip(current_rank_map["symbol"], current_rank_map["rank"]))
        
        # 筛选需卖出的持仓：符合当期排名小于3的标的
        to_exit = [sym for sym in current_hold if rank_lookup.get(sym, 999) < 3]
        # 获取当期排名第一的标的
        top1 = grouped_data[date].filter(pl.col("rank") == 1)["symbol"].item()
        
        # 更新持仓：移除卖出标的，替换为Top1，维持2只持仓
        for sym in to_exit:
            current_hold.remove(sym)
            if top1 not in current_hold:
                current_hold.append(top1)
        # 若持仓不足2，补充当期Top2（避免重复）
        if len(current_hold) < 2:
            top2 = grouped_data[date].filter(pl.col("rank") == 2)["symbol"].item()
            if top2 not in current_hold:
                current_hold.append(top2)
        
        hold_history[date] = current_hold.copy()
    
    # 转换为标准长格式输出
    return (
        pl.DataFrame({"date": hold_history.keys(), "holdings": hold_history.values()})
        .explode("holdings")
        .sort("date")
    )

# 按日期分组后执行策略
strategy_result = ranked_data.group_by("date").map_groups(lambda x: x).pipe(execute_strategy)
print(strategy_result)

3. 输出示例 #

执行后会得到每期的持仓记录，格式如下：

shape: (6, 2)
┌────────────┬──────────┐
│ date       │ holdings │
│ ---        │ ---      │
│ str        │ str      │
├────────────┼──────────┤
│ 2023-01-01 │ A        │
│ 2023-01-01 │ C        │
│ 2023-01-02 │ C        │
│ 2023-01-02 │ A        │
│ 2023-01-03 │ A        │
│ 2023-01-03 │ B        │
└────────────┴──────────┘

关键优化点 #

• 用Polars向量化计算收益率和排名，避免低效的逐行处理
• 仅在日期维度迭代，而非逐标的循环，适配大数据集（当标的数量远多于日期时，效率优势显著）
• 可根据实际业务逻辑调整卖出条件（比如若规则应为“排名≥3时卖出”，只需修改to_exit的判断条件为rank_lookup.get(sym, 999) >= 3）

内容的提问来源于stack exchange，提问作者Andi