Scrape.py

"""
DAY 1 — Oil Intelligence Data Pipeline
Pulls from: Yahoo Finance (prices), EIA API (inventories), NewsAPI (news)
Stores everything in /data as JSON/CSV
"""

import os
from dotenv import load_dotenv
import json
import time
from datetime import datetime, timedelta
from pathlib import Path

import yfinance as yf
import pandas as pd
import requests

# ── Config ──────────────────────────────────────────────────────────────────
load_dotenv()  # Load environment variables from .env file
EIA_API_KEY  = os.getenv("EIA_API_KEY",  "YOUR_EIA_KEY")   # free at eia.gov/opendata
NEWS_API_KEY = os.getenv("NEWS_API_KEY", "YOUR_NEWS_KEY")  # free at newsapi.org

DATA_DIR = Path("data")
DATA_DIR.mkdir(exist_ok=True)

# ── 1. Oil Prices (Yahoo Finance — no key needed) ───────────────────────────
def fetch_prices() -> pd.DataFrame:
    """Pull 2 years of WTI crude + Brent futures."""
    print("📈 Fetching oil prices...")
    tickers = {
        "WTI":   "CL=F",   # WTI Crude Futures
        "Brent": "BZ=F",   # Brent Crude Futures
        "XOM":   "XOM",    # ExxonMobil (proxy for oil sector)
    }
    frames = []
    for name, symbol in tickers.items():
        df = yf.download(symbol, period="2y", progress=False)[["Close", "Volume"]]
        df.columns = [f"{name}_close", f"{name}_volume"]
        frames.append(df)

    prices = pd.concat(frames, axis=1).dropna()
    prices.index.name = "date"

    out = DATA_DIR / "oil_prices.csv"
    prices.to_csv(out)
    print(f"   ✅ Saved {len(prices)} rows → {out}")
    return prices


# ── 2. EIA Petroleum Inventories ────────────────────────────────────────────
EIA_SERIES = {
    "us_crude_inventory":    "PET.WCRSTUS1.W",   # Weekly US crude stocks
    "us_crude_production":   "PET.WCRFPUS2.W",   # Weekly US field production
    "us_gasoline_inventory": "PET.WGTSTUS1.W",   # Weekly gasoline stocks
}

def fetch_eia(series_id: str, name: str) -> list[dict]:
    """Pull a single EIA weekly series (last 2 years)."""
    url = "https://api.eia.gov/v2/seriesid/{sid}".format(sid=series_id)
    params = {
        "api_key": EIA_API_KEY,
        "data[0]": "value",
        "sort[0][column]": "period",
        "sort[0][direction]": "desc",
        "length": 104,   # ~2 years of weekly data
    }
    r = requests.get(url, params=params, timeout=15)
    r.raise_for_status()
    rows = r.json()["response"]["data"]
    for row in rows:
        row["series"] = name
    return rows

def fetch_all_eia() -> pd.DataFrame:
    """Pull all EIA series and combine."""
    print("🛢  Fetching EIA inventory data...")
    all_rows = []
    for name, sid in EIA_SERIES.items():
        try:
            rows = fetch_eia(sid, name)
            all_rows.extend(rows)
            print(f"   ✅ {name}: {len(rows)} weeks")
        except Exception as e:
            print(f"   ⚠️  {name} failed: {e}")
        time.sleep(0.5)   # be polite to the API

    df = pd.DataFrame(all_rows)
    out = DATA_DIR / "eia_inventories.csv"
    df.to_csv(out, index=False)
    print(f"   💾 Saved → {out}")
    return df


# ── 3. Oil News (NewsAPI) ────────────────────────────────────────────────────
NEWS_QUERIES = [
    "crude oil price OPEC",
    "WTI Brent oil market",
    "oil supply demand forecast",
    "oil sanctions pipeline",
]

