pandas для бэкенд-разработчика: логи, Excel/1С, ETL и отчёты

# Задача: найти топ-10 эндпоинтов по времени ответа за последнюю неделю
# Вариант 1: Чистый Python (15+ строк с циклами, dict comprehensions)
# Вариант 2: SQL запрос + ORM (нужно писать миграцию для analytics таблицы)
# Вариант 3: pandas (3 строки кода)
 
import pandas as pd
 
df = pd.read_csv('api_logs.csv')
top_slow = df.nlargest(10, 'response_time')[['endpoint', 'response_time']]
print(top_slow)

# ❌ ПЛОХО: запись в БД через pandas
df.to_sql('users', engine, if_exists='append')
# Проблемы: медленно, нет поддержки транзакций, сложная обработка ошибок
 
# ✅ ХОРОШО: используйте ORM или bulk insert
session.bulk_insert_mappings(User, records)

# ❌ ПЛОХО: загрузка 10GB файла
df = pd.read_csv('huge_file.csv')  # OOM (Out of Memory)
 
# ✅ ХОРОШО: используйте chunked reading, Dask или DuckDB
for chunk in pd.read_csv('huge_file.csv', chunksize=10000):
    process_chunk(chunk)

# ❌ ПЛОХО: pandas в FastAPI endpoint для каждого запроса
@app.get("/stats")
def get_stats():
    df = pd.read_sql("SELECT * FROM orders", engine)  # Медленно!
    return df.groupby('product').sum().to_dict()
 
# ✅ ХОРОШО: предрассчитайте агрегаты или используйте SQL/Redis

# ❌ ПЛОХО: изменение данных через pandas с сохранением в БД
df = pd.read_sql("SELECT * FROM accounts", engine)
df.loc[df['id'] == 123, 'balance'] -= 100  # Нет транзакции!
df.to_sql('accounts', engine, if_exists='replace')  # Потеря данных при ошибке, нет аудита
 
# ✅ ХОРОШО: используйте SQL транзакции с записью операции
with session.begin():
    # Блокируем счёт для обновления
    account = session.query(Account).filter_by(id=123).with_for_update().first()
    account.balance -= 100
 
    # Записываем транзакцию для аудита
    transaction = Transaction(
        account_id=123,
        amount=-100,
        type='withdrawal',
        timestamp=datetime.now()
    )
    session.add(transaction)

# ❌ ПЛОХО: загружаем всё в pandas для простой фильтрации
df = pd.read_sql("SELECT * FROM users", engine)
active_users = df[df['is_active'] == True]
 
# ✅ ХОРОШО: используйте SQL
df = pd.read_sql("SELECT * FROM users WHERE is_active = TRUE", engine)

import pandas as pd
 
# Создание DataFrame из dict
data = {
    'user_id': [1, 2, 3, 4],
    'name': ['Alice', 'Bob', 'Charlie', 'Diana'],
    'age': [25, 30, 35, 28],
    'city': ['Moscow', 'SPb', 'Moscow', 'Kazan']
}
 
df = pd.DataFrame(data)
print(df)
 
#    user_id     name  age     city
# 0        1    Alice   25   Moscow
# 1        2      Bob   30      SPb
# 2        3  Charlie   35   Moscow
# 3        4    Diana   28    Kazan

# Получение колонки
ages = df['age']
print(type(ages))  # <class 'pandas.core.series.Series'>
 
# Операции над Series
print(ages.mean())     # 29.5 (среднее)
print(ages.max())      # 35
print(ages.std())      # 4.2 (стандартное отклонение)

# Фильтрация
moscow_users = df[df['city'] == 'Moscow']
 
# Сортировка
sorted_by_age = df.sort_values('age', ascending=False)
 
# Выбор колонок
names_and_ages = df[['name', 'age']]
 
# Добавление колонки
df['is_adult'] = df['age'] >= 18
 
# Группировка и агрегация
city_stats = df.groupby('city')['age'].agg(['mean', 'count'])

import pandas as pd
from datetime import datetime
 
# 1. Чтение логов (предположим, они в CSV)
df = pd.read_csv(
    'access.log',
    sep='|',
    names=['timestamp', 'method_path', 'status', 'response_time'],
    parse_dates=['timestamp']  # Автоматический парсинг дат
)
 
# 2. Очистка данных (убираем пробелы!)
df['response_time'] = df['response_time'].str.strip().str.replace('ms', '').astype(int)
df[['method', 'endpoint']] = df['method_path'].str.strip().str.split(' ', expand=True)
df['status'] = df['status'].astype(int)
# .str.strip() критичен: логи часто содержат пробелы вокруг разделителей
 
