Excel_Generator/Stats_data_collection/Forex/Scrapping.py

#!/usr/bin/env python3
"""
Comprehensive Forex & Financial Data scraper.
Sources: investing.com, hamariweb.com, tradingeconomics.com, easydata.sbp.org.pk
Uses Playwright headless for JS-rendered pages, requests for static pages.
"""
from playwright.sync_api import sync_playwright
from bs4 import BeautifulSoup
import requests
import json
import os
import re
import time
from datetime import datetime

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
OUTPUT_FILE = os.path.join(SCRIPT_DIR, 'forex_data.json')

HTTP_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}

# ── investing.com (JS-rendered, needs Playwright) ──────────────────────

INVESTING_PAIRS = {
    "USD_PKR": "https://www.investing.com/currencies/usd-pkr",
    "EUR_PKR": "https://www.investing.com/currencies/eur-pkr",
    "EUR_USD": "https://www.investing.com/currencies/eur-usd",
    "CNY_PKR": "https://www.investing.com/currencies/cny-pkr",
}

def scrape_investing_pairs(page):
    """Scrape currency pairs from investing.com using a shared Playwright page."""
    results = {}
    for label, url in INVESTING_PAIRS.items():
        print(f"  Fetching {label} ...")
        try:
            page.goto(url, timeout=60000, wait_until="domcontentloaded")
            page.wait_for_selector('[data-test="instrument-price-last"]', timeout=20000)
            time.sleep(1)
            html = page.content()
            soup = BeautifulSoup(html, "html.parser")

            data = {"last_price": None, "change": None, "percent_change": None}
            el = soup.find(attrs={"data-test": "instrument-price-last"})
            if el:
                data["last_price"] = el.get_text(strip=True)
            el = soup.find(attrs={"data-test": "instrument-price-change"})
            if el:
                data["change"] = el.get_text(strip=True)
            el = soup.find(attrs={"data-test": "instrument-price-change-percent"})
            if el:
                data["percent_change"] = el.get_text(strip=True)

            results[label] = data
        except Exception as e:
            print(f"  Error {label}: {e}")
            results[label] = {"error": str(e)}
    return results


def scrape_usdpkr_forwards(page):
    """Scrape USD/PKR forward rates from investing.com."""
    url = "https://www.investing.com/currencies/usd-pkr-forward-rates"
    print(f"  Fetching USD/PKR Forwards ...")
    try:
        page.goto(url, timeout=60000, wait_until="domcontentloaded")
        page.wait_for_selector("tr[id^='pair_']", timeout=20000)
        time.sleep(1)
        html = page.content()
        soup = BeautifulSoup(html, "html.parser")

        forwards = []
        for tr in soup.find_all("tr", id=re.compile(r'^pair_\d+')):
            cells = tr.find_all("td")
            if len(cells) < 7:
                continue
            name_td = cells[1]
            name = name_td.get_text(strip=True).replace('\xa0', ' ')
            # Only pick 1M-4M forwards
            if "FWD" not in name:
                continue
            bid = cells[2].get_text(strip=True)
            ask = cells[3].get_text(strip=True)
            high = cells[4].get_text(strip=True)
            low = cells[5].get_text(strip=True)
            change = cells[6].get_text(strip=True)
            forwards.append({
                "name": name,
                "bid": bid,
                "ask": ask,
                "high": high,
                "low": low,
                "change": change
            })
        return forwards
    except Exception as e:
        print(f"  Error forwards: {e}")
        return {"error": str(e)}


# ── hamariweb.com Open Market (simple HTTP) ────────────────────────────

def scrape_hamariweb():
    """Scrape open market forex rates from hamariweb.com."""
    url = "https://hamariweb.com/finance/forex/"
    print(f"  Fetching Open Market rates ...")
    try:
        response = requests.get(url, headers=HTTP_HEADERS, timeout=15)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, "html.parser")

        targets = {"usd-to-pkr": "USD_PKR", "eur-to-pkr": "EUR_PKR", "gbp-to-pkr": "GBP_PKR"}
        results = {}

        for a_tag in soup.find_all("a", href=True):
            href = a_tag.get("href", "")
            for slug, key in targets.items():
                if slug in href:
                    tr = a_tag.find_parent("tr")
                    if tr:
                        tds = tr.find_all("td")
                        if len(tds) >= 3:
                            results[key] = {
                                "buying": tds[1].get_text(strip=True),
                                "selling": tds[2].get_text(strip=True)
                            }
        return results
    except Exception as e:
        print(f"  Error hamariweb: {e}")
        return {"error": str(e)}


# ── tradingeconomics.com (simple HTTP) ─────────────────────────────────

