build_dataset.py

"""
build_dataset.py

Reproduces the processed Volve well-condition dataset, the grounded event
labels, and the per-event provenance file directly from Equinor's raw open
production export.

INPUT (obtain from Equinor's open data distribution; not redistributed here):
    Volve_production_data.xlsx   -- the raw production export.
                                    We use only its "Daily Production Data" sheet.

OUTPUTS (written to the current directory):
    volve_well_daily.csv / .parquet   -- cleaned per-well daily telemetry
    volve_grounded_events.csv         -- 236 grounded anomaly events (5 types)
    volve_events_provenance.csv       -- per-event audit trail (well, dates,
                                         channels, event type, PUD rationale)

Every processing decision below is a stated choice, documented in the paper's
Method section and in CODEBOOK.md. Thresholds are declared assumptions, not
facts: a different threshold would produce a different label set.

Usage:
    python build_dataset.py --raw Volve_production_data.xlsx

Dependencies: pandas, numpy, openpyxl, pyarrow  (no GPU required)
"""

import argparse
import numpy as np
import pandas as pd



def load_and_clean(raw_path):
    """Read the Daily Production Data sheet and clean it into a per-well table.

    All telemetry in this study derives from this single sheet; this is the
    provenance anchor. See paper Section 3.
    """
    df = pd.read_excel(raw_path, sheet_name="Daily Production Data")
    df["DATEPRD"] = pd.to_datetime(df["DATEPRD"])

    keep = [
        "DATEPRD", "NPD_WELL_BORE_NAME", "WELL_TYPE", "FLOW_KIND",
        "ON_STREAM_HRS", "AVG_DOWNHOLE_PRESSURE", "AVG_DOWNHOLE_TEMPERATURE",
        "AVG_DP_TUBING", "AVG_ANNULUS_PRESS", "AVG_WHP_P", "AVG_WHT_P",
        "AVG_CHOKE_SIZE_P", "BORE_OIL_VOL", "BORE_GAS_VOL", "BORE_WAT_VOL",
        "BORE_WI_VOL",
    ]
    clean = df[keep].rename(columns={
        "DATEPRD": "date", "NPD_WELL_BORE_NAME": "well", "WELL_TYPE": "well_type",
        "FLOW_KIND": "flow_kind", "ON_STREAM_HRS": "on_stream_hrs",
        "AVG_DOWNHOLE_PRESSURE": "dh_pressure",
        "AVG_DOWNHOLE_TEMPERATURE": "dh_temp", "AVG_DP_TUBING": "dp_tubing",
        "AVG_ANNULUS_PRESS": "annulus_press", "AVG_WHP_P": "whp",
        "AVG_WHT_P": "wht", "AVG_CHOKE_SIZE_P": "choke_pct",
        "BORE_OIL_VOL": "oil_vol", "BORE_GAS_VOL": "gas_vol",
        "BORE_WAT_VOL": "water_vol", "BORE_WI_VOL": "wi_vol",
    })
    clean = clean.sort_values(["well", "date"]).reset_index(drop=True)

    # Derived water cut, used by the water-breakthrough rule below.
    tot = clean["oil_vol"] + clean["water_vol"]
    clean["water_cut"] = np.where(tot > 0, clean["water_vol"] / tot, np.nan)
    return clean



#2 GROUNDED EVENT LAYER

'''
Each event type is tied to a documented mechanism in the Volve field development plan (PUD). A rule proposes a candidate;
the candidate is admitted only because the PUD documents that this mechanism occurs on these wells.
See paper Section 4. Thresholds are declared assumptions.
'''
PUD = {
    "shut_in":
        "PUD S5.3.2/S5.3.6: hydrate & wax risk arises specifically during "
        "well shut-in; on_stream_hrs->0 with live well = documented shut-in.",
    "restart_transient":
        "PUD S5.2/S4.5: gas-lift used to restart wells after shut-in; abrupt "
        "pressure/rate recovery after a stop is a documented restart signature.",
    "water_breakthrough":
        "PUD S4.5.2: injection water breaks through ~2 yr into production; "
        "sustained water-cut rise is the documented breakthrough mechanism.",
    "productivity_loss":
        "PUD S5.3.1/S5.3.4: barium/strontium scaling & asphaltene "
        "precipitation reduce productivity; sustained rate decline at stable "
        "choke.",
    "gaslift_instability":
        "PUD S5.2: gas-lift wells show annulus-pressure instability; anomalous "
        "annulus-pressure deviation is a documented lift-system signature.",
}


