Multi-Bagger DNA Screener — automates the sector + filtration steps of a small-cap multi-bagger framework (TradingView + yfinance). Companion to mphinance.substack.com

multibagger_screener.py

#!/usr/bin/env python3
"""
🚀 Multi-Bagger DNA Screener — The "Find It Before The Analysts" Scanner

Companion piece to Forefront Alpha's "How To Find a Multi-Bagger" series.
He sells the framework: be EARLY, before the headlines, before the analysts,
before the crowd. His named winners — IREN at $7, APLD at $6 — were all small,
under-covered, fast-growing names BEFORE Wall Street showed up.

This screener reverse-engineers that DNA into something you can actually run on
the whole US market. It does not parrot his paywalled steps. It asks one
question with cold data: "What trades like IREN/APLD did, RIGHT NOW?"

Architecture mirrors roic_fortress_screener.py (the FUNNEL):
  Stage 1 → TradingView bulk API: fetch the small/mid-cap US universe in ONE call
  Stage 2 → Cheap pre-filter to the speculative-but-not-junk zone
  Stage 3 → yfinance deep scan for analyst coverage, growth, momentum
  Stage 4 → Multi-Bagger DNA score + tiering

The 5 DNA Axes (0-100 each, weighted):
  1. Room To Run        (25%) — small cap = room to 10x (penalize the giants)
  2. Under The Radar    (25%) — few/no analysts = early (Forefront's whole thesis)
  3. Revenue Accel      (20%) — growing INTO the valuation, fast
  4. Early Not Late     (20%) — trend waking up, but hasn't already run +200%
  5. Volume Ignition    (10%) — the crowd is just starting to arrive

Tiers:
  🚀 ROCKET    (80-100) — Textbook pre-run DNA. This is the hunt.
  🛰️  ORBIT     (65-79)  — Strong early profile. Watchlist + DD.
  ✈️  CLIMBING  (50-64)  — Some DNA. Needs a catalyst.
  🛶 DRIFTING  (30-49)  — Weak signal. Probably already known or going nowhere.
  ⚓ ANCHORED  (0-29)   — No multi-bagger DNA. Too big, too covered, or too dead.

Usage:
    python -m dossier.multibagger_screener                          # Full market hunt
    python -m dossier.multibagger_screener --tickers IREN,APLD,INTC # Score specific names
    python -m dossier.multibagger_screener --sector Technology      # Sector filter
    python -m dossier.multibagger_screener --max-cap 5B             # Cap the ceiling
    python -m dossier.multibagger_screener --top 25                 # Top N results
    python -m dossier.multibagger_screener --json                   # Machine output
    python -m dossier.multibagger_screener --csv multibaggers.csv   # Save CSV

© mphinance + Sam the Quant Ghost — "Find it before the analysts do."
"""

import argparse
import json
import sys
import time
from datetime import datetime
from pathlib import Path

try:
    import numpy as np
except ImportError:
    print("❌ pip install numpy")
    sys.exit(1)

try:
    import yfinance as yf
except ImportError:
    print("❌ pip install yfinance")
    sys.exit(1)

try:
    import requests
except ImportError:
    print("❌ pip install requests")
    sys.exit(1)


# ─── Config ───────────────────────────────────────────────────────
PROJECT_ROOT = Path(__file__).resolve().parent.parent

ROCKET_QUOTES = [
    "Small, ignored, and growing. That's the whole game.",
    "By the time the analysts write it up, you want to already own it.",
    "The crowd is the exit, not the entry.",
    "Multi-baggers don't ring a bell. They just quietly stop being small.",
    "Forefront finds these on instinct. We find them on a Tuesday with a script.",
]

ANCHORED_QUOTES = [
    "Nothing with real DNA today. The market's not serving early.",
    "All the small stuff is either junk or already discovered. Patience.",
    "No rockets on the pad. Cash is a position.",
]


# ═══════════════════════════════════════════════════════════════════
# ████  STAGE 1 — TRADINGVIEW BULK UNIVERSE  ████
# ═══════════════════════════════════════════════════════════════════

TV_SCANNER_URL = "https://scanner.tradingview.com/america/scan"

