feat: Trading Economics API for upcoming events with forecasts
- New services/te_calendar.py: fetch_upcoming(weeks_ahead) calls TE API for 9 countries (USD/EUR/GBP/JPY/AUD/CAD/NZD/CHF/CNY), converts to ff_calendar format, upserts with source='te_api' - New endpoints: GET/POST /api/eco/te-key, POST /api/eco/te-sync, GET /api/eco/te-sync/status - Daily scheduler in main.py: FF live sync + TE sync (if key configured) run 60s after startup then every 24h - CalendarPage: TEPanel with key input (password field, Enter to save, "Get free key" link to tradingeconomics.com/api/login), "Sync upcoming (6 weeks)" button with polling FF HTML scraper kept as fallback but TE API is the primary source for upcoming forecasts (no Cloudflare blocking on server IPs). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
@@ -95,21 +95,33 @@ def startup():
|
||||
from services.institutional_scheduler import start_institutional_scheduler
|
||||
start_institutional_scheduler()
|
||||
|
||||
# Daily FF scraper — runs once at startup, then every 24h
|
||||
# Daily calendars sync — FF live (this week) + TE upcoming (if key configured)
|
||||
import threading, time as _time
|
||||
def _ff_scrape_loop():
|
||||
_time.sleep(30) # let server fully boot first
|
||||
def _daily_calendar_sync():
|
||||
_time.sleep(60) # let server fully boot first
|
||||
while True:
|
||||
try:
|
||||
print("[Startup] FF scrape upcoming weeks …", flush=True)
|
||||
from services.ff_calendar import scrape_upcoming
|
||||
result = scrape_upcoming(weeks_ahead=5)
|
||||
print(f"[FF scrape] done: {result.get('total_upserted', 0)} upserted", flush=True)
|
||||
# FF live JSON — this week actuals
|
||||
from services.ff_calendar import sync_live
|
||||
r = sync_live()
|
||||
print(f"[Daily] FF live sync: {r}", flush=True)
|
||||
except Exception as e:
|
||||
print(f"[FF scrape] loop error: {e}", flush=True)
|
||||
_time.sleep(86400) # 24h
|
||||
print(f"[Daily] FF live sync error: {e}", flush=True)
|
||||
|
||||
threading.Thread(target=_ff_scrape_loop, daemon=True).start()
|
||||
try:
|
||||
# Trading Economics — upcoming forecasts (only if key is set)
|
||||
from services.database import get_config
|
||||
te_key = get_config("te_api_key") or ""
|
||||
if te_key:
|
||||
from services.te_calendar import fetch_upcoming
|
||||
r = fetch_upcoming(weeks_ahead=6)
|
||||
print(f"[Daily] TE sync: {r.get('total_upserted', 0)} upserted", flush=True)
|
||||
except Exception as e:
|
||||
print(f"[Daily] TE sync error: {e}", flush=True)
|
||||
|
||||
_time.sleep(86400) # repeat every 24h
|
||||
|
||||
threading.Thread(target=_daily_calendar_sync, daemon=True).start()
|
||||
|
||||
# Auto-import Forex Factory CSV if file is present (force, then delete CSV)
|
||||
import threading
|
||||
|
||||
@@ -296,6 +296,7 @@ def upcoming_events() -> List[Dict[str, Any]]:
|
||||
_ff_import_status: Dict[str, Any] = {"running": False, "last_result": None}
|
||||
_ff_sync_status: Dict[str, Any] = {"running": False, "last_result": None}
|
||||
_ff_scrape_status: Dict[str, Any] = {"running": False, "last_result": None}
|
||||
_te_sync_status: Dict[str, Any] = {"running": False, "last_result": None}
|
||||
|
||||
# CSV search order: Docker image (/app), uploaded to /tmp, local dev paths
|
||||
_FF_CSV_CANDIDATES = [
|
||||
@@ -419,6 +420,55 @@ def ff_scrape_status_ep() -> Dict[str, Any]:
|
||||
return _ff_scrape_status
|
||||
|
||||
|
||||
# ── Trading Economics calendar (upcoming weeks with forecasts) ─────────────────
|
||||
|
||||
@router.get("/te-key")
|
||||
def get_te_key() -> Dict[str, Any]:
|
||||
from services.te_calendar import check_te_key
|
||||
return check_te_key()
|
||||
|
||||
|
||||
@router.post("/te-key")
|
||||
def save_te_key(key: str = Query(..., description="Trading Economics API key")) -> Dict[str, Any]:
|
||||
from services.database import set_config
|
||||
key = key.strip()
|
||||
if not key:
|
||||
raise HTTPException(400, "Empty key")
|
||||
set_config("te_api_key", key)
|
||||
return {"status": "saved", "preview": key[:4] + "…" + key[-4:]}
|
||||
|
||||
|
||||
def _run_te_sync(weeks_ahead: int):
|
||||
global _te_sync_status
|
||||
_te_sync_status["running"] = True
|
||||
try:
|
||||
from services.te_calendar import fetch_upcoming
|
||||
result = fetch_upcoming(weeks_ahead=weeks_ahead)
|
||||
_te_sync_status["last_result"] = result
|
||||
except Exception as e:
|
||||
logger.error(f"[eco/te-sync] Failed: {e}")
|
||||
_te_sync_status["last_result"] = {"error": str(e)}
|
||||
finally:
|
||||
_te_sync_status["running"] = False
|
||||
|
||||
|
||||
@router.post("/te-sync")
|
||||
def te_sync(
|
||||
background_tasks: BackgroundTasks,
|
||||
weeks: int = Query(6, ge=1, le=12),
|
||||
) -> Dict[str, Any]:
|
||||
"""Fetch upcoming economic events + forecasts from Trading Economics API."""
|
||||
if _te_sync_status["running"]:
|
||||
raise HTTPException(409, "TE sync already running")
|
||||
background_tasks.add_task(_run_te_sync, weeks)
|
||||
return {"status": "started", "weeks_ahead": weeks}
|
||||
|
||||
|
||||
@router.get("/te-sync/status")
|
||||
def te_sync_status_ep() -> Dict[str, Any]:
|
||||
return _te_sync_status
|
||||
|
||||
|
||||
@router.get("/calendar")
|
||||
def ff_calendar(
|
||||
period: str = Query("recent", description="recent|today|tomorrow|yesterday|this_week|next_week|previous_week|this_month|next_month|previous_month|custom"),
|
||||
|
||||
160
backend/services/te_calendar.py
Normal file
160
backend/services/te_calendar.py
Normal file
@@ -0,0 +1,160 @@
|
||||
"""
|
||||
Trading Economics calendar service — upcoming events with consensus forecasts.
|
||||
Free API: https://tradingeconomics.com/api/login (500 calls/month)
|
||||
Endpoint: GET https://api.tradingeconomics.com/calendar/country/united%20states?c=KEY
|
||||
"""
|
||||
import logging
|
||||
from datetime import datetime, timezone, timedelta, date
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_TE_BASE = "https://api.tradingeconomics.com"
|
||||
|
||||
# TE country names → our currency codes
|
||||
_TE_COUNTRY_TO_CCY: Dict[str, str] = {
|
||||
"united states": "USD",
|
||||
"euro area": "EUR",
|
||||
"united kingdom":"GBP",
|
||||
"japan": "JPY",
|
||||
"australia": "AUD",
|
||||
"canada": "CAD",
|
||||
"new zealand": "NZD",
|
||||
"switzerland": "CHF",
|
||||
"china": "CNY",
|
||||
}
|
||||
|
||||
# TE importance (1=low, 2=medium, 3=high) → our impact labels
|
||||
_TE_IMPORTANCE = {1: "low", 2: "medium", 3: "high"}
|
||||
|
||||
_COUNTRIES = list(_TE_COUNTRY_TO_CCY.keys())
|
||||
|
||||
|
||||
def _get_te_key() -> str:
|
||||
from services.database import get_config
|
||||
return (get_config("te_api_key") or "").strip()
|
||||
|
||||
|
||||
def check_te_key() -> Dict[str, Any]:
|
||||
key = _get_te_key()
|
||||
return {
|
||||
"configured": bool(key),
|
||||
"preview": (key[:4] + "…" + key[-4:]) if len(key) > 8 else ("***" if key else ""),
|
||||
}
|
||||
|
||||
|
||||
def _parse_te_datetime(dt_str: str) -> tuple[str, str]:
|
||||
"""Parse TE datetime '2026-07-04T12:30:00' (UTC) → (date, time)."""
|
||||
if not dt_str:
|
||||
return "", "00:00"
|
||||
try:
|
||||
# TE returns UTC datetimes
|
||||
dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
|
||||
return dt.strftime("%Y-%m-%d"), dt.strftime("%H:%M")
|
||||
except Exception:
|
||||
return dt_str[:10] if len(dt_str) >= 10 else "", "00:00"
|
||||
|
||||
|
||||
def fetch_upcoming(weeks_ahead: int = 6) -> Dict[str, Any]:
|
||||
"""
|
||||
Fetch upcoming calendar events from Trading Economics API and upsert into ff_calendar.
|
||||
Covers today → today + weeks_ahead weeks.
|
||||
"""
|
||||
from services.database import get_conn
|
||||
from services.ff_calendar import _upsert_batch, FF_TO_FRED, SUPPORTED_CURRENCIES
|
||||
|
||||
key = _get_te_key()
|
||||
if not key:
|
||||
return {"error": "Trading Economics API key not configured. Register free at tradingeconomics.com/api/login"}
|
||||
|
||||
today = date.today()
|
||||
date_to = today + timedelta(weeks=weeks_ahead)
|
||||
|
||||
params = {
|
||||
"c": key,
|
||||
"d1": str(today),
|
||||
"d2": str(date_to),
|
||||
"importance": "1,2,3", # all impact levels
|
||||
}
|
||||
|
||||
total_inserted = 0
|
||||
conn = get_conn()
|
||||
|
||||
for country in _COUNTRIES:
|
||||
ccy = _TE_COUNTRY_TO_CCY[country]
|
||||
try:
|
||||
url = f"{_TE_BASE}/calendar/country/{country.replace(' ', '%20')}"
|
||||
resp = httpx.get(url, params=params, timeout=20, follow_redirects=True)
|
||||
print(f"[TE calendar] {country}: HTTP {resp.status_code}", flush=True)
|
||||
|
||||
if resp.status_code == 401:
|
||||
conn.close()
|
||||
return {"error": "Invalid Trading Economics API key"}
|
||||
if resp.status_code == 403:
|
||||
conn.close()
|
||||
return {"error": "TE API key quota exceeded or invalid plan"}
|
||||
|
||||
resp.raise_for_status()
|
||||
events = resp.json()
|
||||
if not isinstance(events, list):
|
||||
continue
|
||||
|
||||
batch = []
|
||||
for ev in events:
|
||||
event_name = (ev.get("Event") or ev.get("event") or "").strip()
|
||||
if not event_name:
|
||||
continue
|
||||
|
||||
importance = ev.get("Importance") or ev.get("importance") or 1
|
||||
impact = _TE_IMPORTANCE.get(int(importance), "low")
|
||||
|
||||
dt_str = ev.get("Date") or ev.get("date") or ""
|
||||
ev_date, ev_time = _parse_te_datetime(dt_str)
|
||||
if not ev_date:
|
||||
continue
|
||||
|
||||
actual = _fmt_val(ev.get("Actual") or ev.get("actual"))
|
||||
forecast = _fmt_val(ev.get("Forecast") or ev.get("forecast"))
|
||||
previous = _fmt_val(ev.get("Previous") or ev.get("previous"))
|
||||
|
||||
series_id = FF_TO_FRED.get(event_name)
|
||||
|
||||
batch.append((
|
||||
ev_date, ev_time, ccy, impact, event_name,
|
||||
actual or None, forecast or None, previous or None,
|
||||
series_id, None, "te_api",
|
||||
))
|
||||
|
||||
if batch:
|
||||
_upsert_batch(conn, batch)
|
||||
conn.commit()
|
||||
total_inserted += len(batch)
|
||||
print(f"[TE calendar] {country}: {len(batch)} events upserted", flush=True)
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"[TE calendar] {country}: {e}")
|
||||
print(f"[TE calendar] {country} error: {e}", flush=True)
|
||||
|
||||
conn.close()
|
||||
return {
|
||||
"total_upserted": total_inserted,
|
||||
"date_from": str(today),
|
||||
"date_to": str(date_to),
|
||||
"countries": len(_COUNTRIES),
|
||||
}
|
||||
|
||||
|
||||
def _fmt_val(v: Any) -> str:
|
||||
"""Convert TE numeric value to string, return empty if None/NaN."""
|
||||
if v is None:
|
||||
return ""
|
||||
try:
|
||||
f = float(v)
|
||||
if f != f: # NaN check
|
||||
return ""
|
||||
# Format cleanly: no trailing .0 for integers
|
||||
return str(int(f)) if f == int(f) else str(round(f, 4))
|
||||
except (TypeError, ValueError):
|
||||
return str(v).strip()
|
||||
Reference in New Issue
Block a user