Полное руководство: Работа с большими наборами данных в Pandas

import sqlite3
import pandas as pd

# Создание базы данных SQLite из CSV
conn = sqlite3.connect('credit_risk.db')
df = pd.read_csv('large_dataset.csv')
df.to_sql('credit_data', conn, if_exists='replace', index=False)

# Чтение определенных столбцов из файла Parquet
columns_needed = ['customer_id', 'income', 'credit_score', 'debt_ratio']
df_subset = pd.read_parquet('credit_data.parquet', columns=columns_needed)

# Правильное использование HDFStore для вашего случая
store = pd.HDFStore('credit_data.h5', mode='a')
store.put('credit_data', df, format='table', data_columns=True)

import sqlite3
import pandas as pd
from sqlalchemy import create_engine

# Метод 1: Прямое преобразование CSV в SQLite (эффективно по памяти)
def csv_to_sqlite(csv_path, db_path, table_name, chunksize=10000):
    conn = sqlite3.connect(db_path)
    
    # Чтение CSV по частям для эффективного использования памяти
    for chunk in pd.read_csv(csv_path, chunksize=chunksize):
        chunk.to_sql(table_name, conn, if_exists='append', index=False)
    
    # Создание индексов по часто запрашиваемым столбцам
    conn.execute(f'CREATE INDEX IF NOT EXISTS idx_customer_id ON {table_name}(customer_id)')
    conn.close()

# Использование
csv_to_sqlite('consumer_credit_1gb.csv', 'credit_risk.db', 'credit_data', chunksize=50000)

# Определение схемы для согласованности типов
schema_sql = """
CREATE TABLE credit_data (
    customer_id INTEGER PRIMARY KEY,
    income REAL,
    credit_score INTEGER,
    debt_ratio REAL,
    employment_status TEXT,
    loan_amount REAL,
    -- Добавьте все 1000-2000 столбцов с соответствующими типами
    ...
);
"""

# Выполнение создания схемы
conn = sqlite3.connect('credit_risk.db')
conn.executescript(schema_sql)
conn.close()

import sqlite3

def get_column_subset(conn, table_name, columns, conditions=None):
    """Извлечение определенных столбцов для анализа"""
    query = f"SELECT {', '.join(columns)} FROM {table_name}"
    if conditions:
        query += f" WHERE {conditions}"
    
    return pd.read_sql(query, conn)

# Пример: Получение финансовых столбцов для расчета риска
financial_cols = ['customer_id', 'income', 'credit_score', 'debt_ratio', 'loan_amount']
df_financial = get_column_subset(conn, 'credit_data', financial_cols)

# Обработка столбцов партиями для минимизации использования памяти
def process_in_batches(conn, table_name, all_columns, batch_size=20):
    results = []
    
    for i in range(0, len(all_columns), batch_size):
        batch = all_columns[i:i + batch_size]
        df_batch = get_column_subset(conn, table_name, batch)
        
        # Ваша логика анализа здесь
        processed_data = your_analysis_function(df_batch)
        results.append(processed_data)
    
    return pd.concat(results, axis=1)

# Использование
all_columns = ['col1', 'col2', ..., 'col2000']  # Ваши 1000-2000 столбцов
final_result = process_in_batches(conn, 'credit_data', all_columns)

# Используйте параметризованные запросы для предотвращения SQL-инъекций
def safe_query(conn, table_name, columns, conditions=None, params=None):
    query = f"SELECT {', '.join(columns)} FROM {table_name}"
    if conditions:
        query += f" WHERE {conditions}"
    
    return pd.read_sql(query, conn, params=params)

# Пример с условиями
high_risk_customers = safe_query(
    conn, 'credit_data', 
    ['customer_id', 'credit_score'], 
    "credit_score < 600 AND debt_ratio > 0.4"
)

