Numpy Quantile в Python: использование функции для анализа данных

import numpy as np

# Создаем тестовый набор данных
data = np.array([1, 3, 5, 7, 9, 11, 13, 15, 17, 19])

# Вычисляем квантили
q_25 = np.quantile(data, 0.25) # 25-й процентиль или 1-й квартиль
q_50 = np.quantile(data, 0.50) # 50-й процентиль или медиана
q_75 = np.quantile(data, 0.75) # 75-й процентиль или 3-й квартиль

print(f"25-й процентиль: {q_25}")
print(f"50-й процентиль (медиана): {q_50}")
print(f"75-й процентиль: {q_75}")

25-й процентиль: 5.0
50-й процентиль (медиана): 10.0
75-й процентиль: 15.0

numpy.quantile(a, q, axis=None, out=None, overwrite_input=False, 
method='linear', keepdims=False)

import numpy as np

# Создаем двумерный массив
data_2d = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9]])

# Вычисляем квантили по разным осям
q_all = np.quantile(data_2d, 0.5) # Квантиль для всего массива
q_rows = np.quantile(data_2d, 0.5, axis=0) # Квантили для каждого столбца
q_cols = np.quantile(data_2d, 0.5, axis=1) # Квантили для каждой строки

print(f"Квантиль для всего массива: {q_all}")
print(f"Квантили для каждого столбца: {q_rows}")
print(f"Квантили для каждой строки: {q_cols}")

Квантиль для всего массива: 5.0
Квантили для каждого столбца: [4\. 5. 6.]
Квантили для каждой строки: [2\. 5. 8.]

import numpy as np

data = np.array([1, 3, 5, 7, 9, 11, 13, 15, 17, 19])

# Вычисляем несколько квантилей одновременно
quantiles = np.quantile(data, [0, 0.25, 0.5, 0.75, 1])
print("Минимум, Q1, Медиана, Q3, Максимум:", quantiles)

Минимум, Q1, Медиана, Q3, Максимум: [ 1. 5. 10. 15. 19.]

import numpy as np

data = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])

methods = ['linear', 'lower', 'higher', 'midpoint', 'nearest']
q = 0.35 # 35-й процентиль

for method in methods:
result = np.quantile(data, q, method=method)
print(f"Метод {method}: {result}")

import numpy as np

data = np.array([15, 20, 35, 40, 50, 55, 60, 70, 80, 85, 90, 95, 100, 110, 120])

# Расчет процентилей
p25 = np.percentile(data, 25)
p50 = np.percentile(data, 50)
p75 = np.percentile(data, 75)
p90 = np.percentile(data, 90)
p95 = np.percentile(data, 95)

print(f"25-й процентиль: {p25}")
print(f"50-й процентиль (медиана): {p50}")
print(f"75-й процентиль: {p75}")
print(f"90-й процентиль: {p90}")
print(f"95-й процентиль: {p95}")

import numpy as np
import matplotlib.pyplot as plt

# Имитируем данные о времени загрузки страницы (в мс)
response_times = np.random.lognormal(mean=3.0, sigma=0.5, size=1000)

# Вычисляем ключевые процентили
percentiles = [50, 75, 90, 95, 99]
results = np.percentile(response_times, percentiles)

# Создаем словарь для удобства отображения
percentile_dict = {f"p{p}": round(val, 2) for p, val in zip(percentiles, results)}

print("Процентили времени загрузки страницы (мс):")
for key, value in percentile_dict.items():
print(f"{key}: {value}")

# Визуализация данных
plt.figure(figsize=(10, 6))
plt.hist(response_times, bins=50, alpha=0.7, color='skyblue')
for p, value in zip(percentiles, results):
plt.axvline(x=value, color='r', linestyle='--', alpha=0.7)
plt.text(value+1, plt.ylim()[1]*0.9, f'p{p}: {value:.2f}ms', rotation=90)

plt.title('Распределение времени загрузки страницы с процентилями')
plt.xlabel('Время загрузки (мс)')
plt.ylabel('Частота')
plt.grid(alpha=0.3)

import numpy as np

data = np.array([15, 20, 35, 40, 50, 55, 60, 70, 80, 85, 90, 95, 100, 110, 120, 200, 300])

# Расчет квартилей и IQR
q1 = np.percentile(data, 25)
q3 = np.percentile(data, 75)
iqr = q3 – q1

