pandas_analysis.py - master - LKastornova/Python_lesson

pandas_analysis.py
525 строк22 KB
@lvkastor
Initial commit: Finance Manager 2026 application
11 янв 2026, 19:15
11 янв 2026, 19:15209a1c3
"""
Модуль расширенного анализа финансовых данных с использованием pandas.
"""

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
import warnings
import logging

# Настраиваем логирование
logging.basicConfig(level=logging.DEBUG, format='DEBUG: %(message)s')

# Игнорируем предупреждения SettingWithCopyWarning
warnings.filterwarnings('ignore', category=pd.errors.SettingWithCopyWarning)

def create_financial_dataframe(operations) -> pd.DataFrame:
    """
    Создает pandas DataFrame из списка финансовых операций.
    """
    data = []
    for op in operations:
        data.append({
            'date': op.date,
            'date_only': op.date.date(),
            'type': op.operation_type,
            'amount': op.amount,
            'category': op.category,
            'subcategory': op.subcategory,
            'comment': op.comment,
            'month': op.date.strftime('%Y-%m'),
            'week': op.date.isocalendar()[1],
            'year': op.date.year
        })
    
    df = pd.DataFrame(data)
    if not df.empty:
        df['date'] = pd.to_datetime(df['date'])
        df['date_only'] = pd.to_datetime(df['date_only'])
    return df

def filter_operations_by_period_for_pandas(df, period: str = None) -> pd.DataFrame:
    """
    Фильтрация DataFrame по периоду (совместимая с filter_operations_by_period из analysis.py).
    
    Args:
        df: DataFrame с операциями
        period: Период для анализа ('week', 'month', 'all' или строка с диапазоном дат "dd.mm.yyyy-dd.mm.yyyy")
    
    Returns:
        pd.DataFrame: Отфильтрованный DataFrame
    """
    if df.empty or period is None or period == 'all':
        return df
    
    # Обработка пользовательского диапазона дат
    if isinstance(period, str) and '-' in period and '.' in period:
        try:
            date_parts = period.split('-')
            if len(date_parts) == 2:
                start_str, end_str = date_parts
                start_date = datetime.strptime(start_str.strip(), "%d.%m.%Y")
                end_date = datetime.strptime(end_str.strip(), "%d.%m.%Y")
                
                # Устанавливаем время для правильного сравнения
                start_date = datetime(start_date.year, start_date.month, start_date.day, 0, 0, 0)
                end_date = datetime(end_date.year, end_date.month, end_date.day, 23, 59, 59)
                
                filtered = df[(df['date'] >= start_date) & (df['date'] <= end_date)].copy()
                logging.debug(f"Custom date range filter: {start_date} to {end_date}")
                if not filtered.empty:
                    logging.debug(f"Date range in filtered data: {filtered['date'].min()} to {filtered['date'].max()}")
                return filtered
        except ValueError as e:
            logging.debug(f"Error parsing custom date range {period}: {e}")
            return df
    
    now = datetime.now()
    
    if period == 'week':
        # Начало текущей недели (понедельник)
        start_date = now - timedelta(days=now.weekday())
        start_date = datetime(start_date.year, start_date.month, start_date.day, 0, 0, 0)
        # Конец недели (воскресенье)
        end_date = start_date + timedelta(days=6)
        end_date = datetime(end_date.year, end_date.month, end_date.day, 23, 59, 59)
        
        filtered = df[(df['date'] >= start_date) & (df['date'] <= end_date)].copy()
        logging.debug(f"Filter range: {start_date} to {end_date}")
        if not filtered.empty:
            logging.debug(f"Date range in filtered data: {filtered['date'].min()} to {filtered['date'].max()}")
        return filtered
        
    elif period == 'month':
        start_date = datetime(now.year, now.month, 1, 0, 0, 0)
        # Конец месяца
        if now.month == 12:
            next_month = datetime(now.year + 1, 1, 1, 0, 0, 0)
        else:
            next_month = datetime(now.year, now.month + 1, 1, 0, 0, 0)
        end_date = next_month - timedelta(seconds=1)
        
        filtered = df[(df['date'] >= start_date) & (df['date'] <= end_date)].copy()
        logging.debug(f"Filter range: {start_date} to {end_date}")
        if not filtered.empty:
            logging.debug(f"Date range in filtered data: {filtered['date'].min()} to {filtered['date'].max()}")
        return filtered
    
    return df

