ForexBot2/backtest/research/ff_calendar_scraper.py

"""Forex Factory calendar scraper -> backtest/cache/calendar.parquet.

Fetches weekly calendar pages and parses the event JSON embedded in
`window.calendarComponentStates` (the `days:` array). Event `dateline`
is a unix epoch, so timestamps carry no timezone ambiguity — verified
2026-07-11 against known release times (NFP 12:30/13:30 UTC across DST,
ECB 12:15/13:15) and against EUR/USD M15 volatility: the bar containing
a high-impact event averages 3.5x normal range, and shifting event
times by ±1h destroys that alignment.

Usage:
    python backtest/research/ff_calendar_scraper.py --start 2024-01-01 --end 2026-07-13
    python backtest/research/ff_calendar_scraper.py            # refresh: last stored week -> today

Merges into the existing parquet on event (id, dateline), so refreshing
is idempotent and also picks up revised `actual` values for recent weeks.
Be polite: ~1.6s between requests (a full year is ~52 requests).
"""
import argparse
import json
import time
from datetime import date, timedelta
from pathlib import Path
from typing import List, Optional

import pandas as pd
import requests

CACHE_DIR = Path(__file__).resolve().parent.parent / "cache"
PARQUET = CACHE_DIR / "calendar.parquet"

UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/126.0 Safari/537.36")

FIELDS = ["id", "dateline", "name", "country", "currency", "impactName",
          "timeLabel", "timeMasked", "actual", "forecast", "previous",
          "revision", "actualBetterWorse"]


def week_url(d: date) -> str:
    return (f"https://www.forexfactory.com/calendar?week="
            f"{d.strftime('%b').lower()}{d.day}.{d.year}")


def extract_days(html: str) -> Optional[list]:
    """Balanced-bracket extraction of the `days:` JSON array."""
    i = html.find("days:")
    if i < 0:
        return None
    j = html.find("[", i)
    depth, in_str, esc = 0, False, False
    for k in range(j, len(html)):
        ch = html[k]
        if in_str:
            if esc:
                esc = False
            elif ch == "\\":
                esc = True
            elif ch == '"':
                in_str = False
        else:
            if ch == '"':
                in_str = True
            elif ch == "[":
                depth += 1
            elif ch == "]":
                depth -= 1
                if depth == 0:
                    return json.loads(html[j:k + 1])
    return None


def fetch_week(sess: requests.Session, d: date, retries: int = 4) -> Optional[List[dict]]:
    for attempt in range(retries):
        try:
            r = sess.get(week_url(d), timeout=30)
            if r.status_code == 200:
                days = extract_days(r.text)
                if days is not None:
                    return [{k: e.get(k) for k in FIELDS}
                            for day in days for e in day.get("events", [])]
        except requests.RequestException:
            pass
        time.sleep(5 * (attempt + 1))
    return None


def scrape(start: date, end: date) -> pd.DataFrame:
    sess = requests.Session()
    sess.headers["User-Agent"] = UA
    rows: List[dict] = []
    failed: List[date] = []
    d = start
    while d <= end:
        week_rows = fetch_week(sess, d)
        if week_rows is None:
            failed.append(d)
            print(f"FAILED week {d}")
        else:
            rows.extend(week_rows)
        d += timedelta(days=7)
        time.sleep(1.6)
    if failed:
        print(f"WARNING: {len(failed)} weeks failed: {failed}")
    return pd.DataFrame(rows)


def merge_into_store(new: pd.DataFrame, path: Path = PARQUET) -> pd.DataFrame:
    if path.exists():
        old = pd.read_parquet(path)
        combined = pd.concat([old, new])
    else:
        combined = new
    # keep="last": a refresh overwrites stale rows (actuals fill in
    # after release, revisions land weeks later).
    combined = (combined.drop_duplicates(subset=["id", "dateline"], keep="last")
                .sort_values("dateline").reset_index(drop=True))
    combined.to_parquet(path, index=False)
    return combined


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--start", type=date.fromisoformat, default=None)
    ap.add_argument("--end", type=date.fromisoformat, default=date.today())
    args = ap.parse_args()

    start = args.start
    if start is None:
        if PARQUET.exists():
            last = pd.read_parquet(PARQUET).dateline.max()
            start = pd.Timestamp(last, unit="s").date() - timedelta(days=7)
        else:
            start = date(2024, 1, 1)
    print(f"Scraping {start} -> {args.end}")
    new = scrape(start, args.end)
    combined = merge_into_store(new)
    print(f"Store: {len(combined)} events "
          f"({pd.Timestamp(combined.dateline.min(), unit='s')} -> "
          f"{pd.Timestamp(combined.dateline.max(), unit='s')}) -> {PARQUET}")


if __name__ == "__main__":
    main()