# Определение границ выбросов
lower_bound = q1 – 1.5 * iqr
upper_bound = q3 + 1.5 * iqr

# Поиск выбросов
outliers = data[(data < lower_bound) | (data > upper_bound)]

print(f"Q1: {q1}")
print(f"Q3: {q3}")
print(f"IQR: {iqr}")
print(f"Нижняя граница: {lower_bound}")
print(f"Верхняя граница: {upper_bound}")
print(f"Выбросы: {outliers}")

import numpy as np
import matplotlib.pyplot as plt

# Сгенерируем данные из нескольких распределений для сравнения
np.random.seed(42)
data1 = np.random.normal(loc=5, scale=1, size=100)
data2 = np.random.exponential(scale=1, size=100)
data3 = np.random.lognormal(mean=0, sigma=0.5, size=100)

# Вычисляем статистики для боксплота вручную с помощью np.quantile
def get_boxplot_stats(data):
q1 = np.quantile(data, 0.25)
median = np.quantile(data, 0.5)
q3 = np.quantile(data, 0.75)
iqr = q3 – q1
lower_bound = max(np.min(data), q1 – 1.5 * iqr)
upper_bound = min(np.max(data), q3 + 1.5 * iqr)
return {
'q1': q1,
'median': median,
'q3': q3,
'lower_whisker': lower_bound,
'upper_whisker': upper_bound
}

# Выводим статистики
stats1 = get_boxplot_stats(data1)
stats2 = get_boxplot_stats(data2)
stats3 = get_boxplot_stats(data3)

print("Нормальное распределение:", stats1)
print("Экспоненциальное распределение:", stats2)
print("Логнормальное распределение:", stats3)

# Построение боксплотов
plt.figure(figsize=(10, 6))
plt.boxplot([data1, data2, data3], labels=['Нормальное', 'Экспоненциальное', 'Логнормальное'])
plt.title('Сравнение распределений с помощью боксплотов')
plt.ylabel('Значение')
plt.grid(alpha=0.3)

import numpy as np
from statsmodels.regression.quantile_regression import QuantReg
import matplotlib.pyplot as plt

# Генерируем данные с гетероскедастичностью
np.random.seed(42)
n = 200
x = np.random.uniform(0, 10, size=n)
# Дисперсия ошибки увеличивается с ростом x
e = np.random.normal(0, 0.5 + 0.5 * x, size=n)
y = 2 + 0.5 * x + e

# Подготавливаем данные для регрессии
X = np.column_stack((np.ones(n), x))

# Вычисляем обычную МНК-регрессию
beta_ols = np.linalg.lstsq(X, y, rcond=None)[0]

# Вычисляем квантильную регрессию для разных квантилей
quantiles = [0\.1, 0.25, 0.5, 0.75, 0.9]
models = {}

for q in quantiles:
model = QuantReg(y, X)
result = model.fit(q=q)
models[q] = result.params

# Строим график
plt.figure(figsize=(12, 8))
plt.scatter(x, y, alpha=0.6, label='Наблюдения')

# Добавляем линию МНК
x_plot = np.linspace(0, 10, 100)
y_ols = beta_ols[0] + beta_ols[1] * x_plot
plt.plot(x_plot, y_ols, 'r-', lw=2, label='OLS (среднее)')

# Добавляем линии квантильной регрессии
colors = ['blue', 'green', 'purple', 'orange', 'cyan']
for i, q in enumerate(quantiles):
y_qr = models[q][0] + models[q][1] * x_plot
plt.plot(x_plot, y_qr, color=colors[i], linestyle='--', 
label=f'Квантиль {q}')

plt.title('Сравнение OLS и квантильной регрессии')
plt.xlabel('X')
plt.ylabel('Y')
plt.legend()
plt.grid(alpha=0.3)

import numpy as np

# Сгенерируем данные с выбросами
np.random.seed(42)
data = np.concatenate([
np.random.normal(loc=50, scale=10, size=95), # Основная часть данных
np.random.normal(loc=100, scale=10, size=5) # Выбросы
])

# Вычисляем квартили
q1 = np.quantile(data, 0.25)
q3 = np.quantile(data, 0.75)
iqr = q3 – q1

# Определяем границы выбросов (метод Тьюки)
lower_bound = q1 – 1.5 * iqr
upper_bound = q3 + 1.5 * iqr

# Фильтруем данные
filtered_data = data[(data >= lower_bound) & (data <= upper_bound)]

