init

.dockerignore

@@ -0,0 +1,5 @@
+.venv
+.env
+charts
+__pycache__
+.git

.env.example

@@ -0,0 +1,11 @@
+TELEGRAM_BOT_TOKEN=
+TELEGRAM_CHANNEL_ID=
+IMG_PATH=charts
+WATERMARK_PATH=assets/waternarkResize.png
+N_BARS=100
+INTERVAL=1m
+PAIRS=EURJPY,AUDJPY,AUDCHF,EURUSD,AUDCAD,USDCAD,EURGBP,USDCHF
+SCREENER=forex
+EXCHANGE=FX_IDC
+TV_USERNAME=
+TV_PASSWORD=

.gitignore

@@ -0,0 +1,9 @@
+.venv/
+.env
+__pycache__/
+*.py[cod]
+*.egg-info/
+dist/
+build/
+charts/
+.python-version.bak

.python-version

@@ -0,0 +1 @@
+3.12

Dockerfile

@@ -0,0 +1,34 @@
+# syntax=docker/dockerfile:1
+
+FROM python:3.12-slim-bookworm
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates curl gnupg \
+    && curl -fsSL https://dl.google.com/linux/linux_signing_key.pub \
+        | gpg --dearmor -o /usr/share/keyrings/google-linux.gpg \
+    && echo "deb [arch=amd64 signed-by=/usr/share/keyrings/google-linux.gpg] http://dl.google.com/linux/chrome/deb/ stable main" \
+        > /etc/apt/sources.list.d/google-chrome.list \
+    && apt-get update \
+    && apt-get install -y --no-install-recommends google-chrome-stable \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+ENV UV_COMPILE_BYTECODE=1 \
+    UV_LINK_MODE=copy \
+    UV_PYTHON_DOWNLOADS=0
+
+COPY pyproject.toml uv.lock README.md .python-version ./
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --frozen --no-install-project --no-dev
+
+COPY src ./src
+COPY assets ./assets
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --frozen --no-dev
+
+CMD ["uv", "run", "qpokus"]

README.md

@@ -0,0 +1,53 @@
+# Qpokus back
+
+Telegram-бот сигналов по форекс: технический анализ TradingView, свечной график с полосами Боллинджера.
+
+Два режима:
+
+- `bot` — inline-кнопки пар, сигнал уходит пользователю
+- `publisher` — цикл автопубликации в канал
+
+## Требования
+
+- [uv](https://docs.astral.sh/uv/)
+- Python 3.12 (поставится через `uv python pin`)
+- Chrome или Chromium — нужен Kaleido, чтобы сохранить график в PNG
+
+## Установка
+
+```bash
+cd qpokus-back
+uv sync
+cp .env.example .env
+```
+
+На Windows PowerShell: `Copy-Item .env.example .env`
+
+Заполните `.env`:
+
+- `TELEGRAM_BOT_TOKEN` — токен BotFather
+- `TELEGRAM_CHANNEL_ID` — id канала (нужен для `publisher`)
+
+Опционально: `WATERMARK_PATH`, `TV_USERNAME`, `TV_PASSWORD`, список `PAIRS`.
+
+По умолчанию один процесс: интерактивный бот и посты в канал.
+
+```bash
+uv run qpokus
+```
+
+Отдельно:
+
+```bash
+uv run qpokus bot
+uv run qpokus publisher
+```
+
+## Docker
+
+```bash
+docker build -t qpokus-back .
+docker run --env-file .env qpokus-back
+```
+
+По умолчанию контейнер стартует бота и публикацию в канал.

pyproject.toml

@@ -0,0 +1,26 @@
+[project]
+name = "qpokus"
+version = "0.1.0"
+description = "Telegram forex signal bot from TradingView TA"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "kaleido>=1.3.0",
+    "numpy>=2.5.2",
+    "pandas>=3.0.5",
+    "pillow>=12.3.0",
+    "plotly>=6.9.0",
+    "pydantic-settings>=2.15.0",
+    "pytelegrambotapi>=4.36.1",
+    "python-dotenv>=1.2.1",
+    "tradingview-ta>=3.3.0",
+    "tvdatafeed-enhanced>=2.2.1",
+    "websocket-client>=1.8.0",
+]
+
+[project.scripts]
+qpokus = "qpokus.__main__:main"
+
+[build-system]
+requires = ["uv_build>=0.9.30,<0.10.0"]
+build-backend = "uv_build"

