Spaces:
Running on Zero
Running on Zero
cloud refresh: 06:30 Asia/Tokyo (21:30 UTC); one attempt per slot; status upload failure no longer raises
352d9b7 verified | """Daily cloud refresh run inside the Space: snapshot sources that keep no history (or move daily) and need no | |
| API key, written to the PRIVATE dataset (never to the Space). Schedule: 06:30 Asia/Tokyo (21:30 UTC) by | |
| default; if the Space slept through the slot it catches up on wake when the last run is >20 h old. | |
| Output in the dataset (HF_DATA_REPO): | |
| warehouse/snapshots/<source>/<YYYY-MM-DD>.parquet dated snapshots (append-only; never overwritten by other days) | |
| warehouse/snapshots/_status/latest.json last run time + per-source result | |
| Env: CLOUD_REFRESH=off disables; CLOUD_REFRESH_UTC=HH:MM overrides the slot. | |
| """ | |
| from __future__ import annotations | |
| import datetime as dt | |
| import io | |
| import json | |
| import os | |
| import re | |
| import tempfile | |
| import threading | |
| import time | |
| import traceback | |
| import pandas as pd | |
| import requests | |
| UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36"} | |
| NASDAQ_H = {**UA, "Accept": "application/json, text/plain, */*", "Origin": "https://www.nasdaq.com", "Referer": "https://www.nasdaq.com/"} | |
| FRED_IDS = ["DGS10", "DFII10", "T10YIE", "DGS2", "DFF", "SOFR", "BAMLH0A0HYM2", "BAMLH0A1HYBB", "BAMLH0A3HYC", "BAMLC0A4CBBB", | |
| "BAMLC0A0CM", "BAMLC0A4CBBBEY", "DBAA", "DAAA", "T10Y3M", "T10Y2Y", "DCOILBRENTEU", "DCOILWTICO", "VIXCLS", "NFCI", | |
| "WALCL", "RRPONTSYD", "ICSA", "DTWEXBGS"] | |
| _LOCK = threading.Lock() | |
| def _repo(): | |
| return os.getenv("HF_DATA_REPO", "").strip() | |
| def _api(): | |
| from huggingface_hub import HfApi | |
| return HfApi(token=os.getenv("HF_WRITE_TOKEN") or os.getenv("HF_TOKEN")) | |
| def _get(url, headers=UA, timeout=30, tries=2, params=None): | |
| """Browser-impersonating GET (FRED, SSGA and others reset plain script connections), falling back to requests.""" | |
| last = None | |
| for i in range(tries): | |
| try: | |
| from curl_cffi import requests as cr | |
| r = cr.get(url, params=params, impersonate="chrome", timeout=timeout) | |
| if r.status_code == 200 and r.content: | |
| return r | |
| last = f"HTTP {r.status_code}" | |
| except Exception as e: # curl_cffi missing or connection failure | |
| last = type(e).__name__ | |
| try: | |
| r = requests.get(url, params=params, headers=headers, timeout=timeout) | |
| if r.status_code == 200 and r.content: | |
| return r | |
| last = f"HTTP {r.status_code}" | |
| except requests.RequestException as e2: | |
| last = type(e2).__name__ | |
| time.sleep(3 * (i + 1)) | |
| raise RuntimeError(f"fetch failed ({last}): {url}") | |
| # ---------------------------------------------------------------- sources | |
| def spy_holdings(): | |
| raw = _get("https://www.ssga.com/library-content/products/fund-data/etfs/us/holdings-daily-us-en-spy.xlsx").content | |
| x = pd.read_excel(io.BytesIO(raw), header=None) | |
| hdr = next(i for i in range(10) if (x.iloc[i].astype(str).str.strip() == "Ticker").any()) | |
| df = x.iloc[hdr + 1:].copy(); df.columns = [str(c).strip() for c in x.iloc[hdr]] | |
| return df.dropna(subset=["Ticker"]).astype("string") | |
| def fred(): | |
| frames, fails = [], 0 | |
| for sid in FRED_IDS: | |
| if fails >= 3: | |
| break | |
| try: | |
| d = pd.read_csv(io.BytesIO(_get(f"https://fred.stlouisfed.org/graph/fredgraph.csv?id={sid}").content)) | |
| d.columns = ["period", "value"]; d["value"] = pd.to_numeric(d["value"], errors="coerce") | |
| d["series_id"] = sid; frames.append(d.dropna().tail(400)) # last ~400 obs keeps ICE spreads past FRED's window | |
| fails = 0 | |
| except Exception: | |
| fails += 1 | |
| continue | |
| time.sleep(0.5) | |
| return pd.concat(frames, ignore_index=True) | |
| def cboe_putcall(): | |
| frames = [] | |
| for name in ["totalpc", "equitypc", "indexpc", "vixpc", "spxpc"]: | |
| txt = _get(f"https://cdn.cboe.com/resources/options/volume_and_call_put_ratios/{name}.csv").text.splitlines() | |
| hdr = next(i for i, l in enumerate(txt) if l.upper().startswith("DATE")) | |
| d = pd.read_csv(io.StringIO("\n".join(txt[hdr:]))).tail(30); d["series"] = name | |
| frames.append(d.astype("string")) | |
| return pd.concat(frames, ignore_index=True) | |
| def gpu_prices(): | |
| return pd.read_csv(io.BytesIO(_get("https://getdeploying.com/dataset/gpu-prices/weekly.csv").content)).astype("string") | |
| def openrouter_models(): | |
| rows = [] | |
| for x in _get("https://openrouter.ai/api/v1/models").json().get("data", []): | |
| p = x.get("pricing") or {} | |
| rows.append({"id": x.get("id"), "name": x.get("name"), "created": x.get("created"), "context_length": x.get("context_length"), | |
| "price_prompt": p.get("prompt"), "price_completion": p.get("completion")}) | |
| return pd.DataFrame(rows).astype("string") | |
| def earnings_calendar(days=30): | |
| rows = [] | |
| for k in range(days): | |
| d = dt.date.today() + dt.timedelta(days=k) | |
| if d.weekday() >= 5: | |
| continue | |
| try: | |
| j = _get("https://api.nasdaq.com/api/calendar/earnings", headers=NASDAQ_H, params={"date": d.isoformat()}, tries=1).json() | |
| rows += [{"date": d.isoformat(), **{kk: str(vv) for kk, vv in r.items()}} for r in ((j.get("data") or {}).get("rows") or [])] | |
| except Exception: | |
| pass | |
| time.sleep(0.5) | |
| return pd.DataFrame(rows) | |
| def estimates(spy: pd.DataFrame | None): | |
| if spy is None or spy.empty: | |
| return pd.DataFrame() | |
| fails = 0 | |
| tickers = sorted({str(t).strip().replace(".", "-") for t in spy["Ticker"] if re.fullmatch(r"[A-Z][A-Z.\-]{0,6}", str(t).strip())}) | |
| rows = [] | |
| for t in tickers: | |
| try: | |
| j = _get(f"https://api.nasdaq.com/api/analyst/{t}/earnings-forecast", headers=NASDAQ_H, tries=1).json().get("data") or {} | |
| for k in ("quarterlyForecast", "yearlyForecast"): | |
| rows += [{"ticker": t, "kind": k, **{kk: str(vv) for kk, vv in r.items()}} for r in (j.get(k) or {}).get("rows") or []] | |
| fails = 0 | |
| except Exception: | |
| fails += 1 | |
| if fails >= 10: # endpoint blocked: stop instead of 500 timeouts | |
| break | |
| time.sleep(0.4) | |
| return pd.DataFrame(rows) | |
| def treasury_curves(): | |
| y = dt.date.today().year | |
| frames = [] | |
| for kind, tag in [("daily_treasury_yield_curve", "nominal"), ("daily_treasury_real_yield_curve", "real")]: | |
| url = ("https://home.treasury.gov/resource-center/data-chart-center/interest-rates/" | |
| f"daily-treasury-rates.csv/{y}/all?type={kind}&field_tdr_date_value={y}&page&_format=csv") | |
| d = pd.read_csv(io.BytesIO(_get(url).content)); d["curve"] = tag; frames.append(d.astype("string")) | |
| return pd.concat(frames, ignore_index=True) | |
| def dram_spot(): | |
| from curl_cffi import requests as cffi | |
| r = cffi.Session(impersonate="chrome").get("https://www.dramexchange.com/", timeout=60) | |
| tabs = [t for t in pd.read_html(io.BytesIO(r.content)) if t.shape[1] >= 5 and t.shape[0] >= 2] | |
| for i, t in enumerate(tabs): | |
| t.columns = [" ".join(map(str, c)) if isinstance(c, tuple) else str(c) for c in t.columns] | |
| t["table_no"] = i | |
| return pd.concat(tabs, ignore_index=True).astype("string") if tabs else pd.DataFrame() | |
| # ---------------------------------------------------------------- runner | |
| def run_once() -> dict: | |
| """Fetch every source, upload dated snapshots to the private dataset, write a status file.""" | |
| repo = _repo() | |
| if not repo: | |
| return {"error": "HF_DATA_REPO not set"} | |
| api = _api() | |
| if not api.dataset_info(repo).private: | |
| return {"error": "dataset is public; refusing to write"} | |
| # snapshot dates use the user's calendar (UTC+9, Asia/Tokyo) so they match the local 06:30 run's dates | |
| today = (dt.datetime.utcnow() + dt.timedelta(hours=float(os.getenv("CLOUD_REFRESH_TZ_HOURS", "9")))).date().isoformat() | |
| status, spy = {"started_utc": dt.datetime.utcnow().isoformat(timespec="seconds"), "sources": {}}, None | |
| jobs = [("spy_holdings", spy_holdings), ("fred", fred), ("cboe_putcall", cboe_putcall), ("gpu_prices", gpu_prices), | |
| ("openrouter_models", openrouter_models), ("earnings_calendar", earnings_calendar), ("treasury_curves", treasury_curves), | |
| ("dram_spot", dram_spot), ("estimates_nasdaq", lambda: estimates(spy))] | |
| with tempfile.TemporaryDirectory() as tmp: | |
| for name, fn in jobs: | |
| try: | |
| df = fn() | |
| if name == "spy_holdings": | |
| spy = df | |
| if df is None or df.empty: | |
| status["sources"][name] = "empty"; continue | |
| df["snapshot_date"] = today | |
| p = os.path.join(tmp, f"{name}.parquet") | |
| df.to_parquet(p, index=False) | |
| api.upload_file(path_or_fileobj=p, path_in_repo=f"warehouse/snapshots/{name}/{today}.parquet", repo_id=repo, | |
| repo_type="dataset", commit_message=f"cloud refresh {name} {today}") | |
| status["sources"][name] = f"ok {len(df)} rows" | |
| except Exception as e: | |
| status["sources"][name] = f"error {type(e).__name__}: {str(e)[:120]}" | |
| status["finished_utc"] = dt.datetime.utcnow().isoformat(timespec="seconds") | |
| sp = os.path.join(tmp, "latest.json") | |
| with open(sp, "w") as fh: | |
| json.dump(status, fh, indent=1) | |
| try: | |
| api.upload_file(path_or_fileobj=sp, path_in_repo="warehouse/snapshots/_status/latest.json", repo_id=repo, | |
| repo_type="dataset", commit_message=f"cloud refresh status {today}") | |
| except Exception as e: # e.g. read-only token: report, do not raise | |
| status["status_upload"] = f"error {type(e).__name__}: {str(e)[:120]}" | |
| return status | |
| def _last_run_utc(): | |
| try: | |
| from huggingface_hub import hf_hub_download | |
| p = hf_hub_download(_repo(), "warehouse/snapshots/_status/latest.json", repo_type="dataset", | |
| token=os.getenv("HF_TOKEN"), force_download=True) | |
| return dt.datetime.fromisoformat(json.load(open(p))["finished_utc"]) | |
| except Exception: | |
| return None | |
| def _loop(): | |
| """One attempt per scheduled slot. A failed or partial run is not retried until the next slot, so a read-only | |
| token or a blocked source can never turn into a refetch loop.""" | |
| hh, mm = (int(x) for x in os.getenv("CLOUD_REFRESH_UTC", "21:30").split(":")) | |
| time.sleep(90) # let the app finish starting | |
| attempted_slot = None | |
| while True: | |
| try: | |
| now = dt.datetime.utcnow() | |
| slot = now.replace(hour=hh, minute=mm, second=0, microsecond=0) | |
| if slot > now: | |
| slot -= dt.timedelta(days=1) # most recent scheduled slot | |
| if attempted_slot != slot: | |
| last = _last_run_utc() | |
| if last is None or last < slot: # missed or due: run now (covers waking after sleep) | |
| attempted_slot = slot | |
| with _LOCK: | |
| print("[cloud_refresh] running for slot", slot.isoformat(), flush=True) | |
| try: | |
| print("[cloud_refresh]", json.dumps(run_once())[:800], flush=True) | |
| except Exception: | |
| traceback.print_exc() | |
| else: | |
| attempted_slot = slot | |
| nxt = slot + dt.timedelta(days=1) | |
| time.sleep(max(60, min((nxt - dt.datetime.utcnow()).total_seconds(), 3600))) | |
| except Exception: | |
| traceback.print_exc() | |
| time.sleep(3600) | |
| def start_background(): | |
| if os.getenv("CLOUD_REFRESH", "on").lower() == "off": | |
| return False | |
| threading.Thread(target=_loop, name="cloud_refresh", daemon=True).start() | |
| return True | |