Data Analytics in Oil & Gas: Production Monitoring, Pipeline Integrity, and Upstream Decision Intelligence

Introduction

Nigeria produces approximately 1.2 million barrels of oil per day across onshore fields in the Niger Delta, shallow offshore platforms, and deepwater installations in the Gulf of Guinea. The upstream sector — operated by NNPC and its joint venture partners including Shell Petroleum Development Company (SPDC), TotalEnergies EP Nigeria, Chevron Nigeria Limited, and indigenous operators like Seplat, Oando, and First E&P — generates an extraordinary volume of operational data every second.

A single offshore platform can produce over 2 million sensor readings per hour from wellhead pressure transmitters, flowmeters, temperature probes, vibration sensors, and corrosion monitoring devices. Across the Niger Delta, pipeline networks spanning over 7,000 km are instrumented with SCADA systems that continuously report flow rates, pressure differentials, and leak detection signals.

Yet the industry faces a paradox: operators are data-rich but insight-poor. Legacy historians and siloed SCADA systems capture the data but fail to deliver the real-time analytics and predictive insights that modern upstream operations demand. Production engineers spend hours manually correlating well test data with reservoir models. Pipeline integrity teams rely on scheduled pigging runs rather than continuous condition monitoring. Regulatory reporting to the Nigerian Upstream Petroleum Regulatory Commission (NUPRC, formerly DPR) involves weeks of manual data gathering from spreadsheets and disparate systems.

This guide provides a comprehensive technical framework for building a modern data platform purpose-built for oil and gas operations.

The Oil & Gas Data Challenge

Data Sources and Volume

Upstream oil and gas operations generate data from fundamentally different source systems:

┌─────────────────────────────────────────────────────────────────────────┐
│                OIL & GAS DATA SOURCE TAXONOMY                           │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  REAL-TIME SENSOR DATA              OPERATIONAL DATA                    │
│  ├─ Wellhead sensors (1-10s)        ├─ Well test reports (daily)       │
│  ├─ Flow meters (sub-second)        ├─ Production allocation (daily)   │
│  ├─ Pipeline SCADA (1-5s)           ├─ Maintenance work orders         │
│  ├─ Gas compression (1s)            ├─ Inspection records              │
│  └─ Tank level gauges (10-60s)      └─ HSE incident logs               │
│                                                                         │
│  SUBSURFACE DATA                    REGULATORY & COMMERCIAL             │
│  ├─ Reservoir pressure surveys      ├─ NUPRC production returns        │
│  ├─ Seismic survey data (TB+)       ├─ Gas flaring reports             │
│  ├─ Well log data (LAS/DLIS)        ├─ Environmental compliance        │
│  └─ PVT analysis results            └─ JV accounting & cost recovery   │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Volume Estimates for a Mid-Size Nigerian Operator

Data Source	Frequency	Volume per Day	Annual Volume
Wellhead sensors (200 wells)	5-second intervals	3.5 billion readings	~1.3 trillion
Pipeline SCADA (500 km network)	1-second intervals	8.6 billion readings	~3.1 trillion
Flow meters (custody transfer)	Sub-second	500 million readings	~180 billion
Rotating equipment vibration	10 ms intervals	50 GB raw waveforms	~18 TB
Well test data	Daily/weekly	5,000 records	~1.5 million

A mid-size operator with 200 producing wells generates approximately 50-100 TB of raw sensor data per year.

The Nigerian Context: Unique Challenges

Connectivity constraints: Many Niger Delta well sites have limited VSAT connectivity (2-10 Mbps), requiring edge computing and store-and-forward architectures rather than continuous cloud streaming.

Power reliability: Inconsistent grid power means edge devices must tolerate frequent power cycles. Data buffers must survive unclean shutdowns without data loss.

Regulatory complexity: The Petroleum Industry Act (PIA) 2021 introduced new NUPRC reporting requirements including granular production data submissions, environmental monitoring obligations, and host community development fund accounting.

Joint venture structures: Most Nigerian upstream operations are JV partnerships between NNPC and IOCs. Data platforms must support multi-party access controls, production sharing calculations, and cost recovery tracking across partners.

IoT Sensor Ingestion Architecture

End-to-End Architecture

┌──────────────────────────────────────────────────────────────────────────┐
│               SENSOR-TO-CLOUD INGESTION ARCHITECTURE                     │
├──────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  FIELD LAYER              EDGE LAYER              CLOUD LAYER            │
│                                                                          │
│  ┌──────────┐           ┌──────────────┐        ┌───────────────────┐   │
│  │ Wellhead │──Modbus──►│ Edge Gateway │──MQTT──►│ MQTT Broker       │   │
│  │ Sensors  │  /HART    │ (Ruggedized) │  over   │ (EMQX / HiveMQ)  │   │
│  └──────────┘           │              │  VSAT   └─────────┬─────────┘   │
│                         │ - Protocol   │                   │             │
│  ┌──────────┐           │   translation│                   ▼             │
│  │ Pipeline │──4-20mA──►│ - Local      │        ┌───────────────────┐   │
│  │ SCADA    │  /OPC-UA  │   buffering  │        │ Apache Kafka      │   │
│  │ RTUs     │           │ - Store &    │        │ - raw.wellhead.*  │   │
│  └──────────┘           │   forward    │        │ - raw.pipeline.*  │   │
│                         │ - Compression│        │ - raw.facility.*  │   │
│  ┌──────────┐           └──────────────┘        └─────────┬─────────┘   │
│  │ Facility │                                             │             │
│  │ DCS/PLC  │──OPC-UA──────────────────────────►          │             │
│  └──────────┘                                             ▼             │
│                                                 ┌───────────────────┐   │
│                                                 │ Stream Processing │   │
│                                                 │ (Flink / Kafka    │   │
│                                                 │  Streams)         │   │
│                                                 └────────┬──────────┘   │
│                                                   ┌──────┴──────┐       │
│                                                   ▼             ▼       │
│                                             ┌──────────┐ ┌──────────┐  │
│                                             │TimescaleDB│ │Data Lake │  │
│                                             │(Hot Data) │ │(Iceberg) │  │
│                                             └──────────┘ └──────────┘  │
└──────────────────────────────────────────────────────────────────────────┘

