Trading signal subscription system development

Development of Trading Signal Subscription System

A trading signal subscription system allows users to receive notifications about trading opportunities from analysts or algorithmic systems. Unlike copy trading, the user makes decisions independently — a signal is a recommendation, not automatic execution.

System Components

Signal Providers — signal sources: analyst traders, algorithmic systems, on-chain analytics.

Signal Format — structured message: instrument, direction, entry price, take profit levels, stop loss, timeframe, rationale.

Distribution Engine — signal delivery to all subscribers via different channels.

Subscription Management — subscription management, pricing, payment.

Performance Tracking — tracking results of each signal to calculate provider win rate.

Signal Data Model

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      # or None for 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]   # for futures
    risk_pct: Optional[float] # recommended % of capital
    timeframe: str            # 4h, 1d
    rationale: str            # text explanation
    chart_url: Optional[str]  # screenshot of markup
    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):
        # Get all active subscribers of this provider
        subscribers = await self.subscription_repo.get_active_subscribers(
            provider_id=signal.provider_id
        )

        # Group by preferred notification channels
        by_channel: dict[str, list] = {}
        for sub in subscribers:
            for channel in sub.notification_channels:
                by_channel.setdefault(channel, []).append(sub.user_id)

        # Send via channels in parallel
        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)

        # Log distribution
        await self.signal_repo.mark_distributed(signal.id, len(subscribers))

Telegram Delivery

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}

**Entry:** {'market' if signal.entry_type == 'MARKET' else f'${signal.entry_price:,.2f}'}
**Stop Loss:** ${signal.stop_loss:,.2f}
**Take Profit:**
{tp_lines}

**Timeframe:** {signal.timeframe}
**Risk:** {signal.risk_pct or 1}% of deposit

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

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

        # Batch by 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

Performance Tracking

class SignalPerformanceTracker:
    async def track_signal_outcome(self, signal: TradingSignal):
        """Track signal result based on market data"""
        entry_time = signal.created_at

        # Check if entry was reached
        entry_price = await self.find_entry_price(signal)
        if not entry_price:
            await self.mark_signal_missed(signal.id)
            return

        # Track TP and 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,
        )

Accumulated statistics of results — key indicator for new subscribers. Win rate, average R:R, P&L over time, percentage of hit TP1/TP2/TP3 vs SL — all should be visible on signal provider page.