Files
dzentra_bot/app/src/telegram/handlers/system.py

607 lines
22 KiB
Python
Raw Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# app/src/telegram/handlers/system.py
from __future__ import annotations
from aiogram import F, Router
from aiogram.fsm.context import FSMContext
from aiogram.types import CallbackQuery, InlineKeyboardMarkup, Message
from aiogram.utils.keyboard import InlineKeyboardBuilder
from src.core.system_status import build_system_text, get_system_snapshot, has_system_alerts
from src.core.config import load_settings
from src.core.constants import APP_NAME, APP_VERSION
from src.trading.journal.service import JournalService
from src.trading.auto.service import AutoTradeService
from src.trading.auto.runner import AutoTradeRunner
router = Router(name="system")
def _system_keyboard() -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.button(text="🛠️ Настройки", callback_data="system:management")
builder.button(text=" Информация", callback_data="system:about")
builder.adjust(2)
return builder.as_markup()
def _system_alert_keyboard() -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.button(text="🔁 Обновить", callback_data="system:retry")
builder.button(text="🛠️ Настройки", callback_data="system:management")
builder.button(text=" Информация", callback_data="system:about")
builder.adjust(1, 2)
return builder.as_markup()
async def _render_system_screen(
target_message: Message,
*,
edit_mode: bool,
user_id: int | None,
chat_id: int | None,
action: str,
) -> None:
AutoTradeRunner.set_current_screen("system")
journal = JournalService()
journal.log_ui_info(
event_type="system_open_requested",
message="Запрошено открытие экрана системы.",
screen="system",
action=action,
user_id=user_id,
chat_id=chat_id,
)
snapshot = get_system_snapshot()
is_alert = has_system_alerts(snapshot)
if is_alert:
journal.log_ui_warning(
event_type="system_open_alert",
message="Система загружена с предупреждениями.",
screen="system",
action=action,
user_id=user_id,
chat_id=chat_id,
payload={
"has_alerts": True,
"components": [
{
"name": component.name,
"state": component.state,
"details": component.details,
}
for component in snapshot.components
if component.state != "🟢"
],
},
)
else:
journal.log_ui_info(
event_type="system_open_success",
message="Экран системы загружен.",
screen="system",
action=action,
user_id=user_id,
chat_id=chat_id,
payload={"has_alerts": False},
)
text = build_system_text(include_updated_at=is_alert)
reply_markup = _system_alert_keyboard() if is_alert else _system_keyboard()
if edit_mode:
await target_message.edit_text(text, reply_markup=reply_markup)
else:
await target_message.answer(text, reply_markup=reply_markup)
@router.message(F.text.in_({"🖥️ Система"}))
async def open_system(message: Message, state: FSMContext) -> None:
await state.clear()
user_id = message.from_user.id if message.from_user else None
chat_id = message.chat.id if message.chat else None
await _render_system_screen(
message,
edit_mode=False,
user_id=user_id,
chat_id=chat_id,
action="open",
)
@router.callback_query(F.data == "system:retry")
async def retry_system(callback: CallbackQuery, state: FSMContext) -> None:
await state.clear()
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
user_id = callback.from_user.id if callback.from_user else None
chat_id = callback.message.chat.id if callback.message.chat else None
await _render_system_screen(
callback.message,
edit_mode=True,
user_id=user_id,
chat_id=chat_id,
action="retry",
)
await callback.answer()
@router.callback_query(F.data == "system:management")
async def open_system_management(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>🛠️ Настройки</b>\n\n"
"<b>СИСТЕМА</b>\n\n"
"Выберите раздел:"
)
builder = InlineKeyboardBuilder()
builder.button(text="🤖 Автоторговля", callback_data="settings:auto")
builder.button(text="💹 Торговля", callback_data="settings:trade")
builder.button(text="🌍 Общие", callback_data="settings:general")
builder.button(text="📒 Журнал", callback_data="settings:journal")
builder.button(text="⬅️ Назад", callback_data="system:back")
builder.adjust(2, 2, 1)
await callback.message.edit_text(
text,
reply_markup=builder.as_markup(),
)
await callback.answer()
@router.callback_query(F.data == "settings:auto")
async def open_auto_settings(callback: CallbackQuery) -> None:
AutoTradeRunner.set_current_screen("settings_auto")
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
AutoTradeRunner.unregister_screen(
chat_id=callback.message.chat.id,
message_id=callback.message.message_id,
)
state = AutoTradeService().get_state()
strategy_map = {
"TREND": "📈 Trend Following",
"GRID": "🧩 Grid Trading",
"SCALP": "⚡ Scalping",
}
strategy_ready = state.strategy is not None
symbol_ready = bool(state.symbol)
risk_ready = state.risk_percent is not None
leverage_ready = state.leverage is not None
is_configured = strategy_ready and symbol_ready and risk_ready and leverage_ready
strategy = strategy_map.get(state.strategy or "", "")
symbol = state.symbol or ""
risk = f"{state.risk_percent:.1f}%" if state.risk_percent is not None else ""
leverage = f"x{state.leverage:g}" if state.leverage is not None else ""
strategy_icon = "" if strategy_ready else "👉"
symbol_icon = "" if symbol_ready else "👉"
risk_icon = "" if risk_ready else "👉"
leverage_icon = "" if leverage_ready else "👉"
config_status = (
"Все параметры настроены"
if is_configured
else "⛔️ Настрой все параметры"
)
text = (
"<b>🤖 Автоторговля</b>\n\n"
"<b>СИСТЕМА</b> · Настройки\n\n"
f"{strategy_icon} Стратегия: {strategy}\n"
f"{symbol_icon} Инструмент: {symbol}\n"
f"{risk_icon} Риск: {risk}\n"
f"{leverage_icon} Плечо: {leverage}\n\n"
f"{config_status}"
)
if not is_configured:
text += "\n\nВыберите настройку:"
builder = InlineKeyboardBuilder()
builder.button(text="🧠 Стратегия", callback_data="settings:auto_strategy")
builder.button(text="📈 Инструмент", callback_data="settings:auto_symbol")
builder.button(text="🛡️ Риск", callback_data="settings:auto_risk")
builder.button(text="⚙️ Плечо", callback_data="settings:auto_leverage")
builder.button(text="⬅️ Назад", callback_data="system:management")
builder.button(text="🤖 Автоторговля", callback_data="auto:home")
builder.adjust(2, 2, 2)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data == "settings:auto_strategy")
async def open_auto_strategy_settings(callback: CallbackQuery) -> None:
AutoTradeRunner.set_current_screen("settings_auto")
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>🧠 Стратегия</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Автоторговля\n\n"
"Выберите стратегию:"
)
builder = InlineKeyboardBuilder()
builder.button(text="📈 Trend", callback_data="settings:auto_strategy:trend")
builder.button(text="🧩 Grid", callback_data="settings:auto_strategy:grid")
builder.button(text="⚡ Scalp", callback_data="settings:auto_strategy:scalp")
builder.button(text="⬅️ Назад", callback_data="settings:auto")
builder.adjust(3, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data.startswith("settings:auto_strategy:"))
async def set_auto_strategy(callback: CallbackQuery) -> None:
strategy = callback.data.split(":", 2)[2]
AutoTradeService().set_strategy(strategy.upper())
if callback.message is not None:
await open_auto_settings(callback)
AutoTradeRunner.set_current_screen("settings_auto")
await callback.answer("Стратегия обновлена")
@router.callback_query(F.data == "settings:auto_symbol")
async def open_auto_symbol_settings(callback: CallbackQuery) -> None:
AutoTradeRunner.set_current_screen("settings_auto")
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
settings = load_settings()
text = (
"<b>📈 Инструмент</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Автоторговля\n\n"
"Выберите инструмент:"
)
builder = InlineKeyboardBuilder()
builder.button(text=settings.default_symbol, callback_data=f"settings:auto_symbol:{settings.default_symbol}")
builder.button(text="BTCUSDT", callback_data="settings:auto_symbol:BTCUSDT")
builder.button(text="ETHUSDT", callback_data="settings:auto_symbol:ETHUSDT")
builder.button(text="⬅️ Назад", callback_data="settings:auto")
builder.adjust(1, 2, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data.startswith("settings:auto_symbol:"))
async def set_auto_symbol(callback: CallbackQuery) -> None:
symbol = callback.data.split(":", 2)[2]
AutoTradeService().set_symbol(symbol)
if callback.message is not None:
await open_auto_settings(callback)
AutoTradeRunner.set_current_screen("settings_auto")
await callback.answer("Инструмент обновлён")
@router.callback_query(F.data == "settings:auto_risk")
async def open_auto_risk_settings(callback: CallbackQuery) -> None:
AutoTradeRunner.set_current_screen("settings_auto")
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>🛡️ Риск</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Автоторговля\n\n"
"Выберите риск на сделку:"
)
builder = InlineKeyboardBuilder()
builder.button(text="0.5%", callback_data="settings:auto_risk:0.5")
builder.button(text="1.0%", callback_data="settings:auto_risk:1.0")
builder.button(text="2.0%", callback_data="settings:auto_risk:2.0")
builder.button(text="⬅️ Назад", callback_data="settings:auto")
builder.adjust(3, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data.startswith("settings:auto_risk:"))
async def set_auto_risk(callback: CallbackQuery) -> None:
risk = float(callback.data.split(":", 2)[2])
AutoTradeService().set_risk_percent(risk)
if callback.message is not None:
await open_auto_settings(callback)
AutoTradeRunner.set_current_screen("settings_auto")
await callback.answer("Риск обновлён")
@router.callback_query(F.data == "settings:auto_leverage")
async def open_auto_leverage_settings(callback: CallbackQuery) -> None:
AutoTradeRunner.set_current_screen("settings_auto")
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>⚙️ Плечо</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Автоторговля\n\n"
"Выберите плечо:"
)
builder = InlineKeyboardBuilder()
builder.button(text="x1", callback_data="settings:auto_leverage:1")
builder.button(text="x2", callback_data="settings:auto_leverage:2")
builder.button(text="x3", callback_data="settings:auto_leverage:3")
builder.button(text="x5", callback_data="settings:auto_leverage:5")
builder.button(text="x10", callback_data="settings:auto_leverage:10")
builder.button(text="x20", callback_data="settings:auto_leverage:20")
builder.button(text="⬅️ Назад", callback_data="settings:auto")
builder.adjust(3, 3, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data.startswith("settings:auto_leverage:"))
async def set_auto_leverage(callback: CallbackQuery) -> None:
leverage = float(callback.data.split(":", 2)[2])
AutoTradeService().set_leverage(leverage)
if callback.message is not None:
await open_auto_settings(callback)
AutoTradeRunner.set_current_screen("settings_auto")
await callback.answer("Плечо обновлено")
@router.callback_query(F.data == "settings:trade")
async def open_trade_settings(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>💹 Торговля</b>\n\n"
"<b>СИСТЕМА</b> · Настройки\n\n"
"Инструмент: —\n"
"Тип ордера по умолчанию: —\n"
"Пресеты количества: —\n\n"
"В разработке."
)
builder = InlineKeyboardBuilder()
builder.button(text="⬅️ Назад", callback_data="system:management")
builder.button(text="💹 Торговля", callback_data="trade:home")
builder.adjust(2)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data == "settings:general")
async def open_general_settings(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>🌍 Общие</b>\n\n"
"<b>СИСТЕМА</b> · Настройки\n\n"
"Режим аккаунта: —\n"
"Часовой пояс: —\n"
"Язык интерфейса: ru\n\n"
"В разработке."
)
builder = InlineKeyboardBuilder()
builder.button(text="⬅️ Назад", callback_data="system:management")
builder.adjust(1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data == "settings:journal")
async def open_journal_settings(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
service = JournalService()
total = service.get_total_count()
text = (
"<b>📒 Журнал</b>\n\n"
"<b>СИСТЕМА</b> · Настройки\n\n"
f"📄 Записей: {total}\n"
"📦 Лимит: —\n"
"⏳ Хранение: —\n"
"🗄 Архив: —\n\n"
)
builder = InlineKeyboardBuilder()
builder.button(text="🗑 Очистка", callback_data="journal:clear_confirm")
builder.button(text="🗄 Архив", callback_data="settings:journal_archive")
builder.button(text="📦 Лимит", callback_data="settings:journal_limit")
builder.button(text="⏳ Хранение", callback_data="settings:journal_retention")
builder.button(text="⬅️ Назад", callback_data="system:management")
builder.button(text="📒 Журнал", callback_data="journal:1")
builder.adjust(2, 2, 2)
await callback.message.edit_text(
text,
reply_markup=builder.as_markup(),
)
await callback.answer()
@router.callback_query(F.data == "settings:journal_archive")
async def open_journal_archive_settings(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>🗄 Архив</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Журнал\n\n"
"Автоматический архив журнала: —\n"
"Формат архива: —\n"
"Периодичность: —\n\n"
"В разработке."
)
builder = InlineKeyboardBuilder()
builder.button(text="⬅️ Назад", callback_data="settings:journal")
builder.button(text="📒 Журнал", callback_data="journal:1")
builder.adjust(2)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data == "settings:journal_limit")
async def open_journal_limit_settings(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>📦 Лимит</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Журнал\n\n"
"Текущий лимит: —\n\n"
"Доступные варианты:"
)
builder = InlineKeyboardBuilder()
builder.button(text="1 000", callback_data="settings:journal_limit_stub")
builder.button(text="5 000", callback_data="settings:journal_limit_stub")
builder.button(text="10 000", callback_data="settings:journal_limit_stub")
builder.button(text="", callback_data="settings:journal_limit_stub")
builder.button(text="⬅️ Назад", callback_data="settings:journal")
builder.adjust(2, 2, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data == "settings:journal_retention")
async def open_journal_retention_settings(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
text = (
"<b>⏳ Хранение</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Журнал\n\n"
"Текущий срок хранения: —\n\n"
"Доступные варианты:"
)
builder = InlineKeyboardBuilder()
builder.button(text="7 дней", callback_data="settings:journal_retention_stub")
builder.button(text="30 дней", callback_data="settings:journal_retention_stub")
builder.button(text="90 дней", callback_data="settings:journal_retention_stub")
builder.button(text="", callback_data="settings:journal_retention_stub")
builder.button(text="⬅️ Назад", callback_data="settings:journal")
builder.adjust(2, 2, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
await callback.answer()
@router.callback_query(F.data.in_({
"settings:journal_limit_stub",
"settings:journal_retention_stub",
}))
async def journal_settings_stub(callback: CallbackQuery) -> None:
await callback.answer("Настройка скоро появится", show_alert=True)
@router.callback_query(F.data == "system:back")
async def back_to_system(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
user_id = callback.from_user.id if callback.from_user else None
chat_id = callback.message.chat.id if callback.message.chat else None
await _render_system_screen(
callback.message,
edit_mode=True,
user_id=user_id,
chat_id=chat_id,
action="back",
)
await callback.answer()
@router.callback_query(F.data == "system:about")
async def open_system_about(callback: CallbackQuery) -> None:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return
settings = load_settings()
journal = JournalService()
journal.log_ui_info(
event_type="system_about_opened",
message="Открыта информация о продукте.",
screen="system",
action="about",
user_id=callback.from_user.id if callback.from_user else None,
chat_id=callback.message.chat.id if callback.message.chat else None,
)
text = (
"<b> Информация</b>\n\n"
"<b>СИСТЕМА</b>\n\n"
f"<b>{APP_NAME}</b>\n"
f"Версия: {APP_VERSION}\n"
f"Режим: {'DEMO' if 'demo' in settings.exchange_base_url.lower() else 'LIVE'}\n"
f"Часовой пояс: {settings.tz}\n\n"
"Торговый Telegram-бот для контроля рынка, портфеля, журнала событий "
"и будущей автоторговли."
)
builder = InlineKeyboardBuilder()
builder.button(text="⬅️ Назад", callback_data="system:back")
builder.adjust(1)
await callback.message.edit_text(
text,
reply_markup=builder.as_markup(),
)
await callback.answer()