def fetch_news(days_back: int = 28) -> list[dict]:
    """Pull recent oil news articles."""
    print("📰 Fetching oil news...")
    from_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d")
    articles = []

    for query in NEWS_QUERIES:
        url = "https://newsapi.org/v2/everything"
        params = {
            "q":        query,
            "from":     from_date,
            "language": "en",
            "sortBy":   "relevancy",
            "pageSize": 25,
            "apiKey":   NEWS_API_KEY,
        }
        try:
            r = requests.get(url, params=params, timeout=15)
            r.raise_for_status()
            hits = r.json().get("articles", [])
            # Keep only what we need (avoid storing full content blob)
            for a in hits:
                articles.append({
                    "title":       a.get("title", ""),
                    "source":      a.get("source", {}).get("name", ""),
                    "published_at": a.get("publishedAt", ""),
                    "url":         a.get("url", ""),
                    "description": a.get("description", ""),
                    "content":     a.get("content", ""),   # truncated by NewsAPI (~200 chars)
                    "query":       query,
                })
            print(f"   ✅ '{query}': {len(hits)} articles")
        except Exception as e:
            print(f"   ⚠️  '{query}' failed: {e}")
        time.sleep(0.3)

    # Deduplicate by URL
    seen, unique = set(), []
    for a in articles:
        if a["url"] not in seen:
            seen.add(a["url"])
            unique.append(a)

    out = DATA_DIR / "oil_news.json"
    out.write_text(json.dumps(unique, indent=2))
    print(f"   💾 {len(unique)} unique articles → {out}")
    return unique


# ── 4. Build RAG Document Corpus ─────────────────────────────────────────────
def build_corpus(news: list[dict], prices: pd.DataFrame, eia: pd.DataFrame):
    """
    Combine all data sources into plain-text documents for RAG ingestion.
    Each document = one chunk the vector DB will index.
    """
    print("\n📦 Building RAG corpus...")
    docs = []

    # News articles → one doc each
    for a in news:
        text = f"[NEWS] {a['published_at'][:10]} | {a['source']}\n"
        text += f"Title: {a['title']}\n"
        if a["description"]:
            text += f"{a['description']}\n"
        if a["content"]:
            text += f"{a['content']}\n"
        docs.append({
            "id":     f"news_{len(docs)}",
            "source": "news",
            "date":   a["published_at"][:10],
            "text":   text.strip(),
        })

    # Price data → weekly summary chunks (every 4 weeks)
    prices_reset = prices.reset_index()
    for i in range(0, len(prices_reset), 20):   # ~1-month windows
        chunk = prices_reset.iloc[i:i+20]
        start = str(chunk["date"].iloc[0])[:10]
        end   = str(chunk["date"].iloc[-1])[:10]
        wti_avg   = chunk["WTI_close"].mean()
        brent_avg = chunk["Brent_close"].mean()
        text = (
            f"[PRICE DATA] {start} to {end}\n"
            f"WTI avg close: ${wti_avg:.2f}/bbl\n"
            f"Brent avg close: ${brent_avg:.2f}/bbl\n"
        )
        docs.append({"id": f"price_{i}", "source": "prices", "date": end, "text": text})

    # EIA data → one doc per series per month
    if not eia.empty and "series" in eia.columns:
        for series_name, group in eia.groupby("series"):
            group = group.sort_values("period").tail(12)   # last 12 weeks
            rows_text = "\n".join(
                f"  {row['period']}: {row['value']}" for _, row in group.iterrows()
            )
            text = f"[EIA DATA] {series_name} (recent 12 weeks)\n{rows_text}"
            docs.append({"id": f"eia_{series_name}", "source": "eia", "date": "recent", "text": text})

    out = DATA_DIR / "corpus.json"
    out.write_text(json.dumps(docs, indent=2))
    print(f"   ✅ {len(docs)} documents → {out}")
    return docs


# ── Main ─────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    print("=" * 55)
    print("  OIL INTELLIGENCE — Day 1: Data Pipeline")
    print("=" * 55)

    prices = fetch_prices()
    eia    = fetch_all_eia()
    news   = fetch_news(days_back=28)
    docs   = build_corpus(news, prices, eia)

    print("\n🎉 Done! Files in /data:")
    for f in sorted(DATA_DIR.iterdir()):
        size = f.stat().st_size / 1024
        print(f"   {f.name:30s} {size:6.1f} KB")
    print("\nNext step: run day2_rag.py to embed and index these docs.")