Itertools.groupby в Python: элегантная группировка данных без циклов

itertools.groupby(iterable, key=None)

# Сохранение групп в списки
result = [(key, list(group)) for key, group in itertools.groupby(sorted_data, key_func)]

# Пример с несортированными данными
data = [('A', 1), ('B', 2), ('A', 3), ('C', 4)]

# Группировка без предварительной сортировки
for key, group in itertools.groupby(data, lambda x: x[0]):
print(key, list(group)) # Результат: A [('A', 1)], B [('B', 2)], A [('A', 3)], C [('C', 4)]

# Группировка с предварительной сортировкой
sorted_data = sorted(data, key=lambda x: x[0])
for key, group in itertools.groupby(sorted_data, lambda x: x[0]):
print(key, list(group)) # Результат: A [('A', 1), ('A', 3)], B [('B', 2)], C [('C', 4)]

import itertools

names = ["Anna", "Alex", "Bob", "Brian", "Charlie", "Casey"]
names.sort() # Сортируем список для корректной группировки

for letter, group in itertools.groupby(names, key=lambda x: x[0]):
print(f"Имена на букву {letter}: {', '.join(group)}")

# Результат:
# Имена на букву A: Anna, Alex
# Имена на букву B: Bob, Brian
# Имена на букву C: Charlie, Casey

def compress_sequence(sequence):
return [(key, sum(1 for _ in group)) for key, group in itertools.groupby(sequence)]

original = [1, 1, 1, 2, 2, 3, 3, 3, 3, 1, 1]
compressed = compress_sequence(original)
print(compressed) # [(1, 3), (2, 2), (3, 4), (1, 2)]

# Восстановление последовательности
restored = [item for key, count in compressed for item in [key] * count]
print(restored) # [1, 1, 1, 2, 2, 3, 3, 3, 3, 1, 1]

users = [
{'name': 'John', 'role': 'admin'},
{'name': 'Jane', 'role': 'user'},
{'name': 'Bob', 'role': 'admin'},
{'name': 'Alice', 'role': 'user'}
]

# Сортировка по роли для корректной группировки
sorted_users = sorted(users, key=lambda x: x['role'])

# Группировка пользователей по роли
for role, group in itertools.groupby(sorted_users, key=lambda x: x['role']):
print(f"Роль {role}:")
for user in group:
print(f" – {user['name']}")

def unique_consecutive(iterable, key=None):
"""Оставляет только первый элемент из каждой последовательности дубликатов."""
return [k for k, g in itertools.groupby(iterable, key)]

# Удаление последовательных дубликатов
print(unique_consecutive([1, 1, 2, 3, 3, 3, 4, 1, 1])) # [1, 2, 3, 4, 1]

# Учёт регистра при поиске дубликатов в строках
print(unique_consecutive("AaaBBbCcc", key=str.lower)) # ['A', 'B', 'C']

data = [
('2023', 'Q1', 'Jan', 100),
('2023', 'Q1', 'Feb', 150),
('2023', 'Q1', 'Mar', 200),
('2023', 'Q2', 'Apr', 120),
('2023', 'Q2', 'May', 160),
('2023', 'Q2', 'Jun', 210)
]

# Сортировка по году и кварталу
sorted_data = sorted(data, key=lambda x: (x[0], x[1]))

# Группировка по году
for year, year_group in itertools.groupby(sorted_data, lambda x: x[0]):
print(f"Год: {year}")
# Преобразуем в список для повторного использования
year_data = list(year_group)

# Группировка по кварталу
for quarter, quarter_group in itertools.groupby(year_data, lambda x: x[1]):
total = sum(item[3] for item in quarter_group)
print(f" Квартал {quarter}: {total}")

import itertools

# Обработка данных временного ряда с применением скользящего окна
def find_trends(data, window_size=3):
# Создаём скользящие окна с помощью itertools.islice
windows = []
data_iter = iter(data)
window = list(itertools.islice(data_iter, window_size))

while window:
windows.append(window)
window = window[1:] + list(itertools.islice(data_iter, 1))
if len(window) < window_size:
break

# Определяем тренды для каждого окна
trends = []
for window in windows:
if all(window[i] <= window[i+1] for i in range(len(window)-1)):
trends.append('up')
elif all(window[i] >= window[i+1] for i in range(len(window)-1)):
trends.append('down')
else:
trends.append('mixed')

# Группируем последовательные тренды
result = []
for trend, group in itertools.groupby(zip(range(len(trends)), trends), key=lambda x: x[1]):
indices = [idx for idx, _ in group]
result.append((trend, indices))

