ceodkwk/datacenterStock

1"""2NASDAQ Data Center / AI Infrastructure Stock Crawler3Fetches 2 years of price + financial data using yfinance and saves to JSON.4Usage:5  python crawler.py                  # crawl all tickers6  python crawler.py --ticker CIFR    # refresh single ticker7"""8 9import json10import math11import os12import time13import argparse14from datetime import datetime, timezone15 16import yfinance as yf17import pandas as pd18import numpy as np19 20try:21    import ta22    HAS_TA = True23except ImportError:24    HAS_TA = False25 26TICKERS = [27    "CIFR", "IREN", "APLD", "BTBT", "CLSK", "MARA", "RIOT",28    "CORZ", "HUT", "WULF",29    "NVDA", "AMD", "SMCI", "ANET", "DELL",30    "EQIX", "DLR", "CLS", "VRT",31]32 33DATA_DIR = os.path.join(os.path.dirname(__file__), "data", "stocks")34INDEX_DIR = os.path.join(os.path.dirname(__file__), "data")35 36 37def _to_float(val):38    """Convert value to float, returning None if not numeric."""39    if val is None:40        return None41    try:42        f = float(val)43        return None if math.isnan(f) or math.isinf(f) else f44    except (TypeError, ValueError):45        return None46 47 48def _safe_get(d, key):49    """Get a value from a dict, returning None if missing or NaN."""50    return _to_float(d.get(key))51 52 53def fetch_ticker_info(ticker_obj: yf.Ticker) -> dict:54    """Extract useful fields from ticker.info."""55    try:56        info = ticker_obj.info57    except Exception:58        return {}59 60    fields = [61        "marketCap", "trailingPE", "forwardPE", "enterpriseValue",62        "priceToBook", "debtToEquity", "returnOnEquity", "returnOnAssets",63        "revenuePerShare", "trailingEps", "beta",64        "fiftyTwoWeekHigh", "fiftyTwoWeekLow",65        "sharesOutstanding", "floatShares", "shortRatio",66        "currentPrice", "previousClose",67        "totalRevenue", "grossProfits", "ebitda", "operatingCashflow",68        "freeCashflow", "totalDebt", "totalCash",69        "longName", "shortName", "sector", "industry",70        "country", "fullTimeEmployees", "longBusinessSummary",71    ]72 73    result = {}74    for f in fields:75        val = info.get(f)76        if isinstance(val, (int, float)):77            result[f] = _to_float(val)78        elif isinstance(val, str):79            result[f] = val80        else:81            result[f] = None82    return result83 84 85def fetch_price_history(ticker_obj: yf.Ticker, period: str = "2y") -> list:86    """Return daily OHLCV records for the given period."""87    try:88        df = ticker_obj.history(period=period, auto_adjust=True)89    except Exception:90        return []91 92    if df.empty:93        return []94 95    records = []96    for date, row in df.iterrows():97        close = _to_float(row.get("Close"))98        if close is None:99            continue100        records.append({101            "date": date.strftime("%Y-%m-%d"),102            "open": _to_float(row.get("Open")),103            "high": _to_float(row.get("High")),104            "low": _to_float(row.get("Low")),105            "close": close,106            "volume": int(row["Volume"]) if not pd.isna(row.get("Volume", float("nan"))) else None,107        })108    return records109 110 111def _df_to_records(df) -> list:112    """Convert a yfinance financial DataFrame (rows=metrics, cols=dates) to a list of period dicts."""113    if df is None or df.empty:114        return []115    try:116        transposed = df.T117        transposed.index = pd.to_datetime(transposed.index).strftime("%Y-%m-%d")118        result = []119        for period_str, row in transposed.iterrows():120            record = {"period": period_str}121            for col, val in row.items():122                key = str(col).strip().replace(" ", "_").replace("/", "_").lower()123                record[key] = _to_float(val)124            result.append(record)125        return result126    except Exception:127        return []128 129 130def fetch_income_statement(ticker_obj: yf.Ticker) -> dict:131    """Fetch annual and quarterly income statements."""132    try:133        annual = _df_to_records(ticker_obj.financials)134        quarterly = _df_to_records(ticker_obj.quarterly_financials)135    except Exception:136        annual, quarterly = [], []137    return {"annual": annual, "quarterly": quarterly}138 139 140def fetch_balance_sheet(ticker_obj: yf.Ticker) -> dict:141    try:142        annual = _df_to_records(ticker_obj.balance_sheet)143        quarterly = _df_to_records(ticker_obj.quarterly_balance_sheet)144    except Exception:145        annual, quarterly = [], []146    return {"annual": annual, "quarterly": quarterly}147 148 149def fetch_cash_flow(ticker_obj: yf.Ticker) -> dict:150    try:151        annual = _df_to_records(ticker_obj.cashflow)152        quarterly = _df_to_records(ticker_obj.quarterly_cashflow)153    except Exception:154        annual, quarterly = [], []155    return {"annual": annual, "quarterly": quarterly}156 157 158def compute_technical_indicators(price_history: list) -> dict:159    """Compute SMA, RSI and return metrics from price history."""160    if len(price_history) < 20:161        return {}162 163    closes = pd.Series([r["close"] for r in price_history])164    volumes = pd.Series([r["volume"] or 0 for r in price_history])165 166    def sma(n):167        if len(closes) >= n:168            return _to_float(closes.rolling(n).mean().iloc[-1])169        return None170 171    sma20 = sma(20)172    sma50 = sma(50)173    sma200 = sma(200)174    current = _to_float(closes.iloc[-1])175 176    # RSI177    rsi_val = None178    if HAS_TA and len(closes) >= 15:179        try:180            rsi_series = ta.momentum.RSIIndicator(closes, window=14).rsi()181            rsi_val = _to_float(rsi_series.iloc[-1])182        except Exception:183            pass184    elif len(closes) >= 15:185        # Manual RSI186        delta = closes.diff()187        gain = delta.clip(lower=0).rolling(14).mean()188        loss = (-delta.clip(upper=0)).rolling(14).mean()189        rs = gain / loss.replace(0, float("nan"))190        rsi_series = 100 - (100 / (1 + rs))191        rsi_val = _to_float(rsi_series.iloc[-1])192 193    def pct_change_n_days(n):194        if len(closes) >= n + 1:195            past = _to_float(closes.iloc[-(n + 1)])196            if past and past != 0 and current is not None:197                return round((current - past) / past, 4)198        return None199 200    # Approximate trading days201    change_1m = pct_change_n_days(21)202    change_3m = pct_change_n_days(63)203    change_6m = pct_change_n_days(126)204    change_1y = pct_change_n_days(252)205 206    avg_vol_20d = _to_float(volumes.rolling(20).mean().iloc[-1])207 208    # Trend: bullish if SMA50 > SMA200, bearish if SMA50 < SMA200, else sideways209    if sma50 is not None and sma200 is not None:210        if sma50 > sma200 * 1.02:211            trend = "bullish"212        elif sma50 < sma200 * 0.98:213            trend = "bearish"214        else:215            trend = "sideways"216    else:217        trend = "unknown"218 219    return {220        "sma_20": sma20,221        "sma_50": sma50,222        "sma_200": sma200,223        "rsi_14": rsi_val,224        "price_change_1m": change_1m,225        "price_change_3m": change_3m,226        "price_change_6m": change_6m,227        "price_change_1y": change_1y,228        "avg_volume_20d": avg_vol_20d,229        "trend": trend,230        "current_price": current,231    }232 233 234def crawl_single_ticker(symbol: str) -> dict:235    """Fetch all data for one ticker and save to data/stocks/{symbol}.json."""236    print(f"  Fetching {symbol}...")237    ticker_obj = yf.Ticker(symbol)238 239    info = fetch_ticker_info(ticker_obj)240    price_history = fetch_price_history(ticker_obj)241    tech_indicators = compute_technical_indicators(price_history)242 243    income_stmt = fetch_income_statement(ticker_obj)244    balance_sheet = fetch_balance_sheet(ticker_obj)245    cash_flow = fetch_cash_flow(ticker_obj)246 247    data = {248        "ticker": symbol,249        "name": info.get("longName") or info.get("shortName") or symbol,250        "sector": info.get("sector"),251        "industry": info.get("industry"),252        "last_updated": datetime.now(timezone.utc).isoformat(),253        "info": info,254        "price_history": price_history,255        "technical_indicators": tech_indicators,256        "financials": {257            "income_statement": income_stmt,258            "balance_sheet": balance_sheet,259            "cash_flow": cash_flow,260        },261    }262 263    os.makedirs(DATA_DIR, exist_ok=True)264    out_path = os.path.join(DATA_DIR, f"{symbol}.json")265    with open(out_path, "w", encoding="utf-8") as f:266        json.dump(data, f, ensure_ascii=False, indent=2)267 268    print(f"  Saved {symbol} → {out_path} ({len(price_history)} price records)")269    return data270 271 272def crawl_all(tickers: list = None) -> None:273    """Crawl all tickers and update last_updated.json."""274    if tickers is None:275        tickers = TICKERS276 277    os.makedirs(DATA_DIR, exist_ok=True)278    timestamps = {}279 280    for i, symbol in enumerate(tickers, 1):281        print(f"\n[{i}/{len(tickers)}] Crawling {symbol}...")282        try:283            crawl_single_ticker(symbol)284            timestamps[symbol] = datetime.now(timezone.utc).isoformat()285        except Exception as e:286            print(f"  ERROR crawling {symbol}: {e}")287            timestamps[symbol] = f"error: {e}"288 289        if i < len(tickers):290            time.sleep(1.5)291 292    ts_path = os.path.join(INDEX_DIR, "last_updated.json")293    with open(ts_path, "w", encoding="utf-8") as f:294        json.dump(timestamps, f, indent=2)295 296    print(f"\nDone! Crawled {len(tickers)} tickers.")297 298 299if __name__ == "__main__":300    parser = argparse.ArgumentParser(description="NASDAQ stock data crawler")301    parser.add_argument("--ticker", type=str, help="Crawl a single ticker (e.g. CIFR)")302    args = parser.parse_args()303 304    if args.ticker:305        crawl_single_ticker(args.ticker.upper())306    else:307        crawl_all()308