Urban Water Network Analytics

Introduction

30 Second Summary

Every time you turn on a tap, treated water travels through miles of underground pipes to reach you. In many cities, nearly a third of that water never reaches a customer because it leaks out through aging, cracked infrastructure.

In this project, you will build a complete analytics platform for a fictional water utility that detects leaks using machine learning, forecasts seasonal demand, and helps operations managers decide where to invest limited repair budgets.

What You'll Build

You'll create an end-to-end data analytics pipeline using Python, PostgreSQL, and Power BI that transforms messy IoT sensor data into actionable infrastructure intelligence.

By the end of this project, you'll have:

• A simulated multi-table IoT dataset representing 2 years of 15-minute interval sensor readings across 12 pressure zones, complete with realistic data quality issues and injected anomalies.
• Machine learning models including K-Means clustering for leak anomaly detection and Prophet time-series forecasting for demand prediction, validated against ground-truth data.
• A production-grade PostgreSQL star schema with analytical views powered by advanced window functions, CTEs, and recursive queries, feeding an enterprise Power BI dashboard with what-if investment simulation.
• Secret Mission: Implement Row-Level Security so each zone manager sees only their data, plus build a recursive CTE that traces leak propagation paths through the pipe network graph.

Are there any prerequisites?

This project assumes intermediate Python skills (comfortable with pandas and numpy) and basic SQL knowledge (SELECT, JOIN, WHERE).

You'll also need Power BI Desktop installed, which is available free on Windows.

Before We Start

Before you can detect leaks and forecast demand for AquaFlow Utilities, you need a working analytics environment. This means a Python environment with the right data science packages, a PostgreSQL database ready to receive your star schema, and an organized project folder structure.

In this step, you will create all three foundations so that every later step can focus on the actual analysis.

In this step, get ready to:

• Create a Python virtual environment and install all required data science packages.
• Set up a PostgreSQL database for your star schema.
• Build the project folder structure and verify your Jupyter Notebook imports.

Set up your Python environment

A virtual environment keeps your project's packages isolated from other Python projects on your machine. This prevents version conflicts between projects.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir aquaflow-analytics
cd aquaflow-analytics

• Create a virtual environment by running:

macOS

python3 -m venv .venv

Windows

python -m venv .venv

• Activate the virtual environment:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

You should see (.venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

What is a virtual environment?

A virtual environment is an isolated Python installation that lives inside your project folder. Packages you install here won't interfere with other Python projects on your machine.

• Install all required packages by running this command:

pip install pandas==3.0.3 numpy scikit-learn==1.9.0 prophet==1.3.0 matplotlib seaborn sqlalchemy psycopg2-binary

What are these packages?

• pandas and numpy handle data manipulation and numerical operations.
• scikit-learn provides the K-Means clustering algorithm for leak detection.
• prophet is Facebook's time-series forecasting library for demand prediction.
• matplotlib and seaborn create data visualizations.
• sqlalchemy and psycopg2-binary connect Python to your PostgreSQL database.

The installation may take a few minutes as Prophet downloads its dependencies.

Installation failing?

Make sure your virtual environment is active. You should see (.venv) in your terminal prompt.

If Prophet fails to install, try upgrading pip first with pip install --upgrade pip and then re-run the install command.

Help me troubleshoot my package installation

Create the PostgreSQL database

Your analytics platform needs a dedicated PostgreSQL database to store the star schema you will build in a later step.

• Create the aquaflow_analytics database by running this command:

createdb aquaflow_analytics

What does this command do?

createdb is a PostgreSQL command-line utility that creates a new database on your local server. It connects to your default PostgreSQL instance and creates the database with your system user as the owner.

• Verify the database exists by connecting to it with psql:

psql -d aquaflow_analytics -c "SELECT current_database();"

You should see aquaflow_analytics printed in the output. This confirms your database is ready.

Connection refused or role does not exist?

Make sure the PostgreSQL service is running on your machine. On macOS, check with brew services list. On Windows, check that the PostgreSQL 17 service is running in Services.

If you see a "role does not exist" error, you may need to specify your PostgreSQL superuser with createdb -U postgres aquaflow_analytics.

Help me connect to my PostgreSQL database

Create the project folder structure and verify imports

A clean folder structure keeps your data, notebooks, SQL scripts, and Power BI files organized and easy to find.

• Create the four subdirectories by running:

mkdir data notebooks sql powerbi

• Confirm the folders exist by running:

ls

You should see data, notebooks, sql, and powerbi in the output alongside your .venv folder.

• Create the main notebook file inside the notebooks directory by running:

touch notebooks/aquaflow_analysis.ipynb

• Open notebooks/aquaflow_analysis.ipynb in your code editor.
• Create a new code cell and add the following imports to verify everything is installed correctly:

# Core data manipulation
import pandas as pd
import numpy as np

# Machine learning - clustering for leak detection
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

# Time-series forecasting
from prophet import Prophet

# Visualization
import matplotlib.pyplot as plt
import seaborn as sns

# Database connection
from sqlalchemy import create_engine
import psycopg2

print("All imports successful!")
print(f"pandas version: {pd.__version__}")
print(f"numpy version: {np.__version__}")

What does this code do?

This cell imports every library your project needs and prints version numbers to confirm the installations are working. If any import fails, you will see an error pointing to the specific package that needs fixing.

• KMeans and StandardScaler from scikit-learn will power the leak anomaly detection.
• Prophet will forecast water demand per pressure zone.
• SQLAlchemy provides the bridge between your Python DataFrames and PostgreSQL.

• Run the cell.

You should see "All imports successful!" followed by the pandas and numpy version numbers.

Seeing an import error?

If a specific package fails to import, re-run the pip install command from earlier. Make sure your virtual environment is still active (check for (.venv) in your terminal prompt).

If your editor does not recognize the virtual environment, select it as your Python interpreter. In VS Code, press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows), type "Python: Select Interpreter", and choose the .venv option.

Help me fix my import errors

✔️ Awesome, I've got everything!

Great. Make sure you have saved your notebook file before moving on.

ⓧ I'd like to double check the full code

Here is the complete import cell your notebook should contain:

# Core data manipulation
import pandas as pd
import numpy as np

# Machine learning - clustering for leak detection
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

# Time-series forecasting
from prophet import Prophet

# Visualization
import matplotlib.pyplot as plt
import seaborn as sns

# Database connection
from sqlalchemy import create_engine
import psycopg2

print("All imports successful!")
print(f"pandas version: {pd.__version__}")
print(f"numpy version: {np.__version__}")

Your analytics environment is ready. Next up, you will generate the messy IoT dataset that simulates two years of sensor readings across AquaFlow's pipe network.

Generate the Multi-Table IoT Dataset

Your environment is ready, and your notebook is connected to all the libraries you need. Now it is time to build the foundation of this entire project: the dataset itself.

Real IoT sensor data from a water utility is messy. Sensors drop offline, timestamps drift between timezones, readings go negative when hardware malfunctions, and pressure drops silently indicate leaks. You will simulate all of this intentionally so that the cleaning, clustering, and forecasting steps later have realistic challenges to solve.

In this step, get ready to:

• Generate 3 dimension tables describing the network's sensors, pipe segments, and pressure zones.
• Generate 3 fact tables with intentional data quality issues simulating real-world sensor failures.
• Export all 7 CSV files to your data directory with a summary of injected issues.

Generate the dimension tables

Dimension tables describe the static attributes of your network. They answer "what is this sensor?" and "where is this pipe?" rather than recording measurements over time.

• In your aquaflow_analysis.ipynb notebook, create a new cell and set up the random seed and configuration constants by pasting the following code:

import numpy as np
import pandas as pd
from datetime import datetime, timedelta

# Set random seed for reproducible data generation
np.random.seed(42)

# Configuration constants for the AquaFlow network
NUM_ZONES = 12
NUM_PIPE_SEGMENTS = 500
NUM_SENSORS = 200
NUM_MAINTENANCE_LOGS = 800
LEAK_SEGMENTS = 45  # segments with injected pressure drop anomalies
NULL_RATE = 0.08    # 8% random NULLs in sensor readings

print("Configuration loaded successfully.")
print(f"Generating data for {NUM_ZONES} zones, {NUM_PIPE_SEGMENTS} pipes, {NUM_SENSORS} sensors.")

What does this code do?

This cell locks in a random seed so your generated data is reproducible every time you re-run the notebook. The constants at the top define the scale of the simulated network and will be reused across all generator functions.

• Run the cell. You should see the configuration confirmation message printed below the cell.

• In a new cell, generate the dim_zones table representing 12 pressure zones by pasting this code:

# Generate dim_zones: 12 pressure zones across the AquaFlow service area
zone_names = [
    'Downtown Core', 'Industrial Park', 'Northside Residential',
    'Southside Residential', 'Eastwood Heights', 'Westlake District',
    'Harbor Front', 'University Quarter', 'Airport Zone',
    'Hillcrest Estates', 'Valley Gardens', 'Riverside Commons'
]

dim_zones = pd.DataFrame({
    'zone_id': range(1, NUM_ZONES + 1),
    'zone_name': zone_names,
    'population_served': np.random.randint(12000, 35000, size=NUM_ZONES),
    'elevation_m': np.round(np.random.uniform(15, 180, size=NUM_ZONES), 1),
    'reservoir_id': [f'RES-{i:02d}' for i in range(1, NUM_ZONES + 1)]
})

print(f"dim_zones: {len(dim_zones)} rows")
dim_zones.head()

What does this code do?

Each zone represents a distinct pressure area in the water network, serving a population between 12,000 and 35,000 customers. Elevation matters because water pressure is directly affected by height differences between the reservoir and the delivery point.

• Run the cell. You should see "dim_zones: 12 rows" printed, followed by a preview table showing the first 5 zones.

• In a new cell, generate the dim_pipe_segments table with network graph relationships by pasting this code:

# Generate dim_pipe_segments: 500 pipe segments forming a network graph
materials = ['cast_iron', 'PVC', 'ductile_iron', 'asbestos_cement']
material_weights = [0.35, 0.30, 0.25, 0.10]  # cast iron dominates older networks

dim_pipe_segments = pd.DataFrame({
    'segment_id': range(1, NUM_PIPE_SEGMENTS + 1),
    'material': np.random.choice(materials, size=NUM_PIPE_SEGMENTS, p=material_weights),
    'diameter_mm': np.random.choice([100, 150, 200, 250, 300, 400, 500], size=NUM_PIPE_SEGMENTS),
    'length_m': np.round(np.random.uniform(50, 800, size=NUM_PIPE_SEGMENTS), 1),
    'installation_year': np.random.randint(1955, 2021, size=NUM_PIPE_SEGMENTS),
    'zone_id': np.random.randint(1, NUM_ZONES + 1, size=NUM_PIPE_SEGMENTS)
})

# Create upstream/downstream relationships for network graph traversal
upstream = np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_PIPE_SEGMENTS)
downstream = np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_PIPE_SEGMENTS)

# Avoid self-references (a pipe cannot connect to itself)
for i in range(NUM_PIPE_SEGMENTS):
    while upstream[i] == i + 1:
        upstream[i] = np.random.randint(1, NUM_PIPE_SEGMENTS + 1)
    while downstream[i] == i + 1:
        downstream[i] = np.random.randint(1, NUM_PIPE_SEGMENTS + 1)

# ~10% of segments are terminal (no upstream or downstream connection)
null_mask_up = np.random.random(NUM_PIPE_SEGMENTS) < 0.10
null_mask_down = np.random.random(NUM_PIPE_SEGMENTS) < 0.10

dim_pipe_segments['upstream_segment_id'] = upstream.astype(float)
dim_pipe_segments.loc[null_mask_up, 'upstream_segment_id'] = np.nan
dim_pipe_segments['downstream_segment_id'] = downstream.astype(float)
dim_pipe_segments.loc[null_mask_down, 'downstream_segment_id'] = np.nan

print(f"dim_pipe_segments: {len(dim_pipe_segments)} rows")
print(f"Materials distribution:\n{dim_pipe_segments['material'].value_counts()}")
print(f"\nTerminal segments (no upstream): {null_mask_up.sum()}")
print(f"Terminal segments (no downstream): {null_mask_down.sum()}")
dim_pipe_segments.head()

Why build a network graph?

The upstream and downstream columns let you traverse the pipe network as a directed graph. In Step 7, you will write a recursive CTE in PostgreSQL that follows these relationships to trace how a leak in one segment propagates pressure loss downstream.

Terminal segments (those with NaN connections) represent endpoints like dead-end streets or reservoir intake points.

• Run the cell. You should see "dim_pipe_segments: 500 rows" followed by the material distribution counts showing cast_iron as the most common.

• In a new cell, generate the dim_sensors table by pasting this code:

# Generate dim_sensors: 200 sensors deployed across the pipe network
sensor_types = ['pressure', 'flow', 'temperature']
type_weights = [0.40, 0.40, 0.20]  # pressure and flow are most critical
manufacturers = ['AquaSense', 'FlowTech', 'HydroMetrics', 'PipeStar']
calibration_statuses = ['calibrated', 'due_calibration', 'failed']

# Generate random installation dates spanning 2018-2022
install_start = pd.Timestamp('2018-01-01')
install_dates = install_start + pd.to_timedelta(
    np.random.randint(0, 1500, size=NUM_SENSORS), unit='D'
)

dim_sensors = pd.DataFrame({
    'sensor_id': range(1, NUM_SENSORS + 1),
    'sensor_type': np.random.choice(sensor_types, size=NUM_SENSORS, p=type_weights),
    'zone_id': np.random.randint(1, NUM_ZONES + 1, size=NUM_SENSORS),
    'pipe_segment_id': np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_SENSORS),
    'installation_date': install_dates.strftime('%Y-%m-%d'),
    'manufacturer': np.random.choice(manufacturers, size=NUM_SENSORS),
    'calibration_status': np.random.choice(
        calibration_statuses, size=NUM_SENSORS, p=[0.70, 0.20, 0.10]
    )
})

print(f"dim_sensors: {len(dim_sensors)} rows")
print(f"Sensor types:\n{dim_sensors['sensor_type'].value_counts()}")
print(f"\nCalibration status:\n{dim_sensors['calibration_status'].value_counts()}")
dim_sensors.head()

What does this code do?

Each sensor is assigned a type (pressure, flow, or temperature), a physical location on a pipe segment, and a calibration status. The 10% of sensors with 'failed' calibration status will be useful later when investigating suspicious readings.

• Run the cell. You should see "dim_sensors: 200 rows" with type counts showing roughly 80 pressure sensors, 80 flow sensors, and 40 temperature sensors.

Seeing a NameError?

If you see an error about NUM_SENSORS or NUM_ZONES not being defined, make sure you ran the first configuration cell before this one. Jupyter cells must be executed in order for variables to be available.

help me fix a NameError in my Jupyter notebook where constants from an earlier cell are not defined

Generate the fact tables with intentional messiness

Fact tables record measurements and events over time. The sensor readings table is the largest and most important. You will intentionally inject multiple types of data quality issues that mirror real-world IoT failures.

• In a new cell, generate the base sensor readings data by pasting this code:

# Generate fact_sensor_readings: ~2M rows of 15-min interval data over 2 years
print("Generating ~2M sensor readings. This may take 1-2 minutes...")

# Create the full 2-year timestamp range at 15-minute intervals
start_date = pd.Timestamp('2022-01-01')
end_date = pd.Timestamp('2023-12-31 23:45:00')
timestamps_full = pd.date_range(start_date, end_date, freq='15min')
print(f"Total available timestamps: {len(timestamps_full):,}")

# Each sensor gets ~10,000 readings (contiguous time blocks)
readings_per_sensor = 10000

all_readings = []
for sensor_id in range(1, NUM_SENSORS + 1):
    # Pick a random starting point within the 2-year window
    max_start = len(timestamps_full) - readings_per_sensor
    start_idx = np.random.randint(0, max_start)
    sensor_timestamps = timestamps_full[start_idx:start_idx + readings_per_sensor]

    # Look up this sensor's type to generate appropriate readings
    sensor_type = dim_sensors.loc[
        dim_sensors['sensor_id'] == sensor_id, 'sensor_type'
    ].values[0]

    # Generate realistic readings based on sensor type
    reading_data = {
        'reading_id': range(
            (sensor_id - 1) * readings_per_sensor + 1,
            sensor_id * readings_per_sensor + 1
        ),
        'sensor_id': sensor_id,
        'timestamp': sensor_timestamps,
        'pressure_bar': np.nan,
        'flow_lps': np.nan,
        'temperature_c': np.nan
    }

    if sensor_type == 'pressure':
        # Normal pressure: 3.5 - 5.5 bar with daily variation
        reading_data['pressure_bar'] = np.random.normal(4.5, 0.5, readings_per_sensor)
    elif sensor_type == 'flow':
        # Normal flow: 10-25 liters per second
        reading_data['flow_lps'] = np.random.normal(15, 3, readings_per_sensor)
    else:
        # Temperature: 8-18 degrees C with seasonal variation
        reading_data['temperature_c'] = np.random.normal(12, 3, readings_per_sensor)

    all_readings.append(pd.DataFrame(reading_data))

# Combine all sensor readings into one DataFrame
fact_sensor_readings = pd.concat(all_readings, ignore_index=True)
print(f"\nBase readings generated: {len(fact_sensor_readings):,} rows")
print(f"Date range: {fact_sensor_readings['timestamp'].min()} to {fact_sensor_readings['timestamp'].max()}")

What does this code do?

This generates approximately 2 million rows of sensor data. Each sensor gets a contiguous block of 10,000 readings at 15-minute intervals, placed randomly within the 2-year window (2022-2024).

Only the column matching each sensor's type receives actual values. Pressure sensors produce pressure readings, flow sensors produce flow readings, and temperature sensors produce temperature readings. The other columns remain NaN by design.

• Run the cell. This takes 1-2 minutes depending on your machine. You should see "Base readings generated: 2,000,000 rows" when it finishes.

• In a new cell, inject all the intentional data quality issues by pasting this code:

# Inject data quality issues to simulate real-world IoT messiness
print("Injecting data quality issues...")

# Issue 1: 8% random NULLs across measurement columns
null_mask = np.random.random(len(fact_sensor_readings)) < NULL_RATE
for col in ['pressure_bar', 'flow_lps', 'temperature_c']:
    col_mask = null_mask & fact_sensor_readings[col].notna()
    fact_sensor_readings.loc[col_mask, col] = np.nan
nulls_injected = null_mask.sum()
print(f"  - Random NULLs injected: {nulls_injected:,} ({NULL_RATE*100:.0f}% of rows)")

# Issue 2: 3 sensors with week-long gaps (simulating hardware failure)
gap_sensors = np.random.choice(range(1, NUM_SENSORS + 1), size=3, replace=False)
for sensor_id in gap_sensors:
    sensor_mask = fact_sensor_readings['sensor_id'] == sensor_id
    sensor_indices = fact_sensor_readings[sensor_mask].index
    # Remove a week of readings (7 days * 96 readings/day = 672 readings)
    gap_start = np.random.randint(1000, 8000)
    gap_indices = sensor_indices[gap_start:gap_start + 672]
    fact_sensor_readings = fact_sensor_readings.drop(gap_indices)
print(f"  - Week-long gaps created for sensors: {gap_sensors.tolist()}")

# Issue 3: Pressure drops of 15-40% on 45 specific segments (simulating leaks)
leak_segment_ids = np.random.choice(
    range(1, NUM_PIPE_SEGMENTS + 1), size=LEAK_SEGMENTS, replace=False
)
# Find sensors on these leak segments
leak_sensor_ids = dim_sensors[
    dim_sensors['pipe_segment_id'].isin(leak_segment_ids) &
    (dim_sensors['sensor_type'] == 'pressure')
]['sensor_id'].values

for sensor_id in leak_sensor_ids:
    sensor_mask = fact_sensor_readings['sensor_id'] == sensor_id
    drop_factor = np.random.uniform(0.60, 0.85)  # 15-40% pressure reduction
    fact_sensor_readings.loc[sensor_mask, 'pressure_bar'] *= drop_factor
print(f"  - Pressure drop anomalies injected on {LEAK_SEGMENTS} segments ({len(leak_sensor_ids)} sensors affected)")

# Issue 4: Timezone inconsistencies (30% of sensors report in UTC+5 instead of UTC)
tz_offset_sensors = np.random.choice(
    range(1, NUM_SENSORS + 1), size=int(NUM_SENSORS * 0.30), replace=False
)
for sensor_id in tz_offset_sensors:
    sensor_mask = fact_sensor_readings['sensor_id'] == sensor_id
    fact_sensor_readings.loc[sensor_mask, 'timestamp'] += pd.Timedelta(hours=5)
print(f"  - Timezone offset applied to {len(tz_offset_sensors)} sensors (UTC+5 instead of UTC)")

# Issue 5: Duplicate timestamps (~0.5% of rows duplicated)
num_duplicates = int(len(fact_sensor_readings) * 0.005)
duplicate_indices = np.random.choice(
    fact_sensor_readings.index, size=num_duplicates, replace=False
)
duplicates = fact_sensor_readings.loc[duplicate_indices].copy()
fact_sensor_readings = pd.concat([fact_sensor_readings, duplicates], ignore_index=True)
print(f"  - Duplicate timestamps added: {num_duplicates:,} rows")

# Issue 6: Negative flow values (sensor malfunction on ~1% of flow readings)
flow_mask = fact_sensor_readings['flow_lps'].notna()
flow_indices = fact_sensor_readings[flow_mask].index
neg_count = int(len(flow_indices) * 0.01)
neg_indices = np.random.choice(flow_indices, size=neg_count, replace=False)
fact_sensor_readings.loc[neg_indices, 'flow_lps'] *= -1
print(f"  - Negative flow values injected: {neg_count:,} readings")

# Reset the index after all modifications
fact_sensor_readings = fact_sensor_readings.reset_index(drop=True)
print(f"\nFinal fact_sensor_readings: {len(fact_sensor_readings):,} rows")

