690 lines
25 KiB
Python
690 lines
25 KiB
Python
"""
|
|
Forex Factory calendar service.
|
|
- CSV bulk import from forex_factory_cache.csv (2007-2025)
|
|
- Live sync from nfs.faireconomy.media JSON endpoint (this week)
|
|
- HTML scraper for upcoming weeks (forexfactory.com/calendar?week=...)
|
|
"""
|
|
import csv
|
|
import json
|
|
import logging
|
|
from datetime import datetime, date, timezone, timedelta
|
|
from typing import Any, Dict, List, Optional
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import httpx
|
|
from bs4 import BeautifulSoup
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_ET = ZoneInfo("America/New_York") # FF publishes times in US Eastern
|
|
|
|
_FF_THIS_WEEK_URL = "https://nfs.faireconomy.media/ff_calendar_thisweek.json"
|
|
_FF_NEXT_WEEK_URL = "https://nfs.faireconomy.media/ff_calendar_nextweek.json"
|
|
|
|
# Map FF impact strings → normalized
|
|
_IMPACT_MAP = {
|
|
"High Impact Expected": "high",
|
|
"High": "high",
|
|
"Medium Impact Expected": "medium",
|
|
"Medium": "medium",
|
|
"Low Impact Expected": "low",
|
|
"Low": "low",
|
|
"Non-Economic": "non-economic",
|
|
"Holiday": "non-economic",
|
|
}
|
|
|
|
# Currencies we care about (drop 'All' pseudo-currency)
|
|
SUPPORTED_CURRENCIES = {"USD", "EUR", "GBP", "JPY", "AUD", "CAD", "NZD", "CHF", "CNY"}
|
|
|
|
# ── FF event name → FRED series_id mapping ────────────────────────────────────
|
|
# Exact FF event names → our FRED series catalog
|
|
FF_TO_FRED: dict[str, str] = {
|
|
# Employment
|
|
"Non-Farm Employment Change": "PAYEMS",
|
|
"ADP Non-Farm Employment Change": "ADP_NFP",
|
|
"Unemployment Rate": "UNRATE",
|
|
"Unemployment Claims": "ICSA",
|
|
# Inflation — CPI
|
|
"CPI m/m": "CPIAUCSL",
|
|
"CPI y/y": "CPIAUCSL",
|
|
"Core CPI m/m": "CPILFESL",
|
|
"Core CPI y/y": "CPILFESL",
|
|
# Inflation — PCE
|
|
"Core PCE Price Index m/m": "PCEPILFE",
|
|
"Core PCE Price Index y/y": "PCEPILFE",
|
|
# Monetary policy
|
|
"Federal Funds Rate": "FEDFUNDS",
|
|
"FOMC Statement": "FEDFUNDS",
|
|
# Growth — GDP
|
|
"Advance GDP q/q": "GDPC1",
|
|
"Prelim GDP q/q": "GDPC1",
|
|
"Final GDP q/q": "GDPC1",
|
|
"GDP q/q": "GDPC1",
|
|
}
|
|
|
|
|
|
def _parse_ff_datetime(dt_str: str) -> tuple[str, str]:
|
|
"""
|
|
Parse an ISO datetime with offset (e.g. '2007-01-05T17:00:00+03:30')
|
|
into (event_date YYYY-MM-DD UTC, event_time HH:MM UTC).
|
|
"""
|
|
try:
|
|
dt = datetime.fromisoformat(dt_str)
|
|
dt_utc = dt.astimezone(timezone.utc)
|
|
return dt_utc.strftime("%Y-%m-%d"), dt_utc.strftime("%H:%M")
|
|
except Exception:
|
|
# Fallback: take the date/time literally
|
|
parts = dt_str[:16].split("T")
|
|
return parts[0], parts[1] if len(parts) > 1 else "00:00"
|
|
|
|
|
|
def import_csv(csv_path: str, progress_cb=None) -> Dict[str, Any]:
|
|
"""
|
|
Bulk-import forex_factory_cache.csv into ff_calendar table.
|
|
Skips non-economic events and unsupported currencies.
|
|
Returns {inserted, skipped, total}.
|
|
"""
|
|
from services.database import get_conn
|
|
|
|
conn = get_conn()
|
|
inserted = 0
|
|
skipped = 0
|
|
total = 0
|
|
|
|
try:
|
|
with open(csv_path, encoding="utf-8", errors="replace") as f:
|
|
reader = csv.DictReader(f)
|
|
batch = []
|
|
|
|
for row in reader:
|
|
total += 1
|
|
currency = (row.get("Currency") or "").strip()
|
|
impact_raw = (row.get("Impact") or "").strip()
|
|
impact = _IMPACT_MAP.get(impact_raw, "low")
|
|
|
|
# Skip non-economic and unsupported currencies
|
|
if currency not in SUPPORTED_CURRENCIES:
|
|
skipped += 1
|
|
continue
|
|
if impact == "non-economic":
|
|
skipped += 1
|
|
continue
|
|
|
|
event_name = (row.get("Event") or "").strip()
|
|
if not event_name:
|
|
skipped += 1
|
|
continue
|
|
|
|
dt_str = (row.get("DateTime") or "").strip()
|
|
event_date, event_time = _parse_ff_datetime(dt_str)
|
|
|
|
actual = (row.get("Actual") or "").strip() or None
|
|
forecast = (row.get("Forecast") or "").strip() or None
|
|
previous = (row.get("Previous") or "").strip() or None
|
|
detail = (row.get("Detail") or "").strip() or None
|
|
|
|
series_id = FF_TO_FRED.get(event_name)
|
|
batch.append((
|
|
event_date, event_time, currency, impact, event_name,
|
|
actual, forecast, previous, series_id, detail, "ff_cache",
|
|
))
|
|
|
|
if len(batch) >= 500:
|
|
_upsert_batch(conn, batch)
|
|
inserted += len(batch)
|
|
batch = []
|
|
if progress_cb:
|
|
progress_cb({"inserted": inserted, "total": total})
|
|
|
|
if batch:
|
|
_upsert_batch(conn, batch)
|
|
inserted += len(batch)
|
|
|
|
conn.commit()
|
|
print(f"[FF import] Done: {inserted} inserted, {skipped} skipped, {total} total", flush=True)
|
|
return {"inserted": inserted, "skipped": skipped, "total": total}
|
|
except Exception as e:
|
|
logger.error(f"[FF import] Failed: {e}")
|
|
print(f"[FF import] ERROR: {e}", flush=True)
|
|
return {"error": str(e)}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _upsert_batch(conn, batch: list):
|
|
conn.executemany(
|
|
"""INSERT INTO ff_calendar
|
|
(event_date, event_time, currency, impact, event_name,
|
|
actual_value, forecast_value, previous_value, series_id, detail, source)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?)
|
|
ON CONFLICT(event_date, event_time, currency, event_name)
|
|
DO UPDATE SET
|
|
actual_value = COALESCE(excluded.actual_value, ff_calendar.actual_value),
|
|
forecast_value = COALESCE(excluded.forecast_value, ff_calendar.forecast_value),
|
|
previous_value = COALESCE(excluded.previous_value, ff_calendar.previous_value),
|
|
series_id = COALESCE(excluded.series_id, ff_calendar.series_id),
|
|
source = excluded.source,
|
|
fetched_at = datetime('now')
|
|
""",
|
|
batch,
|
|
)
|
|
|
|
|
|
def sync_live() -> Dict[str, Any]:
|
|
"""
|
|
Fetch this week + next week from faireconomy.media and upsert into ff_calendar.
|
|
Updates actual_value when it becomes available (real-time detection).
|
|
"""
|
|
from services.database import get_conn
|
|
|
|
results = {}
|
|
conn = get_conn()
|
|
|
|
for label, url in [("thisweek", _FF_THIS_WEEK_URL), ("nextweek", _FF_NEXT_WEEK_URL)]:
|
|
try:
|
|
resp = httpx.get(url, timeout=15, follow_redirects=True)
|
|
if resp.status_code == 404:
|
|
results[label] = {"status": "not_available"}
|
|
continue
|
|
resp.raise_for_status()
|
|
events = resp.json()
|
|
|
|
batch = []
|
|
for ev in events:
|
|
currency = (ev.get("country") or "").strip().upper()
|
|
if currency not in SUPPORTED_CURRENCIES:
|
|
continue
|
|
|
|
impact = _IMPACT_MAP.get(ev.get("impact") or "", "low")
|
|
if impact == "non-economic":
|
|
continue
|
|
|
|
event_name = (ev.get("title") or "").strip()
|
|
if not event_name:
|
|
continue
|
|
|
|
dt_str = ev.get("date") or ""
|
|
event_date, event_time = _parse_ff_datetime(dt_str)
|
|
|
|
actual = (ev.get("actual") or "").strip() or None
|
|
forecast = (ev.get("forecast") or "").strip() or None
|
|
previous = (ev.get("previous") or "").strip() or None
|
|
|
|
series_id = FF_TO_FRED.get(event_name)
|
|
batch.append((
|
|
event_date, event_time, currency, impact, event_name,
|
|
actual, forecast, previous, series_id, None, "ff_live",
|
|
))
|
|
|
|
if batch:
|
|
_upsert_batch(conn, batch)
|
|
conn.commit()
|
|
|
|
results[label] = {"status": "ok", "events": len(batch)}
|
|
print(f"[FF sync] {label}: {len(batch)} events upserted", flush=True)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"[FF sync] {label} failed: {e}")
|
|
results[label] = {"status": "error", "error": str(e)}
|
|
|
|
conn.close()
|
|
return results
|
|
|
|
|
|
def get_calendar(
|
|
period: str = "recent",
|
|
date_from: Optional[str] = None,
|
|
date_to: Optional[str] = None,
|
|
currencies: Optional[List[str]] = None,
|
|
impacts: Optional[List[str]] = None,
|
|
limit: int = 300,
|
|
offset: int = 0,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Query ff_calendar with period filter.
|
|
|
|
period options:
|
|
recent | today | tomorrow | yesterday | this_week | next_week |
|
|
previous_week | this_month | next_month | previous_month | custom
|
|
When period=custom, date_from and date_to are used directly.
|
|
"""
|
|
from services.database import get_conn
|
|
|
|
today = datetime.now(timezone.utc).date()
|
|
monday = today - timedelta(days=today.weekday())
|
|
sunday = monday + timedelta(days=6)
|
|
|
|
# For custom period, date_from/date_to come from the caller — don't override
|
|
if period != "custom":
|
|
date_from = None
|
|
date_to = None
|
|
|
|
if period == "custom":
|
|
pass # date_from / date_to already set by caller
|
|
elif period == "today":
|
|
date_from = date_to = str(today)
|
|
elif period == "tomorrow":
|
|
t = today + timedelta(days=1)
|
|
date_from = date_to = str(t)
|
|
elif period == "yesterday":
|
|
t = today - timedelta(days=1)
|
|
date_from = date_to = str(t)
|
|
elif period == "this_week":
|
|
date_from = str(monday)
|
|
date_to = str(sunday)
|
|
elif period == "next_week":
|
|
nm = monday + timedelta(weeks=1)
|
|
date_from = str(nm)
|
|
date_to = str(nm + timedelta(days=6))
|
|
elif period == "previous_week":
|
|
pm = monday - timedelta(weeks=1)
|
|
date_from = str(pm)
|
|
date_to = str(pm + timedelta(days=6))
|
|
elif period == "this_month":
|
|
date_from = str(today.replace(day=1))
|
|
# last day of month
|
|
if today.month == 12:
|
|
last = today.replace(year=today.year + 1, month=1, day=1) - timedelta(days=1)
|
|
else:
|
|
last = today.replace(month=today.month + 1, day=1) - timedelta(days=1)
|
|
date_to = str(last)
|
|
elif period == "next_month":
|
|
if today.month == 12:
|
|
first = today.replace(year=today.year + 1, month=1, day=1)
|
|
else:
|
|
first = today.replace(month=today.month + 1, day=1)
|
|
if first.month == 12:
|
|
last = first.replace(year=first.year + 1, month=1, day=1) - timedelta(days=1)
|
|
else:
|
|
last = first.replace(month=first.month + 1, day=1) - timedelta(days=1)
|
|
date_from = str(first)
|
|
date_to = str(last)
|
|
elif period == "previous_month":
|
|
if today.month == 1:
|
|
first = today.replace(year=today.year - 1, month=12, day=1)
|
|
else:
|
|
first = today.replace(month=today.month - 1, day=1)
|
|
last = today.replace(day=1) - timedelta(days=1)
|
|
date_from = str(first)
|
|
date_to = str(last)
|
|
else: # recent: last 7 days + next 14 days
|
|
date_from = str(today - timedelta(days=7))
|
|
date_to = str(today + timedelta(days=14))
|
|
|
|
conn = get_conn()
|
|
where: List[str] = []
|
|
params: List[Any] = []
|
|
|
|
if date_from:
|
|
where.append("event_date >= ?")
|
|
params.append(date_from)
|
|
if date_to:
|
|
where.append("event_date <= ?")
|
|
params.append(date_to)
|
|
|
|
if currencies:
|
|
ph = ",".join("?" * len(currencies))
|
|
where.append(f"currency IN ({ph})")
|
|
params.extend(currencies)
|
|
|
|
if impacts:
|
|
ph = ",".join("?" * len(impacts))
|
|
where.append(f"impact IN ({ph})")
|
|
params.extend(impacts)
|
|
|
|
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
|
|
|
|
count_sql = f"SELECT COUNT(*) FROM ff_calendar {where_sql}"
|
|
data_sql = (
|
|
f"SELECT * FROM ff_calendar {where_sql} "
|
|
f"ORDER BY event_date ASC, event_time ASC "
|
|
f"LIMIT ? OFFSET ?"
|
|
)
|
|
|
|
try:
|
|
total = conn.execute(count_sql, params).fetchone()[0]
|
|
rows = conn.execute(data_sql, params + [limit, offset]).fetchall()
|
|
events = [dict(r) for r in rows]
|
|
return {
|
|
"total": total,
|
|
"offset": offset,
|
|
"limit": limit,
|
|
"period": period,
|
|
"date_from": date_from,
|
|
"date_to": date_to,
|
|
"events": events,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ── FF HTML scraper for upcoming weeks ────────────────────────────────────────
|
|
#
|
|
# Forex Factory renders its calendar server-side; the HTML table is parseable
|
|
# without JavaScript. Times are in US Eastern Time (ET).
|
|
#
|
|
# URL pattern: https://www.forexfactory.com/calendar?week=jun26.2026
|
|
# (any date in the week works; we use each Monday)
|
|
|
|
_FF_BASE = "https://www.forexfactory.com"
|
|
|
|
_FF_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/125.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
"Accept-Encoding": "gzip, deflate, br",
|
|
"Connection": "keep-alive",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Referer": "https://www.forexfactory.com/",
|
|
}
|
|
|
|
# FF impact icon CSS class suffix → our normalized impact label
|
|
_FF_IMPACT_CSS = {
|
|
"red": "high",
|
|
"ora": "medium", # orange
|
|
"yel": "low", # yellow
|
|
"gra": "non-economic",
|
|
}
|
|
|
|
|
|
def _week_url(monday: date) -> str:
|
|
"""Return FF calendar URL for the week containing 'monday'."""
|
|
month = monday.strftime("%b").lower() # "jun", "jul", …
|
|
return f"{_FF_BASE}/calendar?week={month}{monday.day}.{monday.year}"
|
|
|
|
|
|
def _parse_et_time(time_str: str, ref_date: date) -> tuple[str, str]:
|
|
"""
|
|
Parse an FF time string (e.g. '8:30am', 'Tentative', 'All Day')
|
|
expressed in US Eastern Time and return (event_date_utc, event_time_utc).
|
|
"""
|
|
s = (time_str or "").strip().lower()
|
|
if not s or s in ("all day", "tentative", ""):
|
|
return str(ref_date), "00:00"
|
|
try:
|
|
# FF uses "8:30am" / "12:45pm" format
|
|
fmt = "%I:%M%p" if ":" in s else "%I%p"
|
|
t = datetime.strptime(s, fmt)
|
|
dt_et = datetime(
|
|
ref_date.year, ref_date.month, ref_date.day,
|
|
t.hour, t.minute, tzinfo=_ET,
|
|
)
|
|
dt_utc = dt_et.astimezone(timezone.utc)
|
|
return dt_utc.strftime("%Y-%m-%d"), dt_utc.strftime("%H:%M")
|
|
except Exception:
|
|
return str(ref_date), "00:00"
|
|
|
|
|
|
def _parse_impact(impact_td) -> str:
|
|
if impact_td is None:
|
|
return "low"
|
|
span = impact_td.find("span", class_=lambda c: c and "icon--ff-impact" in c)
|
|
if span:
|
|
cls = " ".join(span.get("class", []))
|
|
for suffix, label in _FF_IMPACT_CSS.items():
|
|
if suffix in cls:
|
|
return label
|
|
# Fallback: check title attribute
|
|
title = (impact_td.get("title") or "").lower()
|
|
if "high" in title: return "high"
|
|
if "medium" in title: return "medium"
|
|
return "low"
|
|
|
|
|
|
def _fetch_via_flaresolverr(url: str) -> Optional[str]:
|
|
"""
|
|
Fetch a page through FlareSolverr (a headless-browser proxy that solves Cloudflare's
|
|
JS challenge) instead of a plain HTTP client. FF started 403'ing our direct requests
|
|
— the response is Cloudflare's "Just a moment..." interstitial, which no amount of
|
|
header/cookie spoofing can pass (confirmed: replaying a real logged-in session's
|
|
cf_clearance cookie from a different IP still 403s — it's IP-bound). FlareSolverr runs
|
|
as a sibling container (see deploy/docker-compose.yml) and is reached over the internal
|
|
docker network; not configured (e.g. local dev without that container) → returns None
|
|
so the caller falls back to the direct httpx path unchanged.
|
|
"""
|
|
import os
|
|
base = os.environ.get("FLARESOLVERR_URL", "http://flaresolverr:8191/v1")
|
|
try:
|
|
resp = httpx.post(
|
|
base,
|
|
json={"cmd": "request.get", "url": url, "maxTimeout": 60000},
|
|
timeout=70,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if data.get("status") != "ok":
|
|
print(f"[FF scrape] FlareSolverr error for {url}: {data.get('message')}", flush=True)
|
|
return None
|
|
return data.get("solution", {}).get("response")
|
|
except Exception as e:
|
|
print(f"[FF scrape] FlareSolverr unavailable ({e}) — falling back to direct fetch", flush=True)
|
|
return None
|
|
|
|
|
|
def _scrape_week(client: httpx.Client, monday: date) -> list[tuple]:
|
|
"""Fetch and parse one week of FF calendar. Returns list of row tuples."""
|
|
url = _week_url(monday)
|
|
print(f"[FF scrape] GET {url}", flush=True)
|
|
|
|
html = _fetch_via_flaresolverr(url)
|
|
if html is None:
|
|
try:
|
|
resp = client.get(url, timeout=20)
|
|
resp.raise_for_status()
|
|
html = resp.text
|
|
except Exception as e:
|
|
print(f"[FF scrape] fetch error {url}: {e}", flush=True)
|
|
return []
|
|
|
|
soup = BeautifulSoup(html, "lxml")
|
|
table = soup.find("table", class_="calendar__table")
|
|
if not table:
|
|
# Try alternate selector (FF occasionally restructures)
|
|
table = soup.find("table", attrs={"id": "calendar"})
|
|
if not table:
|
|
print(f"[FF scrape] table not found for {url}", flush=True)
|
|
return []
|
|
|
|
rows = table.find_all("tr", class_="calendar__row")
|
|
batch: list[tuple] = []
|
|
current_date: date = monday # carry forward
|
|
|
|
for tr in rows:
|
|
# ── Date cell (only filled on day-breaker rows) ─────────────────
|
|
date_td = tr.find("td", class_="calendar__date")
|
|
if date_td and date_td.get_text(strip=True):
|
|
raw = date_td.get_text(" ", strip=True) # e.g. "Fri Jun 27"
|
|
for fmt in ("%a %b %d", "%A %b %d", "%a %B %d"):
|
|
try:
|
|
parsed = datetime.strptime(raw, fmt)
|
|
current_date = date(monday.year, parsed.month, parsed.day)
|
|
# Handle year wrap (Dec → Jan)
|
|
if current_date < monday - timedelta(days=6):
|
|
current_date = date(monday.year + 1, parsed.month, parsed.day)
|
|
break
|
|
except ValueError:
|
|
continue
|
|
|
|
# ── Time ────────────────────────────────────────────────────────
|
|
time_td = tr.find("td", class_="calendar__time")
|
|
time_str = time_td.get_text(strip=True) if time_td else ""
|
|
|
|
# ── Currency ────────────────────────────────────────────────────
|
|
cur_td = tr.find("td", class_="calendar__currency")
|
|
currency = cur_td.get_text(strip=True).upper() if cur_td else ""
|
|
if currency not in SUPPORTED_CURRENCIES:
|
|
continue
|
|
|
|
# ── Impact ──────────────────────────────────────────────────────
|
|
impact_td = tr.find("td", class_="calendar__impact")
|
|
impact = _parse_impact(impact_td)
|
|
if impact == "non-economic":
|
|
continue
|
|
|
|
# ── Event name ──────────────────────────────────────────────────
|
|
event_td = tr.find("td", class_="calendar__event")
|
|
if not event_td:
|
|
continue
|
|
title_span = event_td.find("span", class_="calendar__event-title")
|
|
event_name = (title_span or event_td).get_text(strip=True)
|
|
if not event_name:
|
|
continue
|
|
|
|
# ── Values ──────────────────────────────────────────────────────
|
|
actual = _td_val(tr, "calendar__actual")
|
|
forecast = _td_val(tr, "calendar__forecast")
|
|
previous = _td_val(tr, "calendar__previous")
|
|
|
|
# ── Convert time ET → UTC ────────────────────────────────────────
|
|
ev_date, ev_time = _parse_et_time(time_str, current_date)
|
|
|
|
series_id = FF_TO_FRED.get(event_name)
|
|
batch.append((
|
|
ev_date, ev_time, currency, impact, event_name,
|
|
actual or None, forecast or None, previous or None,
|
|
series_id, None, "ff_scrape",
|
|
))
|
|
|
|
return batch
|
|
|
|
|
|
def _td_val(tr, cls: str) -> str:
|
|
td = tr.find("td", class_=cls)
|
|
if not td:
|
|
return ""
|
|
span = td.find("span")
|
|
text = (span or td).get_text(strip=True)
|
|
return text if text not in ("", "\xa0") else ""
|
|
|
|
|
|
def scrape_upcoming(weeks_ahead: int = 5) -> Dict[str, Any]:
|
|
"""
|
|
Scrape FF calendar HTML for the next N weeks and upsert into ff_calendar.
|
|
Includes the current week so we catch events added after the JSON sync.
|
|
"""
|
|
from services.database import get_conn
|
|
|
|
today_date = date.today()
|
|
monday = today_date - timedelta(days=today_date.weekday())
|
|
|
|
total_inserted = 0
|
|
week_results = {}
|
|
|
|
conn = get_conn()
|
|
with httpx.Client(headers=_FF_HEADERS, follow_redirects=True, timeout=30) as client:
|
|
# Warm up session — get cookies from homepage
|
|
try:
|
|
client.get(_FF_BASE, timeout=10)
|
|
except Exception:
|
|
pass
|
|
|
|
for w in range(weeks_ahead):
|
|
week_monday = monday + timedelta(weeks=w)
|
|
batch = _scrape_week(client, week_monday)
|
|
|
|
if batch:
|
|
_upsert_batch(conn, batch)
|
|
conn.commit()
|
|
total_inserted += len(batch)
|
|
|
|
week_results[str(week_monday)] = len(batch)
|
|
print(f"[FF scrape] week {week_monday}: {len(batch)} events", flush=True)
|
|
|
|
conn.close()
|
|
print(f"[FF scrape] total upserted: {total_inserted}", flush=True)
|
|
return {"weeks_scraped": weeks_ahead, "total_upserted": total_inserted, "by_week": week_results}
|
|
|
|
|
|
def sync_historical_range(from_date: date, to_date: Optional[date] = None) -> Dict[str, Any]:
|
|
"""
|
|
Backfill ff_calendar for a historical date range by scraping FF HTML week by week.
|
|
Skips weeks that already have events with actual_value populated (avoids redundant requests).
|
|
Adds a small delay between requests to be respectful of the server.
|
|
Returns progress stats.
|
|
"""
|
|
import time
|
|
from services.database import get_conn
|
|
|
|
if to_date is None:
|
|
to_date = date.today()
|
|
|
|
# Align from_date to the Monday of its week
|
|
monday_start = from_date - timedelta(days=from_date.weekday())
|
|
monday_today = to_date - timedelta(days=to_date.weekday())
|
|
|
|
# Collect all Mondays in range
|
|
mondays: list[date] = []
|
|
cur = monday_start
|
|
while cur <= monday_today:
|
|
mondays.append(cur)
|
|
cur += timedelta(weeks=1)
|
|
|
|
if not mondays:
|
|
return {"weeks_scraped": 0, "total_upserted": 0, "skipped": 0, "by_week": {}}
|
|
|
|
conn = get_conn()
|
|
|
|
# Pre-check which weeks already have actual values to avoid redundant scraping
|
|
# A week is "done" if it has ≥5 events with actual_value for supported currencies
|
|
weeks_with_actuals: set[str] = set()
|
|
for m in mondays:
|
|
sunday = m + timedelta(days=6)
|
|
cnt = conn.execute(
|
|
"""SELECT COUNT(*) FROM ff_calendar
|
|
WHERE event_date >= ? AND event_date <= ?
|
|
AND actual_value IS NOT NULL""",
|
|
(str(m), str(sunday))
|
|
).fetchone()[0]
|
|
if cnt >= 5:
|
|
weeks_with_actuals.add(str(m))
|
|
|
|
total_inserted = 0
|
|
skipped = 0
|
|
week_results: Dict[str, Any] = {}
|
|
|
|
with httpx.Client(headers=_FF_HEADERS, follow_redirects=True, timeout=30) as client:
|
|
# Warm up session
|
|
try:
|
|
client.get(_FF_BASE, timeout=10)
|
|
time.sleep(1.0)
|
|
except Exception:
|
|
pass
|
|
|
|
for monday in mondays:
|
|
monday_str = str(monday)
|
|
if monday_str in weeks_with_actuals:
|
|
week_results[monday_str] = {"status": "skipped", "count": 0}
|
|
skipped += 1
|
|
print(f"[FF backfill] {monday_str}: already populated, skipping", flush=True)
|
|
continue
|
|
|
|
batch = _scrape_week(client, monday)
|
|
|
|
if batch:
|
|
_upsert_batch(conn, batch)
|
|
conn.commit()
|
|
total_inserted += len(batch)
|
|
week_results[monday_str] = {"status": "ok", "count": len(batch)}
|
|
else:
|
|
week_results[monday_str] = {"status": "empty", "count": 0}
|
|
|
|
print(f"[FF backfill] {monday_str}: {len(batch)} events", flush=True)
|
|
time.sleep(1.5) # respectful delay between requests
|
|
|
|
conn.close()
|
|
total_weeks = len(mondays)
|
|
print(f"[FF backfill] done: {total_inserted} events upserted, {skipped}/{total_weeks} weeks skipped", flush=True)
|
|
return {
|
|
"from_date": str(from_date),
|
|
"to_date": str(to_date),
|
|
"weeks_total": total_weeks,
|
|
"weeks_scraped": total_weeks - skipped,
|
|
"weeks_skipped": skipped,
|
|
"total_upserted": total_inserted,
|
|
"by_week": week_results,
|
|
}
|