Itertools в Python: оптимизация кода с мощными итераторами

from itertools import chain

list1 = [1, 2, 3]
list2 = ['a', 'b', 'c']
list3 = [True, False]

# Обычный подход
result_standard = []
for item in list1:
result_standard.append(item)
for item in list2:
result_standard.append(item)
for item in list3:
result_standard.append(item)

# С использованием itertools
result_itertools = list(chain(list1, list2, list3))

print(result_itertools) # [1, 2, 3, 'a', 'b', 'c', True, False]

from itertools import count, islice

# Создаем бесконечную последовательность чисел с шагом 5
counter = count(10, 5)

# Берем первые 5 элементов
first_five = list(islice(counter, 5))
print(first_five) # [10, 15, 20, 25, 30]

# Применение в генерации ID для записей в БД
records = ["Запись A", "Запись B", "Запись C"]
records_with_ids = list(zip(count(1), records))
print(records_with_ids) # [(1, 'Запись A'), (2, 'Запись B'), (3, 'Запись C')]

from itertools import cycle, islice

# Циклическое повторение дней недели
weekdays = ['Пн', 'Вт', 'Ср', 'Чт', 'Пт', 'Сб', 'Вс']
day_cycle = cycle(weekdays)

# Получаем 10 дней подряд
next_ten_days = list(islice(day_cycle, 10))
print(next_ten_days) # ['Пн', 'Вт', 'Ср', 'Чт', 'Пт', 'Сб', 'Вс', 'Пн', 'Вт', 'Ср']

# Ротация серверов для балансировки нагрузки
servers = ['server1', 'server2', 'server3']
server_pool = cycle(servers)

for _ in range(8):
print(f"Запрос отправлен на {next(server_pool)}")

from itertools import repeat, zip_longest

# Повторяем значение 10 раз
repeated = list(repeat("Python", 5))
print(repeated) # ['Python', 'Python', 'Python', 'Python', 'Python']

# Заполнение недостающих значений в списках разной длины
names = ['Алиса', 'Боб', 'Карл']
default_score = 0

# Создаем структуру "имя: балл по умолчанию"
scores = list(zip(names, repeat(default_score)))
print(scores) # [('Алиса', 0), ('Боб', 0), ('Карл', 0)]

# С указанием конкретных баллов для имеющихся результатов
actual_scores = [42, 65]
all_scores = list(zip_longest(names, actual_scores, fillvalue=0))
print(all_scores) # [('Алиса', 42), ('Боб', 65), ('Карл', 0)]

from itertools import permutations

# Все возможные комбинации пин-кода из 3 цифр
digits = [1, 2, 3]
pin_combinations = list(permutations(digits))
print(pin_combinations) 
# [(1, 2, 3), (1, 3, 2), (2, 1, 3), (2, 3, 1), (3, 1, 2), (3, 2, 1)]

# Перестановки 2 элементов из списка букв
letters = ['A', 'B', 'C', 'D']
two_letter_codes = list(permutations(letters, 2))
print(f"Всего комбинаций: {len(two_letter_codes)}") # Всего комбинаций: 12
print(two_letter_codes[:5]) # Выводим первые 5 комбинаций

from itertools import combinations

# Все возможные пары игроков из команды
team = ['Игрок1', 'Игрок2', 'Игрок3', 'Игрок4']
pairs = list(combinations(team, 2))
print(f"Всего возможных пар: {len(pairs)}") # Всего возможных пар: 6
print(pairs) 
# [('Игрок1', 'Игрок2'), ('Игрок1', 'Игрок3'), ('Игрок1', 'Игрок4'), 
# ('Игрок2', 'Игрок3'), ('Игрок2', 'Игрок4'), ('Игрок3', 'Игрок4')]

# Формирование тестовых выборок разного размера
test_cases = ['case1', 'case2', 'case3', 'case4', 'case5']
for i in range(1, 4):
sample = list(combinations(test_cases, i))
print(f"Комбинации размера {i}: {len(sample)} вариантов")

from itertools import combinations_with_replacement

# Возможные комбинации кубиков с повторениями
dice_values = [1, 2, 3, 4, 5, 6]
dice_rolls = list(combinations_with_replacement(dice_values, 2))
print(f"Всего возможных комбинаций: {len(dice_rolls)}") # 21

# Генерация всех возможных инвестиционных портфелей из 3 активов
assets = ['Акции', 'Облигации', 'Золото']
portfolios = list(combinations_with_replacement(assets, 3))
print(portfolios)
# [('Акции', 'Акции', 'Акции'), ('Акции', 'Акции', 'Облигации'),... и т.д.]

from itertools import product