Edge Gateway Configuration

Each Niger Delta well site deploys a ruggedized edge gateway handling protocol translation and store-and-forward buffering:

# edge-gateway-config.yaml
gateway:
  id: "NG-OML58-WS-017"
  location:
    oml: "OML 58"
    field: "Obiafu-Obrikom"
    coordinates: [5.3847, 6.7231]
  connectivity:
    primary: "vsat"
    bandwidth_mbps: 5
    fallback: "4g-lte"

data_sources:
  - name: "wellhead_pressure"
    protocol: "modbus_tcp"
    host: "192.168.1.10"
    register_map:
      - { address: 40001, tag: "WHP_PSIG", type: "float32", scan_rate_ms: 5000 }
      - { address: 40003, tag: "WHT_DEGF", type: "float32", scan_rate_ms: 5000 }
      - { address: 40005, tag: "CHOKE_PCT", type: "float32", scan_rate_ms: 10000 }

  - name: "flow_meter"
    protocol: "opc_ua"
    endpoint: "opc.tcp://192.168.1.20:4840"
    nodes:
      - { node_id: "ns=2;s=OilFlowRate", tag: "OIL_BOPD", scan_rate_ms: 1000 }
      - { node_id: "ns=2;s=GasFlowRate", tag: "GAS_MSCFD", scan_rate_ms: 1000 }
      - { node_id: "ns=2;s=WaterCut", tag: "BSW_PCT", scan_rate_ms: 5000 }

local_buffer:
  type: "sqlite"
  max_size_gb: 50
  retention_days: 30

compression:
  algorithm: "deadband"
  default_deadband_pct: 1.0
  overrides:
    WHP_PSIG: 0.5
    OIL_BOPD: 0.1

upstream:
  protocol: "mqtt"
  broker: "mqtts://mqtt.gemutanalytics.io:8883"
  topic_prefix: "sensors/oml58/ws017"
  qos: 1
  batch_size: 100
  batch_interval_ms: 5000

store_and_forward:
  enabled: true
  retry_interval_s: 30
  max_queue_depth: 1000000

Sensor Data Model (TimescaleDB)

-- Hypertable for raw sensor readings
CREATE TABLE sensor_readings (
    time          TIMESTAMPTZ      NOT NULL,
    gateway_id    TEXT             NOT NULL,
    oml_block     TEXT             NOT NULL,
    field_name    TEXT             NOT NULL,
    well_id       TEXT,
    tag_name      TEXT             NOT NULL,
    value         DOUBLE PRECISION NOT NULL,
    unit          TEXT             NOT NULL,
    quality       TEXT             DEFAULT 'GOOD'
);

SELECT create_hypertable(
    'sensor_readings', 'time',
    chunk_time_interval => INTERVAL '1 hour'
);

CREATE INDEX idx_sensor_well_tag_time
    ON sensor_readings (well_id, tag_name, time DESC);

-- Compression for historical data (>7 days old)
ALTER TABLE sensor_readings SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'well_id, tag_name',
    timescaledb.compress_orderby = 'time DESC'
);
SELECT add_compression_policy('sensor_readings', compress_after => INTERVAL '7 days');

-- 5-minute continuous aggregate for dashboards
CREATE MATERIALIZED VIEW sensor_5min_avg
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('5 minutes', time) AS bucket,
    well_id, tag_name,
    AVG(value) AS avg_value, MIN(value) AS min_value,
    MAX(value) AS max_value, COUNT(*) AS sample_count
FROM sensor_readings
WHERE quality = 'GOOD'
GROUP BY bucket, well_id, tag_name
WITH NO DATA;

-- Hourly aggregate for trend analysis
CREATE MATERIALIZED VIEW sensor_hourly_avg
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS bucket,
    oml_block, field_name, well_id, tag_name,
    AVG(value) AS avg_value,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY value) AS p95_value,
    COUNT(*) AS sample_count
FROM sensor_readings
WHERE quality = 'GOOD'
GROUP BY bucket, oml_block, field_name, well_id, tag_name
WITH NO DATA;

-- Raw data retention: 90 days; aggregates kept indefinitely
SELECT add_retention_policy('sensor_readings', INTERVAL '90 days');