def analyze_category_statistics(operations, period: str = None) -> Dict:
    """
    Подробный анализ расходов по категориям.
    """
    logging.debug(f"analyze_category_statistics called with period={period}")
    logging.debug(f"Number of operations: {len(operations)}")
    
    df = create_financial_dataframe(operations)
    logging.debug(f"DataFrame shape before filtering: {df.shape}")
    
    if df.empty:
        logging.debug("DataFrame is empty")
        return {}
    
    # Используем совместимую фильтрацию
    if period and period != 'all' and str(period).strip():
        df = filter_operations_by_period_for_pandas(df, period)
        logging.debug(f"DataFrame shape after filtering: {df.shape}")
    
    if df.empty:
        logging.debug("DataFrame is empty after filtering")
        return {}
    
    # Расходы по категориям
    expenses_df = df[df['type'] == 'expense'].copy()
    
    logging.debug(f"Expenses DataFrame shape: {expenses_df.shape}")
    
    if expenses_df.empty:
        logging.debug("No expense operations in filtered data")
        logging.debug(f"Operations types in filtered data: {df['type'].value_counts().to_dict()}")
        return {}
    
    category_stats = {}
    
    # Основные метрики по категориям
    grouped = expenses_df.groupby('category')['amount'].agg([
        ('total', 'sum'),
        ('average', 'mean'),
        ('count', 'count'),
        ('max', 'max'),
        ('min', 'min')
    ]).round(2)
    
    # Процент от общих расходов
    total_expenses = expenses_df['amount'].sum()
    grouped['percentage'] = (grouped['total'] / total_expenses * 100).round(1)
    
    # Сортировка по убыванию суммы
    grouped = grouped.sort_values('total', ascending=False)
    
    # Преобразуем в словарь
    for category, row in grouped.iterrows():
        category_stats[category] = {
            'total': row['total'],
            'average': row['average'],
            'count': row['count'],
            'max': row['max'],
            'min': row['min'],
            'percentage': row['percentage']
        }
    
    # Общая статистика
    category_stats['_overall'] = {
        'total_expenses': total_expenses,
        'average_per_category': grouped['average'].mean().round(2) if not grouped.empty else 0,
        'category_count': len(grouped),
        'most_expensive_category': grouped.index[0] if not grouped.empty else None,
        'most_frequent_category': expenses_df['category'].mode().iloc[0] if not expenses_df['category'].mode().empty else None
    }
    
    logging.debug(f"Category analysis completed. Found {len(category_stats)} categories")
    
    return category_stats

