### 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)