def scrape_tradingeconomics():
    """Scrape Pakistan interest rate, forex reserves, interbank rate."""
    url = "https://tradingeconomics.com/pakistan/interest-rate"
    print(f"  Fetching Pakistan financial indicators ...")
    try:
        headers = {**HTTP_HEADERS, "Accept": "text/html"}
        response = requests.get(url, headers=headers, timeout=15)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, "html.parser")

        results = {}

        # Interest Rate - from the main indicator table
        for a_tag in soup.find_all("a", href=True):
            href = a_tag.get("href", "").lower()
            text = a_tag.get_text(strip=True)
            tr = a_tag.find_parent("tr")
            if not tr:
                continue
            tds = tr.find_all("td")
            if len(tds) < 5:
                continue

            if "interest-rate" in href and "Interest Rate" in text:
                results["interest_rate"] = {
                    "value": tds[1].get_text(strip=True),
                    "previous": tds[2].get_text(strip=True),
                    "unit": "percent"
                }
            elif "foreign-exchange-reserves" in href:
                results["foreign_exchange_reserves"] = {
                    "value": tds[1].get_text(strip=True),
                    "previous": tds[2].get_text(strip=True),
                    "unit": tds[3].get_text(strip=True) if len(tds) > 3 else "",
                    "date": tds[4].get_text(strip=True) if len(tds) > 4 else ""
                }
            elif "interbank-rate" in href:
                results["interbank_rate"] = {
                    "value": tds[1].get_text(strip=True),
                    "previous": tds[2].get_text(strip=True),
                    "unit": "percent"
                }

        return results
    except Exception as e:
        print(f"  Error tradingeconomics: {e}")
        return {"error": str(e)}


# ── SBP easydata KIBID/KIBOR (simple HTTP) ─────────────────────────────

def scrape_sbp_kibor():
    """Scrape latest KIBID and KIBOR (Six-Months) from SBP easydata."""
    url = "https://easydata.sbp.org.pk/apex/f?p=10:211:4932927851621::NO:RP:P211_DATASET_TYPE_CODE,P211_PAGE_ID:TS_GP_BAM_SIRKIBOR_D,1&cs=1883CA5742C889BB27CD0C1C818F1AB8B"
    print(f"  Fetching KIBID/KIBOR ...")
    try:
        response = requests.get(url, headers=HTTP_HEADERS, timeout=15, verify=False)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, "html.parser")

        results = {}

        # Get the latest date from the last header column
        latest_date = None
        header_row = soup.find("tr", attrs={"class": None})
        for th in soup.find_all("th", class_="t20ReportHeader"):
            th_id = th.get("id", "")
            if re.match(r'\d{2}-\w{3}-\d{4}', th_id):
                latest_date = th_id  # keep overwriting, last one is the latest

        for tr in soup.find_all("tr", class_="highlight-row"):
            tds = tr.find_all("td")
            if len(tds) < 3:
                continue

            series_name = ""
            for td in tds:
                text = td.get_text(strip=True)
                if "Six-Months Karachi Interbank" in text:
                    series_name = text
                    break

            if not series_name:
                continue

            # Get the value from the LAST td that contains a span with a number
            last_value = None
            for td in reversed(tds):
                span = td.find("span")
                if span:
                    val = span.get_text(strip=True)
                    try:
                        last_value = float(val)
                        break
                    except ValueError:
                        continue

            if "Bid" in series_name:
                results["KIBID_6M"] = {"name": series_name, "latest_date": latest_date, "latest_value": last_value}
            elif "Offer" in series_name:
                results["KIBOR_6M"] = {"name": series_name, "latest_date": latest_date, "latest_value": last_value}

        return results
    except Exception as e:
        print(f"  Error SBP: {e}")
        return {"error": str(e)}


# ── Main ────────────────────────────────────────────────────────────────

def main():
    print("=" * 50)
    print("Forex & Financial Data Scraper")
    print("=" * 50)

    all_data = {"scraped_at": datetime.now().isoformat()}

    # 1. investing.com pairs + forwards (Playwright headless)
    print("\n[1/4] investing.com (Playwright headless)")
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
        )
        page = context.new_page()

        all_data["investing_pairs"] = scrape_investing_pairs(page)
        all_data["usdpkr_forwards"] = scrape_usdpkr_forwards(page)

        browser.close()

    # 2. hamariweb Open Market
    print("\n[2/4] hamariweb.com Open Market rates")
    all_data["open_market"] = scrape_hamariweb()

    # 3. tradingeconomics Pakistan
    print("\n[3/4] tradingeconomics.com Pakistan indicators")
    all_data["pakistan_indicators"] = scrape_tradingeconomics()

    # 4. SBP KIBID/KIBOR
    print("\n[4/4] SBP easydata KIBID/KIBOR")
    all_data["kibid_kibor"] = scrape_sbp_kibor()

    # Save
    with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
        json.dump(all_data, f, indent=4, ensure_ascii=False)

    print(f"\n{'=' * 50}")
    print(f"All data saved to: {OUTPUT_FILE}")

if __name__ == "__main__":
    import urllib3
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    main()