thinkingEverytime/QuantOracle

1#!/usr/bin/env python32"""Validate freshness + schema of published QuantOracle artifacts.3 4This checker is intended for CI and can run safely on a schedule.5It validates:6  - intraday quotes snapshot7  - EOD latest snapshot metadata8  - news intel daily snapshot9"""10 11from __future__ import annotations12 13import argparse14import json15import os16from dataclasses import dataclass17from datetime import UTC, datetime, time18from pathlib import PurePosixPath19from typing import Any20 21import pytz22import requests23 24 25def _has(v: str | None) -> bool:26    return bool((v or "").strip())27 28 29def _safe_num(v: Any) -> float:30    try:31        return float(v)32    except Exception:33        return 0.034 35 36def _public_url(path: str) -> str | None:37    base = (os.getenv("SUPABASE_URL") or "").strip().rstrip("/")38    bucket = (os.getenv("SUPABASE_BUCKET") or "").strip()39    if not base or not bucket:40        return None41    safe = str(PurePosixPath(path.lstrip("/")))42    return f"{base}/storage/v1/object/public/{bucket}/{safe}"43 44 45def _quotes_url() -> str | None:46    direct = (os.getenv("QUANTORACLE_SUPABASE_QUOTES_URL") or "").strip()47    if direct:48        return direct49    prefix = (os.getenv("QUANTORACLE_EOD_PREFIX") or "eod/nifty50").strip("/")50    return _public_url(f"{prefix}/quotes.json")51 52 53def _eod_latest_url() -> str | None:54    direct = (os.getenv("QUANTORACLE_SUPABASE_EOD_LATEST_URL") or "").strip()55    if direct:56        return direct57    prefix = (os.getenv("QUANTORACLE_EOD_PREFIX") or "eod/nifty50").strip("/")58    return _public_url(f"{prefix}/latest.json")59 60 61def _news_intel_url() -> str | None:62    direct = (os.getenv("QUANTORACLE_NEWS_INTEL_URL") or "").strip()63    if direct:64        return direct65    prefix = (os.getenv("QUANTORACLE_NEWS_PREFIX") or "news/intel").strip("/")66    return _public_url(f"{prefix}/latest.json")67 68 69def _parse_ts(raw: Any) -> datetime | None:70    if not isinstance(raw, str) or not raw.strip():71        return None72    s = raw.strip()73    if s.endswith("Z"):74        s = s[:-1] + "+00:00"75    # Support +0530 variant.76    if len(s) >= 5 and (s[-5] in "+-") and s[-3] != ":":77        s = s[:-2] + ":" + s[-2:]78    try:79        dt = datetime.fromisoformat(s)80    except Exception:81        return None82    if dt.tzinfo is None:83        return dt.replace(tzinfo=UTC)84    return dt.astimezone(UTC)85 86 87def _age_minutes(ts: datetime | None) -> float | None:88    if not ts:89        return None90    return (datetime.now(UTC) - ts).total_seconds() / 60.091 92 93def _is_nse_runtime() -> bool:94    now_ist = datetime.now(pytz.timezone("Asia/Kolkata"))95    if now_ist.weekday() >= 5:96        return False97    t = now_ist.time()98    return time(9, 0) <= t <= time(16, 15)99 100 101@dataclass102class CheckResult:103    name: str104    url: str | None105    ok: bool106    required: bool107    reason: str108    as_of_utc: str | None109    age_minutes: float | None110    status: int | None111 112    def to_dict(self) -> dict[str, Any]:113        return {114            "name": self.name,115            "url": self.url,116            "ok": self.ok,117            "required": self.required,118            "reason": self.reason,119            "as_of_utc": self.as_of_utc,120            "age_minutes": None121            if self.age_minutes is None122            else round(self.age_minutes, 2),123            "status": self.status,124        }125 126 127def _fetch_json(url: str) -> tuple[int, dict[str, Any] | None, str | None]:128    try:129        r = requests.get(url, timeout=15)130        if r.status_code != 200:131            return r.status_code, None, f"HTTP {r.status_code}"132        data = r.json()133        if not isinstance(data, dict):134            return r.status_code, None, "Response is not a JSON object"135        return r.status_code, data, None136    except Exception as e:137        return None, None, str(e)138 139 140def _check_quotes(max_age_minutes: int) -> CheckResult:141    url = _quotes_url()142    required = _is_nse_runtime()143    if not url:144        return CheckResult(145            name="quotes",146            url=None,147            ok=not required,148            required=required,149            reason="Missing quotes URL configuration",150            as_of_utc=None,151            age_minutes=None,152            status=None,153        )154 155    status, data, error = _fetch_json(url)156    if data is None:157        return CheckResult(158            name="quotes",159            url=url,160            ok=not required,161            required=required,162            reason=error or "quotes fetch failed",163            as_of_utc=None,164            age_minutes=None,165            status=status,166        )167 168    as_of = _parse_ts(data.get("as_of_utc")) or _parse_ts(data.get("as_of_ist"))169    age = _age_minutes(as_of)170    quotes = data.get("quotes")171    count = int(data.get("count") or 0)172    shape_ok = isinstance(quotes, dict) and count >= 1173    age_ok = (age is not None) and (age <= max_age_minutes)174 175    ok = shape_ok and (age_ok or (not required))176    reason = "ok"177    if not shape_ok:178        reason = "quotes schema invalid"179    elif required and not age_ok:180        reason = f"quotes stale: age={round(age or 0, 2)}m limit={max_age_minutes}m"181    elif not required and not age_ok:182        reason = "outside NSE hours; stale tolerated"183 184    return CheckResult(185        name="quotes",186        url=url,187        ok=ok,188        required=required,189        reason=reason,190        as_of_utc=as_of.isoformat().replace("+00:00", "Z") if as_of else None,191        age_minutes=age,192        status=status,193    )194 195 196def _check_eod(max_age_minutes: int) -> CheckResult:197    url = _eod_latest_url()198    required = True199    if not url:200        return CheckResult(201            name="eod",202            url=None,203            ok=False,204            required=required,205            reason="Missing EOD latest URL configuration",206            as_of_utc=None,207            age_minutes=None,208            status=None,209        )210 211    status, data, error = _fetch_json(url)212    if data is None:213        return CheckResult(214            name="eod",215            url=url,216            ok=False,217            required=required,218            reason=error or "eod fetch failed",219            as_of_utc=None,220            age_minutes=None,221            status=status,222        )223 224    as_of = _parse_ts(data.get("generated_at_utc")) or _parse_ts(data.get("as_of_utc"))225    age = _age_minutes(as_of)226    shape_ok = all(227        [228            isinstance(data.get("as_of_date"), str),229            isinstance(data.get("universe"), str),230            isinstance(data.get("model_id"), str),231            isinstance(data.get("model_version"), str),232        ]233    )234    age_ok = (age is not None) and (age <= max_age_minutes)235 236    reason = "ok"237    if not shape_ok:238        reason = "eod schema invalid"239    elif not age_ok:240        reason = f"eod stale: age={round(age or 0, 2)}m limit={max_age_minutes}m"241 242    return CheckResult(243        name="eod",244        url=url,245        ok=shape_ok and age_ok,246        required=required,247        reason=reason,248        as_of_utc=as_of.isoformat().replace("+00:00", "Z") if as_of else None,249        age_minutes=age,250        status=status,251    )252 253 254def _check_news(max_age_minutes: int) -> CheckResult:255    url = _news_intel_url()256    required = True257    if not url:258        return CheckResult(259            name="news_intel",260            url=None,261            ok=False,262            required=required,263            reason="Missing news intel URL configuration",264            as_of_utc=None,265            age_minutes=None,266            status=None,267        )268 269    status, data, error = _fetch_json(url)270    if data is None:271        return CheckResult(272            name="news_intel",273            url=url,274            ok=False,275            required=required,276            reason=error or "news intel fetch failed",277            as_of_utc=None,278            age_minutes=None,279            status=status,280        )281 282    as_of = _parse_ts(data.get("as_of_utc")) or _parse_ts(data.get("generated_at_utc"))283    age = _age_minutes(as_of)284 285    items = data.get("items")286    shape_ok = isinstance(items, list) and len(items) > 0287    if shape_ok:288        for item in items[:10]:289            if not isinstance(item, dict):290                shape_ok = False291                break292            if not all(293                [294                    isinstance(item.get("headline"), str),295                    isinstance(item.get("source"), str),296                    isinstance(item.get("impact"), dict),297                    isinstance(item.get("source_tier"), str),298                ]299            ):300                shape_ok = False301                break302 303    age_ok = (age is not None) and (age <= max_age_minutes)304    reason = "ok"305    if not shape_ok:306        reason = "news intel schema invalid"307    elif not age_ok:308        reason = f"news intel stale: age={round(age or 0, 2)}m limit={max_age_minutes}m"309 310    return CheckResult(311        name="news_intel",312        url=url,313        ok=shape_ok and age_ok,314        required=required,315        reason=reason,316        as_of_utc=as_of.isoformat().replace("+00:00", "Z") if as_of else None,317        age_minutes=age,318        status=status,319    )320 321 322def main() -> int:323    ap = argparse.ArgumentParser()324    ap.add_argument("--quotes-max-age-minutes", type=int, default=180)325    ap.add_argument("--eod-max-age-minutes", type=int, default=4320)326    ap.add_argument("--news-max-age-minutes", type=int, default=2160)327    ap.add_argument("--strict", action="store_true")328    args = ap.parse_args()329 330    checks = [331        _check_quotes(args.quotes_max_age_minutes),332        _check_eod(args.eod_max_age_minutes),333        _check_news(args.news_max_age_minutes),334    ]335 336    summary = {337        "all_ok": all(c.ok for c in checks),338        "required_ok": all((c.ok or (not c.required)) for c in checks),339        "nse_runtime": _is_nse_runtime(),340        "config": {341            "supabase_url": _has(os.getenv("SUPABASE_URL")),342            "supabase_bucket": _has(os.getenv("SUPABASE_BUCKET")),343            "quotes_url": _quotes_url(),344            "eod_latest_url": _eod_latest_url(),345            "news_intel_url": _news_intel_url(),346        },347    }348 349    out = {350        "as_of_utc": datetime.now(UTC).isoformat().replace("+00:00", "Z"),351        "checks": {c.name: c.to_dict() for c in checks},352        "summary": summary,353    }354    print(json.dumps(out, indent=2))355 356    if args.strict and not summary["required_ok"]:357        return 1358    return 0359 360 361if __name__ == "__main__":362    raise SystemExit(main())363