scripts/webull_volume_spike.py

import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta

import pandas as pd
import pytz
import requests
import yfinance as yf

CACHE_DIR = "scripts/cache"


def ensure_cache_dir():
    os.makedirs(CACHE_DIR, exist_ok=True)


def get_cache_path(symbol):
    return os.path.join(CACHE_DIR, f"{symbol}.csv")


def is_cache_fresh(symbol, max_age_hours=24):
    """Check if cache file exists and is younger than max_age_hours"""
    cache_path = get_cache_path(symbol)
    if not os.path.exists(cache_path):
        return False
    mtime = datetime.fromtimestamp(os.path.getmtime(cache_path))
    return datetime.now() - mtime < timedelta(hours=max_age_hours)


def load_from_cache(symbol):
    """Load data from cache CSV"""
    cache_path = get_cache_path(symbol)
    df = pd.read_csv(cache_path, parse_dates=["Date"], index_col=0)
    return df


def save_to_cache(symbol, df):
    """Save DataFrame to cache CSV"""
    cache_path = get_cache_path(symbol)
    df.to_csv(cache_path, index=True)


def download_symbol(symbol):
    """Download 60d of data for a symbol, using cache if fresh"""
    if is_cache_fresh(symbol):
        try:
            df = load_from_cache(symbol)
            return symbol, df
        except Exception:
            pass

    df = yf.download(symbol, period="60d", interval="1d", progress=False)
    if df is not None and not (hasattr(df, "empty") and df.empty):
        if isinstance(df.columns, pd.MultiIndex):
            df.columns = df.columns.get_level_values(0)
        save_to_cache(symbol, df)
        return symbol, df
    return symbol, None


def fetch_ranking(rank_type, min_change=0.20):
    """Fetch a specific ranking type (5d, afterMarket, preMarket)"""
    items = []
    page = 1
    while True:
        r = requests.get(
            "https://quotes-gw.webullfintech.com/api/wlas/ranking/v9/rise",
            params={"regionId": 6, "rankType": rank_type, "pageIndex": page, "pageSize": 50},
            headers={"appid": "wb_web_us", "origin": "https://www.webull.com"},
            timeout=10,
        )
        data = r.json()
        batch = data.get("data", [])
        if not batch:
            break

        for item in batch:
            values = item.get("values", {})
            change_ratio = float(values.get("changeRatio", 0))
            pch_ratio = float(values.get("pchRatio", 0))

            # For 5d type, apply filter and stop condition
            if rank_type == "5d":
                if change_ratio < min_change:
                    print(f"  Page {page}: first item < {min_change * 100}% found, stopping pagination")
                    return items

            # For afterMarket/preMarket: DON'T filter by pch_ratio
            # The API returns items in some order, we take top 3 pages to avoid spam
            # WALD with +64% should appear here!
            elif rank_type in ("afterMarket", "preMarket") and page > 3:  # Max 3 pages (150 items)
                return items

            close = float(values.get("close", 0))
            pre_close = float(values.get("preClose", 0))
            today_regular = ((close / pre_close) - 1) if pre_close > 0 else 0

            ticker = item.get("ticker", {})
            items.append(
                {
                    "symbol": ticker.get("symbol", ""),
                    "name": ticker.get("name", ""),
                    "change_5d": change_ratio,
                    "today_regular": today_regular,
                    "today_ah": pch_ratio,
                    "source": rank_type,
                }
            )

        if not data.get("hasMore"):
            break
        page += 1
    return items


def fetch_gainers():
    """Obtiene gainers de múltiples rankings según sesión"""
    gainers = []

    # SIEMPRE: 5d gainers
    print("  Fetching 5d ranking...")
    gainers += fetch_ranking("5d", min_change=0.20)

    # SEGÚN SESIÓN: agregar afterMarket o preMarket
    ny_tz = pytz.timezone("America/New_York")
    now_ny = datetime.now(ny_tz)
    hour = now_ny.hour + now_ny.minute / 60

    # Extended hours: before regular (pre-market 4:00-9:30) or after (after-hours 16:00-23:59)
    if hour < 9.5 or hour >= 16:  # NOT in regular session
        if 4 <= hour < 9.5:
            print("  [Pre-market: adding preMarket ranking]")
            gainers += fetch_ranking("preMarket", min_change=0.20)
        else:  # After-hours or closed (16:00-23:59)
            print("  [After-hours/Closed: adding afterMarket ranking]")
            gainers += fetch_ranking("afterMarket", min_change=0.20)

    # ELIMINAR DUPLICADOS
    seen = set()
    unique_gainers = []
    for g in gainers:
        if g["symbol"] not in seen:
            seen.add(g["symbol"])
            unique_gainers.append(g)

    print(f"  Total unique symbols: {len(unique_gainers)}")
    return unique_gainers


