Doctor Recommendations Quickstart#
This notebook shows how to get cpd doctor recommendations, fall back to a bundled snapshot when live CLI execution is unavailable, and run a recommended pipeline in Python.
import json
import subprocess
from pathlib import Path
import numpy as np
import cpd
try:
import matplotlib.pyplot as plt
HAS_MATPLOTLIB = True
except ImportError:
HAS_MATPLOTLIB = False
print("matplotlib is optional. Install with: python -m pip install matplotlib")
# Build deterministic seasonal + trend + outlier signal and save CSV for CLI interop
rng = np.random.default_rng(2028)
t = np.arange(360, dtype=np.float64)
seasonal = 6.0 * np.sin(2.0 * np.pi * t / 30.0)
trend = 0.03 * t
level_shift = np.where(t >= 190.0, 7.5, 0.0)
signal = 80.0 + seasonal + trend + level_shift + rng.normal(0.0, 1.3, size=t.shape[0])
for idx in (55, 140, 265, 315):
signal[idx] += 10.0
notebook_data_dir = Path("/Users/admin/Documents/Work/claude-doctor-changepoint/cpd/python/examples/notebooks/data")
notebook_data_dir.mkdir(parents=True, exist_ok=True)
doctor_input_csv = notebook_data_dir / "doctor_series.csv"
np.savetxt(doctor_input_csv, signal.reshape(-1, 1), delimiter=",")
doctor_input_csv
# Attempt live doctor CLI recommendation generation
manifest_path = Path("/Users/admin/Documents/Work/claude-doctor-changepoint/cpd/Cargo.toml")
live_output_path = notebook_data_dir / "doctor_recommendations_live.json"
cmd = [
"cargo",
"run",
"--manifest-path",
str(manifest_path),
"-p",
"cpd-cli",
"--",
"doctor",
"--input",
str(doctor_input_csv),
"--objective",
"balanced",
"--min-confidence",
"0.2",
"--output",
str(live_output_path),
]
doctor_payload = None
doctor_source = None
doctor_error = None
try:
completed = subprocess.run(cmd, check=True, capture_output=True, text=True)
doctor_payload = json.loads(live_output_path.read_text(encoding="utf-8"))
doctor_source = f"live_cli ({live_output_path})"
print("Live doctor command succeeded.")
if completed.stdout.strip():
print(completed.stdout.strip())
except Exception as exc:
doctor_error = str(exc)
print("Live doctor command unavailable; will use fallback snapshot.")
# Fallback snapshot path when live CLI is unavailable
snapshot_path = Path("/Users/admin/Documents/Work/claude-doctor-changepoint/cpd/python/examples/notebooks/data/doctor_recommendations_snapshot.json")
if doctor_payload is None:
doctor_payload = json.loads(snapshot_path.read_text(encoding="utf-8"))
doctor_source = f"snapshot ({snapshot_path})"
print("Recommendation source:", doctor_source)
if doctor_error:
print("Live CLI error:", doctor_error)
# Inspect top recommendations
recommendations = doctor_payload.get("recommendations", [])
if not recommendations:
raise RuntimeError("Doctor output did not include any recommendations")
for rec in recommendations[:3]:
print(
f"rank={rec['rank']} confidence={rec['confidence']:.3f} "
f"detector={rec['pipeline']['detector']} cost={rec['pipeline']['cost']}"
)
if rec["warnings"]:
print(" warnings:", "; ".join(rec["warnings"]))
print(" summary:", rec["explanation"]["summary"])
# Execute top recommended pipeline in Python
top_pipeline = recommendations[0]["pipeline"]
doctor_result = cpd.detect_offline(signal, pipeline=top_pipeline, repro_mode="balanced")
print("top pipeline breakpoints:", doctor_result.breakpoints)
print("algorithm:", doctor_result.diagnostics.algorithm)
print("cost_model:", doctor_result.diagnostics.cost_model)
# Optional: visualize the signal and selected breakpoints
if HAS_MATPLOTLIB:
fig, ax = plt.subplots(figsize=(12, 4))
ax.plot(signal, color="black", linewidth=1.2, alpha=0.85)
for cp in [bp for bp in doctor_result.breakpoints if bp < len(signal)]:
ax.axvline(cp, color="tab:red", linestyle="--", linewidth=1.5, alpha=0.9)
ax.set_title("Doctor-selected pipeline result")
ax.set_xlabel("sample")
ax.set_ylabel("value")
plt.show()
else:
print("Skipping plot (matplotlib not installed).")