TV_COLUMNS = [
    "name",                     # 0  ticker
    "description",              # 1  company name
    "close",                    # 2  last price
    "change",                   # 3  % change today
    "volume",                   # 4  today's volume
    "average_volume_30d_calc",  # 5  30d avg volume
    "market_cap_basic",         # 6  market cap
    "sector",                   # 7  sector
    "Perf.Y",                   # 8  1-year performance
    "Perf.6M",                  # 9  6-month performance
    "Perf.3M",                  # 10 3-month performance
    "Perf.1M",                  # 11 1-month performance
    "RSI",                      # 12 RSI(14)
    "SMA50",                    # 13 SMA50
    "SMA200",                   # 14 SMA200
    "Recommend.All",            # 15 TV signal
]


def _tv_fetch_universe(min_cap: float = 100_000_000,
                       max_cap: float = 10_000_000_000) -> list[dict]:
    """
    Fetch the small/mid-cap US equity universe from TradingView.

    NOTE: unlike the fortress screener, we deliberately do NOT require positive
    EPS. Early multi-baggers (IREN, APLD pre-run) are frequently pre-profit —
    requiring earnings would filter out exactly the names we want.
    """
    payload = {
        "filter": [
            {"left": "type", "operation": "in_range", "right": ["stock"]},
            {"left": "subtype", "operation": "in_range",
             "right": ["common", "foreign-issuer"]},
            {"left": "exchange", "operation": "in_range",
             "right": ["NYSE", "NASDAQ", "AMEX"]},
            {"left": "average_volume_30d_calc", "operation": "greater", "right": 150_000},
            {"left": "close", "operation": "greater", "right": 2},
            {"left": "market_cap_basic", "operation": "greater", "right": min_cap},
            {"left": "market_cap_basic", "operation": "less", "right": max_cap},
        ],
        "options": {"lang": "en"},
        "symbols": {"query": {"types": []}, "tickers": []},
        "columns": TV_COLUMNS,
        # Sort smallest-first — the hunt lives at the bottom of the cap ladder
        "sort": {"sortBy": "market_cap_basic", "sortOrder": "asc"},
        "range": [0, 5000],
    }

    resp = requests.post(TV_SCANNER_URL, json=payload, timeout=30)
    resp.raise_for_status()
    data = resp.json()
    rows = data.get("data", [])

    results = []
    for item in rows:
        d = item.get("d", [])
        if len(d) < len(TV_COLUMNS):
            continue
        ticker = d[0]
        if not ticker or d[2] is None:
            continue

        results.append({
            "ticker": ticker,
            "name": d[1] or ticker,
            "price": d[2],
            "change_pct": d[3] or 0,
            "volume": d[4] or 0,
            "avg_vol_30d": d[5] or 0,
            "market_cap": d[6] or 0,
            "sector": d[7] or "Unknown",
            "perf_1y": d[8],
            "perf_6m": d[9],
            "perf_3m": d[10],
            "perf_1m": d[11],
            "rsi": d[12],
            "sma_50": d[13],
            "sma_200": d[14],
            "tv_signal": d[15],
        })

    return results


def _tv_fetch_tickers(tickers: list[dict] | list[str]) -> list[dict]:
    """Fetch specific tickers (for --tickers, e.g. backtesting the known winners)."""
    syms = [t.upper() for t in tickers]
    # TradingView resolves bare symbols here without exchange prefixes
    payload = {
        "filter": [],
        "options": {"lang": "en"},
        "symbols": {"query": {"types": []}, "tickers": syms},
        "columns": TV_COLUMNS,
        "range": [0, len(syms)],
    }
    try:
        resp = requests.post(TV_SCANNER_URL, json=payload, timeout=30)
        resp.raise_for_status()
        rows = resp.json().get("data", [])
    except Exception:
        rows = []

    found = {}
    for item in rows:
        d = item.get("d", [])
        if len(d) < len(TV_COLUMNS) or not d[0]:
            continue
        found[d[0].upper()] = {
            "ticker": d[0], "name": d[1] or d[0], "price": d[2],
            "change_pct": d[3] or 0, "volume": d[4] or 0, "avg_vol_30d": d[5] or 0,
            "market_cap": d[6] or 0, "sector": d[7] or "Unknown",
            "perf_1y": d[8], "perf_6m": d[9], "perf_3m": d[10], "perf_1m": d[11],
            "rsi": d[12], "sma_50": d[13], "sma_200": d[14], "tv_signal": d[15],
        }
    # Fall back to a bare stub so deep_scan still runs even if TV misses it
    out = []
    for s in syms:
        out.append(found.get(s, {"ticker": s, "name": s, "price": 0, "market_cap": 0,
                                 "sector": "Unknown", "avg_vol_30d": 0, "volume": 0,
                                 "perf_6m": None, "perf_1m": None, "sma_50": None}))
    return out