def detect_spikes_parallel(gainers, limit=50, max_workers=5):
    """Detecta spikes usando descarga paralela con ThreadPoolExecutor y cache"""
    results = []
    symbols_to_process = [g["symbol"] for g in gainers[:limit]]
    gainers_dict = {g["symbol"]: g for g in gainers[:limit]}

    ensure_cache_dir()

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_symbol = {executor.submit(download_symbol, sym): sym for sym in symbols_to_process}

        for future in as_completed(future_to_symbol):
            symbol = future_to_symbol[future]
            try:
                sym, df = future.result()
                if df is None or (hasattr(df, "empty") and df.empty) or len(df) < 5:
                    continue

                if "Date" not in df.columns and df.index.name == "Date":
                    df = df.reset_index()

                df = df.tail(60)
                avg_vol = float(df["Volume"].mean())
                last_5 = df.tail(5).reset_index(drop=True)

                for i in range(len(last_5)):
                    row = last_5.iloc[i]
                    prev_idx = df.tail(5).index[i] - 1
                    prev_close = float(df.iloc[prev_idx]["Close"]) if prev_idx >= 0 else float(row["Close"])
                    spike = (float(row["Close"]) / prev_close) - 1
                    vol = float(row["Volume"])

                    if vol > 10_000_000 and vol > avg_vol and spike >= 0.20:
                        gainer = gainers_dict.get(symbol, {})
                        results.append(
                            {
                                "symbol": symbol,
                                "date": str(row["Date"].date())
                                if "Date" in row
                                else str(df.iloc[prev_idx]["Date"].date()),
                                "today_regular": gainer.get("today_regular", 0),
                                "today_ah": gainer.get("today_ah", 0),
                                "gain_5d": gainer.get("change_5d", 0),
                            }
                        )
                        break
            except Exception as e:
                print(f"Error processing {symbol}: {e}")

    results.sort(key=lambda x: x["gain_5d"], reverse=True)
    return results


def get_session_sort_key(spikes):
    """Determine sort column based on market session (NYSE timezone)"""
    ny_tz = pytz.timezone("America/New_York")
    now_ny = datetime.now(ny_tz)
    hour = now_ny.hour + now_ny.minute / 60

    if 9.5 <= hour < 16:
        print("  [Regular session - sorting by Today%]")
        return lambda x: x.get("today_regular", 0), "Today%"
    else:
        print("  [Extended session - sorting by Ext%]")
        return lambda x: x.get("today_ah", 0), "Ext%"


if __name__ == "__main__":
    print("Fetching gainers (filtered >= 20% 5d)...")
    gainers = fetch_gainers()
    print(f"Got {len(gainers)} filtered gainers. Detecting spikes in parallel...")
    spikes = detect_spikes_parallel(gainers, limit=50, max_workers=5)

    # Separate extended movers (from afterMarket/preMarket) that didn't make it to spikes
    extended_movers = []
    for g in gainers:
        if (
            g.get("source") in ("afterMarket", "preMarket")
            and g.get("today_ah", 0) >= 0.20
            and not any(s["symbol"] == g["symbol"] for s in spikes)
        ):
            extended_movers.append(g)

    # Historical Spikes: ALWAYS sort by Today% (regular session)
    spikes.sort(key=lambda x: x.get("today_regular", 0), reverse=True)

    print("\n=== Historical Spikes (Last 5 Days) ===")
    print(f"  {'Sym':4} | {'Sk':3} | {'Day%':>5} | {'Ext%':>5}")
    today = datetime.now().date()
    for s in spikes:
        spike_date = datetime.strptime(s["date"], "%Y-%m-%d").date()
        days_ago = (today - spike_date).days
        day_label = f"D{days_ago}" if days_ago > 0 else "Hoy"
        day_pct = s.get("today_regular", 0) * 100
        ext_pct = s.get("today_ah", 0) * 100
        print(f"  {s['symbol']:4} | {day_label:3} | {day_pct:>5.1f}% | {ext_pct:>5.1f}%")

    if extended_movers:
        print("\n=== Extended Hours Movers ===")
        print(f"  {'Sym':4} | {'Day%':>5} | {'Ext%':>5}")
        # Sort by Ext% descending
        extended_movers.sort(key=lambda x: x.get("today_ah", 0), reverse=True)
        for g in extended_movers:
            day_pct = g.get("today_regular", 0) * 100
            ext_pct = g.get("today_ah", 0) * 100
            print(f"  {g['symbol']:4} | {day_pct:>5.1f}% | {ext_pct:>5.1f}%")