# 3. Топ-10 медленных эндпоинтов
top_slow = (
    df.groupby('endpoint')['response_time']
    .agg(['mean', 'max', 'count'])
    .sort_values('mean', ascending=False)
    .head(10)
)
 
print("Топ-10 медленных эндпоинтов:")
print(top_slow)
 
# 4. Эндпоинты с высоким процентом 5xx ошибок (векторизовано!)
error_rate = (
    df.assign(is_5xx=df['status'] >= 500)
    .groupby('endpoint')['is_5xx']
    .mean()
    .mul(100)
    .sort_values(ascending=False)
    .head(10)
)
 
print("\nЭндпоинты с высоким % ошибок:")
print(error_rate)
 
# 5. Распределение времени ответа по часам
df['hour'] = df['timestamp'].dt.hour
hourly_stats = df.groupby('hour')['response_time'].agg(['mean', 'median', 'count'])
 
print("\nРаспределение по часам:")
print(hourly_stats)

import matplotlib.pyplot as plt
 
# График среднего времени ответа по часам
hourly_stats['mean'].plot(kind='line', title='Среднее время ответа по часам')
plt.xlabel('Час дня')
plt.ylabel('Среднее время (ms)')
plt.savefig('response_time_by_hour.png')

import pandas as pd
 
# 1. Загрузка данных из Excel (разные листы)
sales = pd.read_excel('sales.xlsx', sheet_name='Продажи')
stock = pd.read_excel('sales.xlsx', sheet_name='Остатки')
costs = pd.read_excel('sales.xlsx', sheet_name='Себестоимость')
 
# 2. Очистка названий колонок (1С любит пробелы и кириллицу)
sales.columns = sales.columns.str.strip()
stock.columns = stock.columns.str.strip()
costs.columns = costs.columns.str.strip()
 
# 3. Объединение данных
# Добавляем себестоимость к продажам
df = sales.merge(
    costs[['Артикул', 'Себестоимость']],
    on='Артикул',
    how='left'
)
 
# 4. Расчёт маржинальности (с защитой от деления на 0)
df['Маржа'] = df['Цена продажи'] - df['Себестоимость']
df['Маржинальность %'] = (
    df['Маржа'] / df['Цена продажи'].replace(0, float('nan')) * 100
).round(2)
 
# 5. Анализ: топ товаров по маржинальности
top_margin = (
    df.groupby(['Категория', 'Название товара'])
    .agg({
        'Количество': 'sum',
        'Маржа': 'sum',
        'Маржинальность %': 'mean'
    })
    .sort_values('Маржа', ascending=False)
    .head(20)
)
 
print("Топ-20 товаров по марже:")
print(top_margin)
 
# 6. Анализ: убыточные товары
unprofitable = df[df['Маржа'] < 0].groupby('Категория').agg({
    'Название товара': 'count',
    'Маржа': 'sum'
})
 
print("\nУбыточные товары по категориям:")
print(unprofitable)
 
# 7. Сводная таблица для руководства (требуется колонка "Дата" в исходных данных!)
pivot = df.pivot_table(
    values=['Количество', 'Маржа'],
    index='Категория',
    columns=pd.to_datetime(df['Дата']).dt.month,
    aggfunc='sum',
    fill_value=0
)
 
print("\nПродажи и маржа по месяцам:")
print(pivot)
 
# 8. Экспорт результатов в новый Excel
with pd.ExcelWriter('analysis_results.xlsx') as writer:
    top_margin.to_excel(writer, sheet_name='Топ товары')
    unprofitable.to_excel(writer, sheet_name='Убыточные')
    pivot.to_excel(writer, sheet_name='Динамика по месяцам')
 
print("\n✅ Анализ завершён: analysis_results.xlsx")

import logging
from pathlib import Path
from datetime import datetime
from typing import Optional
import pandas as pd
 
# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
 
class SalesAnalysisError(Exception):
    """Ошибка при анализе продаж."""
    pass
 