# Все возможные комбинации при броске двух кубиков
dice1 = [1, 2, 3, 4, 5, 6]
dice2 = [1, 2, 3, 4, 5, 6]
all_throws = list(product(dice1, dice2))
print(f"Всего комбинаций: {len(all_throws)}") # 36

# Генерация всех возможных RGB цветов (упрощенно)
rgb_values = [0, 128, 255] # Упрощенно для примера
rgb_colors = list(product(rgb_values, repeat=3))
print(f"Возможных цветов: {len(rgb_colors)}") # 27

# Первые 5 цветов
for r, g, b in rgb_colors[:5]:
print(f"RGB: ({r}, {g}, {b})")

from itertools import chain

# Объединение данных из разных источников
users_db1 = [{'id': 1, 'name': 'Алиса'}, {'id': 2, 'name': 'Боб'}]
users_db2 = [{'id': 3, 'name': 'Виктор'}, {'id': 4, 'name': 'Галина'}]
users_cache = [{'id': 5, 'name': 'Дмитрий'}]

# Плохой подход: создание временного списка
all_users_bad = users_db1 + users_db2 + users_cache

# Хороший подход с itertools
all_users = chain(users_db1, users_db2, users_cache)

# Фильтрация по всем источникам без создания промежуточного списка
filtered_users = [user for user in all_users if len(user['name']) > 5]
print(filtered_users) # [{'id': 3, 'name': 'Виктор'}, {'id': 4, 'name': 'Галина'}, {'id': 5, 'name': 'Дмитрий'}]

# Обработка файлов разных форматов
def process_logs():
error_logs = open('error.log')
access_logs = open('access.log')
system_logs = open('system.log')

# Обработка всех логов одним потоком
for log_line in chain(error_logs, access_logs, system_logs):
if "ERROR" in log_line:
print(f"Найдена ошибка: {log_line.strip()}")

# Закрытие файлов
for log_file in [error_logs, access_logs, system_logs]:
log_file.close()

from itertools import groupby
from operator import itemgetter

# Группировка транзакций по дате
transactions = [
{'date': '2023-01-15', 'amount': 200, 'category': 'Еда'},
{'date': '2023-01-15', 'amount': 50, 'category': 'Транспорт'},
{'date': '2023-01-16', 'amount': 100, 'category': 'Развлечения'},
{'date': '2023-01-16', 'amount': 300, 'category': 'Покупки'},
{'date': '2023-01-17', 'amount': 150, 'category': 'Еда'}
]

# Сортировка необходима перед groupby!
transactions.sort(key=itemgetter('date'))

# Группировка и подсчет суммы по дням
for date, group in groupby(transactions, key=itemgetter('date')):
daily_transactions = list(group)
total_amount = sum(t['amount'] for t in daily_transactions)
print(f"Дата: {date}, Количество транзакций: {len(daily_transactions)}, "
f"Общая сумма: {total_amount}")

# Группировка строк лога по уровню важности
log_lines = [
"INFO: Сервер запущен",
"INFO: Получен запрос",
"WARNING: Высокая нагрузка ЦП",
"ERROR: База данных недоступна",
"ERROR: Таймаут соединения"
]

# Сортировка строк по префиксу
sorted_logs = sorted(log_lines, key=lambda x: x.split(":")[0])

# Группировка и подсчет
for level, entries in groupby(sorted_logs, key=lambda x: x.split(":")[0]):
entries_list = list(entries)
print(f"Уровень: {level}, Количество записей: {len(entries_list)}")
for entry in entries_list:
print(f" – {entry}")

from itertools import islice, count

# Чтение только определенных строк из большого файла
def read_specific_lines(filename, start, stop, step=1):
with open(filename, 'r') as file:
for line in islice(file, start, stop, step):
yield line.strip()

# Использование: чтение строк с 100 по 200 с шагом 10
# for line in read_specific_lines('huge_log.txt', 100, 200, 10):
# print(line)

# Получение элементов из бесконечной последовательности
infinite_numbers = count(1) # 1, 2, 3, ... бесконечно

# Получаем числа с 1000 по 1020
thousand_range = list(islice(infinite_numbers, 999, 1020))
print(thousand_range) # [1000, 1001, ..., 1019]

# Обработка больших наборов данных пакетами
def process_large_dataset(dataset, batch_size=1000):
# Создаем итератор по датасету
it = iter(dataset)

# Обрабатываем данные батчами
while True:
# Берем следующую порцию данных
batch = list(islice(it, batch_size))
if not batch:
break

# Здесь обработка пакета
print(f"Обработано {len(batch)} записей")

# Если получили меньше записей, чем ожидали, значит достигли конца
if len(batch) < batch_size:
break

from itertools import islice, tee

# Реализация скользящего окна (sliding window)
def sliding_window(iterable, n):
"""Создает скользящие окна размера n из итерируемого объекта."""
iterables = tee(iterable, n)

