Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)
da0ce7b verified | #!/usr/bin/env python3 | |
| """ | |
| Ingest Federal Reserve Economic Data (FRED) into Aureon's global history DB. | |
| Uses the free FRED API (fredapi library) and stores data in the | |
| macro_indicators table. Supports resumable watermarks so re-runs only | |
| fetch new observations. | |
| Usage examples: | |
| # Ingest all default series (full history back to 1970) | |
| python scripts/python/ingest_fred.py --fred-key YOUR_KEY | |
| # Ingest only interest-rate series | |
| python scripts/python/ingest_fred.py --category rates | |
| # Ingest specific series | |
| python scripts/python/ingest_fred.py --series FEDFUNDS,UNRATE,CPIAUCSL | |
| # Fresh start (ignore watermarks) | |
| python scripts/python/ingest_fred.py --no-resume | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import sys | |
| import time | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any, Dict, List, Optional | |
| # --------------------------------------------------------------------------- | |
| # Repo-root bootstrap (same pattern as other scripts/python/*.py) | |
| # --------------------------------------------------------------------------- | |
| _REPO_ROOT = Path(__file__).resolve().parents[2] | |
| if str(_REPO_ROOT) not in sys.path: | |
| sys.path.insert(0, str(_REPO_ROOT)) | |
| try: | |
| from dotenv import load_dotenv | |
| load_dotenv(_REPO_ROOT / ".env") | |
| except Exception: | |
| pass | |
| from aureon.core import aureon_global_history_db as ghdb | |
| # --------------------------------------------------------------------------- | |
| # Default FRED series organised by category | |
| # --------------------------------------------------------------------------- | |
| SERIES_BY_CATEGORY: Dict[str, List[str]] = { | |
| "rates": [ | |
| "FEDFUNDS", # Fed Funds Rate | |
| "DFF", # Daily Fed Funds | |
| "DFEDTARU", # Fed Funds Upper Target | |
| "DGS1", # 1-Year Treasury | |
| "DGS2", # 2-Year Treasury | |
| "DGS5", # 5-Year Treasury | |
| "DGS10", # 10-Year Treasury | |
| "DGS30", # 30-Year Treasury | |
| "T10Y2Y", # 10Y-2Y Spread (Yield Curve) | |
| "T10Y3M", # 10Y-3M Spread | |
| "BAMLH0A0HYM2", # High Yield Spread | |
| "AAA", # Moody's AAA Corporate Bond Yield | |
| "BAA", # Moody's BAA Corporate Bond Yield | |
| ], | |
| "gdp": [ | |
| "GDP", # Nominal GDP | |
| "GDPC1", # Real GDP | |
| "A191RL1Q225SBEA", # Real GDP % Change | |
| "GDPPOT", # Potential GDP | |
| ], | |
| "inflation": [ | |
| "CPIAUCSL", # CPI All Urban Consumers | |
| "CPILFESL", # Core CPI (less food & energy) | |
| "PCEPI", # PCE Price Index | |
| "PCEPILFE", # Core PCE | |
| "CPALTT01USM657N", # CPI Year-over-Year | |
| "GOLDAMGBD228NLBM", # Gold Price (London Fix) | |
| "DCOILWTICO", # WTI Crude Oil | |
| ], | |
| "employment": [ | |
| "UNRATE", # Unemployment Rate | |
| "PAYEMS", # Nonfarm Payrolls | |
| "ICSA", # Initial Jobless Claims | |
| "JTSJOL", # Job Openings (JOLTS) | |
| "U6RATE", # U-6 Unemployment | |
| ], | |
| "money": [ | |
| "M2SL", # M2 Money Supply | |
| "WALCL", # Fed Balance Sheet | |
| "TOTRESNS", # Total Reserves | |
| ], | |
| "housing": [ | |
| "CSUSHPINSA", # Case-Shiller Home Price Index | |
| "HOUST", # Housing Starts | |
| "MORTGAGE30US", # 30-Year Mortgage Rate | |
| ], | |
| "consumer": [ | |
| "UMCSENT", # Michigan Consumer Sentiment | |
| "RSAFS", # Retail Sales | |
| "INDPRO", # Industrial Production | |
| "DGORDER", # Durable Goods Orders | |
| ], | |
| "financial": [ | |
| "VIXCLS", # VIX | |
| "DTWEXBGS", # Trade Weighted USD Index | |
| "SP500", # S&P 500 | |
| "NASDAQCOM", # NASDAQ Composite | |
| "DEXUSEU", # EUR/USD | |
| "DEXJPUS", # JPY/USD | |
| "DEXUSUK", # GBP/USD | |
| ], | |
| "global": [ | |
| "GFDEBTN", # Federal Debt Total | |
| "FYFSD", # Federal Surplus/Deficit | |
| "STLFSI4", # St. Louis Financial Stress Index | |
| ], | |
| } | |
| # Reverse lookup: series_id -> category | |
| _SERIES_CATEGORY: Dict[str, str] = {} | |
| for _cat, _ids in SERIES_BY_CATEGORY.items(): | |
| for _sid in _ids: | |
| _SERIES_CATEGORY[_sid] = _cat | |
| ALL_DEFAULT_SERIES: List[str] = [ | |
| sid for ids in SERIES_BY_CATEGORY.values() for sid in ids | |
| ] | |
| # --------------------------------------------------------------------------- | |
| # Helpers | |
| # --------------------------------------------------------------------------- | |
| def _ts_ms(dt: datetime) -> int: | |
| """Convert a datetime to epoch milliseconds (UTC).""" | |
| return int(dt.replace(tzinfo=timezone.utc).timestamp() * 1000) | |
| def _iso_to_date(s: str) -> str: | |
| """Validate and return an ISO date string (YYYY-MM-DD).""" | |
| datetime.strptime(s, "%Y-%m-%d") | |
| return s | |
| def _watermark_name(series_id: str) -> str: | |
| return f"fred:{series_id}" | |
| # --------------------------------------------------------------------------- | |
| # Core ingestion | |
| # --------------------------------------------------------------------------- | |
| def ingest_series( | |
| fred, | |
| conn, | |
| series_id: str, | |
| start: str, | |
| end: str, | |
| use_watermark: bool = True, | |
| ) -> int: | |
| """Fetch one FRED series and store observations. Returns row count.""" | |
| # Resolve watermark-based start if resuming | |
| effective_start = start | |
| if use_watermark: | |
| wm = ghdb.get_watermark_ms(conn, _watermark_name(series_id)) | |
| if wm is not None: | |
| # Start one day after the watermark to avoid re-fetching | |
| wm_date = datetime.utcfromtimestamp(wm / 1000).strftime("%Y-%m-%d") | |
| if wm_date > effective_start: | |
| effective_start = wm_date | |
| # Fetch metadata | |
| try: | |
| info = fred.get_series_info(series_id) | |
| except Exception as exc: | |
| print(f" [WARN] Could not fetch metadata for {series_id}: {exc}") | |
| info = {} | |
| series_name = info.get("title", series_id) if isinstance(info, dict) else getattr(info, "title", series_id) | |
| frequency = info.get("frequency_short", "") if isinstance(info, dict) else getattr(info, "frequency_short", "") | |
| units = info.get("units_short", "") if isinstance(info, dict) else getattr(info, "units_short", "") | |
| category = _SERIES_CATEGORY.get(series_id, "other") | |
| # Fetch observations | |
| try: | |
| data = fred.get_series( | |
| series_id, | |
| observation_start=effective_start, | |
| observation_end=end, | |
| ) | |
| except Exception as exc: | |
| print(f" [ERROR] Failed to fetch {series_id}: {exc}") | |
| return 0 | |
| if data is None or data.empty: | |
| print(f" {series_id}: 0 new observations") | |
| return 0 | |
| inserted = 0 | |
| max_obs_ms = 0 | |
| batch_count = 0 | |
| now_ms = int(time.time() * 1000) | |
| for obs_date, value in data.items(): | |
| # Skip NaN values (FRED uses NaN for missing observations) | |
| if value != value: # NaN check | |
| continue | |
| obs_dt = obs_date.to_pydatetime() | |
| obs_ms = _ts_ms(obs_dt) | |
| row = { | |
| "provider": "fred", | |
| "series_id": series_id, | |
| "series_name": series_name, | |
| "category": category, | |
| "region": "US", | |
| "frequency": frequency, | |
| "units": units, | |
| "observation_date_ms": obs_ms, | |
| "value": float(value), | |
| "ingested_at_ms": now_ms, | |
| "raw_json": json.dumps( | |
| { | |
| "series_id": series_id, | |
| "date": obs_dt.strftime("%Y-%m-%d"), | |
| "value": float(value), | |
| }, | |
| default=str, | |
| ), | |
| } | |
| if ghdb.insert_macro_indicator(conn, row): | |
| inserted += 1 | |
| if obs_ms > max_obs_ms: | |
| max_obs_ms = obs_ms | |
| batch_count += 1 | |
| if batch_count % 500 == 0: | |
| conn.commit() | |
| # Final commit for remaining rows | |
| if batch_count % 500 != 0: | |
| conn.commit() | |
| # Update watermark | |
| if max_obs_ms > 0 and use_watermark: | |
| ghdb.set_watermark_ms(conn, _watermark_name(series_id), max_obs_ms) | |
| conn.commit() | |
| print(f" {series_id}: {inserted} inserted ({batch_count} processed, {data.shape[0]} total obs)") | |
| return inserted | |
| # --------------------------------------------------------------------------- | |
| # CLI | |
| # --------------------------------------------------------------------------- | |
| def parse_args() -> argparse.Namespace: | |
| p = argparse.ArgumentParser( | |
| description="Ingest FRED macro indicators into Aureon global history DB.", | |
| ) | |
| p.add_argument("--db", default=None, help="SQLite DB path override") | |
| p.add_argument( | |
| "--no-resume", | |
| action="store_true", | |
| help="Ignore watermarks and re-fetch from --start", | |
| ) | |
| p.add_argument("--start", default="1970-01-01", help="ISO start date (default: 1970-01-01)") | |
| p.add_argument("--end", default=None, help="ISO end date (default: today)") | |
| p.add_argument( | |
| "--series", | |
| default=None, | |
| help="Comma-separated FRED series IDs (overrides defaults)", | |
| ) | |
| p.add_argument( | |
| "--category", | |
| default="all", | |
| help="Category filter: rates,gdp,inflation,employment,money,housing,consumer,financial,global,all", | |
| ) | |
| p.add_argument( | |
| "--fred-key", | |
| default=None, | |
| help="FRED API key (or set env FRED_API_KEY)", | |
| ) | |
| return p.parse_args() | |
| def main() -> None: | |
| args = parse_args() | |
| # ---- API key ---- | |
| api_key = args.fred_key or os.getenv("FRED_API_KEY") | |
| if not api_key: | |
| print("ERROR: FRED API key required. Use --fred-key or set FRED_API_KEY env var.") | |
| print(" Get a free key at https://fred.stlouisfed.org/docs/api/api_key.html") | |
| sys.exit(1) | |
| try: | |
| from fredapi import Fred | |
| except ImportError: | |
| print("ERROR: fredapi not installed. Run: pip install fredapi") | |
| sys.exit(1) | |
| fred = Fred(api_key=api_key) | |
| # ---- Dates ---- | |
| start = _iso_to_date(args.start) | |
| end = _iso_to_date(args.end) if args.end else datetime.utcnow().strftime("%Y-%m-%d") | |
| # ---- Series selection ---- | |
| if args.series: | |
| series_list = [s.strip().upper() for s in args.series.split(",") if s.strip()] | |
| elif args.category == "all": | |
| series_list = list(ALL_DEFAULT_SERIES) | |
| else: | |
| cats = [c.strip().lower() for c in args.category.split(",") if c.strip()] | |
| series_list = [] | |
| for cat in cats: | |
| if cat not in SERIES_BY_CATEGORY: | |
| print(f"WARNING: Unknown category '{cat}', skipping. " | |
| f"Valid: {', '.join(SERIES_BY_CATEGORY.keys())}") | |
| continue | |
| series_list.extend(SERIES_BY_CATEGORY[cat]) | |
| if not series_list: | |
| print("No series to ingest.") | |
| sys.exit(0) | |
| # ---- Connect ---- | |
| conn = ghdb.connect(args.db) | |
| paths = ghdb.resolve_paths(args.db) | |
| print(f"DB: {paths.db_path}") | |
| print(f"Period: {start} to {end}") | |
| print(f"Series: {len(series_list)} series, resume={'off' if args.no_resume else 'on'}") | |
| print() | |
| use_watermark = not args.no_resume | |
| total_inserted = 0 | |
| errors = 0 | |
| for i, sid in enumerate(series_list, 1): | |
| print(f"[{i}/{len(series_list)}] {sid}") | |
| try: | |
| n = ingest_series(fred, conn, sid, start, end, use_watermark=use_watermark) | |
| total_inserted += n | |
| except Exception as exc: | |
| print(f" [ERROR] {sid}: {exc}") | |
| errors += 1 | |
| # FRED rate limit: 120 req/min. Sleep 0.5s between series to stay safe. | |
| if i < len(series_list): | |
| time.sleep(0.5) | |
| print() | |
| print(f"Done. {total_inserted} observations inserted across {len(series_list)} series " | |
| f"({errors} errors).") | |
| conn.close() | |
| if __name__ == "__main__": | |
| main() | |