def analyze_1c_sales(
    file_path: str,
    output_dir: str = 'reports',
    required_sheets: Optional[list] = None
) -> str:
    """
    Автоматический анализ продаж из выгрузки 1С с обработкой ошибок.
 
    Args:
        file_path: Путь к Excel-файлу с выгрузкой из 1С
        output_dir: Директория для сохранения отчёта
        required_sheets: Список обязательных листов (по умолчанию: Продажи, Себестоимость)
 
    Returns:
        Путь к созданному отчёту
 
    Raises:
        SalesAnalysisError: При ошибках валидации или обработки данных
    """
    if required_sheets is None:
        required_sheets = ['Продажи', 'Себестоимость']
 
    try:
        logger.info(f"Начинаем анализ файла: {file_path}")
 
        # 1. Проверка существования файла
        if not Path(file_path).exists():
            raise SalesAnalysisError(f"Файл не найден: {file_path}")
 
        # 2. Создаём директорию для отчётов
        Path(output_dir).mkdir(parents=True, exist_ok=True)
 
        # 3. Загрузка данных с валидацией
        try:
            excel_file = pd.ExcelFile(file_path)
            available_sheets = excel_file.sheet_names
 
            # Проверяем наличие обязательных листов
            missing_sheets = set(required_sheets) - set(available_sheets)
            if missing_sheets:
                raise SalesAnalysisError(
                    f"Отсутствуют обязательные листы: {missing_sheets}. "
                    f"Доступные листы: {available_sheets}"
                )
 
            sales = pd.read_excel(excel_file, sheet_name='Продажи')
            costs = pd.read_excel(excel_file, sheet_name='Себестоимость')
 
        except ValueError as e:
            raise SalesAnalysisError(f"Ошибка чтения Excel-файла: {e}")
 
        # 4. Валидация структуры данных
        required_sales_cols = ['Артикул', 'Количество', 'Цена продажи', 'Категория', 'Название товара']
        required_costs_cols = ['Артикул', 'Себестоимость']
 
        missing_sales_cols = set(required_sales_cols) - set(sales.columns)
        missing_costs_cols = set(required_costs_cols) - set(costs.columns)
 
        if missing_sales_cols:
            raise SalesAnalysisError(f"В листе 'Продажи' отсутствуют колонки: {missing_sales_cols}")
        if missing_costs_cols:
            raise SalesAnalysisError(f"В листе 'Себестоимость' отсутствуют колонки: {missing_costs_cols}")
 
        logger.info(f"Загружено {len(sales)} записей о продажах, {len(costs)} записей о себестоимости")
 
        # 5. Очистка данных
        sales.columns = sales.columns.str.strip()
        costs.columns = costs.columns.str.strip()
 
        # Удаляем пустые строки
        sales = sales.dropna(subset=['Артикул'])
        costs = costs.dropna(subset=['Артикул'])
 
        # 6. Объединение и расчёт маржи
        df = sales.merge(
            costs[['Артикул', 'Себестоимость']],
            on='Артикул',
            how='left'
        )
 
        # Проверяем наличие товаров без себестоимости
        missing_costs = df['Себестоимость'].isna().sum()
        if missing_costs > 0:
            logger.warning(f"Найдено {missing_costs} товаров без себестоимости")
            df['Себестоимость'] = df['Себестоимость'].fillna(0)
 
        df['Маржа'] = df['Цена продажи'] - df['Себестоимость']
        # Защита от деления на 0
        df['Маржинальность %'] = (
            df['Маржа'] / df['Цена продажи'].replace(0, float('nan')) * 100
        ).round(2)
 
        # 7. Анализ
        top_margin = df.groupby(['Категория', 'Название товара']).agg({
            'Количество': 'sum',
            'Маржа': 'sum',
            'Маржинальность %': 'mean'
        }).sort_values('Маржа', ascending=False).head(20)
 
        unprofitable = df[df['Маржа'] < 0].groupby('Категория').agg({
            'Название товара': 'count',
            'Маржа': 'sum'
        })
 
        # 8. Экспорт отчёта
        timestamp = datetime.now().strftime('%Y%m%d_%H%M')
        output_file = Path(output_dir) / f"sales_analysis_{timestamp}.xlsx"
 
        with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
            top_margin.to_excel(writer, sheet_name='Топ товары')
 
            if not unprofitable.empty:
                unprofitable.to_excel(writer, sheet_name='Убыточные')
            else:
                logger.info("Убыточных товаров не найдено")
 
            # Сводная статистика
            summary = pd.DataFrame({
                'Метрика': [
                    'Всего продаж',
                    'Общая маржа',
                    'Средняя маржинальность',
                    'Убыточных товаров'
                ],
                'Значение': [
                    f"{df['Количество'].sum():,.0f} шт",
                    f"{df['Маржа'].sum():,.2f} ₽",
                    f"{df['Маржинальность %'].mean():.2f}%",
                    len(df[df['Маржа'] < 0])
                ]
            })
            summary.to_excel(writer, sheet_name='Сводка', index=False)
 
        logger.info(f"✅ Отчёт успешно создан: {output_file}")
        return str(output_file)
 
    except SalesAnalysisError:
        raise
    except Exception as e:
        logger.error(f"Неожиданная ошибка при анализе: {e}", exc_info=True)
        raise SalesAnalysisError(f"Ошибка обработки данных: {e}")
 
# Использование с обработкой ошибок
try:
    report = analyze_1c_sales('sales.xlsx')
    print(f"✅ Отчёт готов: {report}")
except SalesAnalysisError as e:
    print(f"❌ Ошибка: {e}")
    # Отправка уведомления администратору
