Spaces:
Runtime error
Runtime error
File size: 3,090 Bytes
b717bee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """Load calibrated V5 model + exact predict_return_signal from notebook"""
import joblib
from huggingface_hub import hf_hub_download # type: ignore
import numpy as np
import pandas as pd
from src.config import MODEL_PATH, HF_REPO_ID
from src.inference_features import build_live_v5_features
def load_model():
if not HF_REPO_ID or HF_REPO_ID.startswith("yourusername/"):
model = joblib.load(MODEL_PATH)
print("Model loaded from local file")
return model
try:
path = hf_hub_download(repo_id=HF_REPO_ID, filename="v5_lgbm_24h_dualfilter.joblib")
model = joblib.load(path)
print("✅ Model loaded from Hugging Face Hub")
except Exception:
model = joblib.load(MODEL_PATH)
print("✅ Model loaded from local file")
return model
def predict_return_signal(model, X):
"""Exact function from latest notebook (handles calibrated classifier)"""
if isinstance(model, dict) and "model" in model: # calibrated classifier
proba = model["model"].predict_proba(X)[:, 1]
calibrator = model.get("calibrator")
if calibrator is None:
return (proba - 0.5) * 2.0 * 0.006
# Apply calibrator
idx = np.searchsorted(calibrator["edges"][1:-1], proba, side="right")
return calibrator["values"][idx]
else:
return model.predict(X) # type: ignore
def core_model(model):
return model.get("model") if isinstance(model, dict) and "model" in model else model
def model_feature_cols(model) -> list[str] | None:
if isinstance(model, dict):
cols = model.get("feature_cols") or model.get("features")
if cols is not None:
return list(cols)
return None
def build_live_matrix(df_raw: pd.DataFrame, model):
df_feat, feature_cols = build_live_v5_features(df_raw)
latest = df_feat.iloc[-1:].copy()
trained_feature_cols = model_feature_cols(model)
if trained_feature_cols is not None:
missing = [col for col in trained_feature_cols if col not in latest.columns]
if missing:
raise ValueError(f"Live data is missing {len(missing)} trained features, e.g. {missing[:10]}")
X = latest[trained_feature_cols]
else:
X = latest[feature_cols]
expected_features = getattr(core_model(model), "n_features_in_", None)
if expected_features is not None and X.shape[1] != expected_features:
raise ValueError(
f"Live feature count is {X.shape[1]}, but the model expects {expected_features}. "
"Re-export the notebook model as a dict containing feature_cols=feat_v5."
)
return df_feat, latest, X
def classifier_positive_proba(model, X) -> float | None:
base_model = core_model(model)
if hasattr(base_model, "predict_proba"):
return float(base_model.predict_proba(X)[:, 1][0])
return None
def predict_next_8h(df_raw: pd.DataFrame, model): # type: ignore
"""Main inference entry point"""
_, _, X = build_live_matrix(df_raw, model)
pred = predict_return_signal(model, X)
return float(pred[0])
|