Ingestion Performance Benchmarks

On a 3-node TimescaleDB cluster (16 vCPUs, 64 GB RAM, NVMe SSD per node):

Metric	Result
Raw ingestion rate	580,000 rows/second
Compressed storage ratio	12:1
Query latency (single well, 24 hours)	8 ms
Query latency (field-wide, 7 days)	120 ms
Storage per well-year (compressed)	2.1 GB
Storage for 200 wells, 1 year	~420 GB compressed

Real-Time Production Monitoring

Well Performance Dashboard Queries

-- Current production status for all wells in an OML block
SELECT
    w.well_id, w.well_name, w.field_name,
    latest.whp_psig AS wellhead_pressure,
    latest.oil_bopd AS oil_rate,
    latest.gas_mscfd AS gas_rate,
    latest.bsw_pct AS water_cut,
    latest.gas_mscfd / NULLIF(latest.oil_bopd, 0) AS gor_scf_bbl,
    latest.reading_time AS last_reading,
    CASE
        WHEN latest.reading_time < NOW() - INTERVAL '30 minutes' THEN 'STALE'
        WHEN latest.whp_psig < w.min_whp_threshold THEN 'LOW_PRESSURE'
        WHEN latest.bsw_pct > w.max_bsw_threshold THEN 'HIGH_WATERCUT'
        WHEN latest.oil_bopd < w.min_production_threshold * 0.7 THEN 'UNDERPERFORMING'
        ELSE 'NORMAL'
    END AS well_status
FROM well_master w
CROSS JOIN LATERAL (
    SELECT
        MAX(CASE WHEN tag_name = 'WHP_PSIG' THEN value END) AS whp_psig,
        MAX(CASE WHEN tag_name = 'OIL_BOPD' THEN value END) AS oil_bopd,
        MAX(CASE WHEN tag_name = 'GAS_MSCFD' THEN value END) AS gas_mscfd,
        MAX(CASE WHEN tag_name = 'BSW_PCT' THEN value END) AS bsw_pct,
        MAX(time) AS reading_time
    FROM sensor_5min_avg
    WHERE well_id = w.well_id AND bucket >= NOW() - INTERVAL '15 minutes'
) latest
WHERE w.oml_block = 'OML 58' AND w.status = 'PRODUCING'
ORDER BY latest.oil_bopd DESC NULLS LAST;

Production Trend Analysis

Identifying wells with declining or improving production trends over a rolling 30-day window:

-- 30-day production trend with linear regression
WITH daily_production AS (
    SELECT
        well_id,
        time_bucket('1 day', bucket) AS production_date,
        AVG(avg_value) FILTER (WHERE tag_name = 'OIL_BOPD') AS avg_oil_bopd,
        AVG(avg_value) FILTER (WHERE tag_name = 'BSW_PCT') AS avg_bsw_pct,
        AVG(avg_value) FILTER (WHERE tag_name = 'WHP_PSIG') AS avg_whp_psig
    FROM sensor_hourly_avg
    WHERE oml_block = 'OML 58'
      AND bucket >= NOW() - INTERVAL '30 days'
    GROUP BY well_id, time_bucket('1 day', bucket)
)
SELECT
    well_id,
    ROUND(AVG(avg_oil_bopd)::numeric, 1) AS avg_oil_bopd,
    ROUND(AVG(avg_bsw_pct)::numeric, 1) AS avg_water_cut,
    -- Linear regression slope: bopd change per day
    ROUND((REGR_SLOPE(avg_oil_bopd, EXTRACT(EPOCH FROM production_date)) * 86400)::numeric, 2)
        AS daily_decline_bopd,
    ROUND(REGR_R2(avg_oil_bopd, EXTRACT(EPOCH FROM production_date))::numeric, 3)
        AS trend_confidence,
    CASE
        WHEN REGR_SLOPE(avg_oil_bopd, EXTRACT(EPOCH FROM production_date)) * 86400 < -5
             AND REGR_R2(avg_oil_bopd, EXTRACT(EPOCH FROM production_date)) > 0.6
            THEN 'DECLINING'
        WHEN REGR_SLOPE(avg_oil_bopd, EXTRACT(EPOCH FROM production_date)) * 86400 > 5
             AND REGR_R2(avg_oil_bopd, EXTRACT(EPOCH FROM production_date)) > 0.6
            THEN 'IMPROVING'
        ELSE 'STABLE'
    END AS production_trend
FROM daily_production
GROUP BY well_id
ORDER BY daily_decline_bopd ASC;

Anomaly Detection (Apache Flink)

Real-time anomaly detection on sensor streams using exponentially weighted moving average (EWMA) with adaptive thresholds:

# flink_anomaly_detector.py
import math
from dataclasses import dataclass
from datetime import datetime

from pyflink.datastream.functions import KeyedProcessFunction, RuntimeContext
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.common.typeinfo import Types


@dataclass
class AnomalyAlert:
    well_id: str
    tag_name: str
    alert_type: str        # STATISTICAL_DEVIATION, RAPID_CHANGE, FLATLINE
    severity: str          # WARNING, CRITICAL
    current_value: float
    expected_range: tuple[float, float]
    deviation_sigma: float
    timestamp: str
    description: str