Why inject these specific issues?

Each injected issue maps to a real problem that water utilities face daily. Random NULLs come from packet loss over cellular networks. Week-long gaps happen when a sensor's battery dies. Timezone drift occurs when maintenance crews replace hardware without syncing the clock.

The 45 segments with pressure drops are your ground truth for leak detection. In Step 4, you will use K-Means clustering to find these segments without knowing the answer in advance. Storing leak_segment_ids lets you validate your model's accuracy later.

• Run the cell. You should see each issue type printed with counts, ending with the final row count (approximately 2 million).

• In a new cell, generate the remaining two fact tables by pasting this code:

# Generate fact_maintenance_logs: 800 maintenance events over 2 years
event_types = ['repair', 'inspection', 'replacement']
event_weights = [0.45, 0.40, 0.15]  # repairs most common, replacements rare

maint_dates = pd.date_range('2022-01-01', '2023-12-31', periods=NUM_MAINTENANCE_LOGS)

fact_maintenance_logs = pd.DataFrame({
    'log_id': range(1, NUM_MAINTENANCE_LOGS + 1),
    'segment_id': np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_MAINTENANCE_LOGS),
    'event_type': np.random.choice(event_types, size=NUM_MAINTENANCE_LOGS, p=event_weights),
    'event_date': maint_dates.strftime('%Y-%m-%d'),
    'cost': np.round(np.random.uniform(500, 25000, size=NUM_MAINTENANCE_LOGS), 2),
    'crew_id': [f'CREW-{np.random.randint(1, 16):02d}' for _ in range(NUM_MAINTENANCE_LOGS)],
    'notes': np.random.choice([
        'Routine inspection completed',
        'Emergency repair - pipe burst',
        'Scheduled replacement - end of life',
        'Leak detected during patrol',
        'Valve replacement',
        'Pressure regulator adjustment',
        'Corrosion treatment applied',
        'Joint seal replacement'
    ], size=NUM_MAINTENANCE_LOGS)
})

print(f"fact_maintenance_logs: {len(fact_maintenance_logs)} rows")
print(f"Event types:\n{fact_maintenance_logs['event_type'].value_counts()}")

# Generate fact_customer_consumption: monthly consumption per zone with seasonality
months = pd.date_range('2022-01', '2023-12', freq='MS')  # month starts
consumption_rows = []

for zone_id in range(1, NUM_ZONES + 1):
    base_consumption = np.random.randint(50000, 150000)  # base m3 per month
    customer_count = dim_zones.loc[
        dim_zones['zone_id'] == zone_id, 'population_served'
    ].values[0] // 3  # approximate households

    for month in months:
        # Add seasonal variation (higher in summer months)
        seasonal_factor = 1.0 + 0.3 * np.sin(2 * np.pi * (month.month - 1) / 12)
        noise = np.random.normal(1.0, 0.05)
        monthly_consumption = int(base_consumption * seasonal_factor * noise)

        consumption_rows.append({
            'zone_id': zone_id,
            'year_month': month.strftime('%Y-%m'),
            'consumption_m3': monthly_consumption,
            'customer_count': customer_count
        })

fact_customer_consumption = pd.DataFrame(consumption_rows)
print(f"\nfact_customer_consumption: {len(fact_customer_consumption)} rows")
print(f"Months covered: {fact_customer_consumption['year_month'].nunique()}")
fact_customer_consumption.head()

What does this code do?

The maintenance logs table records 800 events spread across the 2-year period. Costs range from $500 (minor inspections) to $25,000 (full segment replacements). These will correlate with pipe age and material in your risk scoring model.

The customer consumption table uses a sine function to create realistic seasonal demand patterns. Summer months peak at 30% above the annual average. This seasonality is what Prophet will decompose in Step 5.

• Run the cell. You should see "fact_maintenance_logs: 800 rows" and "fact_customer_consumption: 288 rows" (12 zones times 24 months).

Cell taking too long or running out of memory?

If the sensor readings generation cell stalls, try restarting the kernel and running all cells again from the top. Jupyter can accumulate memory from failed partial runs. Make sure no other heavy applications are consuming memory.

help me debug a Jupyter notebook that is running out of memory when generating a large DataFrame

Export all datasets to CSV

With all 7 tables generated in memory, you need to persist them to disk so they survive kernel restarts and can be loaded into PostgreSQL later.

• In a new cell, export all tables and print the data quality summary by pasting this code:

# Export all datasets to the /data directory
import os

data_dir = 'data'
os.makedirs(data_dir, exist_ok=True)

# Write dimension tables
dim_zones.to_csv(f'{data_dir}/dim_zones.csv', index=False)
dim_pipe_segments.to_csv(f'{data_dir}/dim_pipe_segments.csv', index=False)
dim_sensors.to_csv(f'{data_dir}/dim_sensors.csv', index=False)

# Write fact tables
fact_sensor_readings.to_csv(f'{data_dir}/fact_sensor_readings.csv', index=False)
fact_maintenance_logs.to_csv(f'{data_dir}/fact_maintenance_logs.csv', index=False)
fact_customer_consumption.to_csv(f'{data_dir}/fact_customer_consumption.csv', index=False)

# Also save the leak segment IDs for validation in Step 4
pd.DataFrame({'segment_id': leak_segment_ids}).to_csv(
    f'{data_dir}/ground_truth_leaks.csv', index=False
)

print("All 7 datasets exported to /data directory.")
print(f"\n{'='*60}")
print("DATA QUALITY SUMMARY - Injected Issues")
print(f"{'='*60}")
print(f"Total sensor readings:        {len(fact_sensor_readings):>12,}")
print(f"Random NULLs (8% rate):       {nulls_injected:>12,}")
print(f"Sensors with week-long gaps:  {len(gap_sensors):>12}")
print(f"Segments with leak anomalies: {LEAK_SEGMENTS:>12}")
print(f"Sensors with TZ offset:       {len(tz_offset_sensors):>12}")
print(f"Duplicate rows added:         {num_duplicates:>12,}")
print(f"Negative flow values:         {neg_count:>12,}")
print(f"{'='*60}")
print(f"\nDimension tables:")
print(f"  dim_zones:              {len(dim_zones):>6} rows")
print(f"  dim_pipe_segments:      {len(dim_pipe_segments):>6} rows")
print(f"  dim_sensors:            {len(dim_sensors):>6} rows")
print(f"\nFact tables:")
print(f"  fact_sensor_readings:   {len(fact_sensor_readings):>12,} rows")
print(f"  fact_maintenance_logs:  {len(fact_maintenance_logs):>6} rows")
print(f"  fact_customer_consumption: {len(fact_customer_consumption):>4} rows")

What does this code do?

This exports all tables as CSV files for portability and later loading into PostgreSQL. The ground truth leaks file is a bonus export that stores which segments have injected anomalies. You will use it in Step 4 to validate your K-Means clustering accuracy.

The summary printout gives you a reference card of every data quality issue you need to address in Step 3 (data cleaning). Keep this output visible as you work through the cleaning logic.

• Run the cell. You should see the full data quality summary table printed, confirming all 7 files were exported.
• Verify the CSV files exist by running this in a new cell:

# Verify all files were created successfully
for filename in os.listdir(data_dir):
    filepath = os.path.join(data_dir, filename)
    size_mb = os.path.getsize(filepath) / (1024 * 1024)
    print(f"  {filename:40s} {size_mb:>8.2f} MB")

• Run the cell. You should see all CSV files listed with their sizes. The fact_sensor_readings.csv file should be the largest (roughly 80-120 MB).

Not seeing all 7 files?

If some files are missing, check that you ran every cell above in order. Each cell depends on variables created by previous cells. Try selecting **Kernel** then **Restart & Run All** to execute the entire notebook from scratch.

help me figure out why some of my CSV exports are missing from the data directory

✔️ Awesome, I've got everything!

Great. Double check you have saved your notebook before moving on.

ⓧ I'd like to double check the full code

Here is the complete code for all cells in this step combined. Compare it against your notebook:

import numpy as np
import pandas as pd
from datetime import datetime, timedelta

# Set random seed for reproducible data generation
np.random.seed(42)

# Configuration constants for the AquaFlow network
NUM_ZONES = 12
NUM_PIPE_SEGMENTS = 500
NUM_SENSORS = 200
NUM_MAINTENANCE_LOGS = 800
LEAK_SEGMENTS = 45  # segments with injected pressure drop anomalies
NULL_RATE = 0.08    # 8% random NULLs in sensor readings

print("Configuration loaded successfully.")
print(f"Generating data for {NUM_ZONES} zones, {NUM_PIPE_SEGMENTS} pipes, {NUM_SENSORS} sensors.")

# Generate dim_zones: 12 pressure zones across the AquaFlow service area
zone_names = [
    'Downtown Core', 'Industrial Park', 'Northside Residential',
    'Southside Residential', 'Eastwood Heights', 'Westlake District',
    'Harbor Front', 'University Quarter', 'Airport Zone',
    'Hillcrest Estates', 'Valley Gardens', 'Riverside Commons'
]

dim_zones = pd.DataFrame({
    'zone_id': range(1, NUM_ZONES + 1),
    'zone_name': zone_names,
    'population_served': np.random.randint(12000, 35000, size=NUM_ZONES),
    'elevation_m': np.round(np.random.uniform(15, 180, size=NUM_ZONES), 1),
    'reservoir_id': [f'RES-{i:02d}' for i in range(1, NUM_ZONES + 1)]
})

print(f"dim_zones: {len(dim_zones)} rows")
dim_zones.head()

# Generate dim_pipe_segments: 500 pipe segments forming a network graph
materials = ['cast_iron', 'PVC', 'ductile_iron', 'asbestos_cement']
material_weights = [0.35, 0.30, 0.25, 0.10]  # cast iron dominates older networks

dim_pipe_segments = pd.DataFrame({
    'segment_id': range(1, NUM_PIPE_SEGMENTS + 1),
    'material': np.random.choice(materials, size=NUM_PIPE_SEGMENTS, p=material_weights),
    'diameter_mm': np.random.choice([100, 150, 200, 250, 300, 400, 500], size=NUM_PIPE_SEGMENTS),
    'length_m': np.round(np.random.uniform(50, 800, size=NUM_PIPE_SEGMENTS), 1),
    'installation_year': np.random.randint(1955, 2021, size=NUM_PIPE_SEGMENTS),
    'zone_id': np.random.randint(1, NUM_ZONES + 1, size=NUM_PIPE_SEGMENTS)
})

# Create upstream/downstream relationships for network graph traversal
upstream = np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_PIPE_SEGMENTS)
downstream = np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_PIPE_SEGMENTS)

# Avoid self-references (a pipe cannot connect to itself)
for i in range(NUM_PIPE_SEGMENTS):
    while upstream[i] == i + 1:
        upstream[i] = np.random.randint(1, NUM_PIPE_SEGMENTS + 1)
    while downstream[i] == i + 1:
        downstream[i] = np.random.randint(1, NUM_PIPE_SEGMENTS + 1)

# ~10% of segments are terminal (no upstream or downstream connection)
null_mask_up = np.random.random(NUM_PIPE_SEGMENTS) < 0.10
null_mask_down = np.random.random(NUM_PIPE_SEGMENTS) < 0.10

dim_pipe_segments['upstream_segment_id'] = upstream.astype(float)
dim_pipe_segments.loc[null_mask_up, 'upstream_segment_id'] = np.nan
dim_pipe_segments['downstream_segment_id'] = downstream.astype(float)
dim_pipe_segments.loc[null_mask_down, 'downstream_segment_id'] = np.nan

print(f"dim_pipe_segments: {len(dim_pipe_segments)} rows")
print(f"Materials distribution:\n{dim_pipe_segments['material'].value_counts()}")
print(f"\nTerminal segments (no upstream): {null_mask_up.sum()}")
print(f"Terminal segments (no downstream): {null_mask_down.sum()}")
dim_pipe_segments.head()

# Generate dim_sensors: 200 sensors deployed across the pipe network
sensor_types = ['pressure', 'flow', 'temperature']
type_weights = [0.40, 0.40, 0.20]  # pressure and flow are most critical
manufacturers = ['AquaSense', 'FlowTech', 'HydroMetrics', 'PipeStar']
calibration_statuses = ['calibrated', 'due_calibration', 'failed']

# Generate random installation dates spanning 2018-2022
install_start = pd.Timestamp('2018-01-01')
install_dates = install_start + pd.to_timedelta(
    np.random.randint(0, 1500, size=NUM_SENSORS), unit='D'
)

dim_sensors = pd.DataFrame({
    'sensor_id': range(1, NUM_SENSORS + 1),
    'sensor_type': np.random.choice(sensor_types, size=NUM_SENSORS, p=type_weights),
    'zone_id': np.random.randint(1, NUM_ZONES + 1, size=NUM_SENSORS),
    'pipe_segment_id': np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_SENSORS),
    'installation_date': install_dates.strftime('%Y-%m-%d'),
    'manufacturer': np.random.choice(manufacturers, size=NUM_SENSORS),
    'calibration_status': np.random.choice(
        calibration_statuses, size=NUM_SENSORS, p=[0.70, 0.20, 0.10]
    )
})

print(f"dim_sensors: {len(dim_sensors)} rows")
print(f"Sensor types:\n{dim_sensors['sensor_type'].value_counts()}")
print(f"\nCalibration status:\n{dim_sensors['calibration_status'].value_counts()}")
dim_sensors.head()

# Generate fact_sensor_readings: ~2M rows of 15-min interval data over 2 years
print("Generating ~2M sensor readings. This may take 1-2 minutes...")

# Create the full 2-year timestamp range at 15-minute intervals
start_date = pd.Timestamp('2022-01-01')
end_date = pd.Timestamp('2023-12-31 23:45:00')
timestamps_full = pd.date_range(start_date, end_date, freq='15min')
print(f"Total available timestamps: {len(timestamps_full):,}")

# Each sensor gets ~10,000 readings (contiguous time blocks)
readings_per_sensor = 10000

all_readings = []
for sensor_id in range(1, NUM_SENSORS + 1):
    # Pick a random starting point within the 2-year window
    max_start = len(timestamps_full) - readings_per_sensor
    start_idx = np.random.randint(0, max_start)
    sensor_timestamps = timestamps_full[start_idx:start_idx + readings_per_sensor]

    # Look up this sensor's type to generate appropriate readings
    sensor_type = dim_sensors.loc[
        dim_sensors['sensor_id'] == sensor_id, 'sensor_type'
    ].values[0]

    # Generate realistic readings based on sensor type
    reading_data = {
        'reading_id': range(
            (sensor_id - 1) * readings_per_sensor + 1,
            sensor_id * readings_per_sensor + 1
        ),
        'sensor_id': sensor_id,
        'timestamp': sensor_timestamps,
        'pressure_bar': np.nan,
        'flow_lps': np.nan,
        'temperature_c': np.nan
    }

    if sensor_type == 'pressure':
        # Normal pressure: 3.5 - 5.5 bar with daily variation
        reading_data['pressure_bar'] = np.random.normal(4.5, 0.5, readings_per_sensor)
    elif sensor_type == 'flow':
        # Normal flow: 10-25 liters per second
        reading_data['flow_lps'] = np.random.normal(15, 3, readings_per_sensor)
    else:
        # Temperature: 8-18 degrees C with seasonal variation
        reading_data['temperature_c'] = np.random.normal(12, 3, readings_per_sensor)

    all_readings.append(pd.DataFrame(reading_data))

# Combine all sensor readings into one DataFrame
fact_sensor_readings = pd.concat(all_readings, ignore_index=True)
print(f"\nBase readings generated: {len(fact_sensor_readings):,} rows")
print(f"Date range: {fact_sensor_readings['timestamp'].min()} to {fact_sensor_readings['timestamp'].max()}")

# Inject data quality issues to simulate real-world IoT messiness
print("Injecting data quality issues...")

# Issue 1: 8% random NULLs across measurement columns
null_mask = np.random.random(len(fact_sensor_readings)) < NULL_RATE
for col in ['pressure_bar', 'flow_lps', 'temperature_c']:
    col_mask = null_mask & fact_sensor_readings[col].notna()
    fact_sensor_readings.loc[col_mask, col] = np.nan
nulls_injected = null_mask.sum()
print(f"  - Random NULLs injected: {nulls_injected:,} ({NULL_RATE*100:.0f}% of rows)")

# Issue 2: 3 sensors with week-long gaps (simulating hardware failure)
gap_sensors = np.random.choice(range(1, NUM_SENSORS + 1), size=3, replace=False)
for sensor_id in gap_sensors:
    sensor_mask = fact_sensor_readings['sensor_id'] == sensor_id
    sensor_indices = fact_sensor_readings[sensor_mask].index
    # Remove a week of readings (7 days * 96 readings/day = 672 readings)
    gap_start = np.random.randint(1000, 8000)
    gap_indices = sensor_indices[gap_start:gap_start + 672]
    fact_sensor_readings = fact_sensor_readings.drop(gap_indices)
print(f"  - Week-long gaps created for sensors: {gap_sensors.tolist()}")

# Issue 3: Pressure drops of 15-40% on 45 specific segments (simulating leaks)
leak_segment_ids = np.random.choice(
    range(1, NUM_PIPE_SEGMENTS + 1), size=LEAK_SEGMENTS, replace=False
)
# Find sensors on these leak segments
leak_sensor_ids = dim_sensors[
    dim_sensors['pipe_segment_id'].isin(leak_segment_ids) &
    (dim_sensors['sensor_type'] == 'pressure')
]['sensor_id'].values

for sensor_id in leak_sensor_ids:
    sensor_mask = fact_sensor_readings['sensor_id'] == sensor_id
    drop_factor = np.random.uniform(0.60, 0.85)  # 15-40% pressure reduction
    fact_sensor_readings.loc[sensor_mask, 'pressure_bar'] *= drop_factor
print(f"  - Pressure drop anomalies injected on {LEAK_SEGMENTS} segments ({len(leak_sensor_ids)} sensors affected)")

# Issue 4: Timezone inconsistencies (30% of sensors report in UTC+5 instead of UTC)
tz_offset_sensors = np.random.choice(
    range(1, NUM_SENSORS + 1), size=int(NUM_SENSORS * 0.30), replace=False
)
for sensor_id in tz_offset_sensors:
    sensor_mask = fact_sensor_readings['sensor_id'] == sensor_id
    fact_sensor_readings.loc[sensor_mask, 'timestamp'] += pd.Timedelta(hours=5)
print(f"  - Timezone offset applied to {len(tz_offset_sensors)} sensors (UTC+5 instead of UTC)")

# Issue 5: Duplicate timestamps (~0.5% of rows duplicated)
num_duplicates = int(len(fact_sensor_readings) * 0.005)
duplicate_indices = np.random.choice(
    fact_sensor_readings.index, size=num_duplicates, replace=False
)
duplicates = fact_sensor_readings.loc[duplicate_indices].copy()
fact_sensor_readings = pd.concat([fact_sensor_readings, duplicates], ignore_index=True)
print(f"  - Duplicate timestamps added: {num_duplicates:,} rows")

# Issue 6: Negative flow values (sensor malfunction on ~1% of flow readings)
flow_mask = fact_sensor_readings['flow_lps'].notna()
flow_indices = fact_sensor_readings[flow_mask].index
neg_count = int(len(flow_indices) * 0.01)
neg_indices = np.random.choice(flow_indices, size=neg_count, replace=False)
fact_sensor_readings.loc[neg_indices, 'flow_lps'] *= -1
print(f"  - Negative flow values injected: {neg_count:,} readings")

# Reset the index after all modifications
fact_sensor_readings = fact_sensor_readings.reset_index(drop=True)
print(f"\nFinal fact_sensor_readings: {len(fact_sensor_readings):,} rows")

# Generate fact_maintenance_logs: 800 maintenance events over 2 years
event_types = ['repair', 'inspection', 'replacement']
event_weights = [0.45, 0.40, 0.15]  # repairs most common, replacements rare

maint_dates = pd.date_range('2022-01-01', '2023-12-31', periods=NUM_MAINTENANCE_LOGS)

fact_maintenance_logs = pd.DataFrame({
    'log_id': range(1, NUM_MAINTENANCE_LOGS + 1),
    'segment_id': np.random.randint(1, NUM_PIPE_SEGMENTS + 1, size=NUM_MAINTENANCE_LOGS),
    'event_type': np.random.choice(event_types, size=NUM_MAINTENANCE_LOGS, p=event_weights),
    'event_date': maint_dates.strftime('%Y-%m-%d'),
    'cost': np.round(np.random.uniform(500, 25000, size=NUM_MAINTENANCE_LOGS), 2),
    'crew_id': [f'CREW-{np.random.randint(1, 16):02d}' for _ in range(NUM_MAINTENANCE_LOGS)],
    'notes': np.random.choice([
        'Routine inspection completed',
        'Emergency repair - pipe burst',
        'Scheduled replacement - end of life',
        'Leak detected during patrol',
        'Valve replacement',
        'Pressure regulator adjustment',
        'Corrosion treatment applied',
        'Joint seal replacement'
    ], size=NUM_MAINTENANCE_LOGS)
})

print(f"fact_maintenance_logs: {len(fact_maintenance_logs)} rows")
print(f"Event types:\n{fact_maintenance_logs['event_type'].value_counts()}")

# Generate fact_customer_consumption: monthly consumption per zone with seasonality
months = pd.date_range('2022-01', '2023-12', freq='MS')  # month starts
consumption_rows = []