src/qpokus/__init__.py

@@ -0,0 +1,3 @@
+"""Telegram forex signal bot based on TradingView technical analysis."""
+
+__version__ = "0.1.0"

src/qpokus/__main__.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+
+
+def main(argv: list[str] | None = None) -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    parser = argparse.ArgumentParser(prog="qpokus", description="QP forex Telegram signals")
+    parser.add_argument(
+        "command",
+        nargs="?",
+        default="all",
+        choices=("all", "bot", "publisher"),
+        help="all: bot + channel in one process (default)",
+    )
+    args = parser.parse_args(argv)
+
+    if args.command == "bot":
+        from qpokus.bot import run_bot
+
+        run_bot()
+        return
+
+    if args.command == "publisher":
+        from qpokus.publisher import run_publisher
+
+        run_publisher()
+        return
+
+    from qpokus.app import run_all
+
+    run_all()
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])

src/qpokus/app.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import logging
+import threading
+
+import telebot
+
+from qpokus.bot import register_handlers
+from qpokus.config import load_settings
+from qpokus.engine import SignalEngine
+from qpokus.publisher import publisher_loop
+
+logger = logging.getLogger(__name__)
+
+
+def run_all() -> None:
+    settings = load_settings()
+    bot = telebot.TeleBot(settings.telegram_bot_token)
+    engine = SignalEngine(settings)
+    register_handlers(bot, engine, list(settings.pairs))
+
+    if settings.telegram_channel_id:
+        threading.Thread(
+            target=publisher_loop,
+            args=(bot, engine, settings.telegram_channel_id),
+            daemon=True,
+            name="qpokus-publisher",
+        ).start()
+    else:
+        logger.warning("TELEGRAM_CHANNEL_ID is empty, channel publisher is off")
+
+    logger.info("Interactive bot started")
+    bot.infinity_polling()

src/qpokus/bot.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import logging
+
+import telebot
+from telebot.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message
+
+from qpokus.config import load_settings
+from qpokus.engine import SignalEngine
+from qpokus.telegram_send import send_signal
+
+logger = logging.getLogger(__name__)
+
+
+def gen_markup(symbols: list[str]) -> InlineKeyboardMarkup:
+    markup = InlineKeyboardMarkup()
+    markup.row_width = 2
+    markup.add(*[InlineKeyboardButton(text=symbol, callback_data=symbol) for symbol in symbols])
+    return markup
+
+
+def register_handlers(bot: telebot.TeleBot, engine: SignalEngine, symbols: list[str]) -> None:
+    @bot.message_handler(commands=["help", "start"])
+    def send_welcome(message: Message) -> None:
+        bot.send_message(
+            message.chat.id,
+            "Выбирай на чем бабло проебешь",
+            reply_markup=gen_markup(symbols),
+        )
+
+    @bot.callback_query_handler(func=lambda call: True)
+    def callback_query(call: CallbackQuery) -> None:
+        symbol = call.data or ""
+        bot.answer_callback_query(call.id)
+        try:
+            pair = engine.market.pair_by_symbol(symbol)
+        except KeyError:
+            bot.send_message(call.message.chat.id, "Неизвестная пара")
+            return
+
+        chat_id = call.message.chat.id
+        try:
+            bot.edit_message_text(
+                chat_id=chat_id,
+                message_id=call.message.id,
+                text=f"{call.message.text} - {symbol}",
+            )
+        except Exception:
+            logger.exception("Failed to edit callback message")
+
+        try:
+            payload = engine.build_signal(pair)
+            send_signal(bot, chat_id, payload)
+        except Exception:
+            logger.exception("Failed to build signal for %s", symbol)
+            bot.send_message(chat_id, "Не удалось отправить")
+
+
+def run_bot() -> None:
+    settings = load_settings()
+    bot = telebot.TeleBot(settings.telegram_bot_token)
+    engine = SignalEngine(settings)
+    register_handlers(bot, engine, list(settings.pairs))
+    logger.info("Interactive bot started")
+    bot.infinity_polling()

