feat: calendar

This commit is contained in:
OpenSquared
2026-07-02 11:27:45 +02:00
parent 8098104f5b
commit 5081514898
6 changed files with 286 additions and 118 deletions

View File

@@ -106,7 +106,19 @@ def startup():
from services.institutional_scheduler import start_institutional_scheduler from services.institutional_scheduler import start_institutional_scheduler
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 import threading, time as _time
def _calendar_sync_loop(): def _calendar_sync_loop():
_time.sleep(60) # let server fully boot _time.sleep(60) # let server fully boot

View File

@@ -486,88 +486,56 @@ def fmp_sync_status_ep() -> Dict[str, Any]:
return _te_sync_status 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") _ff_page_status: Dict[str, Any] = {"running": False, "last_result": None}
async def te_html_upload(file: UploadFile = File(...)) -> Dict[str, Any]: _FF_PAGE_TMP = os.path.join(os.path.dirname(__file__), "..", "data", "ff_page.html")
"""Upload calendar_data.txt (TE HTML page) from the browser."""
dest = "/tmp/calendar_data.txt"
@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: try:
content = await file.read() content = await file.read()
with open(dest, "wb") as f: with open(_FF_PAGE_TMP, "wb") as f:
f.write(content) f.write(content)
size_mb = round(len(content) / 1_048_576, 1) size_mb = round(len(content) / 1_048_576, 2)
print(f"[TE HTML upload] Saved {size_mb} MB to {dest}", flush=True) print(f"[FF page upload] Saved {size_mb} MB", flush=True)
return {"status": "uploaded", "path": dest, "size_mb": size_mb} return {"status": "uploaded", "size_mb": size_mb}
except Exception as e: except Exception as e:
raise HTTPException(500, f"Upload failed: {e}") raise HTTPException(500, f"Upload failed: {e}")
def _run_te_html_import(): def _run_ff_page_import():
global _te_html_status global _ff_page_status
_te_html_status["running"] = True _ff_page_status["running"] = True
try: try:
from services.te_html_parser import import_html_file from services.ff_page_parser import import_ff_page_file
path = _find_te_html() result = import_ff_page_file(os.path.abspath(_FF_PAGE_TMP))
if not path: _ff_page_status["last_result"] = result
_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
except Exception as e: except Exception as e:
logger.error(f"[eco/te-html-import] Failed: {e}") logger.error(f"[eco/ff-page-import] Failed: {e}")
_te_html_status["last_result"] = {"error": str(e)} _ff_page_status["last_result"] = {"error": str(e)}
finally: finally:
_te_html_status["running"] = False _ff_page_status["running"] = False
@router.post("/te-html-import") @router.post("/ff-page-import")
def te_html_import(background_tasks: BackgroundTasks) -> Dict[str, Any]: def ff_page_import(background_tasks: BackgroundTasks) -> Dict[str, Any]:
"""Parse calendar_data.txt (TE HTML) and upsert into ff_calendar.""" """Parse uploaded FF calendar HTML and upsert into ff_calendar (ForexFactory format)."""
if _te_html_status["running"]: if _ff_page_status["running"]:
raise HTTPException(409, "TE HTML import already running") raise HTTPException(409, "FF page import already running")
if not _find_te_html(): if not os.path.exists(_FF_PAGE_TMP):
raise HTTPException(404, "calendar_data.txt not found — upload via POST /api/eco/te-html-upload") raise HTTPException(404, "No FF page uploaded yet — POST /api/eco/ff-page-upload first")
background_tasks.add_task(_run_te_html_import) background_tasks.add_task(_run_ff_page_import)
return {"status": "started"} return {"status": "started"}
@router.get("/te-html-import/status") @router.get("/ff-page-import/status")
def te_html_import_status() -> Dict[str, Any]: def ff_page_import_status() -> Dict[str, Any]:
return _te_html_status return _ff_page_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
# ── Unified calendar sync ───────────────────────────────────────────────────── # ── Unified calendar sync ─────────────────────────────────────────────────────
@@ -585,7 +553,7 @@ def calendar_sync_ep(
background_tasks: BackgroundTasks, background_tasks: BackgroundTasks,
weeks: int = Query(8, ge=1, le=12), weeks: int = Query(8, ge=1, le=12),
) -> Dict[str, Any]: ) -> 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 from services.calendar_sync import get_sync_status
if get_sync_status()["running"]: if get_sync_status()["running"]:
raise HTTPException(409, "Calendar sync already running") raise HTTPException(409, "Calendar sync already running")
@@ -688,6 +656,23 @@ def series_history(
conn.close() 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") @router.get("/ff-stats")
def ff_stats() -> Dict[str, Any]: def ff_stats() -> Dict[str, Any]:
"""Quick inventory of ff_calendar table.""" """Quick inventory of ff_calendar table."""

View File

@@ -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. Called by the background loop in main.py and manually via POST /api/eco/calendar-sync.
""" """
import logging import logging
@@ -19,35 +21,38 @@ _sync_status: Dict[str, Any] = {
def calendar_sync(weeks_ahead: int = 8) -> Dict[str, Any]: def calendar_sync(weeks_ahead: int = 8) -> Dict[str, Any]:
""" """
Sync upcoming economic events: Sync upcoming economic events from ForexFactory only:
1. Try FXStreet (no API key, ~300-400 events over weeks_ahead weeks) - FF live JSON covers this week + next week (with actuals as they publish)
2. Fall back to FF live JSON if FXStreet returns 403 or fails - FF HTML scraper covers weeks 3..weeks_ahead for forecasts
""" """
global _sync_status global _sync_status
_sync_status["running"] = True _sync_status["running"] = True
_sync_status["last_run"] = datetime.now(timezone.utc).isoformat() _sync_status["last_run"] = datetime.now(timezone.utc).isoformat()
result: Dict[str, Any] = {} result: Dict[str, Any] = {}
source = "unknown" source = "ff"
try: try:
from services.fxstreet_calendar import fetch_upcoming as fxs_fetch from services.ff_calendar import sync_live, scrape_upcoming
r = fxs_fetch(weeks_ahead=weeks_ahead)
if r.get("error"): # Step 1 — live JSON (thisweek + nextweek) — picks up actuals in real time
logger.warning(f"[calendar_sync] FXStreet error: {r['error']} — falling back to FF live") r_live = sync_live()
from services.ff_calendar import sync_live result["live"] = r_live
r_ff = sync_live()
result = {**r_ff, "_fallback": True, "_fxs_error": r["error"]} # Step 2 — HTML scraper for weeks 3..N ahead (forecasts only, no actuals yet)
source = "ff_fallback" scrape_weeks = max(0, weeks_ahead - 2)
if scrape_weeks > 0:
r_scrape = scrape_upcoming(weeks_ahead=scrape_weeks)
result["scrape"] = r_scrape
else: else:
result = r result["scrape"] = {"skipped": True}
source = "fxstreet"
except Exception as e: except Exception as e:
logger.error(f"[calendar_sync] Exception: {e}") logger.error(f"[calendar_sync] FF sync failed: {e}")
result = {"error": str(e)} result = {"error": str(e)}
source = "error" source = "error"
# After calendar upsert, log any forecast/actual changes to macro_series_log # Log forecast/actual changes to macro_series_log
try: try:
from services.database import get_conn from services.database import get_conn
from services.macro_series_log import scan_and_log_all from services.macro_series_log import scan_and_log_all

View File

@@ -163,6 +163,7 @@ def _upsert_batch(conn, batch: list):
forecast_value = COALESCE(excluded.forecast_value, ff_calendar.forecast_value), forecast_value = COALESCE(excluded.forecast_value, ff_calendar.forecast_value),
previous_value = COALESCE(excluded.previous_value, ff_calendar.previous_value), previous_value = COALESCE(excluded.previous_value, ff_calendar.previous_value),
series_id = COALESCE(excluded.series_id, ff_calendar.series_id), series_id = COALESCE(excluded.series_id, ff_calendar.series_id),
source = excluded.source,
fetched_at = datetime('now') fetched_at = datetime('now')
""", """,
batch, batch,

View File

@@ -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

View File

@@ -167,9 +167,9 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () =>
const [syncMsg, setSyncMsg] = useState<string | null>(null) const [syncMsg, setSyncMsg] = useState<string | null>(null)
const calPollRef = useRef<ReturnType<typeof setInterval> | null>(null) const calPollRef = useRef<ReturnType<typeof setInterval> | null>(null)
const [teHtmlMsg, setTeHtmlMsg] = useState<string | null>(null) const [ffPageMsg, setFfPageMsg] = useState<string | null>(null)
const [teHtmlBusy, setTeHtmlBusy] = useState(false) const [ffPageBusy, setFfPageBusy] = useState(false)
const teHtmlPollRef = useRef<ReturnType<typeof setInterval> | null>(null) const ffPagePollRef = useRef<ReturnType<typeof setInterval> | null>(null)
const fileInputId = useId() const fileInputId = useId()
useEffect(() => { useEffect(() => {
@@ -193,7 +193,7 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () =>
return () => { return () => {
if (pollRef.current) clearInterval(pollRef.current) if (pollRef.current) clearInterval(pollRef.current)
if (calPollRef.current) clearInterval(calPollRef.current) if (calPollRef.current) clearInterval(calPollRef.current)
if (teHtmlPollRef.current) clearInterval(teHtmlPollRef.current) if (ffPagePollRef.current) clearInterval(ffPagePollRef.current)
} }
}, [onImported]) }, [onImported])
@@ -210,12 +210,11 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () =>
const r = s.last_result || {} const r = s.last_result || {}
if (r.error) { if (r.error) {
setSyncMsg(`Erreur: ${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 { } 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() onImported()
} }
@@ -223,35 +222,35 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () =>
} catch { setSyncMsg('sync error'); setCalStatus(s => ({ ...s, running: false })) } } catch { setSyncMsg('sync error'); setCalStatus(s => ({ ...s, running: false })) }
} }
const uploadAndImportTeHtml = async (file: File) => { const uploadAndImportFfPage = async (file: File) => {
setTeHtmlBusy(true) setFfPageBusy(true)
setTeHtmlMsg('uploading…') setFfPageMsg('uploading…')
try { try {
const form = new FormData() const form = new FormData()
form.append('file', file) form.append('file', file)
const up = await fetch(`${API}/api/eco/te-html-upload`, { method: 'POST', body: form }).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) { setTeHtmlMsg(`Upload error: ${up.error}`); setTeHtmlBusy(false); return } if (up.error) { setFfPageMsg(`Upload error: ${up.error}`); setFfPageBusy(false); return }
setTeHtmlMsg(`uploaded (${up.size_mb} MB) — importing…`) setFfPageMsg(`uploaded (${up.size_mb} MB) — importing…`)
await fetch(`${API}/api/eco/te-html-import`, { method: 'POST' }) await fetch(`${API}/api/eco/ff-page-import`, { method: 'POST' })
teHtmlPollRef.current = setInterval(async () => { ffPagePollRef.current = setInterval(async () => {
const s = await fetch(`${API}/api/eco/te-html-import/status`).then(x => x.json()) const s = await fetch(`${API}/api/eco/ff-page-import/status`).then(x => x.json())
if (!s.running) { if (!s.running) {
clearInterval(teHtmlPollRef.current!) clearInterval(ffPagePollRef.current!)
setTeHtmlBusy(false) setFfPageBusy(false)
const r = s.last_result || {} const r = s.last_result || {}
if (r.error) { setTeHtmlMsg(`Error: ${r.error}`); return } if (r.error) { setFfPageMsg(`Error: ${r.error}`); return }
setTeHtmlMsg(`TE HTML: ${r.total} events (${r.date_from}${r.date_to})`) setFfPageMsg(`FF Page: ${r.inserted} events (${r.date_from}${r.date_to})`)
onImported() onImported()
} }
}, 2000) }, 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 hasData = stats.total > 0
const res = importStatus.last_result 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 ( return (
<div className="bg-slate-800 border border-slate-700 rounded-lg p-4 mb-4 flex flex-wrap items-center gap-4 text-sm"> <div className="bg-slate-800 border border-slate-700 rounded-lg p-4 mb-4 flex flex-wrap items-center gap-4 text-sm">
@@ -284,27 +283,27 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () =>
{calStatus.running ? 'Syncing…' : 'Sync Now'} {calStatus.running ? 'Syncing…' : 'Sync Now'}
</button> </button>
{/* TE HTML import */} {/* FF Page import — save forexfactory.com/calendar as HTML (Ctrl+U) and upload */}
<label <label
htmlFor={fileInputId} htmlFor={fileInputId}
className={clsx( className={clsx(
'px-3 py-1.5 rounded flex items-center gap-1.5 text-white cursor-pointer', 'px-3 py-1.5 rounded flex items-center gap-1.5 text-white cursor-pointer',
teHtmlBusy ffPageBusy
? 'bg-amber-800 opacity-50 cursor-not-allowed' ? 'bg-amber-800 opacity-50 cursor-not-allowed'
: 'bg-amber-700 hover:bg-amber-600' : 'bg-amber-700 hover:bg-amber-600'
)} )}
title="Import Trading Economics HTML page (calendar_data.txt) — 1000 events with forecasts" title="Import Trading Economics HTML page (calendar_data.txt) — 1000 events with forecasts"
> >
<Upload size={13} /> <Upload size={13} />
Import TE HTML Import FF Page
</label> </label>
<input <input
id={fileInputId} id={fileInputId}
type="file" type="file"
accept=".txt,.html" accept=".txt,.html"
className="hidden" className="hidden"
disabled={teHtmlBusy} disabled={ffPageBusy}
onChange={e => { const f = e.target.files?.[0]; if (f) uploadAndImportTeHtml(f); e.target.value = '' }} onChange={e => { const f = e.target.files?.[0]; if (f) uploadAndImportFfPage(f); e.target.value = '' }}
/> />
{res && !importStatus.running && ( {res && !importStatus.running && (
@@ -317,9 +316,9 @@ function ImportPanel({ stats, onImported }: { stats: FFStats; onImported: () =>
{syncMsg} {syncMsg}
</span> </span>
)} )}
{teHtmlMsg && ( {ffPageMsg && (
<span className={clsx('text-xs', teHtmlMsg.startsWith('✓') ? 'text-emerald-400' : teHtmlMsg.startsWith('Error') ? 'text-red-400' : 'text-yellow-400')}> <span className={clsx('text-xs', ffPageMsg.startsWith('✓') ? 'text-emerald-400' : ffPageMsg.startsWith('Error') ? 'text-red-400' : 'text-yellow-400')}>
{teHtmlMsg} {ffPageMsg}
</span> </span>
)} )}
</div> </div>