# ═══════════════════════════════════════════════════════════════════
# ████  STAGE 2 — SPECULATIVE-ZONE PRE-FILTER  ████
# ═══════════════════════════════════════════════════════════════════

def speculative_prefilter(stocks: list[dict], sector_filter: str | None = None,
                          verbose: bool = True) -> list[dict]:
    """
    Cheap pre-filter to the speculative-but-not-junk zone using TV's
    pre-computed fields, before the expensive yfinance deep scan.
    """
    total = len(stocks)
    if verbose:
        print(f"\n  ┌─ DNA FUNNEL: {total} small/mid-cap US stocks loaded")

    if sector_filter:
        prev = len(stocks)
        sl = sector_filter.lower()
        stocks = [s for s in stocks if sl in (s.get("sector") or "").lower()]
        if verbose:
            print(f"  ├─ Sector: '{sector_filter}' ──────────→ {len(stocks)} survive ({prev - len(stocks)} cut)")

    # Not a falling knife: price must be holding above (or near) its SMA50.
    # The DNA is "waking up," not "bleeding out."
    prev = len(stocks)
    kept = []
    for s in stocks:
        sma50 = s.get("sma_50")
        price = s.get("price") or 0
        if sma50 is None or sma50 <= 0:
            kept.append(s)  # missing data — let deep scan decide
        elif price >= sma50 * 0.90:
            kept.append(s)
    stocks = kept
    if verbose:
        print(f"  ├─ Holding ≥90% of SMA50 (no knives) ─→ {len(stocks)} survive ({prev - len(stocks)} cut)")

    # Not already a finished multi-bagger: drop anything already up >250% in 6M.
    # We want the pre-run, not the victory lap.
    prev = len(stocks)
    kept = []
    for s in stocks:
        p6 = s.get("perf_6m")
        if p6 is None or p6 <= 250:
            kept.append(s)
    stocks = kept
    if verbose:
        print(f"  ├─ 6M perf ≤ +250% (not the exit) ────→ {len(stocks)} survive ({prev - len(stocks)} cut)")

    if verbose:
        pct = (1 - len(stocks) / total) * 100 if total > 0 else 0
        print(f"  └─ FUNNEL COMPLETE: {len(stocks)} candidates ({pct:.0f}% eliminated)\n")

    return stocks


# ═══════════════════════════════════════════════════════════════════
# ████  STAGE 3 — DEEP SCAN + DNA SCORING  ████
# ═══════════════════════════════════════════════════════════════════

def _safe_get(d: dict, key: str, default=0):
    val = d.get(key, default)
    if val is None:
        return default
    try:
        f = float(val)
        if np.isnan(f) or np.isinf(f):
            return default
        return f
    except (ValueError, TypeError):
        return default


def _score_room_to_run(market_cap: float) -> float:
    """Smaller = more room to 10x. The giants are anchored by their own size."""
    if market_cap <= 0:
        return 0
    b = market_cap / 1_000_000_000  # in $B
    if b <= 0.5:   return 100   # nano/micro — maximum room
    if b <= 1.0:   return 95
    if b <= 2.0:   return 85
    if b <= 3.0:   return 70
    if b <= 5.0:   return 55
    if b <= 8.0:   return 35
    if b <= 12.0:  return 20
    return 8                    # >$12B — multi-bagging from here is a tall order


def _score_under_radar(n_analysts: float) -> float:
    """
    Forefront's whole thesis: 'How do you know if a stock will grow if there
    are no analyst ratings?' Few/no analysts = early = the opportunity.
    """
    n = int(n_analysts or 0)
    if n == 0:   return 100   # totally uncovered — the purest 'early'
    if n <= 2:   return 95
    if n <= 4:   return 85
    if n <= 6:   return 70
    if n <= 9:   return 50
    if n <= 14:  return 30
    if n <= 20:  return 15
    return 5                  # 20+ analysts — Wall Street already showed up


