feat(live): add live screens for market and portfolio

app/src/telegram/live/runner.py

@@ -0,0 +1,133 @@
+# app/src/telegram/live/runner.py
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+from typing import Callable
+
+from aiogram import Bot
+
+
+@dataclass(slots=True)
+class LiveScreen:
+    # имя live-экрана: market / portfolio / journal
+    screen: str
+
+    # Telegram bot instance
+    bot: Bot
+
+    # чат, где находится live-экран
+    chat_id: int
+
+    # сообщение, которое нужно автообновлять
+    message_id: int
+
+    # функция сборки текста экрана
+    render_text: Callable[[], str]
+
+    # функция сборки клавиатуры экрана
+    render_markup: Callable[[], object]
+
+    # интервал обновления в секундах
+    interval_seconds: int = 5
+
+
+class LiveScreenRunner:
+    _screens: dict[str, LiveScreen] = {}
+    _tasks: dict[str, asyncio.Task] = {}
+    _current_screen: str | None = None
+
+    # переключить активный экран
+    @classmethod
+    def set_current_screen(cls, screen: str) -> None:
+        cls._current_screen = screen
+
+    # зарегистрировать live-экран
+    @classmethod
+    def register_screen(cls, live_screen: LiveScreen) -> None:
+        cls._screens[live_screen.screen] = live_screen
+
+    # удалить старый live-экран из Telegram
+    @classmethod
+    async def delete_screen(
+        cls,
+        *,
+        screen: str,
+        bot: Bot,
+        chat_id: int,
+    ) -> None:
+        live_screen = cls._screens.get(screen)
+
+        if live_screen is None:
+            return
+
+        if live_screen.chat_id != chat_id:
+            return
+
+        try:
+            await bot.delete_message(
+                chat_id=live_screen.chat_id,
+                message_id=live_screen.message_id,
+            )
+        except Exception:
+            pass
+
+        cls._screens.pop(screen, None)
+
+    # запустить автообновление экрана
+    @classmethod
+    def start(cls, screen: str) -> None:
+        task = cls._tasks.get(screen)
+
+        if task is not None and not task.done():
+            return
+
+        cls._tasks[screen] = asyncio.create_task(cls._worker(screen))
+
+    # остановить автообновление экрана
+    @classmethod
+    def stop(cls, screen: str) -> None:
+        task = cls._tasks.get(screen)
+
+        if task is None:
+            return
+
+        task.cancel()
+        cls._tasks.pop(screen, None)
+
+    # фоновый цикл обновления одного экрана
+    @classmethod
+    async def _worker(cls, screen: str) -> None:
+        while True:
+            await cls._refresh_screen(screen)
+            await asyncio.sleep(cls._screen_interval(screen))
+
+    # получить интервал обновления экрана
+    @classmethod
+    def _screen_interval(cls, screen: str) -> int:
+        live_screen = cls._screens.get(screen)
+        if live_screen is None:
+            return 5
+
+        return live_screen.interval_seconds
+
+    # обновить Telegram-сообщение live-экрана
+    @classmethod
+    async def _refresh_screen(cls, screen: str) -> None:
+        if cls._current_screen != screen:
+            return
+
+        live_screen = cls._screens.get(screen)
+        if live_screen is None:
+            return
+
+        try:
+            await live_screen.bot.edit_message_text(
+                chat_id=live_screen.chat_id,
+                message_id=live_screen.message_id,
+                text=live_screen.render_text(),
+                reply_markup=live_screen.render_markup(),
+            )
+        except Exception:
+            pass