src/qpokus/charts.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pandas as pd
+import plotly.graph_objs as go
+from PIL import Image
+from plotly.subplots import make_subplots
+
+from qpokus.indicators import BollingerBands
+from qpokus.market import Pair
+
+
+def _json_x(values: object) -> list[str]:
+    """Kaleido/orjson cannot serialize pandas.Timestamp."""
+    return pd.to_datetime(values).strftime("%Y-%m-%d %H:%M:%S").tolist()
+
+
+def _json_y(values: object) -> list[float]:
+    return pd.Series(values, dtype="float64").tolist()
+
+
+def render_chart(
+    df: pd.DataFrame,
+    bands: BollingerBands,
+    pair: Pair,
+    dest: Path,
+    watermark_path: Path | None = None,
+) -> Path:
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    fig = make_subplots(shared_xaxes=True, shared_yaxes=True, vertical_spacing=0.1)
+    fig.update_layout(
+        xaxis_rangeslider_visible=False,
+        template="plotly_dark",
+        width=2400,
+        height=1080,
+        title=f"{pair.exchange}:{pair.symbol}",
+        showlegend=False,
+    )
+    x_ohlc = _json_x(df.index)
+    x_bands = _json_x(bands.x)
+    fig.add_trace(
+        go.Candlestick(
+            x=x_ohlc,
+            open=_json_y(df["open"]),
+            high=_json_y(df["high"]),
+            low=_json_y(df["low"]),
+            close=_json_y(df["close"]),
+            name=f"{pair.exchange}:{pair.symbol}",
+        )
+    )
+    fig.add_trace(
+        go.Scatter(x=x_bands, y=_json_y(bands.bb), fill=None, mode="lines", name="BB")
+    )
+    fig.add_trace(
+        go.Scatter(
+            x=x_bands,
+            y=_json_y(bands.pstd),
+            fill=None,
+            mode="lines",
+            line_color="indigo",
+            name="pSTD",
+        )
+    )
+    fig.add_trace(
+        go.Scatter(
+            x=x_bands,
+            y=_json_y(bands.mstd),
+            fill="tonexty",
+            mode="lines",
+            line_color="indigo",
+            name="mSTD",
+        )
+    )
+    fig.write_image(str(dest))
+    if watermark_path is not None and watermark_path.exists():
+        apply_watermark(dest, watermark_path)
+    return dest
+
+
+def apply_watermark(
+    image_path: Path,
+    watermark_path: Path,
+    position: tuple[int, int] = (0, 0),
+    transparency: int = 10,
+) -> None:
+    base_image = Image.open(image_path).convert("RGBA")
+    watermark = Image.open(watermark_path).convert("RGBA")
+    width, height = base_image.size
+    alpha = watermark.split()[3].point(lambda i: i * transparency / 100.0)
+    watermark.putalpha(alpha)
+    transparent = Image.new("RGBA", (width, height), (0, 0, 0, 0))
+    transparent.paste(base_image, (0, 0))
+    transparent.paste(watermark, position, mask=watermark)
+    transparent.save(image_path)

src/qpokus/config.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from typing import Annotated
+
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
+
+_PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+
+DEFAULT_PAIRS = (
+    "EURJPY",
+    "AUDJPY",
+    "AUDCHF",
+    "EURUSD",
+    "AUDCAD",
+    "USDCAD",
+    "EURGBP",
+    "USDCHF",
+)
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=_PROJECT_ROOT / ".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    telegram_bot_token: str
+    telegram_channel_id: str = ""
+    img_path: Path = Path("charts")
+    watermark_path: Path | None = None
+    n_bars: int = 100
+    interval: str = "1m"
+    pairs: Annotated[list[str], NoDecode] = Field(default_factory=lambda: list(DEFAULT_PAIRS))
+    screener: str = "forex"
+    exchange: str = "FX_IDC"
+    tv_username: str | None = None
+    tv_password: str | None = None
+
+    @field_validator("pairs", mode="before")
+    @classmethod
+    def split_pairs(cls, value: object) -> object:
+        if isinstance(value, str):
+            return [item.strip().upper() for item in value.split(",") if item.strip()]
+        return value
+
+    @field_validator("watermark_path", mode="before")
+    @classmethod
+    def empty_watermark(cls, value: object) -> object:
+        if value in ("", None):
+            return None
+        return value
+
+    @field_validator("tv_username", "tv_password", mode="before")
+    @classmethod
+    def empty_optional(cls, value: object) -> object:
+        if value == "":
+            return None
+        return value
+
+    def ensure_img_dir(self) -> Path:
+        path = self.img_path
+        if not path.is_absolute():
+            path = _PROJECT_ROOT / path
+        path.mkdir(parents=True, exist_ok=True)
+        self.img_path = path
+        if self.watermark_path is not None and not self.watermark_path.is_absolute():
+            self.watermark_path = _PROJECT_ROOT / self.watermark_path
+        return path
+
+
+def load_settings() -> Settings:
+    return Settings()