def _score_revenue_accel(rev_growth: float, earnings_growth: float) -> float:
    """
    Multi-baggers grow INTO their valuation (Forefront's Step 2: 'best path to
    large revenue'). But reward DURABLE hypergrowth, not accounting noise: a
    +1600% number is almost always a near-zero base, not a real growth engine,
    so it gets DISCOUNTED, not crowned.
    """
    score = 0
    if rev_growth >= 300:    score = 50   # noisy tiny base — suspicious, not rewarded
    elif rev_growth >= 120:  score = 75   # strong but verify it's real
    elif rev_growth >= 40:   score = 100  # the durable hypergrowth sweet spot
    elif rev_growth >= 25:   score = 88
    elif rev_growth >= 15:   score = 68
    elif rev_growth >= 5:    score = 45
    elif rev_growth >= 0:    score = 25
    else:                    score = 8
    # Earnings inflection bonus — turning the corner from losses is rocket fuel
    if 0 < earnings_growth < 500 and earnings_growth >= 50:
        score = min(100, score + 10)
    return score


def _score_early_not_late(price: float, sma50: float, sma200: float,
                          perf_6m: float, perf_1m: float) -> float:
    """
    The sweet spot: trend turning up (above SMA50) and structurally healthy
    (near/above SMA200), but NOT already parabolic. We want the base breakout,
    not the third leg.
    """
    score = 40
    if sma50 and price >= sma50:
        score += 20                      # trend is up
    if sma200 and price >= sma200:
        score += 15                      # above the long-term line = real strength
    elif sma200 and price >= sma200 * 0.85:
        score += 8                       # reclaiming it
    # Penalize names that already ran hard — the entry has passed
    p6 = perf_6m or 0
    if p6 > 150:
        score -= 25
    elif p6 > 80:
        score -= 10
    elif 10 <= p6 <= 80:
        score += 10                      # healthy, building, not blown out
    # A calm last month (digesting) is better than a vertical spike
    p1 = perf_1m or 0
    if p1 > 60:
        score -= 10
    return max(0, min(100, score))


def _score_volume_ignition(volume: float, avg_vol_30d: float) -> float:
    """The crowd is just starting to arrive: today's volume vs the 30d baseline."""
    if not avg_vol_30d or avg_vol_30d <= 0:
        return 30
    ratio = volume / avg_vol_30d
    if ratio >= 3.0:   return 100
    if ratio >= 2.0:   return 85
    if ratio >= 1.5:   return 70
    if ratio >= 1.0:   return 50
    if ratio >= 0.6:   return 35
    return 20


# ─── Forefront Step 1: Sector Analysis ────────────────────────────
# His green list (where every big find came from) vs his "paint-dry" avoid list.
# Keyed on yfinance's clean sector taxonomy. This is the literal automation of
# his first move: pick the right sector before you ever look at a ticker.
FOREFRONT_GREEN_SECTORS = {
    "Technology",             # software, cyber, cloud, AI, semis, hardware
    "Industrials",            # aerospace, defense, machinery, robotics
    "Energy",                 # oil, gas, solar, exploration, refining
    "Healthcare",             # drugs, experimental pharma, treatments
    "Basic Materials",        # connectors, batteries, materials, mining
    "Communication Services", # gaming, comms-tech (his NOK/GOOGL lane)
}
# Explicitly the "paint-dry" sectors he says to avoid:
#   Financial Services, Real Estate, Consumer Defensive (staples),
#   Utilities, Consumer Cyclical (apparel / fast food / retail)

WEIGHTS = {
    "room_to_run":   0.25,
    "under_radar":   0.25,
    "revenue_accel": 0.20,
    "early_not_late": 0.20,
    "volume_ignition": 0.10,
}