for zone_id in range(1, NUM_ZONES + 1):
    base_consumption = np.random.randint(50000, 150000)  # base m3 per month
    customer_count = dim_zones.loc[
        dim_zones['zone_id'] == zone_id, 'population_served'
    ].values[0] // 3  # approximate households

    for month in months:
        # Add seasonal variation (higher in summer months)
        seasonal_factor = 1.0 + 0.3 * np.sin(2 * np.pi * (month.month - 1) / 12)
        noise = np.random.normal(1.0, 0.05)
        monthly_consumption = int(base_consumption * seasonal_factor * noise)

        consumption_rows.append({
            'zone_id': zone_id,
            'year_month': month.strftime('%Y-%m'),
            'consumption_m3': monthly_consumption,
            'customer_count': customer_count
        })

fact_customer_consumption = pd.DataFrame(consumption_rows)
print(f"\nfact_customer_consumption: {len(fact_customer_consumption)} rows")
print(f"Months covered: {fact_customer_consumption['year_month'].nunique()}")
fact_customer_consumption.head()

# Export all datasets to the /data directory
import os

data_dir = 'data'
os.makedirs(data_dir, exist_ok=True)

# Write dimension tables
dim_zones.to_csv(f'{data_dir}/dim_zones.csv', index=False)
dim_pipe_segments.to_csv(f'{data_dir}/dim_pipe_segments.csv', index=False)
dim_sensors.to_csv(f'{data_dir}/dim_sensors.csv', index=False)

# Write fact tables
fact_sensor_readings.to_csv(f'{data_dir}/fact_sensor_readings.csv', index=False)
fact_maintenance_logs.to_csv(f'{data_dir}/fact_maintenance_logs.csv', index=False)
fact_customer_consumption.to_csv(f'{data_dir}/fact_customer_consumption.csv', index=False)

# Also save the leak segment IDs for validation in Step 4
pd.DataFrame({'segment_id': leak_segment_ids}).to_csv(
    f'{data_dir}/ground_truth_leaks.csv', index=False
)

print("All 7 datasets exported to /data directory.")
print(f"\n{'='*60}")
print("DATA QUALITY SUMMARY - Injected Issues")
print(f"{'='*60}")
print(f"Total sensor readings:        {len(fact_sensor_readings):>12,}")
print(f"Random NULLs (8% rate):       {nulls_injected:>12,}")
print(f"Sensors with week-long gaps:  {len(gap_sensors):>12}")
print(f"Segments with leak anomalies: {LEAK_SEGMENTS:>12}")
print(f"Sensors with TZ offset:       {len(tz_offset_sensors):>12}")
print(f"Duplicate rows added:         {num_duplicates:>12,}")
print(f"Negative flow values:         {neg_count:>12,}")
print(f"{'='*60}")
print(f"\nDimension tables:")
print(f"  dim_zones:              {len(dim_zones):>6} rows")
print(f"  dim_pipe_segments:      {len(dim_pipe_segments):>6} rows")
print(f"  dim_sensors:            {len(dim_sensors):>6} rows")
print(f"\nFact tables:")
print(f"  fact_sensor_readings:   {len(fact_sensor_readings):>12,} rows")
print(f"  fact_maintenance_logs:  {len(fact_maintenance_logs):>6} rows")
print(f"  fact_customer_consumption: {len(fact_customer_consumption):>4} rows")

# Verify all files were created successfully
for filename in os.listdir(data_dir):
    filepath = os.path.join(data_dir, filename)
    size_mb = os.path.getsize(filepath) / (1024 * 1024)
    print(f"  {filename:40s} {size_mb:>8.2f} MB")

Your raw dataset is complete. You have 7 interconnected tables with realistic messiness baked in. Next up, you will clean this data using domain-aware strategies that handle each type of issue differently.

Clean Data & Engineer Features

In the previous step, you generated a realistic but messy IoT dataset simulating AquaFlow Utilities' sensor network. Now it is time to turn that raw chaos into analysis-ready data.

Real-world sensor data is never clean. Sensors report in different timezones, hardware failures create gaps, and malfunctioning equipment produces impossible values like negative flow rates. Before you can detect leaks or forecast demand, you need to fix these structural issues and engineer features that expose hidden patterns.

In this step, get ready to:

• Fix structural anomalies in the sensor readings (timezone inconsistencies, duplicates, impossible values).
• Apply domain-aware imputation strategies based on gap length.
• Engineer analytical features like rolling statistics, pressure differentials, and night flow ratios.

Handle structural anomalies

The raw sensor readings have three structural problems that must be fixed before any analysis. Some sensors report timestamps in UTC while others use local time. Hardware glitches produce duplicate readings, and broken flow meters report negative values.

• In your notebook aquaflow_analysis.ipynb, create a new cell and load the raw sensor readings by running this code:

import pandas as pd
import numpy as np

# Load the raw sensor readings
df = pd.read_csv('data/fact_sensor_readings.csv')

print(f"Raw dataset shape: {df.shape}")
print(f"\nColumn dtypes:\n{df.dtypes}")
print(f"\nSample timestamps (first 5):")
print(df['timestamp'].head())

What does this code do?

This loads the ~2 million row sensor readings file and prints basic diagnostics. You will see that the timestamp column is stored as a string, not a datetime object. Some entries include timezone offsets while others do not.

• Run the cell. You should see the DataFrame shape and a mix of timestamp formats in the output.

• Standardize all timestamps to UTC and remove duplicates by adding this code in a new cell:

# Standardize all timestamps to UTC
# utc=True handles mixed naive/aware timestamps by converting everything to UTC
df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)

# Sort by sensor and time for consistent deduplication
df = df.sort_values(['sensor_id', 'timestamp']).reset_index(drop=True)

# Remove duplicate readings (keep first occurrence per sensor + timestamp)
before_dedup = len(df)
df = df.drop_duplicates(subset=['sensor_id', 'timestamp'], keep='first')
after_dedup = len(df)

print(f"Removed {before_dedup - after_dedup} duplicate readings")
print(f"Dataset shape after dedup: {df.shape}")

What does this code do?

The pd.to_datetime() function with utc=True converts all timestamps to UTC. Timezone-naive inputs are localized as UTC, and timezone-aware inputs with offsets are converted to equivalent UTC times.

The deduplication keeps only the first reading when the same sensor reports at the same timestamp. Sorting first ensures deterministic results regardless of original file order.

• Run the cell. You should see a message showing how many duplicates were removed.

• Handle negative flow values by adding this code in a new cell:

# Handle negative flow values (physically impossible - sensor malfunction)
# Replace with NaN and create a flag column for investigation
negative_flow_mask = df['flow_lps'] < 0
print(f"Negative flow readings found: {negative_flow_mask.sum()}")

# Flag the rows before replacing
df['negative_flow_flag'] = negative_flow_mask.astype(int)

# Replace negative flows with NaN
df.loc[negative_flow_mask, 'flow_lps'] = np.nan

print(f"Negative flows replaced with NaN and flagged")
print(f"\nStructural cleaning summary:")
print(f"  - All timestamps converted to UTC")
print(f"  - {before_dedup - after_dedup} duplicates removed")
print(f"  - {negative_flow_mask.sum()} negative flow values flagged and set to NaN")

Why replace with NaN instead of dropping?

A negative flow reading means the sensor malfunctioned at that moment, but the pressure and temperature values from the same timestamp are still valid. Replacing just the flow value with NaN preserves the other columns. The flag column lets you investigate patterns in sensor failures later.

• Run the cell. You should see a count of negative flow readings that were flagged and replaced.

Seeing a timezone parsing error?

If you get a ValueError about mixed timezones, make sure you included utc=True in the pd.to_datetime() call. Without this parameter, pandas cannot handle a column that mixes timezone-aware and timezone-naive strings.

Help me debug my timezone conversion error.

Handle missing values with domain-aware imputation

Not all missing data is the same. A 30-minute gap likely means a brief network hiccup, while a week-long gap means the sensor was offline for maintenance. Using the same fill strategy for both would introduce false data.

• Implement the gap-aware imputation logic by adding this code in a new cell:

def classify_and_fill_gaps(group):
    """
    Apply different imputation strategies based on gap length:
    - Short gaps (< 4 hours): forward-fill (sensor was briefly interrupted)
    - Medium gaps (4-24 hours): linear interpolation (gradual change assumed)
    - Long gaps (> 24 hours): leave as NaN and flag as sensor_offline
    """
    # Ensure sorted by timestamp
    group = group.sort_values('timestamp')

    # Calculate time differences between consecutive readings
    time_diff = group['timestamp'].diff()

    # Identify gap boundaries
    short_gap_limit = pd.Timedelta(hours=4)
    medium_gap_limit = pd.Timedelta(hours=24)

    # Create sensor_offline flag for long gaps
    group['sensor_offline_flag'] = 0
    long_gaps = time_diff > medium_gap_limit
    group.loc[long_gaps, 'sensor_offline_flag'] = 1

    # Columns to impute (numeric sensor values only)
    value_cols = ['pressure_bar', 'flow_lps', 'temperature_c']

    for col in value_cols:
        # Step 1: Forward-fill short gaps (limit to 16 readings = 4 hours at 15-min intervals)
        group[col] = group[col].ffill(limit=16)

        # Step 2: Linear interpolation for medium gaps (limit to 96 readings = 24 hours)
        group[col] = group[col].interpolate(
            method='linear',
            limit=96,
            limit_direction='forward'
        )
        # Long gaps (> 24h) remain as NaN naturally since both fill methods have limits

    return group

# Apply gap-aware imputation per sensor
print("Applying gap-aware imputation (this may take a minute)...")
df = df.groupby('sensor_id', group_keys=False).apply(classify_and_fill_gaps)
print("Imputation complete!")

What does this imputation strategy do?

The function processes each sensor independently using groupby. Short gaps get forward-filled because a brief interruption likely means the value did not change much. Medium gaps get linearly interpolated because the value probably transitioned smoothly.

Long gaps (over 24 hours) stay as NaN because you cannot reliably guess what happened during a week-long outage. The sensor_offline_flag column marks these periods for exclusion from downstream analysis.

• Run the cell. This processes ~2 million rows grouped by 200 sensors, so it takes about 30-60 seconds.

• Create a data quality scorecard to verify your imputation worked by adding this code in a new cell:

# Build a data quality scorecard showing completeness per sensor
value_cols = ['pressure_bar', 'flow_lps', 'temperature_c']

scorecard = df.groupby('sensor_id')[value_cols].apply(
    lambda x: x.notna().mean()
).reset_index()

# Rename columns for clarity
scorecard.columns = ['sensor_id', 'pressure_completeness', 'flow_completeness', 'temp_completeness']

# Add overall completeness score
scorecard['overall_completeness'] = scorecard[['pressure_completeness', 'flow_completeness', 'temp_completeness']].mean(axis=1)

# Sort to see worst sensors first
scorecard = scorecard.sort_values('overall_completeness', ascending=True)

print("Data Quality Scorecard (worst 10 sensors):")
print(scorecard.head(10).to_string(index=False))
print(f"\nAverage completeness across all sensors: {scorecard['overall_completeness'].mean():.1%}")

What is this scorecard showing?

The scorecard calculates what percentage of readings are non-null for each sensor after imputation. The 3 sensors with week-long gaps will show the lowest completeness scores. Most sensors should be above 95% complete after your imputation filled the short and medium gaps.

• Run the cell. You should see a table showing the 10 worst sensors by completeness, with most sensors above 90%.

Imputation taking too long?

If the groupby().apply() call takes more than 5 minutes, check that you sorted and deduplicated the DataFrame in the previous substep. Unsorted data with duplicates dramatically increases processing time.

Help me speed up my pandas groupby apply operation on 2 million rows.

Engineer analytical features

Raw pressure and flow values on their own do not reveal leaks. You need derived features that capture patterns over time. Rolling statistics show trends, pressure differentials expose drops between connected pipes, and night flow ratios flag suspicious water movement during low-usage hours.

• Calculate rolling statistics per sensor by adding this code in a new cell:

# Calculate rolling statistics per sensor
# Window sizes: 1h=4, 6h=24, 24h=96 readings (at 15-min intervals)
print("Calculating rolling statistics...")

df = df.sort_values(['sensor_id', 'timestamp']).reset_index(drop=True)

for window_name, window_size in [('1h', 4), ('6h', 24), ('24h', 96)]:
    # Rolling mean and std for pressure
    df[f'rolling_mean_{window_name}'] = df.groupby('sensor_id')['pressure_bar'].transform(
        lambda x: x.rolling(window=window_size, min_periods=1).mean()
    )
    df[f'rolling_std_{window_name}'] = df.groupby('sensor_id')['pressure_bar'].transform(
        lambda x: x.rolling(window=window_size, min_periods=1).std()
    )

print("Rolling statistics complete!")
print(f"New columns: {[c for c in df.columns if 'rolling' in c]}")

Why three different window sizes?

A 1-hour window captures sudden pressure drops that indicate an active pipe burst. A 6-hour window smooths out normal daily fluctuation and highlights sustained anomalies. A 24-hour window reveals slow degradation trends that develop over days.

The rolling_std columns are especially useful. A pipe segment with consistently high pressure variance is likely experiencing intermittent leaks.

• Run the cell. You should see a list of 6 new rolling columns in the output.

• Calculate pressure differentials and night flow ratio by adding this code in a new cell:

# Load pipe segment data for upstream/downstream relationships
pipes = pd.read_csv('data/dim_pipe_segments.csv')
sensors = pd.read_csv('data/dim_sensors.csv')

# Map each sensor to its pipe segment
df = df.merge(
    sensors[['sensor_id', 'pipe_segment_id', 'sensor_type']],
    on='sensor_id',
    how='left'
)

# Calculate pressure differential between upstream and downstream sensors
# on the same pipe segment (requires daily aggregation)
daily_pressure = df[df['sensor_type'] == 'pressure'].groupby(
    ['pipe_segment_id', df['timestamp'].dt.date]
)['pressure_bar'].mean().reset_index()
daily_pressure.columns = ['pipe_segment_id', 'date', 'avg_pressure']

# Merge with pipe topology to get upstream/downstream relationships
daily_pressure = daily_pressure.merge(
    pipes[['segment_id', 'upstream_segment_id']],
    left_on='pipe_segment_id',
    right_on='segment_id',
    how='left'
)

# Calculate differential: upstream pressure - current segment pressure
upstream_pressure = daily_pressure[['pipe_segment_id', 'date', 'avg_pressure']].rename(
    columns={'pipe_segment_id': 'upstream_segment_id', 'avg_pressure': 'upstream_pressure'}
)
daily_pressure = daily_pressure.merge(upstream_pressure, on=['upstream_segment_id', 'date'], how='left')
daily_pressure['pressure_differential'] = daily_pressure['upstream_pressure'] - daily_pressure['avg_pressure']

# Map pressure differential back to main DataFrame
df['date'] = df['timestamp'].dt.date
df = df.merge(
    daily_pressure[['pipe_segment_id', 'date', 'pressure_differential']].drop_duplicates(),
    on=['pipe_segment_id', 'date'],
    how='left'
)

print(f"Pressure differential calculated for {daily_pressure['pressure_differential'].notna().sum()} segment-days")
print(f"Mean pressure differential: {daily_pressure['pressure_differential'].mean():.4f} bar")

What does pressure differential reveal?

A healthy pipe network shows small, consistent pressure drops as water flows downstream (due to friction losses). A sudden increase in pressure differential between an upstream and downstream sensor means water is escaping between them. This is one of the strongest indicators of a leak location.

• Run the cell. You should see a count of segment-days with valid pressure differentials and the mean value.

• Calculate the night flow ratio and temporal features, then export the cleaned dataset by adding this code in a new cell:

# Calculate night flow ratio per pipe segment per day
# Night flow ratio = avg flow during 2AM-4AM / daily avg flow
# High ratio = suspected leak (water flowing at night when demand is low)
df['hour_of_day'] = df['timestamp'].dt.hour
df['day_of_week'] = df['timestamp'].dt.dayofweek
df['month'] = df['timestamp'].dt.month
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
df['is_holiday'] = 0  # Placeholder - can be enriched with holiday calendar

# Calculate night flow ratio
night_mask = df['hour_of_day'].between(2, 3)  # 2AM-4AM
night_flow = df[night_mask].groupby(
    ['pipe_segment_id', 'date']
)['flow_lps'].mean().reset_index().rename(columns={'flow_lps': 'night_flow'})

daily_flow = df.groupby(
    ['pipe_segment_id', 'date']
)['flow_lps'].mean().reset_index().rename(columns={'flow_lps': 'daily_avg_flow'})

flow_ratios = night_flow.merge(daily_flow, on=['pipe_segment_id', 'date'], how='left')
flow_ratios['night_flow_ratio'] = flow_ratios['night_flow'] / flow_ratios['daily_avg_flow']

# Merge night flow ratio back to main DataFrame
df = df.merge(
    flow_ratios[['pipe_segment_id', 'date', 'night_flow_ratio']].drop_duplicates(),
    on=['pipe_segment_id', 'date'],
    how='left'
)

# Drop intermediate columns and save
df = df.drop(columns=['date', 'sensor_type'])

# Export cleaned dataset
df.to_csv('data/fact_sensor_readings_clean.csv', index=False)

print(f"\nFinal cleaned dataset shape: {df.shape}")
print(f"\nNew feature columns:")
feature_cols = ['rolling_mean_1h', 'rolling_std_1h', 'rolling_mean_6h', 'rolling_std_6h',
                'rolling_mean_24h', 'rolling_std_24h', 'pressure_differential',
                'night_flow_ratio', 'hour_of_day', 'day_of_week', 'month',
                'is_weekend', 'is_holiday']
for col in feature_cols:
    print(f"  - {col}")
print(f"\nSaved to: data/fact_sensor_readings_clean.csv")

What is night flow ratio?

In a healthy water network, demand drops dramatically between 2AM and 4AM because most customers are asleep. If a pipe segment still shows high flow during these hours, water is going somewhere it should not.

A night flow ratio close to 1.0 means flow is constant day and night. That strongly suggests a leak. A healthy segment typically shows a ratio between 0.2 and 0.4.

• Run the cell. You should see the final dataset shape, a list of all 13 new feature columns, and a confirmation that the file was saved to data/fact_sensor_readings_clean.csv.

Getting a merge error on pipe_segment_id?

If you see a KeyError for pipe_segment_id, check that the merge with the sensors DataFrame completed successfully in the pressure differential cell. The column comes from joining dim_sensors.csv to your main DataFrame.

Help me debug my merge error when calculating night flow ratio.

✔️ Awesome, I've got everything!

Great. Double check you have saved all cells in aquaflow_analysis.ipynb and that data/fact_sensor_readings_clean.csv exists in your project folder.

ⓧ I'd like to double check the full code

Here is the complete cleaning and feature engineering code for this step, all in one view:

import pandas as pd
import numpy as np

# Load the raw sensor readings
df = pd.read_csv('data/fact_sensor_readings.csv')

print(f"Raw dataset shape: {df.shape}")
print(f"\nColumn dtypes:\n{df.dtypes}")
print(f"\nSample timestamps (first 5):")
print(df['timestamp'].head())

# --- Structural Cleaning ---

# Standardize all timestamps to UTC
df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)

# Sort by sensor and time for consistent deduplication
df = df.sort_values(['sensor_id', 'timestamp']).reset_index(drop=True)

# Remove duplicate readings (keep first occurrence per sensor + timestamp)
before_dedup = len(df)
df = df.drop_duplicates(subset=['sensor_id', 'timestamp'], keep='first')
after_dedup = len(df)

print(f"Removed {before_dedup - after_dedup} duplicate readings")
print(f"Dataset shape after dedup: {df.shape}")

# Handle negative flow values
negative_flow_mask = df['flow_lps'] < 0
print(f"Negative flow readings found: {negative_flow_mask.sum()}")

df['negative_flow_flag'] = negative_flow_mask.astype(int)
df.loc[negative_flow_mask, 'flow_lps'] = np.nan

print(f"Negative flows replaced with NaN and flagged")

# --- Gap-Aware Imputation ---

def classify_and_fill_gaps(group):
    group = group.sort_values('timestamp')
    time_diff = group['timestamp'].diff()

    short_gap_limit = pd.Timedelta(hours=4)
    medium_gap_limit = pd.Timedelta(hours=24)

    group['sensor_offline_flag'] = 0
    long_gaps = time_diff > medium_gap_limit
    group.loc[long_gaps, 'sensor_offline_flag'] = 1

    value_cols = ['pressure_bar', 'flow_lps', 'temperature_c']

    for col in value_cols:
        group[col] = group[col].ffill(limit=16)
        group[col] = group[col].interpolate(
            method='linear',
            limit=96,
            limit_direction='forward'
        )

    return group

print("Applying gap-aware imputation (this may take a minute)...")
df = df.groupby('sensor_id', group_keys=False).apply(classify_and_fill_gaps)
print("Imputation complete!")

# --- Data Quality Scorecard ---

value_cols = ['pressure_bar', 'flow_lps', 'temperature_c']

scorecard = df.groupby('sensor_id')[value_cols].apply(
    lambda x: x.notna().mean()
).reset_index()

scorecard.columns = ['sensor_id', 'pressure_completeness', 'flow_completeness', 'temp_completeness']
scorecard['overall_completeness'] = scorecard[['pressure_completeness', 'flow_completeness', 'temp_completeness']].mean(axis=1)
scorecard = scorecard.sort_values('overall_completeness', ascending=True)

print("Data Quality Scorecard (worst 10 sensors):")
print(scorecard.head(10).to_string(index=False))
print(f"\nAverage completeness across all sensors: {scorecard['overall_completeness'].mean():.1%}")

# --- Feature Engineering ---

# Rolling statistics
df = df.sort_values(['sensor_id', 'timestamp']).reset_index(drop=True)

for window_name, window_size in [('1h', 4), ('6h', 24), ('24h', 96)]:
    df[f'rolling_mean_{window_name}'] = df.groupby('sensor_id')['pressure_bar'].transform(
        lambda x: x.rolling(window=window_size, min_periods=1).mean()
    )
    df[f'rolling_std_{window_name}'] = df.groupby('sensor_id')['pressure_bar'].transform(
        lambda x: x.rolling(window=window_size, min_periods=1).std()
    )

