analysis.py - master - LKastornova/Python_lesson

analysis.py
524 строки21 KB
@lvkastor
Initial commit: Finance Manager 2026 application
11 янв 2026, 19:15
11 янв 2026, 19:15209a1c3
"""
Модуль анализа финансовых данных и визуализации.

Предоставляет функции для анализа финансовых операций и создания графиков:
- Расчет баланса и статистики
- Анализ расходов по категориям
- Фильтрация операций по периодам
- Создание различных типов графиков (круговые, столбчатые)
"""

import pandas as pd
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('TkAgg')
from collections import defaultdict
import numpy as np
import logging

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

# Глобальная переменная для отслеживания открытых фигур
_open_figures = []

def close_all_figures():
    """Закрывает все открытые фигуры matplotlib."""
    global _open_figures
    for fig in _open_figures:
        try:
            plt.close(fig)
        except:
            pass
    _open_figures = []

def track_figure(fig):
    """Добавляет фигуру в список отслеживания."""
    global _open_figures
    _open_figures.append(fig)
    if len(_open_figures) > 10:
        for _ in range(len(_open_figures) - 10):
            old_fig = _open_figures.pop(0)
            try:
                plt.close(old_fig)
            except:
                pass

def calculate_balance(operations):
    """Рассчитывает текущий финансовый баланс."""
    income = sum(op.amount for op in operations if op.operation_type == 'income')
    expense = sum(op.amount for op in operations if op.operation_type == 'expense')
    return income - expense

def analyze_expenses_by_category(operations, period=None):
    """Анализирует расходы по категориям."""
    logging.debug(f"analyze_expenses_by_category called with period={period}")
    
    # Если period == 'all', не фильтруем
    if period and period != 'all':
        filtered_ops = filter_operations_by_period(operations, period)
    else:
        filtered_ops = operations
    
    expenses = defaultdict(float)
    for op in filtered_ops:
        if op.operation_type == 'expense':
            expenses[op.category] += op.amount
    
    logging.debug(f"Found {len(expenses)} expense categories")
    return dict(expenses)

def analyze_income_by_category(operations, period=None):
    """Анализирует доходы по категориям."""
    logging.debug(f"analyze_income_by_category called with period={period}")
    
    # Если period == 'all', не фильтруем
    if period and period != 'all':
        filtered_ops = filter_operations_by_period(operations, period)
    else:
        filtered_ops = operations
    
    income = defaultdict(float)
    for op in filtered_ops:
        if op.operation_type == 'income':
            income[op.category] += op.amount
    
    logging.debug(f"Found {len(income)} income categories")
    return dict(income)

def filter_operations_by_period(operations, period='month'):
    """
    Фильтрация операций по периоду.
    
    Args:
        operations: Список операций для фильтрации
        period: Период ('week', 'month', или None для всех)
    
    Returns:
        list: Отфильтрованные операции
    """
    logging.debug(f"filter_operations_by_period called with period={period}, operations count={len(operations)}")
    
    if not operations:
        logging.debug("No operations to filter")
        return operations
    
    # Если период - строка с диапазоном дат (формат: "dd.mm.yyyy-dd.mm.yyyy")
    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)
                
                logging.debug(f"Custom date range filter: {start_date} to {end_date}")
                filtered = [op for op in operations if start_date <= op.date <= end_date]
                logging.debug(f"Filtered operations count: {len(filtered)}")
                return filtered
        except ValueError as e:
            logging.debug(f"Error parsing custom date range {period}: {e}")
    
    # Если period == 'all', не фильтруем
    if period == 'all':
        logging.debug("Period is 'all', no filtering applied")
        return operations
        
    now = datetime.now()
    logging.debug(f"Current 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)
        
        logging.debug(f"Week filter: {start_date} to {end_date}")
        filtered = [op for op in operations if start_date <= op.date <= end_date]
        logging.debug(f"Filtered operations count: {len(filtered)}")
        
        # Отладочный вывод дат операций
        for i, op in enumerate(filtered[:5]):  # Показываем первые 5
            logging.debug(f"  Operation {i}: {op.date} - {op.category} - {op.amount}")
        
        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)
        
        logging.debug(f"Month filter: {start_date} to {end_date}")
        filtered = [op for op in operations if start_date <= op.date <= end_date]
        logging.debug(f"Filtered operations count: {len(filtered)}")
        return filtered
    else:
        logging.debug(f"No filtering applied for period={period}")
        return operations

