"Деньги — это не главное в жизни, но лучше вести их учет в Telegram-боте, чем не вести вообще" — наверное, так сказал бы какой-нибудь современный философ

Привет, хабровчане! Сегодня мы создадим полноценного финансового бота для Telegram, который поможет вам и вашим пользователям следить за расходами, долгами и вести финансовую аналитику. Пристегните ремни — будет интересно!

? Подготовка и структура проекта

Структура проекта

Для начала создадим структуру нашего проекта:

finance_bot/
├── requirements.txt      # Зависимости проекта
├── config.py            # Конфигурация и константы
├── database.py          # Модели и подключение к БД
├── admin.py             # Административные функции
├── utils.py            # Вспомогательные функции
├── handlers.py         # Обработчики команд бота
├── migrations/         # Миграции базы данных
└── bot.py             # Основной файл бота

Зависимости

Создаем файл requirements.txt:

python-telegram-bot==20.7
SQLAlchemy==2.0.23
pandas==2.1.3
openpyxl==3.1.2
python-dotenv==1.0.0
alembic==1.12.1
psycopg2-binary==2.9.9  # Для PostgreSQL
aiohttp==3.9.1          # Для асинхронных запросов

Настройка окружения

# Создаем виртуальное окружение
python -m venv venv
source venv/bin/activate  # для Linux/macOS
venv\Scripts\activate     # для Windows

# Устанавливаем зависимости
pip install -r requirements.txt

# Создаем .env файл
touch .env  # для Linux/macOS
echo. > .env  # для Windows

Конфигурация

Создаем файл config.py:

import os
from dotenv import load_dotenv

# Загружаем переменные окружения
load_dotenv()

# Базовые настройки
BOT_TOKEN = os.getenv('BOT_TOKEN')
if not BOT_TOKEN:
    raise ValueError("Не задан BOT_TOKEN в .env файле!")

# Настройки базы данных
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///finance_bot.db')

# Список администраторов
ADMIN_IDS = [int(id_) for id_ in os.getenv('ADMIN_IDS', '').split(',') if id_]

# Настройки логирования
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
LOG_FILE = os.getenv('LOG_FILE', 'bot.log')

# Лимиты и константы
MAX_DESCRIPTION_LENGTH = 500
MAX_EXPENSE_AMOUNT = 1000000
CURRENCY_SYMBOL = '₽'

? База данных и модели

Создание моделей

В файле database.py определяем наши модели:

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, Boolean, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, scoped_session
from datetime import datetime
from config import DATABASE_URL

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    telegram_id = Column(Integer, unique=True, nullable=False)
    username = Column(String(64))
    first_name = Column(String(64))
    last_name = Column(String(64))
    created_at = Column(DateTime, default=datetime.utcnow)
    is_active = Column(Boolean, default=True)
    
    expenses = relationship("Expense", back_populates="user", cascade="all, delete-orphan")
    categories = relationship("Category", back_populates="user", cascade="all, delete-orphan")
    debts = relationship("Debt", back_populates="user", cascade="all, delete-orphan")
    
    # Индекс для быстрого поиска по telegram_id
    __table_args__ = (Index('idx_telegram_id', telegram_id),)

class Category(Base):
    __tablename__ = 'categories'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'))
    name = Column(String(64), nullable=False)
    description = Column(String(256))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    user = relationship("User", back_populates="categories")
    expenses = relationship("Expense", back_populates="category")
    
    __table_args__ = (
        Index('idx_user_category', user_id, name),  # Для быстрого поиска категорий пользователя
    )

class Expense(Base):
    __tablename__ = 'expenses'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'))
    category_id = Column(Integer, ForeignKey('categories.id', ondelete='SET NULL'))
    amount = Column(Float, nullable=False)
    description = Column(String(500))
    date = Column(DateTime, default=datetime.utcnow)
    
    user = relationship("User", back_populates="expenses")
    category = relationship("Category", back_populates="expenses")
    
    __table_args__ = (
        Index('idx_user_expenses', user_id, date),  # Для быстрой выборки расходов по датам
    )