def add_new_columns_to_sqlite(conn, table_name, df_new_columns):
    """Добавление новых вычисляемых столбцов в существующую таблицу"""
    # Создание временной таблицы с новыми столбцами
    temp_table = f"{table_name}_temp_{pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')}"
    df_new_columns.to_sql(temp_table, conn, if_exists='replace', index=False)
    
    # Объединение с исходной таблицей и обновление
    update_query = f"""
    UPDATE {table_name} t
    SET {', '.join([f"{col}=s.{col}" for col in df_new_columns.columns])}
    FROM {temp_table} s
    WHERE t.customer_id = s.customer_id
    """
    
    conn.execute(update_query)
    conn.execute(f"DROP TABLE {temp_table}")
    conn.commit()

def create_risk_features(df):
    """Создание новых признаков на основе бизнес-логики"""
    df['risk_score'] = (
        (df['credit_score'] / 850) * 0.3 +
        (1 - df['debt_ratio']) * 0.4 +
        (df['income'] / df['loan_amount']) * 0.3
    )
    
    df['risk_category'] = pd.cut(
        df['risk_score'],
        bins=[0, 0.3, 0.6, 1.0],
        labels=['Высокий риск', 'Средний риск', 'Низкий риск']
    )
    
    df['income_stability'] = (
        np.where(df['employment_status'] == 'Employed', 1, 0) +
        np.where(df['years_at_job'] > 2, 1, 0)
    )
    
    return df

# Использование
df_subset = get_column_subset(conn, 'credit_data', 
    ['customer_id', 'credit_score', 'debt_ratio', 'income', 'loan_amount', 
     'employment_status', 'years_at_job'])

df_with_features = create_risk_features(df_subset)
add_new_columns_to_sqlite(conn, 'credit_data', df_with_features[['customer_id', 'risk_score', 'risk_category', 'income_stability']])

def batch_update_features(conn, table_name, feature_updates, batch_size=1000):
    """Обновление базы данных партиями для больших наборов признаков"""
    customer_ids = pd.read_sql(f"SELECT customer_id FROM {table_name}", conn)['customer_id']
    
    for i in range(0, len(customer_ids), batch_size):
        batch_ids = customer_ids[i:i + batch_size]
        batch_condition = f"customer_id IN ({','.join(map(str, batch_ids))})"
        
        for column, value in feature_updates.items():
            update_query = f"""
            UPDATE {table_name}
            SET {column} = {value}
            WHERE {batch_condition}
            """
            conn.execute(update_query)
        
        conn.commit()
        print(f"Обработана партия {i//batch_size + 1}")

# Использование
feature_updates = {
    'risk_score': '(credit_score / 850) * 0.3 + (1 - debt_ratio) * 0.4',
    'risk_level': "CASE WHEN credit_score < 600 THEN 'High' WHEN credit_score < 700 THEN 'Medium' ELSE 'Low' END"
}
batch_update_features(conn, 'credit_data', feature_updates)

class LargeDatasetProcessor:
    def __init__(self, db_path, table_name):
        self.db_path = db_path
        self.table_name = table_name
        self.conn = sqlite3.connect(db_path)
        
    def process_column_groups(self, column_groups, feature_functions):
        """Обработка нескольких групп столбцов с их функциями признаков"""
        for group_name, columns in column_groups.items():
            print(f"Обработка столбцов {group_name}...")
            
            # Получение подмножества данных
            df = pd.read_sql(
                f"SELECT {', '.join(columns)} FROM {self.table_name}", 
                self.conn
            )
            
            # Применение инженерии признаков
            if group_name in feature_functions:
                df = feature_functions[group_name](df)
                
                # Сохранение новых признаков обратно в базу данных
                new_cols = [col for col in df.columns if col not in columns]
                if new_cols:
                    self.add_features(df[['customer_id'] + new_cols])
    
    def add_features(self, df_features):
        """Добавление новых признаков в базу данных"""
        df_features.to_sql(
            f"{self.table_name}_new_features", 
            self.conn, 
            if_exists='replace', 
            index=False
        )
        
        # Объединение новых признаков в основную таблицу
        merge_query = f"""
        UPDATE {self.table_name} t
        SET {', '.join([f"{col}=n.{col}" for col in df_features.columns if col != 'customer_id'])}
        FROM {self.table_name}_new_features n
        WHERE t.customer_id = n.customer_id
        """
        self.conn.execute(merge_query)
        self.conn.execute(f"DROP TABLE {self.table_name}_new_features")
        self.conn.commit()

