264 lines
12 KiB
Python
264 lines
12 KiB
Python
import csv
|
||
from datetime import datetime, timedelta
|
||
|
||
def load_transactions(filepath: str) -> list[dict]:
|
||
"""Загружает список транзакций из файла."""
|
||
transactions = []
|
||
try:
|
||
with open(filepath, 'r', encoding='utf-8') as file:
|
||
reader = csv.DictReader(file)
|
||
for row in reader:
|
||
try:
|
||
transaction = {
|
||
'id': int(row['id']),
|
||
'date': row['date'],
|
||
'amount': float(row['amount']),
|
||
'category': row['category'],
|
||
'description': row['description']
|
||
}
|
||
transactions.append(transaction)
|
||
except (ValueError, KeyError):
|
||
# Пропускаем строки с некорректным форматом
|
||
continue
|
||
print(f"Загружено транзакций: {len(transactions)}")
|
||
return transactions
|
||
except FileNotFoundError:
|
||
print(f"Ошибка: Файл {filepath} не найден")
|
||
return []
|
||
except Exception as e:
|
||
print(f"Ошибка при чтении файла: {e}")
|
||
return []
|
||
|
||
def filter_by_date_range(transactions: list[dict], start_date: str, end_date: str) -> list[dict]:
|
||
"""Фильтрует транзакции по диапазону дат."""
|
||
return [t for t in transactions if start_date <= t['date'] <= end_date]
|
||
|
||
def filter_by_category(transactions: list[dict], categories: list[str], exclude: bool = False) -> list[dict]:
|
||
"""
|
||
Фильтрует транзакции по категориям.
|
||
|
||
Параметры:
|
||
- transactions: список транзакций.
|
||
- categories:список категорий для фильтрации
|
||
- exclude: если True - исключает указанные категории, если False - оставляет только их
|
||
|
||
Возвращает: отфильтрованый список транзакций
|
||
"""
|
||
categories_lower = [c.lower() for c in categories]
|
||
if exclude:
|
||
return [t for t in transactions if t['category'].lower() not in categories_lower]
|
||
else:
|
||
return [t for t in transactions if t['category'].lower() in categories_lower]
|
||
|
||
def filter_by_amount_threshold(transactions: list[dict], min_amount: float = None, max_amount: float = None) -> list[dict]:
|
||
"""Фильтрует транзакции по диапазону сумм"""
|
||
result = transactions
|
||
if min_amount is not None:
|
||
result = [t for t in result if t['amount'] >= min_amount]
|
||
if max_amount is not None:
|
||
result = [t for t in result if t['amount'] <= max_amount]
|
||
return result
|
||
|
||
def calculate_balance(transactions: list[dict]) -> float:
|
||
"""Вычисляет общий баланс"""
|
||
balance = sum(t['amount'] for t in transactions)
|
||
return round(balance, 2)
|
||
|
||
def group_by_category(transactions: list[dict]) -> dict[str, float]:
|
||
"""Группирует транзакции по категориям"""
|
||
groups = {}
|
||
for t in transactions:
|
||
category = t['category']
|
||
groups[category] = groups.get(category, 0) + t['amount']
|
||
return {cat: round(amount, 2) for cat, amount in groups.items()}
|
||
|
||
def get_top_expenses(transactions: list[dict], n: int) -> list[dict]:
|
||
"""Возвращает n транзакций с наибольшими расходами."""
|
||
expenses = [t for t in transactions if t['amount'] < 0]
|
||
expenses.sort(key=lambda x: x['amount']) # Сортировка от наиболее отрицательных
|
||
return expenses[:n]
|
||
|
||
def get_average_daily_expenses(transactions: list[dict], days: int = 30) -> float:
|
||
"""Вычисляет среднюю сумму расходов за последние days дней."""
|
||
if not transactions:
|
||
return 0.0
|
||
|
||
# Находим самую позднюю дату
|
||
latest_date = max(t['date'] for t in transactions)
|
||
latest_date_obj = datetime.strptime(latest_date, "%Y-%m-%d")
|
||
start_date_obj = latest_date_obj - timedelta(days=days - 1)
|
||
start_date = start_date_obj.strftime("%Y-%m-%d")
|
||
|
||
# Фильтруем расходы за указанный период
|
||
expenses_in_period = [
|
||
t for t in transactions
|
||
if t['amount'] < 0 and start_date <= t['date'] <= latest_date
|
||
]
|
||
|
||
if not expenses_in_period:
|
||
return 0.0
|
||
|
||
total_expenses = abs(sum(t['amount'] for t in expenses_in_period))
|
||
return round(total_expenses / days, 2)
|
||
|
||
def find_transactions_by_text(transactions: list[dict], keyword: str) -> list[dict]:
|
||
"""находит транзакции по ключевому слову в описании."""
|
||
keyword_lower = keyword.lower()
|
||
return [t for t in transactions if keyword_lower in t['description'].lower()]
|
||
|
||
def generate_summary_report(transactions: list[dict]) -> str:
|
||
"""генерирует текстовый отчет по транзакциям."""
|
||
if not transactions:
|
||
return "Нет данных для формирования отчета"
|
||
|
||
total_count = len(transactions)
|
||
incomes = [t for t in transactions if t['amount'] > 0]
|
||
expenses = [t for t in transactions if t['amount'] < 0]
|
||
|
||
total_income = sum(t['amount'] for t in incomes)
|
||
total_expense = abs(sum(t['amount'] for t in expenses))
|
||
balance = total_income - total_expense
|
||
|
||
# Топ-3 категории по расходам
|
||
expenses_by_category = {}
|
||
for t in expenses:
|
||
expenses_by_category[t['category']] = expenses_by_category.get(t['category'], 0) + abs(t['amount'])
|
||
top_categories = sorted(expenses_by_category.items(), key=lambda x: x[1], reverse=True)[:3]
|
||
|
||
# Самая крупная расходная транзакция
|
||
largest_expense = min(expenses, key=lambda x: x['amount']) if expenses else None
|
||
|
||
# Формируем отчёт
|
||
report = []
|
||
report.append("=" * 50)
|
||
report.append("ФИНАНСОВЫЙ ОТЧЕТ")
|
||
report.append("=" * 50)
|
||
report.append(f"Всего транзакций: {total_count}")
|
||
report.append(f"Общий доход: {total_income:.2f} руб.")
|
||
report.append(f"Общий расход: {total_expense:.2f} руб.")
|
||
report.append(f"Итоговый баланс: {balance:.2f} руб.")
|
||
report.append("-" * 50)
|
||
report.append("ТОП-3 КАТЕГОРИИ ПО РАСХОДАМ:")
|
||
for i, (cat, amount) in enumerate(top_categories, 1):
|
||
report.append(f" {i}. {cat}: {amount:.2f} руб.")
|
||
|
||
if largest_expense:
|
||
report.append("-" * 50)
|
||
report.append("САМАЯ КРУПНАЯ РАСХОДНАЯ ТРАНЗАКЦИЯ:")
|
||
report.append(f" Дата: {largest_expense['date']}")
|
||
report.append(f" Категория: {largest_expense['category']}")
|
||
report.append(f" Сумма: {largest_expense['amount']:.2f} руб.")
|
||
report.append(f" Описание: {largest_expense['description']}")
|
||
report.append("=" * 50)
|
||
|
||
return "\n".join(report)
|
||
|
||
def main():
|
||
"""Главная функция, демонстрирующая работу всех функций."""
|
||
print("=" * 60)
|
||
print("СИСТЕМА УПРАВЛЕНИЯ ЛИЧНЫМИ ФИНАНСАМИ")
|
||
print("=" * 60)
|
||
|
||
# 1.Загрузка данных
|
||
print("\n1. ЗАГРУЗКА ДАННЫХ")
|
||
print("-" * 40)
|
||
transactions = load_transactions('data/transactions.txt')
|
||
|
||
if not transactions:
|
||
print("Нет данных для обработки. Завершение работы.")
|
||
return
|
||
|
||
# 2. Фильтрация по дате (последние 90 дней)
|
||
print("\n2. ФИЛЬТРАЦИЯ ПО ДАТЕ (последние 90 дней)")
|
||
print("-" * 40)
|
||
|
||
# Находим самую позднюю и самую раннюю дату
|
||
latest_date = max(t['date'] for t in transactions)
|
||
earliest_date = min(t['date'] for t in transactions)
|
||
latest_date_obj = datetime.strptime(latest_date, "%Y-%m-%d")
|
||
|
||
# Вычисляем дату 90 дней назад
|
||
start_date_obj = latest_date_obj - timedelta(days=89)
|
||
start_date = start_date_obj.strftime("%Y-%m-%d")
|
||
end_date = latest_date
|
||
|
||
# Если start_date раньше earliest_date, то используем earliest_date
|
||
if start_date < earliest_date:
|
||
start_date = earliest_date
|
||
print(f"Примечание: Данных за последние 90 дней недостаточно")
|
||
print(f"Используем все доступные данные с {start_date} по {end_date}")
|
||
else:
|
||
print(f"Период: с {start_date} по {end_date}")
|
||
|
||
filtered_by_date = filter_by_date_range(transactions, start_date, end_date)
|
||
print(f"Транзакций после фильтрации по дате: {len(filtered_by_date)}")
|
||
|
||
# 3. Фильтрация по категориям (исключаем "Перевод" и "Инвестиции")
|
||
print("\n3. ФИЛЬТРАЦИЯ ПО КАТЕГОРИЯМ (исключаем 'Перевод' и 'Инвестиции')")
|
||
print("-" * 40)
|
||
|
||
excluded_categories = ["Перевод", "Инвестиции"]
|
||
filtered_by_category_data = filter_by_category(filtered_by_date, excluded_categories, exclude=True)
|
||
print(f"Транзакций после фильтрации по категориям: {len(filtered_by_category_data)}")
|
||
|
||
# 4. Расчёт баланса
|
||
print("\n4. РАСЧЁТ БАЛАНСА")
|
||
print("-" * 40)
|
||
balance = calculate_balance(filtered_by_category_data)
|
||
print(f"Баланс за период: {balance:.2f} руб.")
|
||
|
||
# 5. Топ-5 расходов
|
||
print("\n5. ТОП-5 РАСХОДОВ")
|
||
print("-" * 40)
|
||
top_expenses = get_top_expenses(filtered_by_category_data, 5)
|
||
if top_expenses:
|
||
for i, expense in enumerate(top_expenses, 1):
|
||
print(f"{i}. {expense['date']} | {expense['category']} | {expense['amount']:.2f} руб. | {expense['description']}")
|
||
else:
|
||
print("Нет расходов за указанный период")
|
||
|
||
# 6. Поиск транзакций по ключевому слову "кафе"
|
||
print("\n6. ПОИСК ПО КЛЮЧЕВОМУ СЛОВУ 'кафе'")
|
||
print("-" * 40)
|
||
found_transactions = find_transactions_by_text(filtered_by_category_data, "кафе")
|
||
if found_transactions:
|
||
print(f"Найдено транзакций: {len(found_transactions)}")
|
||
for t in found_transactions:
|
||
print(f" {t['date']} | {t['category']} | {t['amount']:.2f} руб. | {t['description']}")
|
||
else:
|
||
print("Транзакции с ключевым словом 'кафе' не найдены")
|
||
|
||
# 7. Среднедневные расходы за последние 30 дней
|
||
print("\n7. СРЕДНЕДНЕВНЫЕ РАСХОДЫ (последние 30 дней)")
|
||
print("-" * 40)
|
||
avg_expenses = get_average_daily_expenses(filtered_by_category_data, 30)
|
||
print(f"Среднедневные расходы: {avg_expenses:.2f} руб.")
|
||
|
||
# 8. Группировка по категориям
|
||
print("\n8. ГРУППИРОВКА ПО КАТЕГОРИЯМ")
|
||
print("-" * 40)
|
||
grouped = group_by_category(filtered_by_category_data)
|
||
for category, amount in sorted(grouped.items(), key=lambda x: x[1], reverse=True):
|
||
print(f" {category}: {amount:.2f} руб.")
|
||
|
||
# 9. Итоговый отчет
|
||
print("\n9. ИТОГОВЫЙ ОТЧЕТ")
|
||
print("-" * 40)
|
||
report = generate_summary_report(filtered_by_category_data)
|
||
print(report)
|
||
|
||
# 10. Демонстрация фильтрации по сумме
|
||
print("\n10. ДЕМОНСТРАЦИЯ ФИЛЬТРАЦИИ ПО СУММЕ")
|
||
print("-" * 40)
|
||
large_expenses = filter_by_amount_threshold(filtered_by_category_data, max_amount=-500)
|
||
print(f"Расходы более 500 руб. (сумма < -500): {len(large_expenses)} транзакций")
|
||
for expense in large_expenses[:3]: # Показываем первые 3
|
||
print(f" {expense['date']} | {expense['category']} | {expense['amount']:.2f} руб.")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("ОБРАБОТКА ЗАВЕРШЕНА")
|
||
print("=" * 60)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|