etsi-watchdog
etsi-watchdog is a lightweight Python library for detecting tabular data drift using statistical tests like PSI. It's modular, production-ready, and ideal for ML validation pipelines and scheduled monitoring tasks.
Installation
pip install etsi-watchdog
Core Components
| Class |
Purpose |
DriftCheck |
Compare reference vs. current dataset using PSI or other drift algorithms. |
Monitor |
Rolling monitoring system to detect drift over time with logging support. |
DriftComparator |
Compares drift results between two different data/model versions. |
DriftCheck API
| Method |
Description |
check.run(current_df, features) |
Runs drift detection on a list of features from current_df vs the reference. |
Monitor API
| Method |
Description |
monitor.enable_logging(path) |
Enable CSV-based logging of drift results to the specified path. |
monitor.watch_rolling(df, window, freq, features) |
Monitors drift over a time-indexed DataFrame. Aggregates and checks drift for each period. |
DriftResult API
| Attribute / Method |
Description |
.is_drifted |
Returns True if the score exceeds threshold. |
.summary() |
Prints a short summary like “Drift Detected” with score and method name. |
.plot() |
Renders a histogram plot of PSI or other scores (requires matplotlib). |
.to_json(path) |
Saves result as a structured JSON file. If no path is passed, returns a dict. |
Example: One-time Check
from etsi.watchdog import DriftCheck
check = DriftCheck(reference_df)
results = check.run(current_df, features=["age", "salary"])
for feature, result in results.items():
print(result.summary())
result.plot()
result.to_json(f"logs/drift_{feature}.json")
Example: Scheduled Monitoring
from etsi.watchdog import Monitor
monitor = Monitor(reference_df)
monitor.enable_logging("logs/rolling_log.csv")
monitor.watch_rolling(
df=live_df,
window=50,
freq="D",
features=["age", "salary"]
)
Example: Version Comparison
from etsi.watchdog import DriftComparator
comp = DriftComparator(run1_results, run2_results)
diff = comp.diff()
for feature, delta in diff.items():
print(f"{feature}: Δ PSI = {delta:+.4f}")
Resources