src/qpokus/engine.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+import threading
+
+from qpokus.charts import render_chart
+from qpokus.config import Settings
+from qpokus.indicators import bollinger_bands
+from qpokus.market import Analysis, MarketClient, Pair
+from qpokus.messages import format_signal, is_directional, is_strong
+
+
+@dataclass(frozen=True)
+class SignalPayload:
+    analysis: Analysis
+    text: str
+    image_path: Path
+
+
+class SignalEngine:
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+        self.market = MarketClient(settings)
+        self._lock = threading.Lock()
+        settings.ensure_img_dir()
+
+    def chart_path(self, pair: Pair) -> Path:
+        return self.settings.img_path / f"{pair.exchange}_{pair.symbol}.png"
+
+    def build_signal(self, pair: Pair, analysis: Analysis | None = None) -> SignalPayload:
+        if analysis is None:
+            analysis = self.market.get_analysis(pair)
+        with self._lock:
+            ohlc = self.market.get_ohlc(pair)
+            bands = bollinger_bands(ohlc)
+            image_path = render_chart(
+                ohlc,
+                bands,
+                pair,
+                self.chart_path(pair),
+                watermark_path=self.settings.watermark_path,
+            )
+        return SignalPayload(
+            analysis=analysis,
+            text=format_signal(analysis),
+            image_path=image_path,
+        )
+
+    def pick_channel_pair(self) -> Analysis | None:
+        analyses = self.market.get_analyses(self.market.pairs())
+        for analysis in analyses:
+            if is_strong(analysis.recommendation):
+                return analysis
+        for analysis in analyses:
+            if is_directional(analysis.recommendation):
+                return analysis
+        return None

src/qpokus/indicators.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+import pandas as pd
+
+
+@dataclass(frozen=True)
+class BollingerBands:
+    x: np.ndarray
+    bb: np.ndarray
+    pstd: np.ndarray
+    mstd: np.ndarray
+
+
+def bollinger_bands(
+    df: pd.DataFrame,
+    *,
+    window: int = 10,
+    value_type: str = "low",
+    k_dev: float = 2.0,
+) -> BollingerBands:
+    """SMA midline and windowed std bands, matching the reference notebook."""
+    base = df.reset_index(drop=True)
+    col = base[value_type].to_numpy(dtype=float)
+    sma = np.convolve(col, np.ones(window), "valid") / window
+    std = np.array(
+        [np.std(col[i : i + window], ddof=1) for i in range(len(col) - window + 1)],
+        dtype=float,
+    )
+    dates = np.array(base["date"][window - 1:].to_list())
+    return BollingerBands(
+        x=dates,
+        bb=sma,
+        pstd=sma + std * k_dev,
+        mstd=sma - std * k_dev,
+    )

src/qpokus/intervals.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import tradingview_ta
+from tvDatafeed import Interval as DfInterval
+
+
+@dataclass(frozen=True)
+class Timeframe:
+    ta: str
+    df: DfInterval
+
+
+INTERVALS: dict[str, Timeframe] = {
+    "1m": Timeframe(tradingview_ta.Interval.INTERVAL_1_MINUTE, DfInterval.in_1_minute),
+    "5m": Timeframe(tradingview_ta.Interval.INTERVAL_5_MINUTES, DfInterval.in_5_minute),
+    "15m": Timeframe(tradingview_ta.Interval.INTERVAL_15_MINUTES, DfInterval.in_15_minute),
+    "30m": Timeframe(tradingview_ta.Interval.INTERVAL_30_MINUTES, DfInterval.in_30_minute),
+    "1h": Timeframe(tradingview_ta.Interval.INTERVAL_1_HOUR, DfInterval.in_1_hour),
+    "2h": Timeframe(tradingview_ta.Interval.INTERVAL_2_HOURS, DfInterval.in_2_hour),
+    "4h": Timeframe(tradingview_ta.Interval.INTERVAL_4_HOURS, DfInterval.in_4_hour),
+    "1d": Timeframe(tradingview_ta.Interval.INTERVAL_1_DAY, DfInterval.in_daily),
+    "1w": Timeframe(tradingview_ta.Interval.INTERVAL_1_WEEK, DfInterval.in_weekly),
+    "1M": Timeframe(tradingview_ta.Interval.INTERVAL_1_MONTH, DfInterval.in_monthly),
+}
+
+
+def get_timeframe(key: str) -> Timeframe:
+    try:
+        return INTERVALS[key]
+    except KeyError as exc:
+        known = ", ".join(INTERVALS)
+        raise ValueError(f"Unknown interval {key!r}. Expected one of: {known}") from exc

