PowerMind FeatureHub User Guide - Python SDK

The powermind_featurehub Python library allows you to automate FeatureHub workflows from scripts, data science notebooks, batch training jobs, and real-time model serving services.

Import and Client Setup

To connect to FeatureHub, import and instantiate the client:

from powermind_featurehub import FeatureHubClient
from powermind_featurehub import FeatureHubAPIError, FeatureHubAuthError

client = FeatureHubClient(
    base_url="http://localhost:8000",
    timeout=30.0,
    retries=2,
    retry_backoff=0.3,
    verify_ssl=True,
)

Client Parameters

base_url: The address of the FeatureHub web API.
token: Optional JWT token string if already authenticated.
integration_api_key: Optional integration API key for server authentication.
workspace_id: Optional workspace ID to lock the client to a specific project.
timeout: Maximum request duration in seconds (default 30.0).
retries: Number of retries on transient network failures (default 2).
retry_backoff: Backoff multiplier in seconds (default 0.3).
verify_ssl: Verify HTTPS certificates (default True).
ssl_context: Optional custom ssl.SSLContext for private certificate authorities.

For a trusted internal or development deployment that uses a self-signed certificate:

client = FeatureHubClient(
    base_url="https://featurehub.internal",
    integration_api_key="fh_key_xxxxxxxxxxxxxxxxxxxxxxxxxx",
    verify_ssl=False,
)

Keep SSL verification enabled for production/public endpoints. Prefer a custom ssl_context with your private CA when possible.

Authentication Strategies

Best for notebooks, administrative scripts, and experiments:

client = FeatureHubClient("http://localhost:8000")
login_info = client.login("engineer@example.com", "password123")

print(f"Logged in user: {login_info['user']}")
print(f"Active Workspace ID: {client.workspace_id}")

Calling login() sets the client's internal JWT and switches the active workspace_id to the user's default workspace.

2. Service Integration API Key (Recommended for Production)

Best for inference servers and automated pipelines:

# Pass the integration key to the constructor
client = FeatureHubClient(
    base_url="http://localhost:8000",
    integration_api_key="fh_key_xxxxxxxxxxxxxxxxxxxxxxxxxx",
)

With an integration key, you do not need to call login(). The server automatically determines your workspace.

3. Session Logout

To terminate a session and clear local client authentication state:

client.logout()

Calling logout() invalidates the active JWT on the server (recording a logout audit event) and resets the client's internal token, current_user, and workspace_id properties to None.

End-to-End Walkthrough Script

Here is a complete Python script demonstrating how to upload a CSV file, auto-generate features, materialize them, and run an online serving lookup:

from powermind_featurehub import FeatureHubClient

# 1. Initialize and Login
client = FeatureHubClient("http://localhost:8000")
client.login("demo@example.com", "password123")

# 2. Upload CSV Data Source
datasource = client.upload_csv_datasource(
    name="customer-scoring-data",
    file_path="examples/data/customer_features.csv",
)

# 3. Fetch Inferred Schema and Column Suggestions
schema = client.schema_suggestions(datasource["id"])
join_key = schema["recommended_join_key"]
timestamp_col = schema["recommended_event_timestamp_column"]

# Filter out key columns to get candidate feature columns
feature_columns = [
    col["name"] for col in schema["columns"]
    if col["suggested_feature"] and col["name"] not in {join_key, timestamp_col}
]

# 4. Create Feature View
feature_view = client.create_simple_feature_view(
    datasource_id=datasource["id"],
    name="customer_activity",
    join_key=join_key,
    event_timestamp_column=timestamp_col,
    include_columns=feature_columns,
    derived_features=[
        {
            "name": "avg_order_value",
            "expression": "total_spend / nullif(order_count, 0)",
            "dtype": "float64",
        }
    ],
    description="Customer activity features created from script."
)

# 5. Load Features to Online Low-Latency serving database
client.materialize_feature_view(feature_view["id"])

# 6. Bundle Features into a Stable Service Contract
feature_refs = [
    f"{feature_view['name']}:{feature['name']}"
    for feature in feature_view["features"]
]
service = client.create_feature_service(
    name="customer_inference_service",
    feature_refs=feature_refs,
    description="Serving contract for real-time predictions."
)

# 7. Query Real-Time Feature Values
response = client.get_online_features(
    feature_service_name=service["name"],
    entity_rows=[{"customer_id": "C001"}]
)

print("Retrieved online features:")
print(response["results"])

Detailed SDK Method Reference

Authentication Actions

login(identifier, password, workspace_id=None): Authenticates user credentials and stores the JWT.
register_user(email, username, password, full_name=None): Registers a new user account.
logout(): Terminates the user session on the server and clears local auth state.
me(): Retrieves current authenticated user details.

