基于Python Pandas实现每日回撤(DrawDown)持续时长的技术问询

I get it—you want to compute the daily drawdown duration (how many consecutive days we’ve been in a drawdown from the last peak) for your time series, just like you did in Excel. Let’s break down how to implement this in Python using pandas, which is ideal for this kind of tabular data work.

Key Logic Recap #

First, let’s align on what the Duration columns represent:

• For each series (A and B), we track the running maximum (peak) of the cumulative value up to each day.
• If the current cumulative value is below this peak: the duration is the previous day’s duration + 1.
• If the current cumulative value hits or exceeds the peak: reset the duration to 0, and update the running maximum to the current cumulative value.

Step-by-Step Implementation #

Let’s start with sample data matching your Excel table, then build the code step by step.

1. Import Pandas and Load Data #

First, set up your environment and create a DataFrame with your sample data:

import pandas as pd

# Sample data matching your example
data = {
    'A': [3.5, -2.1, 0.7, -1.1, 2.4, 1.3],
    'B': [2.2, 1.8, 0.7, -1.8, 3.2, -1.8]
}

df = pd.DataFrame(data)

2. Calculate Cumulative Values #

Compute the cumulative sums for A and B (your Cum(A) and Cum(B) columns):

df['Cum(A)'] = df['A'].cumsum()
df['Cum(B)'] = df['B'].cumsum()

3. Compute Running Peaks #

Next, calculate the running maximum (peak) for each cumulative series:

df['Peak(A)'] = df['Cum(A)'].cummax()
df['Peak(B)'] = df['Cum(B)'].cummax()

4. Calculate Drawdown Values (Optional) #

If you want to replicate the DD(A) and D(B) columns (drawdown amount), compute them as:

df['DD(A)'] = df['Cum(A)'] - df['Peak(A)']
df['D(B)'] = df['Cum(B)'] - df['Peak(B)']

5. Compute Drawdown Duration #

This is the core part. We’ll use a custom function with vectorized operations to track consecutive drawdown days efficiently:

def calculate_duration(cum_series):
    peak = cum_series.cummax()
    # Create a mask where current value is below peak (in drawdown)
    drawdown_mask = cum_series < peak
    # Reset the counter whenever drawdown status changes
    reset_groups = (drawdown_mask != drawdown_mask.shift()).cumsum()
    # Count consecutive days in each drawdown group
    duration = drawdown_mask.groupby(reset_groups).cumsum()
    return duration

# Apply the function to both cumulative series
df['Duration(A)'] = calculate_duration(df['Cum(A)'])
df['Duration(B)'] = calculate_duration(df['Cum(B)'])

6. View the Result #

Print the DataFrame to verify it matches your Excel example:

print(df)

The output will look exactly like your table:

A    B  Cum(A)  Cum(B)  Peak(A)  Peak(B)  DD(A)  D(B)  Duration(A)  Duration(B)
0  3.5  2.2     3.5     2.2      3.5      2.2    0.0   0.0            0            0
1 -2.1  1.8     1.4     4.0      3.5      4.0   -2.1   0.0            1            0
2  0.7  0.7     2.1     4.7      3.5      4.7   -1.4   0.0            2            0
3 -1.1 -1.8     1.0     2.9      3.5      4.7   -2.5  -1.8            3            1
4  2.4  3.2     3.4     6.1      3.5      6.1   -0.1   0.0            4            0
5  1.3 -1.8     4.7     4.3      4.7      6.1    0.0  -1.8            0            1

Quick Explanation of the Duration Function #

• drawdown_mask: A boolean series where True means we’re in a drawdown (current cumulative value < peak).
• reset_groups: Increments every time the drawdown status changes (e.g., from non-drawdown to drawdown), grouping consecutive drawdown periods.
• duration: Counts consecutive days in each drawdown group using cumsum(), which resets automatically when the group changes.

This approach is fast even for large datasets, as it avoids slow loops and uses pandas’ optimized vectorized operations.

内容的提问来源于stack exchange，提问作者B.A.