import requests
import logging
import matplotlib.pyplot as plt
from io import BytesIO
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes, CallbackQueryHandler, MessageHandler
from telegram.ext.filters import Text, Command
from collections import defaultdict
from datetime import datetime, timedelta
import asyncio
# Настройки логирования
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
)
logger = logging.getLogger(__name__)
# Структура для хранения пользовательских данных
user_data = defaultdict(
lambda: {
"
)
# Получение доступных валют
def get_supported_currencies() -> list:
return ["bitcoin", "ethereum", "litecoin", "dogecoin", "usd", "eur", "rub"]
# Функция для получения курсов
async def fetch_exchange_rates(from_currency: str) -> dict:
url = f"https://api.coingecko.com/api/v3/simple/price?ids={from_currency}&vs_currencies=usd,eur,rub"
response = requests.get(url)
if response.status_code != 200:
logger.error(
f"Ошибка при запросе данных для валюты {from_currency}: {response.text}"
)
return None
_rates.get(currency, 0))
timestamps.append(datetime.now().strftime("%H:%M:%S"))
plt.figure(figsize=(10, 5))
for currency, values in rates.items():
plt.plot(timestamps, values, label=currency.upper())
plt.title(f"Курс {from_currency.upper()} к другим валютам")
plt.xlabel("Время")
plt.ylabel("Курс")
plt.legend()
plt.xticks(rotation=45)
plt.tight_layout()
buf = BytesIO()
plt.savefig(buf, format="png")
buf.seek(0)
return buf
# Обработка команды /start
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
await update.message.reply_text(
"Привет! Я криптообменник. Используйте команду /exchange для обмена валют."
)
# Обработка команды /exchange
async def exchange(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
keyboard = [
[
InlineKeyboardButton(currency, callback_data=currency)
for currency in get_supported_currencies()
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
"Выберите валюту для обмена:", reply_markup=reply_markup
)
# Обработка выбора валюты
async def button(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
query = update.callback_query
await query.answer()
from_currency = query.data
context.user_data["from_currency"] = from_currency # Сохраняем выбранную валюту
await query.message.reply_text(
f"Вы выбрали: {from_currency.upper()}. Введите сумму и целевую валюту (пример: 0.01 usd)."
)
# Обработка обмена валюты
async def handle_exchange(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
from_currency = context.user_data.get("from_currency")
user_id = update.effective_user.id
if not from_currency:
await update.message.reply_text(
"Сначала выберите валюту с помощью /exchange."
)message.text.split()
amount = float(amount)
except (ValueError, TypeError):
await update.message.reply_text(
"Неверный формат ввода. Введите сумму и целевую валюту (пример: 0.01 usd)."
)
return
rates = await fetch_exchange_rates(from_currency)
if rates is None:
await update.message.reply_text(
f"Не удалось получить курсы для {from_currency.upper()}"
)
return
if to_currency.lower() not in rates:
await update.message.reply_text(f"Неверная целевая валюта. Доступные валюты: {', '.join(rates.keys())}")
return
converted_amount = amount * rates[to_currency.lower()]
user_data[user_id]["history"].append(
{
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"from_currency": from_currency,
"to_currency": to_currency,
"amount": amount,
"converted_amount": converted_amount,
}
)
await update.message.reply_text(
f"{amount} {from_currency.upper()} = {converted_amount:.2f} {to_currency.upper()}"
)
# Обработка команды /history
async def history(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user_id = update.effective_user.id
history = user_data[user_id]["history"]
if not history:
await update.message.reply_text("История обменов пуста.")
return
history_text = "\n".join(
f"{item['timestamp']} - {item['amount']} {item['from_currency'].upper()} = {item['converted_amount']:.2f} {item['to_currency'].upper()}"
for item in history
)
await update.message.reply_text(f"История обменов:\n{history_text}")
# Обработка команды /chart
async def chart(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
from_currency = context.user_data.get("from_currency")
if not from_currency:
await update.message.reply_text(
"Сначала выберите валюту с помощью /exchange."
)
return
фик для {from_currency.upper()}"
)
return
await update.message.reply_photo(photo=plot)
# Обработка команды /subscribe
async def subscribe(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user_id = update.effective_user.id
from_currency = context.user_data.get("from_currency")
if not from_currency:
await update.message.reply_text(
"Сначала выберите валюту с помощью /exchange."
)
return
user_data[user_id]["subscriptions"][from_currency] = datetime.now()
await update.message.reply_text(
f"Вы подписались на уведомления об изменениях курса {from_currency.upper()}"
)
# Обработка команды /unsubscribe
async def unsubscribe(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
user_id = update.effective_user.id
from_currency = context.user_data.get("from_currency")
if not from_currency:
await update.message.reply_text(
"Сначала выберите валюту с помощью /exchange."
)
return
if from_currency in user_data[user_id]["subscriptions"]:
del user_data[user_id]["subscriptions"][from_currency]
await update.message.reply_text(
f"Вы отписались от уведомлений об изменениях курса {from_currency.upper()}"
)
else:
await update.message.reply_text(
f"Вы не подписаны на уведомления об изменениях курса {from_currency.upper()}"
)
# Проверка подписок на уведомления
async def check_subscriptions(context: ContextTypes.DEFAULT_TYPE):
for user_id, user_data_ in user_data.items():
for from_currency, subscription_time in user_data_["subscriptions"].items():
if datetime.now() - subscription_time > timedelta(minutes=5):
rates = await fetch_exchange_rates(from_currency)
if rates is None:
continue
for to_currency, rate in rates.items():
if to_currency.lower() in ["usd", "eur", "rub"]:
context.bot.send_message(
chat_id=user_id,
text=f"Изменение курса {from_currency.upper()} к {to_currency.upper()}: {rate:.2f}"
)
# Регистрируем обработчики команд
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("exchange", exchange))
application.add_handler(CommandHandler("history", history))
application.add_handler(CommandHandler("chart", chart))
application.add_handler(CommandHandler("subscribe", subscribe))
application.add_handler(CommandHandler("unsubscribe", unsubscribe))
# Регистрируем обработчик нажатия кнопок
application.add_handler(CallbackQueryHandler(button))
application.add_handler(MessageHandler(Text & ~Command, handle_exchange))
# Запускаем цикл проверки подписок каждые 5 минут
job_queue = application.job_queue
job_queue.run_repeating(check_subscriptions, interval=300, first=0)
# Запускаем бота
await applic