print("Rolling statistics complete!")

# Pressure differential
pipes = pd.read_csv('data/dim_pipe_segments.csv')
sensors = pd.read_csv('data/dim_sensors.csv')

df = df.merge(
    sensors[['sensor_id', 'pipe_segment_id', 'sensor_type']],
    on='sensor_id',
    how='left'
)

daily_pressure = df[df['sensor_type'] == 'pressure'].groupby(
    ['pipe_segment_id', df['timestamp'].dt.date]
)['pressure_bar'].mean().reset_index()
daily_pressure.columns = ['pipe_segment_id', 'date', 'avg_pressure']

daily_pressure = daily_pressure.merge(
    pipes[['segment_id', 'upstream_segment_id']],
    left_on='pipe_segment_id',
    right_on='segment_id',
    how='left'
)

upstream_pressure = daily_pressure[['pipe_segment_id', 'date', 'avg_pressure']].rename(
    columns={'pipe_segment_id': 'upstream_segment_id', 'avg_pressure': 'upstream_pressure'}
)
daily_pressure = daily_pressure.merge(upstream_pressure, on=['upstream_segment_id', 'date'], how='left')
daily_pressure['pressure_differential'] = daily_pressure['upstream_pressure'] - daily_pressure['avg_pressure']

df['date'] = df['timestamp'].dt.date
df = df.merge(
    daily_pressure[['pipe_segment_id', 'date', 'pressure_differential']].drop_duplicates(),
    on=['pipe_segment_id', 'date'],
    how='left'
)

print(f"Pressure differential calculated for {daily_pressure['pressure_differential'].notna().sum()} segment-days")

# Night flow ratio and temporal features
df['hour_of_day'] = df['timestamp'].dt.hour
df['day_of_week'] = df['timestamp'].dt.dayofweek
df['month'] = df['timestamp'].dt.month
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
df['is_holiday'] = 0

night_mask = df['hour_of_day'].between(2, 3)
night_flow = df[night_mask].groupby(
    ['pipe_segment_id', 'date']
)['flow_lps'].mean().reset_index().rename(columns={'flow_lps': 'night_flow'})

daily_flow = df.groupby(
    ['pipe_segment_id', 'date']
)['flow_lps'].mean().reset_index().rename(columns={'flow_lps': 'daily_avg_flow'})

flow_ratios = night_flow.merge(daily_flow, on=['pipe_segment_id', 'date'], how='left')
flow_ratios['night_flow_ratio'] = flow_ratios['night_flow'] / flow_ratios['daily_avg_flow']

df = df.merge(
    flow_ratios[['pipe_segment_id', 'date', 'night_flow_ratio']].drop_duplicates(),
    on=['pipe_segment_id', 'date'],
    how='left'
)

# Drop intermediate columns and save
df = df.drop(columns=['date', 'sensor_type'])
df.to_csv('data/fact_sensor_readings_clean.csv', index=False)

print(f"\nFinal cleaned dataset shape: {df.shape}")
print(f"Saved to: data/fact_sensor_readings_clean.csv")

Your sensor data is now clean, imputed, and enriched with 13 analytical features. Next up, you will use K-Means clustering on these features to detect which pipe segments are actively leaking.

Detect Leaks with K-Means Clustering

Your cleaned sensor data is ready, with all timestamps standardized, missing values handled intelligently, and analytical features engineered.

Now comes the detective work. AquaFlow Utilities knows it's losing 28% of its water somewhere in the network, but 500 pipe segments is too many to inspect manually. You need an algorithm that can identify suspicious patterns across all segments at once. K-Means clustering is an unsupervised learning technique that groups data points by similarity. In this step, you'll use it to separate pipe segments into risk categories based on their pressure and flow behavior.

In this step, get ready to:

• Prepare a feature matrix by aggregating sensor data to daily level per pipe segment.
• Run K-Means clustering to classify segments into Normal, Degrading, Suspected Leak, and Critical Leak categories.
• Validate your clustering results against the injected ground-truth leaks and export risk scores.

Prepare the feature matrix for clustering

K-Means needs a table where each row is one pipe segment and each column is a numeric feature describing that segment's behavior. Right now, your data has millions of rows at the sensor-reading level. You need to aggregate it down to one summary row per segment per day, then average across all days.

• In your aquaflow_analysis.ipynb notebook, create a new cell and aggregate the clean sensor data to daily level per pipe segment by pasting this code:

# --- Step 4: K-Means Clustering for Leak Detection ---
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import confusion_matrix

# Load the cleaned sensor data
df_clean = pd.read_csv('data/fact_sensor_readings_clean.csv', parse_dates=['timestamp'])

# Merge with sensor metadata to get pipe_segment_id
df_sensors = pd.read_csv('data/dim_sensors.csv')
df_merged = df_clean.merge(df_sensors[['sensor_id', 'pipe_segment_id']], on='sensor_id', how='left')

# Aggregate to daily level per pipe segment
df_merged['date'] = df_merged['timestamp'].dt.date

daily_segment = df_merged.groupby(['pipe_segment_id', 'date']).agg(
    avg_pressure_drop=('pressure_differential', 'mean'),
    pressure_std=('pressure_differential', 'std'),
    night_flow_ratio=('night_flow_ratio', 'mean'),
    min_pressure=('reading_value', 'min'),
    flow_variance=('reading_value', 'var')
).reset_index()

print(f"Daily segment data: {daily_segment.shape[0]} rows, {daily_segment['pipe_segment_id'].nunique()} unique segments")

What does this code do?

This cell performs the critical transformation from sensor-level readings to segment-level features. Each pipe segment gets one row per day with five aggregated metrics.

• avg_pressure_drop captures the mean pressure differential across the segment. A consistently negative differential suggests a leak.
• night_flow_ratio is high when water flows during 2AM-4AM (when demand should be near zero). High night flow strongly indicates a leak.
• flow_variance captures unstable flow patterns that accompany structural pipe failures.

• Run the cell. You should see a print statement confirming the shape of your daily data and the number of unique segments (should be close to 500).

• In the next cell, collapse the daily data into a single feature vector per segment and standardize it by pasting this code:

# Collapse daily data into one feature vector per segment
segment_features = daily_segment.groupby('pipe_segment_id').agg(
    avg_pressure_drop=('avg_pressure_drop', 'mean'),
    pressure_std=('pressure_std', 'mean'),
    night_flow_ratio=('night_flow_ratio', 'mean'),
    min_pressure=('min_pressure', 'mean'),
    flow_variance=('flow_variance', 'mean')
).reset_index()

# Drop rows with NaN values (segments with insufficient data)
segment_features = segment_features.dropna()

# Select the 5 features for clustering
feature_cols = ['avg_pressure_drop', 'pressure_std', 'night_flow_ratio', 'min_pressure', 'flow_variance']
X = segment_features[feature_cols]

# Standardize features so each contributes equally to distance calculations
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

print(f"Feature matrix shape: {X_scaled.shape}")
print(f"Features: {feature_cols}")

Why standardize features?

K-Means uses Euclidean distance to assign points to clusters. Without standardization, features with larger numeric ranges (like flow_variance) would dominate the distance calculation, drowning out features with smaller ranges (like night_flow_ratio).

StandardScaler transforms each feature to have mean=0 and standard deviation=1. This ensures every feature contributes proportionally to the clustering.

• Run the cell. You should see a feature matrix shape of approximately (500, 5), meaning 500 segments with 5 features each.

Seeing far fewer than 500 segments?

If your segment count is significantly lower than 500, check that the merge on sensor_id succeeded. Run df_merged['pipe_segment_id'].isna().sum() to see how many readings couldn't be matched to a segment.

Also confirm your dim_sensors.csv file has the pipe_segment_id column populated for all 200 sensors.

Help me debug my feature matrix

Run K-Means and identify anomaly clusters

With your feature matrix ready, you need to determine the right number of clusters. The Elbow Method plots the within-cluster sum of squares (inertia) for different values of k. The "elbow" in the curve is where adding more clusters stops providing meaningful improvement.

• Create a new cell and generate the elbow plot by pasting this code:

# Elbow Method: test k=2 through k=10
inertias = []
k_range = range(2, 11)

for k in k_range:
    kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
    kmeans.fit(X_scaled)
    inertias.append(kmeans.inertia_)

# Plot the elbow curve
plt.figure(figsize=(10, 6))
plt.plot(k_range, inertias, 'bo-', linewidth=2, markersize=8)
plt.xlabel('Number of Clusters (k)')
plt.ylabel('Inertia (Within-Cluster Sum of Squares)')
plt.title('Elbow Method for Optimal k')
plt.xticks(k_range)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('data/elbow_plot.png', dpi=150)
plt.show()

print("Look for the 'elbow' where the curve bends sharply.")

What does this code do?

The loop fits a K-Means model for each value of k from 2 to 10. The inertia_ attribute on each fitted model stores the sum of squared distances from each point to its assigned cluster center. Lower inertia means tighter clusters.

The elbow appears around k=4 for this dataset because there are four natural behavior patterns: normal pipes, degrading pipes, suspected leaks, and critical leaks.

• Run the cell. You should see a line chart where inertia drops steeply from k=2 to k=4, then the rate of decrease slows significantly.

Don't see a clear elbow?

If the curve looks smooth without a sharp bend, your data might have overlapping clusters. Proceed with k=4 anyway. The four categories (Normal, Degrading, Suspected Leak, Critical Leak) are domain-driven, not just statistical.

Help me interpret my elbow plot

• In the next cell, fit the final K-Means model with k=4 and assign cluster labels by pasting this code:

# Fit final KMeans model with 4 clusters
kmeans_final = KMeans(n_clusters=4, random_state=42, n_init=10)
segment_features['cluster'] = kmeans_final.fit_predict(X_scaled)

# Examine cluster centers to determine which cluster maps to which label
cluster_centers = pd.DataFrame(
    scaler.inverse_transform(kmeans_final.cluster_centers_),
    columns=feature_cols
)
print("Cluster Centers (original scale):")
print(cluster_centers.round(3))
print(f"\nCluster sizes: {segment_features['cluster'].value_counts().sort_index().to_dict()}")

What does this code do?

The fit_predict() method fits the model and returns cluster assignments (0, 1, 2, or 3) for each segment in one call.

Inverse-transforming the cluster centers shows you the average feature values for each cluster in their original units. This helps you determine which numeric cluster (0-3) corresponds to which domain label. The cluster with the highest avg_pressure_drop and night_flow_ratio is your "Critical Leak" cluster.

• Run the cell. Examine the cluster centers output to identify which numeric cluster has the most extreme pressure drops and highest night flow ratios.

• In the next cell, map numeric clusters to domain labels based on the cluster center analysis by pasting this code:

# Map clusters to domain labels based on cluster center severity
# Sort clusters by avg_pressure_drop to assign labels from least to most severe
cluster_severity = cluster_centers['avg_pressure_drop'].sort_values()
label_map = {
    cluster_severity.index[0]: 'Normal',
    cluster_severity.index[1]: 'Degrading',
    cluster_severity.index[2]: 'Suspected Leak',
    cluster_severity.index[3]: 'Critical Leak'
}

segment_features['cluster_label'] = segment_features['cluster'].map(label_map)

print("Cluster label distribution:")
print(segment_features['cluster_label'].value_counts())
print(f"\nLabel mapping: {label_map}")

Why sort by avg_pressure_drop for labeling?

K-Means assigns arbitrary numeric labels (0, 1, 2, 3) that change between runs. Sorting by severity lets you consistently map numbers to meaningful domain labels. The cluster with the lowest (most negative) average pressure drop has the most severe leaks.

• Run the cell. You should see the distribution of segments across the four categories, with the majority labeled "Normal" and a smaller number in the leak categories.

All segments assigned to one cluster?

If nearly all segments end up in a single cluster, the standardization step may have failed. Go back and confirm that X_scaled has mean close to 0 and std close to 1 for each column. Run print(X_scaled.mean(axis=0), X_scaled.std(axis=0)) to verify.

Help me fix my cluster distribution

Visualize and export cluster results

Your clusters are assigned. Now you need to visualize the separation between clusters and validate that your algorithm actually found the leaks you injected during data generation.

• Create a new cell and generate the scatter plot of clusters by pasting this code:

# Scatter plot: pressure_drop vs night_flow_ratio colored by cluster
plt.figure(figsize=(12, 8))
colors = {'Normal': '#2ecc71', 'Degrading': '#f39c12', 'Suspected Leak': '#e74c3c', 'Critical Leak': '#8e44ad'}

for label, color in colors.items():
    mask = segment_features['cluster_label'] == label
    plt.scatter(
        segment_features.loc[mask, 'avg_pressure_drop'],
        segment_features.loc[mask, 'night_flow_ratio'],
        c=color,
        label=f"{label} (n={mask.sum()})",
        alpha=0.7,
        s=60
    )

plt.xlabel('Average Pressure Drop')
plt.ylabel('Night Flow Ratio')
plt.title('Pipe Segment Clusters: Pressure Drop vs Night Flow Ratio')
plt.legend(loc='upper right', fontsize=10)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('data/cluster_scatter_plot.png', dpi=150)
plt.show()

print("Scatter plot saved to data/cluster_scatter_plot.png")

What should I see?

You should see four distinct color groups. The green "Normal" cluster sits in the lower-left (low pressure drop, low night flow). The purple "Critical Leak" cluster appears in the upper-right (high pressure drop, high night flow ratio). The orange and red clusters sit between these extremes.

• Run the cell. Confirm you see the four colored clusters with clear separation between at least the Normal and Critical Leak groups.

• In the next cell, validate your results against the ground truth and export the risk scores by pasting this code:

# Validate against injected ground-truth leaks from data generation
# injected_leak_segments was created in Step 2 (list of 45 segment IDs)
actual_leak = segment_features['pipe_segment_id'].isin(injected_leak_segments).astype(int)
detected_leak = segment_features['cluster_label'].isin(['Suspected Leak', 'Critical Leak']).astype(int)

# Print confusion matrix
cm = confusion_matrix(actual_leak, detected_leak)
print("Confusion Matrix (Actual vs Detected):")
print(f"  TN={cm[0,0]}, FP={cm[0,1]}")
print(f"  FN={cm[1,0]}, TP={cm[1,1]}")
print(f"\nDetection rate: {cm[1,1]}/{cm[1,0]+cm[1,1]} = {cm[1,1]/(cm[1,0]+cm[1,1])*100:.1f}%")
print(f"False positive rate: {cm[0,1]}/{cm[0,0]+cm[0,1]} = {cm[0,1]/(cm[0,0]+cm[0,1])*100:.1f}%")

What does the confusion matrix tell you?

The confusion matrix compares your algorithm's predictions against the 45 segments where you deliberately injected leak anomalies in Step 2.

• TP (True Positives): segments with injected leaks that K-Means correctly flagged as Suspected or Critical Leak.
• FN (False Negatives): segments with injected leaks that K-Means missed (labeled Normal or Degrading).
• FP (False Positives): healthy segments that K-Means incorrectly flagged as leaks.

A detection rate above 80% is a strong result for unsupervised learning. Some false positives are expected because K-Means finds patterns in the data that may not perfectly align with the injected anomalies.

• Run the cell. You should see the confusion matrix with a detection rate (TP rate) above 80%.

• In the next cell, calculate risk scores and export the final CSV by pasting this code:

# Calculate risk score (0-100) based on distance from the Normal cluster center
normal_center = kmeans_final.cluster_centers_[label_map[cluster_severity.index[0]] == 'Normal']

# Use distance from Normal center, normalized to 0-100 scale
from scipy.spatial.distance import cdist
distances = cdist(X_scaled, kmeans_final.cluster_centers_[cluster_severity.index[0]].reshape(1, -1), metric='euclidean').flatten()

# Normalize to 0-100
risk_scores = (distances - distances.min()) / (distances.max() - distances.min()) * 100
segment_features['risk_score'] = risk_scores.round(1)

# Export to CSV
export_df = segment_features[['pipe_segment_id', 'cluster_label', 'risk_score']].copy()
export_df.columns = ['segment_id', 'cluster_label', 'risk_score']
export_df.to_csv('data/segment_risk_scores.csv', index=False)

print(f"Exported {len(export_df)} segment risk scores to data/segment_risk_scores.csv")
print(f"\nRisk score statistics by cluster:")
print(export_df.groupby('cluster_label')['risk_score'].describe().round(1))

How is the risk score calculated?

The risk score measures how far each segment is from the "Normal" cluster center in the standardized feature space. A segment that behaves identically to a healthy pipe scores near 0. A segment with extreme pressure drops and high night flow scores near 100.

This continuous score is more actionable than a binary label. It lets operations managers prioritize inspection by severity rather than treating all "Suspected Leak" segments equally.

• Run the cell. You should see confirmation that 500 segment risk scores were exported, along with a summary table showing "Critical Leak" segments averaging a risk score above 80.

• Verify the file exists by running this in a new cell:

# Quick verification of the exported file
verify = pd.read_csv('data/segment_risk_scores.csv')
print(verify.head(10))
print(f"\nShape: {verify.shape}")
print(f"Columns: {list(verify.columns)}")

• Run the cell. Confirm you see 500 rows with columns segment_id, cluster_label, and risk_score.

Seeing a FileNotFoundError?

Make sure your data/ directory exists relative to where your notebook is running. Check your notebook's working directory by running import os; print(os.getcwd()) in a cell.

Help me fix my file export

✔️ Awesome, I've got everything!

Great. Make sure you've saved your notebook with all the clustering cells executed.

ⓧ I'd like to double check the full code

Here is the complete K-Means clustering code for this step. All of these cells should appear in your notebook after the data cleaning and feature engineering code from the previous step.

# --- Step 4: K-Means Clustering for Leak Detection ---
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import confusion_matrix

# Load the cleaned sensor data
df_clean = pd.read_csv('data/fact_sensor_readings_clean.csv', parse_dates=['timestamp'])

# Merge with sensor metadata to get pipe_segment_id
df_sensors = pd.read_csv('data/dim_sensors.csv')
df_merged = df_clean.merge(df_sensors[['sensor_id', 'pipe_segment_id']], on='sensor_id', how='left')

# Aggregate to daily level per pipe segment
df_merged['date'] = df_merged['timestamp'].dt.date

daily_segment = df_merged.groupby(['pipe_segment_id', 'date']).agg(
    avg_pressure_drop=('pressure_differential', 'mean'),
    pressure_std=('pressure_differential', 'std'),
    night_flow_ratio=('night_flow_ratio', 'mean'),
    min_pressure=('reading_value', 'min'),
    flow_variance=('reading_value', 'var')
).reset_index()

print(f"Daily segment data: {daily_segment.shape[0]} rows, {daily_segment['pipe_segment_id'].nunique()} unique segments")

# Collapse daily data into one feature vector per segment
segment_features = daily_segment.groupby('pipe_segment_id').agg(
    avg_pressure_drop=('avg_pressure_drop', 'mean'),
    pressure_std=('pressure_std', 'mean'),
    night_flow_ratio=('night_flow_ratio', 'mean'),
    min_pressure=('min_pressure', 'mean'),
    flow_variance=('flow_variance', 'mean')
).reset_index()

# Drop rows with NaN values (segments with insufficient data)
segment_features = segment_features.dropna()

# Select the 5 features for clustering
feature_cols = ['avg_pressure_drop', 'pressure_std', 'night_flow_ratio', 'min_pressure', 'flow_variance']
X = segment_features[feature_cols]

# Standardize features so each contributes equally to distance calculations
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

print(f"Feature matrix shape: {X_scaled.shape}")
print(f"Features: {feature_cols}")

# Elbow Method: test k=2 through k=10
inertias = []
k_range = range(2, 11)

for k in k_range:
    kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
    kmeans.fit(X_scaled)
    inertias.append(kmeans.inertia_)

# Plot the elbow curve
plt.figure(figsize=(10, 6))
plt.plot(k_range, inertias, 'bo-', linewidth=2, markersize=8)
plt.xlabel('Number of Clusters (k)')
plt.ylabel('Inertia (Within-Cluster Sum of Squares)')
plt.title('Elbow Method for Optimal k')
plt.xticks(k_range)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('data/elbow_plot.png', dpi=150)
plt.show()

print("Look for the 'elbow' where the curve bends sharply.")

# Fit final KMeans model with 4 clusters
kmeans_final = KMeans(n_clusters=4, random_state=42, n_init=10)
segment_features['cluster'] = kmeans_final.fit_predict(X_scaled)

# Examine cluster centers to determine which cluster maps to which label
cluster_centers = pd.DataFrame(
    scaler.inverse_transform(kmeans_final.cluster_centers_),
    columns=feature_cols
)
print("Cluster Centers (original scale):")
print(cluster_centers.round(3))
print(f"\nCluster sizes: {segment_features['cluster'].value_counts().sort_index().to_dict()}")

# Map clusters to domain labels based on cluster center severity
# Sort clusters by avg_pressure_drop to assign labels from least to most severe
cluster_severity = cluster_centers['avg_pressure_drop'].sort_values()
label_map = {
    cluster_severity.index[0]: 'Normal',
    cluster_severity.index[1]: 'Degrading',
    cluster_severity.index[2]: 'Suspected Leak',
    cluster_severity.index[3]: 'Critical Leak'
}

segment_features['cluster_label'] = segment_features['cluster'].map(label_map)

print("Cluster label distribution:")
print(segment_features['cluster_label'].value_counts())
print(f"\nLabel mapping: {label_map}")

# Scatter plot: pressure_drop vs night_flow_ratio colored by cluster
plt.figure(figsize=(12, 8))
colors = {'Normal': '#2ecc71', 'Degrading': '#f39c12', 'Suspected Leak': '#e74c3c', 'Critical Leak': '#8e44ad'}