except Exception as e:
    print(f"❌ Критическая ошибка: {e}")
    # Логирование в Sentry/другую систему мониторинга

import pandas as pd
from sqlalchemy import create_engine
 
# 1. Подключение к БД
engine = create_engine('postgresql://user:password@localhost:5432/mydb')
 
# 2. Загрузка данных
query = """
    SELECT
        u.id,
        u.email,
        u.created_at,
        u.city,
        COUNT(o.id) as orders_count,
        SUM(o.total) as total_spent
    FROM users u
    LEFT JOIN orders o ON u.id = o.user_id
    WHERE u.created_at >= '2025-01-01'
    GROUP BY u.id, u.email, u.created_at, u.city
"""
 
df = pd.read_sql(query, engine)
 
# 3. Трансформация данных
df['created_at'] = pd.to_datetime(df['created_at'])
df['registration_month'] = df['created_at'].dt.to_period('M')
 
# Категоризация пользователей (векторизованный подход)
import numpy as np
 
# ✅ Используем np.select вместо apply(): в 10-100 раз быстрее!
conditions = [
    df['total_spent'] > 10000,      # Условие для VIP
    df['orders_count'] > 5,          # Условие для Active
    df['orders_count'] > 0           # Условие для Buyer
]
choices = ['VIP', 'Active', 'Buyer']
df['category'] = np.select(conditions, choices, default='Registered')
 
# Почему это быстрее?
# apply() вызывает Python-функцию для КАЖДОЙ строки (построчно)
# np.select() выполняет операцию сразу над всем массивом в C/NumPy
# На 100k строк: apply() ~2 сек, np.select() ~20 мс
 
# 4. Добавление данных из внешнего источника (например, город -> регион)
city_to_region = {
    'Moscow': 'Central',
    'SPb': 'Northwest',
    'Kazan': 'Volga'
}
df['region'] = df['city'].map(city_to_region)
 
# 5. Экспорт в Excel с форматированием
with pd.ExcelWriter('users_report.xlsx', engine='openpyxl') as writer:
    # Основной отчёт
    df.to_excel(writer, sheet_name='Users', index=False)
 
    # Сводная таблица по категориям
    summary = df.groupby('category').agg({
        'id': 'count',
        'total_spent': 'sum'
    }).rename(columns={'id': 'users_count'})
 
    summary.to_excel(writer, sheet_name='Summary')

import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
 
def daily_etl_pipeline(date: datetime):
    """
    Ежедневный ETL: загружает данные за день, обрабатывает, сохраняет.
    """
    # 1. Загрузка данных
    raw_data_path = f'raw_data/{date.strftime("%Y-%m-%d")}.csv'
    df = pd.read_csv(raw_data_path)
 
    # 2. Валидация
    required_columns = ['user_id', 'event_type', 'timestamp']
    missing = set(required_columns) - set(df.columns)
    if missing:
        raise ValueError(f"Missing columns: {missing}")
 
    # 3. Очистка
    df = df.dropna(subset=['user_id'])  # Удаляем строки без user_id
    df = df.drop_duplicates()           # Удаляем дубликаты
 
    # 4. Преобразование типов
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['user_id'] = df['user_id'].astype(int)
 
    # 5. Фильтрация аномалий
    df = df[df['timestamp'].dt.date == date.date()]  # Только данные за нужный день
 
    # 6. Агрегация
    daily_stats = df.groupby('user_id').agg({
        'event_type': 'count',
        'timestamp': ['min', 'max']
    }).reset_index()
 
    # 7. Сохранение результата
    output_path = f'processed_data/{date.strftime("%Y-%m-%d")}_stats.parquet'
    daily_stats.to_parquet(output_path, compression='snappy')
 
    return daily_stats
 
# Запуск пайплайна
yesterday = datetime.now() - timedelta(days=1)
stats = daily_etl_pipeline(yesterday)
print(f"Processed {len(stats)} users")

import pandas as pd
import re
 