class WellAnomalyDetector(KeyedProcessFunction):
    """EWMA-based anomaly detection per well/tag stream.

    Adapts to each well's operating profile, handling the gradual
    drift common in oil wells (declining pressure over months)
    while catching sudden spikes or sensor failures.
    """

    ALPHA = 0.01            # EWMA smoothing factor
    SIGMA_THRESHOLD = 3.5   # Std deviations for anomaly
    MIN_SAMPLES = 100       # Warmup before alerting

    def open(self, ctx: RuntimeContext) -> None:
        self.ewma_mean = ctx.get_state(
            ValueStateDescriptor("ewma_mean", Types.DOUBLE()))
        self.ewma_var = ctx.get_state(
            ValueStateDescriptor("ewma_var", Types.DOUBLE()))
        self.count = ctx.get_state(
            ValueStateDescriptor("count", Types.LONG()))
        self.last_value = ctx.get_state(
            ValueStateDescriptor("last_val", Types.DOUBLE()))
        self.last_ts = ctx.get_state(
            ValueStateDescriptor("last_ts", Types.LONG()))

    def process_element(self, reading, ctx):
        n = self.count.value() or 0
        mean = self.ewma_mean.value()
        var = self.ewma_var.value()

        if n == 0:
            self.ewma_mean.update(reading.value)
            self.ewma_var.update(0.0)
            self.count.update(1)
            self.last_value.update(reading.value)
            self.last_ts.update(int(reading.timestamp.timestamp()))
            return

        # Update EWMA statistics
        diff = reading.value - mean
        new_mean = mean + self.ALPHA * diff
        new_var = (1 - self.ALPHA) * (var + self.ALPHA * diff * diff)
        sigma = math.sqrt(new_var) if new_var > 0 else 0.001

        self.ewma_mean.update(new_mean)
        self.ewma_var.update(new_var)
        self.count.update(n + 1)

        if n + 1 < self.MIN_SAMPLES:
            self.last_value.update(reading.value)
            self.last_ts.update(int(reading.timestamp.timestamp()))
            return

        deviation = abs(diff) / sigma

        # Check 1: Statistical deviation
        if deviation > self.SIGMA_THRESHOLD:
            severity = "CRITICAL" if deviation > 5.0 else "WARNING"
            yield AnomalyAlert(
                well_id=reading.well_id,
                tag_name=reading.tag_name,
                alert_type="STATISTICAL_DEVIATION",
                severity=severity,
                current_value=reading.value,
                expected_range=(
                    round(new_mean - self.SIGMA_THRESHOLD * sigma, 2),
                    round(new_mean + self.SIGMA_THRESHOLD * sigma, 2),
                ),
                deviation_sigma=round(deviation, 2),
                timestamp=reading.timestamp.isoformat(),
                description=(
                    f"{reading.tag_name} at {reading.value:.2f} deviates "
                    f"{deviation:.1f}sigma from expected {new_mean:.2f}"
                ),
            )

        # Check 2: Flatline detection (sensor stuck >10 min)
        prev = self.last_value.value()
        prev_ts = self.last_ts.value()
        if prev is not None and reading.value == prev:
            stuck_s = int(reading.timestamp.timestamp()) - (prev_ts or 0)
            if stuck_s > 600:
                yield AnomalyAlert(
                    well_id=reading.well_id,
                    tag_name=reading.tag_name,
                    alert_type="FLATLINE",
                    severity="WARNING",
                    current_value=reading.value,
                    expected_range=(round(new_mean - sigma, 2),
                                    round(new_mean + sigma, 2)),
                    deviation_sigma=0.0,
                    timestamp=reading.timestamp.isoformat(),
                    description=(
                        f"{reading.tag_name} stuck at {reading.value:.2f} "
                        f"for {stuck_s}s — possible sensor failure"
                    ),
                )

        self.last_value.update(reading.value)
        self.last_ts.update(int(reading.timestamp.timestamp()))

Detection Performance

Testing against 6 months of historical data from OML 58 (45 producing wells):

Metric	Value
True positive rate (equipment failures)	92%
False positive rate	3.2%
Mean time to detection (vs. manual)	4.2 hours earlier
Processing latency (end-to-end)	< 500 ms

EWMA outperforms static thresholds by producing 15-20x fewer false positives because it adapts to each well's individual operating profile.

Predictive Maintenance for Pipeline Integrity

The Pipeline Challenge

Nigeria's pipeline infrastructure — the Trans-Niger Pipeline, Nembe Creek Trunk Line, and hundreds of smaller flow lines, many built in the 1960s-1980s — traverses the Niger Delta's swamp terrain and mangrove forests. Corrosion, third-party interference, and material fatigue are persistent threats.

Predictive Corrosion Model

The model combines sensor data, ILI history, environmental factors, and pipeline metadata to estimate corrosion growth rate and remaining useful life:

# pipeline_integrity_model.py
import numpy as np
from dataclasses import dataclass
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_absolute_error, r2_score
from sklearn.preprocessing import StandardScaler


