Monitor New Token Launches on Twitter: Early Alpha Detection

queries_eth = [
    "0x AND (just launched OR stealth launch OR fair launch) -filter:retweets",
    "contract address 0x AND (dex OR uniswap OR liquidity) -filter:retweets",
    "new token 0x AND (etherscan OR dextools) -filter:retweets",
]

queries_sol = [
    "(pump.fun OR raydium) AND (just launched OR new token OR stealth) -filter:retweets",
    "solana AND CA AND (launch OR deployed OR live) -filter:retweets",
    "SOL AND contract AND (dexscreener OR birdeye OR jupiter) -filter:retweets",
]

queries_base = [
    "base chain AND (just launched OR new token OR stealth launch) -filter:retweets",
    "0x AND (base OR basescan) AND (launch OR deployed) -filter:retweets",
    "base AND (aerodrome OR uniswap) AND new token -filter:retweets",
]

import requests
import time
import re
import json
from datetime import datetime, timedelta

XCROP_API_KEY = "xc_live_your_api_key_here"
BASE_URL = "https://xcrop.io/api/v2"
HEADERS = {
    "Authorization": f"Bearer {XCROP_API_KEY}",
    "Content-Type": "application/json"
}

# Patterns to detect contract addresses
CA_PATTERNS = {
    "ETH": r"0x[a-fA-F0-9]{40}",
    "SOL": r"[1-9A-HJ-NP-Za-km-z]{32,44}",
}

# Search queries for new launches
SEARCH_QUERIES = [
    "just launched AND (contract OR CA OR 0x) -filter:retweets",
    "stealth launch AND (token OR coin OR dex) -filter:retweets",
    "fair launch AND (uniswap OR raydium OR pump.fun) -filter:retweets",
    "new token AND (liquidity added OR LP locked) -filter:retweets",
    "deploy AND (contract address OR CA) AND (sol OR eth OR base) -filter:retweets",
]


def search_tweets(query, count=20):
    """Search Twitter via XCROP API."""
    response = requests.post(
        f"{BASE_URL}/search",
        headers=HEADERS,
        json={"query": query, "count": count}
    )
    if response.status_code != 200:
        print(f"Search error: {response.status_code}")
        return []
    return response.json().get("data", [])


def extract_addresses(text):
    """Extract potential contract addresses from tweet text."""
    found = {}
    for chain, pattern in CA_PATTERNS.items():
        matches = re.findall(pattern, text)
        if matches:
            found[chain] = matches
    return found


def calculate_engagement_score(tweet):
    """Score a tweet based on engagement velocity."""
    likes = tweet.get("likes", 0)
    retweets = tweet.get("retweets", 0)
    replies = tweet.get("replies", 0)
    views = tweet.get("views", 1)

    # High engagement relative to views = organic interest
    engagement_rate = (likes + retweets * 2 + replies * 3) / max(views, 1)

    # Bonus for quote tweets (people discussing it)
    quote_bonus = tweet.get("quotes", 0) * 5

    return round((engagement_rate * 10000) + quote_bonus, 2)


def check_trending_boost():
    """Check if any crypto topics are trending."""
    response = requests.get(
        f"{BASE_URL}/trending",
        headers={"Authorization": f"Bearer {XCROP_API_KEY}"}
    )
    if response.status_code != 200:
        return []

    topics = response.json().get("data", [])
    crypto_keywords = ["token", "coin", "launch", "airdrop", "dex", "nft",
                       "defi", "sol", "eth", "base", "pump"]

    return [
        t for t in topics
        if any(kw in t.get("name", "").lower() for kw in crypto_keywords)
    ]


def analyze_author(username):
    """Quick check on tweet author — new accounts shilling = red flag."""
    response = requests.get(
        f"{BASE_URL}/users/{username}",
        headers={"Authorization": f"Bearer {XCROP_API_KEY}"}
    )
    if response.status_code != 200:
        return None

    user = response.json().get("data", {})
    return {
        "username": username,
        "followers": user.get("followers", 0),
        "following": user.get("following", 0),
        "tweets": user.get("tweets_count", 0),
        "created": user.get("created_at", ""),
        "verified": user.get("verified", False),
    }


