pawnlib.metrics package

from pawnlib.metrics import *

Submodules

pawnlib.metrics.tracker module

from pawnlib.metrics import tracker

class TPSCalculator(history_size=50, sleep_time=2, variable_time=False)

Bases: object

A class to calculate Transactions Per Second (TPS) based on block height changes over time.

This class tracks the block height and calculates TPS over a configurable history size. It supports both fixed interval calculations and variable time-based calculations.

Parameters:

• history_size (int) – Number of recent TPS values to keep for averaging. |default| 50

• sleep_time (int) – Fixed interval between API calls in seconds. |default| 2

• variable_time (bool) – If True, calculate TPS based on actual elapsed time. |default| False

Example

# Initialize TPSCalculator with default parameters
tps_calculator = TPSCalculator()

# Simulate block height updates
current_height = 100
current_time = time.time()

# Calculate TPS with a new block height
current_tps, average_tps = tps_calculator.calculate_tps(current_height, current_time)

print(f"Current TPS: {current_tps}, Average TPS: {average_tps}")

# Reset the calculator if needed
tps_calculator.reset()

Initializes the TPSCalculator.

Parameters:

• history_size (int) – Number of recent TPS values to keep for averaging. |default| 50

• sleep_time (int) – Fixed interval between API calls in seconds. |default| 2

• variable_time (bool) – If True, calculate TPS based on actual elapsed time. |default| False

__init__(history_size=50, sleep_time=2, variable_time=False)

Initializes the TPSCalculator.

Parameters:

• history_size (int) – Number of recent TPS values to keep for averaging. |default| 50

• sleep_time (int) – Fixed interval between API calls in seconds. |default| 2

• variable_time (bool) – If True, calculate TPS based on actual elapsed time. |default| False

calculate_tps(current_height, current_time=None)

Calculates the current TPS and updates the TPS history.

Parameters:

• current_height (int) – The current block height.

• current_time (float, optional) – The current timestamp. If None and variable_time is False, uses fixed sleep_time. |default| None

Returns:

Tuple of (current_tps, average_tps)

Return type:

tuple(float, float)

get_average_tps()

Calculates the average TPS from the history.

Returns:

The average TPS.

Return type:

float

processed_tx()

Returns the total number of transactions since monitoring started.

Returns:

Total transactions as integer.

Return type:

int

last_n_tx()

Returns the number of transactions in the last n seconds (sleep_time).

Returns:

Transactions count in last n seconds.

Return type:

float

reset()

Resets the TPSCalculator to its initial state.

This clears all stored data and resets counters.

Returns:

None

class SyncSpeedTracker(history_size=10)

Bases: object

A class to track the synchronization speed of a blockchain node.

This class calculates the average number of blocks synchronized per second using a moving average approach based on recent block height and time differences.

Parameters:

history_size (int) – The number of records to keep for calculating the moving average. |default| 10

Example

# Initialize the tracker with a history size of 10
sync_tracker = SyncSpeedTracker(history_size=10)

# Update the tracker with new block height and timestamp
current_height = 150
current_time = time.time()
sync_tracker.update(current_height, current_time)

# Get the average synchronization speed (blocks per second)
avg_sync_speed = sync_tracker.get_average_sync_speed()
print(f"Average Sync Speed: {avg_sync_speed} blocks/second")

Initializes the SyncSpeedTracker.

Parameters:

history_size (int) – The number of records to store for calculating the moving average. |default| 10

__init__(history_size=10)

Initializes the SyncSpeedTracker.

Parameters:

history_size (int) – The number of records to store for calculating the moving average. |default| 10

update(current_height, current_time)

Updates the synchronization speed tracker with new block height and timestamp.

Parameters:

• current_height (int) – The current block height.

• current_time (float) – The current timestamp.

get_average_sync_speed()

Calculates the average number of blocks synchronized per second using a moving average.

Returns:

The average synchronization speed in blocks per second. Returns None if insufficient data.

Return type:

float or None

class BlockDifferenceTracker(history_size=100)

Bases: object

A class to track and calculate the average block difference over a configurable history size.

This class is useful for monitoring the difference in block heights between nodes or systems.

Parameters:

history_size (int) – The number of recent block differences to track. |default| 100

Example

# Initialize the tracker with a history size of 50
block_tracker = BlockDifferenceTracker(history_size=50)

# Add block differences
block_tracker.add_difference(5)
block_tracker.add_difference(3)

# Get the average block difference
avg_difference = block_tracker.get_average_difference()
print(f"Average Block Difference: {avg_difference}")

Initializes the BlockDifferenceTracker.

