April 4, 2026 19 min read Tutorial Python

Automating Trade Alerts with Python and Flask

A step-by-step guide to building a trade alert system that monitors market signals on a schedule and sends notifications via email and Telegram. We will use Flask for the web framework, APScheduler for scheduling, yfinance for market data, and the SEC EDGAR API for insider filings.

1. Architecture Overview

The system has four components:

Flask web application — serves a dashboard for viewing alerts and system status, and exposes an API for configuration.
APScheduler background scheduler — runs signal checks on a configurable interval (e.g., every 15 minutes during market hours) without blocking the web server.
Signal checking engine — fetches market data, evaluates conditions (technical indicators, insider activity), and generates alerts when conditions are met.
Notification layer — sends alerts via email (SMTP) and/or Telegram, with deduplication to prevent alert fatigue.

This architecture is simple enough to run on a single machine (a VPS, a Raspberry Pi, or a cloud service like Railway) while being cleanly separated so that each component can be tested and modified independently.

2. Project Setup

Create a project directory and install the dependencies:

mkdir trade-alerts && cd trade-alerts
python -m venv venv
source venv/bin/activate        # macOS/Linux
# venv\Scripts\activate         # Windows

pip install flask apscheduler yfinance requests

The core dependencies are:

flask — lightweight web framework
apscheduler — the Advanced Python Scheduler, specifically BackgroundScheduler which runs jobs in background threads
yfinance — Yahoo Finance API wrapper for fetching OHLCV data
requests — HTTP client for SEC EDGAR and Telegram API calls

Create a .env file for configuration (never commit this to version control):

# .env
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
[email protected]
SMTP_PASSWORD=your-app-password
[email protected]

TELEGRAM_BOT_TOKEN=your-bot-token
TELEGRAM_CHAT_ID=your-chat-id

SEC_USER_AGENT=YourName [email protected]
WATCHLIST=AAPL,MSFT,NVDA,GOOGL,AMZN

Gmail App Passwords

Gmail requires an “App Password” for SMTP access when two-factor authentication is enabled. Go to your Google Account → Security → 2-Step Verification → App passwords, and generate a 16-character password specifically for this application. Use that password in SMTP_PASSWORD, not your regular Gmail password.

3. The Flask App with APScheduler

The foundation is a Flask app with a BackgroundScheduler that starts when the application starts:

import os
import json
import logging
from datetime import datetime, timedelta
from flask import Flask, jsonify, render_template_string
from apscheduler.schedulers.background import BackgroundScheduler

app = Flask(__name__)
scheduler = BackgroundScheduler()

# In-memory store for recent alerts
alert_history = []

def check_signals():
    """Main signal checking function — called by the scheduler."""
    logging.info("Running signal check at %s", datetime.now())
    watchlist = os.environ.get("WATCHLIST", "").split(",")
    for symbol in watchlist:
        symbol = symbol.strip()
        if not symbol:
            continue
        try:
            alerts = evaluate_symbol(symbol)
            for alert in alerts:
                if not is_duplicate(alert):
                    send_alert(alert)
                    alert_history.append(alert)
        except Exception as e:
            logging.error("Error checking %s: %s", symbol, e)

# Schedule the job: every 15 minutes
scheduler.add_job(
    check_signals,
    "interval",
    minutes=15,
    id="signal_check",
    max_instances=1,
    coalesce=True,
)
scheduler.start()

@app.route("/")
def dashboard():
    return render_template_string(DASHBOARD_HTML, alerts=alert_history[-50:])

@app.route("/api/alerts")
def api_alerts():
    return jsonify(alert_history[-100:])

@app.route("/health")
def health():
    return jsonify({"status": "ok", "scheduler_running": scheduler.running})

A few important details about BackgroundScheduler:

max_instances=1 ensures that if a signal check takes longer than 15 minutes, a second instance will not start concurrently. This prevents duplicate alerts and excessive API usage.
coalesce=True means that if the scheduler missed a run (e.g., the machine was asleep), it will only fire once when it wakes up rather than catching up with all missed runs.
The scheduler runs in a background thread, so it does not block Flask from serving HTTP requests.

4. Signal Checking: Technical Indicators

The evaluate_symbol function fetches market data and checks conditions. We use yfinance to fetch 6 months of daily OHLCV data and compute technical indicators directly:

