12. Trend Detection

Trend detection identifies long-term directional movement in data, filtering out noise and seasonality to reveal underlying patterns.

Moving Average Trend Lines

Simple moving averages reveal direction:

# Multiple timeframe trends
df['trend_7d'] = df['value'].rolling(window=7).mean()
df['trend_30d'] = df['value'].rolling(window=30).mean()
df['trend_90d'] = df['value'].rolling(window=90).mean()

# Trend direction indicator
df['trend_direction'] = np.where(df['trend_30d'] > df['trend_30d'].shift(30), 'upward', 'downward')

Linear Regression for Trend Slope

Calculate the slope of the trend line:

from scipy import stats

def calculate_trend(series):
    x = np.arange(len(series))
    slope, intercept, r_value, p_value, std_err = stats.linregress(x, series)
    return {
        'slope': slope,
        'r_squared': r_value**2,
        'p_value': p_value,
        'trend_per_period': slope / series.mean() * 100  # % change per period
    }

trend_metrics = calculate_trend(df['revenue'].dropna())
print(f"Daily trend: {trend_metrics['slope']:.2f}")
print(f"R²: {trend_metrics['r_squared']:.3f}")
print(f"p-value: {trend_metrics['p_value']:.4f}")

Detecting Change Points

Significant shifts in underlying patterns:

# Cumulative sum (CUSUM) for shift detection
def cusum_detect(data, threshold=5):
    mean = data.mean()
    cusum_pos = []
    cusum_neg = []
    
    for val in data:
        cusum_pos.append(max(0, cusum_pos[-1] + val - mean if cusum_pos else val - mean))
        cusum_neg.append(min(0, cusum_neg[-1] + val - mean if cusum_neg else val - mean))
    
    return cusum_pos, cusum_neg

cusum_pos, cusum_neg = cusum_detect(df['metric'])

# Find change points where CUSUM exceeds threshold
change_points = [i for i, (p, n) in enumerate(zip(cusum_pos, cusum_neg)) 
                 if abs(p) > threshold or abs(n) > threshold]

Local Trend with Polynomial Fitting

from numpy.polynomial import polynomial as P

# Fit quadratic trend (captures acceleration/deceleration)
degree = 2
x = np.arange(len(df))
coeffs = np.polyfit(x, df['value'].values, degree)
poly = np.poly1d(coeffs)

df['poly_trend'] = poly(x)

# Detect inflection points where second derivative changes sign
second_derivative = np.diff(np.diff(poly(x)))
inflection_indices = np.where(np.diff(np.sign(second_derivative)) != 0)[0] + 2

Visualization with Confidence Bands

plt.figure(figsize=(12, 6))
plt.plot(df.index, df['value'], alpha=0.4, label='Actual')
plt.plot(df.index, df['trend_30d'], linewidth=2, label='30-day Trend')

# Confidence band using rolling standard deviation
upper_band = df['trend_30d'] + 2 * df['value'].rolling(30).std()
lower_band = df['trend_30d'] - 2 * df['value'].rolling(30).std()
plt.fill_between(df.index, lower_band, upper_band, alpha=0.2, label='95% CI')

plt.legend()
plt.title('Trend with Confidence Band')