print(f"Исходный размер данных: {len(data)}")
print(f"Размер отфильтрованных данных: {len(filtered_data)}")
print(f"Удалено выбросов: {len(data) – len(filtered_data)}")
print(f"Границы фильтрации: [{lower_bound:.2f}, {upper_bound:.2f}]")

import numpy as np
import time
from concurrent.futures import ProcessPoolExecutor
import pandas as pd

# Функция для измерения времени выполнения
def time_quantile_calculation(func, repeats=5):
times = []
for _ in range(repeats):
start_time = time.time()
_ = func()
end_time = time.time()
times.append(end_time – start_time)
return np.mean(times)

# Генерируем большой массив данных
np.random.seed(42)
data_large = np.random.normal(size=10**7)

# 1. Стандартный расчет
def standard_quantile():
return np.quantile(data_large, [0\.25, 0.5, 0.75])

# 2. Расчет с overwrite_input=True
def overwrite_quantile():
# Создаем копию для корректного измерения в цикле
data_copy = data_large.copy()
return np.quantile(data_copy, [0\.25, 0.5, 0.75], overwrite_input=True)

# 3. Расчет по чанкам (приближение)
def chunked_quantile(chunk_size=10**6):
chunks = [data_large[i:i+chunk_size] for i in range(0, len(data_large), chunk_size)]
# Вычисляем квантили для каждого чанка
chunk_quantiles = np.array([np.quantile(chunk, [0\.25, 0.5, 0.75]) for chunk in chunks])
# Усредняем квантили по чанкам (приближение)
return chunk_quantiles.mean(axis=0)

# 4. Параллельный расчет по чанкам
def parallel_chunked_quantile(chunk_size=10**6):
chunks = [data_large[i:i+chunk_size] for i in range(0, len(data_large), chunk_size)]

def process_chunk(chunk):
return np.quantile(chunk, [0\.25, 0.5, 0.75])

with ProcessPoolExecutor() as executor:
results = list(executor.map(process_chunk, chunks))

return np.mean(results, axis=0)

# 5. Использование pandas (может быть быстрее для некоторых операций)
def pandas_quantile():
return pd.Series(data_large).quantile([0\.25, 0.5, 0.75]).values

# Измеряем время выполнения каждого метода
times = {
"Стандартный np.quantile()": time_quantile_calculation(standard_quantile),
"С overwrite_input=True": time_quantile_calculation(overwrite_quantile),
"По чанкам": time_quantile_calculation(chunked_quantile),
"Параллельный по чанкам": time_quantile_calculation(parallel_chunked_quantile),
"Pandas": time_quantile_calculation(pandas_quantile)
}

# Выводим результаты
for method, elapsed_time in times.items():
print(f"{method}: {elapsed_time:.4f} секунд")

# Пример с использованием t-digest (требуется установка: pip install tdigest)
import numpy as np
from tdigest import TDigest

# Создаем большой массив данных
np.random.seed(42)
data_stream = np.random.normal(size=10**7)

# Инициализируем t-digest
digest = TDigest()

# Обрабатываем данные порциями
chunk_size = 10**5
for i in range(0, len(data_stream), chunk_size):
chunk = data_stream[i:i+chunk_size]
digest.batch_update(chunk)

# Получаем приблизительные квантили
approx_q25 = digest.quantile(0.25)
approx_median = digest.quantile(0.5)
approx_q75 = digest.quantile(0.75)

# Сравниваем с точными значениями
exact_quantiles = np.quantile(data_stream, [0\.25, 0.5, 0.75])

print("Приблизительные квантили (t-digest):")
print(f"Q25: {approx_q25:.4f}, Медиана: {approx_median:.4f}, Q75: {approx_q75:.4f}")

print("\nТочные квантили (np.quantile):")
print(f"Q25: {exact_quantiles[0]:.4f}, Медиана: {exact_quantiles[1]:.4f}, Q75: {exact_quantiles[2]:.4f}")

# Вычисляем относительную ошибку аппроксимации
errors = [abs(a – e) / abs(e) * 100 for a, e in zip(
[approx_q25, approx_median, approx_q75], exact_quantiles)]

print("\nОтносительная ошибка аппроксимации (%):")
print(f"Q25: {errors[0]:.2f}%, Медиана: {errors[1]:.2f}%, Q75: {errors[2]:.2f}%")