def calculate_period_totals(operations, period=None):
    """Расчет итогов за период."""
    logging.debug(f"calculate_period_totals called with period={period}")
    
    # Если period == 'all', не фильтруем
    if period and period != 'all':
        filtered_ops = filter_operations_by_period(operations, period)
    else:
        filtered_ops = operations
    
    total_income = sum(op.amount for op in filtered_ops if op.operation_type == 'income')
    total_expense = sum(op.amount for op in filtered_ops if op.operation_type == 'expense')
    profit = total_income - total_expense
    
    logging.debug(f"Totals: income={total_income}, expense={total_expense}, profit={profit}")
    
    return {
        'total_income': total_income,
        'total_expense': total_expense,
        'profit': profit,
        'operation_count': len(filtered_ops)
    }

def create_expense_pie_chart(operations, period=None):
    """Создание круговой диаграммы расходов."""
    logging.debug(f"create_expense_pie_chart called with period={period}")
    
    # Если period == 'all', передаем None
    period_for_analysis = None if period == 'all' else period
    expenses = analyze_expenses_by_category(operations, period_for_analysis)
    
    logging.debug(f"Expenses data for pie chart: {expenses}")
    
    if not expenses:
        logging.debug("No expenses data for pie chart")
        return None
    
    close_all_figures()
    
    sorted_items = sorted(expenses.items(), key=lambda x: x[1], reverse=True)
    categories = [item[0] for item in sorted_items]
    amounts = [item[1] for item in sorted_items]
    total = sum(amounts)
    
    percentages = [amount / total * 100 for amount in amounts]
    
    fig, ax = plt.subplots(figsize=(14, 10))
    
    # Определяем порог для группировки мелких категорий
    threshold_percentage = 3.0  # Процентный порог для группировки
    threshold_amount = total * threshold_percentage / 100  # Абсолютный порог
    
    # Разделяем категории на основные и мелкие
    main_categories = []
    main_amounts = []
    small_categories = []
    small_amounts = []
    
    for category, amount, pct in zip(categories, amounts, percentages):
        if pct >= threshold_percentage:
            main_categories.append(category)
            main_amounts.append(amount)
        else:
            small_categories.append(category)
            small_amounts.append(amount)
    
    # Если есть мелкие категории, группируем их в "Прочие"
    if small_amounts:
        main_categories.append("Прочие")
        main_amounts.append(sum(small_amounts))
        percentages = [amount / total * 100 for amount in main_amounts]
    
    explode = [0.05 if i == 0 else 0 for i in range(len(main_categories))]
    
    # Создаем круговую диаграмму
    wedges, texts, autotexts = ax.pie(
        main_amounts, 
        labels=main_categories, 
        autopct=lambda pct: f'{pct:.1f}%',
        startangle=90,
        explode=explode,
        shadow=True,
        wedgeprops=dict(width=0.5, edgecolor='w', linewidth=2),
        pctdistance=0.85,
        textprops=dict(fontsize=11, weight='bold'),
        labeldistance=1.1
    )
    
    # Настраиваем отображение процентов
    for autotext in autotexts:
        autotext.set_color('white')
        autotext.set_fontweight('bold')
        autotext.set_fontsize(10)
    
    # Настраиваем отображение подписей
    for text in texts:
        text.set_fontsize(11)
        text.set_ha('center')
    
    ax.axis('equal')
    
    # Заголовок
    title = 'РАСХОДЫ ПО КАТЕГОРИЯМ'
    if period and period != 'all':
        title += f' ({period})'
    plt.title(title, fontsize=16, fontweight='bold', pad=20)
    
    # Добавляем легенду с деталями для "Прочих"
    if small_categories and small_amounts:
        legend_texts = []
        for category, amount in zip(small_categories, small_amounts):
            pct = amount / total * 100
            legend_texts.append(f"{category}: {amount:.2f} руб. ({pct:.1f}%)")
        
        # Создаем рамку с детализацией мелких категорий
        details = "Детализация 'Прочих':\n" + "\n".join(legend_texts)
        
        # Добавляем текстовый блок
        ax.text(1.25, 0.5, details, 
                transform=ax.transAxes, fontsize=9,
                bbox=dict(boxstyle="round,pad=0.5", 
                         facecolor="lightyellow", 
                         alpha=0.9,
                         edgecolor='orange',
                         linewidth=1),
                verticalalignment='center',
                linespacing=1.5)
    
    # Добавляем общую информацию
    info_text = f"ОБЩАЯ СУММА РАСХОДОВ: {total:.2f} руб.\n"
    info_text += f"Количество категорий: {len(main_categories)}"
    if small_categories:
        info_text += f" ({len(small_categories)} объединены в 'Прочие')"
    
    plt.figtext(0.5, 0.02, info_text, 
                ha='center', fontsize=11, weight='bold', style='italic',
                bbox=dict(boxstyle="round,pad=0.5", 
                         facecolor="lightblue", 
                         alpha=0.7,
                         edgecolor='blue',
                         linewidth=1))
    
    # Оптимизируем расположение
    plt.tight_layout(rect=[0, 0.05, 1, 0.95])
    
    track_figure(fig)
    logging.debug("Pie chart created successfully")
    return fig