return result

# Пример использования
data = [1, 2, 3, 4, 3, 2, 1, 1, 2, 3]
print(find_trends(data))
# Вывод: [('up', [0, 1]), ('down', [3, 4, 5]), ('mixed', [6]), ('up', [7, 8])]

# Анализ структурированных логов
logs = [
{"timestamp": "2023-10-01 10:15:23", "service": "auth", "level": "ERROR", "message": "Authentication failed"},
{"timestamp": "2023-10-01 10:15:24", "service": "auth", "level": "INFO", "message": "User retry"},
{"timestamp": "2023-10-01 10:16:01", "service": "db", "level": "WARNING", "message": "Slow query detected"},
{"timestamp": "2023-10-01 10:17:15", "service": "auth", "level": "ERROR", "message": "Rate limit exceeded"}
]

# Сортировка логов по сервису и уровню
sorted_logs = sorted(logs, key=lambda x: (x["service"], x["level"]))

# Группировка по сервису и уровню
grouped_logs = {}
for (service, level), entries in itertools.groupby(sorted_logs, key=lambda x: (x["service"], x["level"])):
if service not in grouped_logs:
grouped_logs[service] = {}
grouped_logs[service][level] = list(entries)

# Анализ результатов
for service, levels in grouped_logs.items():
print(f"Сервис: {service}")
for level, entries in levels.items():
print(f" {level}: {len(entries)} записей")
if level == "ERROR":
print(" Детали ошибок:")
for entry in entries:
print(f" – {entry['timestamp']}: {entry['message']}")

# Обработка большого файла построчно
def process_log_file(filename):
def parse_line(line):
# Предполагаем формат: timestamp service level message
parts = line.strip().split(' ', 3)
if len(parts) < 4:
return None
return {"timestamp": parts[0], "service": parts[1], "level": parts[2], "message": parts[3]}

# Генератор, читающий и парсящий файл построчно
def log_entries():
with open(filename, 'r') as f:
for line in f:
entry = parse_line(line)
if entry:
yield entry

# Группировка записей по сервисам
entries = sorted(log_entries(), key=lambda x: x["service"])

for service, group in itertools.groupby(entries, key=lambda x: x["service"]):
error_count = 0
warning_count = 0

for entry in group:
if entry["level"] == "ERROR":
error_count += 1
elif entry["level"] == "WARNING":
warning_count += 1

if error_count > 0:
print(f"Сервис {service}: {error_count} ошибок, {warning_count} предупреждений")

# Пример вызова
# process_log_file("application.log")

from operator import itemgetter
import functools

# Анализ продаж по категориям
sales = [
{"product": "Laptop", "category": "Electronics", "price": 1200, "units": 5},
{"product": "Phone", "category": "Electronics", "price": 800, "units": 10},
{"product": "Desk", "category": "Furniture", "price": 350, "units": 2},
{"product": "Chair", "category": "Furniture", "price": 120, "units": 8},
{"product": "Monitor", "category": "Electronics", "price": 300, "units": 3}
]

# Сортировка по категории
sorted_sales = sorted(sales, key=itemgetter("category"))

# Группировка по категории
category_groups = {
category: list(items) 
for category, items in itertools.groupby(sorted_sales, key=itemgetter("category"))
}

# Функциональный подход для анализа каждой категории
def analyze_category(items):
total_revenue = sum(item["price"] * item["units"] for item in items)
avg_price = sum(item["price"] for item in items) / len(items) if items else 0
total_units = sum(item["units"] for item in items)

return {
"total_revenue": total_revenue,
"avg_price": round(avg_price, 2),
"total_units": total_units,
"products": [item["product"] for item in items]
}

# Применение анализа к каждой категории
category_analysis = {
category: analyze_category(items) 
for category, items in category_groups.items()
}

# Вывод результатов
for category, analysis in category_analysis.items():
print(f"Категория: {category}")
print(f" Товары: {', '.join(analysis['products'])}")
print(f" Общая выручка: ${analysis['total_revenue']}")
print(f" Средняя цена: ${analysis['avg_price']}")
print(f" Всего продано: {analysis['total_units']} единиц")

# Предположим, у нас есть данные по частоте букв в тексте
text = "to be or not to be that is the question"
char_counts = [(char, len(list(group))) for char, group in itertools.groupby(sorted(text.replace(" ", "")))]