@dataclass
class SegmentFeatures:
    segment_id: str
    age_years: float
    wall_thickness_mm: float
    environment_type_encoded: int    # swamp, land, river_crossing, offshore
    has_cathodic_protection: int
    avg_cp_potential_mv: float
    min_cp_potential_mv: float
    avg_operating_pressure_psi: float
    pressure_cycling_count_30d: int
    soil_resistivity_ohm_cm: float
    rainfall_annual_mm: float        # Niger Delta: 2,000-4,000 mm/yr
    coating_condition_score: float   # 1-5 from inspection
    last_ili_max_depth_pct: float
    years_since_last_ili: float


FEATURE_NAMES = [
    "age_years", "wall_thickness_mm", "environment_type_encoded",
    "has_cathodic_protection", "avg_cp_potential_mv", "min_cp_potential_mv",
    "avg_operating_pressure_psi", "pressure_cycling_count_30d",
    "soil_resistivity_ohm_cm", "rainfall_annual_mm",
    "coating_condition_score", "last_ili_max_depth_pct",
    "years_since_last_ili",
]


class PipelineIntegrityPredictor:
    """Predicts corrosion growth rate and remaining useful life.

    Key predictive factors for Niger Delta pipelines:
    - CP potential readings (most important single feature)
    - Pipeline age and material grade
    - Soil/water resistivity (highly variable in swamp terrain)
    - Annual rainfall (correlates with external corrosion rate)
    """

    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=500, max_depth=6, learning_rate=0.05,
            subsample=0.8, min_samples_leaf=10, random_state=42,
        )
        self.scaler = StandardScaler()

    def train(self, segments: list[SegmentFeatures], rates: np.ndarray) -> dict:
        X = np.array([[getattr(s, f) for f in FEATURE_NAMES] for s in segments])
        X_scaled = self.scaler.fit_transform(X)

        cv_scores = []
        for tr, va in TimeSeriesSplit(n_splits=5).split(X_scaled):
            self.model.fit(X_scaled[tr], rates[tr])
            preds = self.model.predict(X_scaled[va])
            cv_scores.append({
                "mae": mean_absolute_error(rates[va], preds),
                "r2": r2_score(rates[va], preds),
            })

        self.model.fit(X_scaled, rates)
        return {
            "cv_mae": np.mean([s["mae"] for s in cv_scores]),
            "cv_r2": np.mean([s["r2"] for s in cv_scores]),
            "feature_importance": dict(sorted(
                zip(FEATURE_NAMES, self.model.feature_importances_),
                key=lambda x: x[1], reverse=True,
            )),
        }

    def predict(self, segment: SegmentFeatures) -> dict:
        X = np.array([[getattr(segment, f) for f in FEATURE_NAMES]])
        rate = self.model.predict(self.scaler.transform(X))[0]

        current_depth = (
            segment.last_ili_max_depth_pct
            + (rate / segment.wall_thickness_mm) * 100 * segment.years_since_last_ili
        )
        remaining_wall = 100 - current_depth

        # ASME B31G simplified: repair at 60%, failure at 80%
        yrs_to_repair = max(0, (60 - current_depth) * segment.wall_thickness_mm / (rate * 100)) if rate > 0 else float("inf")
        yrs_to_failure = max(0, (80 - current_depth) * segment.wall_thickness_mm / (rate * 100)) if rate > 0 else float("inf")

        if yrs_to_failure < 2 or current_depth > 70:
            risk = "CRITICAL"
        elif yrs_to_repair < 3 or current_depth > 50:
            risk = "HIGH"
        elif yrs_to_repair < 5:
            risk = "MEDIUM"
        else:
            risk = "LOW"

        return {
            "segment_id": segment.segment_id,
            "predicted_rate_mmpy": round(rate, 3),
            "estimated_depth_pct": round(current_depth, 1),
            "remaining_wall_pct": round(remaining_wall, 1),
            "years_to_repair": round(yrs_to_repair, 1),
            "years_to_failure": round(yrs_to_failure, 1),
            "risk_level": risk,
        }

Model Performance on Nigerian Pipeline Data

Trained on 8 years of ILI comparison data from 1,200 segments across OML 11, OML 25, and OML 58:

Metric	Value
Cross-validated MAE (corrosion rate)	0.08 mm/year
Cross-validated R-squared	0.81
Top feature: CP potential (min)	18.3% importance
Top feature: Pipeline age	14.7% importance
Top feature: Soil resistivity	12.1% importance
Top feature: Rainfall annual	9.8% importance

The model correctly identified 87% of segments requiring repair within 2 years, with an 11% false positive rate. Compared to time-based scheduling, it reduced total ILI runs by 35%.

Time-Series Analytics for Reservoir Performance

Decline Curve Analysis

Decline curve analysis (DCA) is the workhorse of reservoir engineering. A data platform can automate DCA across hundreds of wells simultaneously, replacing manual Excel-based analysis:

# decline_curve_analysis.py
import numpy as np
from scipy.optimize import curve_fit
from dataclasses import dataclass


@dataclass
class DCAResult:
    well_id: str
    model_type: str          # exponential, hyperbolic, harmonic
    qi: float                # Initial rate (BOPD)
    di: float                # Initial decline rate (1/year)
    b_factor: float          # Arps b-factor
    r_squared: float
    eur_mbbl: float          # Estimated Ultimate Recovery
    remaining_reserves_mbbl: float


def exponential(t, qi, di):
    return qi * np.exp(-di * t)