for label, color in colors.items():
    mask = segment_features['cluster_label'] == label
    plt.scatter(
        segment_features.loc[mask, 'avg_pressure_drop'],
        segment_features.loc[mask, 'night_flow_ratio'],
        c=color,
        label=f"{label} (n={mask.sum()})",
        alpha=0.7,
        s=60
    )

plt.xlabel('Average Pressure Drop')
plt.ylabel('Night Flow Ratio')
plt.title('Pipe Segment Clusters: Pressure Drop vs Night Flow Ratio')
plt.legend(loc='upper right', fontsize=10)
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('data/cluster_scatter_plot.png', dpi=150)
plt.show()

print("Scatter plot saved to data/cluster_scatter_plot.png")

# Validate against injected ground-truth leaks from data generation
# injected_leak_segments was created in Step 2 (list of 45 segment IDs)
actual_leak = segment_features['pipe_segment_id'].isin(injected_leak_segments).astype(int)
detected_leak = segment_features['cluster_label'].isin(['Suspected Leak', 'Critical Leak']).astype(int)

# Print confusion matrix
cm = confusion_matrix(actual_leak, detected_leak)
print("Confusion Matrix (Actual vs Detected):")
print(f"  TN={cm[0,0]}, FP={cm[0,1]}")
print(f"  FN={cm[1,0]}, TP={cm[1,1]}")
print(f"\nDetection rate: {cm[1,1]}/{cm[1,0]+cm[1,1]} = {cm[1,1]/(cm[1,0]+cm[1,1])*100:.1f}%")
print(f"False positive rate: {cm[0,1]}/{cm[0,0]+cm[0,1]} = {cm[0,1]/(cm[0,0]+cm[0,1])*100:.1f}%")

# Calculate risk score (0-100) based on distance from the Normal cluster center
from scipy.spatial.distance import cdist
distances = cdist(X_scaled, kmeans_final.cluster_centers_[cluster_severity.index[0]].reshape(1, -1), metric='euclidean').flatten()

# Normalize to 0-100
risk_scores = (distances - distances.min()) / (distances.max() - distances.min()) * 100
segment_features['risk_score'] = risk_scores.round(1)

# Export to CSV
export_df = segment_features[['pipe_segment_id', 'cluster_label', 'risk_score']].copy()
export_df.columns = ['segment_id', 'cluster_label', 'risk_score']
export_df.to_csv('data/segment_risk_scores.csv', index=False)

print(f"Exported {len(export_df)} segment risk scores to data/segment_risk_scores.csv")
print(f"\nRisk score statistics by cluster:")
print(export_df.groupby('cluster_label')['risk_score'].describe().round(1))

# Quick verification of the exported file
verify = pd.read_csv('data/segment_risk_scores.csv')
print(verify.head(10))
print(f"\nShape: {verify.shape}")
print(f"Columns: {list(verify.columns)}")

Your leak detection model is complete. You now have a scored list of every pipe segment in the network, ranked by how likely it is to be leaking. Next up, you'll use Facebook Prophet to forecast future water demand and identify which zones are at risk of exceeding capacity.

Forecast Water Demand with Prophet

Your K-Means clustering model has identified which pipe segments are leaking. Now you need to answer a different question: when will demand exceed supply?

Water utilities need to predict seasonal demand spikes weeks in advance to avoid capacity shortfalls. In this step, you will build a time-series forecasting model using Prophet that predicts daily water demand 90 days into the future for each pressure zone.

In this step, get ready to:

• Aggregate flow sensor data into daily demand per zone and format it for Prophet.
• Fit Prophet models with custom regressors and evaluate forecast accuracy.
• Export forecasts and identify the top 3 zones at risk of exceeding capacity.

Prepare demand data for Prophet

Prophet expects a DataFrame with exactly two columns: ds (a datestamp) and y (the numeric value to forecast). You need to aggregate the raw 15-minute flow readings into a single daily total per zone.

• In your aquaflow_analysis.ipynb notebook, create a new cell below the clustering section and add the following code to aggregate flow data per zone:

# --- Step 5: Time-Series Demand Forecasting with Prophet ---
import logging
from prophet import Prophet
from sklearn.metrics import mean_absolute_error

# Suppress verbose Prophet/cmdstanpy output
logging.getLogger('prophet').setLevel(logging.WARNING)
logging.getLogger('cmdstanpy').disabled = True

# Load cleaned sensor readings and sensor metadata
readings = pd.read_csv('data/fact_sensor_readings_clean.csv', parse_dates=['timestamp'])
sensors = pd.read_csv('data/dim_sensors.csv')

# Filter for flow sensors only
flow_sensor_ids = sensors[sensors['sensor_type'] == 'flow']['sensor_id'].tolist()
flow_readings = readings[readings['sensor_id'].isin(flow_sensor_ids)]

# Merge to get zone_id for each reading
flow_readings = flow_readings.merge(sensors[['sensor_id', 'zone_id']], on='sensor_id')

# Aggregate to daily total flow per zone
daily_flow = flow_readings.groupby(
    [flow_readings['timestamp'].dt.date, 'zone_id']
)['value'].sum().reset_index()
daily_flow.columns = ['ds', 'zone_id', 'y']
daily_flow['ds'] = pd.to_datetime(daily_flow['ds'])

print(f"Daily flow records: {len(daily_flow)}")
print(f"Date range: {daily_flow['ds'].min()} to {daily_flow['ds'].max()}")
print(f"Zones: {daily_flow['zone_id'].nunique()}")

What does this code do?

This code filters the cleaned sensor readings to keep only flow sensors (not pressure or temperature). It merges each reading with its zone assignment from dim_sensors.csv, then aggregates all 15-minute readings into a single daily sum per zone.

The output DataFrame has the two columns Prophet requires: ds for the date and y for total daily flow, plus zone_id so you can model each zone independently.

• Run the cell.
• You should see output confirming the number of daily flow records, the 2-year date range, and 12 zones.

Seeing an ImportError for prophet?

Make sure you installed Prophet in Step 1 with pip install prophet. Prophet requires the CmdStan backend, which is included automatically with the pip package on Python 3.12.

If you see a message about cmdstanpy, try restarting your kernel and running the cell again.

Help me troubleshoot my Prophet import error

• In the next cell, add the code to prepare the temperature regressor and create a train/validation split:

# Calculate the split point: first 20 months training, last 4 months validation
split_date = daily_flow['ds'].min() + pd.DateOffset(months=20)

# Prepare temperature regressor from temperature sensors
temp_sensor_ids = sensors[sensors['sensor_type'] == 'temperature']['sensor_id'].tolist()
temp_readings = readings[readings['sensor_id'].isin(temp_sensor_ids)]
temp_readings = temp_readings.merge(sensors[['sensor_id', 'zone_id']], on='sensor_id')

# Daily average temperature per zone
daily_temp = temp_readings.groupby(
    [temp_readings['timestamp'].dt.date, 'zone_id']
)['value'].mean().reset_index()
daily_temp.columns = ['ds', 'zone_id', 'temperature']
daily_temp['ds'] = pd.to_datetime(daily_temp['ds'])

# Merge temperature into daily flow data
daily_flow = daily_flow.merge(daily_temp, on=['ds', 'zone_id'], how='left')
daily_flow['temperature'] = daily_flow['temperature'].fillna(daily_flow['temperature'].mean())

# Create is_holiday feature (weekends as proxy for reduced commercial demand)
daily_flow['is_holiday'] = daily_flow['ds'].dt.dayofweek.isin([5, 6]).astype(int)

print(f"Training period: {daily_flow['ds'].min()} to {split_date}")
print(f"Validation period: {split_date} to {daily_flow['ds'].max()}")
print(f"Features added: temperature, is_holiday")

Why add temperature and holidays as regressors?

Water demand is heavily influenced by weather. Hot days drive up residential irrigation and cooling usage. Weekends and holidays shift commercial vs. residential consumption patterns. Adding these as regressors helps Prophet distinguish between genuine demand trends and weather-driven spikes.

• Run the cell.
• You should see the training period covering approximately 20 months and a 4-month validation window at the end.

Getting a merge error or NaN values everywhere?

Check that your dim_sensors.csv has a sensor_type column with the value temperature. If the column name or value is slightly different (e.g., temp), adjust the filter string to match your generated data.

Help me fix my temperature merge issue

Fit Prophet models and generate forecasts

Prophet models one time series at a time. Since you have 12 pressure zones, you will loop through each zone, fit a model, and collect the forecasts into a single DataFrame.

• In a new cell, add the Prophet model fitting loop that trains on each zone and calculates accuracy metrics:

# Fit a Prophet model for each zone and generate 90-day forecasts
all_forecasts = []

