anomaly/strange_cryptos.py

### Python Code

import requests
import pandas as pd
import numpy as np
import time
import schedule
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime

# Configuration
CRYPTO_LIST = ['solana', 'dogecoin', 'ethereum', 'ripple', 'tron']  # CoinGecko IDs
CRYPTO_LABELS = {
    'solana': 'SOL',
    'dogecoin': 'DOGE',
    'ethereum': 'ETH',
    'ripple': 'XRP',
    'tron': 'TRX'
}
CURRENCY = 'usd'
API_URL = 'https://api.coingecko.com/api/v3/simple/price'
CHECK_INTERVAL_SECONDS = 60  # Check every 60 seconds
Z_SCORE_THRESHOLD = 3.0
HISTORY_WINDOW = 100

EMAIL_FROM = 'vortify-lc@algometic.com'
EMAIL_TO = 'larry1chan@qq.com'
EMAIL_PASSWORD = 'g33kPoppy!'
SMTP_SERVER = 'hwsmtp.exmail.qq.com'
SMTP_PORT = 465

# In-memory storage for historical prices per crypto
price_history = {crypto: [] for crypto in CRYPTO_LIST}

def fetch_prices():
    """Fetch current prices for all cryptos from CoinGecko API."""
    try:
        ids = ','.join(CRYPTO_LIST)
        params = {'ids': ids, 'vs_currencies': CURRENCY}
        response = requests.get(API_URL, params=params)
        response.raise_for_status()
        data = response.json()
        return {crypto: data.get(crypto, {}).get(CURRENCY) for crypto in CRYPTO_LIST}
    except Exception as e:
        print(f"Error fetching prices: {e}")
        return {}

def detect_anomaly(crypto, current_price):
    """Detect if the current price is an anomaly based on Z-score."""
    history = price_history[crypto]
    if len(history) < 2:
        return False, 0.0
    df = pd.DataFrame(history, columns=['price'])
    mean = df['price'].mean()
    std_dev = df['price'].std()
    if std_dev == 0:
        return False, 0.0
    z_score = (current_price - mean) / std_dev
    is_anomaly = abs(z_score) > Z_SCORE_THRESHOLD
    return is_anomaly, z_score

def send_email_alert(crypto, current_price, z_score):
    """Send email alert for anomaly."""
    symbol = CRYPTO_LABELS.get(crypto, crypto.upper())
    subject = f"Anomaly Detected in {symbol} Price!"
    body = f"""
    Alert Time: {datetime.now()}
    Crypto: {symbol}
    Current Price: ${current_price:.4f}
    Z-Score: {z_score:.2f} (Threshold: {Z_SCORE_THRESHOLD})
    This indicates a potential anomalous movement (spike or drop).
    """
    msg = MIMEMultipart()
    msg['From'] = EMAIL_FROM
    msg['To'] = EMAIL_TO
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'plain'))
    try:
        server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT)
        server.starttls()
        server.login(EMAIL_FROM, EMAIL_PASSWORD)
        server.sendmail(EMAIL_FROM, EMAIL_TO, msg.as_string())
        server.quit()
        print(f"Email alert sent for {symbol}.")
    except Exception as e:
        e.printstacktrace()
        print(f"Error sending email for {symbol}: {e}")

def monitor():
    """Main monitoring function: Fetch prices, check for anomaly, alert if needed."""
    current_prices = fetch_prices()
    for crypto, price in current_prices.items():
        if price is None:
            continue
        # Append to history (maintain fixed window size)
        price_history[crypto].append(price)
        if len(price_history[crypto]) > HISTORY_WINDOW:
            price_history[crypto].pop(0)
        is_anomaly, z_score = detect_anomaly(crypto, price)
        symbol = CRYPTO_LABELS.get(crypto, crypto.upper())
        print(f"{datetime.now()} - {symbol} Price: ${price:.4f} | Z-Score: {z_score:.2f}")
        if is_anomaly:
            print(f"ANOMALY DETECTED for {symbol}! Z-Score: {z_score:.2f}")
            send_email_alert(crypto, price, z_score)

# Schedule the monitor to run every CHECK_INTERVAL_SECONDS
schedule.every(CHECK_INTERVAL_SECONDS).seconds.do(monitor)

print("Starting multi-crypto price anomaly monitor...")
while True:
    schedule.run_pending()
    time.sleep(1)