def analyze_monthly_trends(operations, months: int = 6, period: str = None) -> Dict:
    """
    Анализ месячных трендов доходов и расходов.
    
    Args:
        operations: Список операций
        months: Количество месяцев для анализа
        period: Период для фильтрации ('week', 'month', 'all' или строка с диапазоном дат)
    """
    logging.debug(f"analyze_monthly_trends called with period={period}, months={months}")
    
    df = create_financial_dataframe(operations)
    
    if df.empty:
        logging.debug("DataFrame is empty")
        return {}
    
    # Фильтрация по периоду (если задан)
    if period and period != 'all' and str(period).strip():
        df = filter_operations_by_period_for_pandas(df, period)
        logging.debug(f"DataFrame shape after period filtering: {df.shape}")
    
    if df.empty:
        logging.debug("DataFrame is empty after filtering")
        return {}
    
    # Ограничиваем период последними N месяцами (только если не задан конкретный период)
    if not period or period == 'all':
        cutoff_date = datetime.now() - timedelta(days=months*30)
        df = df[df['date'] >= cutoff_date].copy()
        logging.debug(f"DataFrame shape after months filtering: {df.shape}")
    
    if df.empty:
        logging.debug("DataFrame is empty after months filtering")
        return {}
    
    # Группировка по месяцам
    df['month_year'] = df['date'].dt.to_period('M')
    monthly_stats = df.groupby(['month_year', 'type'])['amount'].sum().unstack(fill_value=0)
    
    # Добавляем баланс
    monthly_stats['balance'] = monthly_stats.get('income', 0) - monthly_stats.get('expense', 0)
    
    # Процентное изменение
    monthly_stats = monthly_stats.copy()
    monthly_stats.loc[:, 'income_change'] = monthly_stats.get('income', 0).pct_change() * 100
    monthly_stats.loc[:, 'expense_change'] = monthly_stats.get('expense', 0).pct_change() * 100
    monthly_stats.loc[:, 'balance_change'] = monthly_stats['balance'].pct_change() * 100
    
    # Форматируем для вывода
    trends = {}
    for idx, row in monthly_stats.iterrows():
        month_str = str(idx)
        trends[month_str] = {
            'income': float(row.get('income', 0)),
            'expense': float(row.get('expense', 0)),
            'balance': float(row['balance']),
            'income_change': float(row['income_change']) if not pd.isna(row['income_change']) else None,
            'expense_change': float(row['expense_change']) if not pd.isna(row['expense_change']) else None,
            'balance_change': float(row['balance_change']) if not pd.isna(row['balance_change']) else None
        }
    
    logging.debug(f"Monthly trends analysis completed. Found {len(trends)} months")
    
    return trends

def analyze_daily_spending_patterns(operations, period: str = None) -> Dict:
    """
    Анализ паттернов ежедневных расходов.
    
    Args:
        operations: Список операций
        period: Период для фильтрации ('week', 'month', 'all' или строка с диапазоном дат)
    """
    logging.debug(f"analyze_daily_spending_patterns called with period={period}")
    
    df = create_financial_dataframe(operations)
    
    if df.empty:
        logging.debug("DataFrame is empty")
        return {}
    
    # Фильтрация по периоду (если задан)
    if period and period != 'all' and str(period).strip():
        df = filter_operations_by_period_for_pandas(df, period)
        logging.debug(f"DataFrame shape after period filtering: {df.shape}")
    
    if df.empty:
        logging.debug("DataFrame is empty after filtering")
        return {}
    
    # Только расходы
    expenses_df = df[df['type'] == 'expense'].copy()
    
    logging.debug(f"Expenses DataFrame shape: {expenses_df.shape}")
    
    if expenses_df.empty:
        logging.debug("No expense operations in filtered data")
        return {}
    
    # Расходы по дням недели
    expenses_df.loc[:, 'weekday'] = expenses_df['date'].dt.day_name()
    expenses_df.loc[:, 'weekday_num'] = expenses_df['date'].dt.dayofweek
    
    # Статистика по дням недели
    weekday_stats = expenses_df.groupby('weekday_num').agg({
        'amount': ['sum', 'mean', 'count', 'std'],
        'weekday': 'first'
    }).round(2)
    
    # Сортировка по дням недели
    weekday_stats = weekday_stats.sort_index()
    
    patterns = {}
    days_map = {
        0: 'Понедельник',
        1: 'Вторник', 
        2: 'Среда',
        3: 'Четверг',
        4: 'Пятница',
        5: 'Суббота',
        6: 'Воскресенье'
    }
    
    for idx, row in weekday_stats.iterrows():
        day_name = days_map[idx]
        patterns[day_name] = {
            'total': float(row[('amount', 'sum')]),
            'average': float(row[('amount', 'mean')]),
            'count': int(row[('amount', 'count')]),
            'std_dev': float(row[('amount', 'std')]) if not pd.isna(row[('amount', 'std')]) else 0,
            'day_of_week': idx
        }
    
    # Находим самый дорогой и самый дешевый день
    if patterns:
        max_day = max(patterns.items(), key=lambda x: x[1]['total'])
        min_day = min(patterns.items(), key=lambda x: x[1]['total'])
        
        patterns['_summary'] = {
            'most_expensive_day': max_day[0],
            'most_expensive_amount': max_day[1]['total'],
            'cheapest_day': min_day[0],
            'cheapest_amount': min_day[1]['total'],
            'average_daily_spending': sum(p['total'] for p in patterns.values() if isinstance(p, dict) and 'total' in p) / 7 if patterns else 0
        }
    
    logging.debug(f"Daily patterns analysis completed. Found patterns for {len(patterns)} days")
    
    return patterns

