Python Trading Bot: Real-Time RSI Alerts Using Market Masters API

Traders waste hours glued to charts. A five-minute Python script fixes that. This bot pulls live BTC prices from the Market Masters API, runs RSI, and pings your Telegram when it drops below 30. Oversold signal. Full code, no frameworks.

Tested on free tier. Handles crypto volatility where RSI shines: quick reversals off extremes.

Setup

pip install requests pandas pandas_ta python-telegram-bot

Three keys needed:

Market Masters API key: Sign up at marketmasters.ai. Free tier gives live prices for 2500+ cryptos. Premium unlocks AI signals, Elliott Waves ($39.99/mo, 30-day trial).
Telegram bot token: Message @BotFather.
Your Telegram chat ID: Forward a message to @userinfobot.

Core Script

Save as rsi_bot.py:

import requests
import pandas as pd
import pandas_ta as ta
import telegram
import time
import logging
from datetime import datetime

# Config
API_KEY = 'your_marketmasters_api_key_here'
TELEGRAM_TOKEN = 'your_telegram_bot_token'
CHAT_ID = 'your_chat_id'  # int or str
SYMBOL = 'BTCUSDT'  # or ETHUSDT, etc.
RSI_PERIOD = 14
OVERSOLD = 30
OVERBOUGHT = 70
CHECK_INTERVAL = 60  # seconds

logging.basicConfig(level=logging.INFO)
bot = telegram.Bot(token=TELEGRAM_TOKEN)

def fetch_ohlcv(symbol, limit=100):
    """Fetch last N 1m candles from Market Masters API."""
    url = f"https://api.marketmasters.ai/v1/market/klines?symbol={symbol}&interval=1m&limit={limit}"
    headers = {'Authorization': f'Bearer {API_KEY}'}
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        resp.raise_for_status()
        data = resp.json().get('data', [])
        if not data:
            logging.warning("No data returned")
            return None
        df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
        df['close'] = pd.to_numeric(df['close'])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        return df
    except Exception as e:
        logging.error(f"API error: {e}")
        return None

def calculate_rsi(df, period=RSI_PERIOD):
    """Compute RSI on close prices."""
    rsi = ta.rsi(df['close'], length=period)
    return rsi.iloc[-1]  # Latest value

while True:
    df = fetch_ohlcv(SYMBOL)
    if df is None or len(df) < RSI_PERIOD:
        logging.info("Insufficient data, retrying...")
        time.sleep(CHECK_INTERVAL)
        continue

    rsi = calculate_rsi(df)
    logging.info(f"{SYMBOL} RSI: {rsi:.2f}")

    if rsi < OVERSOLD:
        msg = f"🚨 {SYMBOL} RSI {rsi:.1f} - Oversold. Consider long. Price: ${df['close'].iloc[-1]:.2f}"
        bot.send_message(chat_id=CHAT_ID, text=msg, parse_mode='HTML')
        logging.info("Alert sent")
    elif rsi > OVERBOUGHT:
        msg = f"🔴 {SYMBOL} RSI {rsi:.1f} - Overbought. Consider short. Price: ${df['close'].iloc[-1]:.2f}"
        bot.send_message(chat_id=CHAT_ID, text=msg, parse_mode='HTML')
        logging.info("Alert sent")

    time.sleep(CHECK_INTERVAL)