Workspace Actions

create_workspace(name, description=None): Creates a new workspace.
list_workspaces(limit=50, offset=0): Lists workspaces accessible to the user.
set_workspace(workspace_id): Switches the client's current workspace.
sync_workspace(workspace_id=None): Syncs registry objects for the current or provided workspace.
delete_workspace(workspace_id): Deletes the workspace and its contents.

Datasource Actions

list_datasources(limit=50, offset=0): Lists datasources in the active workspace.
register_datasource(name, source_type, connection_config): Registers a remote database or cloud storage file.
upload_csv_datasource(name, file_path): Uploads a local CSV file.
preview_datasource(datasource_id, limit=10): Inspects sample rows.
schema_suggestions(datasource_id, limit=100): Retrieves suggested keys and columns.
ingest_stream(datasource_id, records, sync=True): Ingests streaming records into the staging and online stores. Set sync=False to queue records asynchronously.
delete_datasource(datasource_id): Deletes a datasource and related generated assets.

Feature View Actions

list_feature_views(active_only=True, limit=50, offset=0): Lists feature views.
create_feature_view(payload): Creates a feature view using raw parameters.
create_simple_feature_view(datasource_id, name, join_key, event_timestamp_column, include_columns, derived_features=None, description=None): Simplified helper for stored features.
preview_simple_feature_view(payload, limit=10): Previews a simple feature view before creating it.
preview_feature_view(feature_view_id, limit=10): Previews an existing feature view.
materialize_feature_view(feature_view_id): Loads stored features into the online low-latency serving store.
update_feature_view_schedule(feature_view_id, schedule): Configures background materialization schedule (e.g. "every_5s", "every_30s", "hourly", "daily", or None).
get_feature_view_drift_logs(feature_view_id): Retrieves data drift warning logs and quality checks for the feature view.
reset_feature_view_baseline(feature_view_id): Recalculates and replaces the drift reference baseline with the current live data snapshot. Use this after intentional data changes, schema updates, or model retraining.
update_feature_view_drift_thresholds(feature_view_id, warning_threshold_psi, drift_threshold_psi): Updates the PSI alert thresholds for a feature view. warning_threshold_psi (default 0.10) triggers a yellow warning; drift_threshold_psi (default 0.20) triggers a red critical alert.
activate_feature_view(feature_view_id): Sets a specific version active.
list_feature_view_versions(name_or_base, limit=50, offset=0): Lists versions of a feature view.
get_active_feature_view_offline_source(name_or_base): Gets offline source metadata for integrations.
delete_feature_view(feature_view_id): Deletes a feature view.

Feature Service Actions

list_feature_services(limit=50, offset=0): Lists feature services.
create_feature_service(name, feature_refs, description=None, version=None, activate=True): Bundles features.
activate_feature_service(feature_service_id): Sets a specific version active.
list_feature_service_versions(name_or_base, limit=50, offset=0): Lists service versions.
delete_feature_service(feature_service_id): Deletes a feature service.

Online Serving Actions

get_online_features(feature_service_name=None, entity_rows=None, request_features=None, feature_refs=None): Queries online database and executes real-time calculations.
get_request_only_features(feature_service_name, request_features): Evaluates request-only/on-demand features using provided inputs, without database lookup.

Key and Export Actions

create_integration_key(name, expires_in_days=None): Generates an API key.
list_integration_keys(limit=50, offset=0): Lists integration keys.
revoke_integration_key(key_id): Revokes an integration key.
delete_integration_key(key_id): Deletes an integration key.
create_export(feature_view_name): Starts a background historical export.
list_exports(limit=50, offset=0): Lists exports.
get_export(export_id): Gets export details. Requires workspace_id.
latest_export(feature_view_name): Gets the latest ready export. Requires workspace_id.
download_export(export_id, destination): Downloads a completed export as a CSV.
delete_export(export_id): Deletes an export.

Guidance and Observability Actions

get_feature_view_requirements(): Gets required fields and feature view creation guidance.
get_datasource_strategies(): Gets recommendations for datasets with or without IDs/timestamps.
get_feature_types_guide(): Gets the feature type guide.
get_scaling_guide(): Gets scale() expression guidance.
get_workflow_checklist(): Gets the creation workflow checklist.
get_audit_logs(): Retrieves drift detection and other platform event logs.
get_metrics_text(): Gets Prometheus metrics as plain text.
admin_update_user_password(user_id, new_password): Overrides any user's password (requires administrator privileges).

Data Drift Monitoring