src/qpokus/market.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import logging
+import time
+from collections.abc import Callable
+from typing import TypeVar
+
+import pandas as pd
+from tradingview_ta import TA_Handler, get_multiple_analysis
+from tvDatafeed import TvDatafeed
+
+from qpokus.config import Settings
+from qpokus.intervals import get_timeframe
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+@dataclass(frozen=True)
+class Pair:
+    symbol: str
+    screener: str
+    exchange: str
+
+
+@dataclass(frozen=True)
+class Analysis:
+    pair: Pair
+    recommendation: str
+    open_price: float
+
+
+def _looks_like_rate_limit(exc: BaseException) -> bool:
+    return "429" in str(exc) or isinstance(exc, json.JSONDecodeError)
+
+
+def _call_with_retry(fn: Callable[[], T], *, attempts: int = 4, base_delay: float = 8.0) -> T:
+    last: BaseException | None = None
+    for step in range(attempts):
+        try:
+            return fn()
+        except Exception as exc:
+            last = exc
+            if not _looks_like_rate_limit(exc) or step == attempts - 1:
+                raise
+            delay = base_delay * (2**step)
+            logger.warning("TradingView rate limit, retry in %ss (%s)", delay, exc)
+            time.sleep(delay)
+    assert last is not None
+    raise last
+
+
+class MarketClient:
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+        self.timeframe = get_timeframe(settings.interval)
+        if settings.tv_username and settings.tv_password:
+            self.tv = TvDatafeed(settings.tv_username, settings.tv_password)
+        else:
+            self.tv = TvDatafeed()
+
+    def pairs(self) -> list[Pair]:
+        return [
+            Pair(symbol=symbol, screener=self.settings.screener, exchange=self.settings.exchange)
+            for symbol in self.settings.pairs
+        ]
+
+    def pair_by_symbol(self, symbol: str) -> Pair:
+        wanted = symbol.upper()
+        for pair in self.pairs():
+            if pair.symbol == wanted:
+                return pair
+        raise KeyError(f"Unknown symbol: {symbol}")
+
+    def get_analysis(self, pair: Pair) -> Analysis:
+        def fetch() -> Analysis:
+            result = TA_Handler(
+                symbol=pair.symbol,
+                screener=pair.screener,
+                exchange=pair.exchange,
+                interval=self.timeframe.ta,
+                timeout=30,
+            ).get_analysis()
+            return Analysis(
+                pair=pair,
+                recommendation=str(result.summary["RECOMMENDATION"]),
+                open_price=float(result.indicators["open"]),
+            )
+
+        return _call_with_retry(fetch)
+
+    def get_analyses(self, pairs: list[Pair]) -> list[Analysis]:
+        if not pairs:
+            return []
+        symbols = [f"{pair.exchange}:{pair.symbol}" for pair in pairs]
+        screener = pairs[0].screener
+
+        def fetch() -> dict:
+            return get_multiple_analysis(
+                screener=screener,
+                interval=self.timeframe.ta,
+                symbols=symbols,
+                timeout=30,
+            )
+
+        raw = _call_with_retry(fetch)
+        by_key = {key.upper(): value for key, value in (raw or {}).items()}
+        analyses: list[Analysis] = []
+        for pair in pairs:
+            item = by_key.get(f"{pair.exchange}:{pair.symbol}".upper())
+            if item is None:
+                logger.warning("No TradingView analysis for %s", pair.symbol)
+                continue
+            analyses.append(
+                Analysis(
+                    pair=pair,
+                    recommendation=str(item.summary["RECOMMENDATION"]),
+                    open_price=float(item.indicators["open"]),
+                )
+            )
+        return analyses
+
+    def get_ohlc(self, pair: Pair) -> pd.DataFrame:
+        def fetch() -> pd.DataFrame:
+            df = self.tv.get_hist(
+                symbol=pair.symbol,
+                exchange=pair.exchange,
+                interval=self.timeframe.df,
+                n_bars=self.settings.n_bars,
+            )
+            if df is None or df.empty:
+                raise RuntimeError(f"No OHLC data for {pair.exchange}:{pair.symbol}")
+            out = df.copy()
+            out["date"] = out.index
+            return out
+
+        return _call_with_retry(fetch)