def create_income_expense_bar_chart(operations, period=None):
    """Создание столбчатой диаграммы доходов и расходов по времени."""
    logging.debug(f"create_income_expense_bar_chart called with period={period}")
    
    if not operations:
        logging.debug("No operations for bar chart")
        return None
    
    # Если period == 'all', передаем None
    period_for_filter = None if period == 'all' else period
    filtered_ops = operations
    if period_for_filter:
        filtered_ops = filter_operations_by_period(operations, period_for_filter)
    
    logging.debug(f"Filtered operations count for bar chart: {len(filtered_ops)}")
    
    if not filtered_ops:
        logging.debug("No operations after filtering for bar chart")
        return None
    
    close_all_figures()
    
    df = pd.DataFrame([{
        'date': op.date.date(),
        'type': op.operation_type,
        'amount': op.amount,
        'category': op.category
    } for op in filtered_ops])
    
    df['date_str'] = df['date'].astype(str)
    grouped = df.groupby(['date_str', 'type'])['amount'].sum().unstack(fill_value=0)
    
    grouped = grouped.sort_index()
    
    fig, ax = plt.subplots(figsize=(14, 8))
    
    colors = {'income': '#4CAF50', 'expense': '#F44336'}
    
    x = np.arange(len(grouped))
    width = 0.35
    
    if 'income' in grouped.columns:
        income_bars = ax.bar(x - width/2, grouped['income'], width, 
                           label='Доходы', color=colors['income'], 
                           edgecolor='black', linewidth=1, alpha=0.8)
        for bar in income_bars:
            height = bar.get_height()
            if height > 0:
                ax.text(bar.get_x() + bar.get_width()/2., height + max(grouped.max())*0.005,
                       f'{height:.0f}', ha='center', va='bottom', 
                       fontsize=8, fontweight='bold')
    
    if 'expense' in grouped.columns:
        expense_bars = ax.bar(x + width/2, grouped['expense'], width, 
                            label='Расходы', color=colors['expense'], 
                            edgecolor='black', linewidth=1, alpha=0.8)
        for bar in expense_bars:
            height = bar.get_height()
            if height > 0:
                ax.text(bar.get_x() + bar.get_width()/2., height + max(grouped.max())*0.005,
                       f'{height:.0f}', ha='center', va='bottom', 
                       fontsize=8, fontweight='bold')
    
    # Заголовок
    title = 'ДОХОДЫ И РАСХОДЫ ПО ДАТАМ'
    if period and period != 'all':
        title += f' ({period})'
    plt.title(title, fontsize=16, fontweight='bold', pad=20)
    
    plt.xlabel('Дата', fontsize=12, fontweight='bold')
    plt.ylabel('Сумма (руб.)', fontsize=12, fontweight='bold')
    
    # Настройка оси X
    if len(grouped) > 15:
        step = max(1, len(grouped) // 10)
        xticks = x[::step]
        xlabels = grouped.index[::step]
        ax.set_xticks(xticks)
        ax.set_xticklabels(xlabels, rotation=45, ha='right', fontsize=9)
    else:
        ax.set_xticks(x)
        ax.set_xticklabels(grouped.index, rotation=45, ha='right', fontsize=9)
    
    # Сетка
    ax.yaxis.grid(True, linestyle='--', alpha=0.5, linewidth=0.5)
    ax.set_axisbelow(True)
    
    # Легенда
    ax.legend(loc='upper left', fontsize=10, framealpha=0.9)
    
    # Общая информация
    total_income = grouped.get('income', pd.Series([0])).sum()
    total_expense = grouped.get('expense', pd.Series([0])).sum()
    balance = total_income - total_expense
    
    info_text = f"Доходы: {total_income:.2f} руб. | "
    info_text += f"Расходы: {total_expense:.2f} руб. | "
    info_text += f"Баланс: {balance:.2f} руб."
    
    plt.figtext(0.5, 0.02, info_text, 
               ha='center', fontsize=11, weight='bold', style='italic',
               bbox=dict(boxstyle="round,pad=0.5", 
                        facecolor="lightgreen", 
                        alpha=0.7,
                        edgecolor='green',
                        linewidth=1))
    
    plt.tight_layout(rect=[0, 0.05, 1, 0.95])
    
    track_figure(fig)
    logging.debug("Bar chart created successfully")
    return fig

def create_top_expenses_chart(operations, top_n=10, period=None):
    """Столбчатая диаграмма самых больших расходов."""
    logging.debug(f"create_top_expenses_chart called with period={period}, top_n={top_n}")
    
    # Если period == 'all', передаем None
    period_for_filter = None if period == 'all' else period
    filtered_ops = filter_operations_by_period(operations, period_for_filter) if period_for_filter else operations
    expense_ops = [op for op in filtered_ops if op.operation_type == 'expense']
    
    logging.debug(f"Expense operations count: {len(expense_ops)}")
    
    if not expense_ops:
        logging.debug("No expense operations for top chart")
        return None
    
    close_all_figures()
    
    expense_ops.sort(key=lambda x: x.amount, reverse=True)
    top_ops = expense_ops[:min(top_n, len(expense_ops))]
    
    logging.debug(f"Top {len(top_ops)} expenses selected")
    
    fig, ax = plt.subplots(figsize=(14, 8))
    
    descriptions = []
    for op in top_ops:
        if op.subcategory:
            desc = f"{op.category}\n({op.subcategory})"
        else:
            desc = op.category
        if op.comment and len(op.comment) > 0:
            desc += f"\n'{op.comment[:20]}...'" if len(op.comment) > 20 else f"\n'{op.comment}'"
        if len(desc) > 50:
            desc = desc[:47] + "..."
        descriptions.append(desc)
    
    amounts = [op.amount for op in top_ops]
    dates = [op.date.strftime('%d.%m.%Y') for op in top_ops]
    
    colors = plt.cm.Reds(np.linspace(0.4, 0.9, len(top_ops)))
    
    bars = ax.bar(range(len(top_ops)), amounts, color=colors, 
                  edgecolor='black', linewidth=1, alpha=0.8)
    ax.set_xlabel('Категория расходов', fontsize=12, fontweight='bold')
    ax.set_ylabel('Сумма (руб.)', fontsize=12, fontweight='bold')
    
    title = f'ТОП-{len(top_ops)} САМЫХ БОЛЬШИХ РАСХОДОВ'
    if period and period != 'all':
        title += f' ({period})'
    ax.set_title(title, fontsize=16, fontweight='bold', pad=20)
    
    ax.set_xticks(range(len(top_ops)))
    ax.set_xticklabels(descriptions, rotation=45, ha='right', fontsize=10)
    
    ax.yaxis.grid(True, linestyle='--', alpha=0.5, linewidth=0.5)
    ax.set_axisbelow(True)
    
    for bar, amount, date in zip(bars, amounts, dates):
        height = bar.get_height()
        ax.text(bar.get_x() + bar.get_width()/2., height + max(amounts)*0.005,
               f'{amount:.2f}\n{date}', ha='center', va='bottom', 
               fontsize=9, fontweight='bold')
    
    total_top = sum(amounts)
    all_expenses = sum(op.amount for op in expense_ops)
    percentage = (total_top / all_expenses * 100) if all_expenses > 0 else 0
    
    info_text = f"Сумма ТОП-{len(top_ops)}: {total_top:.2f} руб. "
    info_text += f"({percentage:.1f}% от всех расходов)"
    
    plt.figtext(0.5, 0.02, info_text, 
               ha='center', fontsize=11, weight='bold', style='italic',
               bbox=dict(boxstyle="round,pad=0.5", 
                        facecolor="lightblue", 
                        alpha=0.7,
                        edgecolor='blue',
                        linewidth=1))
    
    if period and period != 'all':
        plt.figtext(0.02, 0.98, f'Период: {period}', 
                   ha='left', fontsize=10, style='italic',
                   transform=ax.transAxes,
                   bbox=dict(boxstyle="round,pad=0.3", 
                            facecolor="white", 
                            alpha=0.8,
                            edgecolor='gray',
                            linewidth=0.5))
    
    plt.tight_layout(rect=[0, 0.05, 1, 0.95])
    
    track_figure(fig)
    logging.debug("Top expenses chart created successfully")
    return fig