import yfinance as yf
import numpy as np

def evaluate_symbol(symbol):
    """Check a symbol for alert conditions. Returns a list of alert dicts."""
    alerts = []
    ticker = yf.Ticker(symbol)
    df = ticker.history(period="6mo")

    if df.empty or len(df) < 50:
        return alerts

    close = df["Close"]
    volume = df["Volume"]

    # 200-day moving average (use available data)
    ma200 = close.rolling(window=min(200, len(close))).mean()

    # RSI-14
    delta = close.diff()
    gain = delta.where(delta > 0, 0.0).rolling(14).mean()
    loss = (-delta.where(delta < 0, 0.0)).rolling(14).mean()
    rs = gain / loss.replace(0, np.nan)
    rsi = 100 - (100 / (1 + rs))

    # Volume ratio: current vs 20-day average
    avg_volume = volume.rolling(20).mean()
    volume_ratio = volume / avg_volume.replace(0, np.nan)

    latest = df.index[-1]
    current_close = close.iloc[-1]
    current_rsi = rsi.iloc[-1]
    current_vol_ratio = volume_ratio.iloc[-1]
    current_ma200 = ma200.iloc[-1]

    # Condition 1: RSI oversold (below 30)
    if current_rsi < 30:
        alerts.append({
            "symbol": symbol,
            "type": "RSI_OVERSOLD",
            "message": f"{symbol}: RSI at {current_rsi:.1f} (oversold)",
            "timestamp": datetime.now().isoformat(),
            "price": round(current_close, 2),
        })

    # Condition 2: Price crosses above 200 MA
    if len(close) >= 2:
        prev_close = close.iloc[-2]
        prev_ma200 = ma200.iloc[-2]
        if prev_close < prev_ma200 and current_close > current_ma200:
            alerts.append({
                "symbol": symbol,
                "type": "MA200_CROSSOVER",
                "message": f"{symbol}: Price crossed above 200 MA "
                           f"(${current_close:.2f} > ${current_ma200:.2f})",
                "timestamp": datetime.now().isoformat(),
                "price": round(current_close, 2),
            })

    # Condition 3: Unusual volume (> 2x 20-day average)
    if current_vol_ratio > 2.0:
        alerts.append({
            "symbol": symbol,
            "type": "UNUSUAL_VOLUME",
            "message": f"{symbol}: Volume {current_vol_ratio:.1f}x "
                       f"above 20-day average",
            "timestamp": datetime.now().isoformat(),
            "price": round(current_close, 2),
        })

    return alerts

These three conditions — RSI oversold, moving average crossover, and unusual volume — are simple but effective starting points. You can add more conditions as needed: Bollinger Band touches, MACD crossovers, ATR breakouts, and so on. The key is that each condition is a clear, testable rule that generates a structured alert object.

5. Insider Signal Monitoring

The SEC provides a public API for accessing EDGAR filings. Form 4 filings disclose insider transactions (purchases and sales by officers, directors, and 10%+ shareholders). We can monitor for new insider purchases — particularly cluster buying (multiple insiders buying within a short window), which academic research has shown to be a statistically significant positive signal.

import requests
import time

SEC_HEADERS = {
    "User-Agent": os.environ.get("SEC_USER_AGENT", "AlertBot [email protected]"),
    "Accept-Encoding": "gzip, deflate",
}

def check_insider_activity(symbol):
    """Check SEC EDGAR for recent insider purchases."""
    alerts = []

    # Get company CIK from ticker
    url = f"https://efts.sec.gov/LATEST/search-index?q=%22{symbol}%22&dateRange=custom&startdt={start_date}&enddt={end_date}&forms=4"

    # A simpler approach: use the company tickers JSON
    tickers_url = "https://www.sec.gov/files/company_tickers.json"
    resp = requests.get(tickers_url, headers=SEC_HEADERS)
    time.sleep(0.2)  # SEC rate limit: max 10 requests per second

    if resp.status_code != 200:
        return alerts

    tickers = resp.json()
    cik = None
    for entry in tickers.values():
        if entry.get("ticker", "").upper() == symbol.upper():
            cik = str(entry["cik_str"]).zfill(10)
            break

    if not cik:
        return alerts

    # Fetch recent filings for this CIK
    filings_url = (
        f"https://data.sec.gov/submissions/CIK{cik}.json"
    )
    resp = requests.get(filings_url, headers=SEC_HEADERS)
    time.sleep(0.2)

    if resp.status_code != 200:
        return alerts

    data = resp.json()
    recent = data.get("filings", {}).get("recent", {})
    forms = recent.get("form", [])
    dates = recent.get("filingDate", [])

    # Count Form 4 filings in the last 7 days
    cutoff = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
    recent_form4_count = sum(
        1 for f, d in zip(forms, dates)
        if f == "4" and d >= cutoff
    )

    if recent_form4_count >= 3:
        alerts.append({
            "symbol": symbol,
            "type": "INSIDER_CLUSTER",
            "message": f"{symbol}: {recent_form4_count} Form 4 filings "
                       f"in the last 7 days (cluster buying signal)",
            "timestamp": datetime.now().isoformat(),
            "price": None,
        })

    return alerts

