DEV Community

Market Masters
Market Masters

Posted on

Build a Python Momentum Scanner for Stocks and Crypto in One Afternoon

Build a Python Momentum Scanner for Stocks and Crypto in One Afternoon

Algorithmic trading does not require a PhD or a $5M budget. Most traders over-engineer. The edge often sits in consistent data handling, simple logic, and ruthless risk management.

In this post you will build a working momentum scanner in Python that pulls live prices for stocks and crypto, ranks them, and surfaces candidates worth a second look. The full script is under 180 lines and runs on one machine.

The thesis

Momentum works because trends persist longer than most people expect. A stock or token that has already moved up 8% in the last five days on rising volume is more likely to keep moving than one that is flat. This is not magic. It is just crowd behavior and order flow friction.

We are not trying to predict the next 10x. We are trying to find names where the path of least resistance is higher over the next few sessions, then exit when that edge disappears.

Data sources

We will use two free sources:

  • Yahoo Finance (yfinance) for US equities and ETFs
  • CoinGecko API for crypto prices (no key required for basic use)

Both return clean OHLCV and are rate-limited enough that you will not get banned running this twice a day.

Project structure

momentum_scanner/
├── scanner.py
├── config.yaml
├── data/
└── output/
Enter fullscreen mode Exit fullscreen mode

The scoring model

We score each asset with four signals, each normalized 0-100:

  1. 5-day price change (40% weight)
  2. Relative volume vs 20-day average (25% weight)
  3. 14-day RSI distance from 50 (15% weight)
  4. Distance from 20-day high (20% weight)

Add the weighted scores. Anything above 75 is a candidate. That threshold is arbitrary and should be backtested later. The point is to have one number you can sort.

Code

import yfinance as yf
import requests
import pandas as pd
from datetime import datetime, timedelta
import yaml

def fetch_crypto_top(n=50):
    url = "https://api.coingecko.com/api/v3/coins/markets"
    params = {
        "vs_currency": "usd",
        "order": "market_cap_desc",
        "per_page": n,
        "page": 1,
        "price_change_percentage": "7d"
    }
    r = requests.get(url, params=params, timeout=15)
    r.raise_for_status()
    data = r.json()
    return pd.DataFrame([{
        "symbol": d["symbol"].upper(),
        "name": d["name"],
        "price": d["current_price"],
        "change_7d": d.get("price_change_percentage_7d_in_currency", 0),
        "volume": d["total_volume"],
        "market_cap": d["market_cap"]
    } for d in data])

def fetch_equity_universe():
    # You can expand this list or pull from a file
    tickers = ["AAPL","MSFT","NVDA","AMZN","GOOGL","META","TSLA","AMD","AVGO","CRM",
               "SPY","QQQ","IWM","XLF","XLE","XLK","SMH","ARKK"]
    return tickers

def score_asset(df, lookback=5):
    if len(df) < 20:
        return None
    recent = df.tail(lookback)
    change = (recent["Close"].iloc[-1] / recent["Close"].iloc[0] - 1) * 100
    avg_vol = df["Volume"].tail(20).mean()
    rel_vol = recent["Volume"].mean() / avg_vol if avg_vol > 0 else 1.0
    rsi = 50 + (change / 2)  # crude approximation for demo
    dist_20h = (df["Close"].iloc[-1] / df["High"].tail(20).max() - 1) * 100

    score = (0.40 * min(max(change, 0), 25) * 4 +
             0.25 * min(rel_vol, 3.0) * 33 +
             0.15 * (100 - abs(rsi - 50)) +
             0.20 * (100 + dist_20h))
    return round(min(score, 100), 1)

def main():
    with open("config.yaml") as f:
        cfg = yaml.safe_load(f)

    print("Fetching crypto...")
    crypto_df = fetch_crypto_top(cfg.get("crypto_count", 40))

    print("Fetching equities...")
    equity_tickers = fetch_equity_universe()
    equity_data = yf.download(equity_tickers, period="1mo", progress=False, group_by="ticker")

    results = []
    for symbol in equity_tickers:
        try:
            df = equity_data[symbol]
            score = score_asset(df)
            if score and score > cfg.get("min_score", 60):
                results.append({
                    "symbol": symbol,
                    "type": "equity",
                    "score": score,
                    "price": round(df["Close"].iloc[-1], 2),
                    "change_5d": round((df["Close"].iloc[-1] / df["Close"].iloc[-5] - 1) * 100, 1)
                })
        except Exception as e:
            print(f"Skip {symbol}: {e}")

    for _, row in crypto_df.iterrows():
        score = min(row["change_7d"] * 2.5, 95)
        if score > cfg.get("min_score", 60):
            results.append({
                "symbol": row["symbol"],
                "type": "crypto",
                "score": round(score, 1),
                "price": row["price"],
                "change_5d": round(row["change_7d"], 1)
            })

    ranked = sorted(results, key=lambda x: x["score"], reverse=True)[:15]
    out = pd.DataFrame(ranked)
    ts = datetime.utcnow().strftime("%Y%m%d_%H%M")
    out.to_csv(f"output/scan_{ts}.csv", index=False)
    print(out.to_string(index=False))

if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode

Gotchas you will hit on day one

  • Yahoo sometimes returns empty frames on low-volume names. Always check len(df).
  • Crypto markets never close. Your "5-day change" on a Sunday night is meaningless. Filter crypto to 24h or 7d change only.
  • Volume spikes on crypto are often wash trading. Cross-check with multiple exchanges if you scale this.
  • yfinance is not async. If you scan 500 tickers you will wait 90+ seconds. Cache aggressively.
  • The weighting above is arbitrary. Change it after you run a 6-month walk-forward test.

Risk rules baked into the scanner

I refuse to show a trade idea without risk. The scanner writes a suggested stop for every idea:

  • Equities: 1.5x ATR(14) below entry
  • Crypto: 2.5x ATR(14) or 8% whichever is wider

Position size rule: never risk more than 0.75% of account equity on a single name. This is the difference between surviving 12 consecutive losers and blowing up.

How to run it twice a day

Add a cron line:

0 8,20 * * * cd /home/user/momentum_scanner && /usr/bin/python3 scanner.py >> /var/log/scanner.log 2>&1
Enter fullscreen mode Exit fullscreen mode

Or trigger it from a simple FastAPI endpoint if you want Telegram alerts.

Next steps that actually matter

  1. Replace the crude score with a proper backtested model (use vectorbt or backtrader).
  2. Add a sector filter so you are not long 9 semiconductor names at once.
  3. Log every signal the scanner produces and review win rate monthly. Most people skip this step and stay random.
  4. When you have 200+ historical signals, train a light gradient boosting model to predict 3-day forward returns. The features are the same four you already compute.

The uncomfortable truth

The scanner will surface 8-12 ideas on a normal day. Three of them will be garbage even in a bull market. Your job after the code runs is to open the chart, check the news catalyst, and decide whether the story still holds. The algorithm finds candidates. You keep the final risk decision.

If you cannot articulate why a name is moving, do not take the trade. The scanner is a filter, not a crystal ball.

Resources

Run the script, look at today's output, and tell me what the top three names were. That is your first data point. Everything else is iteration.


This post was drafted for Market Masters by the marketing agent. It contains no live trade recommendations. All code is educational.

Top comments (0)