def validate_and_clean_csv(file_path: str) -> pd.DataFrame:
    """
    Валидация и очистка CSV с клиентами.
    """
 
    # 1. Загрузка
    df = pd.read_csv(file_path)
 
    # 2. Проверка обязательных колонок
    required = ['name', 'email', 'phone']
    missing = set(required) - set(df.columns)
    if missing:
        raise ValueError(f"Missing required columns: {missing}")
 
    # 3. Удаление пустых строк
    df = df.dropna(subset=required)
 
    # 4. Валидация email (векторизовано!)
    email_pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    df['email_valid'] = df['email'].str.match(email_pattern, na=False)
 
    invalid_emails = df[~df['email_valid']]
    if len(invalid_emails) > 0:
        print(f"⚠️ Found {len(invalid_emails)} invalid emails:")
        print(invalid_emails[['name', 'email']].head())
 
    df = df[df['email_valid']].drop('email_valid', axis=1)
 
    # 5. Очистка телефонов (приведение к единому формату)
    def clean_phone(phone):
        # Удаляем все кроме цифр
        digits = re.sub(r'\D', '', str(phone))
        # Добавляем +7 если начинается с 8 или 9
        if digits.startswith('8') and len(digits) == 11:
            digits = '7' + digits[1:]
        elif digits.startswith('9') and len(digits) == 10:
            digits = '7' + digits
        return f"+{digits}"
 
    df['phone'] = df['phone'].apply(clean_phone)
 
    # 6. Удаление дубликатов по email
    duplicates = df[df.duplicated(subset=['email'], keep=False)]
    if len(duplicates) > 0:
        print(f"⚠️ Found {len(duplicates)} duplicate emails")
        print(duplicates[['name', 'email']])
 
    df = df.drop_duplicates(subset=['email'], keep='first')
 
    # 7. Приведение имени к title case
    df['name'] = df['name'].str.strip().str.title()
 
    return df
 
# Использование
try:
    clean_df = validate_and_clean_csv('clients.csv')
    clean_df.to_csv('clients_clean.csv', index=False)
    print(f"✅ Successfully cleaned {len(clean_df)} records")
except Exception as e:
    print(f"❌ Validation failed: {e}")

def create_validation_report(df: pd.DataFrame) -> dict:
    """
    Создаёт детальный отчёт о качестве данных.
    """
 
    report = {
        'total_rows': len(df),
        'missing_values': df.isnull().sum().to_dict(),
        'duplicates': {
            'email': df['email'].duplicated().sum(),
            'phone': df['phone'].duplicated().sum()
        },
        'data_types': df.dtypes.astype(str).to_dict(),
        'unique_values': {
            col: df[col].nunique()
            for col in df.columns
        }
    }
 
    # Статистика для числовых колонок
    numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns
    if len(numeric_cols) > 0:
        report['numeric_stats'] = df[numeric_cols].describe().to_dict()
 
    # Топ-10 самых частых значений для категориальных колонок
    categorical_cols = df.select_dtypes(include=['object']).columns
    report['top_values'] = {
        col: df[col].value_counts().head(10).to_dict()
        for col in categorical_cols
    }
 
    return report
 
# Генерация отчёта
report = create_validation_report(df)
print(f"Total rows: {report['total_rows']}")
print(f"Missing values: {report['missing_values']}")
print(f"Email duplicates: {report['duplicates']['email']}")

import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime
 
def generate_monthly_report(year: int, month: int):
    """
    Генерирует ежемесячный отчёт для бизнеса.
    """
 
    engine = create_engine('postgresql://user:password@localhost:5432/mydb')
 
    # 1. Загрузка данных за месяц (безопасный параметризованный запрос)
    from datetime import datetime
    from dateutil.relativedelta import relativedelta
 
    # Создаём границы периода
    start_date = datetime(year, month, 1)
    end_date = start_date + relativedelta(months=1)
 
    # ✅ Параметризованный запрос (защита от SQL-инъекции)
    query = """
        SELECT
            o.id as order_id,
            o.created_at,
            o.total,
            o.status,
            u.id as user_id,
            u.email,
            u.city
        FROM orders o
        JOIN users u ON o.user_id = u.id
        WHERE o.created_at >= :start_date AND o.created_at < :end_date
    """
 
    df = pd.read_sql(query, engine, params={'start_date': start_date, 'end_date': end_date})
    df['created_at'] = pd.to_datetime(df['created_at'])
    df['date'] = df['created_at'].dt.date
 
    # 2. Создание Excel с множеством листов
    output_file = f'monthly_report_{year}_{month:02d}.xlsx'
 
    with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
        # Лист 1: Общая статистика
        summary = pd.DataFrame({
            'Метрика': [
                'Всего заказов',
                'Сумма продаж',
                'Средний чек',
                'Уникальных клиентов',
                'Завершённых заказов',
                'Отменённых заказов'
            ],
            'Значение': [
                len(df),
                f"{df['total'].sum():,.2f} ₽",
                f"{df['total'].mean():,.2f} ₽",
                df['user_id'].nunique(),
                len(df[df['status'] == 'completed']),
                len(df[df['status'] == 'cancelled'])
            ]
        })
        summary.to_excel(writer, sheet_name='Общая статистика', index=False)
 
        # Лист 2: Топ-10 клиентов
        top_clients = (
            df.groupby(['user_id', 'email'])
            .agg({
                'order_id': 'count',
                'total': 'sum'
            })
            .rename(columns={
                'order_id': 'Количество заказов',
                'total': 'Сумма покупок'
            })
            .sort_values('Сумма покупок', ascending=False)
            .head(10)
            .reset_index()
        )
        top_clients.to_excel(writer, sheet_name='Топ-10 клиентов', index=False)
 
        # Лист 3: Динамика по дням
        daily_sales = (
            df.groupby('date')
            .agg({
                'order_id': 'count',
                'total': 'sum'
            })
            .rename(columns={
                'order_id': 'Заказов',
                'total': 'Выручка'
            })
            .reset_index()
        )
        daily_sales.to_excel(writer, sheet_name='Динамика по дням', index=False)
 
        # Лист 4: Разбивка по городам
        city_stats = (
            df.groupby('city')
            .agg({
                'order_id': 'count',
                'total': ['sum', 'mean'],
                'user_id': 'nunique'
            })
            .round(2)
        )
        city_stats.columns = ['Заказов', 'Выручка', 'Средний чек', 'Клиентов']
        city_stats.to_excel(writer, sheet_name='По городам')
 
    print(f"✅ Отчёт сохранён: {output_file}")
    return output_file
 