# Создание простой текстовой гистограммы
for char, count in char_counts:
if char.isalpha(): # Игнорируем не-буквы
bar = '#' * count
print(f"{char}: {bar} ({count})")

# Исходный подход с использованием словарей
def analyze_user_paths_old(logs):
user_paths = {}
for log in logs:
user_id = log["user_id"]
page = log["page"]
if user_id not in user_paths:
user_paths[user_id] = []
user_paths[user_id].append(page)

# Анализ путей каждого пользователя
results = {}
for user_id, path in user_paths.items():
# Создание последовательностей страниц
sequences = []
current_sequence = []
for page in path:
if not current_sequence or page != current_sequence[-1]:
current_sequence.append(page)
if len(current_sequence) == 3:
sequences.append(tuple(current_sequence))
current_sequence = current_sequence[1:]

results[user_id] = sequences

return results

# Оптимизированный подход с `itertools.groupby()`
def analyze_user_paths_new(logs):
# Сортировка логов по пользователю и времени
sorted_logs = sorted(logs, key=lambda x: (x["user_id"], x["timestamp"]))

results = {}
# Группировка по пользователю
for user_id, user_logs in itertools.groupby(sorted_logs, key=lambda x: x["user_id"]):
# Извлечение последовательности страниц
pages = [log["page"] for log in user_logs]

# Удаление последовательных дубликатов
unique_pages = [page for page, _ in itertools.groupby(pages)]

# Создание триплетов страниц (скользящее окно размером 3)
sequences = []
for i in range(len(unique_pages) – 2):
sequences.append(tuple(unique_pages[i:i+3]))

results[user_id] = sequences

return results

# Исходный подход с использованием циклов
def detect_trends_old(price_data, threshold=0.05):
trends = []
current_trend = None
trend_start = 0

for i in range(1, len(price_data)):
price_change = (price_data[i] – price_data[i-1]) / price_data[i-1]

if price_change > threshold:
new_trend = "up"
elif price_change < -threshold:
new_trend = "down"
else:
new_trend = "sideways"

if new_trend != current_trend:
if current_trend is not None:
trends.append({
"trend": current_trend,
"start": trend_start,
"end": i – 1,
"duration": i – trend_start
})
current_trend = new_trend
trend_start = i

# Добавляем последний тренд
if current_trend is not None:
trends.append({
"trend": current_trend,
"start": trend_start,
"end": len(price_data) – 1,
"duration": len(price_data) – trend_start
})

return trends

# Оптимизированный подход с `itertools.groupby()`
def detect_trends_new(price_data, threshold=0.05):
# Вычисляем изменения цен
changes = []
for i in range(1, len(price_data)):
price_change = (price_data[i] – price_data[i-1]) / price_data[i-1]
if price_change > threshold:
changes.append(("up", i))
elif price_change < -threshold:
changes.append(("down", i))
else:
changes.append(("sideways", i))

# Группируем по тренду
trends = []
for trend, group in itertools.groupby(changes, key=lambda x: x[0]):
group = list(group)
start_idx = group[0][1]
end_idx = group[-1][1]
trends.append({
"trend": trend,
"start": start_idx,
"end": end_idx,
"duration": end_idx – start_idx + 1
})

return trends

# Исходное решение
def rle_compress_old(data):
if not data:
return []

result = []
current_char = data[0]
count = 1

for char in data[1:]:
if char == current_char:
count += 1
else:
result.append((current_char, count))
current_char = char
count = 1

# Добавляем последовательность
result.append((current_char, count))
return result

# Оптимизированное решение с `groupby()`
def rle_compress_new(data):
return [(char, sum(1 for _ in group)) for char, group in itertools.groupby(data)]

# Измерение производительности
import time
import random

# Генерация тестовых данных
def generate_test_data(size, runs=50):
chars = ['A', 'B', 'C', 'D', 'E']
data = []
for _ in range(runs):
char = random.choice(chars)
run_length = random.randint(size // (runs * 2), size // runs)
data.extend([char] * run_length)
return data[:size]

# Тестирование производительности
test_data = generate_test_data(1_000_000)

start = time.time()
old_result = rle_compress_old(test_data)
old_time = time.time() – start

start = time.time()
new_result = rle_compress_new(test_data)
new_time = time.time() – start

print(f"Старый алгоритм: {old_time:.4f} с")
print(f"Новый алгоритм: {new_time:.4f} с")
print(f"Ускорение: {(old_time / new_time):.2f}x")

# Проверка корректности
assert old_result == new_result, "Результаты алгоритмов различаются!"