rate-catcher 0.1.0

Pattern-based time-series prediction using moving average and slope analysis

🎯 RateCatcher

A Python library for pattern-based time-series prediction using moving average and slope analysis.

📖 Overview

RateCatcher learns patterns from historical data by analyzing moving averages and their slopes, then uses these learned patterns to predict future events. The core idea: when certain combinations of MA values and slopes occur, they often correlate with specific outcomes.

graph LR
    A[Time Series Data] --> B[Calculate MA + Slope]
    B --> C[Bucket into Categories]
    C --> D[Learn Triggers]
    D --> E[Predict Future Events]
    E --> F[num_activated per timestamp]

🚀 Installation

pip install rate-catcher

Or install in development mode:

git clone https://github.com/Arseni1919/RateCatcherLibrary.git
cd RateCatcherLibrary
pip install -e .

📚 Core Workflow

1️⃣ Generate Example Data

Start with test data to understand the library. generate_dummy_data() creates two signals:

• sig_1: Sinusoidal baseline signal (200-350 per minute)
• sig_2: Normally low (10-20), but spikes to ~170 when sig_1 crosses 340

from rate_catcher import generate_dummy_data

df = generate_dummy_data(n_minutes=1000)

Returns: DataFrame with columns [timestamp, signal_name, count]

---

2️⃣ Define Your Metric Function

A metric function evaluates how "interesting" each timestamp is. It receives a timestamp, data, and memory dict, then returns a float.

import pandas as pd

def metric_func(timestamp, data, memory):
    if 'max' not in memory:
        sig_2_data = data[data['signal_name'] == 'sig_2']
        memory['max'] = sig_2_data['count'].max()
    max_val = memory['max']
    sig_2_data = data[data['signal_name'] == 'sig_2']
    next_5_minutes = sig_2_data[
        (sig_2_data['timestamp'] > timestamp) &
        (sig_2_data['timestamp'] <= timestamp + pd.Timedelta(minutes=5))
    ]
    if len(next_5_minutes) == 0:
        return 0.0
    max_next_5 = next_5_minutes['count'].max()
    return float(max_next_5 / max_val)

This example looks ahead 5 minutes and returns the ratio of peak activity to overall max. High values (near 1.0) indicate upcoming spikes.

---

3️⃣ Analyze Single MA Value

Use get_ma_triggers() to analyze patterns at a specific moving average window.

from rate_catcher import get_ma_triggers

output = get_ma_triggers(
    main_signal_name='sig_1',
    data=df,
    ma_value=20,
    metric_function=metric_func,
    ma_buckets_number=10,
    slope_buckets_number=20
)
ma_df, slope_labels, ma_labels, triggers, ma_buckets, slope_buckets = output

Parameters:

• main_signal_name: Signal to analyze
• ma_value: Moving average window size
• metric_function: Your metric function
• ma_buckets_number: Divide MA range into N buckets (default 10)
• slope_buckets_number: Divide slope range -90° to +90° into N buckets (default 20)

Returns:

• ma_df: DataFrame with [timestamp, ma] (starts at index ma_value)
• slope_labels: List of slope bucket labels (e.g., "0-9", "9-18")
• ma_labels: List of MA bucket labels (e.g., "200-230")
• triggers: Dict {(slope_label, ma_label): [(timestamp, metric_value), ...]}
• ma_buckets: List of (min, max) tuples for MA ranges
• slope_buckets: List of (min, max) tuples for slope ranges

Example output:

triggers[("18-27", "280-310")] = [
    (Timestamp('2024-01-01 00:15:00'), 0.95),
    (Timestamp('2024-01-01 02:30:00'), 0.87),
    ...
]

This means: when MA is between 280-310 AND slope is 18-27°, the metric values were [0.95, 0.87, ...].

---

4️⃣ Learn from Multiple MA Values

learn_all_triggers() repeats the analysis across many MA windows to find patterns at different time scales.

from rate_catcher import learn_all_triggers

all_triggers = learn_all_triggers(
    main_signal_name='sig_1',
    data=learning_data,
    metric_function=metric_func,
    ma_values=list(range(5, 51, 5)),
    ma_buckets_number=10,
    slope_buckets_number=20
)

Parameters:

• ma_values: List of MA windows to analyze (default: 1-100)
• Other params same as get_ma_triggers()

Returns: Dict {ma_value: (ma_df, slope_labels, ma_labels, triggers, ma_buckets, slope_buckets)}

Shows progress bar via tqdm. For ma_values=[5,10,15,...,50], this analyzes 10 different MA windows.

---

5️⃣ Predict with Learned Triggers

predict_with_all_triggers() applies learned patterns to new data.

from rate_catcher import predict_with_all_triggers

predictions_df = predict_with_all_triggers(
    main_signal_name='sig_1',
    data=prediction_data,
    all_triggers=all_triggers,
    minimum_count=3,
    minimum_rate=0.5,
    aggregation_method="mean"
)

Parameters:

• all_triggers: Output from learn_all_triggers()
• minimum_count: Trigger must have occurred ≥N times during learning
• minimum_rate: Trigger's aggregated metric value must be ≥ threshold
• aggregation_method: How to aggregate metric values:
  • "mean" (default): average of all occurrences
  • "max": maximum value
  • "min": minimum value
  • Custom function: lambda values: np.percentile(values, 75)

Returns: DataFrame with [timestamp, num_activated]

