### 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 = 'dogecoin' # CoinGecko ID for DOGE CURRENCY = 'usd' # Base currency API_URL = f'https://api.coingecko.com/api/v3/simple/price?ids={CRYPTO}&vs_currencies={CURRENCY}' CHECK_INTERVAL_SECONDS = 60 # Check every 60 seconds Z_SCORE_THRESHOLD = 3.0 # Flag as anomaly if |Z-score| > 3 HISTORY_WINDOW = 100 # Keep last 100 prices for mean/std calculation # Email configuration (replace with your details) 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 price_history = [] def fetch_price(): """Fetch current price from CoinGecko API.""" try: response = requests.get(API_URL) response.raise_for_status() data = response.json() price = data[CRYPTO][CURRENCY] return price except Exception as e: print(f"Error fetching price: {e}") return None def detect_anomaly(current_price): """Detect if the current price is an anomaly based on Z-score.""" if len(price_history) < 2: # Need at least 2 points for std dev return False, 0.0 # Calculate mean and std dev from history df = pd.DataFrame(price_history, columns=['price']) mean = df['price'].mean() std_dev = df['price'].std() if std_dev == 0: # Avoid division by zero 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(current_price, z_score): """Send email alert for anomaly.""" subject = f"Anomaly Detected in {CRYPTO.upper()} Price!" body = f""" Alert Time: {datetime.now()} 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("Email alert sent successfully.") except Exception as e: print(f"Error sending email: {e}") def monitor(): """Main monitoring function: Fetch price, check for anomaly, alert if needed.""" current_price = fetch_price() if current_price is None: return # Append to history (maintain fixed window size) price_history.append(current_price) if len(price_history) > HISTORY_WINDOW: price_history.pop(0) is_anomaly, z_score = detect_anomaly(current_price) print(f"{datetime.now()} - Current {CRYPTO.upper()} Price: ${current_price:.4f} | Z-Score: {z_score:.2f}") if is_anomaly: print(f"ANOMALY DETECTED! Z-Score: {z_score:.2f}") send_email_alert(current_price, z_score) # Schedule the monitor to run every CHECK_INTERVAL_SECONDS schedule.every(CHECK_INTERVAL_SECONDS).seconds.do(monitor) # Run the monitoring loop print(f"Starting {CRYPTO.upper()} price anomaly monitor...") while True: schedule.run_pending() time.sleep(1) # Sleep to avoid high CPU usage