src/model.py

"""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])