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