""" 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()