src/qpokus/messages.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import random
+
+from qpokus.market import Analysis
+
+STRONG_BUY = {"STRONG_BUY", "BUY"}
+STRONG_SELL = {"STRONG_SELL", "SELL"}
+
+
+def is_directional(recommendation: str) -> bool:
+    return recommendation in STRONG_BUY | STRONG_SELL
+
+
+def is_strong(recommendation: str) -> bool:
+    return recommendation in {"STRONG_BUY", "STRONG_SELL"}
+
+
+def format_signal(analysis: Analysis) -> str:
+    recommendation = analysis.recommendation
+    if recommendation in STRONG_BUY:
+        color = "🟢"
+        direction = "ВВЕРХ"
+    elif recommendation in STRONG_SELL:
+        color = "🔴"
+        direction = "ВНИЗ"
+    else:
+        color = "🟣"
+        direction = "ПЛАТО"
+
+    if recommendation == "NEUTRAL":
+        return f"""
+Пара: *#{analysis.pair.symbol}*
+
+Сигнал: *{direction}* {color}
+Цена открытия: *{analysis.open_price}*
+_Резюме_:
+Технический анализ — *{direction}*
+"""
+
+    deal_minutes = random.choice(range(1, 5))
+    strength = random.choice(range(65, 85))
+    overlap = random.choice(range(10, 25))
+    return f"""
+Пара: *#{analysis.pair.symbol}*
+
+Сигнал: *{direction}* {color}
+Цена открытия: *{analysis.open_price}*
+Время сдеки: *{deal_minutes} мин.*⏰
+Сила направления движения в сторону хода — {strength}%
+_Резюме_:
+Рассматриваем вход в рынок — *{direction}*.
+Время выхода по таймингу рассчитано на *{deal_minutes} мин.*
+Вероятность использования перекрытия — *{overlap}%*.
+Технический анализ — *{direction}*
+"""

src/qpokus/publisher.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import logging
+import random
+import time
+
+import telebot
+
+from qpokus.config import load_settings
+from qpokus.engine import SignalEngine
+from qpokus.telegram_send import send_signal
+
+logger = logging.getLogger(__name__)
+
+
+def publisher_loop(bot: telebot.TeleBot, engine: SignalEngine, channel_id: str) -> None:
+    logger.info("Publisher started for channel %s", channel_id)
+    while True:
+        try:
+            picked = engine.pick_channel_pair()
+            if picked is None:
+                logger.info("No directional signal, skip send")
+            else:
+                payload = engine.build_signal(picked.pair, analysis=picked)
+                send_signal(bot, channel_id, payload)
+        except Exception as exc:
+            logger.exception("Failed to publish a signal")
+            if "429" in str(exc):
+                pause = 300
+                logger.info("TradingView 429, sleeping %s seconds", pause)
+                time.sleep(pause)
+                continue
+        pause = random.choice(range(1, 5)) * 60
+        logger.info("Sleeping %s seconds", pause)
+        time.sleep(pause)
+
+
+def run_publisher() -> None:
+    settings = load_settings()
+    if not settings.telegram_channel_id:
+        raise SystemExit("TELEGRAM_CHANNEL_ID is required for publisher mode")
+
+    bot = telebot.TeleBot(settings.telegram_bot_token)
+    engine = SignalEngine(settings)
+    publisher_loop(bot, engine, settings.telegram_channel_id)

src/qpokus/telegram_send.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import telebot
+
+from qpokus.engine import SignalPayload
+
+logger = logging.getLogger(__name__)
+
+
+def send_signal(bot: telebot.TeleBot, chat_id: int | str, payload: SignalPayload) -> None:
+    logger.info("Sending %s to %s", payload.analysis.pair.symbol, chat_id)
+    with Path(payload.image_path).open("rb") as photo:
+        bot.send_photo(
+            chat_id,
+            photo,
+            caption=payload.text,
+            parse_mode="Markdown",
+        )