Upload build_dataset.py

build_dataset.py

@@ -0,0 +1,221 @@
+
+"""
+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()