# Использование
processor = LargeDatasetProcessor('credit_risk.db', 'credit_data')

column_groups = {
    'financial': ['customer_id', 'income', 'credit_score', 'debt_ratio'],
    'employment': ['customer_id', 'employment_status', 'years_at_job', 'industry'],
    'loan_details': ['customer_id', 'loan_amount', 'loan_term', 'interest_rate']
}

feature_functions = {
    'financial': create_risk_features,
    'employment': lambda df: df.assign(
        employment_stability=(df['years_at_job'] > 2).astype(int)
    )
}

processor.process_column_groups(column_groups, feature_functions)

import time
import psutil

def monitor_performance(func):
    """Декоратор для мониторинга производительности функции"""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        start_mem = psutil.Process().memory_info().rss / 1024 / 1024  # MB
        
        result = func(*args, **kwargs)
        
        end_time = time.time()
        end_mem = psutil.Process().memory_info().rss / 1024 / 1024  # MB
        
        print(f"Функция {func.__name__}:")
        print(f"  Время: {end_time - start_time:.2f} секунд")
        print(f"  Память: {end_mem - start_mem:.2f} MB прирост")
        print(f"  Пиковая память: {end_mem:.2f} MB")
        
        return result
    return wrapper

# Использование с вашими функциями
@monitor_performance
def process_large_dataset():
    # Ваша логика обработки здесь
    pass