# Генерация отчёта
report_file = generate_monthly_report(2025, 12)

from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
 
def format_excel_report(file_path: str):
    """
    Добавляет форматирование к Excel отчёту.
    """
 
    wb = load_workbook(file_path)
 
    # Форматирование для всех листов
    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]
 
        # Заголовки: жирный шрифт, серый фон
        header_fill = PatternFill(start_color="CCCCCC", end_color="CCCCCC", fill_type="solid")
        header_font = Font(bold=True)
 
        for cell in ws[1]: # Первая строка (заголовки)
            cell.fill = header_fill
            cell.font = header_font
            cell.alignment = Alignment(horizontal='center')
 
        # Автоширина колонок
        for column in ws.columns:
            max_length = 0
            column_letter = column[0].column_letter
            for cell in column:
                try:
                    if len(str(cell.value)) > max_length:
                        max_length = len(str(cell.value))
                except:
                    pass
            adjusted_width = min(max_length + 2, 50)
            ws.column_dimensions[column_letter].width = adjusted_width
 
    wb.save(file_path)
    print(f"✅ Форматирование применено: {file_path}")
 
# Применение форматирования
format_excel_report(report_file)

# Быстрая проверка: сколько пользователей зарегистрировано за неделю?
df = pd.read_sql("SELECT * FROM users WHERE created_at >= NOW() - INTERVAL '7 days'", engine)
print(df.groupby(df['created_at'].dt.date).size())

# Объединение данных из БД и CSV
users_db = pd.read_sql("SELECT * FROM users", engine)
purchases_csv = pd.read_csv('external_purchases.csv')
merged = users_db.merge(purchases_csv, on='user_id')

# Pivot таблица, агрегация по нескольким измерениям
pivot = df.pivot_table(
    values='revenue',
    index='date',
    columns=['city', 'category'],
    aggfunc=['sum', 'mean']
)

df.to_excel('report.xlsx', sheet_name='Sales')

# Запись в БД в продакшене
# ❌ df.to_sql('users', engine)  # Медленно для больших данных
# ✅ Используйте bulk_insert_mappings или COPY

# Требуется ACID, rollback, constraints
with session.begin():
    session.add(user)
    session.add(order)

# Обработка 10 млн строк: используйте SQL + индексы
# pandas загрузит всё в память и может упасть

# Используйте SQL для фильтрации, pandas для анализа
query = """
    SELECT * FROM orders
    WHERE created_at >= '2025-01-01'
    AND status = 'completed'
    LIMIT 100000
"""
df = pd.read_sql(query, engine)  # SQL: фильтрация
top_products = df.groupby('product_id')['quantity'].sum().nlargest(10)  # pandas: агрегация

import pandas as pd
 
df = pd.read_csv('data.csv')
result = (
    df.groupby('category')['sales']
    .sum()
    .sort_values(ascending=False)
    .head(10)
)

import polars as pl
 
# В 5-10 раз быстрее pandas!
df = pl.read_csv('data.csv')
result = (
    df.lazy()  # Ленивое выполнение
    .group_by('category')
    .agg(pl.col('sales').sum())
    .sort('sales', descending=True)
    .limit(10)
    .collect()  # Выполнение
)

import duckdb
 
# SQL синтаксис, работает с pandas DataFrame
df = pd.read_csv('data.csv')
result = duckdb.sql("""
    SELECT category, SUM(sales) as total_sales
    FROM df
    GROUP BY category
    ORDER BY total_sales DESC
    LIMIT 10
""").to_df()

# ✅ Parquet (snappy): золотая середина для большинства задач
df.to_parquet('data.parquet', compression='snappy', index=False)
df = pd.read_parquet('data.parquet')
 
