thinkingEverytime/QuantOracle

1"""Cross-sectional ranking from a feature snapshot + a model."""2 3from __future__ import annotations4 5from pathlib import Path6 7import numpy as np8import pandas as pd9 10from quant.registry import data_root, latest_dir, load_meta11 12try:13    import duckdb  # type: ignore14except Exception:  # pragma: no cover15    duckdb = None16 17 18FEATURES = [19    "ret_1d",20    "ret_5d",21    "ret_20d",22    "vol_20d",23    "price_sma20",24    "price_sma50",25    "rsi_14",26]27 28 29def features_path() -> Path:30    return data_root() / "features.parquet"31 32def _require_duckdb():33    if duckdb is None:34        raise RuntimeError("duckdb is required for feature snapshot reads. Install: `pip install duckdb`.")35 36 37def latest_feature_date() -> pd.Timestamp | None:38    p = features_path()39    if not p.exists():40        return None41    _require_duckdb()42    con = duckdb.connect(database=":memory:")43    d = con.execute("SELECT MAX(Date) FROM read_parquet(?)", [str(p)]).fetchone()[0]44    con.close()45    return pd.to_datetime(d) if d else None46 47 48def load_feature_snapshot(date: pd.Timestamp | None = None) -> pd.DataFrame:49    p = features_path()50    if not p.exists():51        return pd.DataFrame()52    _require_duckdb()53 54    if date is None:55        date = latest_feature_date()56        if date is None:57            return pd.DataFrame()58 59    con = duckdb.connect(database=":memory:")60    df = con.execute("SELECT * FROM read_parquet(?) WHERE Date = ?", [str(p), date]).df()61    con.close()62    if df.empty:63        return df64    df["Date"] = pd.to_datetime(df["Date"])65    return df66 67 68def load_ridge_latest(horizon: int = 5):69    d = latest_dir(f"ridge_h{horizon}")70    if not d:71        return None, None72    meta = load_meta(d)73    z = np.load(d / "model.npz", allow_pickle=True)74    return meta, {"w": z["w"], "mu": z["mu"], "sig": z["sig"], "features": list(z["features"])}75 76 77def predict_ridge(snapshot: pd.DataFrame, model: dict) -> pd.DataFrame:78    if snapshot.empty:79        return pd.DataFrame()80    feats = model["features"]81    X = snapshot[feats].to_numpy(dtype=float)82    Xz = (X - model["mu"]) / model["sig"]83    yhat = Xz @ model["w"]84    out = snapshot[["symbol"]].copy()85    out["pred"] = yhat86    out["risk"] = snapshot["vol_20d"].astype(float).clip(lower=1e-6)87    return out88 89 90def top_bottom(preds: pd.DataFrame, n: int = 10) -> tuple[pd.DataFrame, pd.DataFrame]:91    if preds.empty:92        return pd.DataFrame(), pd.DataFrame()93    p = preds.sort_values("pred", ascending=False)94    return p.head(n), p.tail(n).sort_values("pred", ascending=True)95