for i, it in enumerate(iterables):
for _ in range(i):
next(it, None)

return zip(*iterables)

# Применение: находим скользящие средние в списке цен
prices = [100, 105, 110, 108, 115, 120, 125]
windows = list(sliding_window(prices, 3))
print("Окна размера 3:", windows) # [(100, 105, 110), (105, 110, 108), ...]

# Вычисляем скользящие средние
moving_averages = [sum(window)/len(window) for window in windows]
print("Скользящие средние:", moving_averages) # [105\.0, 107.67, ...]

from itertools import combinations, chain, groupby
import time

# Задача: найти все пары элементов с заданной суммой
def find_pairs_naive(numbers, target_sum):
"""Наивная реализация поиска пар с заданной суммой."""
result = []
n = len(numbers)

for i in range(n):
for j in range(i+1, n):
if numbers[i] + numbers[j] == target_sum:
result.append((numbers[i], numbers[j]))

return result

def find_pairs_optimized(numbers, target_sum):
"""Оптимизированная реализация с использованием combinations."""
return [(a, b) for a, b in combinations(numbers, 2) if a + b == target_sum]

# Сравнение производительности
test_numbers = list(range(1000))
target = 500

# Тест производительности
start_time = time.time()
naive_result = find_pairs_naive(test_numbers, target)
naive_time = time.time() – start_time
print(f"Наивный подход: найдено {len(naive_result)} пар за {naive_time:.4f} сек")

start_time = time.time()
optimized_result = find_pairs_optimized(test_numbers, target)
optimized_time = time.time() – start_time
print(f"Оптимизированный подход: найдено {len(optimized_result)} пар за {optimized_time:.4f} сек")
print(f"Ускорение в {naive_time/optimized_time:.2f} раз")

from itertools import chain, islice, groupby, combinations
from collections import Counter

# Задача: найти наиболее часто встречающиеся N-граммы в тексте
def find_top_ngrams(text, n, top=5):
"""
Находит top наиболее часто встречающихся последовательностей из n слов.
"""
# Разбиваем текст на слова и очищаем от пунктуации
words = [word.strip('.,!?:;()[]{}"\'-').lower()
for word in text.split() if word.strip('.,!?:;()[]{}"\'-')]

# Создаем N-граммы с помощью скользящего окна
ngrams = sliding_window(words, n)

# Подсчитываем частоту каждой N-граммы
ngram_freq = Counter(map(tuple, ngrams))

# Возвращаем top наиболее часто встречающихся N-грамм
return ngram_freq.most_common(top)

# Пример текста
sample_text = """
Python — это мощный язык программирования. Python прост в освоении.
Python имеет эффективные структуры данных высокого уровня. Python использует 
элегантный синтаксис, что делает программы написанные на нём легкочитаемыми.
"""

# Находим самые частые биграммы (пары слов)
top_bigrams = find_top_ngrams(sample_text, 2, 3)
print("Топ-3 биграммы:")
for ngram, count in top_bigrams:
print(f" {' '.join(ngram)}: {count} раз")

from itertools import groupby, chain
import csv
from operator import itemgetter

# Создаем конвейер обработки данных
def data_pipeline():
# 1. Чтение данных из нескольких источников
def read_csv_files(filenames):
for filename in filenames:
with open(filename, 'r') as file:
reader = csv.DictReader(file)
for row in reader:
yield row

# 2. Фильтрация данных
def filter_data(rows, min_value):
for row in rows:
if float(row.get('value', 0)) >= min_value:
yield row

# 3. Трансформация данных
def transform_data(rows):
for row in rows:
# Добавляем вычисляемое поле
row['doubled_value'] = float(row['value']) * 2
yield row

# 4. Группировка и агрегация
def aggregate_by_category(rows):
# Сначала сортируем по категории
sorted_rows = sorted(rows, key=itemgetter('category'))

# Затем группируем и агрегируем
for category, group in groupby(sorted_rows, key=itemgetter('category')):
group_list = list(group)
total = sum(float(row['value']) for row in group_list)
avg = total / len(group_list)
yield {
'category': category,
'count': len(group_list),
'total': total,
'average': avg
}

# Соединение этапов в конвейер
files = ['sales_q1.csv', 'sales_q2.csv', 'sales_q3.csv']
data = read_csv_files(files)
filtered_data = filter_data(data, 100) # Фильтр по минимальному значению
transformed_data = transform_data(filtered_data)
results = aggregate_by_category(transformed_data)

return list(results) # В реальном коде можно возвращать итератор

# В реальной ситуации вызов будет примерно такой:
# results = data_pipeline()
# for result in results:
# print(f"Категория: {result['category']}, "
# f"Количество: {result['count']}, "
# f"Среднее значение: {result['average']:.2f}")