Parameters:

history_size (int) – Number of recent block differences to track. |default| 100

__init__(history_size=100)

Initializes the BlockDifferenceTracker.

Parameters:

history_size (int) – Number of recent block differences to track. |default| 100

add_difference(block_difference)

Adds a new block difference to the tracker.

Parameters:

block_difference (int) – The current block difference.

get_average_difference()

Calculates the average block difference.

Returns:

The average of the tracked block differences.

Return type:

float

class LatencyTracker(history_size=100)

Bases: object

A class to track and analyze latency measurements.

This class stores latency values and provides methods to calculate average, minimum, and maximum latencies.

Parameters:

history_size (int) – The number of recent latency measurements to track. |default| 100

Example

# Initialize the tracker with a history size of 100
latency_tracker = LatencyTracker(history_size=100)

# Add latency measurements
latency_tracker.add_latency(120)
latency_tracker.add_latency(150)

# Get latency statistics
avg_latency = latency_tracker.get_average_latency()
min_latency = latency_tracker.get_min_latency()
max_latency = latency_tracker.get_max_latency()

print(f"Average Latency: {avg_latency} ms")
print(f"Min Latency: {min_latency} ms")
print(f"Max Latency: {max_latency} ms")

Initializes the LatencyTracker.

Parameters:

history_size (int) – The number of recent latency measurements to track. |default| 100

__init__(history_size=100)

Initializes the LatencyTracker.

Parameters:

history_size (int) – The number of recent latency measurements to track. |default| 100

add_latency(latency)

Adds a new latency measurement to the tracker.

Parameters:

latency (float) – The measured latency in milliseconds.

get_average_latency()

Calculates the average latency from the tracked measurements.

Returns:

The average latency in milliseconds.

Return type:

float

get_min_latency()

Returns the minimum recorded latency.

Returns:

The minimum latency in milliseconds, or None if no data is available.

Return type:

float or None

get_max_latency()

Returns the maximum recorded latency.

Returns:

The maximum latency in milliseconds, or None if no data is available.

Return type:

float or None

class ErrorRateTracker

Bases: object

A class to track and calculate error rates for API requests or operations.

This class monitors total requests and failed requests to compute an error rate percentage.

Example

# Initialize the tracker
error_tracker = ErrorRateTracker()

# Record requests (success=True for successful requests)
error_tracker.record_request(success=True)
error_tracker.record_request(success=False)

# Get the current error rate
error_rate = error_tracker.get_error_rate()
print(f"Error Rate: {error_rate:.2f}%")

Initializes the ErrorRateTracker.

Tracks total and failed requests for calculating error rates.

__init__()

Initializes the ErrorRateTracker.

Tracks total and failed requests for calculating error rates.

record_request(success=True)

Records a request and whether it was successful or not.

Parameters:

success (bool) – True if the request was successful; False otherwise. Defaults to True. |default| True

get_error_rate()

Calculates the error rate as a percentage of failed requests over total requests.

Returns:

The error rate percentage. Returns 0 if no requests have been recorded.

Return type:

float

class ThroughputTracker(history_size=100)

Bases: object

A class to track and calculate throughput (requests per second).

This class records timestamps of requests and calculates the throughput over a configurable history size.

Parameters:

history_size (int) – The number of recent timestamps to track. |default| 100

Example

# Initialize the tracker with a history size of 100
throughput_tracker = ThroughputTracker(history_size=100)

# Record requests with timestamps
throughput_tracker.record_request()
time.sleep(1)
throughput_tracker.record_request()

# Get the current throughput
throughput = throughput_tracker.get_throughput()
print(f"Throughput: {throughput} requests/second")

Initializes the ThroughputTracker.

Parameters:

history_size (int) – The number of recent timestamps to track. |default| 100

__init__(history_size=100)

Initializes the ThroughputTracker.

Parameters:

history_size (int) – The number of recent timestamps to track. |default| 100

record_request(timestamp=None)

Records the timestamp of a request.

Parameters:

timestamp (float, optional) – The timestamp of the request. Defaults to the current time. |default| None

get_throughput()

Calculates the throughput (requests per second).

Returns:

The calculated throughput. Returns 0 if insufficient data.

Return type:

float

class PeriodicMetricLogger(interval=10)

Bases: object

A class for logging metrics at regular intervals.

This class logs provided metrics only if a specified interval has passed since the last log.

Parameters:

interval (int) – The logging interval in seconds. |default| 10

Example

# Initialize the logger with a 10-second interval
metric_logger = PeriodicMetricLogger(interval=10)

