求助：当回撤突破指定阈值时调整时间序列的实现方法

需求说明 #

当回撤（Drawdown）突破指定数值时调整时间序列：

• 当回撤达到4%时，将原时间序列缩减50%
• 当回撤回落至2%时，将时间序列恢复至100%

已实现随机时间序列生成及回撤计算代码，需补充基于回撤的序列调整逻辑。

完整实现代码 #

import numpy as np
import math
import pandas as pd
np.random.seed(1234)

sigma = 0.07
N = 3
IR = 1
daily_sigma = sigma / math.sqrt(252)
daily_mu = daily_sigma * IR / math.sqrt(252)
days = pd.bdate_range(pd.Timestamp(2022, 1, 1), pd.Timestamp(2022, 12, 31))
T = len(days)

def do_drawdown(cum_pnl_max, cum_pnl):
    return np.maximum(cum_pnl_max.add(-cum_pnl), 0)

def get_daily_gross(target_ir):
    return pd.DataFrame(np.random.normal(loc=daily_mu * target_ir, scale=daily_sigma, size=(N, T)), 
                        columns=pd.Series(days, name='date'), index=pd.Series(np.arange(N), name='obs'))

# 生成原始时间序列并计算回撤
daily_net = get_daily_gross(1)
cum_pnl = daily_net.cumsum(axis=1)
cum_pnl_max = cum_pnl.cummax(axis=1)
drawdown = do_drawdown(cum_pnl_max, cum_pnl)

# 定义回撤阈值
DOWNSCALE_THRESHOLD = 0.04  # 触发缩减的回撤阈值（4%）
RESCALE_THRESHOLD = 0.02     # 触发恢复的回撤阈值（2%）

# 初始化调整乘数：初始状态为100%
adjustment_multiplier = pd.DataFrame(1.0, index=daily_net.index, columns=daily_net.columns)

# 遍历时间点，根据回撤状态更新调整乘数
for i in range(1, T):
    prev_multiplier = adjustment_multiplier.iloc[:, i-1]
    current_drawdown = drawdown.iloc[:, i]
    
    # 状态切换逻辑
    # 前一状态正常且当前回撤≥4%，切换为缩减状态（乘数0.5）
    downscale_mask = (prev_multiplier == 1.0) & (current_drawdown >= DOWNSCALE_THRESHOLD)
    # 前一状态缩减且当前回撤≤2%，切换为正常状态（乘数1.0）
    rescale_mask = (prev_multiplier == 0.5) & (current_drawdown <= RESCALE_THRESHOLD)
    
    # 更新乘数
    adjustment_multiplier.iloc[:, i] = np.where(downscale_mask, 0.5, 
                                                np.where(rescale_mask, 1.0, prev_multiplier))

# 应用调整乘数到原始时间序列
adjusted_daily_net = daily_net * adjustment_multiplier

实现思路解析 #

1. 状态跟踪机制：通过adjustment_multiplier记录每个观测序列在每个时间点的调整状态（1.0代表正常，0.5代表缩减），避免因回撤短暂波动导致状态反复切换。
2. 阈值判断逻辑：
   • 仅当处于正常状态且回撤达到4%时，触发序列缩减
   • 仅当处于缩减状态且回撤回落至2%以下时，触发序列恢复
3. 高效批量处理：利用numpy的np.where和pandas掩码操作，批量处理所有观测序列的状态更新，比逐行循环更高效。
4. 结果应用：直接将调整乘数与原始日度序列相乘，保留原始数据的结构和索引，得到最终调整后的时间序列。

效果验证示例 #

可以通过以下代码查看调整效果：

# 查看单个观测序列的调整状态分布
print(adjustment_multiplier.iloc[0].value_counts())

# 对比原始序列与调整后序列的累计收益
import matplotlib.pyplot as plt
plt.figure(figsize=(12,6))
cum_pnl.iloc[0].plot(label='原始累计收益')
adjusted_daily_net.cumsum(axis=1).iloc[0].plot(label='调整后累计收益')
plt.legend()
plt.title('单序列原始与调整后累计收益对比')
plt.show()

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