# --- Catastrophes minieres britanniques, 1851-1962 (Jarrett 1979) ---
disasters = np.array([
4,5,4,0,1,4,3,4,0,6, 3,3,4,0,2,6,3,3,5,4,
5,3,1,4,4,1,5,5,3,4, 2,5,2,2,3,4,2,1,3,2,
2,1,1,1,1,3,0,0,1,0, 1,1,0,0,3,1,0,3,2,2,
0,1,1,1,0,1,0,1,0,0, 0,2,1,0,0,0,1,1,0,2,
3,3,1,1,2,1,1,1,1,2, 4,2,0,0,1,4,0,0,0,1,
0,0,0,0,1,0,0,1,0,1, 0,1
])
N2 = disasters.size
annees = 1851 + np.arange(N2)
t2 = np.arange(N2)
print(f"Serie catastrophes miniers : {N2} annees ({annees[0]}..{annees[-1]})")
with pm.Model() as modele_disasters:
cp2 = pm.DiscreteUniform("cp2", lower=0, upper=N2 - 1)
early_rate = pm.Exponential("early_rate", lam=1.0) # taux avant (prior faible)
late_rate = pm.Exponential("late_rate", lam=1.0) # taux apres
rate = pm.math.switch(t2 <= cp2, early_rate, late_rate)
pm.Poisson("disasters_obs", mu=rate, observed=disasters)
idata_d = pm.sample(
2000, tune=1500, chains=4, cores=1, random_seed=42,
target_accept=0.95, progressbar=False, idata_kwargs={"log_likelihood": False},
)
cp2_samples = idata_d.posterior["cp2"].values.flatten().astype(int)
cp2_mode = int(np.bincount(cp2_samples, minlength=N2).argmax())
er = float(idata_d.posterior["early_rate"].mean())
lr = float(idata_d.posterior["late_rate"].mean())
print(f"\nAnnee de rupture detectee : {1851 + cp2_mode} (indice {cp2_mode})")
print(f"Taux AVANT : {er:5.2f} catastrophes/an")
print(f"Taux APRES : {lr:5.2f} catastrophes/an")
print(f"Rapport de taux : {er / max(lr, 1e-6):.1f}x")
print()
rapport_convergence(idata_d, "idata_d - catastrophes minieres (cp2 discret en Metropolis, taux en NUTS)")