anomaly/strange_doge.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 = '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