多列多周期滚动收益率分位数计算的最优实现方案问询

核心思路 #

针对多列时间序列，根据指定的周期字典（支持日/周等周期），通过周期统一转换+向量化函数封装+批量合并的方式实现高效计算，避免冗余循环，同时兼容无负收益这类边界场景。

完整代码示例 #

import pandas as pd
import numpy as np

# 1. 构造示例数据（多列资产价格时间序列）
dates = pd.date_range(start='2023-01-01', periods=100, freq='B')  # 按交易日生成日期
price_df = pd.DataFrame(
    np.random.randn(100, 3).cumsum(axis=0) + 100,
    index=dates,
    columns=['AssetA', 'AssetB', 'AssetC']
)

# 2. 定义周期字典：key为列名，value为日数或'W'（代表周，按5交易日计算）
period_dict = {'AssetA': 7, 'AssetB': 'W', 'AssetC': 14}

# 3. 周期转换函数：将周/日统一转为交易日数
def convert_to_trading_days(period):
    if period == 'W':
        return 5  # 可根据实际市场交易日规则调整（如A股、港股均为5天）
    else:
        return int(period)

# 转换所有周期为交易日数
trading_periods = {col: convert_to_trading_days(p) for col, p in period_dict.items()}

# 4. 核心计算函数：单列滚动负收益分位数统计
def get_neg_return_quantiles(series, period, quantiles=[0.1, 0.25, 0.5]):
    # 计算指定周期的滚动收益率：(当前价 / N期前价格) - 1
    rolling_returns = series / series.shift(period) - 1
    # 筛选负收益并去除空值
    negative_returns = rolling_returns[rolling_returns < 0].dropna()
    # 处理无负收益的边界情况
    if negative_returns.empty:
        return pd.Series([np.nan]*len(quantiles), index=quantiles)
    # 返回分位数结果
    return negative_returns.quantile(quantiles)

# 5. 批量计算并合并结果
final_result = pd.concat(
    [get_neg_return_quantiles(price_df[col], trading_periods[col]) for col in trading_periods],
    axis=1,
    keys=trading_periods.keys()
)

print(final_result)

实现优势 #

• 高效性：利用pandas向量化操作+concat批量合并，比显式for循环更贴合pandas优化逻辑，大数据量下性能更优；
• 灵活性：周期转换函数可扩展支持月（'M'）、季度（'Q'）等周期，只需添加对应交易日数映射即可；
• 鲁棒性：处理了无负收益的边界场景，避免空值报错；
• 可读性：函数封装后逻辑清晰，分位数列表等参数可灵活调整，便于后续维护。

注意事项 #

• 若需要计算滚动窗口内的负收益分位数（而非全序列），可将rolling_returns替换为series.rolling(window=period).apply(lambda x: (x.iloc[-1]/x.iloc[0])-1)，再对滚动结果做分位数统计；
• 交易日数的转换需根据实际市场规则调整（如美股部分节假日后的交易日数可能不同）。

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