From 5081514898340c274b785d441b756f3516ed7594 Mon Sep 17 00:00:00 2001 From: OpenSquared Date: Thu, 2 Jul 2026 11:27:45 +0200 Subject: [PATCH] feat: calendar --- backend/main.py | 14 ++- backend/routers/eco.py | 119 +++++++++----------- backend/services/calendar_sync.py | 39 ++++--- backend/services/ff_calendar.py | 1 + backend/services/ff_page_parser.py | 166 ++++++++++++++++++++++++++++ frontend/src/pages/CalendarPage.tsx | 65 ++++++----- 6 files changed, 286 insertions(+), 118 deletions(-) create mode 100644 backend/services/ff_page_parser.py diff --git a/backend/main.py b/backend/main.py index 4d99d86..a01c6ca 100644 --- a/backend/main.py +++ b/backend/main.py @@ -106,7 +106,19 @@ def startup(): from services.institutional_scheduler import start_institutional_scheduler start_institutional_scheduler() - # Calendar sync loop — FXStreet primary, FF fallback; interval from calendar_refresh_h config (default 6h) + # One-time migration: remove FXStreet-sourced rows (decimal format) — FF sync below will refill from FF + try: + from services.database import get_conn as _get_conn + _mc = _get_conn() + _deleted = _mc.execute("DELETE FROM ff_calendar WHERE source='fxstreet'").rowcount + _mc.commit() + _mc.close() + if _deleted: + print(f"[Startup] Purged {_deleted} FXStreet rows — will refill from FF on next calendar sync", flush=True) + except Exception as _e: + print(f"[Startup] FXStreet purge skipped: {_e}", flush=True) + + # Calendar sync loop — FF only (live JSON + HTML scraper); interval from calendar_refresh_h config (default 6h) import threading, time as _time def _calendar_sync_loop(): _time.sleep(60) # let server fully boot diff --git a/backend/routers/eco.py b/backend/routers/eco.py index dd7c5f7..150f04c 100644 --- a/backend/routers/eco.py +++ b/backend/routers/eco.py @@ -486,88 +486,56 @@ def fmp_sync_status_ep() -> Dict[str, Any]: return _te_sync_status -# ── Trading Economics HTML import ───────────────────────────────────────────── +# ── ForexFactory page HTML import ───────────────────────────────────────────── +# User saves the FF calendar page source (Ctrl+U → Save As, or copy-paste HTML) +# and uploads it. We extract window.calendarComponentStates[1].days JSON. -@router.post("/te-html-upload") -async def te_html_upload(file: UploadFile = File(...)) -> Dict[str, Any]: - """Upload calendar_data.txt (TE HTML page) from the browser.""" - dest = "/tmp/calendar_data.txt" +_ff_page_status: Dict[str, Any] = {"running": False, "last_result": None} +_FF_PAGE_TMP = os.path.join(os.path.dirname(__file__), "..", "data", "ff_page.html") + + +@router.post("/ff-page-upload") +async def ff_page_upload(file: UploadFile = File(...)) -> Dict[str, Any]: + """Upload the ForexFactory calendar page HTML (Ctrl+U → Save As, any week).""" try: content = await file.read() - with open(dest, "wb") as f: + with open(_FF_PAGE_TMP, "wb") as f: f.write(content) - size_mb = round(len(content) / 1_048_576, 1) - print(f"[TE HTML upload] Saved {size_mb} MB to {dest}", flush=True) - return {"status": "uploaded", "path": dest, "size_mb": size_mb} + size_mb = round(len(content) / 1_048_576, 2) + print(f"[FF page upload] Saved {size_mb} MB", flush=True) + return {"status": "uploaded", "size_mb": size_mb} except Exception as e: raise HTTPException(500, f"Upload failed: {e}") -def _run_te_html_import(): - global _te_html_status - _te_html_status["running"] = True +def _run_ff_page_import(): + global _ff_page_status + _ff_page_status["running"] = True try: - from services.te_html_parser import import_html_file - path = _find_te_html() - if not path: - _te_html_status["last_result"] = {"error": "calendar_data.txt not found — upload first"} - return - result = import_html_file(path) - _te_html_status["last_result"] = result + from services.ff_page_parser import import_ff_page_file + result = import_ff_page_file(os.path.abspath(_FF_PAGE_TMP)) + _ff_page_status["last_result"] = result except Exception as e: - logger.error(f"[eco/te-html-import] Failed: {e}") - _te_html_status["last_result"] = {"error": str(e)} + logger.error(f"[eco/ff-page-import] Failed: {e}") + _ff_page_status["last_result"] = {"error": str(e)} finally: - _te_html_status["running"] = False + _ff_page_status["running"] = False -@router.post("/te-html-import") -def te_html_import(background_tasks: BackgroundTasks) -> Dict[str, Any]: - """Parse calendar_data.txt (TE HTML) and upsert into ff_calendar.""" - if _te_html_status["running"]: - raise HTTPException(409, "TE HTML import already running") - if not _find_te_html(): - raise HTTPException(404, "calendar_data.txt not found — upload via POST /api/eco/te-html-upload") - background_tasks.add_task(_run_te_html_import) +@router.post("/ff-page-import") +def ff_page_import(background_tasks: BackgroundTasks) -> Dict[str, Any]: + """Parse uploaded FF calendar HTML and upsert into ff_calendar (ForexFactory format).""" + if _ff_page_status["running"]: + raise HTTPException(409, "FF page import already running") + if not os.path.exists(_FF_PAGE_TMP): + raise HTTPException(404, "No FF page uploaded yet — POST /api/eco/ff-page-upload first") + background_tasks.add_task(_run_ff_page_import) return {"status": "started"} -@router.get("/te-html-import/status") -def te_html_import_status() -> Dict[str, Any]: - return _te_html_status - - -# ── FXStreet calendar (no API key required) ─────────────────────────────────── - -def _run_fxs_sync(weeks_ahead: int): - global _fxs_sync_status - _fxs_sync_status["running"] = True - try: - from services.fxstreet_calendar import fetch_upcoming - result = fetch_upcoming(weeks_ahead=weeks_ahead) - _fxs_sync_status["last_result"] = result - except Exception as e: - logger.error(f"[eco/fxs-sync] Failed: {e}") - _fxs_sync_status["last_result"] = {"error": str(e)} - finally: - _fxs_sync_status["running"] = False - - -@router.post("/fxs-sync") -def fxs_sync( - background_tasks: BackgroundTasks, - weeks: int = Query(6, ge=1, le=12), -) -> Dict[str, Any]: - """Fetch upcoming events + forecasts from FXStreet (no API key needed).""" - if _fxs_sync_status["running"]: - raise HTTPException(409, "FXStreet sync already running") - background_tasks.add_task(_run_fxs_sync, weeks) - return {"status": "started", "weeks_ahead": weeks} - - -@router.get("/fxs-sync/status") -def fxs_sync_status_ep() -> Dict[str, Any]: - return _fxs_sync_status +@router.get("/ff-page-import/status") +def ff_page_import_status() -> Dict[str, Any]: + return _ff_page_status # ── Unified calendar sync ───────────────────────────────────────────────────── @@ -585,7 +553,7 @@ def calendar_sync_ep( background_tasks: BackgroundTasks, weeks: int = Query(8, ge=1, le=12), ) -> Dict[str, Any]: - """Trigger unified calendar sync (FXStreet primary → FF fallback) in background.""" + """Trigger unified calendar sync (FF live JSON thisweek+nextweek + FF HTML scraper) in background.""" from services.calendar_sync import get_sync_status if get_sync_status()["running"]: raise HTTPException(409, "Calendar sync already running") @@ -688,6 +656,23 @@ def series_history( conn.close() +@router.post("/ff-purge-fxstreet") +def ff_purge_fxstreet() -> Dict[str, Any]: + """ + Delete all rows sourced from FXStreet (source='fxstreet'). + Run once to clean up corrupted/decimal-format data, then POST /calendar-sync to refill from FF. + """ + from services.database import get_conn + conn = get_conn() + try: + count = conn.execute("SELECT COUNT(*) FROM ff_calendar WHERE source='fxstreet'").fetchone()[0] + conn.execute("DELETE FROM ff_calendar WHERE source='fxstreet'") + conn.commit() + return {"deleted": count, "next_step": "POST /api/eco/calendar-sync to refill from ForexFactory"} + finally: + conn.close() + + @router.get("/ff-stats") def ff_stats() -> Dict[str, Any]: """Quick inventory of ff_calendar table.""" diff --git a/backend/services/calendar_sync.py b/backend/services/calendar_sync.py index d78bae4..8192a6c 100644 --- a/backend/services/calendar_sync.py +++ b/backend/services/calendar_sync.py @@ -1,5 +1,7 @@ """ -Unified calendar sync — FXStreet (primary) with FF live fallback. +Unified calendar sync — ForexFactory only. + 1. FF live JSON (nfs.faireconomy.media) → this week + next week (actual values, precise) + 2. FF HTML scraper → weeks 3..N ahead (upcoming forecasts) Called by the background loop in main.py and manually via POST /api/eco/calendar-sync. """ import logging @@ -19,35 +21,38 @@ _sync_status: Dict[str, Any] = { def calendar_sync(weeks_ahead: int = 8) -> Dict[str, Any]: """ - Sync upcoming economic events: - 1. Try FXStreet (no API key, ~300-400 events over weeks_ahead weeks) - 2. Fall back to FF live JSON if FXStreet returns 403 or fails + Sync upcoming economic events from ForexFactory only: + - FF live JSON covers this week + next week (with actuals as they publish) + - FF HTML scraper covers weeks 3..weeks_ahead for forecasts """ global _sync_status _sync_status["running"] = True _sync_status["last_run"] = datetime.now(timezone.utc).isoformat() result: Dict[str, Any] = {} - source = "unknown" + source = "ff" try: - from services.fxstreet_calendar import fetch_upcoming as fxs_fetch - r = fxs_fetch(weeks_ahead=weeks_ahead) - if r.get("error"): - logger.warning(f"[calendar_sync] FXStreet error: {r['error']} — falling back to FF live") - from services.ff_calendar import sync_live - r_ff = sync_live() - result = {**r_ff, "_fallback": True, "_fxs_error": r["error"]} - source = "ff_fallback" + from services.ff_calendar import sync_live, scrape_upcoming + + # Step 1 — live JSON (thisweek + nextweek) — picks up actuals in real time + r_live = sync_live() + result["live"] = r_live + + # Step 2 — HTML scraper for weeks 3..N ahead (forecasts only, no actuals yet) + scrape_weeks = max(0, weeks_ahead - 2) + if scrape_weeks > 0: + r_scrape = scrape_upcoming(weeks_ahead=scrape_weeks) + result["scrape"] = r_scrape else: - result = r - source = "fxstreet" + result["scrape"] = {"skipped": True} + except Exception as e: - logger.error(f"[calendar_sync] Exception: {e}") + logger.error(f"[calendar_sync] FF sync failed: {e}") result = {"error": str(e)} source = "error" - # After calendar upsert, log any forecast/actual changes to macro_series_log + # Log forecast/actual changes to macro_series_log try: from services.database import get_conn from services.macro_series_log import scan_and_log_all diff --git a/backend/services/ff_calendar.py b/backend/services/ff_calendar.py index 518484e..c8aedf8 100644 --- a/backend/services/ff_calendar.py +++ b/backend/services/ff_calendar.py @@ -163,6 +163,7 @@ def _upsert_batch(conn, batch: list): 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, diff --git a/backend/services/ff_page_parser.py b/backend/services/ff_page_parser.py new file mode 100644 index 0000000..460cbaa --- /dev/null +++ b/backend/services/ff_page_parser.py @@ -0,0 +1,166 @@ +""" +Parser for ForexFactory calendar page source (HTML). +FF embeds a rich JSON object directly in the page: + window.calendarComponentStates[1] = { days: [...] } + +Each event has: dateline (unix ts UTC), currency (ISO), name, impactName, +actual, forecast, previous, revision. + +Usage: + rows = parse_ff_page(html_string) + # → list of tuples ready for ff_calendar._upsert_batch() +""" +import json +import logging +import re +from datetime import datetime, timezone + +logger = logging.getLogger(__name__) + +# Currencies we care about (same set as ff_calendar.py) +_SUPPORTED = {"USD", "EUR", "GBP", "JPY", "AUD", "CAD", "NZD", "CHF", "CNY"} + +# Impact labels FF uses (impactName field in the embedded JSON) +_IMPACT_OK = {"high", "medium", "low"} + + +def _extract_days_json(html: str) -> list: + """ + Pull the value of `days` array from window.calendarComponentStates[1]. + Returns parsed list or raises ValueError. + """ + # Find the start of the days array + m = re.search(r'window\.calendarComponentStates\s*\[\s*1\s*\]\s*=\s*\{\s*days\s*:\s*(\[)', html) + if not m: + raise ValueError("window.calendarComponentStates[1] not found in HTML") + + start = m.start(1) + depth = 0 + end = start + in_string = False + escape = False + + for i in range(start, len(html)): + ch = html[i] + if escape: + escape = False + continue + if ch == '\\' and in_string: + escape = True + continue + if ch == '"': + in_string = not in_string + continue + if in_string: + continue + if ch == '[': + depth += 1 + elif ch == ']': + depth -= 1 + if depth == 0: + end = i + 1 + break + + raw = html[start:end] + return json.loads(raw) + + +def parse_ff_page(html: str) -> tuple[list, dict]: + """ + Parse FF calendar HTML page source. + Returns (rows, stats) where rows are tuples for _upsert_batch(). + """ + days = _extract_days_json(html) + + rows = [] + skipped = 0 + date_min = date_max = None + + for day in days: + for ev in day.get("events", []): + dateline = ev.get("dateline") + if not dateline: + skipped += 1 + continue + + currency = (ev.get("currency") or "").strip() + if currency not in _SUPPORTED: + skipped += 1 + continue + + impact = (ev.get("impactName") or "low").lower() + if impact not in _IMPACT_OK: + skipped += 1 + continue + + name = (ev.get("name") or "").strip() + if not name: + skipped += 1 + continue + + dt = datetime.fromtimestamp(int(dateline), tz=timezone.utc) + event_date = dt.strftime("%Y-%m-%d") + # "All Day" events: FF sets timeMasked=true — normalize to "00:00" + # so they match rows inserted by the HTML scraper (same convention) + event_time = "00:00" if ev.get("timeMasked") else dt.strftime("%H:%M") + + if date_min is None or event_date < date_min: + date_min = event_date + if date_max is None or event_date > date_max: + date_max = event_date + + actual = (ev.get("actual") or "").strip() or None + forecast = (ev.get("forecast") or "").strip() or None + revision = (ev.get("revision") or "").strip() + previous = revision or (ev.get("previous") or "").strip() or None + + rows.append(( + event_date, event_time, currency, impact, name, + actual, forecast, previous, + None, None, "ff_page", + )) + + stats = { + "total": len(rows) + skipped, + "inserted": len(rows), + "skipped": skipped, + "date_from": date_min, + "date_to": date_max, + } + return rows, stats + + +def import_ff_page_file(path: str) -> dict: + """Read an FF calendar HTML file and upsert into ff_calendar.""" + from services.database import get_conn + from services.ff_calendar import _upsert_batch + + try: + with open(path, encoding="utf-8", errors="replace") as f: + html = f.read() + except Exception as e: + return {"error": f"Cannot read file: {e}"} + + try: + rows, stats = parse_ff_page(html) + except ValueError as e: + return {"error": str(e)} + except Exception as e: + logger.error(f"[ff_page_parser] parse error: {e}") + return {"error": f"Parse error: {e}"} + + if not rows: + return {**stats, "error": "No usable events found — check the HTML is a full FF calendar page"} + + conn = get_conn() + try: + _upsert_batch(conn, rows) + conn.commit() + logger.info(f"[ff_page_parser] {stats['inserted']} rows upserted ({stats['date_from']} → {stats['date_to']})") + except Exception as e: + logger.error(f"[ff_page_parser] DB error: {e}") + return {**stats, "error": f"DB error: {e}"} + finally: + conn.close() + + return stats diff --git a/frontend/src/pages/CalendarPage.tsx b/frontend/src/pages/CalendarPage.tsx index 8b9c378..97368eb 100644 --- a/frontend/src/pages/CalendarPage.tsx +++ b/frontend/src/pages/CalendarPage.tsx @@ -167,9 +167,9 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () => const [syncMsg, setSyncMsg] = useState(null) const calPollRef = useRef | null>(null) - const [teHtmlMsg, setTeHtmlMsg] = useState(null) - const [teHtmlBusy, setTeHtmlBusy] = useState(false) - const teHtmlPollRef = useRef | null>(null) + const [ffPageMsg, setFfPageMsg] = useState(null) + const [ffPageBusy, setFfPageBusy] = useState(false) + const ffPagePollRef = useRef | null>(null) const fileInputId = useId() useEffect(() => { @@ -193,7 +193,7 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () => return () => { if (pollRef.current) clearInterval(pollRef.current) if (calPollRef.current) clearInterval(calPollRef.current) - if (teHtmlPollRef.current) clearInterval(teHtmlPollRef.current) + if (ffPagePollRef.current) clearInterval(ffPagePollRef.current) } }, [onImported]) @@ -210,12 +210,11 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () => const r = s.last_result || {} if (r.error) { setSyncMsg(`Erreur: ${r.error}`) - } else if (s.source === 'ff_fallback') { - const tw = r.thisweek?.events ?? '?' - const nw = r.nextweek?.events ?? '?' - setSyncMsg(`✓ FF fallback: ${tw}+${nw} events`) } else { - setSyncMsg(`✓ FXStreet: ${r.total_upserted} events`) + const tw = r.live?.thisweek?.events ?? r.live?.events ?? '?' + const nw = r.live?.nextweek?.events ?? '?' + const sc = r.scrape?.total_upserted ?? 0 + setSyncMsg(`✓ FF: ${tw}+${nw} live, ${sc} upcoming`) } onImported() } @@ -223,35 +222,35 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () => } catch { setSyncMsg('sync error'); setCalStatus(s => ({ ...s, running: false })) } } - const uploadAndImportTeHtml = async (file: File) => { - setTeHtmlBusy(true) - setTeHtmlMsg('uploading…') + const uploadAndImportFfPage = async (file: File) => { + setFfPageBusy(true) + setFfPageMsg('uploading…') try { const form = new FormData() form.append('file', file) - const up = await fetch(`${API}/api/eco/te-html-upload`, { method: 'POST', body: form }).then(x => x.json()) - if (up.error) { setTeHtmlMsg(`Upload error: ${up.error}`); setTeHtmlBusy(false); return } - setTeHtmlMsg(`uploaded (${up.size_mb} MB) — importing…`) - await fetch(`${API}/api/eco/te-html-import`, { method: 'POST' }) - teHtmlPollRef.current = setInterval(async () => { - const s = await fetch(`${API}/api/eco/te-html-import/status`).then(x => x.json()) + const up = await fetch(`${API}/api/eco/ff-page-upload`, { method: 'POST', body: form }).then(x => x.json()) + if (up.error) { setFfPageMsg(`Upload error: ${up.error}`); setFfPageBusy(false); return } + setFfPageMsg(`uploaded (${up.size_mb} MB) — importing…`) + await fetch(`${API}/api/eco/ff-page-import`, { method: 'POST' }) + ffPagePollRef.current = setInterval(async () => { + const s = await fetch(`${API}/api/eco/ff-page-import/status`).then(x => x.json()) if (!s.running) { - clearInterval(teHtmlPollRef.current!) - setTeHtmlBusy(false) + clearInterval(ffPagePollRef.current!) + setFfPageBusy(false) const r = s.last_result || {} - if (r.error) { setTeHtmlMsg(`Error: ${r.error}`); return } - setTeHtmlMsg(`✓ TE HTML: ${r.total} events (${r.date_from} → ${r.date_to})`) + if (r.error) { setFfPageMsg(`Error: ${r.error}`); return } + setFfPageMsg(`✓ FF Page: ${r.inserted} events (${r.date_from} → ${r.date_to})`) onImported() } }, 2000) - } catch { setTeHtmlBusy(false); setTeHtmlMsg('error') } + } catch { setFfPageBusy(false); setFfPageMsg('error') } } - useEffect(() => () => { if (teHtmlPollRef.current) clearInterval(teHtmlPollRef.current) }, []) + useEffect(() => () => { if (ffPagePollRef.current) clearInterval(ffPagePollRef.current) }, []) const hasData = stats.total > 0 const res = importStatus.last_result - const srcLabel = calStatus.source === 'ff_fallback' ? 'FF fallback' : calStatus.source === 'fxstreet' ? 'FXStreet' : null + const srcLabel = calStatus.source === 'ff' ? 'ForexFactory' : calStatus.source ?? null return (
@@ -284,27 +283,27 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () => {calStatus.running ? 'Syncing…' : 'Sync Now'} - {/* TE HTML import */} + {/* FF Page import — save forexfactory.com/calendar as HTML (Ctrl+U) and upload */} { const f = e.target.files?.[0]; if (f) uploadAndImportTeHtml(f); e.target.value = '' }} + disabled={ffPageBusy} + onChange={e => { const f = e.target.files?.[0]; if (f) uploadAndImportFfPage(f); e.target.value = '' }} /> {res && !importStatus.running && ( @@ -317,9 +316,9 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () => {syncMsg} )} - {teHtmlMsg && ( - - {teHtmlMsg} + {ffPageMsg && ( + + {ffPageMsg} )}