class Debt(Base):
    __tablename__ = 'debts'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id', ondelete='CASCADE'))
    amount = Column(Float, nullable=False)
    description = Column(String(500))
    creditor_name = Column(String(64))  # Имя кредитора
    due_date = Column(DateTime)
    is_paid = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    user = relationship("User", back_populates="debts")
    
    __table_args__ = (
        Index('idx_user_debts', user_id, is_paid),  # Для быстрого поиска неоплаченных долгов
    )

# Создаем подключение к базе данных
engine = create_engine(DATABASE_URL)

# Создаем фабрику сессий
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)

def init_db():
    """Инициализация базы данных"""
    Base.metadata.create_all(engine)

def get_session():
    """Получение сессии базы данных"""
    session = Session()
    try:
        yield session
    finally:
        session.close()

Миграции

Для управления миграциями используем Alembic. Создадим базовую структуру:

# Инициализация Alembic
alembic init migrations

# Создание первой миграции
alembic revision --autogenerate -m "Initial migration"

# Применение миграций
alembic upgrade head

?‍♂️ Административный модуль

В файле admin.py реализуем функции для управления ботом:

from database import Session, User, Expense, Category, Debt
from datetime import datetime
import pandas as pd
import io
from typing import List, Dict, Any
import logging
from config import ADMIN_IDS

logger = logging.getLogger(__name__)

def is_admin(telegram_id: int) -> bool:
    """Проверяет, является ли пользователь администратором"""
    return telegram_id in ADMIN_IDS

def get_users_stats() -> List[Dict[str, Any]]:
    """Получает статистику по всем пользователям"""
    session = Session()
    try:
        users = session.query(User).all()
        
        stats = []
        for user in users:
            # Считаем общую сумму расходов
            total_expenses = sum(expense.amount for expense in user.expenses)
            
            # Считаем общую сумму долгов
            total_debts = sum(debt.amount for debt in user.debts if not debt.is_paid)
            
            # Статистика по категориям
            category_stats = {}
            for expense in user.expenses:
                category_name = expense.category.name if expense.category else "Без категории"
                category_stats[category_name] = category_stats.get(category_name, 0) + expense.amount
            
            stats.append({
                'telegram_id': user.telegram_id,
                'username': user.username,
                'first_name': user.first_name,
                'last_name': user.last_name,
                'total_expenses': total_expenses,
                'total_debts': total_debts,
                'expenses_count': len(user.expenses),
                'debts_count': len([d for d in user.debts if not d.is_paid]),
                'categories': category_stats,
                'created_at': user.created_at,
                'is_active': user.is_active
            })
        
        return stats
    except Exception as e:
        logger.error(f"Error getting users stats: {e}")
        raise
    finally:
        session.close()

def export_users_to_excel() -> io.BytesIO:
    """Экспортирует статистику пользователей в Excel"""
    try:
        stats = get_users_stats()
        df = pd.DataFrame(stats)
        
        # Создаем буфер для записи Excel файла
        output = io.BytesIO()
        
        # Записываем DataFrame в Excel
        with pd.ExcelWriter(output, engine='openpyxl') as writer:
            df.to_excel(writer, sheet_name='Users', index=False)
            
            # Автоматическая настройка ширины столбцов
            worksheet = writer.sheets['Users']
            for idx, col in enumerate(df.columns):
                max_length = max(
                    df[col].astype(str).apply(len).max(),
                    len(str(col))
                ) + 2
                worksheet.column_dimensions[chr(65 + idx)].width = max_length
        
        output.seek(0)
        return output
    except Exception as e:
        logger.error(f"Error exporting users to Excel: {e}")
        raise

def get_user_expenses(telegram_id: int, start_date: datetime = None, end_date: datetime = None):
    """Получает расходы конкретного пользователя за период"""
    session = Session()
    try:
        user = session.query(User).filter_by(telegram_id=telegram_id).first()
        if not user:
            return None
        
        query = session.query(Expense).filter_by(user_id=user.id)
        
        if start_date:
            query = query.filter(Expense.date >= start_date)
        if end_date:
            query = query.filter(Expense.date <= end_date)
        
        return query.all()
    except Exception as e:
        logger.error(f"Error getting user expenses: {e}")
        raise
    finally:
        session.close()