# Log metrics periodically
metrics = {"TPS": 50.5, "Latency": 120}
metric_logger.log_metrics(metrics)

Initializes the PeriodicMetricLogger.

Parameters:

interval (int) – The logging interval in seconds. |default| 10

__init__(interval=10)

Initializes the PeriodicMetricLogger.

Parameters:

interval (int) – The logging interval in seconds. |default| 10

log_metrics(metrics)

Logs the provided metrics if the logging interval has passed.

Parameters:

metrics (dict) – A dictionary of metric names and their values.

class SpikeDetector(threshold, history_size=10)

Bases: object

A class to detect spikes in a series of values.

A spike is detected when the difference between consecutive values exceeds a specified threshold.

Parameters:

• threshold (float) – The change amount considered as a spike.

• history_size (int) – The maximum number of recent values to store. |default| 10

Example

# Initialize the spike detector with a threshold of 10
spike_detector = SpikeDetector(threshold=10, history_size=5)

# Add values and detect spikes
spike_detector.add_value(50)
spike_detector.add_value(65)

is_spike = spike_detector.detect_spike()
print(f"Spike Detected: {is_spike}")

Initializes the SpikeDetector.

Parameters:

• threshold (float) – The change amount considered as a spike.

• history_size (int) – The maximum number of recent values to store. |default| 10

__init__(threshold, history_size=10)

Initializes the SpikeDetector.

Parameters:

• threshold (float) – The change amount considered as a spike.

• history_size (int) – The maximum number of recent values to store. |default| 10

add_value(value)

Adds a new value to the tracker.

Parameters:

value (float) – The new value to track.

detect_spike()

Detects whether a spike occurred based on recent values.

Returns:

True if a spike is detected; False otherwise.

Return type:

bool

class AnomalyDetector(baseline, tolerance=0.2)

Bases: object

A class to detect anomalies in data based on a baseline and tolerance.

An anomaly is detected when a value deviates from the baseline by more than the allowed tolerance.

Parameters:

• baseline (float) – The reference value for normal data (e.g., average TPS).

• tolerance (float) – The acceptable deviation from the baseline as a fraction (e.g., 0.2 for ±20%). Defaults to 0.2 (±20%). |default| 0.2

Example

# Initialize the anomaly detector with a baseline of 50 and tolerance of 20%
anomaly_detector = AnomalyDetector(baseline=50, tolerance=0.2)

# Detect anomalies in new values
is_anomaly = anomaly_detector.detect_anomaly(65)
print(f"Anomaly Detected: {is_anomaly}")

Initializes the AnomalyDetector.

Parameters:

• baseline (float) – The reference value for normal data (e.g., average TPS).

• tolerance (float) – The acceptable deviation from the baseline as a fraction. Defaults to 0.2 (±20%). |default| 0.2

__init__(baseline, tolerance=0.2)

Initializes the AnomalyDetector.

Parameters:

• baseline (float) – The reference value for normal data (e.g., average TPS).

• tolerance (float) – The acceptable deviation from the baseline as a fraction. Defaults to 0.2 (±20%). |default| 0.2

detect_anomaly(value)

Detects whether a given value is an anomaly based on the baseline and tolerance.

Parameters:

value (float) – The value to check for anomalies.

Returns:

True if the value is an anomaly; False otherwise.

Return type:

bool

Example

# Check if a value is an anomaly
is_anomaly = anomaly_detector.detect_anomaly(75)
print(f"Anomaly Detected: {is_anomaly}")

class TrendAnalyzer(history_size=10)

Bases: object

A class to analyze trends in a series of values.

This class tracks a series of values and determines whether the trend is upward, downward, or stable.

Parameters:

history_size (int) – The maximum number of recent values to store for trend analysis. |default| 10

Example

# Initialize the trend analyzer with a history size of 5
trend_analyzer = TrendAnalyzer(history_size=5)

# Add values to the trend analyzer
trend_analyzer.add_value(10)
trend_analyzer.add_value(15)
trend_analyzer.add_value(20)

# Get the current trend
trend = trend_analyzer.get_trend()
print(f"Current Trend: {trend}")

Initializes the TrendAnalyzer.

Parameters:

history_size (int) – The maximum number of recent values to store for trend analysis. |default| 10

__init__(history_size=10)

Initializes the TrendAnalyzer.

Parameters:

history_size (int) – The maximum number of recent values to store for trend analysis. |default| 10

add_value(value)

Adds a new value to the tracker.

Parameters:

value (float or int) – The new value to track.

get_trend()

Calculates the trend based on the stored values.