def deep_scan_dna(tv_data: dict, forefront_only: bool = True) -> dict | None:
    """Pull analyst coverage + growth from yfinance, then score the DNA."""
    ticker = tv_data["ticker"]
    try:
        stock = yf.Ticker(ticker)
        info = stock.info or {}
    except Exception:
        info = {}

    market_cap = float(info.get("marketCap") or tv_data.get("market_cap") or 0)
    price = float(info.get("currentPrice") or info.get("regularMarketPrice")
                  or tv_data.get("price") or 0)
    if market_cap <= 0 or price <= 0:
        return None

    sector = info.get("sector") or tv_data.get("sector", "Unknown")
    # Forefront Step 1: only hunt in his green sectors, skip the paint-dry ones.
    if forefront_only and sector not in FOREFRONT_GREEN_SECTORS:
        return None

    n_analysts = _safe_get(info, "numberOfAnalystOpinions", 0)
    rev_growth = _safe_get(info, "revenueGrowth", 0) * 100
    earnings_growth = _safe_get(info, "earningsGrowth", 0) * 100
    sma50 = _safe_get(info, "fiftyDayAverage", tv_data.get("sma_50") or 0)
    sma200 = _safe_get(info, "twoHundredDayAverage", tv_data.get("sma_200") or 0)
    perf_6m = tv_data.get("perf_6m")
    perf_1m = tv_data.get("perf_1m")
    volume = tv_data.get("volume") or _safe_get(info, "volume", 0)
    avg_vol = tv_data.get("avg_vol_30d") or _safe_get(info, "averageVolume", 0)

    scores = {
        "room_to_run":     _score_room_to_run(market_cap),
        "under_radar":     _score_under_radar(n_analysts),
        "revenue_accel":   _score_revenue_accel(rev_growth, earnings_growth),
        "early_not_late":  _score_early_not_late(price, sma50, sma200, perf_6m, perf_1m),
        "volume_ignition": _score_volume_ignition(volume, avg_vol),
    }
    dna = round(sum(scores[k] * WEIGHTS[k] for k in WEIGHTS), 1)

    if dna >= 80:
        tier, emoji = "ROCKET", "🚀"
    elif dna >= 65:
        tier, emoji = "ORBIT", "🛰️"
    elif dna >= 50:
        tier, emoji = "CLIMBING", "✈️"
    elif dna >= 30:
        tier, emoji = "DRIFTING", "🛶"
    else:
        tier, emoji = "ANCHORED", "⚓"

    # Data-quality flags — surfaced, never hidden. yfinance fundamentals on
    # micro-caps are noisy; a published screen should say so.
    flags = []
    if rev_growth >= 300:
        flags.append("noisy_growth")        # likely tiny-base, verify by hand
    if int(n_analysts or 0) == 0:
        flags.append("zero_coverage")       # purest 'early' — but also unverified
    if market_cap < 150_000_000:
        flags.append("micro_cap")           # thin, illiquid, gap risk

    return {
        "ticker": ticker,
        "name": tv_data.get("name", ticker),
        "sector": sector,
        "price": round(price, 2),
        "market_cap": market_cap,
        "n_analysts": int(n_analysts or 0),
        "rev_growth": round(rev_growth, 1),
        "earnings_growth": round(earnings_growth, 1),
        "perf_6m": perf_6m,
        "perf_1m": perf_1m,
        "dna_score": dna,
        "tier": tier,
        "emoji": emoji,
        "flags": flags,
        "axes": {k: round(v, 0) for k, v in scores.items()},
    }


# ═══════════════════════════════════════════════════════════════════
# ████  OUTPUT  ████
# ═══════════════════════════════════════════════════════════════════

def _fmt_cap(mc: float) -> str:
    if mc >= 1_000_000_000:
        return f"${mc / 1_000_000_000:.1f}B"
    return f"${mc / 1_000_000:.0f}M"