def build_events(clean):
    """Rule-derived candidates, admitted with PUD corroboration. See Section 4."""
    op = clean[clean.well_type == "OP"].copy().sort_values(["well", "date"])
    events = []

    def add(well, t0, t1, etype, chans, detail):
        events.append(dict(
            well=well, onset=t0, offset=t1, event_type=etype,
            channels=";".join(chans), pud_rationale=PUD[etype], detail=detail,
        ))

    for well, g in op.groupby("well"):
        g = g.sort_values("date").reset_index(drop=True)
        osh = g["on_stream_hrs"].fillna(0).values
        dates = g["date"].values
        oil = g["oil_vol"].replace(0, np.nan).values
        wc = g["water_cut"].values
        ann = g["annulus_press"].replace(0, np.nan).values
        choke = g["choke_pct"].replace(0, np.nan).values
        n = len(g)

        # SHUT-IN: on_stream_hrs ~0 for >=2 consecutive days (declared: 2 days)
        shut = (osh < 1.0)
        i = 0
        while i < n:
            if shut[i]:
                j = i
                while j < n and shut[j]:
                    j += 1
                if j - i >= 2:
                    add(well, dates[i], dates[j - 1], "shut_in",
                        ["on_stream_hrs"], f"{j-i}-day zero-onstream span")
                    # (2) RESTART: recovery day immediately after shut-in ends
                    if j < n and not np.isnan(oil[j]):
                        add(well, dates[j], dates[min(j + 1, n - 1)],
                            "restart_transient",
                            ["on_stream_hrs", "whp", "oil_vol"],
                            "rate recovery post shut-in")
                i = j
            else:
                i += 1

        # WATER BREAKTHROUGH: 7-day water cut crosses & holds >=0.5 (declared)
        wcs = pd.Series(wc).rolling(7, min_periods=3).mean().values
        above = wcs >= 0.5
        for k in range(1, n):
            if above[k] and not above[k - 1]:
                end = min(k + 14, n)
                if np.nanmean(wcs[k:end]) >= 0.5:
                    add(well, dates[k], dates[end - 1], "water_breakthrough",
                        ["water_cut", "water_vol"], "water-cut sustained >=0.5")
                    break  # first breakthrough only

        # PRODUCTIVITY LOSS: oil down >30% over 30d at stable choke (declared)
        oil_s = pd.Series(oil).rolling(7, min_periods=3).mean().values
        ch_s = pd.Series(choke).rolling(7, min_periods=3).mean().values
        for k in range(30, n):
            base, cur = oil_s[k - 30], oil_s[k]
            if np.isfinite(base) and np.isfinite(cur) and base > 50:
                if cur < 0.7 * base:
                    if (np.isfinite(ch_s[k]) and np.isfinite(ch_s[k - 30])
                            and ch_s[k] >= 0.8 * ch_s[k - 30]):
                        add(well, dates[k - 30], dates[k], "productivity_loss",
                            ["oil_vol", "choke_pct"],
                            f"oil {base:.0f}->{cur:.0f} at stable choke")
                        break  # first sustained decline only

        #  GAS-LIFT INSTABILITY: annulus pressure >3 MAD from well median
        if np.isfinite(np.nanmedian(ann)):
            med = np.nanmedian(ann)
            mad = np.nanmedian(np.abs(ann - med)) or 1
            z = np.abs(ann - med) / (1.4826 * mad)
            spikes = z > 3
            k = 0
            while k < n:
                if spikes[k] and np.isfinite(ann[k]):
                    j = k
                    while j < n and spikes[j]:
                        j += 1
                    add(well, dates[k], dates[j - 1], "gaslift_instability",
                        ["annulus_press"], f"annulus z>3 ({j-k}d)")
                    k = j
                else:
                    k += 1

    ev = pd.DataFrame(events)
    ev["onset"] = pd.to_datetime(ev["onset"])
    ev["offset"] = pd.to_datetime(ev["offset"])
    ev["duration_days"] = (ev["offset"] - ev["onset"]).dt.days + 1
    ev.insert(0, "event_id", [f"VOLVE-EV-{i:04d}" for i in range(len(ev))])
    return ev


# 3. MAIN
def main():
    parser = argparse.ArgumentParser(
        description="Build the processed Volve dataset, grounded events, and "
                    "provenance from Equinor's raw production export.")
    parser.add_argument("--raw", default="Volve_production_data.xlsx",
                        help="Path to Equinor's raw Volve_production_data.xlsx")
    args = parser.parse_args()

    print(f"Reading raw export: {args.raw}")
    clean = load_and_clean(args.raw)
    clean.to_parquet("volve_well_daily.parquet", index=False)
    clean.to_csv("volve_well_daily.csv", index=False)
    print(f"  wrote volve_well_daily.csv/.parquet  ({len(clean)} daily rows)")

    ev = build_events(clean)
    ev.to_csv("volve_grounded_events.csv", index=False)
    print(f"  wrote volve_grounded_events.csv      ({len(ev)} grounded events)")

    # The provenance file is the same events with the audit columns made explicit.
    prov = ev[["event_id", "well", "onset", "offset", "event_type",
               "channels", "pud_rationale", "detail"]]
    prov.to_csv("volve_events_provenance.csv", index=False)
    print(f"  wrote volve_events_provenance.csv    ({len(prov)} provenance rows)")

    print("\nEvent breakdown by type:")
    for etype, count in ev.event_type.value_counts().items():
        print(f"  {etype:22s}: {count}")


if __name__ == "__main__":
    main()