def safe_database_operation(operation_func, *args, max_retries=3, **kwargs):
    """Безопасное выполнение операций с базой данных с логикой повторных попыток"""
    for attempt in range(max_retries):
        try:
            result = operation_func(*args, **kwargs)
            return result
        except sqlite3.Error as e:
            print(f"Попытка {attempt + 1} не удалась: {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Экспоненциальный бэкофф
                # Переподключение к базе данных
                conn = sqlite3.connect(kwargs.get('db_path', 'credit_risk.db'))
                kwargs['conn'] = conn
            else:
                raise
        except MemoryError:
            print("Произошла ошибка памяти. Попробуйте уменьшить размер фрагмента.")
            raise
        except Exception as e:
            print(f"Неожиданная ошибка: {e}")
            raise

# Использование
try:
    safe_database_operation(
        csv_to_sqlite,
        'large_file.csv', 'credit_risk.db', 'credit_data',
        chunksize=30000, max_retries=5
    )
except Exception as e:
    print(f"Не удалось обработать данные: {e}")
    # Реализуйте запасной вариант или оповещение

import sqlite3
import pandas as pd
import numpy as np
from datetime import datetime

class CreditRiskModelingPipeline:
    def __init__(self, db_path='credit_risk.db'):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path)
        
    def initialize_database(self, csv_file):
        """Настройка базы данных из исходного CSV-файла"""
        print("Инициализация базы данных...")
        
        # Создание таблицы с оптимизированной схемой
        self.conn.execute("""
        CREATE TABLE IF NOT EXISTS credit_data (
            customer_id INTEGER PRIMARY KEY,
            credit_score INTEGER,
            income REAL,
            debt_ratio REAL,
            employment_status TEXT,
            loan_amount REAL,
            loan_term INTEGER,
            interest_rate REAL,
            -- Добавьте все ваши 1000-2000 полей
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """)
        
        # Загрузка данных по частям
        chunk_size = 50000
        for i, chunk in enumerate(pd.read_csv(csv_file, chunksize=chunk_size)):
            chunk.to_sql('credit_data', self.conn, if_exists='append', index=False)
            print(f"Загружен фрагмент {i + 1}")
            
        # Создание индексов
        self.conn.execute("CREATE INDEX IF NOT EXISTS idx_credit_score ON credit_data(credit_score)")
        self.conn.execute("CREATE INDEX IF NOT EXISTS idx_income ON credit_data(income)")
        self.conn.commit()
        
    def calculate_risk_features(self):
        """Расчет комплексных признаков риска"""
        print("Расчет признаков риска...")
        
        # Признаки финансового риска
        df_financial = pd.read_sql("""
            SELECT customer_id, credit_score, income, debt_ratio, 
                   loan_amount, loan_term, interest_rate
            FROM credit_data
        """, self.conn)
        
        df_financial['risk_score'] = (
            (df_financial['credit_score'] / 850) * 0.35 +
            (1 - df_financial['debt_ratio']) * 0.25 +
            (df_financial['income'] / 100000) * 0.2 +
            (1 - df_financial['interest_rate'] / 0.2) * 0.2
        )
        
        df_financial['monthly_payment_ratio'] = (
            df_financial['loan_amount'] * 
            (df_financial['interest_rate'] / 12) / 
            (1 - (1 + df_financial['interest_rate'] / 12) ** (-df_financial['loan_term']))
        ) / df_financial['income']
        
        # Признаки стабильности занятости
        df_employment = pd.read_sql("""
            SELECT customer_id, employment_status, years_at_job, industry
            FROM credit_data
        """, self.conn)
        
        df_employment['employment_stability'] = (
            (df_employment['years_at_job'] > 2).astype(int) * 0.6 +
            (df_employment['employment_status'] == 'Employed').astype(int) * 0.4
        )
        
        # Объединение всех признаков
        df_features = pd.merge(df_financial, df_employment, on='customer_id')
        
        # Сохранение признаков в базе данных
        feature_cols = [col for col in df_features.columns if col != 'customer_id']
        for col in feature_cols:
            self.conn.execute(f"""
                ALTER TABLE credit_data ADD COLUMN {col} REAL
            """)
        
        # Обновление новыми признаками
        for _, row in df_features.iterrows():
            update_values = ', '.join([f"{col}={row[col]}" for col in feature_cols])
            self.conn.execute(f"""
                UPDATE credit_data 
                SET {update_values} 
                WHERE customer_id = {row['customer_id']}
            """)
        
        self.conn.commit()
        print("Признаки риска рассчитаны и успешно сохранены")
        
    def analyze_risk_segments(self):
        """Анализ различных сегментов риска"""
        print("Анализ сегментов риска...")
        
        # Получение данных с новыми признаками
        df_analysis = pd.read_sql("""
            SELECT customer_id, credit_score, risk_score, monthly_payment_ratio,
                   employment_stability, risk_category
            FROM credit_data
        """, self.conn)
        
        # Анализ сегментов
        risk_segments = {
            'Низкий риск': df_analysis[df_analysis['risk_score'] > 0.7],
            'Средний риск': df_analysis[
                (df_analysis['risk_score'] >= 0.4) & 
                (df_analysis['risk_score'] <= 0.7)
            ],
            'Высокий риск': df_analysis[df_analysis['risk_score'] < 0.4]
        }
        
        # Генерация инсайтов
        for segment_name, segment_data in risk_segments.items():
            avg_risk = segment_data['risk_score'].mean()
            avg_credit_score = segment_data['credit_score'].mean()
            count = len(segment_data)
            
            print(f"\nСегмент {segment_name}:")
            print(f"  Количество: {count} клиентов")
            print(f"  Средний балл риска: {avg_risk:.3f}")
            print(f"  Средний кредитный балл: {avg_credit_score:.0f}")
            
            # Сохранение анализа сегмента
            segment_data.to_sql(
                f'{segment_name.lower().replace(" ", "_")}_customers',
                self.conn, if_exists='replace', index=False
            )
        
        self.conn.commit()
        
    def close(self):
        """Очистка подключения к базе данных"""
        self.conn.close()

# Полное использование рабочего процесса
if __name__ == "__main__":
    # Инициализация конвейера
    pipeline = CreditRiskModelingPipeline()
    
    # Обработка данных (выполнить один раз для настройки базы данных)
    pipeline.initialize_database('consumer_credit_data.csv')
    
    # Расчет признаков риска
    pipeline.calculate_risk_features()
    
    # Анализ сегментов риска
    pipeline.analyze_risk_segments()
    
    # Очистка
    pipeline.close()
    
    print("Конвейер моделирования кредитного риска успешно завершен!")