def hyperbolic(t, qi, di, b):
    return qi / np.power(1 + b * di * t, 1.0 / b)

def harmonic(t, qi, di):
    return qi / (1 + di * t)


def fit_decline_curve(
    well_id: str,
    dates: np.ndarray,
    rates: np.ndarray,
    cumulative: np.ndarray,
    forecast_years: int = 20,
    economic_limit_bopd: float = 10.0,
) -> DCAResult:
    """Fit best Arps decline model to production history."""
    t_years = (dates - dates[0]).astype("timedelta64[D]").astype(float) / 365.25
    mask = rates > 0
    t, q = t_years[mask], rates[mask]

    best_model, best_r2 = None, -np.inf

    for name, func, p0, bounds in [
        ("exponential", exponential, [q[0], 0.1],
         ([0, 0.001], [q[0]*2, 2.0])),
        ("hyperbolic", hyperbolic, [q[0], 0.1, 0.5],
         ([0, 0.001, 0.01], [q[0]*2, 2.0, 1.0])),
        ("harmonic", harmonic, [q[0], 0.1],
         ([0, 0.001], [q[0]*2, 2.0])),
    ]:
        try:
            popt, _ = curve_fit(func, t, q, p0=p0, bounds=bounds, maxfev=10000)
            pred = func(t, *popt)
            ss_res = np.sum((q - pred) ** 2)
            ss_tot = np.sum((q - np.mean(q)) ** 2)
            r2 = 1 - ss_res / ss_tot if ss_tot > 0 else 0

            if r2 > best_r2:
                best_r2 = r2
                b = popt[2] if name == "hyperbolic" else (1.0 if name == "harmonic" else 0.0)
                best_model = {"type": name, "qi": popt[0], "di": popt[1], "b": b, "r2": r2}
        except RuntimeError:
            continue

    # Calculate EUR via numerical integration
    dt = 1.0 / 365.25
    t_fcast = np.arange(0, t[-1] + forecast_years, dt)
    if best_model["type"] == "hyperbolic":
        q_fcast = hyperbolic(t_fcast, best_model["qi"], best_model["di"], best_model["b"])
    elif best_model["type"] == "harmonic":
        q_fcast = harmonic(t_fcast, best_model["qi"], best_model["di"])
    else:
        q_fcast = exponential(t_fcast, best_model["qi"], best_model["di"])

    q_fcast = np.where(q_fcast >= economic_limit_bopd, q_fcast, 0)
    eur_mbbl = np.sum(q_fcast * dt * 365.25) / 1000
    cum_mbbl = cumulative[-1] / 1000

    return DCAResult(
        well_id=well_id,
        model_type=best_model["type"],
        qi=round(best_model["qi"], 1),
        di=round(best_model["di"], 4),
        b_factor=round(best_model["b"], 3),
        r_squared=round(best_model["r2"], 4),
        eur_mbbl=round(eur_mbbl, 1),
        remaining_reserves_mbbl=round(max(0, eur_mbbl - cum_mbbl), 1),
    )

Field-Level Reserves Summary

-- Aggregated DCA results for field-level reserves reporting
SELECT
    w.field_name,
    COUNT(*) AS producing_wells,
    ROUND(SUM(d.qi_bopd)) AS current_rate_bopd,
    ROUND(SUM(d.eur_mbbl)) AS total_eur_mbbl,
    ROUND(SUM(d.remaining_reserves_mbbl)) AS remaining_mbbl,
    ROUND(SUM(d.cumulative_mbbl) / NULLIF(SUM(d.eur_mbbl), 0) * 100, 1)
        AS recovery_factor_pct,
    ROUND(SUM(d.di_annual * d.qi_bopd) / NULLIF(SUM(d.qi_bopd), 0) * 100, 1)
        AS weighted_decline_pct
FROM well_master w
JOIN decline_curve_results d ON w.well_id = d.well_id
WHERE d.forecast_date = (SELECT MAX(forecast_date) FROM decline_curve_results)
  AND w.oml_block = 'OML 58'
GROUP BY w.field_name
ORDER BY current_rate_bopd DESC;

Regulatory Reporting Automation

NUPRC Reporting Obligations

Report	Frequency	Current Pain Point
Daily Production Return	Daily	Manual collation from field reports
Monthly Production Report	Monthly	5-10 day preparation time
Gas Flare Report	Monthly	Incomplete metering data
Environmental Monitoring	Quarterly	Paper-based field reporting
JV Cost Recovery Statement	Monthly	Multi-system reconciliation

Automated Reporting Pipeline