def print_report(results: list[dict], top: int):
    if not results:
        print(f"\n  ⚓ {ANCHORED_QUOTES[0]}\n")
        return

    results = sorted(results, key=lambda r: r["dna_score"], reverse=True)[:top]

    print("\n" + "═" * 78)
    print("  🚀 MULTI-BAGGER DNA — Find It Before The Analysts")
    print("     Companion to Forefront Alpha's multi-bagger framework")
    print("═" * 78)
    print(f"  {'TICKER':<8}{'DNA':>5}  {'TIER':<10}{'CAP':>8}{'ANALYSTS':>9}"
          f"{'REVGRW':>8}  SECTOR")
    print("  " + "─" * 74)
    FLAG_MARK = {"noisy_growth": "⚠rev", "zero_coverage": "⚠cov", "micro_cap": "⚠µ"}
    for r in results:
        rg = f"{r['rev_growth']:+.0f}%" if r['rev_growth'] else "n/a"
        tags = " ".join(FLAG_MARK.get(f, "") for f in r.get("flags", [])).strip()
        sec = r['sector'][:14]
        print(f"  {r['ticker']:<8}{r['dna_score']:>5.0f}  "
              f"{r['emoji']} {r['tier']:<8}{_fmt_cap(r['market_cap']):>8}"
              f"{r['n_analysts']:>9}{rg:>8}  {sec:<15}{tags}")

    top_pick = results[0]
    quote = ROCKET_QUOTES[top_pick['dna_score'] >= 80 and
                          int(top_pick['dna_score']) % len(ROCKET_QUOTES) or 0]
    print("  " + "─" * 74)
    print(f"  💬 Sam: \"{quote}\"")
    print("═" * 78 + "\n")


# ═══════════════════════════════════════════════════════════════════
# ████  MAIN  ████
# ═══════════════════════════════════════════════════════════════════

def main():
    ap = argparse.ArgumentParser(description="Multi-Bagger DNA Screener")
    ap.add_argument("--tickers", help="Comma-separated tickers to score directly")
    ap.add_argument("--sector", help="Sector filter (e.g. Technology)")
    ap.add_argument("--min-cap", default="100M", help="Min market cap (e.g. 100M)")
    ap.add_argument("--max-cap", default="10B", help="Max market cap (e.g. 10B)")
    ap.add_argument("--top", type=int, default=25, help="Top N results")
    ap.add_argument("--all-sectors", action="store_true",
                    help="Disable Forefront Step 1 sector gate (scan every sector)")
    ap.add_argument("--json", action="store_true", help="Machine-readable JSON output")
    ap.add_argument("--csv", help="Save results to CSV path")
    args = ap.parse_args()

    def parse_cap(s: str) -> float:
        s = s.strip().upper()
        mult = 1
        if s.endswith("B"):
            mult, s = 1_000_000_000, s[:-1]
        elif s.endswith("M"):
            mult, s = 1_000_000, s[:-1]
        return float(s) * mult

    verbose = not args.json
    t0 = time.time()

    if args.tickers:
        syms = [t.strip() for t in args.tickers.split(",") if t.strip()]
        if verbose:
            print(f"\n🚀 Scoring {len(syms)} tickers for multi-bagger DNA...")
        universe = _tv_fetch_tickers(syms)
        candidates = universe
    else:
        if verbose:
            print("\n🚀 Multi-Bagger DNA Screener — hunting the whole small/mid-cap market...")
        universe = _tv_fetch_universe(parse_cap(args.min_cap), parse_cap(args.max_cap))
        candidates = speculative_prefilter(universe, args.sector, verbose)

    results = []
    for i, c in enumerate(candidates):
        if verbose and not args.tickers and i % 25 == 0:
            print(f"    deep-scanning {i}/{len(candidates)}...", end="\r")
        r = deep_scan_dna(c, forefront_only=not args.all_sectors)
        if r:
            results.append(r)

    if verbose:
        print(f"\n  Scanned {len(candidates)} candidates in {time.time() - t0:.0f}s, "
              f"{len(results)} scored.")

    results.sort(key=lambda r: r["dna_score"], reverse=True)

    if args.csv:
        import csv
        with open(args.csv, "w", newline="") as f:
            w = csv.writer(f)
            w.writerow(["ticker", "name", "dna_score", "tier", "market_cap",
                        "n_analysts", "rev_growth", "perf_6m", "sector"])
            for r in results[:args.top]:
                w.writerow([r["ticker"], r["name"], r["dna_score"], r["tier"],
                            r["market_cap"], r["n_analysts"], r["rev_growth"],
                            r["perf_6m"], r["sector"]])
        if verbose:
            print(f"  💾 Saved {min(len(results), args.top)} rows → {args.csv}")

    if args.json:
        print(json.dumps(results[:args.top], indent=2, default=str))
    else:
        print_report(results, args.top)


if __name__ == "__main__":
    main()