Розробка системи підписки на торгові сигнали

Розробка системи підписки на торговельні сигнали

Система підписки на торговельні сигнали дозволяє користувачам отримувати сповіщення про торговельні можливості від аналітиків або алгоритмічних систем. На відміну від copy trading, користувач приймає рішення самостійно — сигнал це рекомендація, не автоматичне виконання.

Компоненти системи

Signal Providers — джерела сигналів: аналітики-трейдери, алгоритмічні системи, on-chain аналітика.

Signal Format — структуроване повідомлення: інструмент, напрямок, ціна входу, рівні take profit, stop loss, timeframe, обґрунтування.

Distribution Engine — доставка сигналу всім підписникам через різні канали.

Subscription Management — керування підписками, тарифи, оплата.

Performance Tracking — відстеження результатів кожного сигналу для розрахунку win rate провайдера.

Модель даних сигналу

from pydantic import BaseModel
from decimal import Decimal
from datetime import datetime
from typing import Optional

class TradingSignal(BaseModel):
    id: str
    provider_id: str
    symbol: str               # BTC/USDT
    exchange: str             # binance
    direction: str            # LONG / SHORT
    entry_type: str           # MARKET / LIMIT / ZONE
    entry_price: Decimal      # або None для market
    entry_zone_low: Optional[Decimal]
    entry_zone_high: Optional[Decimal]
    take_profit_levels: list[Decimal]  # [tp1, tp2, tp3]
    stop_loss: Decimal
    leverage: Optional[int]   # для фьючерсів
    risk_pct: Optional[float] # рекомендуємий % капіталу
    timeframe: str            # 4h, 1d
    rationale: str            # текстове обґрунтування
    chart_url: Optional[str]  # скриншот розмітки
    expires_at: Optional[datetime]
    created_at: datetime = datetime.utcnow()

Distribution Engine

class SignalDistributor:
    def __init__(self, telegram_bot, email_service, push_service, websocket_hub):
        self.channels = {
            'telegram': telegram_bot,
            'email': email_service,
            'push': push_service,
            'websocket': websocket_hub,
        }

    async def distribute(self, signal: TradingSignal):
        # Отримуємо всіх активних підписників цього провайдера
        subscribers = await self.subscription_repo.get_active_subscribers(
            provider_id=signal.provider_id
        )

        # Групуємо за предпочтительними каналами сповіщень
        by_channel: dict[str, list] = {}
        for sub in subscribers:
            for channel in sub.notification_channels:
                by_channel.setdefault(channel, []).append(sub.user_id)

        # Паралельно розсилаємо по каналах
        tasks = []
        for channel, user_ids in by_channel.items():
            handler = self.channels.get(channel)
            if handler:
                tasks.append(handler.send_signal(signal, user_ids))

        await asyncio.gather(*tasks, return_exceptions=True)

        # Логуємо факт розсилання
        await self.signal_repo.mark_distributed(signal.id, len(subscribers))

Telegram доставка

class TelegramSignalBot:
    def format_signal(self, signal: TradingSignal) -> str:
        tp_lines = '\n'.join(
            f"  TP{i+1}: ${tp:,.2f}"
            for i, tp in enumerate(signal.take_profit_levels)
        )

        return f"""
📊 **{signal.symbol}** — {signal.direction}

**Вхід:** {'рыночний' if signal.entry_type == 'MARKET' else f'${signal.entry_price:,.2f}'}
**Stop Loss:** ${signal.stop_loss:,.2f}
**Take Profit:**
{tp_lines}

**Таймфрейм:** {signal.timeframe}
**Риск:** {signal.risk_pct or 1}% від депозиту

📝 {signal.rationale}
        """.strip()

    async def send_signal(self, signal: TradingSignal, user_ids: list[str]):
        text = self.format_signal(signal)

        # Батчами по 30 (Telegram rate limit)
        for batch in chunks(user_ids, 30):
            tasks = [
                self.bot.send_message(user_id, text, parse_mode='Markdown')
                for user_id in batch
            ]
            await asyncio.gather(*tasks, return_exceptions=True)
            await asyncio.sleep(1)  # rate limit

Відстеження продуктивності

class SignalPerformanceTracker:
    async def track_signal_outcome(self, signal: TradingSignal):
        """Відстежуємо результат сигналу за ринковими даними"""
        entry_time = signal.created_at

        # Перевіряємо чи досягнута ціна входу
        entry_price = await self.find_entry_price(signal)
        if not entry_price:
            await self.mark_signal_missed(signal.id)
            return

        # Відстежуємо TP та SL
        outcome = await self.monitor_until_close(
            symbol=signal.symbol,
            direction=signal.direction,
            entry=entry_price,
            tp_levels=signal.take_profit_levels,
            sl=signal.stop_loss,
        )

        await self.signal_repo.save_outcome(
            signal_id=signal.id,
            entry_price=entry_price,
            exit_price=outcome.exit_price,
            exit_reason=outcome.reason,  # 'TP1', 'TP2', 'SL', 'EXPIRED'
            pnl_pct=outcome.pnl_pct,
        )

Накопичена статистика результатів — ключовий показник для нових підписників. Win rate, середній R:R, P&L за часом, відсоток сработав TP1/TP2/TP3 vs SL — все це повинно бути видно на сторінці провайдера сигналів.