# ✅ CSV: только для обмена с внешними системами
df.to_csv('export.csv', index=False)
 
# ❌ Pickle: НЕ используйте для долгосрочного хранения
# Причины: несовместим между версиями pandas, уязвимость безопасности

# ❌ Плохо: всё загружается как object (строки)
df = pd.read_csv('large_file.csv')
 
# ✅ Хорошо: указываем типы явно
df = pd.read_csv(
    'large_file.csv',
    dtype={
        'user_id': 'int32',  # int32 вместо int64 если значения < 2 млрд
        'category': 'category',  # Экономит память для повторяющихся строк
        'price': 'float32'
    },
    parse_dates=['created_at']
)
 
# Проверка использования памяти
print(df.memory_usage(deep=True))
 
# Пример экономии памяти
# int64 → int32: экономия 50% памяти (8 байт → 4 байта)
# object → category: экономия 90% для повторяющихся значений

# ❌ Плохо: загружает весь файл в память
df = pd.read_csv('huge_file.csv')  # 10GB файл: OOM
 
# ✅ Хорошо: обрабатываем по частям
chunks = []
for chunk in pd.read_csv('huge_file.csv', chunksize=10000):
    # Обработка chunk
    filtered = chunk[chunk['status'] == 'active']
    chunks.append(filtered)
 
df = pd.concat(chunks, ignore_index=True)

# Медленно: каждая строка: вызов Python-функции
df['price_with_tax'] = df.apply(lambda row: row['price'] * 1.2, axis=1)
 
# Что происходит:
# for row in df.iterrows(): # Python цикл
#     result = lambda(row)    # Вызов Python-функции
#     # GIL блокирует параллелизм
#     # Overhead интерпретатора на каждой итерации

# Быстро: операция над всем массивом сразу
df['price_with_tax'] = df['price'] * 1.2
 
# Что происходит:
# Операция делегируется в NumPy (написан на C)
# Использует SIMD инструкции процессора (параллельная обработка)
# Нет overhead Python-интерпретатора
# Работает в 10-100 раз быстрее

import pandas as pd
import numpy as np
import time
 
# Создаём тестовый DataFrame (100k строк)
df = pd.DataFrame({
    'price': np.random.randint(100, 10000, 100000),
    'quantity': np.random.randint(1, 100, 100000)
})
 
# Тест 1: apply() с lambda
start = time.time()
df['total_apply'] = df.apply(lambda row: row['price'] * row['quantity'], axis=1)
print(f"apply(): {time.time() - start:.3f} сек")  # ~1.2 сек
 
# Тест 2: Векторизованная операция
start = time.time()
df['total_vector'] = df['price'] * df['quantity']
print(f"vectorized: {time.time() - start:.3f} сек")  # ~0.003 сек
 
# Разница: 400x быстрее! ⚡

# ✅ Допустимо: сложная логика, которую нельзя векторизовать
def complex_transformation(row):
    # Логика с вызовом внешнего API, регулярных выражений, etc.
    if row['status'] == 'pending' and row['amount'] > threshold:
        return external_api_call(row['id'])
    return None
 
df['result'] = df.apply(complex_transformation, axis=1)
 
# Но даже здесь лучше:
# 1. Сначала отфильтровать нужные строки
# 2. Применить apply() только к ним
filtered = df[(df['status'] == 'pending') & (df['amount'] > threshold)]
filtered['result'] = filtered.apply(complex_transformation, axis=1)

# ❌ Медленно: apply для категоризации
df['category'] = df.apply(
    lambda row: 'expensive' if row['price'] > 1000 else 'cheap',
    axis=1
)
 
# ✅ Быстро: np.where (для 2 вариантов)
df['category'] = np.where(df['price'] > 1000, 'expensive', 'cheap')
 
# ✅ Быстро: np.select (для 3+ вариантов)
conditions = [
    df['price'] > 5000,
    df['price'] > 1000,
    df['price'] > 500
]
choices = ['premium', 'expensive', 'medium']
df['category'] = np.select(conditions, choices, default='cheap')
 
# ✅ Быстро: pd.cut (для диапазонов)
df['price_range'] = pd.cut(
    df['price'],
    bins=[0, 500, 1000, 5000, float('inf')],
    labels=['cheap', 'medium', 'expensive', 'premium']
)

# ❌ ОШИБКА: модификация потенциальной копии
subset = df[df['sales'] > 100]
subset['discount'] = 0.1  # ⚠️ SettingWithCopyWarning!
# Изменения могут не примениться или применить к оригиналу
 
# ✅ ПРАВИЛЬНО: явное указание намерений
# Способ 1: модифицируем оригинальный DataFrame
df.loc[df['sales'] > 100, 'discount'] = 0.1
 
