OpenFin/backend/services/te_html_parser.py

"""
Parser for Trading Economics economic calendar HTML page.
Converts scraped TE HTML → ff_calendar rows (no API key required).

Column map in TE HTML table rows:
  td[0]  = time (AM/PM, Europe/Zurich timezone) + date as CSS class
  td[3]  = country code (JP, US, DE, EA, ...)
  td[4]  = event name <a> + period <span class="calendar-reference">
  td[5]  = actual value
  td[6]  = previous (+ revised if any)
  td[7]  = consensus (analyst forecast — this is what we store as forecast_value)
  td[8]  = TE model forecast (ignored)
"""
import logging
import re
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)

# Countries → major currency (Eurozone countries all → EUR)
_COUNTRY_TO_CCY: Dict[str, str] = {
    "united states":   "USD",
    "euro area":       "EUR",
    "germany":         "EUR",
    "france":          "EUR",
    "italy":           "EUR",
    "spain":           "EUR",
    "netherlands":     "EUR",
    "austria":         "EUR",
    "portugal":        "EUR",
    "belgium":         "EUR",
    "greece":          "EUR",
    "finland":         "EUR",
    "ireland":         "EUR",
    "slovakia":        "EUR",
    "united kingdom":  "GBP",
    "japan":           "JPY",
    "china":           "CNY",
    "australia":       "AUD",
    "canada":          "CAD",
    "new zealand":     "NZD",
    "switzerland":     "CHF",
}

# High-impact categories
_HIGH_CATS = {
    "interest rate", "non farm payrolls", "inflation rate",
    "gdp growth rate", "gdp annual growth rate", "unemployment rate",
}
# Medium-impact categories
_MEDIUM_CATS = {
    "manufacturing pmi", "non manufacturing pmi", "balance of trade",
    "retail sales mom", "consumer confidence", "business confidence",
    "core inflation rate", "core inflation rate mom",
    "producer price inflation mom", "monthly gdp mom",
    "job offers", "zew economic sentiment index",
}

_TZ_ZURICH = None

def _get_tz():
    global _TZ_ZURICH
    if _TZ_ZURICH is None:
        try:
            from zoneinfo import ZoneInfo
            _TZ_ZURICH = ZoneInfo("Europe/Zurich")
        except Exception:
            import pytz
            _TZ_ZURICH = pytz.timezone("Europe/Zurich")
    return _TZ_ZURICH


def _parse_time_utc(date_str: str, time_str: str) -> Tuple[str, str]:
    """
    Convert 'YYYY-MM-DD' + '02:30 PM' (Europe/Zurich) → ('YYYY-MM-DD', 'HH:MM') UTC.
    Falls back to storing the raw time if parsing fails.
    """
    if not time_str or not date_str:
        return date_str, "00:00"
    try:
        tz = _get_tz()
        dt_local = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %I:%M %p")
        dt_local = dt_local.replace(tzinfo=tz)
        dt_utc = dt_local.astimezone(timezone.utc)
        return dt_utc.strftime("%Y-%m-%d"), dt_utc.strftime("%H:%M")
    except Exception:
        return date_str, "00:00"


def _clean_value(v: Optional[str]) -> Optional[str]:
    """Normalize numeric string: strip whitespace/units, return None if empty."""
    if not v:
        return None
    v = v.strip()
    if not v or v in ("-", "—", "N/A", ""):
        return None
    return v


def _infer_impact(category: str) -> str:
    cat = (category or "").lower().strip()
    if cat in _HIGH_CATS:
        return "high"
    if cat in _MEDIUM_CATS:
        return "medium"
    return "low"


