#позволять
Менее 10 публикаций
Отлично! Добавим Telegram-бота, который будет:
- 📣 Присылать #алерты при обнаружении дрейфа данных
- 📈#Отправлять отчёты о качестве #модели
- 🤖 #Позволять запрашивать предсказания через #чат
- 💾 #Логировать #взаимодействия
---
#
# 🤖 Telegram-бот для онлайн-обучения
> Используем `python-telegram-bot` (v20+)
---
## 1. #Установка
Добавь в `requirements.txt`:
```txt
python-telegram-bot~=20.6
requests
pandas
```
---
## 2. Создание бота в Telegram
1. Напиши [@BotFather](https://t.me/BotFather)
2. Команда: `/newbot`
3. Придумай имя: `OnlineMLBot`
4. Получи токен: `7391234567:AAHabcde...`
---
## 3. `bot/alert_bot.py`
```python
# bot/alert_bot.py
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, ContextTypes, filters
import requests
import os
import json
import pandas as pd
import numpy as np
# Настройки
TELEGRAM_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "ВАШ_ТОКЕН_ЗДЕСЬ")
ADMIN_CHAT_ID = int(os.getenv("ADMIN_CHAT_ID", "123456789")) # Ваш ID
API_URL = "http://api:8000" # Docker network
# Логирование
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Глобальные данные
user_states = {} # Для диалогов
last_drift_alert = None
# --- Команды ---
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"🤖 Привет! Я бот для мониторинга онлайн-модели.\n"
"Доступные команды:\n"
"/status — статус модели\n"
"/predict — сделать предсказание\n"
"/help — справка"
)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"📌 Команды:\n"
"/status — текущее состояние модели\n"
"/predict [числа через запятую] — предсказать (пример: /predict 0.1, -0.5, 2.0)\n"
"/alert_drift — отправить тестовый алерт о дрейфе"
)
async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
resp = requests.get(f"{API_URL}/status", timeout=5)
data = resp.json()
msg = (
"📊 *Статус модели*\n"
f"Модель загружена: {'✅' if data['model_loaded'] else '❌'}\n"
f"Размер буфера: {data['buffer_size']}\n"
f"Последнее обучение: недавно\n"
f"API: {resp.status_code}"
)
await update.message.reply_text(msg, parse_mode='Markdown')
except Exception as e:
await update.message.reply_text(f"❌ Ошибка: {e}")
async def predict(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
if not context.args:
await update.message.reply_text("Введите признаки: `/predict 0.1, -0.5, 2.0`")
return
raw = ",".join(context.args)
features = [float(x.strip()) for x in raw.split(",")]
resp = requests.post(f"{API_URL}/predict", json={"features": features, "target": 0})
pred = resp.json()["prediction"]
await update.message.reply_text(
f"🎯 Предсказание: `{pred:.4f}`\n"
f"Признаки: {features}",
parse_mode='Markdown'
)
except Exception as e:
await update.message.reply_text(f"❌ Ошибка: {e}")
# --- Алерт о дрейфе ---
async def send_drift_alert(application: Application, n_drifted: int, columns: list):
global last_drift_alert
msg = (
"⚠️ *ОБНАРУЖЕН ДРЕЙФ ДАННЫХ!*\n\n"
f"Количество дрейфующих признаков: *{n_drifted}* 📉\n"
f"Признаки: `{', '.join(columns[:5])}`\n\n"
"Рекомендуется:\n"
"1. Проверить источник данных\n"
"2. Дообучить модель\n"
"3. Запустить анализ\n\n"
"Команды:\n"
"/status — проверить модель\n"
"/predict — протестировать"
)
try:
await application.bot.send_message(
7
Нравится
Комментировать
Делитесь опытом в Пульсе
Как больше зарабатывать и эффективнее тратить
Отлично! Добавим Telegram-бота, который будет:
- 📣 #Присылать #алерты при обнаружении #дрейфа данных
- 📈 #Отправлять отчёты о #качестве модели
- 🤖 #Позволять запрашивать #предсказания через чат
- 💾 #Логировать #взаимодействия
---
# 🤖 Telegram-бот для онлайн-обучения
> Используем `python-telegram-bot` (v20+)
---
## 1. Установка
Добавь в `requirements.txt`:
```txt
python-telegram-bot~=20.6
requests
pandas
```
---
## 2. Создание бота в Telegram
1. Напиши [@BotFather](https://t.me/BotFather)
2. Команда: `/newbot`
3. Придумай имя: `OnlineMLBot`
4. Получи токен: `7391234567:AAHabcde...`
---
## 3. `bot/alert_bot.py`
```python
# bot/alert_bot.py
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, ContextTypes, filters
import requests
import os
import json
import pandas as pd
import numpy as np
# Настройки
TELEGRAM_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "ВАШ_ТОКЕН_ЗДЕСЬ")
ADMIN_CHAT_ID = int(os.getenv("ADMIN_CHAT_ID", "123456789")) # Ваш ID
API_URL = "http://api:8000" # Docker network
# Логирование
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Глобальные данные
user_states = {} # Для диалогов
last_drift_alert = None
# --- Команды ---
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"🤖 Привет! Я бот для мониторинга онлайн-модели.\n"
"Доступные команды:\n"
"/status — статус модели\n"
"/predict — сделать предсказание\n"
"/help — справка"
)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"📌 Команды:\n"
"/status — текущее состояние модели\n"
"/predict [числа через запятую] — предсказать (пример: /predict 0.1, -0.5, 2.0)\n"
"/alert_drift — отправить тестовый алерт о дрейфе"
)
async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
resp = requests.get(f"{API_URL}/status", timeout=5)
data = resp.json()
msg = (
"📊 *Статус модели*\n"
f"Модель загружена: {'✅' if data['model_loaded'] else '❌'}\n"
f"Размер буфера: {data['buffer_size']}\n"
f"Последнее обучение: недавно\n"
f"API: {resp.status_code}"
)
await update.message.reply_text(msg, parse_mode='Markdown')
except Exception as e:
await update.message.reply_text(f"❌ Ошибка: {e}")
async def predict(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
if not context.args:
await update.message.reply_text("Введите признаки: `/predict 0.1, -0.5, 2.0`")
return
raw = ",".join(context.args)
features = [float(x.strip()) for x in raw.split(",")]
resp = requests.post(f"{API_URL}/predict", json={"features": features, "target": 0})
pred = resp.json()["prediction"]
await update.message.reply_text(
f"🎯 Предсказание: `{pred:.4f}`\n"
f"Признаки: {features}",
parse_mode='Markdown'
)
except Exception as e:
await update.message.reply_text(f"❌ Ошибка: {e}")
# --- Алерт о дрейфе ---
async def send_drift_alert(application: Application, n_drifted: int, columns: list):
global last_drift_alert
msg = (
"⚠️ *ОБНАРУЖЕН ДРЕЙФ ДАННЫХ!*\n\n"
f"Количество дрейфующих признаков: *{n_drifted}* 📉\n"
f"Признаки: `{', '.join(columns[:5])}`\n\n"
"Рекомендуется:\n"
"1. Проверить источник данных\n"
"2. Дообучить модель\n"
"3. Запустить анализ\n\n"
"Команды:\n"
"/status — проверить модель\n"
"/predict — протестировать"
)
try:
await application.bot.send_message(
chat_id=ADMIN_CHAT_ID,
text=msg,
parse_mode='Markdown'
5
Нравится
Комментировать