┌──────────────────────────────────────────────────────────────────────┐
│              REGULATORY REPORTING AUTOMATION                         │
├──────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  DATA SOURCES             PROCESSING             OUTPUT              │
│                                                                      │
│  ┌────────────┐     ┌────────────────┐    ┌─────────────────────┐   │
│  │ TimescaleDB│────►│                │───►│ NUPRC Daily Return  │   │
│  │ (Sensors)  │     │  Apache        │    │ (Automated XML)     │   │
│  └────────────┘     │  Airflow DAGs  │    └─────────────────────┘   │
│  ┌────────────┐     │                │    ┌─────────────────────┐   │
│  │ Production │────►│  - Validate    │───►│ Monthly Production  │   │
│  │ Database   │     │  - Reconcile   │    │ Report (PDF + CSV)  │   │
│  └────────────┘     │  - Allocate    │    └─────────────────────┘   │
│  ┌────────────┐     │  - Format      │    ┌─────────────────────┐   │
│  │ Flare      │────►│  - Audit trail │───►│ Gas Flare Report    │   │
│  │ Meters     │     │                │    └─────────────────────┘   │
│  └────────────┘     └────────────────┘                              │
│  ┌────────────┐            │              ┌─────────────────────┐   │
│  │ SAP / JDE  │────────────┘         ────►│ JV Cost Recovery    │   │
│  │ (Finance)  │                           └─────────────────────┘   │
│  └────────────┘                                                     │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

Gas Flaring Compliance

-- Gas flaring compliance tracking (PIA Section 104: $2/MSCF penalty)
WITH daily_flare AS (
    SELECT
        time_bucket('1 day', bucket) AS flare_date,
        oml_block,
        SUM(avg_value) FILTER (WHERE tag_name = 'GAS_FLARE_MSCFD') AS flared,
        SUM(avg_value) FILTER (WHERE tag_name = 'GAS_PRODUCED_MSCFD') AS produced
    FROM sensor_hourly_avg
    WHERE bucket >= DATE_TRUNC('month', CURRENT_DATE)
      AND tag_name IN ('GAS_FLARE_MSCFD', 'GAS_PRODUCED_MSCFD')
    GROUP BY time_bucket('1 day', bucket), oml_block
)
SELECT
    flare_date, oml_block,
    ROUND(flared, 1) AS flare_mscfd,
    ROUND((1 - flared / NULLIF(produced, 0)) * 100, 1) AS utilization_pct,
    SUM(flared) OVER (PARTITION BY oml_block ORDER BY flare_date) AS mtd_flare,
    CASE
        WHEN (1 - flared / NULLIF(produced, 0)) >= 0.95 THEN 'COMPLIANT'
        WHEN (1 - flared / NULLIF(produced, 0)) >= 0.90 THEN 'WARNING'
        ELSE 'NON_COMPLIANT'
    END AS status,
    ROUND(SUM(flared) OVER (PARTITION BY oml_block ORDER BY flare_date) * 2.0, 0)
        AS estimated_penalty_usd
FROM daily_flare
ORDER BY flare_date DESC;

JV Production Sharing Calculation

-- PSC production sharing (NNPC / Contractor split)
WITH monthly AS (
    SELECT
        oml_block,
        DATE_TRUNC('month', production_date) AS month,
        SUM(allocated_oil_bbl) AS oil_bbl,
        AVG(brent_price_usd) AS price
    FROM daily_production_allocated
    WHERE production_date >= DATE_TRUNC('year', CURRENT_DATE)
    GROUP BY oml_block, DATE_TRUNC('month', production_date)
),
costs AS (
    SELECT
        oml_block,
        DATE_TRUNC('month', posting_date) AS month,
        SUM(amount_usd) FILTER (WHERE cost_category = 'OPEX') AS opex,
        SUM(amount_usd) FILTER (WHERE cost_category = 'CAPEX') AS capex,
        SUM(amount_usd) AS total_cost
    FROM jv_cost_ledger
    WHERE approved = TRUE
    GROUP BY oml_block, DATE_TRUNC('month', posting_date)
)
SELECT
    m.oml_block, m.month,
    ROUND(m.oil_bbl) AS production_bbl,
    ROUND(m.oil_bbl * m.price) AS gross_revenue,
    ROUND(m.oil_bbl * m.price * 0.10) AS royalty,          -- 10% onshore
    ROUND(LEAST(c.total_cost, m.oil_bbl * m.price * 0.72)) AS cost_recovery,
    ROUND((m.oil_bbl * m.price * 0.90 - LEAST(c.total_cost, m.oil_bbl * m.price * 0.72)) * 0.60)
        AS nnpc_profit_oil,          -- 60% NNPC share
    ROUND((m.oil_bbl * m.price * 0.90 - LEAST(c.total_cost, m.oil_bbl * m.price * 0.72)) * 0.40)
        AS contractor_profit_oil     -- 40% contractor share
FROM monthly m
LEFT JOIN costs c ON m.oml_block = c.oml_block AND m.month = c.month
ORDER BY m.month DESC;

Environmental Monitoring: Spill Detection

Oil spill detection is a critical environmental and operational concern in the Niger Delta. A data-driven approach integrates pipeline sensor anomalies, satellite imagery analysis, and community reporting:

-- Spill incident tracking and response KPIs
CREATE TABLE spill_incidents (
    incident_id         SERIAL PRIMARY KEY,
    detection_time      TIMESTAMPTZ NOT NULL,
    detection_method    TEXT NOT NULL,   -- SENSOR, SATELLITE, COMMUNITY_REPORT
    oml_block           TEXT NOT NULL,
    pipeline_segment    TEXT,
    location_lat        NUMERIC(9,6),
    location_lon        NUMERIC(9,6),
    estimated_volume_bbl NUMERIC(10,2),
    spill_cause         TEXT,           -- CORROSION, THIRD_PARTY, EQUIPMENT_FAILURE
    environment_type    TEXT,           -- LAND, SWAMP, WATERWAY
    response_status     TEXT DEFAULT 'DETECTED',
    response_time_hrs   NUMERIC(8,2),
    jiv_date            DATE,           -- Joint Investigation Visit
    nuprc_notification  TIMESTAMPTZ
);