def generate_financial_report(operations, period: str = None) -> str:
    """
    Генерирует текстовый финансовый отчет.
    """
    from analysis import calculate_balance, calculate_period_totals
    
    totals = calculate_period_totals(operations, period)
    category_stats = analyze_category_statistics(operations, period)
    monthly_trends = analyze_monthly_trends(operations, months=3, period=period)  # Последние 3 месяца
    daily_patterns = analyze_daily_spending_patterns(operations, period)
    
    report = []
    report.append("=" * 60)
    report.append(" " * 20 + "ФИНАНСОВЫЙ ОТЧЕТ")
    report.append("=" * 60)
    if period and period != 'all':
        report.append(f"Период анализа: {period}")
    else:
        report.append("Период анализа: Все данные")
    report.append(f"Дата формирования: {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}")
    report.append("=" * 60)
    report.append("")
    
    # Общие показатели
    report.append("ОБЩИЕ ПОКАЗАТЕЛИ:")
    report.append("-" * 40)
    report.append(f"  Доходы: {totals['total_income']:>15.2f} руб.")
    report.append(f"  Расходы: {totals['total_expense']:>14.2f} руб.")
    report.append(f"  Прибыль: {totals['profit']:>15.2f} руб.")
    report.append(f"  Количество операций: {totals['operation_count']:>7}")
    report.append("")
    
    # Расходы по категориям
    if category_stats and '_overall' in category_stats:
        report.append("РАСХОДЫ ПО КАТЕГОРИЯМ:")
        report.append("-" * 40)
        
        # Заголовок таблицы
        report.append(f"{'Категория':<20} {'Сумма':>12} {'Среднее':>12} {'Кол-во':>8} {'Доля':>6}")
        report.append("-" * 58)
        
        for category, stats in category_stats.items():
            if category != '_overall':
                report.append(f"  {category:<18} {stats['total']:>12.2f} {stats['average']:>12.2f} "
                            f"{stats['count']:>8} {stats['percentage']:>6.1f}%")
        
        report.append("")
        report.append("СВОДНАЯ СТАТИСТИКА РАСХОДОВ:")
        report.append(f"  Всего расходов: {category_stats['_overall']['total_expenses']:.2f} руб.")
        report.append(f"  Категорий: {category_stats['_overall']['category_count']}")
        report.append(f"  Самая дорогая категория: {category_stats['_overall']['most_expensive_category']}")
        report.append(f"  Самая частая категория: {category_stats['_overall']['most_frequent_category']}")
        report.append("")
    
    # Месячные тренды
    if monthly_trends:
        report.append("МЕСЯЧНЫЕ ТРЕНДЫ (последние 3 месяца):")
        report.append("-" * 40)
        
        for month, data in list(monthly_trends.items())[-3:]:  # Последние 3 месяца
            report.append(f"  {month}:")
            report.append(f"    Доходы: {data['income']:>10.2f} руб.")
            report.append(f"    Расходы: {data['expense']:>9.2f} руб.")
            report.append(f"    Баланс: {data['balance']:>10.2f} руб.")
            
            if data['income_change'] is not None:
                change_icon = "↑" if data['income_change'] > 0 else "↓"
                report.append(f"    Изменение доходов: {data['income_change']:>6.1f}% {change_icon}")
            
            if data['expense_change'] is not None:
                change_icon = "↑" if data['expense_change'] > 0 else "↓"
                report.append(f"    Изменение расходов: {data['expense_change']:>5.1f}% {change_icon}")
        report.append("")
    
    # Анализ дней недели
    if daily_patterns and '_summary' in daily_patterns:
        report.append("АНАЛИЗ РАСХОДОВ ПО ДНЯМ НЕДЕЛИ:")
        report.append("-" * 40)
        
        # Вывод по дням недели
        days_order = ['Понедельник', 'Вторник', 'Среда', 'Четверг', 
                     'Пятница', 'Суббота', 'Воскресенье']
        
        for day in days_order:
            if day in daily_patterns and day != '_summary':
                data = daily_patterns[day]
                report.append(f"  {day:<12} {data['total']:>10.2f} руб. "
                            f"(среднее: {data['average']:.2f}, операций: {data['count']})")
        
        report.append("")
        report.append("  ИТОГИ:")
        report.append(f"    Самый дорогой день: {daily_patterns['_summary']['most_expensive_day']} "
                    f"({daily_patterns['_summary']['most_expensive_amount']:.2f} руб.)")
        report.append(f"    Самый дешевый день: {daily_patterns['_summary']['cheapest_day']} "
                    f"({daily_patterns['_summary']['cheapest_amount']:.2f} руб.)")
        report.append(f"    Средние расходы в день: {daily_patterns['_summary']['average_daily_spending']:.2f} руб.")
        report.append("")
    
    report.append("=" * 60)
    report.append(" " * 15 + "ОТЧЕТ СФОРМИРОВАН АВТОМАТИЧЕСКИ")
    report.append("=" * 60)
    
    return "\n".join(report)