? Обработка пользовательских команд

В файле handlers.py реализуем обработчики команд:

from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ContextTypes, CommandHandler, MessageHandler, CallbackQueryHandler, filters
from database import Session, User, Expense, Category, Debt
from datetime import datetime
import logging
from config import MAX_EXPENSE_AMOUNT, CURRENCY_SYMBOL
from utils import format_currency

logger = logging.getLogger(__name__)

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработчик команды /start"""
    session = Session()
    try:
        user = session.query(User).filter_by(telegram_id=update.effective_user.id).first()
        
        if not user:
            # Создаем нового пользователя
            user = User(
                telegram_id=update.effective_user.id,
                username=update.effective_user.username,
                first_name=update.effective_user.first_name,
                last_name=update.effective_user.last_name
            )
            session.add(user)
            
            # Создаем базовые категории
            default_categories = ['Продукты', 'Транспорт', 'Развлечения', 'Здоровье']
            for category_name in default_categories:
                category = Category(
                    user_id=user.id,
                    name=category_name
                )
                session.add(category)
            
            session.commit()
            welcome_message = "? Добро пожаловать в Finance Bot!\n\n" \
                            "Я помогу вам вести учет расходов и долгов. Вот что я умею:\n" \
                            "? /expense - Добавить расход\n" \
                            "? /debt - Добавить долг\n" \
                            "? /stats - Посмотреть статистику\n" \
                            "? /categories - Управление категориями"
        else:
            welcome_message = f"С возвращением, {user.first_name}!\n" \
                            "Чем могу помочь?"
        
        await update.message.reply_text(welcome_message)
        
    except Exception as e:
        logger.error(f"Error in start command: {e}")
        await update.message.reply_text("Произошла ошибка. Попробуйте позже.")
    finally:
        session.close()

async def add_expense(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработчик команды /expense"""
    session = Session()
    try:
        user = session.query(User).filter_by(telegram_id=update.effective_user.id).first()
        categories = session.query(Category).filter_by(user_id=user.id).all()
        
        # Создаем клавиатуру с категориями
        keyboard = []
        row = []
        for i, category in enumerate(categories):
            row.append(InlineKeyboardButton(
                category.name,
                callback_data=f"category_{category.id}"
            ))
            if len(row) == 2 or i == len(categories) - 1:
                keyboard.append(row)
                row = []
        
        reply_markup = InlineKeyboardMarkup(keyboard)
        
        await update.message.reply_text(
            "Выберите категорию расхода:",
            reply_markup=reply_markup
        )
        
    except Exception as e:
        logger.error(f"Error in add_expense: {e}")
        await update.message.reply_text("Произошла ошибка. Попробуйте позже.")
    finally:
        session.close()

# Добавьте остальные обработчики команд здесь...

? Аналитика и экспорт данных

В файле utils.py добавим вспомогательные функции:

from typing import List, Dict, Any
import pandas as pd
import io
from datetime import datetime, timedelta
import calendar
from config import CURRENCY_SYMBOL

def format_currency(amount: float) -> str:
    """Форматирует сумму в читаемый вид"""
    return f"{amount:,.2f} {CURRENCY_SYMBOL}".replace(',', ' ')

def generate_expense_report(expenses: List[Dict[str, Any]], title: str = "Отчет по расходам") -> io.BytesIO:
    """Генерирует Excel отчет по расходам"""
    df = pd.DataFrame(expenses)
    
    # Создаем буфер для Excel файла
    output = io.BytesIO()
    
    # Создаем Excel writer
    with pd.ExcelWriter(output, engine='openpyxl') as writer:
        # Записываем основные данные
        df.to_excel(writer, sheet_name='Расходы', index=False)
        
        # Получаем рабочий лист
        worksheet = writer.sheets['Расходы']
        
        # Форматируем заголовок
        worksheet.merge_cells('A1:E1')
        worksheet['A1'] = title
        
        # Добавляем сводную таблицу
        pivot = pd.pivot_table(
            df,
            values='amount',
            index='category',
            aggfunc='sum'
        )
        
        pivot.to_excel(writer, sheet_name='Сводка', startrow=1)
        
    output.seek(0)
    return output