FeatureHub automatically monitors each feature's statistical distribution on every materialization cycle and compares it against a reference baseline. This is designed to give you early warning when real-world data begins to diverge from what your model was trained on.

How the Baseline Works

Event	What Happens
First materialization	FeatureHub captures the full statistical profile of your data (distribution histogram, mean, std, null ratio, row count) and saves it as the reference baseline.
Every subsequent materialization	The current data distribution is compared to that baseline using PSI (Population Stability Index).
Until you reset	The original creation-time baseline is the permanent reference. Your data could change 100 times — it is always compared to that first snapshot.
After `reset_feature_view_baseline()`	The current live data becomes the new baseline going forward.

This model mirrors industry-standard practice: the baseline is stable during a model's lifetime and is only replaced intentionally when you retrain or redeploy a model.

Drift Severity Levels

FeatureHub classifies drift into two severity levels based on the PSI value:

Severity	Default PSI Range	Badge	Meaning
No drift	PSI < 0.10	Stable	Distribution is stable. No action needed.
Warning	0.10 ≤ PSI < 0.20	Drift Warning	Minor shift detected. Monitor closely.
Critical	PSI ≥ 0.20	Drift Critical	Significant shift. Consider retraining.
Low confidence	< 50 rows	Low Confidence	Sample too small for PSI; falls back to mean shift.

These thresholds are the industry defaults used in banking and credit scoring. You can customise them per feature view (see below).

Reading Drift Logs

logs = client.get_feature_view_drift_logs(feature_view_id)
for log in logs["items"]:
    print(f"{log['log_type']} | Feature: {log['feature_name']} | {log['details']}")

Each log entry has a log_type: - data_drift_warning — PSI crossed the warning threshold - data_drift_detected — PSI crossed the critical threshold - data_drift_low_confidence — result based on mean shift due to small sample - schema_drift — a column was added or removed

Resetting the Reference Baseline

Use this after retraining your model on new data, major ETL changes, or intentional schema updates:

# After retraining your model on July data, adopt the current distribution as the new baseline
client.reset_feature_view_baseline(feature_view_id)
print("Baseline reset. Future drift checks compare against current live data.")

Important: This is a permanent, destructive operation. The old baseline is replaced and cannot be recovered. Only reset when you intentionally want to accept the current data distribution as the new normal — for example, after deploying a retrained model.

Customising PSI Thresholds

Different use cases require different sensitivity. Financial fraud models may need tighter thresholds (0.05 / 0.10), while lower-stakes models can use the defaults:

# Tighter thresholds for a financial risk model
client.update_feature_view_drift_thresholds(
    feature_view_id=feature_view_id,
    warning_threshold_psi=0.05,   # Warn earlier
    drift_threshold_psi=0.10,     # Critical earlier
)

# Looser thresholds for a low-stakes recommendation model
client.update_feature_view_drift_thresholds(
    feature_view_id=feature_view_id,
    warning_threshold_psi=0.15,
    drift_threshold_psi=0.30,
)

Thresholds are stored per feature view and take effect on the next materialization cycle. You can also set them from the UI: open the Feature Statistics modal (stats icon on any feature view) and edit the Warning PSI and Critical PSI fields.

Complete Drift Monitoring Example

from powermind_featurehub import FeatureHubClient

client = FeatureHubClient("http://localhost:8000")
client.login("mlops@example.com", "password123")

fv_id = "<your-feature-view-id>"

# 1. Check current drift status
logs = client.get_feature_view_drift_logs(fv_id)
critical = [l for l in logs["items"] if l["log_type"] == "data_drift_detected"]
if critical:
    print(f"WARNING: {len(critical)} features have critical drift — consider retraining.")

# 2. After model retraining and new deployment, adopt current data as new baseline
client.reset_feature_view_baseline(fv_id)
print("Baseline updated to current live distribution.")

# 3. Optionally tighten thresholds for the new model
client.update_feature_view_drift_thresholds(
    feature_view_id=fv_id,
    warning_threshold_psi=0.08,
    drift_threshold_psi=0.15,
)
print("Drift thresholds updated.")

Error Handling

The SDK raises two custom exceptions: 1. FeatureHubAuthError: Raised when requests are executed without credentials. 2. FeatureHubAPIError: Raised when the server returns an HTTP error code.

from powermind_featurehub import FeatureHubAPIError, FeatureHubAuthError

try:
    response = client.get_online_features(
        feature_service_name="customer_inference_service",
        entity_rows=[{"customer_id": "C999"}]
    )
except FeatureHubAuthError:
    print("Authentication failed. Please log in or set an API key.")
except FeatureHubAPIError as exc:
    print(f"API Error ({exc.status_code}): {exc.message}")