# Способ 2: создаём явную копию
subset = df[df['sales'] > 100].copy()
subset['discount'] = 0.1

# ❌ ПЛОХО: 1GB памяти для повторяющихся строк
df = pd.read_csv('data.csv')
print(df['country'].memory_usage(deep=True))  # 1GB
 
# ✅ ХОРОШО: 100MB памяти с category
df['country'] = df['country'].astype('category')
print(df['country'].memory_usage(deep=True))  # 100MB
 
# Экономия: 10x для колонок с малым числом уникальных значений

# ❌ ПЛОХО: падает на больших файлах
df = pd.read_csv('huge_10gb.csv')  # OutOfMemoryError
 
# ✅ ХОРОШО: обработка по частям
result = []
for chunk in pd.read_csv('huge_10gb.csv', chunksize=50000):
    # Фильтруем и обрабатываем только нужное
    filtered = chunk[chunk['status'] == 'active']
    result.append(filtered)
 
df = pd.concat(result, ignore_index=True)

# ❌ МЕДЛЕННО: pandas угадывает типы для каждой колонки
df = pd.read_csv('data.csv')  # 10 секунд
 
# ✅ БЫСТРО: явно указываем типы
dtypes = {
    'id': 'int32',
    'price': 'float32',
    'category': 'category',
    'created_at': 'str'  # Парсим отдельно
}
df = pd.read_csv(
    'data.csv',
    dtype=dtypes,
    parse_dates=['created_at']
)  # 2 секунды
 
# Выигрыш: 5x скорость + меньше памяти

# ❌ КОШМАР: 1000 строк = 5 секунд
total = 0
for index, row in df.iterrows():
    total += row['price'] * row['quantity']
 
# ✅ ОТЛИЧНО: 1000 строк = 0.001 секунда
total = (df['price'] * df['quantity']).sum()
 
# Разница: 5000x быстрее!

# ❌ ПЛОХО: всё в одной функции
def process_sales(df):
    # 50 строк нечитаемого кода
    df['margin'] = df['price'] - df['cost']
    df['category'] = df.apply(lambda x: categorize(x), axis=1)
    df['bonus'] = df.apply(lambda x: calculate_bonus(x), axis=1)
    # ... ещё 40 строк
    return df
 
# ✅ ХОРОШО: разделение ответственности
class SalesProcessor:
    @staticmethod
    def calculate_margin(df):
        """Расчёт маржи."""
        return df['price'] - df['cost']
 
    @staticmethod
    def categorize_products(df):
        """Категоризация товаров."""
        conditions = [df['margin'] > 100, df['margin'] > 50]
        choices = ['Premium', 'Standard']
        return np.select(conditions, choices, default='Budget')
 
    @staticmethod
    def calculate_bonuses(df):
        """Расчёт бонусов."""
        return np.where(df['category'] == 'Premium', df['margin'] * 0.1, 0)
 
    def process(self, df):
        """Основной метод обработки."""
        df = df.copy()
        df['margin'] = self.calculate_margin(df)
        df['category'] = self.categorize_products(df)
        df['bonus'] = self.calculate_bonuses(df)
        return df

import pandas as pd
from typing import List, Dict, Optional
 
def load_sales_data(file_path: str) -> pd.DataFrame:
    """Загружает данные о продажах из CSV."""
 
    return pd.read_csv(file_path)
 
def filter_by_category(
    df: pd.DataFrame,
    categories: List[str]
) -> pd.DataFrame:
    """Фильтрует данные по категориям."""
 
    return df[df['category'].isin(categories)]
 
def calculate_metrics(df: pd.DataFrame) -> Dict[str, float]:
    """Рассчитывает метрики."""
 
    return {
        'total_sales': float(df['sales'].sum()),
        'avg_price': float(df['price'].mean()),
        'max_quantity': int(df['quantity'].max())
    }

def get_column(df: pd.DataFrame, column: str) -> pd.Series:
    """Возвращает колонку как Series."""
 
    return df[column]
 
def process_series(series: pd.Series) -> pd.Series:
    """Обрабатывает Series."""
 
    return series.fillna(0).astype(int)

def find_product(
    df: pd.DataFrame,
    product_id: int
) -> Optional[pd.Series]:
    """Ищет продукт по ID."""
 
    result = df[df['id'] == product_id]
    if result.empty:
        return None
    return result.iloc[0]

pip install pandas-stubs

from pandas import DataFrame, Series
from typing import Literal
 
def aggregate_data(
    df: DataFrame,
    column: str,
    method: Literal['sum', 'mean', 'count']
) -> float:
    """Агрегирует данные с проверкой типов."""
 
    if method == 'sum':
        return float(df[column].sum())
    elif method == 'mean':
        return float(df[column].mean())
    else:
        return float(df[column].count())