def get_month_statistics(expenses: List[Dict[str, Any]], month: int = None, year: int = None) -> Dict[str, Any]:
    """Получает статистику за месяц"""
    if month is None:
        month = datetime.now().month
    if year is None:
        year = datetime.now().year
        
    # Получаем первый и последний день месяца
    last_day = calendar.monthrange(year, month)[1]
    start_date = datetime(year, month, 1)
    end_date = datetime(year, month, last_day, 23, 59, 59)
    
    # Фильтруем расходы за месяц
    month_expenses = [
        expense for expense in expenses
        if start_date <= expense['date'] <= end_date
    ]
    
    # Считаем статистику
    total_amount = sum(expense['amount'] for expense in month_expenses)
    avg_daily = total_amount / last_day if month_expenses else 0
    
    # Группируем по категориям
    categories = {}
    for expense in month_expenses:
        category = expense['category']
        categories[category] = categories.get(category, 0) + expense['amount']
    
    return {
        'total_amount': total_amount,
        'avg_daily': avg_daily,
        'expenses_count': len(month_expenses),
        'categories': categories,
        'period': f"{calendar.month_name[month]} {year}"
    }

? Деплой и поддержка

Настройка логирования

В основном файле бота (bot.py):

import logging
import logging.handlers
import os
from config import LOG_LEVEL, LOG_FILE

def setup_logging():
    """Настройка системы логирования"""
    # Создаем директорию для логов, если её нет
    log_dir = os.path.dirname(LOG_FILE)
    if log_dir and not os.path.exists(log_dir):
        os.makedirs(log_dir)
    
    # Настраиваем форматирование
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    # Настраиваем ротацию логов
    file_handler = logging.handlers.RotatingFileHandler(
        LOG_FILE,
        maxBytes=10485760,  # 10MB
        backupCount=5,
        encoding='utf-8'
    )
    file_handler.setFormatter(formatter)
    
    # Настраиваем вывод в консоль
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    
    # Настраиваем корневой логгер
    root_logger = logging.getLogger()
    root_logger.setLevel(LOG_LEVEL)
    root_logger.addHandler(file_handler)
    root_logger.addHandler(console_handler)

Основной файл бота

Создаем файл bot.py:

from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, filters
import asyncio
from config import BOT_TOKEN
from handlers import (
    start_command, add_expense, add_debt,
    show_statistics, manage_categories
)
from database import init_db
import logging

logger = logging.getLogger(__name__)

async def main():
    """Запуск бота"""
    try:
        # Инициализируем базу данных
        init_db()
        
        # Создаем приложение
        application = Application.builder().token(BOT_TOKEN).build()
        
        # Добавляем обработчики команд
        application.add_handler(CommandHandler("start", start_command))
        application.add_handler(CommandHandler("expense", add_expense))
        application.add_handler(CommandHandler("debt", add_debt))
        application.add_handler(CommandHandler("stats", show_statistics))
        application.add_handler(CommandHandler("categories", manage_categories))
        
        # Запускаем бота
        await application.run_polling()
        
    except Exception as e:
        logger.error(f"Error starting bot: {e}")
        raise

if __name__ == '__main__':
    # Настраиваем логирование
    setup_logging()
    
    # Запускаем бота
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Bot stopped by user")
    except Exception as e:
        logger.error(f"Bot crashed: {e}")

? Заключение

Что можно улучшить?

1. Добавить поддержку разных валют

2. Реализовать систему напоминаний о долгах

3. Добавить графики расходов

4. Реализовать экспорт в разные форматы

5. Добавить распознавание чеков по фото

Советы по безопасности

1. Регулярно делайте бэкапы базы данных

2. Храните чувствительные данные в переменных окружения

3. Регулярно обновляйте зависимости

4. Мониторьте ошибки и логи

Полезные ресурсы

• Документация python-telegram-bot

• SQLAlchemy Tutorial

• Pandas Documentation

• Python Logging Guide

Надеюсь, этот гайд поможет вам создать своего финансового помощника. Если у вас возникнут вопросы — пишите в комментариях.