-- Quarterly spill response dashboard
SELECT
    DATE_TRUNC('quarter', detection_time) AS quarter,
    oml_block,
    COUNT(*) AS total_incidents,
    COUNT(*) FILTER (WHERE spill_cause = 'CORROSION') AS corrosion_spills,
    COUNT(*) FILTER (WHERE spill_cause = 'THIRD_PARTY') AS third_party_spills,
    ROUND(SUM(estimated_volume_bbl)::numeric, 0) AS total_volume_bbl,
    ROUND(AVG(response_time_hrs)::numeric, 1) AS avg_response_hrs,
    ROUND(COUNT(*) FILTER (WHERE response_time_hrs <= 24)::numeric
        / COUNT(*) * 100, 1) AS pct_responded_24hrs
FROM spill_incidents
WHERE detection_time >= CURRENT_DATE - INTERVAL '1 year'
GROUP BY DATE_TRUNC('quarter', detection_time), oml_block
ORDER BY quarter DESC;

Integrating pipeline sensor anomaly alerts with the spill incident database enables correlation analysis between detected pressure drops, flow rate changes, and confirmed spill events — improving the predictive model's sensitivity over time.

Platform Cost and ROI

Infrastructure Sizing (200 wells, 500 km pipeline, 2 facilities)

MONTHLY COST ESTIMATE

EDGE LAYER
├─ 50 edge gateways (amortized)                     $2,500
├─ VSAT connectivity (50 sites x $300)               $15,000
└─ Subtotal                                           $17,500

CLOUD INGESTION
├─ MQTT Broker cluster (EMQX, 3 nodes)               $1,200
├─ Kafka cluster (6 brokers, managed)                 $4,800
├─ Flink cluster (stream processing)                  $2,400
└─ Subtotal                                           $8,400

DATA STORAGE
├─ TimescaleDB cluster (3 nodes, 2TB NVMe each)      $3,600
├─ Data lake (S3/Azure Blob, ~5TB/month)              $120
└─ Subtotal                                           $3,720

ANALYTICS & REPORTING
├─ Grafana Enterprise                                 $1,500
├─ Apache Airflow (orchestration)                     $800
├─ ML model serving                                   $600
└─ Subtotal                                           $2,900

───────────────────────────────────────────────────────
TOTAL MONTHLY INFRASTRUCTURE                          $32,520
TOTAL ANNUAL INFRASTRUCTURE                           $390,240
+ Engineering (3 data engineers + 1 PE)               $340,000
───────────────────────────────────────────────────────
TRUE ANNUAL TCO                                       $730,240

ROI Benchmarks

Benefit Area	Improvement	Annual Value
Reduced unplanned downtime	30-45% fewer shutdowns	$2-4M
Production allocation accuracy	2-5% improvement	$500K-1.5M
Predictive maintenance savings	35% fewer ILI runs	$1-2M
Regulatory compliance automation	80% time reduction	$200-400K
Gas flaring penalty avoidance	10-20% flare reduction	$500K-2M
Total estimated annual benefit		$4.2-9.9M

Against a TCO of ~$730K/year, the platform delivers an estimated 6-13x return on investment.

Conclusion

Building a modern data platform for upstream oil and gas requires addressing the full spectrum — from ruggedized edge computing at remote Niger Delta well sites to automated NUPRC regulatory submissions. The key engineering decisions:

1. Edge-first architecture: Design for intermittent connectivity and store-and-forward resilience — the Niger Delta demands it
2. TimescaleDB for sensor analytics: Full SQL compatibility and strong compression make it ideal for teams transitioning from legacy historians
3. EWMA-based anomaly detection: Adaptive statistical methods outperform static thresholds by 15-20x in false positive reduction
4. Predictive pipeline integrity: Combining ILI history, CP readings, and environmental factors achieves 0.08 mm/year MAE on corrosion prediction
5. Automated regulatory reporting: End-to-end pipelines from sensors to NUPRC XML eliminate weeks of manual preparation
6. JV-ready data governance: Multi-party access controls and auditable lineage are non-negotiable in Nigerian JV operations

These patterns are applicable globally — operators in the Middle East, North Sea, and Americas face similar challenges. The Nigerian context, with its unique combination of connectivity constraints, environmental sensitivity, and JV regulatory complexity, makes it one of the most demanding deployment environments. A platform that succeeds here will succeed anywhere.

References

• Petroleum Industry Act (PIA) 2021 - Nigerian Upstream Petroleum Regulatory Commission
• ASME B31G - Remaining Strength of Corroded Pipelines
• API 1160 - Managing System Integrity for Hazardous Liquid Pipelines
• TimescaleDB Documentation
• Apache Kafka Documentation
• Apache Flink Documentation
• Arps Decline Curve Analysis - SPE PetroWiki
• DNV-RP-F101 - Corroded Pipelines
• Nigeria Extractive Industries Transparency Initiative (NEITI)