def parse_html(html: str) -> List[Dict[str, Any]]:
    """
    Parse TE calendar HTML, return list of event dicts ready to upsert.
    Each dict: event_date, event_time, currency, impact, event_name,
                actual, forecast, previous
    """
    try:
        from bs4 import BeautifulSoup
    except ImportError:
        raise RuntimeError("beautifulsoup4 not installed — run: pip install beautifulsoup4")

    soup = BeautifulSoup(html, "html.parser")
    rows = soup.find_all("tr", attrs={"data-event": True})
    logger.info(f"[TE HTML parser] {len(rows)} rows found")

    events = []
    skipped = 0

    for row in rows:
        country = (row.get("data-country") or "").lower().strip()
        ccy = _COUNTRY_TO_CCY.get(country)
        if not ccy:
            skipped += 1
            continue

        category = (row.get("data-category") or "").lower().strip()
        impact = _infer_impact(category)

        tds = row.find_all("td")
        if len(tds) < 8:
            continue

        # Date from td[0] CSS class (e.g. "2025-04-01")
        td0_classes = tds[0].get("class") or []
        date_str = ""
        for c in td0_classes:
            if re.match(r"\d{4}-\d{2}-\d{2}", c):
                date_str = c
                break
        if not date_str:
            continue

        # Time from td[0] text
        time_raw = tds[0].get_text(separator=" ").strip()
        time_match = re.search(r"\d{1,2}:\d{2}\s*[AP]M", time_raw, re.I)
        time_str = time_match.group(0).strip() if time_match else ""

        ev_date, ev_time = _parse_time_utc(date_str, time_str)

        # Event name from td[4] <a class="calendar-event">
        ev_link = tds[4].find("a", class_="calendar-event")
        if ev_link:
            event_name = ev_link.get_text(strip=True)
        else:
            event_name = tds[4].get_text(strip=True)
            # Strip appended period reference (MAR, Q1, APR, etc.)
            event_name = re.sub(r"\s+(JAN|FEB|MAR|APR|MAY|JUN|JUL|AUG|SEP|OCT|NOV|DEC|Q[1-4])$", "", event_name, flags=re.I)
        event_name = event_name.strip()
        if not event_name:
            continue

        # Values
        actual   = _clean_value(tds[5].get_text(strip=True))
        previous_span = tds[6].find("span", id="previous")
        if previous_span:
            previous = _clean_value(previous_span.get_text(strip=True))
        else:
            previous = _clean_value(tds[6].get_text(strip=True))
        consensus = _clean_value(tds[7].get_text(strip=True))

        events.append({
            "event_date":  ev_date,
            "event_time":  ev_time,
            "currency":    ccy,
            "impact":      impact,
            "event_name":  event_name,
            "actual":      actual,
            "forecast":    consensus,   # consensus = analyst forecast
            "previous":    previous,
        })

    logger.info(f"[TE HTML parser] {len(events)} valid events, {skipped} skipped")
    return events


def import_html_file(file_path: str) -> Dict[str, Any]:
    """
    Parse TE calendar HTML file and upsert all events into ff_calendar.
    Returns summary dict.
    """
    from services.database import get_conn
    from services.ff_calendar import _upsert_batch, FF_TO_FRED
    import os

    if not os.path.exists(file_path):
        return {"error": f"File not found: {file_path}"}

    print(f"[TE HTML] Reading {file_path} …", flush=True)
    with open(file_path, encoding="utf-8", errors="replace") as f:
        html = f.read()
    print(f"[TE HTML] File size: {len(html):,} chars", flush=True)

    events = parse_html(html)
    if not events:
        return {"error": "No events parsed — check file format"}

    batch = []
    for ev in events:
        series_id = FF_TO_FRED.get(ev["event_name"])
        batch.append((
            ev["event_date"], ev["event_time"], ev["currency"],
            ev["impact"], ev["event_name"],
            ev["actual"], ev["forecast"], ev["previous"],
            series_id, None, "te_html",
        ))

    CHUNK = 500
    conn = get_conn()
    for i in range(0, len(batch), CHUNK):
        _upsert_batch(conn, batch[i:i + CHUNK])
        conn.commit()
    conn.close()

    by_ccy: Dict[str, int] = {}
    for ev in events:
        by_ccy[ev["currency"]] = by_ccy.get(ev["currency"], 0) + 1

    print(f"[TE HTML] {len(batch)} events upserted", flush=True)
    return {
        "total":     len(batch),
        "by_currency": by_ccy,
        "date_from": min(ev["event_date"] for ev in events),
        "date_to":   max(ev["event_date"] for ev in events),
    }