Example:

   timestamp            num_activated
0  2024-01-01 08:20:00  0
1  2024-01-01 08:25:00  3
2  2024-01-01 08:30:00  7

For each timestamp, counts how many learned triggers are activated:

1. Calculate MA at that timestamp for each learned MA window
2. Calculate slope
3. Check if (slope, MA) matches any learned trigger
4. If trigger exists AND meets minimum_count AND aggregated metric ≥ minimum_rate → +1

---

6️⃣ End-to-End Prediction

predict() combines learning and prediction in one call.

from rate_catcher import predict

predictions_df = predict(
    main_signal_name='sig_1',
    learning_data=learning_data,
    prediction_data=prediction_data,
    metric_function=metric_func,
    ma_values=list(range(5, 51, 5)),
    minimum_count=3,
    minimum_rate=0.5,
    aggregation_method="mean"
)

Convenience function = learn_all_triggers() + predict_with_all_triggers().

---

7️⃣ Visualize Results

Plot Signals with Metric

from plots import plot_signals_and_metric

metric_values = []
memory = {}
for ts in df['timestamp'].unique():
    metric_val = metric_func(ts, df, memory)
    metric_values.append({'timestamp': ts, 'metric': metric_val})
metric_df = pd.DataFrame(metric_values)

plot_signals_and_metric(df, metric_df)

Creates two subplots:

• Top: sig_1 (blue) and sig_2 (red) time series
• Bottom: metric values (green)

Plot MA Analysis with Heatmap

from plots import plot_ma_triggers

output = get_ma_triggers('sig_1', df, 20, metric_func)
ma_df, slope_labels, ma_labels, triggers, ma_buckets, slope_buckets = output

plot_ma_triggers('sig_1', df, ma_df, slope_labels, ma_labels, triggers)

Creates two subplots:

• Top: Original signal (blue) with MA overlay (red dashed)
• Bottom: Heatmap showing average metric values for each (slope, MA) combination

Plot Predictions

from plots import plot_predictions

plot_predictions('sig_1', prediction_data, predictions_df)

Overlays predictions on signal data:

• Blue line (left axis): signal counts
• Orange bars (right axis): number of activated triggers

---

🔧 Utility Functions

The library includes helper functions:

from rate_catcher import (
    apply_moving_average,
    calculate_slope_degrees,
    find_bucket_index,
    create_bucket_labels
)

• apply_moving_average(series, window): Rolling mean
• calculate_slope_degrees(ma_current, ma_next): Angle in degrees (-90 to +90)
• find_bucket_index(value, buckets): Which bucket does value fall into?
• create_bucket_labels(buckets): Generate "min-max" labels

---

🎨 Custom Aggregation

Define how to aggregate metric values for each trigger:

import numpy as np

def custom_aggregation(values):
    return np.percentile(values, 90)

predictions_df = predict_with_all_triggers(
    main_signal_name='sig_1',
    data=prediction_data,
    all_triggers=all_triggers,
    minimum_count=5,
    minimum_rate=0.7,
    aggregation_method=custom_aggregation
)

Or use built-in strings: "mean", "max", "min"

---

📊 Example Workflow

from rate_catcher import (
    generate_dummy_data,
    metric_func,
    predict,
    plot_predictions
)

df = generate_dummy_data(n_minutes=1000)

timestamps = sorted(df['timestamp'].unique())
split_idx = len(timestamps) // 2
split_ts = timestamps[split_idx]

learning_data = df[df['timestamp'] < split_ts]
prediction_data = df[df['timestamp'] >= split_ts]

predictions_df = predict(
    main_signal_name='sig_1',
    learning_data=learning_data,
    prediction_data=prediction_data,
    metric_function=metric_func,
    ma_values=list(range(10, 31, 5)),
    minimum_count=3,
    minimum_rate=0.5
)

plot_predictions('sig_1', prediction_data, predictions_df)

print(f"Timestamps with activations: {(predictions_df['num_activated'] > 0).sum()}")
print(f"Max activations: {predictions_df['num_activated'].max()}")

---

🧪 Testing

Run tests from the project root:

uv run python tests/test_predict.py
uv run python tests/test_aggregation_methods.py

---

📝 API Reference Summary

Function	Purpose	Returns
generate_dummy_data()	Create test data	DataFrame
metric_func()	Example metric (customize!)	float
aggregation_func()	Example aggregation	float
get_ma_triggers()	Analyze single MA window	Tuple (ma_df, labels, triggers, buckets)
learn_all_triggers()	Analyze multiple MA windows	Dict {ma_value: tuple}
predict_with_all_triggers()	Predict using learned patterns	DataFrame [timestamp, num_activated]
predict()	Learn + predict in one call	DataFrame [timestamp, num_activated]
plot_signals_and_metric()	Visualize signals + metric	Plot
plot_ma_triggers()	Visualize MA analysis	Plot with heatmap
plot_predictions()	Visualize predictions	Plot with dual axes

---

🎓 Key Concepts

• Moving Average (MA): Smooths signal by averaging over a window
• Slope: Rate of change between consecutive MA values, measured in degrees
• Buckets: Divide continuous ranges into discrete categories
• Triggers: Specific (slope, MA) combinations that correlate with outcomes
• Metric Function: Evaluates "importance" of each timestamp
• Aggregation Method: Combines multiple metric values into one representative value

---

🤝 Contributing

Follow code conventions in CLAUDE.md.