← Volver al hub para desarrolladores

Python: analizar la calidad del agua con la API de ZipCheckup

Use Python para obtener, analizar y visualizar datos de calidad del agua para cualquier código postal de EE. UU.

Configuración

Instalación de las librerías necesarias:

pip install requests pandas matplotlib

Eso es todo. La API de ZipCheckup es un endpoint REST estándar — no se requiere SDK ni clave de API para el nivel gratuito.

Prueba rápida

import requests

resp = requests.get("https://zipcheckup.com/api/v1/zip/90210")
data = resp.json()

print(f"ZIP: {data['zip']}")
print(f"Score: {data['score']}/100 (Grade: {data['grade']})")
print(f"Violations: {data['violations']}")
print(f"Lead level: {data['lead_level']}")

Salida:

ZIP: 90210
Score: 72/100 (Grade: B)
Violations: 3
Lead level: low

Obtener datos para un solo código postal

Aquí hay una función completa que maneja errores y devuelve un diccionario limpio:

import requests

API_BASE = "https://zipcheckup.com/api/v1"

def get_zip_report(zip_code: str) -> dict | None:
    """Obtener el reporte de ZipCheckup para un solo código postal."""
    url = f"{API_BASE}/zip/{zip_code}"
    resp = requests.get(url, timeout=10)

    if resp.status_code == 404:
        print(f"ZIP {zip_code} not found")
        return None
    if resp.status_code == 429:
        print("Rate limit reached (100/day on free tier)")
        return None

    resp.raise_for_status()
    return resp.json()


# Uso
report = get_zip_report("60614")
if report:
    print(f"{report['zip']}: {report['grade']} ({report['score']})")
    for component, score in report['components'].items():
        print(f"  {component}: {score}")

Obtener varios códigos postales en lote

Para comparar datos entre códigos postales, obténgalos en un bucle con un pequeño retardo para respetar los límites de tasa:

import time
import requests

API_BASE = "https://zipcheckup.com/api/v1"

def get_batch_reports(zip_codes: list[str], delay: float = 0.5) -> list[dict]:
    """Obtener reportes para múltiples códigos postales con limitación de tasa."""
    results = []

    for i, zip_code in enumerate(zip_codes):
        try:
            resp = requests.get(f"{API_BASE}/zip/{zip_code}", timeout=10)
            if resp.status_code == 200:
                results.append(resp.json())
                print(f"[{i+1}/{len(zip_codes)}] {zip_code}: OK")
            elif resp.status_code == 429:
                print(f"Rate limit hit at ZIP #{i+1}. Collected {len(results)} results.")
                break
            else:
                print(f"[{i+1}/{len(zip_codes)}] {zip_code}: HTTP {resp.status_code}")
        except requests.RequestException as e:
            print(f"[{i+1}/{len(zip_codes)}] {zip_code}: Error - {e}")

        if i < len(zip_codes) - 1:
            time.sleep(delay)

    return results


# Ejemplo: comparar barrios de Chicago
chicago_zips = ["60601", "60614", "60622", "60637", "60657", "60660"]
reports = get_batch_reports(chicago_zips)

Construir un DataFrame de pandas

Convierta las respuestas de la API a un formato tabular para el análisis:

import pandas as pd

def reports_to_dataframe(reports: list[dict]) -> pd.DataFrame:
    """Convertir una lista de reportes de ZipCheckup a un DataFrame de pandas."""
    rows = []
    for r in reports:
        row = {
            "zip": r["zip"],
            "score": r["score"],
            "grade": r["grade"],
            "violations": r["violations"],
            "lead_level": r["lead_level"],
            "radon_zone": r["radon_zone"],
            "flood_claims": r["flood_claims"],
        }
        # Aplanar las puntuaciones por componente
        for comp, val in r.get("components", {}).items():
            row[f"comp_{comp}"] = val
        rows.append(row)

    return pd.DataFrame(rows)


df = reports_to_dataframe(reports)
print(df.to_string(index=False))