for zone_id in sorted(daily_flow['zone_id'].unique()):
    zone_data = daily_flow[daily_flow['zone_id'] == zone_id].copy()

    # Split into train and validation sets
    train = zone_data[zone_data['ds'] < split_date]
    validation = zone_data[zone_data['ds'] >= split_date]

    # Instantiate Prophet with weekly and yearly seasonality
    model = Prophet(
        weekly_seasonality=True,
        yearly_seasonality=True,
        daily_seasonality=False
    )

    # Add custom regressors
    model.add_regressor('temperature')
    model.add_regressor('is_holiday')

    # Fit the model on training data
    model.fit(train[['ds', 'y', 'temperature', 'is_holiday']])

    # Create future dataframe covering validation + 90 days ahead
    future = model.make_future_dataframe(periods=len(validation) + 90)

    # Add regressor values to the future dataframe
    future = future.merge(
        zone_data[['ds', 'temperature', 'is_holiday']], on='ds', how='left'
    )
    future['temperature'] = future['temperature'].fillna(zone_data['temperature'].mean())
    future['is_holiday'] = future['is_holiday'].fillna(
        future['ds'].dt.dayofweek.isin([5, 6]).astype(int)
    )

    # Generate predictions
    forecast = model.predict(future)
    forecast['zone_id'] = zone_id

    # Calculate accuracy on the validation set
    val_forecast = forecast[forecast['ds'].isin(validation['ds'])]
    val_merged = val_forecast.merge(validation[['ds', 'y']], on='ds')

    mae = mean_absolute_error(val_merged['y'], val_merged['yhat'])
    mape = (abs(val_merged['y'] - val_merged['yhat']) / val_merged['y']).mean() * 100

    print(f"Zone {zone_id}: MAE = {mae:.2f}, MAPE = {mape:.2f}%")

    all_forecasts.append(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper', 'zone_id']])

# Keep the last model and forecast for decomposition plot
last_model = model
last_forecast = forecast

print(f"\nAll {daily_flow['zone_id'].nunique()} zone models fitted successfully.")

What does this code do?

For each zone, the loop creates a fresh Prophet model with weekly and yearly seasonality patterns enabled. The add_regressor calls tell Prophet that temperature and holiday status influence demand beyond what seasonality alone explains.

After fitting, make_future_dataframe extends the timeline beyond the training data. The model then predicts yhat (point estimate), yhat_lower, and yhat_upper (uncertainty bounds). MAE measures average error in the same units as flow. MAPE expresses error as a percentage of actual demand.

• Run the cell. This takes 1-3 minutes as Prophet fits 12 separate models.
• You should see MAE and MAPE printed for each zone, followed by a confirmation that all 12 models fitted successfully.

What MAPE values should I expect?

For daily water demand forecasting, MAPE values between 5-15% are typical. Values above 20% suggest the model is struggling with that zone's pattern. Since this is simulated data with injected anomalies, some zones may show higher error where leaks distort flow readings.

Model fitting is extremely slow or hanging?

Prophet's optimization can be slow if the training set is very large. Verify that your training data per zone has fewer than 1,000 rows (roughly 20 months of daily data). If your aggregation produced 15-minute intervals instead of daily, revisit the groupby in the previous cell.

Help me debug slow Prophet model fitting

Export forecast results and identify capacity risks

The final piece is saving the forecasts to CSV, visualizing how Prophet decomposed the demand signal, and flagging zones that risk exceeding capacity in the next 90 days.

• In a new cell, add the export and capacity analysis code:

# Combine all zone forecasts and export to CSV
demand_forecast = pd.concat(all_forecasts, ignore_index=True)
demand_forecast.to_csv('data/demand_forecast.csv', index=False)
print(f"Exported demand_forecast.csv: {len(demand_forecast)} rows")
print(f"Columns: {list(demand_forecast.columns)}")

# Create decomposition plot showing trend + seasonality components
fig = last_model.plot_components(last_forecast)
fig.savefig('data/demand_decomposition.png', dpi=150, bbox_inches='tight')
print("Saved decomposition plot to data/demand_decomposition.png")

What is a decomposition plot?

Prophet breaks the forecast into additive components: an overall trend (is demand growing or shrinking?), weekly seasonality (which days of the week peak?), and yearly seasonality (which months peak?). The decomposition plot shows each component separately so you can understand what drives demand.

• Run the cell.
• You should see a row count for the exported CSV and a multi-panel decomposition chart rendered inline.

• In a final cell for this step, add the capacity risk identification code:

# Identify top 3 zones at risk of exceeding capacity in next 90 days
# Define capacity threshold as 120% of historical average daily flow per zone
zone_capacity = daily_flow.groupby('zone_id')['y'].mean() * 1.2

# Get future-only predictions (beyond the historical data range)
future_start = daily_flow['ds'].max() + pd.Timedelta(days=1)
future_forecasts = demand_forecast[demand_forecast['ds'] >= future_start]

# Calculate max forecasted demand per zone vs capacity threshold
zone_max_forecast = future_forecasts.groupby('zone_id')['yhat_upper'].max()
capacity_risk = (zone_max_forecast / zone_capacity).sort_values(ascending=False)

print("Top 3 zones at risk of exceeding capacity in next 90 days:")
print("=" * 60)
for zone_id in capacity_risk.head(3).index:
    risk_pct = capacity_risk[zone_id] * 100
    print(f"  Zone {zone_id}: forecasted peak at {risk_pct:.1f}% of capacity threshold")

print(f"\nDemand forecasting complete. Results saved to data/demand_forecast.csv")

How does the capacity threshold work?

The code defines capacity as 120% of each zone's historical average daily flow. This is a simplified threshold. In production, utilities would use actual infrastructure capacity ratings. The upper confidence bound (yhat_upper) represents the worst-case forecast. Zones where this exceeds the threshold need proactive attention.

• Run the cell.
• You should see the top 3 at-risk zones printed with their forecasted peak as a percentage of capacity.

• Verify that data/demand_forecast.csv exists by running:

# Verify the exported file
check = pd.read_csv('data/demand_forecast.csv')
print(f"demand_forecast.csv: {len(check)} rows, columns: {list(check.columns)}")

• You should see the file has columns ds, yhat, yhat_lower, yhat_upper, and zone_id.

CSV is empty or has unexpected row count?

Check that the all_forecasts list was populated. If the model fitting loop encountered an error partway through, some zones may be missing. Re-run the fitting cell and check for errors in the output.

Also ensure you ran the cells in order. The demand_forecast variable depends on all_forecasts from the fitting cell completing successfully.

Help me troubleshoot my demand forecast export

✔️ Awesome, I've got everything!

Great. Make sure you have saved your notebook before moving on.

ⓧ I'd like to double check the full code

Here is all the code added to your notebook in this step. Compare it against what you have:

# --- Step 5: Time-Series Demand Forecasting with Prophet ---
import logging
from prophet import Prophet
from sklearn.metrics import mean_absolute_error

# Suppress verbose Prophet/cmdstanpy output
logging.getLogger('prophet').setLevel(logging.WARNING)
logging.getLogger('cmdstanpy').disabled = True

# Load cleaned sensor readings and sensor metadata
readings = pd.read_csv('data/fact_sensor_readings_clean.csv', parse_dates=['timestamp'])
sensors = pd.read_csv('data/dim_sensors.csv')

# Filter for flow sensors only
flow_sensor_ids = sensors[sensors['sensor_type'] == 'flow']['sensor_id'].tolist()
flow_readings = readings[readings['sensor_id'].isin(flow_sensor_ids)]

# Merge to get zone_id for each reading
flow_readings = flow_readings.merge(sensors[['sensor_id', 'zone_id']], on='sensor_id')

# Aggregate to daily total flow per zone
daily_flow = flow_readings.groupby(
    [flow_readings['timestamp'].dt.date, 'zone_id']
)['value'].sum().reset_index()
daily_flow.columns = ['ds', 'zone_id', 'y']
daily_flow['ds'] = pd.to_datetime(daily_flow['ds'])

print(f"Daily flow records: {len(daily_flow)}")
print(f"Date range: {daily_flow['ds'].min()} to {daily_flow['ds'].max()}")
print(f"Zones: {daily_flow['zone_id'].nunique()}")

# Calculate the split point: first 20 months training, last 4 months validation
split_date = daily_flow['ds'].min() + pd.DateOffset(months=20)

# Prepare temperature regressor from temperature sensors
temp_sensor_ids = sensors[sensors['sensor_type'] == 'temperature']['sensor_id'].tolist()
temp_readings = readings[readings['sensor_id'].isin(temp_sensor_ids)]
temp_readings = temp_readings.merge(sensors[['sensor_id', 'zone_id']], on='sensor_id')

# Daily average temperature per zone
daily_temp = temp_readings.groupby(
    [temp_readings['timestamp'].dt.date, 'zone_id']
)['value'].mean().reset_index()
daily_temp.columns = ['ds', 'zone_id', 'temperature']
daily_temp['ds'] = pd.to_datetime(daily_temp['ds'])

# Merge temperature into daily flow data
daily_flow = daily_flow.merge(daily_temp, on=['ds', 'zone_id'], how='left')
daily_flow['temperature'] = daily_flow['temperature'].fillna(daily_flow['temperature'].mean())

# Create is_holiday feature (weekends as proxy for reduced commercial demand)
daily_flow['is_holiday'] = daily_flow['ds'].dt.dayofweek.isin([5, 6]).astype(int)

print(f"Training period: {daily_flow['ds'].min()} to {split_date}")
print(f"Validation period: {split_date} to {daily_flow['ds'].max()}")
print(f"Features added: temperature, is_holiday")

# Fit a Prophet model for each zone and generate 90-day forecasts
all_forecasts = []

for zone_id in sorted(daily_flow['zone_id'].unique()):
    zone_data = daily_flow[daily_flow['zone_id'] == zone_id].copy()

    # Split into train and validation sets
    train = zone_data[zone_data['ds'] < split_date]
    validation = zone_data[zone_data['ds'] >= split_date]

    # Instantiate Prophet with weekly and yearly seasonality
    model = Prophet(
        weekly_seasonality=True,
        yearly_seasonality=True,
        daily_seasonality=False
    )

    # Add custom regressors
    model.add_regressor('temperature')
    model.add_regressor('is_holiday')

    # Fit the model on training data
    model.fit(train[['ds', 'y', 'temperature', 'is_holiday']])

    # Create future dataframe covering validation + 90 days ahead
    future = model.make_future_dataframe(periods=len(validation) + 90)

    # Add regressor values to the future dataframe
    future = future.merge(
        zone_data[['ds', 'temperature', 'is_holiday']], on='ds', how='left'
    )
    future['temperature'] = future['temperature'].fillna(zone_data['temperature'].mean())
    future['is_holiday'] = future['is_holiday'].fillna(
        future['ds'].dt.dayofweek.isin([5, 6]).astype(int)
    )

    # Generate predictions
    forecast = model.predict(future)
    forecast['zone_id'] = zone_id

    # Calculate accuracy on the validation set
    val_forecast = forecast[forecast['ds'].isin(validation['ds'])]
    val_merged = val_forecast.merge(validation[['ds', 'y']], on='ds')

    mae = mean_absolute_error(val_merged['y'], val_merged['yhat'])
    mape = (abs(val_merged['y'] - val_merged['yhat']) / val_merged['y']).mean() * 100

    print(f"Zone {zone_id}: MAE = {mae:.2f}, MAPE = {mape:.2f}%")

    all_forecasts.append(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper', 'zone_id']])

# Keep the last model and forecast for decomposition plot
last_model = model
last_forecast = forecast

print(f"\nAll {daily_flow['zone_id'].nunique()} zone models fitted successfully.")

# Combine all zone forecasts and export to CSV
demand_forecast = pd.concat(all_forecasts, ignore_index=True)
demand_forecast.to_csv('data/demand_forecast.csv', index=False)
print(f"Exported demand_forecast.csv: {len(demand_forecast)} rows")
print(f"Columns: {list(demand_forecast.columns)}")

# Create decomposition plot showing trend + seasonality components
fig = last_model.plot_components(last_forecast)
fig.savefig('data/demand_decomposition.png', dpi=150, bbox_inches='tight')
print("Saved decomposition plot to data/demand_decomposition.png")

# Identify top 3 zones at risk of exceeding capacity in next 90 days
# Define capacity threshold as 120% of historical average daily flow per zone
zone_capacity = daily_flow.groupby('zone_id')['y'].mean() * 1.2

# Get future-only predictions (beyond the historical data range)
future_start = daily_flow['ds'].max() + pd.Timedelta(days=1)
future_forecasts = demand_forecast[demand_forecast['ds'] >= future_start]

# Calculate max forecasted demand per zone vs capacity threshold
zone_max_forecast = future_forecasts.groupby('zone_id')['yhat_upper'].max()
capacity_risk = (zone_max_forecast / zone_capacity).sort_values(ascending=False)

print("Top 3 zones at risk of exceeding capacity in next 90 days:")
print("=" * 60)
for zone_id in capacity_risk.head(3).index:
    risk_pct = capacity_risk[zone_id] * 100
    print(f"  Zone {zone_id}: forecasted peak at {risk_pct:.1f}% of capacity threshold")

print(f"\nDemand forecasting complete. Results saved to data/demand_forecast.csv")

You now have leak detection (clustering) and demand forecasting (Prophet) working together. Next up, you will load all of this data into a PostgreSQL star schema and write production-grade SQL views that power your final dashboard.

Design & Load the Star Schema in PostgreSQL

Your Python analysis has produced three powerful outputs: cleaned sensor readings, K-Means cluster risk scores, and Prophet demand forecasts. Right now all of this intelligence lives in flat CSV files. To power an interactive Power BI dashboard, you need a structured database that can answer complex cross-cutting questions fast.

A star schema is the standard data model for analytics databases. Fact tables hold measurable events (sensor readings, leak detections, forecasts), while dimension tables hold the descriptive context (dates, zones, pipe segments). In this step you will design both layers in PostgreSQL, then bulk-load every table from Python.

In this step, get ready to:

• Create 5 dimension tables with primary keys, NOT NULL constraints, and CHECK constraints.
• Create 3 fact tables with foreign keys and performance indexes.
• Bulk-load all tables from Python using SQLAlchemy and verify referential integrity.

Create dimension tables with constraints

Dimension tables hold the descriptive attributes that give meaning to raw numbers. Each dimension has a primary key that fact tables will reference, plus constraints that enforce data quality at the database level.

• Inside your project's sql folder, create a new file called ddl_dimensions.sql and paste the following DDL:

-- Dimension: Date (generated programmatically, not from CSV)
CREATE TABLE dim_date (
    date_key INTEGER PRIMARY KEY,
    date DATE NOT NULL UNIQUE,
    year INTEGER NOT NULL,
    month INTEGER NOT NULL CHECK (month BETWEEN 1 AND 12),
    day INTEGER NOT NULL CHECK (day BETWEEN 1 AND 31),
    fiscal_quarter INTEGER NOT NULL CHECK (fiscal_quarter BETWEEN 1 AND 4),
    week_number INTEGER NOT NULL CHECK (week_number BETWEEN 1 AND 53),
    is_holiday BOOLEAN NOT NULL DEFAULT FALSE,
    day_of_week INTEGER NOT NULL CHECK (day_of_week BETWEEN 0 AND 6)
);

-- Dimension: Zone
CREATE TABLE dim_zone (
    zone_id INTEGER PRIMARY KEY,
    zone_name VARCHAR(100) NOT NULL,
    population_served INTEGER NOT NULL CHECK (population_served > 0),
    elevation_m NUMERIC(6,2) NOT NULL,
    reservoir_id INTEGER NOT NULL
);

-- Dimension: Pipe Segment
CREATE TABLE dim_pipe_segment (
    segment_id INTEGER PRIMARY KEY,
    material VARCHAR(50) NOT NULL CHECK (material IN ('cast_iron', 'PVC', 'ductile_iron', 'asbestos_cement')),
    diameter_mm NUMERIC(6,1) NOT NULL CHECK (diameter_mm > 0),
    length_m NUMERIC(8,2) NOT NULL CHECK (length_m > 0),
    installation_year INTEGER NOT NULL CHECK (installation_year BETWEEN 1900 AND 2030),
    zone_id INTEGER NOT NULL REFERENCES dim_zone(zone_id),
    upstream_segment_id INTEGER,
    downstream_segment_id INTEGER
);

-- Dimension: Sensor
CREATE TABLE dim_sensor (
    sensor_id INTEGER PRIMARY KEY,
    sensor_type VARCHAR(50) NOT NULL CHECK (sensor_type IN ('pressure', 'flow', 'temperature')),
    zone_id INTEGER NOT NULL REFERENCES dim_zone(zone_id),
    pipe_segment_id INTEGER NOT NULL REFERENCES dim_pipe_segment(segment_id),
    installation_date DATE NOT NULL,
    manufacturer VARCHAR(100) NOT NULL,
    calibration_status VARCHAR(50) NOT NULL
);

-- Dimension: Maintenance Type
CREATE TABLE dim_maintenance_type (
    maintenance_type_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    event_type VARCHAR(50) NOT NULL UNIQUE
);

What does this DDL do?

Each CREATE TABLE statement defines one dimension of your star schema. The CHECK constraints enforce data quality rules directly in the database. For example, month BETWEEN 1 AND 12 rejects any invalid month value before it ever reaches a dashboard.

The REFERENCES clauses on dim_pipe_segment and dim_sensor create foreign key relationships back to dim_zone. This enforces referential integrity so no sensor can reference a zone that does not exist.

Notice that dim_date uses a manually assigned date_key (you will generate it from Python), while dim_maintenance_type uses GENERATED ALWAYS AS IDENTITY to auto-increment its key.

• Save sql/ddl_dimensions.sql.

Why does table creation order matter?

PostgreSQL requires that a referenced table already exists before you can create a foreign key pointing to it. That is why dim_zone is created before dim_pipe_segment and dim_sensor (both reference dim_zone).

Create fact tables with foreign keys

Fact tables store the measurable events your dashboard will aggregate. Each fact row links to dimension tables through foreign keys, and indexes on those keys make cross-table joins fast.

• In the sql folder, create a new file called ddl_facts.sql and paste the following DDL:

-- Fact: Daily Sensor Readings (aggregated from 15-min intervals)
CREATE TABLE fact_sensor_daily (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    date_key INTEGER NOT NULL REFERENCES dim_date(date_key),
    sensor_id INTEGER NOT NULL REFERENCES dim_sensor(sensor_id),
    zone_id INTEGER NOT NULL REFERENCES dim_zone(zone_id),
    segment_id INTEGER NOT NULL REFERENCES dim_pipe_segment(segment_id),
    avg_pressure NUMERIC(8,3),
    min_pressure NUMERIC(8,3),
    max_pressure NUMERIC(8,3),
    avg_flow NUMERIC(10,3),
    avg_temperature NUMERIC(6,2),
    reading_count INTEGER NOT NULL,
    avg_pressure_drop NUMERIC(8,3),
    night_flow_ratio NUMERIC(6,4)
);

CREATE INDEX idx_fact_sensor_daily_date ON fact_sensor_daily(date_key);
CREATE INDEX idx_fact_sensor_daily_zone ON fact_sensor_daily(zone_id);
CREATE INDEX idx_fact_sensor_daily_segment ON fact_sensor_daily(segment_id);

-- Fact: Leak Detection (from K-Means clustering results)
CREATE TABLE fact_leak_detection (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    segment_id INTEGER NOT NULL REFERENCES dim_pipe_segment(segment_id),
    date_key INTEGER NOT NULL REFERENCES dim_date(date_key),
    zone_id INTEGER NOT NULL REFERENCES dim_zone(zone_id),
    cluster_label VARCHAR(50) NOT NULL,
    risk_score NUMERIC(5,2) NOT NULL CHECK (risk_score BETWEEN 0 AND 100)
);

CREATE INDEX idx_fact_leak_detection_date ON fact_leak_detection(date_key);
CREATE INDEX idx_fact_leak_detection_zone ON fact_leak_detection(zone_id);
CREATE INDEX idx_fact_leak_detection_segment ON fact_leak_detection(segment_id);

-- Fact: Demand Forecast (from Prophet model outputs)
CREATE TABLE fact_demand_forecast (
    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    date_key INTEGER NOT NULL REFERENCES dim_date(date_key),
    zone_id INTEGER NOT NULL REFERENCES dim_zone(zone_id),
    yhat NUMERIC(12,3) NOT NULL,
    yhat_lower NUMERIC(12,3) NOT NULL,
    yhat_upper NUMERIC(12,3) NOT NULL
);

CREATE INDEX idx_fact_demand_forecast_date ON fact_demand_forecast(date_key);
CREATE INDEX idx_fact_demand_forecast_zone ON fact_demand_forecast(zone_id);

What do the indexes do?

The CREATE INDEX statements build lookup structures on date_key, zone_id, and segment_id columns. Without indexes, PostgreSQL would scan every row in a fact table to answer a filtered query. With indexes, lookups on these high-cardinality columns become near-instant.

The CHECK (risk_score BETWEEN 0 AND 100) constraint on fact_leak_detection ensures no invalid risk scores slip through during loading.

• Save sql/ddl_facts.sql.

Not sure where to save the SQL files?

Both files go in the sql/ folder you created in Step 1. If you skipped creating it, make the folder now with mkdir sql from your project root.

Help me find where my sql folder should be

Load data from Python into PostgreSQL

With the schema defined, you will now execute the DDL from your notebook using SQLAlchemy, generate the dim_date table programmatically, and bulk-load every table from your CSV files.

• In aquaflow_analysis.ipynb, create a new cell and establish the database connection by running this code:

from sqlalchemy import create_engine, text
import pandas as pd
import numpy as np
from datetime import date, timedelta

# Connect to PostgreSQL using psycopg2 driver
engine = create_engine(
    'postgresql+psycopg2://postgres:your-password-here@localhost:5432/aquaflow_analytics'
)

# Test the connection
with engine.connect() as conn:
    result = conn.execute(text("SELECT version()"))
    print("Connected:", result.fetchone()[0])

What does this code do?

The create_engine call builds a connection pool to your local PostgreSQL instance. The connection string format is postgresql+psycopg2://user:password@host:port/database.

Replace your-password-here with the password you set during PostgreSQL installation. The test query confirms the connection is live before you run any DDL.

• Run the cell. You should see a line starting with Connected: PostgreSQL 17 followed by version details.

Connection refused or authentication failed?

If you see a connection refused error, confirm PostgreSQL is running. On Windows, check the Services app for a running postgresql-x64-17 service.

If authentication fails, double-check the password in your connection string matches what you set during installation. Help me troubleshoot my PostgreSQL connection

• In a new cell, execute both DDL files and generate the dim_date dimension by running this code:

# Execute dimension DDL
with open('sql/ddl_dimensions.sql', 'r') as f:
    ddl_dims = f.read()

with engine.connect() as conn:
    conn.execute(text(ddl_dims))
    conn.commit()
    print("Dimension tables created")

# Execute fact DDL
with open('sql/ddl_facts.sql', 'r') as f:
    ddl_facts = f.read()

with engine.connect() as conn:
    conn.execute(text(ddl_facts))
    conn.commit()
    print("Fact tables created")

# Generate dim_date programmatically (2 years of dates)
start_date = date(2022, 1, 1)
end_date = date(2023, 12, 31)
dates = pd.date_range(start_date, end_date, freq='D')

# US federal holidays (simplified list)
holidays = [
    '2022-01-01', '2022-01-17', '2022-02-21', '2022-05-30',
    '2022-07-04', '2022-09-05', '2022-11-24', '2022-12-25',
    '2023-01-01', '2023-01-16', '2023-02-20', '2023-05-29',
    '2023-07-04', '2023-09-04', '2023-11-23', '2023-12-25'
]

# Build the date dimension DataFrame
dim_date = pd.DataFrame({
    'date_key': range(1, len(dates) + 1),
    'date': dates,
    'year': dates.year,
    'month': dates.month,
    'day': dates.day,
    'fiscal_quarter': dates.quarter,
    'week_number': dates.isocalendar().week.astype(int),
    'is_holiday': [d.strftime('%Y-%m-%d') in holidays for d in dates],
    'day_of_week': dates.dayofweek
})

# Load dim_date into PostgreSQL
dim_date.to_sql('dim_date', engine, if_exists='append', index=False)
print(f"dim_date loaded: {len(dim_date)} rows")

What does this code do?

The first two blocks read your SQL files and execute them against the live database, creating all 8 tables (5 dimensions, 3 facts) with their constraints and indexes.

The dim_date generation creates 730 rows (2 full years) with fiscal quarters, ISO week numbers, and holiday flags. This table does not come from a CSV because date dimensions are always generated to ensure completeness with no gaps.

The to_sql() method uses if_exists='append' because the table already exists from the DDL. It inserts rows without dropping or recreating the schema.

• Run the cell. You should see three confirmation lines ending with dim_date loaded: 730 rows.

Seeing a 'relation already exists' error?

This means you ran the DDL cell more than once. Add DROP TABLE IF EXISTS statements before each CREATE TABLE in your SQL files, or connect to pgAdmin and drop the tables manually before re-running.

Help me fix the 'relation already exists' error

• In a new cell, load the remaining dimension tables from your CSV files by running this code:

# Load dim_zone from CSV
dim_zones = pd.read_csv('data/dim_zones.csv')
dim_zones.to_sql('dim_zone', engine, if_exists='append', index=False)
print(f"dim_zone loaded: {len(dim_zones)} rows")

# Load dim_pipe_segment from CSV
dim_segments = pd.read_csv('data/dim_pipe_segments.csv')
dim_segments.to_sql('dim_pipe_segment', engine, if_exists='append', index=False)
print(f"dim_pipe_segment loaded: {len(dim_segments)} rows")

# Load dim_sensor from CSV
dim_sensors = pd.read_csv('data/dim_sensors.csv')
dim_sensors.to_sql('dim_sensor', engine, if_exists='append', index=False)
print(f"dim_sensor loaded: {len(dim_sensors)} rows")

# Load dim_maintenance_type (extract distinct types from maintenance logs)
maintenance_logs = pd.read_csv('data/fact_maintenance_logs.csv')
event_types = maintenance_logs['event_type'].unique()
dim_maint_type = pd.DataFrame({'event_type': event_types})
dim_maint_type.to_sql('dim_maintenance_type', engine, if_exists='append', index=False)
print(f"dim_maintenance_type loaded: {len(dim_maint_type)} rows")

What does this code do?

Each to_sql() call reads a CSV and bulk-inserts its rows into the matching PostgreSQL table. The foreign key constraints you defined in the DDL are enforced during insert. If a sensor references a zone_id that does not exist in dim_zone, PostgreSQL will reject the row.

The dim_maintenance_type table is derived from the distinct event_type values in your maintenance logs. This avoids duplicating long strings across every maintenance record.

• Run the cell. You should see four printed confirmations: 12 zone rows, 500 pipe segment rows, 200 sensor rows, and a small number of maintenance type rows.

Foreign key violation on dim_sensor or dim_pipe_segment?

Loading order matters. You must load dim_zone before dim_pipe_segment, and dim_pipe_segment before dim_sensor. If you rearranged the code, restore the order shown above and re-run.

Help me fix this foreign key constraint error

• In a new cell, aggregate your sensor readings to the daily level and load all three fact tables by running this code:

# Aggregate clean sensor readings to daily level per sensor
readings = pd.read_csv('data/fact_sensor_readings_clean.csv', parse_dates=['timestamp'])
readings['date'] = readings['timestamp'].dt.date

fact_sensor_daily = readings.groupby(['date', 'sensor_id']).agg(
    avg_pressure=('pressure', 'mean'),
    min_pressure=('pressure', 'min'),
    max_pressure=('pressure', 'max'),
    avg_flow=('flow', 'mean'),
    avg_temperature=('temperature', 'mean'),
    reading_count=('pressure', 'count'),
    avg_pressure_drop=('pressure_drop', 'mean'),
    night_flow_ratio=('night_flow_ratio', 'mean')
).reset_index()

# Map dates to date_key using dim_date lookup
date_key_map = dim_date.set_index('date')['date_key'].to_dict()
fact_sensor_daily['date_key'] = fact_sensor_daily['date'].map(
    lambda d: date_key_map.get(pd.Timestamp(d))
)

# Add zone_id and segment_id from sensor dimension
sensor_lookup = dim_sensors.set_index('sensor_id')[['zone_id', 'pipe_segment_id']]
fact_sensor_daily = fact_sensor_daily.merge(
    sensor_lookup, left_on='sensor_id', right_index=True
)
fact_sensor_daily.rename(columns={'pipe_segment_id': 'segment_id'}, inplace=True)

# Drop rows outside our date range and cast date_key to int
fact_sensor_daily.dropna(subset=['date_key'], inplace=True)
fact_sensor_daily['date_key'] = fact_sensor_daily['date_key'].astype(int)

# Select columns matching the table schema
fact_sensor_daily_load = fact_sensor_daily[[
    'date_key', 'sensor_id', 'zone_id', 'segment_id',
    'avg_pressure', 'min_pressure', 'max_pressure',
    'avg_flow', 'avg_temperature', 'reading_count',
    'avg_pressure_drop', 'night_flow_ratio'
]]
fact_sensor_daily_load.to_sql('fact_sensor_daily', engine, if_exists='append', index=False)
print(f"fact_sensor_daily loaded: {len(fact_sensor_daily_load):,} rows")

# Load fact_leak_detection from cluster results
risk_scores = pd.read_csv('data/segment_risk_scores.csv')
risk_scores['date_key'] = int(dim_date['date_key'].max())  # detection date = last date in range
segment_zone = dim_segments.set_index('segment_id')['zone_id']
risk_scores['zone_id'] = risk_scores['segment_id'].map(segment_zone)

fact_leak_load = risk_scores[['segment_id', 'date_key', 'zone_id', 'cluster_label', 'risk_score']]
fact_leak_load.to_sql('fact_leak_detection', engine, if_exists='append', index=False)
print(f"fact_leak_detection loaded: {len(fact_leak_load)} rows")

# Load fact_demand_forecast from Prophet outputs
forecast = pd.read_csv('data/demand_forecast.csv', parse_dates=['ds'])
forecast['date_key'] = forecast['ds'].map(
    lambda d: date_key_map.get(pd.Timestamp(d.date()) if hasattr(d, 'date') else pd.Timestamp(d))
)
forecast.dropna(subset=['date_key'], inplace=True)
forecast['date_key'] = forecast['date_key'].astype(int)

fact_forecast_load = forecast[['date_key', 'zone_id', 'yhat', 'yhat_lower', 'yhat_upper']]
fact_forecast_load.to_sql('fact_demand_forecast', engine, if_exists='append', index=False)
print(f"fact_demand_forecast loaded: {len(fact_forecast_load):,} rows")

What does this code do?

The fact_sensor_daily block aggregates your ~2 million 15-minute readings down to daily averages per sensor. It maps each date to the integer date_key from dim_date and looks up each sensor's zone and pipe segment from the dimension table.

The fact_leak_detection block loads your K-Means cluster results directly. It assigns all detections the last date in the range because clustering produced a single point-in-time risk assessment.

The fact_demand_forecast block maps Prophet's ds dates to date_key values. Rows with dates outside your dim_date range are dropped to satisfy the foreign key constraint.

• Run the cell. This may take 30-60 seconds due to the large sensor readings file. You should see three printed confirmations with row counts.

Seeing a CHECK constraint or foreign key error during loading?

A CHECK constraint error (e.g., on risk_score) means a value in your CSV falls outside the allowed range. Inspect the failing rows with a filter: risk_scores[risk_scores['risk_score'] > 100].

A foreign key error means a date_key or zone_id in your fact data does not exist in the corresponding dimension table. Make sure all dimension cells ran successfully before loading facts.

Help me debug this constraint violation

• In a new cell, verify row counts and run a sanity-check join query by running this code:

# Verify row counts across all tables
with engine.connect() as conn:
    tables = [
        'dim_date', 'dim_zone', 'dim_pipe_segment', 'dim_sensor',
        'dim_maintenance_type', 'fact_sensor_daily',
        'fact_leak_detection', 'fact_demand_forecast'
    ]
    print("Row Counts:")
    print("-" * 40)
    for table in tables:
        count = conn.execute(text(f"SELECT COUNT(*) FROM {table}")).scalar()
        print(f"  {table}: {count:,} rows")

# Sanity-check join: top 5 zones by average leak risk
sanity_query = """
SELECT
    dz.zone_name,
    COUNT(fld.id) AS leak_count,
    ROUND(AVG(fld.risk_score), 2) AS avg_risk_score
FROM fact_leak_detection fld
JOIN dim_zone dz ON fld.zone_id = dz.zone_id
WHERE fld.cluster_label IN ('Suspected Leak', 'Critical Leak')
GROUP BY dz.zone_name
ORDER BY avg_risk_score DESC
LIMIT 5;
"""

result = pd.read_sql(sanity_query, engine)
print("\nTop 5 Zones by Leak Risk:")
print(result.to_string(index=False))

What does this code do?

The first block loops through every table and prints its row count. This confirms that all your data landed in the database without silent failures.

The sanity-check query joins fact_leak_detection to dim_zone and filters for high-risk clusters. If this returns results, your foreign keys are working and the star schema is ready for Power BI.

• Run the cell. You should see a table of row counts (730 for dim_date, 12 for dim_zone, 500 for dim_pipe_segment, 200 for dim_sensor) followed by a 5-row results table showing zone names and risk scores.
• Save aquaflow_analysis.ipynb.

Sanity query returns zero rows?

Check that your segment_risk_scores.csv contains the exact cluster labels Suspected Leak and Critical Leak (case-sensitive). If your Step 4 used different labels, update the WHERE clause to match.

Help me debug why my sanity query returns no rows

✔️ Awesome, I've got everything!

Great. Double check you have saved both SQL files and the notebook.

ⓧ I'd like to double check the full code

Here is the full loading code that should be in your notebook cells for this step:

# Cell 1: Connect to PostgreSQL
from sqlalchemy import create_engine, text
import pandas as pd
import numpy as np
from datetime import date, timedelta

engine = create_engine(
    'postgresql+psycopg2://postgres:your-password-here@localhost:5432/aquaflow_analytics'
)

with engine.connect() as conn:
    result = conn.execute(text("SELECT version()"))
    print("Connected:", result.fetchone()[0])

# Cell 2: Execute DDL and generate dim_date
with open('sql/ddl_dimensions.sql', 'r') as f:
    ddl_dims = f.read()

with engine.connect() as conn:
    conn.execute(text(ddl_dims))
    conn.commit()
    print("Dimension tables created")

with open('sql/ddl_facts.sql', 'r') as f:
    ddl_facts = f.read()

with engine.connect() as conn:
    conn.execute(text(ddl_facts))
    conn.commit()
    print("Fact tables created")

start_date = date(2022, 1, 1)
end_date = date(2023, 12, 31)
dates = pd.date_range(start_date, end_date, freq='D')

holidays = [
    '2022-01-01', '2022-01-17', '2022-02-21', '2022-05-30',
    '2022-07-04', '2022-09-05', '2022-11-24', '2022-12-25',
    '2023-01-01', '2023-01-16', '2023-02-20', '2023-05-29',
    '2023-07-04', '2023-09-04', '2023-11-23', '2023-12-25'
]

dim_date = pd.DataFrame({
    'date_key': range(1, len(dates) + 1),
    'date': dates,
    'year': dates.year,
    'month': dates.month,
    'day': dates.day,
    'fiscal_quarter': dates.quarter,
    'week_number': dates.isocalendar().week.astype(int),
    'is_holiday': [d.strftime('%Y-%m-%d') in holidays for d in dates],
    'day_of_week': dates.dayofweek
})

dim_date.to_sql('dim_date', engine, if_exists='append', index=False)
print(f"dim_date loaded: {len(dim_date)} rows")

# Cell 3: Load remaining dimensions from CSV
dim_zones = pd.read_csv('data/dim_zones.csv')
dim_zones.to_sql('dim_zone', engine, if_exists='append', index=False)
print(f"dim_zone loaded: {len(dim_zones)} rows")

dim_segments = pd.read_csv('data/dim_pipe_segments.csv')
dim_segments.to_sql('dim_pipe_segment', engine, if_exists='append', index=False)
print(f"dim_pipe_segment loaded: {len(dim_segments)} rows")

dim_sensors = pd.read_csv('data/dim_sensors.csv')
dim_sensors.to_sql('dim_sensor', engine, if_exists='append', index=False)
print(f"dim_sensor loaded: {len(dim_sensors)} rows")

maintenance_logs = pd.read_csv('data/fact_maintenance_logs.csv')
event_types = maintenance_logs['event_type'].unique()
dim_maint_type = pd.DataFrame({'event_type': event_types})
dim_maint_type.to_sql('dim_maintenance_type', engine, if_exists='append', index=False)
print(f"dim_maintenance_type loaded: {len(dim_maint_type)} rows")

# Cell 4: Aggregate and load fact tables
readings = pd.read_csv('data/fact_sensor_readings_clean.csv', parse_dates=['timestamp'])
readings['date'] = readings['timestamp'].dt.date

fact_sensor_daily = readings.groupby(['date', 'sensor_id']).agg(
    avg_pressure=('pressure', 'mean'),
    min_pressure=('pressure', 'min'),
    max_pressure=('pressure', 'max'),
    avg_flow=('flow', 'mean'),
    avg_temperature=('temperature', 'mean'),
    reading_count=('pressure', 'count'),
    avg_pressure_drop=('pressure_drop', 'mean'),
    night_flow_ratio=('night_flow_ratio', 'mean')
).reset_index()

date_key_map = dim_date.set_index('date')['date_key'].to_dict()
fact_sensor_daily['date_key'] = fact_sensor_daily['date'].map(
    lambda d: date_key_map.get(pd.Timestamp(d))
)

sensor_lookup = dim_sensors.set_index('sensor_id')[['zone_id', 'pipe_segment_id']]
fact_sensor_daily = fact_sensor_daily.merge(
    sensor_lookup, left_on='sensor_id', right_index=True
)
fact_sensor_daily.rename(columns={'pipe_segment_id': 'segment_id'}, inplace=True)

fact_sensor_daily.dropna(subset=['date_key'], inplace=True)
fact_sensor_daily['date_key'] = fact_sensor_daily['date_key'].astype(int)

fact_sensor_daily_load = fact_sensor_daily[[
    'date_key', 'sensor_id', 'zone_id', 'segment_id',
    'avg_pressure', 'min_pressure', 'max_pressure',
    'avg_flow', 'avg_temperature', 'reading_count',
    'avg_pressure_drop', 'night_flow_ratio'
]]
fact_sensor_daily_load.to_sql('fact_sensor_daily', engine, if_exists='append', index=False)
print(f"fact_sensor_daily loaded: {len(fact_sensor_daily_load):,} rows")

risk_scores = pd.read_csv('data/segment_risk_scores.csv')
risk_scores['date_key'] = int(dim_date['date_key'].max())
segment_zone = dim_segments.set_index('segment_id')['zone_id']
risk_scores['zone_id'] = risk_scores['segment_id'].map(segment_zone)

fact_leak_load = risk_scores[['segment_id', 'date_key', 'zone_id', 'cluster_label', 'risk_score']]
fact_leak_load.to_sql('fact_leak_detection', engine, if_exists='append', index=False)
print(f"fact_leak_detection loaded: {len(fact_leak_load)} rows")

forecast = pd.read_csv('data/demand_forecast.csv', parse_dates=['ds'])
forecast['date_key'] = forecast['ds'].map(
    lambda d: date_key_map.get(pd.Timestamp(d.date()) if hasattr(d, 'date') else pd.Timestamp(d))
)
forecast.dropna(subset=['date_key'], inplace=True)
forecast['date_key'] = forecast['date_key'].astype(int)

fact_forecast_load = forecast[['date_key', 'zone_id', 'yhat', 'yhat_lower', 'yhat_upper']]
fact_forecast_load.to_sql('fact_demand_forecast', engine, if_exists='append', index=False)
print(f"fact_demand_forecast loaded: {len(fact_forecast_load):,} rows")

# Cell 5: Verify and sanity check
with engine.connect() as conn:
    tables = [
        'dim_date', 'dim_zone', 'dim_pipe_segment', 'dim_sensor',
        'dim_maintenance_type', 'fact_sensor_daily',
        'fact_leak_detection', 'fact_demand_forecast'
    ]
    print("Row Counts:")
    print("-" * 40)
    for table in tables:
        count = conn.execute(text(f"SELECT COUNT(*) FROM {table}")).scalar()
        print(f"  {table}: {count:,} rows")

sanity_query = """
SELECT
    dz.zone_name,
    COUNT(fld.id) AS leak_count,
    ROUND(AVG(fld.risk_score), 2) AS avg_risk_score
FROM fact_leak_detection fld
JOIN dim_zone dz ON fld.zone_id = dz.zone_id
WHERE fld.cluster_label IN ('Suspected Leak', 'Critical Leak')
GROUP BY dz.zone_name
ORDER BY avg_risk_score DESC
LIMIT 5;
"""

result = pd.read_sql(sanity_query, engine)
print("\nTop 5 Zones by Leak Risk:")
print(result.to_string(index=False))

Your star schema is live and loaded with clean, analysis-ready data. Next up, you will build production-grade SQL views on top of this schema using advanced window functions, CTEs, and recursive queries that transform these raw tables into actionable intelligence.

Build Production SQL Analytical Views

Your star schema is loaded and your data is clean. Now you need to transform that raw data into answers that an operations manager can act on.

In this step, you will build three production-grade PostgreSQL views that surface leak risk rankings, demand trend alerts, and pipe network propagation paths. Each view showcases a different advanced SQL technique: window functions, nested CTEs, and recursive CTEs.

In this step, get ready to:

• Build a leak-risk ranking view using window functions.
• Build a demand trend analysis view using nested CTEs.
• Build a pipe network traversal view using a recursive CTE.

Build the leak-risk ranking view with window functions

Operations managers need to see which pipe segments are most at risk within each pressure zone. A flat list of risk scores is not enough. They need within-zone rankings, pressure trend comparisons, and an overall zone risk total to prioritize crews.

• Create a new file called vw_leak_risk_ranking.sql inside your sql/ directory.
• Paste the following SQL into the file:

CREATE OR REPLACE VIEW vw_leak_risk_ranking AS
SELECT
    fl.segment_id,
    fl.zone_id,
    fl.risk_score,
    -- Rank segments within each zone by descending risk
    ROW_NUMBER() OVER (
        PARTITION BY fl.zone_id
        ORDER BY fl.risk_score DESC
    ) AS zone_rank,
    -- Compare current pressure to 7 days ago for trend detection
    LAG(fsd.avg_pressure, 7) OVER (
        PARTITION BY fl.segment_id
        ORDER BY fsd.date_key
    ) AS pressure_7d_ago,
    -- Aggregate total risk across all segments in the zone
    SUM(fl.risk_score) OVER (
        PARTITION BY fl.zone_id
    ) AS zone_risk_total,
    -- Classify segments into priority tiers for crew dispatch
    CASE
        WHEN fl.risk_score >= 80 THEN 'Critical'
        WHEN fl.risk_score >= 60 THEN 'High'
        WHEN fl.risk_score >= 40 THEN 'Medium'
        ELSE 'Low'
    END AS priority_tier
FROM fact_leak_detection fl
JOIN fact_sensor_daily fsd
    ON fl.segment_id = fsd.segment_id
    AND fl.date_key = fsd.date_key;

What does this SQL do?

This view combines leak detection results with daily sensor readings to produce a ranked, prioritized list of at-risk pipe segments.

• ROW_NUMBER() OVER (PARTITION BY zone_id ...) assigns a rank within each zone so managers see their top-priority segments first.
• LAG(avg_pressure, 7) looks back 7 rows (7 days of daily data) per segment to show the pressure reading from a week ago. A large drop between then and now signals active degradation.
• SUM(risk_score) OVER (PARTITION BY zone_id) gives a zone-level risk total so executives can compare zones at a glance.
• The CASE WHEN block maps numeric scores to human-readable priority tiers that dispatch crews understand.

• Save sql/vw_leak_risk_ranking.sql.
• Execute the view against your database by running:

psql -d aquaflow_analytics -f sql/vw_leak_risk_ranking.sql

• Verify the view works by running a quick query:

psql -d aquaflow_analytics -c "SELECT * FROM vw_leak_risk_ranking LIMIT 5;"

You should see a table with columns segment_id, zone_id, risk_score, zone_rank, pressure_7d_ago, zone_risk_total, and priority_tier.

Getting a column not found error?

• Make sure fact_leak_detection and fact_sensor_daily both have a segment_id column. If your schema uses pipe_segment_id instead, update the join condition to match.
• Confirm the date_key column exists in both tables. The join requires matching dates.
• Check that you connected to the aquaflow_analytics database, not the default postgres database.

Help me debug my vw_leak_risk_ranking view. I'm getting a column not found error when I try to create it.

Build the demand trend analysis view with nested CTEs

Forecasting demand is only useful if you can compare it to historical trends and flag zones approaching capacity limits. This view layers three CTEs on top of each other to produce a single row per zone per day with rolling averages, year-over-year growth, and alert levels.

• Create a new file called vw_demand_trends.sql inside your sql/ directory.
• Paste the following SQL into the file:

CREATE OR REPLACE VIEW vw_demand_trends AS
WITH rolling_avg AS (
    -- CTE 1: Calculate 30-day rolling average demand per zone
    SELECT
        zone_id,
        date_key AS date,
        yhat AS daily_demand,
        AVG(yhat) OVER (
            PARTITION BY zone_id
            ORDER BY date_key
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS rolling_30d_avg
    FROM fact_demand_forecast
),
yoy_growth AS (
    -- CTE 2: Calculate year-over-year growth percentage
    SELECT
        r.zone_id,
        r.date,
        r.rolling_30d_avg,
        CASE
            WHEN LAG(r.rolling_30d_avg, 365) OVER (
                PARTITION BY r.zone_id ORDER BY r.date
            ) > 0
            THEN (
                (r.rolling_30d_avg - LAG(r.rolling_30d_avg, 365) OVER (
                    PARTITION BY r.zone_id ORDER BY r.date
                )) / LAG(r.rolling_30d_avg, 365) OVER (
                    PARTITION BY r.zone_id ORDER BY r.date
                )
            ) * 100
            ELSE NULL
        END AS yoy_growth_pct
    FROM rolling_avg r
),
capacity_breach AS (
    -- CTE 3: Flag zones where demand exceeds capacity threshold
    SELECT
        yg.zone_id,
        yg.date,
        yg.rolling_30d_avg,
        yg.yoy_growth_pct,
        CASE
            WHEN yg.rolling_30d_avg > dz.population_served * 0.5
            THEN TRUE
            ELSE FALSE
        END AS capacity_breach_flag
    FROM yoy_growth yg
    JOIN dim_zone dz ON yg.zone_id = dz.zone_id
)
-- Final SELECT: apply alert level conditional logic
SELECT
    zone_id,
    date,
    rolling_30d_avg,
    yoy_growth_pct,
    capacity_breach_flag,
    CASE
        WHEN capacity_breach_flag = TRUE AND yoy_growth_pct > 15
            THEN 'Red'
        WHEN capacity_breach_flag = TRUE OR yoy_growth_pct > 10
            THEN 'Amber'
        ELSE 'Green'
    END AS alert_level
FROM capacity_breach;

What does this SQL do?

This view stacks three CTEs to progressively build up a demand intelligence layer:

• CTE 1 (rolling_avg): Smooths daily noise by calculating a 30-day rolling average using a window frame of ROWS BETWEEN 29 PRECEDING AND CURRENT ROW.
• CTE 2 (yoy_growth): Uses LAG(..., 365) to compare today's rolling average with the same day last year. The result is a percentage growth figure.
• CTE 3 (capacity_breach): Joins to dim_zone and compares demand against a capacity threshold (50% of population served) to produce a boolean flag.
• The final SELECT combines the breach flag and growth rate into a traffic-light alert level (Red, Amber, Green) that dashboards can consume directly.

• Save sql/vw_demand_trends.sql.
• Execute the view by running:

psql -d aquaflow_analytics -f sql/vw_demand_trends.sql

• Verify the view returns data by running:

psql -d aquaflow_analytics -c "SELECT * FROM vw_demand_trends WHERE alert_level = 'Red' LIMIT 5;"

You should see rows with columns zone_id, date, rolling_30d_avg, yoy_growth_pct, capacity_breach_flag, and alert_level. If no Red alerts exist, try querying without the WHERE filter.

View creates but returns no rows?

• The LAG(..., 365) window needs at least 365 rows of history per zone. If your forecast table has fewer rows, yoy_growth_pct will be NULL for the first year. This is expected behavior.
• Check that fact_demand_forecast has a zone_id column matching dim_zone.zone_id.
• If capacity_breach_flag is always FALSE, the threshold multiplier (0.5) may need adjusting for your simulated data.

Help me debug my vw_demand_trends view that returns no rows or only NULL growth values.

Build the pipe network traversal view with a recursive CTE

When a pipe segment has a critical leak, water pressure drops propagate downstream through the network. A recursive CTE can traverse the pipe graph starting from any critical segment and follow downstream connections to show every segment affected.

• Create a new file called vw_leak_propagation_path.sql inside your sql/ directory.
• Paste the following SQL into the file:

CREATE OR REPLACE VIEW vw_leak_propagation_path AS
WITH RECURSIVE propagation AS (
    -- Base case: start from segments flagged as Critical Leak
    SELECT
        ps.segment_id AS start_segment_id,
        ps.segment_id AS current_segment_id,
        0 AS depth,
        0::NUMERIC AS cumulative_distance_m
    FROM dim_pipe_segment ps
    JOIN fact_leak_detection fl
        ON ps.segment_id = fl.segment_id
    WHERE fl.cluster_label = 'Critical Leak'

    UNION ALL

    -- Recursive step: traverse downstream connections
    SELECT
        p.start_segment_id,
        ds.segment_id AS current_segment_id,
        p.depth + 1,
        p.cumulative_distance_m + ds.length_m
    FROM propagation p
    JOIN dim_pipe_segment ds
        ON ds.upstream_segment_id = p.current_segment_id
    WHERE p.depth < 20  -- Depth limiter prevents infinite recursion
)
SELECT
    start_segment_id,
    current_segment_id,
    depth,
    cumulative_distance_m,
    -- Count total affected segments per starting leak point
    COUNT(*) OVER (PARTITION BY start_segment_id) AS affected_segment_count
FROM propagation
WHERE depth > 0;

What does this SQL do?

This recursive CTE models your pipe network as a directed graph and walks it from each critical leak point downstream.

• The base case selects every segment where cluster_label = 'Critical Leak' and initializes depth at 0 with zero distance.
• The recursive step joins dim_pipe_segment on upstream_segment_id = current_segment_id to find the next downstream segment. It increments depth and adds length_m to the running distance.
• The WHERE p.depth < 20 clause prevents infinite loops if your pipe graph has circular references.
• The final SELECT filters out depth 0 (the leak source itself) and uses a window function to count total affected downstream segments per leak point.

Why a depth limiter?

Recursive CTEs keep iterating until the recursive step returns no new rows. If your pipe network has a cycle (segment A feeds B feeds C feeds A), the query would loop forever. The WHERE p.depth < 20 condition guarantees termination after 20 hops, which is more than enough to model real-world propagation effects.

• Save sql/vw_leak_propagation_path.sql.
• Execute the view by running:

psql -d aquaflow_analytics -f sql/vw_leak_propagation_path.sql

• Verify the view returns propagation paths by running:

psql -d aquaflow_analytics -c "SELECT * FROM vw_leak_propagation_path ORDER BY start_segment_id, depth LIMIT 10;"

You should see rows showing start_segment_id, current_segment_id, increasing depth values, growing cumulative_distance_m, and the total affected_segment_count per leak source.

View returns zero rows?

• Check that fact_leak_detection contains rows where cluster_label is exactly 'Critical Leak' (case-sensitive). Run SELECT DISTINCT cluster_label FROM fact_leak_detection; to confirm.
• Confirm that dim_pipe_segment has populated upstream_segment_id values. If all are NULL, the recursive step has nothing to traverse.
• Verify that the segment IDs in fact_leak_detection match those in dim_pipe_segment. A foreign key mismatch means the base case JOIN finds nothing.

Help me debug my recursive vw_leak_propagation_path view that returns no rows.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files in your sql/ directory and that each view is queryable.

ⓧ I'd like to double check the full code

Here are all three view files for comparison.

-- sql/vw_leak_risk_ranking.sql
CREATE OR REPLACE VIEW vw_leak_risk_ranking AS
SELECT
    fl.segment_id,
    fl.zone_id,
    fl.risk_score,
    ROW_NUMBER() OVER (
        PARTITION BY fl.zone_id
        ORDER BY fl.risk_score DESC
    ) AS zone_rank,
    LAG(fsd.avg_pressure, 7) OVER (
        PARTITION BY fl.segment_id
        ORDER BY fsd.date_key
    ) AS pressure_7d_ago,
    SUM(fl.risk_score) OVER (
        PARTITION BY fl.zone_id
    ) AS zone_risk_total,
    CASE
        WHEN fl.risk_score >= 80 THEN 'Critical'
        WHEN fl.risk_score >= 60 THEN 'High'
        WHEN fl.risk_score >= 40 THEN 'Medium'
        ELSE 'Low'
    END AS priority_tier
FROM fact_leak_detection fl
JOIN fact_sensor_daily fsd
    ON fl.segment_id = fsd.segment_id
    AND fl.date_key = fsd.date_key;

-- sql/vw_demand_trends.sql
CREATE OR REPLACE VIEW vw_demand_trends AS
WITH rolling_avg AS (
    SELECT
        zone_id,
        date_key AS date,
        yhat AS daily_demand,
        AVG(yhat) OVER (
            PARTITION BY zone_id
            ORDER BY date_key
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS rolling_30d_avg
    FROM fact_demand_forecast
),
yoy_growth AS (
    SELECT
        r.zone_id,
        r.date,
        r.rolling_30d_avg,
        CASE
            WHEN LAG(r.rolling_30d_avg, 365) OVER (
                PARTITION BY r.zone_id ORDER BY r.date
            ) > 0
            THEN (
                (r.rolling_30d_avg - LAG(r.rolling_30d_avg, 365) OVER (
                    PARTITION BY r.zone_id ORDER BY r.date
                )) / LAG(r.rolling_30d_avg, 365) OVER (
                    PARTITION BY r.zone_id ORDER BY r.date
                )
            ) * 100
            ELSE NULL
        END AS yoy_growth_pct
    FROM rolling_avg r
),
capacity_breach AS (
    SELECT
        yg.zone_id,
        yg.date,
        yg.rolling_30d_avg,
        yg.yoy_growth_pct,
        CASE
            WHEN yg.rolling_30d_avg > dz.population_served * 0.5
            THEN TRUE
            ELSE FALSE
        END AS capacity_breach_flag
    FROM yoy_growth yg
    JOIN dim_zone dz ON yg.zone_id = dz.zone_id
)
SELECT
    zone_id,
    date,
    rolling_30d_avg,
    yoy_growth_pct,
    capacity_breach_flag,
    CASE
        WHEN capacity_breach_flag = TRUE AND yoy_growth_pct > 15
            THEN 'Red'
        WHEN capacity_breach_flag = TRUE OR yoy_growth_pct > 10
            THEN 'Amber'
        ELSE 'Green'
    END AS alert_level
FROM capacity_breach;

-- sql/vw_leak_propagation_path.sql
CREATE OR REPLACE VIEW vw_leak_propagation_path AS
WITH RECURSIVE propagation AS (
    SELECT
        ps.segment_id AS start_segment_id,
        ps.segment_id AS current_segment_id,
        0 AS depth,
        0::NUMERIC AS cumulative_distance_m
    FROM dim_pipe_segment ps
    JOIN fact_leak_detection fl
        ON ps.segment_id = fl.segment_id
    WHERE fl.cluster_label = 'Critical Leak'

    UNION ALL

    SELECT
        p.start_segment_id,
        ds.segment_id AS current_segment_id,
        p.depth + 1,
        p.cumulative_distance_m + ds.length_m
    FROM propagation p
    JOIN dim_pipe_segment ds
        ON ds.upstream_segment_id = p.current_segment_id
    WHERE p.depth < 20
)
SELECT
    start_segment_id,
    current_segment_id,
    depth,
    cumulative_distance_m,
    COUNT(*) OVER (PARTITION BY start_segment_id) AS affected_segment_count
FROM propagation
WHERE depth > 0;

Your analytical views are ready. Each one turns raw fact table data into decision-ready intelligence without moving or duplicating data. Next up, you will connect these views to Power BI and build advanced DAX measures on top of them.

Build the Power BI Data Model & DAX Measures

Your SQL analytical views are live and queryable. The raw sensor chaos has been transformed into ranked leak-risk tables, demand trend summaries, and graph traversals.

Now you need to bring all of that intelligence into an interactive visual layer. In this step, you'll connect Power BI Desktop to your PostgreSQL database, configure a proper star schema, and write DAX measures that power executive-level KPIs and a What-If investment simulator.

In this step, get ready to:

• Import all tables from PostgreSQL and configure star schema relationships.
• Write DAX measures for leak detection, demand forecasting, and time intelligence.
• Create a What-If parameter for capital investment simulation.

Import data and configure the star schema in Power BI

Power BI needs a direct connection to your PostgreSQL database so it can pull in all eight tables. You'll configure relationships that mirror the star schema you designed in SQL.

• Press the Windows key to open your search bar.
• Type Power BI Desktop and press Enter to open it.
• If this is your first launch, close any welcome dialogs or sign-in prompts.

• Connect to your PostgreSQL database by clicking Get Data on the Home ribbon.
• In the Get Data window, select Database in the left pane.
• Select PostgreSQL database and click Connect.
• In the Server field, enter localhost.
• In the Database field, enter aquaflow_analytics.
• Leave the data connectivity mode as Import and click OK.
• When prompted for credentials, select Database on the left, then enter the PostgreSQL username and password you set up in Step 1.
• Click Connect.

What if I see an encryption prompt?

If Power BI shows a message about an unencrypted connection, click OK to proceed. Your PostgreSQL instance is running locally, so encryption is not required for this project.

• In the Navigator window that appears, expand the public schema.
• Select all 8 tables: dim_date, dim_sensor, dim_pipe_segment, dim_zone, dim_maintenance_type, fact_sensor_daily, fact_leak_detection, and fact_demand_forecast.
• Click Load to import all tables into the model.

Loading may take a minute depending on your machine. Once complete, all 8 tables appear in the Fields pane on the right.

• Switch to Model view by clicking the third icon on the left sidebar (the one showing three connected boxes).
• Power BI may auto-detect some relationships. Delete any auto-detected relationships by right-clicking them and selecting Delete.
• Create new relationships by dragging foreign key columns from fact tables to the primary key columns on dimension tables.

What relationships should I create?

Each relationship should be many-to-one, single cross-filter direction, from the fact table to the dimension table.

• fact_sensor_daily[date_key] to dim_date[date_key].
• fact_sensor_daily[sensor_id] to dim_sensor[sensor_id].
• fact_sensor_daily[zone_id] to dim_zone[zone_id].
• fact_leak_detection[segment_id] to dim_pipe_segment[segment_id].
• fact_leak_detection[zone_id] to dim_zone[zone_id].
• fact_demand_forecast[zone_id] to dim_zone[zone_id].
• fact_demand_forecast[date_key] to dim_date[date_key].

For each relationship, double-click the line and confirm it shows Many to one (*:1) with Single cross-filter direction.

• Mark dim_date as the Date Table for time intelligence. In the Fields pane, right-click dim_date and select Mark as date table then Mark as date table again in the submenu.
• In the dialog that appears, select your date column (e.g. Date) from the dropdown as the unique identifier.
• Click OK.

Why mark a Date Table?

DAX time intelligence functions like SAMEPERIODLASTYEAR and TOTALYTD require a designated Date Table with a continuous date range. Without this marking, those functions return errors or incorrect results.

• Hide foreign key columns from report view. In Model view, select each foreign key column (e.g. date_key, zone_id, sensor_id, segment_id on fact tables), right-click, and select Hide in report view.
• Save your file as aquaflow_analysis.pbix in your /powerbi directory by pressing Ctrl+S.

You should now see all 8 tables in Model view with relationship lines connecting fact tables to their dimensions.

Don't see any tables in Navigator?

• Confirm PostgreSQL is running. Open a terminal and run psql -d aquaflow_analytics -c "SELECT 1;" to verify the database is reachable.
• Double-check you entered localhost for Server and aquaflow_analytics for Database.
• Verify your credentials match the PostgreSQL user you created in Step 1.

Help me troubleshoot my Power BI PostgreSQL connection.

Write advanced DAX measures

With the data model configured, you'll now create DAX measures that calculate the KPIs your dashboard needs. These range from simple aggregations to time-intelligence calculations that compare performance across periods.

• Switch to Report view by clicking the first icon on the left sidebar.
• Create the first base measure. On the Modeling tab in the ribbon, click New measure.
• In the formula bar that appears at the top, type the following DAX formula and press Enter:

Total Leaks Detected = COUNTROWS(FILTER(fact_leak_detection, fact_leak_detection[cluster_label] IN {"Suspected Leak", "Critical Leak"}))

• Create a second measure by clicking New measure again on the Modeling tab:

Avg Pressure Drop = AVERAGE(fact_sensor_daily[avg_pressure_drop])

• Create the third base measure:

Non-Revenue Water % = DIVIDE([Total Water Lost], [Total Water Produced], 0)

What do these base measures calculate?

• Total Leaks Detected counts rows in the leak detection fact table where the cluster label is either "Suspected Leak" or "Critical Leak". FILTER returns only those rows, and COUNTROWS tallies them.
• Avg Pressure Drop gives the mean pressure drop across all daily sensor readings in the current filter context.
• Non-Revenue Water % uses DIVIDE with a zero fallback to avoid division-by-zero errors. It calculates the fraction of water lost vs. produced.

• Create the first time intelligence measure by clicking New measure:

Leaks YoY Change = CALCULATE([Total Leaks Detected], SAMEPERIODLASTYEAR(dim_date[Date])) - [Total Leaks Detected]

• Create the next time intelligence measure:

Demand YTD = TOTALYTD([Total Demand], dim_date[Date])

• Create the rolling average measure:

Rolling 30-Day Avg Pressure = CALCULATE(AVERAGE(fact_sensor_daily[avg_pressure]), DATESINPERIOD(dim_date[Date], MAX(dim_date[Date]), -30, DAY))

• Create the dynamic filtering measure:

Critical Segments in Zone = CALCULATE(COUNTROWS(fact_leak_detection), FILTER(fact_leak_detection, fact_leak_detection[risk_score] >= 80))

What do the time intelligence and dynamic measures do?

• Leaks YoY Change uses CALCULATE with SAMEPERIODLASTYEAR to shift the filter context back one year, then subtracts the current period's leaks to show the year-over-year difference.
• Demand YTD wraps TOTALYTD to accumulate demand from the start of the year through the currently selected date.
• Rolling 30-Day Avg Pressure uses DATESINPERIOD to create a sliding 30-day window ending at the latest date in the current context.
• Critical Segments in Zone uses CALCULATE with FILTER to count only segments whose risk score is 80 or above. This responds dynamically to zone slicer selections.

• Verify your measures appear in the Fields pane. Each measure shows a calculator icon next to its name.
• Save the file with Ctrl+S.

Seeing a time intelligence error?

• Confirm you marked dim_date as the Date Table in the previous substep. Time intelligence functions like SAMEPERIODLASTYEAR and TOTALYTD require this designation.
• Ensure your dim_date table has no missing dates and the date column contains unique values spanning full years.
• Check that the column you selected as the date identifier has a Date data type, not Text.

Help me fix my DAX time intelligence errors in Power BI.

Create a What-If parameter for investment simulation

Operations managers need to answer one critical question: "If we invest $X million in pipe replacement, how much Non-Revenue Water can we eliminate?" A What-If parameter creates a slider that lets stakeholders adjust an investment amount and see projected outcomes in real time.

• On the Modeling tab, click New parameter.
• Select Numeric range as the parameter type.
• Configure the parameter with these values:

• Name: Investment Amount ($M).
• Data type: Whole number.
• Minimum: 1.
• Maximum: 20.
• Increment: 1.

• Check the Add slicer to this page checkbox.
• Click Create.

Power BI creates a disconnected parameter table and a slicer on your report page. The slicer lets users slide between $1M and $20M.

What is a What-If parameter?

A What-If parameter creates a single-column calculated table and a measure returning the currently selected value. The table is disconnected from your star schema, which means it doesn't filter fact tables directly. Instead, you reference the parameter's value measure inside other DAX calculations to model scenarios.

• Create the Segments Fixable measure. Click New measure on the Modeling tab and enter:

Segments Fixable = DIVIDE([Investment Amount Value], [Avg Repair Cost Per Segment])

• Create the Projected NRW Reduction measure:

Projected NRW Reduction = [Segments Fixable] * [Avg Leak Volume Per Segment] / [Total Water Produced]

How does the investment simulation work?

When a user moves the investment slider to $10M, the [Investment Amount Value] measure returns 10. Segments Fixable divides that by the average cost to repair one pipe segment. Projected NRW Reduction then estimates how much of the Non-Revenue Water would be eliminated by fixing that many segments.

This lets stakeholders answer: "If we invest $X, NRW drops by Y%."

• Save the file with Ctrl+S.
• Verify the slicer appears on the report page and that dragging it updates any card or visual referencing Segments Fixable or Projected NRW Reduction.

Parameter slicer not affecting visuals?

• The What-If parameter table is disconnected by design. It does not directly filter fact tables. Its value flows through the DAX measures you wrote (Segments Fixable, Projected NRW Reduction).
• Ensure you referenced [Investment Amount Value] (the auto-generated measure name with "Value" suffix) inside your formulas, not the table name.
• Add a Card visual to the canvas and assign Projected NRW Reduction to it. Moving the slider should update the card.

Help me debug my What-If parameter in Power BI.

Your data model is fully configured with a proper star schema, seven analytical DAX measures, and an interactive What-If simulator. Next up, you'll design the enterprise dashboard pages that bring all of this to life visually.

Design the Enterprise Dashboard

Your data model is complete with all the measures and relationships an operations manager needs to make investment decisions. Now it is time to bring those numbers to life.

In this step, you will build three dashboard pages that tell a complete story: an executive summary for leadership, a leak detection deep-dive for field teams, and a capacity planning view for long-term forecasting.

In this step, get ready to:

• Build Page 1 with KPI cards, a dual-axis line chart, a donut chart, and a treemap.
• Build Page 2 with a scatter plot, a risk-ranked table with conditional formatting, slicers, and a bar chart.
• Build Page 3 with a forecast area chart, year-over-year bars, a capacity breach card, and seasonal decomposition.

Build Page 1: Executive Summary

The Executive Summary gives leadership a single-glance health check of the entire water network. Every visual on this page answers one question: how bad is the leak problem, and where are we headed?

• In Power BI Desktop, right-click the current page tab at the bottom of the canvas.
• Select Rename Page and type Executive Summary.

• Click an empty area of the canvas, then select the Card visual from the Visualizations pane.
• Drag Total Leaks Detected from the Fields pane into the card's Fields well.
• Resize the card to approximately one quarter of the page width and position it in the top-left corner.
• Create a second card and drag Non-Revenue Water % into it.
• Create a third card with a manual text label showing your Prophet MAPE value (Demand Forecast Accuracy).
• Create a fourth card and drag Critical Segments in Zone into it.
• Arrange all four cards in a row across the top of the page.

What are these KPI cards showing?

Each card displays a single headline number. Together they answer four executive questions: how many leaks exist, what percentage of water is lost, how accurate the demand forecast is, and how many pipe segments need immediate attention.

• Click an empty area below the KPI cards and select the Line chart visual.
• Drag dim_date[Date] to the X-axis well.
• Drag Demand YTD to the Y-axis well (actual demand).
• Drag fact_demand_forecast[yhat] to the Secondary Y-axis well (forecast line).
• In the Format visual pane, expand Y-axis and toggle Secondary axis to On.
• Set the chart title to Monthly Demand: Actual vs Forecast.

• Select the Donut chart visual from the Visualizations pane.
• Drag fact_leak_detection[cluster_label] to the Legend well.
• Drag Total Leaks Detected to the Values well.
• Set the title to Leak Severity Distribution.
• Position the donut chart to the right of the line chart.

• Select the Treemap visual.
• Drag dim_zone[zone_name] to the Category well.
• Drag fact_leak_detection[risk_score] (aggregated as Average) to the Values well.
• Set the title to Zone Risk Overview.
• Position the treemap in the bottom-right area of the canvas.

• Press Ctrl+S to save your report.

Your Executive Summary page should now show four KPI cards at the top, a dual-axis line chart on the left, a donut chart beside it, and a treemap in the lower section.

Cards showing blank or zero?

Make sure the measure you dragged into the card's field well matches the exact measure name from the previous step. If a card shows blank, click it, check the Fields well, and confirm it references your DAX measure and not a raw column.

Help me troubleshoot blank card visuals

Build Page 2: Leak Detection Deep Dive

The deep-dive page gives field engineers granular visibility into which pipe segments are failing and why. This is where the K-Means cluster results become actionable.

• Click the + icon next to your page tabs at the bottom of the canvas to add a new page.
• Right-click the new tab and rename it to Leak Detection Deep Dive.

• Select the Scatter chart visual from the Visualizations pane.
• Drag fact_leak_detection[avg_pressure_drop] to the X-axis well.
• Drag fact_leak_detection[night_flow_ratio] to the Y-axis well.
• Drag fact_leak_detection[cluster_label] to the Legend well to color each point by its cluster.
• Drag fact_leak_detection[segment_id] to the Details well so each dot represents one pipe segment.
• Set the title to Pressure Drop vs Night Flow Ratio.

Why these two axes?

High pressure drops combined with high night flow ratios are the signature of an active leak. Night flow should be minimal in a healthy pipe because residential usage is near zero between 2AM and 4AM.

Segments that cluster in the top-right corner of this scatter plot are your highest-priority repair targets.

• Select the Table visual.
• Drag these fields into the Columns well: dim_pipe_segment[segment_id], dim_zone[zone_name], dim_pipe_segment[material], dim_pipe_segment[installation_year], fact_leak_detection[risk_score].
• Click the drop-down arrow on risk_score in the Columns well and select Sort descending.
• In the Filters pane for this visual, set a Top N filter on risk_score to show only the top 20 rows.

Now apply conditional formatting to make risk scores visually scannable.

• With the table selected, right-click risk_score in the Columns well of the Visualizations pane.
• Select Conditional formatting, then Background color.
• In the dialog, set Format style to Rules.
• Add Rule 1: if value is greater than or equal to 80, set the background color to red (critical risk).
• Add Rule 2: if value is greater than or equal to 50 and less than 80, set the background color to yellow (moderate risk).
• Add Rule 3: if value is less than 50, set the background color to green (low risk).
• Click OK to apply the formatting.

How does rules-based conditional formatting work?

Power BI evaluates each cell value against your rules in order. When a value matches a rule's condition, that rule's color is applied to the cell background.

This gives you an instant visual heatmap: red cells need immediate attention, yellow cells are degrading, and green cells are stable.

• Select the Clustered bar chart visual.
• Drag dim_pipe_segment[material] to the Y-axis well.
• Drag Total Leaks Detected to the X-axis well.
• Set the title to Leaks by Pipe Material.

• Select the Slicer visual.
• Drag dim_zone[zone_name] into the slicer's Field well to create the zone selector.
• Add a second Slicer visual and drag dim_date[Date] into it. Power BI automatically renders this as a date range slider.
• Position both slicers at the top of the page so they filter all visuals below.
• Press Ctrl+S to save.

Click different zones in the slicer. All four visuals on the page should update to show only that zone's data.

Slicer not filtering a visual?

Make sure the slicer's field is connected through a relationship in your data model. Click Model view in the left sidebar and confirm dim_zone has a relationship line to your fact tables. If a line is missing, drag zone_id from one table to the other to create it.

Help me fix my slicer filtering

Build Page 3: Demand Forecasting and Capacity Planning

The forecasting page helps planners see where demand is headed and which zones will breach capacity first. This is where your Prophet model outputs become decision-ready visuals.

• Add a new page by clicking the + icon next to your page tabs.
• Rename it to Demand Forecasting.

• Select the Area chart visual.
• Drag fact_demand_forecast[ds] to the X-axis well.
• Drag fact_demand_forecast[yhat] to the Y-axis well.
• Drag fact_demand_forecast[yhat_upper] and fact_demand_forecast[yhat_lower] to the Y-axis well as additional lines.
• Set the title to Demand Forecast with Confidence Interval.

What is the confidence ribbon?

The area between yhat_upper and yhat_lower represents Prophet's 80% confidence interval. If actual demand moves outside this band, it signals an unexpected event like a pipe burst or extreme weather.

• Select the Clustered bar chart visual.
• Drag dim_zone[zone_name] to the Y-axis well.
• Drag Demand YTD to the X-axis well.
• Set the title to Zone-Level Demand Growth YoY.

Now add the capacity breach card for the most at-risk zone.

• Select the Card visual.
• Create a new measure in the fact_demand_forecast table by clicking New measure in the Modeling tab.
• Enter the following DAX formula in the formula bar:

Days Until Capacity Breach = 
    -- Calculate remaining days before the most at-risk zone exceeds capacity
    DATEDIFF(
        TODAY(),
        MIN(fact_demand_forecast[ds]),
        DAY
    )

What does this measure do?

This is a simplified measure that calculates the difference between today and the earliest forecast date in the filtered context. In a production scenario, you would compare each zone's forecast to its capacity threshold and return the first date where yhat exceeds the limit.

For this project, it demonstrates how a card can surface a single critical number for decision-makers.

• Drag the Days Until Capacity Breach measure into the card's Fields well.
• Position the card prominently in the top-right area of the page.

• Select the Line chart visual.
• Drag fact_demand_forecast[ds] to the X-axis well.
• Drag fact_demand_forecast[trend], fact_demand_forecast[weekly], and fact_demand_forecast[yearly] to the Y-axis well as separate lines.
• Set the title to Seasonal Decomposition.
• Press Ctrl+S to save your report.

Navigate back to Page 1 and confirm the What-If parameter slider (Investment Amount ($M)) is visible. Drag the slider left and right.

The Segments Fixable and Projected NRW Reduction measures should update dynamically as you move the slider.

Pin the slicer to all pages

If you want the What-If slider accessible on every page, right-click it and select Pin to all pages. This lets executives simulate investment scenarios regardless of which page they are viewing.

What-If slider not updating measures?

The What-If parameter creates a disconnected table. Your Segments Fixable and Projected NRW Reduction measures must reference the parameter's value measure (created automatically when you set up the What-If parameter). Check that your DAX formulas use the parameter's generated measure name.

Help me debug my What-If parameter

Your enterprise dashboard is complete. You now have three interconnected pages that take raw IoT sensor data all the way through to actionable intelligence for operations managers, field engineers, and executive leadership.

Secret mission

RLS & Recursive Leak Propagation

Ready to take your analytics platform into enterprise security and graph analytics territory? In this secret mission, you will implement Row-Level Security so each zone manager sees only their zone's data. You will also enhance the recursive CTE to model how leaks propagate through the pipe network, and build a composite Network Health Score that weighs multiple risk factors into a single actionable metric.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. Decide whether to keep your resources for portfolio demos or delete them to free up disk space.

Resources you used:

• PostgreSQL database aquaflow_analytics (local, ~50MB).
• Python virtual environment with packages (pandas, scikit-learn, prophet, etc.).
• Project folder with /data, /notebooks, /sql, and /powerbi directories.
• Power BI file aquaflow_analysis.pbix (inside the /powerbi directory).

Keep everything running

No action needed. Choose this if you plan to use this project for portfolio demos or want to continue exploring the data.

• The PostgreSQL database takes minimal disk space (~50MB) and does not run unless the PostgreSQL service is active.
• Your Power BI file and Python virtual environment remain on disk and incur no costs.
• If you want to stop the PostgreSQL service from running in the background, you can stop it from Services (Windows) without deleting data.

Pause - I'll come back to this later

Since all resources are local, there is nothing to pause for cost savings. The only action is stopping the PostgreSQL service to free up memory.

• Press the Windows key, type Services, and press Enter.
• Find postgresql-x64-17 in the list (or the version you installed).
• Right-click it and select Stop.
• When you return, right-click the same service and select Start to resume.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below to fully clean up.

Drop the PostgreSQL database

• Open pgAdmin or a psql terminal connected to the default postgres database (not aquaflow_analytics).
• Run the following command to drop the project database:

DROP DATABASE IF EXISTS aquaflow_analytics;

Delete the project folder and virtual environment

Open a terminal and navigate to where your project folder lives (the Desktop, if you followed the setup steps).

macOS

• Delete the project folder by running:

cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your project folder name"]]

• If your virtual environment is inside the project folder, this command removes it too.
• If you created the virtual environment elsewhere, navigate to that location and run the same command on the environment folder.

Windows

• Delete the project folder by running:

cd ~\Desktop
Remove-Item -Recurse -Force [[PROJECT_FOLDER="your project folder name"]]

• If your virtual environment is inside the project folder, this command removes it too.
• If you created the virtual environment elsewhere, navigate to that location and run the same command on the environment folder.

Confirm cleanup

• In psql, run \l to list databases. Confirm aquaflow_analytics no longer appears.
• In your file explorer, confirm the project folder is gone from your Desktop.

Nice Work!

Nice Work!

You just built a complete end-to-end analytics platform for a water utility losing 28% of its treated water to leaks. This is a portfolio-ready project that demonstrates Senior Data Analyst skills across Python, SQL, and Power BI.

You've learned how to:

• Generate and clean a realistic multi-table IoT dataset, then apply K-Means clustering to detect pressure anomalies and Prophet to forecast seasonal water demand.
• Design a star schema in PostgreSQL and write production-grade analytical views using advanced window functions, nested CTEs, and recursive CTEs for graph traversal.
• Build an enterprise Power BI dashboard with advanced DAX measures using CALCULATE, time intelligence functions, and a What-If parameter that lets stakeholders simulate capital investment scenarios.
• Secret Mission: Implemented Row-Level Security so each zone manager sees only their own data, and enhanced a recursive CTE to model leak propagation paths with diminishing water loss across the pipe network.

Ready to quiz yourself?