def run_scan():
    """Run a single scan cycle."""
    print(f"\n{'='*60}")
    print(f"Scan at {datetime.now().strftime('%H:%M:%S')}")
    print(f"{'='*60}")

    all_signals = []

    # 1. Search for launch tweets
    for query in SEARCH_QUERIES:
        tweets = search_tweets(query, count=20)
        for tweet in tweets:
            addresses = extract_addresses(tweet.get("text", ""))
            if not addresses:
                continue

            score = calculate_engagement_score(tweet)
            signal = {
                "tweet_id": tweet.get("id"),
                "text": tweet.get("text", "")[:200],
                "author": tweet.get("author", {}).get("username", "unknown"),
                "addresses": addresses,
                "engagement_score": score,
                "likes": tweet.get("likes", 0),
                "retweets": tweet.get("retweets", 0),
                "created_at": tweet.get("created_at", ""),
            }
            all_signals.append(signal)

        time.sleep(1)  # Respect rate limits

    # 2. Check trending for crypto surge
    trending_crypto = check_trending_boost()
    if trending_crypto:
        print(f"\nCrypto trending: {[t['name'] for t in trending_crypto[:5]]}")

    # 3. Deduplicate by contract address
    seen_addresses = set()
    unique_signals = []
    for signal in all_signals:
        for chain, addrs in signal["addresses"].items():
            for addr in addrs:
                if addr not in seen_addresses:
                    seen_addresses.add(addr)
                    unique_signals.append(signal)

    # 4. Sort by engagement score
    unique_signals.sort(key=lambda s: s["engagement_score"], reverse=True)

    # 5. Display results
    if not unique_signals:
        print("No new token signals detected.")
        return

    print(f"\nFound {len(unique_signals)} potential launches:\n")
    for i, signal in enumerate(unique_signals[:10], 1):
        print(f"{i}. @{signal['author']} (score: {signal['engagement_score']})")
        print(f"   {signal['text'][:120]}...")
        for chain, addrs in signal["addresses"].items():
            print(f"   {chain}: {addrs[0]}")
        print(f"   Engagement: {signal['likes']} likes, "
              f"{signal['retweets']} RTs")
        print()

    return unique_signals


# ── Main Loop ──────────────────────────────────────────────────

if __name__ == "__main__":
    print("Token Launch Detector — powered by XCROP API")
    print("Scanning every 2 minutes...\n")

    while True:
        try:
            signals = run_scan()

            # Optional: send alerts (Telegram, Discord, etc.)
            # if signals:
            #     send_telegram_alert(signals[0])

        except KeyboardInterrupt:
            print("\nStopped.")
            break
        except Exception as e:
            print(f"Error: {e}")

        time.sleep(120)  # Scan every 2 minutes

# Only surface tweets with meaningful engagement
MIN_LIKES = 5
MIN_RETWEETS = 2

filtered = [
    s for s in signals
    if s["likes"] >= MIN_LIKES or s["retweets"] >= MIN_RETWEETS
]

# Skip brand-new accounts (common for scam launches)
def is_suspicious_account(author_info):
    if not author_info:
        return True

    # Account less than 30 days old
    created = datetime.strptime(author_info["created"], "%Y-%m-%dT%H:%M:%S.000Z")
    age_days = (datetime.now() - created).days
    if age_days < 30:
        return True

    # Very low followers but high tweet count = bot pattern
    if author_info["followers"] < 50 and author_info["tweets"] > 5000:
        return True

    return False

# Track seen contracts across scans to avoid re-alerting
seen_contracts = {}  # address -> first_seen_timestamp

def is_new_contract(address):
    if address in seen_contracts:
        return False
    seen_contracts[address] = datetime.now()
    return True

KOL_WATCHLIST = ["CryptoHayes", "DefiIgnas", "Pentosh1", "blaboratory"]

def check_kol_mentions():
    """Monitor KOL timelines for token mentions."""
    tweets = []
    for kol in KOL_WATCHLIST:
        response = requests.get(
            f"{BASE_URL}/users/{kol}/tweets?count=20",
            headers={"Authorization": f"Bearer {XCROP_API_KEY}"}
        )
        if response.status_code == 200:
            tweets += response.json().get("data", [])
    kol_signals = []

    for tweet in tweets:
        addresses = extract_addresses(tweet.get("text", ""))
        if addresses:
            kol_signals.append({
                "kol": tweet.get("author", {}).get("username"),
                "followers": tweet.get("author", {}).get("followers", 0),
                "addresses": addresses,
                "text": tweet.get("text", "")[:200],
                "tweet_id": tweet.get("id"),
            })

    return kol_signals