The trend is determined as: - “upward”: If all differences between consecutive values are positive. - “downward”: If all differences between consecutive values are negative. - “stable”: If there is no consistent upward or downward pattern.

Returns:

The calculated trend (“upward”, “downward”, or “stable”).

Return type:

str

Example

# Analyze the trend
trend = trend_analyzer.get_trend()
print(f"Current Trend: {trend}")

class RollingAverageCalculator(window_size=5)

Bases: object

A class to calculate the rolling (moving) average of a series of values.

This class maintains a fixed-size window of recent values and calculates the average of the values within the window.

Parameters:

window_size (int) – The size of the rolling window. |default| 5

Example

# Initialize the rolling average calculator with a window size of 5
avg_calculator = RollingAverageCalculator(window_size=5)

# Add values to the calculator
avg_calculator.add_value(10)
avg_calculator.add_value(20)
avg_calculator.add_value(30)

# Get the current rolling average
average = avg_calculator.get_average()
print(f"Rolling Average: {average}")

Initializes the RollingAverageCalculator.

Parameters:

window_size (int) – The size of the rolling window. |default| 5

__init__(window_size=5)

Initializes the RollingAverageCalculator.

Parameters:

window_size (int) – The size of the rolling window. |default| 5

add_value(value)

Adds a new value to the rolling window.

Parameters:

value (float or int) – The new value to add.

get_average()

Calculates and returns the rolling average of the stored values.

Returns:

The rolling average. Returns 0 if no values are stored.

Return type:

float

Example

# Calculate the rolling average
average = avg_calculator.get_average()
print(f"Rolling Average: {average}")

class ThresholdNotifier(threshold, alert_callback)

Bases: object

A class to monitor values and trigger a notification when a threshold is exceeded.

This class checks if a given value exceeds a predefined threshold and executes a callback function when the threshold is crossed.

Parameters:

• threshold (float or int) – The threshold value to monitor.

• alert_callback (callable) – A callback function to execute when the threshold is exceeded.

Example

# Define an alert callback function
def alert(value):
    print(f"Alert! Value exceeded: {value}")

# Initialize the notifier with a threshold of 100
notifier = ThresholdNotifier(threshold=100, alert_callback=alert)

# Check values and trigger alerts if necessary
notifier.check_and_notify(120)  # This will trigger the alert
notifier.check_and_notify(80)   # This will not trigger the alert

Initializes the ThresholdNotifier.

Parameters:

• threshold (float or int) – The threshold value to monitor.

• alert_callback (callable) – A callback function to execute when the threshold is exceeded.

__init__(threshold, alert_callback)

Initializes the ThresholdNotifier.

Parameters:

• threshold (float or int) – The threshold value to monitor.

• alert_callback (callable) – A callback function to execute when the threshold is exceeded.

check_and_notify(value)

Checks if the given value exceeds the threshold and triggers the callback if it does.

Parameters:

value (float or int) – The value to check against the threshold.

Example

notifier.check_and_notify(150)  # Triggers the callback if value > threshold

class RateLimiter(max_calls, time_period)

Bases: object

A class to enforce rate limits on operations.

This class ensures that a maximum number of calls can be made within a specified time period. If the limit is exceeded, further calls are not allowed until the time window resets.

Parameters:

• max_calls (int) – The maximum number of calls allowed within the time period.

• time_period (float or int) – The time period (in seconds) for rate limiting.

Example

# Initialize a rate limiter allowing 5 calls per 10 seconds
rate_limiter = RateLimiter(max_calls=5, time_period=10)

# Check if calls are allowed
for i in range(10):
    if rate_limiter.is_allowed():
        print(f"Call {i + 1}: Allowed")
    else:
        print(f"Call {i + 1}: Rate limit exceeded")
    time.sleep(1)  # Simulate time delay between calls

Initializes the RateLimiter.

Parameters:

• max_calls (int) – The maximum number of calls allowed within the time period.

• time_period (float or int) – The time period (in seconds) for rate limiting.

__init__(max_calls, time_period)

Initializes the RateLimiter.

Parameters:

• max_calls (int) – The maximum number of calls allowed within the time period.

• time_period (float or int) – The time period (in seconds) for rate limiting.

is_allowed()

Checks if a new call is allowed under the rate limit.

Removes expired calls from the queue based on the current time and time period, then determines if another call can be made.

Returns:

True if the call is allowed; False otherwise.

Return type:

bool

Example

if rate_limiter.is_allowed():
    print("Call allowed")
else:
    print("Rate limit exceeded")

calculate_reset_percentage(data)

calculate_pruning_percentage(data)