Salida:

   zip  score grade  violations lead_level  radon_zone  flood_claims  comp_water_quality  comp_lead_risk  comp_radon_risk  comp_flood_risk  comp_air_quality
 60601     65     C           5        low           2           312                  62              80               58               45                68
 60614     72     B           3        low           2           148                  78              85               62               55                71
 60622     68     C           4        low           2           205                  70              82               60               50                69
 60637     41     D           9       high           1           487                  35              42               52               38                55
 60657     74     B           2        low           2           132                  80              88               64               58                72
 60660     70     B           3        low           2           178                  72              84               61               53                70

Operaciones útiles con DataFrame

# Ordenar por puntuación (peor primero)
df.sort_values("score", ascending=True)

# Filtrar a códigos postales de alto riesgo
high_risk = df[df["score"] < 50]

# Puntuación promedio en todos los códigos postales
print(f"Average score: {df['score'].mean():.1f}")

# Distribución de calificaciones
print(df["grade"].value_counts())

# Peor componente en todos los códigos postales
comp_cols = [c for c in df.columns if c.startswith("comp_")]
worst = df[comp_cols].mean().idxmin()
print(f"Weakest component: {worst} ({df[comp_cols].mean().min():.1f})")

Visualización con matplotlib

Gráfico de barras: comparación de códigos postales

import matplotlib.pyplot as plt

def plot_zip_scores(df: pd.DataFrame, title: str = "ZipCheckup Scores"):
    """Crear un gráfico de barras horizontal que compara puntuaciones por código postal."""
    df_sorted = df.sort_values("score", ascending=True)

    colors = []
    for grade in df_sorted["grade"]:
        if grade == "A":
            colors.append("#2E7D32")
        elif grade == "B":
            colors.append("#558B2F")
        elif grade == "C":
            colors.append("#F9A825")
        elif grade == "D":
            colors.append("#EF6C00")
        else:
            colors.append("#C62828")

    fig, ax = plt.subplots(figsize=(10, max(4, len(df) * 0.6)))
    bars = ax.barh(df_sorted["zip"], df_sorted["score"], color=colors, height=0.6)

    # Agregar etiquetas de puntuación
    for bar, score, grade in zip(bars, df_sorted["score"], df_sorted["grade"]):
        ax.text(bar.get_width() + 1, bar.get_y() + bar.get_height()/2,
                f"{score} ({grade})", va="center", fontsize=11)

    ax.set_xlabel("Home Safety Score (0-100)")
    ax.set_title(title)
    ax.set_xlim(0, 105)
    ax.spines["top"].set_visible(False)
    ax.spines["right"].set_visible(False)
    plt.tight_layout()
    plt.savefig("zip_scores.png", dpi=150)
    plt.show()


plot_zip_scores(df, "Chicago Neighborhood Water Quality Scores")

Gráfico de radar: desglose por componente

import numpy as np

def plot_component_radar(report: dict):
    """Trazar un gráfico de radar de las puntuaciones por componente para un solo código postal."""
    components = report["components"]
    labels = [k.replace("_", " ").title() for k in components.keys()]
    values = list(components.values())

    # Cerrar el polígono
    angles = np.linspace(0, 2 * np.pi, len(labels), endpoint=False).tolist()
    values += values[:1]
    angles += angles[:1]

    fig, ax = plt.subplots(figsize=(6, 6), subplot_kw=dict(polar=True))
    ax.fill(angles, values, alpha=0.25, color="#1a6b8a")
    ax.plot(angles, values, color="#1a6b8a", linewidth=2)
    ax.set_xticks(angles[:-1])
    ax.set_xticklabels(labels, fontsize=11)
    ax.set_ylim(0, 100)
    ax.set_title(f"ZIP {report['zip']} — Component Scores", pad=20, fontsize=14)
    plt.tight_layout()
    plt.savefig(f"radar_{report['zip']}.png", dpi=150)
    plt.show()


# Trazar el radar para un solo código postal
plot_component_radar(reports[0])

Gráfico interactivo de Plotly (opcional)

Para tableros web, use Plotly en lugar de matplotlib:

# pip install plotly
import plotly.express as px

fig = px.bar(
    df.sort_values("score"),
    x="score",
    y="zip",
    orientation="h",
    color="grade",
    color_discrete_map={"A": "#2E7D32", "B": "#558B2F", "C": "#F9A825", "D": "#EF6C00", "F": "#C62828"},
    title="ZipCheckup Scores by ZIP Code",
    labels={"score": "Home Safety Score", "zip": "ZIP Code"},
)
fig.update_layout(yaxis={"categoryorder": "total ascending"})
fig.write_html("scores_interactive.html")
fig.show()

