24/7 Crypto Monitoring: Building Always-On Analysis Systems

VPS (Ubuntu, $5/mo)
  |
  [Python Daemon]
  |-- Every 5 min: check indicators for watchlist
  |-- Compare against alert rules
  |-- If triggered: send notification
  |-- Log everything to file
  |
  v
Telegram Bot / Discord Webhook / Email

from dataclasses import dataclass
from enum import Enum

class Condition(Enum):
    ABOVE = "above"
    BELOW = "below"
    CROSSES_ABOVE = "crosses_above"
    CROSSES_BELOW = "crosses_below"

@dataclass
class AlertRule:
    symbol: str
    indicator: str
    timeframe: str
    condition: Condition
    threshold: float
    message: str
    cooldown_minutes: int = 60  # Don't re-alert within this window

# Example rules
RULES = [
    AlertRule("BTCUSD", "RSI_14", "H4", Condition.ABOVE, 80,
             "BTC H4 RSI overbought: {'{'+'value:.1f'+'}'}"),
    AlertRule("BTCUSD", "RSI_14", "H4", Condition.BELOW, 25,
             "BTC H4 RSI oversold: {'{'+'value:.1f'+'}'}"),
    AlertRule("ETHUSD", "RSI_14", "H4", Condition.BELOW, 30,
             "ETH H4 RSI oversold: {'{'+'value:.1f'+'}'}"),
    AlertRule("BTCUSD", "MACD", "D1", Condition.CROSSES_ABOVE, 0,
             "BTC Daily MACD bullish crossover!"),
    AlertRule("XAUUSD", "RSI_14", "D1", Condition.ABOVE, 75,
             "Gold Daily RSI overbought: {'{'+'value:.1f'+'}'}"),
]

import requests
import time
import logging
from datetime import datetime, timedelta

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[
        logging.FileHandler("monitor.log"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger("crypto_monitor")

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://tickatlas.com/v1"

class CryptoMonitor:
    def __init__(self, rules: list[AlertRule]):
        self.rules = rules
        self.last_alert: dict[str, datetime] = {}
        self.previous_values: dict[str, float] = {}

    def fetch_value(self, symbol: str, indicator: str, tf: str) -> float:
        resp = requests.get(
            f"{BASE_URL}/indicator",
            headers={"X-API-Key": API_KEY},
            params={"symbol": symbol, "indicator": indicator, "timeframe": tf},
        )
        resp.raise_for_status()
        value = resp.json()["data"]["value"]
        return value if isinstance(value, (int, float)) else 0.0

    def check_condition(self, rule: AlertRule, value: float) -> bool:
        key = f"{rule.symbol}:{rule.indicator}:{rule.timeframe}"
        prev = self.previous_values.get(key)
        self.previous_values[key] = value

        if rule.condition == Condition.ABOVE:
            return value > rule.threshold
        elif rule.condition == Condition.BELOW:
            return value < rule.threshold
        elif rule.condition == Condition.CROSSES_ABOVE:
            return prev is not None and prev <= rule.threshold < value
        elif rule.condition == Condition.CROSSES_BELOW:
            return prev is not None and prev >= rule.threshold > value
        return False

    def is_cooled_down(self, rule: AlertRule) -> bool:
        key = f"{rule.symbol}:{rule.indicator}:{rule.condition.value}"
        last = self.last_alert.get(key)
        if last is None:
            return True
        return datetime.utcnow() - last > timedelta(minutes=rule.cooldown_minutes)

    def run_cycle(self) -> list[str]:
        triggered = []
        for rule in self.rules:
            try:
                value = self.fetch_value(rule.symbol, rule.indicator, rule.timeframe)
                if self.check_condition(rule, value) and self.is_cooled_down(rule):
                    msg = rule.message.format(value=value)
                    triggered.append(msg)
                    key = f"{rule.symbol}:{rule.indicator}:{rule.condition.value}"
                    self.last_alert[key] = datetime.utcnow()
                    logger.info(f"ALERT: {msg}")
            except Exception as e:
                logger.error(f"Error checking {rule.symbol} {rule.indicator}: {e}")
        return triggered

TELEGRAM_TOKEN = "YOUR_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"

def send_telegram(message: str):
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
    requests.post(url, json={
        "chat_id": TELEGRAM_CHAT_ID,
        "text": f"🔔 {message}",
        "parse_mode": "HTML",
    })

# Main loop
monitor = CryptoMonitor(RULES)
logger.info("Crypto monitor started. Checking every 5 minutes.")

while True:
    alerts = monitor.run_cycle()
    for alert in alerts:
        send_telegram(alert)
    time.sleep(300)  # 5 minutes

# On your VPS (Ubuntu)
sudo apt install python3-pip
pip3 install requests

# Create a systemd service
sudo tee /etc/systemd/system/crypto-monitor.service > /dev/null << 'EOF'
[Unit]
Description=Crypto Market Monitor
After=network.target

[Service]
ExecStart=/usr/bin/python3 /opt/monitor/crypto_monitor.py
WorkingDirectory=/opt/monitor
Restart=always
RestartSec=30
User=monitor

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl enable crypto-monitor
sudo systemctl start crypto-monitor
sudo systemctl status crypto-monitor