SEC EDGAR Rate Limits

The SEC requires a valid User-Agent header with your name and email, and limits requests to 10 per second. Always include a time.sleep(0.2) between requests to stay within limits. Violating rate limits may result in your IP being temporarily blocked.

Note that counting Form 4 filings is a simplified approach. A more thorough implementation would parse the XML of each Form 4 to determine whether the transactions are purchases (transaction code “P”) or sales (transaction code “S”), the dollar amounts, and the insider’s role (CEO purchases carry more signal weight than a director buying a token amount). Alpha Suite performs this full parsing in its signal generation pipeline.

6. Sending Alerts via Email

Python’s standard library includes smtplib and email.mime for sending emails. Here is a function that sends an alert via Gmail’s SMTP server:

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_email_alert(alert):
    """Send an alert via email using SMTP."""
    smtp_host = os.environ.get("SMTP_HOST", "smtp.gmail.com")
    smtp_port = int(os.environ.get("SMTP_PORT", 587))
    smtp_user = os.environ.get("SMTP_USER")
    smtp_pass = os.environ.get("SMTP_PASSWORD")
    recipient = os.environ.get("ALERT_EMAIL")

    if not all([smtp_user, smtp_pass, recipient]):
        logging.warning("Email not configured, skipping")
        return

    subject = f"Trade Alert: {alert['symbol']} - {alert['type']}"

    body = f"""
    Symbol: {alert['symbol']}
    Type: {alert['type']}
    Time: {alert['timestamp']}
    Price: {alert.get('price', 'N/A')}

    {alert['message']}
    """

    msg = MIMEMultipart()
    msg["From"] = smtp_user
    msg["To"] = recipient
    msg["Subject"] = subject
    msg.attach(MIMEText(body, "plain"))

    try:
        with smtplib.SMTP(smtp_host, smtp_port) as server:
            server.starttls()
            server.login(smtp_user, smtp_pass)
            server.sendmail(smtp_user, recipient, msg.as_string())
        logging.info("Email sent for %s", alert["symbol"])
    except Exception as e:
        logging.error("Failed to send email: %s", e)

The starttls() call upgrades the connection to TLS encryption. This is required by Gmail and most modern SMTP servers. Port 587 is the standard port for SMTP with STARTTLS. Port 465 uses implicit SSL (use smtplib.SMTP_SSL instead of smtplib.SMTP with starttls() for that port).

7. Sending Alerts via Telegram

Telegram is often a better choice than email for trade alerts because notifications appear instantly on your phone. Setting up a Telegram bot is straightforward:

Open Telegram and search for @BotFather.
Send /newbot and follow the prompts to name your bot.
BotFather will give you an API token (a string like 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11).
Start a conversation with your new bot (send it any message).
Get your chat ID by visiting https://api.telegram.org/bot{TOKEN}/getUpdates in a browser and looking for the chat.id field in the response.

The sending function is simple:

def send_telegram_alert(alert):
    """Send an alert via Telegram Bot API."""
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    chat_id = os.environ.get("TELEGRAM_CHAT_ID")

    if not token or not chat_id:
        logging.warning("Telegram not configured, skipping")
        return

    text = (
        f"*{alert['type']}* | {alert['symbol']}\n"
        f"Price: ${alert.get('price', 'N/A')}\n"
        f"{alert['message']}\n"
        f"_{alert['timestamp']}_"
    )

    url = f"https://api.telegram.org/bot{token}/sendMessage"
    payload = {
        "chat_id": chat_id,
        "text": text,
        "parse_mode": "Markdown",
    }

    try:
        resp = requests.post(url, json=payload, timeout=10)
        if resp.status_code == 200:
            logging.info("Telegram alert sent for %s", alert["symbol"])
        else:
            logging.error("Telegram API error: %s", resp.text)
    except Exception as e:
        logging.error("Failed to send Telegram alert: %s", e)