def analyze_subcategory_statistics(operations, period: str = None) -> Dict:
    """
    Анализ расходов по подкатегориям.
    """
    df = create_financial_dataframe(operations)
    
    if df.empty:
        return {}
    
    # Используем совместимую фильтрацию
    if period and period != 'all':
        df = filter_operations_by_period_for_pandas(df, period)
    
    # Расходы с подкатегориями
    expenses_df = df[(df['type'] == 'expense') & (df['subcategory'].notna())].copy()
    
    if expenses_df.empty:
        return {}
    
    # Группировка по категории и подкатегории
    grouped = expenses_df.groupby(['category', 'subcategory'])['amount'].agg([
        ('total', 'sum'),
        ('average', 'mean'),
        ('count', 'count'),
        ('max', 'max'),
        ('min', 'min')
    ]).round(2)
    
    # Сортировка по убыванию суммы
    grouped = grouped.sort_values('total', ascending=False)
    
    # Преобразуем в словарь
    subcategory_stats = {}
    for (category, subcategory), row in grouped.iterrows():
        key = f"{category} - {subcategory}"
        subcategory_stats[key] = {
            'category': category,
            'subcategory': subcategory,
            'total': row['total'],
            'average': row['average'],
            'count': row['count'],
            'max': row['max'],
            'min': row['min']
        }
    
    return subcategory_stats

def get_financial_summary(operations, period: str = None) -> Dict:
    """
    Возвращает сводную финансовую информацию.
    """
    from analysis import calculate_period_totals
    
    totals = calculate_period_totals(operations, period)
    category_stats = analyze_category_statistics(operations, period)
    
    summary = {
        'period': period or 'all',
        'income': totals['total_income'],
        'expense': totals['total_expense'],
        'profit': totals['profit'],
        'operation_count': totals['operation_count'],
        'categories_count': 0,
        'most_expensive_category': None,
        'largest_income': 0,
        'largest_expense': 0
    }
    
    # Находим самую дорогую категорию
    if category_stats and '_overall' in category_stats:
        summary['categories_count'] = category_stats['_overall']['category_count']
        summary['most_expensive_category'] = category_stats['_overall']['most_expensive_category']
    
    # Находим самую крупную операцию дохода и расхода
    if operations:
        income_ops = [op for op in operations if op.operation_type == 'income']
        expense_ops = [op for op in operations if op.operation_type == 'expense']
        
        if income_ops:
            summary['largest_income'] = max(op.amount for op in income_ops)
        if expense_ops:
            summary['largest_expense'] = max(op.amount for op in expense_ops)
    
    return summary