I have a model, and it has some features that are more important than others.
How to fetch the top features of a model created via Model studio in a programmatic way ?
Here are two code snippets to show how to use the public API to do so.
End result example
The first one working from transforms.
'''
The Service user required: api:usage:models-read
You need to allow the service user for “models-read” in the Developer console.
You need to whitelist the project in Developer Console.
You need to grant access to the project (Viewer/Editor/etc.)
'''
"""Retrieve top feature importances from the latest model experiment."""
from io import BytesIO
from urllib.parse import urlparse
import base64
import json
import numpy as np
import polars as pl
import palantir_models as pm
from foundry_sdk import Config, ConfidentialClientAuth, FoundryClient
from palantir_models.transforms import ModelInput
from transforms.api import LightweightOutput, Output, transform
from transforms.external.systems import ResolvedSource, Source, external_systems
import logging
TOP_N = 20
REQUEST_TIMEOUT_SECONDS = 60
MODE_INPUT_RID = "ri.models.main.model.769d82f8-b9d8-4ba5-a152-68687cc6f18f"
@external_systems(data_source=Source("ri.magritte..source.02548bca-84e8-43fd-a30d-75cad150e1ef"))
@transform.using(
# model=ModelInput("ri.models.main.model.769d82f8-b9d8-4ba5-a152-68687cc6f18f"),
output=Output("ri.foundry.main.dataset.993cde5a-d80a-43e6-8ead-e69f30f8f540"),
)
def compute(
data_source: ResolvedSource,
# model: pm.ModelAdapter,
output: LightweightOutput,
) -> None:
## Step 1. Get the token of the service user
host = "https://YOUR_DOMAIN.com"
auth = ConfidentialClientAuth(
client_id=data_source.get_secret("additionalSecretServiceUserId"),
client_secret=data_source.get_secret("additionalSecretServiceUserSecret"),
hostname=host,
should_refresh=True,
# scopes=["api:usage:models-read"], # Max scope for now
)
# auth.sign_in_as_service_user()
client = FoundryClient(auth=auth, hostname=host)
## Step 2. Fetch the most important and latest features
model_rid = MODE_INPUT_RID
search_response = client.models.Model.Experiment.search(
model_rid,
order_by={"field": "CREATED_TIME", "direction": "DESC"},
page_size=1,
preview=True,
request_timeout=REQUEST_TIMEOUT_SECONDS,
)
logging.info(search_response)
if not search_response.data:
raise RuntimeError(f"No experiments found for {model_rid}.")
experiment_rid = search_response.data[0].rid
series_response = client.models.Model.Experiment.Series.parquet(
model_rid,
experiment_rid,
"feature_importance",
preview=True,
request_timeout=REQUEST_TIMEOUT_SECONDS,
)
feature_series = pl.read_parquet(BytesIO(series_response))
trace = json.loads(feature_series["value"][0])["data"][0]
feature_names = trace["y"]
encoded_scores = trace["x"]
if isinstance(encoded_scores, dict):
importance_scores = np.frombuffer(
base64.b64decode(encoded_scores["bdata"]),
dtype=np.dtype(encoded_scores["dtype"]),
)
else:
importance_scores = np.asarray(encoded_scores, dtype=float)
ranked_features = sorted(
zip(feature_names, importance_scores),
key=lambda pair: pair[1],
reverse=True,
)[:TOP_N]
rows = [
{
"model_rid": model_rid,
"experiment_rid": experiment_rid,
"feature_rank": rank,
"feature_name": str(feature_name),
"feature_importance": float(importance),
}
for rank, (feature_name, importance) in enumerate(ranked_features, start=1)
]
result = pl.DataFrame(
rows,
schema={
"model_rid": pl.String,
"experiment_rid": pl.String,
"feature_rank": pl.Int64,
"feature_name": pl.String,
"feature_importance": pl.Float64,
},
strict=False,
)
output.write_table(result)
Example python script to fetch the top features of an experiment on a model (from local, for testing purposes, hardcoded token)
from urllib.parse import quote
import requests
from io import BytesIO
import polars as pl
import base64
import json
import numpy as np
# Temporary local testing only — do not commit this token.
TOKEN ="eyJwb...ew"
BASE_URL = "https://YOUR_INSTANCE.com"
MODEL_RID = "ri.models.main.model.769d82f8-b9d8-4ba5-a152-68687cc6f18f"
HEADERS = {
"Authorization": f"Bearer {TOKEN}",
"Accept": "application/json",
}
# 1. Search for the latest experiment.
search_response = requests.post(
f"{BASE_URL}/api/v2/models/{MODEL_RID}/experiments/search",
params={"preview": "true"},
headers={
**HEADERS,
"Content-Type": "application/json",
},
json={
"orderBy": {
"field": "CREATED_TIME",
"direction": "DESC",
},
"pageSize": 1,
},
timeout=60,
)
search_response.raise_for_status()
experiments = search_response.json()["data"]
if not experiments:
raise RuntimeError("No experiments were found for this model.")
experiment_rid = experiments[0]["rid"]
print(f"Latest experiment: {experiment_rid}")
# 2. Download the feature-importance series as Parquet.
feature_series = "feature_importance"
encoded_series_name = quote(feature_series, safe="")
series_response = requests.get(
f"{BASE_URL}/api/v2/models/{MODEL_RID}/experiments/{experiment_rid}/series/{encoded_series_name}/parquet",
params={"preview": "true"},
headers={
"Authorization": f"Bearer {TOKEN}",
"Accept": "application/octet-stream",
},
timeout=60,
)
series_response.raise_for_status()
# Reads directly from memory—no temporary file is created.
features = pl.read_parquet(BytesIO(series_response.content))
# 3. Extract the features
trace = json.loads(features["value"][0])["data"][0]
names = trace["y"]
scores = np.frombuffer(
base64.b64decode(trace["x"]["bdata"]),
dtype=trace["x"]["dtype"],
)
top_features = [
name
for name, score in sorted(
zip(names, scores),
key=lambda pair: pair[1],
reverse=True,
)[:20]
]
print(top_features)