test_telegram.py

#!/usr/bin/env python3
"""
Test Telegram Integration for All Alert Types
"""

import os
import requests
from dotenv import load_dotenv
from services.telegram import send_telegram

load_dotenv()

BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")

def test_telegram_direct():
    """Test telegram API directly"""
    print("🔍 Testing Telegram API directly...")

    if not BOT_TOKEN or not CHAT_ID:
        print("❌ Telegram credentials not found in .env")
        return False

    try:
        url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
        data = {
            "chat_id": CHAT_ID,
            "text": "🤖 Telegram Integration Test - Direct API call"
        }

        response = requests.post(url, data=data, timeout=10)

        if response.status_code == 200:
            result = response.json()
            if result.get('ok'):
                print("✅ Direct API call successful")
                return True
            else:
                print(f"❌ API returned error: {result}")
                return False
        else:
            print(f"❌ HTTP error: {response.status_code} - {response.text}")
            return False

    except Exception as e:
        print(f"❌ Direct API call failed: {e}")
        return False

def test_service_function():
    """Test the telegram service function"""
    print("\n🔍 Testing Telegram service function...")

    try:
        send_telegram("🤖 Telegram Integration Test - Service function")
        print("✅ Service function called (check Telegram for message)")
        return True
    except Exception as e:
        print(f"❌ Service function failed: {e}")
        return False

def test_all_alert_types():
    """Test all types of alerts that the bot sends"""
    print("\n📨 Testing all alert types...")

    alerts = [
        # Trading session alerts
        ("🚀 BTCUSDT LONG @ $84,200", "Trade entry alert"),
        ("🎯 TP: $86,355 (+2.5%)", "Take profit alert"),
        ("🛡️ SL: $83,258 (-1%)", "Stop loss alert"),

        # Trade execution alerts
        ("💰 BTCUSDT TP HIT @ $86,355", "Take profit hit"),
        ("💸 BTCUSDT SL HIT @ $83,258", "Stop loss hit"),
        ("🔄 BTCUSDT position closed", "Position closure"),

        # Session management alerts
        ("▶️ Started trading session for BTCUSDT (18h duration)", "Session start"),
        ("⏹️ Stopped trading session for BTCUSDT", "Session stop"),

        # System alerts
        ("🚨 EMERGENCY STOP activated - All trading halted", "Emergency stop"),
        ("📊 Daily P&L: -$45.67 (Limit: -$100)", "Daily P&L report"),

        # Session reports
        ("🏆 TRADING SESSION REPORT - BTCUSDT\n• Duration: 18 hours\n• Total Trades: 45\n• Win Rate: 71.1%\n• Total P&L: $15.23", "Session report"),

        # Analysis alerts
        ("🎯 TRADE SIGNAL READY - BTCUSDT", "Signal detection"),
        ("📈 EMA 9: 84120.45, EMA 21: 84095.23", "Technical analysis"),
        ("📊 RSI 14: 67.8", "RSI reading"),
        ("💥 Volume spike detected", "Volume alert"),
    ]

    sent_count = 0
    for message, alert_type in alerts:
        try:
            send_telegram(f"🧪 TEST ALERT - {alert_type}\n{message}")
            sent_count += 1
            print(f"✅ Sent: {alert_type}")
        except Exception as e:
            print(f"❌ Failed to send {alert_type}: {e}")

    print(f"\n📊 Sent {sent_count}/{len(alerts)} test alerts")
    return sent_count == len(alerts)

def get_bot_info():
    """Get telegram bot information"""
    print("\n🤖 Checking Telegram bot info...")

    if not BOT_TOKEN:
        print("❌ Bot token not configured")
        return False

    try:
        url = f"https://api.telegram.org/bot{BOT_TOKEN}/getMe"
        response = requests.get(url, timeout=10)

        if response.status_code == 200:
            result = response.json()
            if result.get('ok'):
                bot = result['result']
                print("✅ Bot connected successfully")
                print(f"   Name: {bot.get('first_name', 'Unknown')}")
                print(f"   Username: @{bot.get('username', 'Unknown')}")
                print(f"   Can read messages: {bot.get('can_read_all_group_messages', False)}")
                return True
            else:
                print(f"❌ Bot info error: {result}")
                return False
        else:
            print(f"❌ HTTP error: {response.status_code}")
            return False

    except Exception as e:
        print(f"❌ Failed to get bot info: {e}")
        return False

def main():
    print("🔔 Telegram Integration Test Suite")
    print("=" * 50)

    # Test bot connection
    bot_ok = get_bot_info()

    if not bot_ok:
        print("\n❌ Cannot proceed without bot connection")
        return

    # Test direct API
    direct_ok = test_telegram_direct()

    # Test service function
    service_ok = test_service_function()

    # Test all alert types
    alerts_ok = test_all_alert_types()

    # Summary
    print("\n" + "=" * 50)
    print("📊 TELEGRAM INTEGRATION TEST RESULTS:")
    print(f"🤖 Bot Connection: {'✅ PASS' if bot_ok else '❌ FAIL'}")
    print(f"🔗 Direct API: {'✅ PASS' if direct_ok else '❌ FAIL'}")
    print(f"⚙️ Service Function: {'✅ PASS' if service_ok else '❌ FAIL'}")
    print(f"📨 All Alerts: {'✅ PASS' if alerts_ok else '❌ FAIL'}")

    all_pass = bot_ok and direct_ok and service_ok and alerts_ok

    if all_pass:
        print("\n🎉 ALL TESTS PASSED!")
        print("📱 Check your Telegram for test messages")
        print("✅ Telegram integration is fully working")
    else:
        print("\n⚠️ Some tests failed. Check your configuration:")
        print("1. Verify TELEGRAM_BOT_TOKEN in .env")
        print("2. Verify TELEGRAM_CHAT_ID in .env")
        print("3. Send a message to your bot and run get_chat_id.py")
        print("4. Check internet connection")

    print("\n🔧 Current Configuration:")
    print(f"   Bot Token: {BOT_TOKEN[:20]}..." if BOT_TOKEN else "   Bot Token: Not set")
    print(f"   Chat ID: {CHAT_ID}" if CHAT_ID else "   Chat ID: Not set")

if __name__ == "__main__":
    main()