The Telegram Bot API is a simple HTTP POST. The parse_mode: "Markdown" parameter enables basic formatting: *bold* and _italic_. For more complex formatting, use "HTML" parse mode instead.

Unified Send Function

def send_alert(alert):
    """Send alert via all configured channels."""
    send_email_alert(alert)
    send_telegram_alert(alert)

8. Alert Deduplication

Without deduplication, you will receive the same alert every 15 minutes as long as the condition remains true (e.g., RSI stays below 30). This is the fastest way to make alerts useless — alert fatigue causes you to ignore all notifications, including the important ones.

A simple deduplication strategy uses a cooldown period per symbol per alert type:

# Track when we last alerted on each (symbol, type) pair
alert_cooldowns = {}
COOLDOWN_HOURS = 24

def is_duplicate(alert):
    """Check if we've already sent this alert recently."""
    key = (alert["symbol"], alert["type"])
    last_sent = alert_cooldowns.get(key)

    if last_sent:
        elapsed = datetime.now() - last_sent
        if elapsed < timedelta(hours=COOLDOWN_HOURS):
            logging.debug(
                "Suppressing duplicate alert for %s/%s "
                "(last sent %s ago)",
                alert["symbol"], alert["type"], elapsed,
            )
            return True

    alert_cooldowns[key] = datetime.now()
    return False

With a 24-hour cooldown, you will receive at most one alert per symbol per condition per day. You can make this more sophisticated: use a shorter cooldown for high-urgency alerts (insider cluster buying) and a longer cooldown for routine signals (RSI oversold in a trending market).

9. Persistent Alert Storage

The in-memory alert_history list works for development, but alerts are lost when the process restarts. For production, persist alerts to a file or database. A JSON file is the simplest approach:

import json
from pathlib import Path

ALERT_FILE = Path("alert_history.json")

def save_alerts():
    """Persist alert history to disk."""
    with open(ALERT_FILE, "w") as f:
        json.dump(alert_history[-500:], f, indent=2)

def load_alerts():
    """Load alert history from disk."""
    global alert_history
    if ALERT_FILE.exists():
        with open(ALERT_FILE, "r") as f:
            alert_history = json.load(f)

# Call load_alerts() at startup
load_alerts()

For a more robust solution, use SQLite. Python includes SQLite support in the standard library (sqlite3 module), so there is no additional dependency. A single SQLite file can handle millions of alert records with indexed queries.

import sqlite3

DB_PATH = "alerts.db"

def init_db():
    """Create the alerts table if it doesn't exist."""
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS alerts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            symbol TEXT NOT NULL,
            type TEXT NOT NULL,
            message TEXT,
            price REAL,
            timestamp TEXT NOT NULL,
            sent_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.execute("""
        CREATE INDEX IF NOT EXISTS idx_alerts_symbol_type
        ON alerts (symbol, type, sent_at)
    """)
    conn.commit()
    conn.close()

def store_alert(alert):
    """Save an alert to the database."""
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "INSERT INTO alerts (symbol, type, message, price, timestamp) "
        "VALUES (?, ?, ?, ?, ?)",
        (alert["symbol"], alert["type"], alert["message"],
         alert.get("price"), alert["timestamp"]),
    )
    conn.commit()
    conn.close()

10. Web Dashboard

A simple dashboard lets you check system status, view recent alerts, and see the active watchlist without checking your email or Telegram. Here is a minimal template served by Flask:

DASHBOARD_HTML = """
<!DOCTYPE html>
<html>
<head>
    <title>Trade Alerts Dashboard</title>
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <style>
        body { font-family: monospace; background: #0a0b0e;
               color: #c4c8ce; padding: 2rem; }
        h1 { color: #00d4aa; }
        table { border-collapse: collapse; width: 100%; margin-top: 1rem; }
        th, td { padding: 0.5rem 1rem; text-align: left;
                 border-bottom: 1px solid #2a2e35; }
        th { color: #e8eaed; }
        .badge { padding: 0.2rem 0.5rem; border-radius: 4px;
                 font-size: 0.8rem; font-weight: bold; }
        .RSI_OVERSOLD { background: rgba(255,71,87,0.2); color: #ff4757; }
        .MA200_CROSSOVER { background: rgba(0,212,170,0.2); color: #00d4aa; }
        .UNUSUAL_VOLUME { background: rgba(255,179,71,0.2); color: #ffb347; }
        .INSIDER_CLUSTER { background: rgba(79,172,254,0.2); color: #4facfe; }
    </style>
</head>
<body>
    <h1>Trade Alerts</h1>
    <table>
        <tr><th>Time</th><th>Symbol</th><th>Type</th>
            <th>Price</th><th>Message</th></tr>
        {% for a in alerts | reverse %}
        <tr>
            <td>{{ a.timestamp[:16] }}</td>
            <td><strong>{{ a.symbol }}</strong></td>
            <td><span class="badge {{ a.type }}">{{ a.type }}</span></td>
            <td>{{ a.price or '-' }}</td>
            <td>{{ a.message }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>
"""

This dashboard uses Jinja2 templating (built into Flask) to render the alert history as an HTML table. The CSS classes on the badge elements color-code each alert type for quick visual scanning.

11. Putting It All Together

The complete app.py combines all the components. Here is the main entry point:

if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )

    # Load environment variables
    from dotenv import load_dotenv
    load_dotenv()

    # Initialize database
    init_db()

    # Load existing alerts
    load_alerts()

    # Start Flask (scheduler is already running)
    app.run(host="0.0.0.0", port=5000, debug=False)

Important

Set debug=False in production. Flask’s debug mode uses a reloader that restarts the process on code changes, which will cause APScheduler to start duplicate instances. If you need debug mode during development, set use_reloader=False explicitly: app.run(debug=True, use_reloader=False).

12. Deployment Options

The alert system needs to run continuously to be useful. Here are practical deployment options:

VPS (Virtual Private Server)

A small VPS (1 CPU, 1 GB RAM) from providers like DigitalOcean, Linode, or Hetzner is sufficient. Use systemd to run the app as a service that starts on boot and restarts on failure:

# /etc/systemd/system/trade-alerts.service
[Unit]
Description=Trade Alert System
After=network.target

[Service]
User=deploy
WorkingDirectory=/opt/trade-alerts
ExecStart=/opt/trade-alerts/venv/bin/python app.py
Restart=always
RestartSec=10
EnvironmentFile=/opt/trade-alerts/.env

[Install]
WantedBy=multi-user.target

Railway

Railway is a cloud platform that can deploy directly from a Git repository. Create a Procfile in your project root:

web: python app.py

Set your environment variables in the Railway dashboard and deploy. Railway will keep the process running and restart it if it crashes. This is how Alpha Suite deploys its own signal engine.

Raspberry Pi

A Raspberry Pi (any model with network access) is an excellent choice for a personal alert system. It runs 24/7 with minimal power consumption (~5 watts). Install the project in the same way as a VPS and use systemd to manage the process. The Pi 4 with 2 GB RAM is more than sufficient for this workload.

13. Extensions and Next Steps

The basic system described here is a foundation. Here are natural extensions:

Market hours awareness. Only run signal checks during U.S. market hours (9:30 AM to 4:00 PM ET on weekdays, excluding holidays). Use the cron trigger in APScheduler instead of interval for precise scheduling.
Configurable watchlist via the web dashboard. Add Flask routes to add/remove symbols from the watchlist without editing the .env file.
Signal strength scoring. Instead of binary alerts (condition met or not), compute a signal strength score that combines multiple indicators. Alert only when the composite score exceeds a threshold.
Backtesting integration. Log all signals and their outcomes (what happened to the price after the alert). This lets you evaluate which conditions actually lead to profitable trades and tune your thresholds.
Multi-channel routing. Send high-urgency alerts (insider cluster buying) to both email and Telegram, but routine alerts (RSI oversold) only to the dashboard.

Building your own alert system gives you complete control over what signals you monitor, how they are evaluated, and how you are notified. It is one of the most practical applications of Python in trading — a project that provides daily value from the first day it runs.