Guardar resultados en CSV

def save_reports_csv(df: pd.DataFrame, filename: str = "zipcheckup_results.csv"):
    """Guardar el DataFrame en un archivo CSV."""
    df.to_csv(filename, index=False)
    print(f"Saved {len(df)} records to {filename}")


save_reports_csv(df, "chicago_water_quality.csv")

La salida CSV:

zip,score,grade,violations,lead_level,radon_zone,flood_claims,comp_water_quality,comp_lead_risk,comp_radon_risk,comp_flood_risk,comp_air_quality
60601,65,C,5,low,2,312,62,80,58,45,68
60614,72,B,3,low,2,148,78,85,62,55,71
...

Ejemplo: ¿qué códigos postales de California tienen los peores niveles de plomo?

Aquí hay un script completo que obtiene datos para códigos postales de California e identifica las áreas con mayor riesgo de plomo:

import time
import requests
import pandas as pd

API_BASE = "https://zipcheckup.com/api/v1"

# Muestra de códigos postales de California (ciudades principales)
ca_zips = [
    "90001", "90012", "90210", "90401",  # área de Los Ángeles
    "92101", "92126",                     # San Diego
    "94102", "94110", "94158",            # San Francisco
    "95814",                               # Sacramento
    "93701",                               # Fresno
    "92374",                               # Redlands
    "93305",                               # Bakersfield
    "95202",                               # Stockton
]


def fetch_california_data(zips: list[str]) -> pd.DataFrame:
    """Obtener y analizar códigos postales de California."""
    results = []

    for i, z in enumerate(zips):
        try:
            resp = requests.get(f"{API_BASE}/zip/{z}", timeout=10)
            if resp.status_code == 200:
                data = resp.json()
                results.append({
                    "zip": data["zip"],
                    "score": data["score"],
                    "grade": data["grade"],
                    "lead_level": data["lead_level"],
                    "lead_risk_score": data["components"]["lead_risk"],
                    "violations": data["violations"],
                    "water_quality": data["components"]["water_quality"],
                })
                print(f"[{i+1}/{len(zips)}] {z}: lead_risk={data['components']['lead_risk']}")
            if resp.status_code == 429:
                print("Rate limit — stopping")
                break
        except Exception as e:
            print(f"Error for {z}: {e}")

        time.sleep(0.5)

    return pd.DataFrame(results)


# Obtener los datos
df_ca = fetch_california_data(ca_zips)

# Ordenar por riesgo de plomo (peor primero)
df_ca_sorted = df_ca.sort_values("lead_risk_score", ascending=True)

print("\n=== California ZIPs by Lead Risk (worst first) ===")
print(df_ca_sorted[["zip", "lead_level", "lead_risk_score", "violations"]].to_string(index=False))

# Estadísticas resumidas
print(f"\nTotal ZIPs analyzed: {len(df_ca)}")
print(f"High lead level: {len(df_ca[df_ca['lead_level'] == 'high'])}")
print(f"Average lead risk score: {df_ca['lead_risk_score'].mean():.1f}")
print(f"Worst ZIP: {df_ca_sorted.iloc[0]['zip']} (lead risk: {df_ca_sorted.iloc[0]['lead_risk_score']})")

# Guardar en CSV
df_ca.to_csv("california_lead_analysis.csv", index=False)
print("\nSaved to california_lead_analysis.csv")

Próximos pasos

  • Referencia completa de la API: vea todos los endpoints disponibles, parámetros de consulta y campos de respuesta en /api/docs/
  • Datos a nivel estatal: use /api/v1/state/{code} para obtener estadísticas agregadas de un estado entero
  • Clasificaciones: use /api/v1/rankings/{metric} para obtener clasificaciones nacionales o a nivel estatal
  • Widgets para incrustar: ¿no quiere programar? Incruste una tarjeta de puntuación en su sitio web con una sola línea de HTML
  • Automatizar: conecte con Zapier/Make para flujos de datos sin código

Límites de tasa: el nivel gratuito permite 100 solicitudes por día. Para análisis por lotes más grandes, escriba a [email protected] sobre el acceso al nivel Pro (10,000 solicitudes/día).