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

704 lines
26 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.config import load_settings
from src.core.constants import APP_NAME, APP_VERSION
from src.core.system_status import build_system_text, get_system_snapshot, has_system_alerts
from src.telegram.live.active_screen import ActiveScreenManager
from src.telegram.live.runner import LiveScreenRunner, ScreenRegistry, StaticScreen
from src.trading.auto.service import AutoTradeService
from src.trading.journal.service import JournalService
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()
def _register_system_screen(message: Message, screen: str = "system") -> None:
LiveScreenRunner.unregister_message(
chat_id=message.chat.id,
message_id=message.message_id,
)
ScreenRegistry.unregister_message(
chat_id=message.chat.id,
message_id=message.message_id,
)
ScreenRegistry.register_screen(
StaticScreen(
screen=screen,
bot=message.bot,
chat_id=message.chat.id,
message_id=message.message_id,
)
)
ActiveScreenManager.register(
screen=screen,
message=message,
)
async def _prepare_system_from_message(message: Message, screen: str = "system") -> None:
await ActiveScreenManager.prepare_new_screen(
screen=screen,
bot=message.bot,
chat_id=message.chat.id,
)
async def _prepare_system_from_callback(
callback: CallbackQuery,
screen: str = "system",
) -> bool:
if callback.message is None:
await callback.answer("Сообщение не найдено", show_alert=True)
return False
await ActiveScreenManager.prepare_new_screen(
screen=screen,
bot=callback.message.bot,
chat_id=callback.message.chat.id,
keep_message_id=callback.message.message_id,
)
return True
async def _render_system_screen(
target_message: Message,
*,
edit_mode: bool,
user_id: int | None,
chat_id: int | None,
action: str,
) -> None:
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)
_register_system_screen(target_message, screen="system")
return
sent_message = await target_message.answer(text, reply_markup=reply_markup)
_register_system_screen(sent_message, screen="system")
@router.message(F.text.in_({"🖥️ Система"}))
async def open_system(message: Message, state: FSMContext) -> None:
await state.clear()
await _prepare_system_from_message(message, screen="system")
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 not await _prepare_system_from_callback(callback, screen="system"):
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 not await _prepare_system_from_callback(callback, screen="system"):
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())
_register_system_screen(callback.message, screen="system")
await callback.answer()
@router.callback_query(F.data == "settings:auto")
async def open_auto_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_auto"):
return
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_trend_strategy = (state.strategy or "").upper() == "TREND"
sl_ready = state.stop_loss_percent is not None and state.stop_loss_percent > 0
is_configured = (
strategy_ready
and symbol_ready
and risk_ready
and leverage_ready
and (not is_trend_strategy or sl_ready)
)
strategy = strategy_map.get(state.strategy or "", "")
symbol = ""
if state.symbol:
base = state.symbol.split("_", 1)[0].upper()
if "/" in base:
symbol = base.split("/", 1)[0]
else:
for suffix in ("USDT", "USD", "EUR", "BTC"):
if base.endswith(suffix) and len(base) > len(suffix):
base = base[: -len(suffix)]
break
symbol = base
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 ""
max_reserved = (
f"{state.max_reserved_balance_percent:g}%"
if state.max_reserved_balance_percent is not None
else "off"
)
sl = f"{state.stop_loss_percent:g}%" if state.stop_loss_percent is not None else "off"
tp = f"{state.take_profit_percent:g}%" if state.take_profit_percent is not None else "off"
ml = f"{state.max_loss_usd:g} USD" if state.max_loss_usd is not None else "off"
strategy_icon = "" if strategy_ready else "⚠️"
symbol_icon = "" if symbol_ready else "⚠️"
risk_icon = "" if risk_ready else "⚠️"
leverage_icon = "" if leverage_ready else "⚠️"
sl_icon = "" if sl_ready else "⚠️"
if is_trend_strategy:
risk_controls_block = (
"<b>Защита позиции:</b>\n"
f"{sl_icon} Stop Loss · <b>{'required' if not sl_ready else sl}</b>\n"
f"✅ Take Profit · {tp}\n"
f"✅ Max Loss · {ml}"
)
else:
risk_controls_block = (
"<b>Защита позиции:</b>\n"
f"✅ Stop Loss · {sl}\n"
f"✅ Take Profit · {tp}\n"
f"✅ Max Loss · {ml}"
)
config_status = "Все параметры настроены" if is_configured else "⚠️ Настрой все параметры"
text = (
"<b>🤖 Автоторговля</b>\n\n"
"<b>СИСТЕМА</b> · Настройки\n\n"
f"{strategy_icon} Стратегия: <b>{strategy}</b>\n"
f"{symbol_icon} Актив: <b>{symbol}</b>\n"
f"{risk_icon} Риск на сделку: <b>{risk}</b>\n"
f"{leverage_icon} Плечо: <b>{leverage}</b>\n\n"
f"✅ Лимит на сделку: <b>{max_reserved}</b>\n\n"
f"{risk_controls_block}\n\n"
f"{config_status}"
)
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_leverage")
builder.button(text="🏦 Лимит", callback_data="settings:auto_max_reserved")
builder.button(text="🛡️ Риск", callback_data="settings:auto_risk")
builder.button(text="🧯 Защита", callback_data="auto:risk")
builder.button(text="🤖 Автоторговля", callback_data="auto:home")
builder.button(text="⬅️ Назад", callback_data="system:management")
builder.adjust(2, 2, 2, 2)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
_register_system_screen(callback.message, screen="settings_auto")
await callback.answer()
@router.callback_query(F.data == "settings:auto_strategy")
async def open_auto_strategy_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_auto"):
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())
_register_system_screen(callback.message, screen="settings_auto")
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())
await open_auto_settings(callback)
await callback.answer("Стратегия обновлена")
@router.callback_query(F.data == "settings:auto_symbol")
async def open_auto_symbol_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_auto"):
return
text = (
"<b>💱 Актив</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Автоторговля\n\n"
"Выберите актив:"
)
builder = InlineKeyboardBuilder()
builder.button(text="BTC", callback_data="settings:auto_symbol:BTC/USD_LEVERAGE")
builder.button(text="ETH", callback_data="settings:auto_symbol:ETH/USD_LEVERAGE")
builder.button(text="LTC", callback_data="settings:auto_symbol:LTC/USD_LEVERAGE")
builder.button(text="XRP", callback_data="settings:auto_symbol:XRP/USD_LEVERAGE")
builder.button(text="⬅️ Назад", callback_data="settings:auto")
builder.adjust(2, 2, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
_register_system_screen(callback.message, screen="settings_auto")
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)
await open_auto_settings(callback)
await callback.answer("Актив обновлён")
@router.callback_query(F.data == "settings:auto_risk")
async def open_auto_risk_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_auto"):
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())
_register_system_screen(callback.message, screen="settings_auto")
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)
await open_auto_settings(callback)
await callback.answer("Риск обновлён")
@router.callback_query(F.data == "settings:auto_leverage")
async def open_auto_leverage_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_auto"):
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())
_register_system_screen(callback.message, screen="settings_auto")
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)
await open_auto_settings(callback)
await callback.answer("Плечо обновлено")
@router.callback_query(F.data == "settings:trade")
async def open_trade_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_trade"):
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())
_register_system_screen(callback.message, screen="settings_trade")
await callback.answer()
@router.callback_query(F.data == "settings:general")
async def open_general_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_general"):
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())
_register_system_screen(callback.message, screen="settings_general")
await callback.answer()
@router.callback_query(F.data == "settings:journal")
async def open_journal_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_journal"):
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())
_register_system_screen(callback.message, screen="settings_journal")
await callback.answer()
@router.callback_query(F.data == "settings:journal_archive")
async def open_journal_archive_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_journal"):
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())
_register_system_screen(callback.message, screen="settings_journal")
await callback.answer()
@router.callback_query(F.data == "settings:journal_limit")
async def open_journal_limit_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_journal"):
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())
_register_system_screen(callback.message, screen="settings_journal")
await callback.answer()
@router.callback_query(F.data == "settings:journal_retention")
async def open_journal_retention_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_journal"):
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())
_register_system_screen(callback.message, screen="settings_journal")
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 not await _prepare_system_from_callback(callback, screen="system"):
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 not await _prepare_system_from_callback(callback, screen="system_about"):
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())
_register_system_screen(callback.message, screen="system_about")
await callback.answer()
@router.callback_query(F.data == "settings:auto_max_reserved")
async def open_auto_max_reserved_settings(callback: CallbackQuery) -> None:
if not await _prepare_system_from_callback(callback, screen="settings_auto"):
return
text = (
"<b>🏦 Лимит на сделку</b>\n\n"
"<b>СИСТЕМА</b> · Настройки · Автоторговля\n\n"
"Максимальная доля баланса, которую можно зарезервировать под позицию:"
)
builder = InlineKeyboardBuilder()
builder.button(text="25%", callback_data="settings:auto_max_reserved:25")
builder.button(text="50%", callback_data="settings:auto_max_reserved:50")
builder.button(text="75%", callback_data="settings:auto_max_reserved:75")
builder.button(text="100%", callback_data="settings:auto_max_reserved:100")
builder.button(text="off", callback_data="settings:auto_max_reserved:off")
builder.button(text="⬅️ Назад", callback_data="settings:auto")
builder.adjust(2, 2, 1, 1)
await callback.message.edit_text(text, reply_markup=builder.as_markup())
_register_system_screen(callback.message, screen="settings_auto")
await callback.answer()
@router.callback_query(F.data.startswith("settings:auto_max_reserved:"))
async def set_auto_max_reserved(callback: CallbackQuery) -> None:
raw_value = callback.data.split(":", 2)[2]
value = None if raw_value == "off" else float(raw_value)
AutoTradeService().set_max_reserved_balance_percent(value)
await open_auto_settings(callback)
await callback.answer("Max Reserved обновлён")