gavindiaz/GoldRegime_X
1
0
Fork 0
mirror of https://github.com/lucasmos/GoldRegime_X.git synced 2026-11-02 06:26:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove obsolete patch scripts for Explorer notebook cells and add new bfg.jar file.

Browse Source
This commit is contained in:
Luke Mosoti
2026-07-14 07:40:12 +03:00
parent aaa843681c
commit 0f449a0eda
8 changed files with 313 additions and 16901 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+22
View File
@@ -33,3 +33,25 @@ mql5/*.ex5
.idea/
nul
.kilo/
# Too large for GitHub (>100 MB hard limit)
pipeline_verification_bundle/reports/candidate_portfolios.json
pipeline_verification_bundle/reports/candidate_portfolios.json
pipeline_verification_bundle/reports/candidate_trades_export.csv
pipeline_verification_bundle/reports/candidate_trades_export.csv
# Also exclude any regenerated copies with size suffixes / timestamps
**/candidate_portfolios*.json
**/candidate_trades_export*.csv
cd C:\GoldRegime_X
@"
# --- Files >100 MB that GitHub rejects ---
pipeline_verification_bundle/reports/candidate_portfolios.json
pipeline_verification_bundle/reports/candidate_trades_export.csv
"@ | Add-Content -Path .gitignore -Encoding UTF8
git add .gitignore
git commit -m "Ignore files exceeding GitHub 100MB limit"
bfg.jar
BIN
View File
Binary file not shown.
pipeline_verification_bundle/GoldRegimeX_Explorer_fixed.ipynb
+135 -2753
View File
File diff suppressed because it is too large Load Diff
pipeline_verification_bundle/GoldRegimeX_Explorer_fixed.orig.ipynb
-7741
View File
File diff suppressed because it is too large Load Diff
pipeline_verification_bundle/Strategy_Tester_fixed.ipynb
+156 -47
View File
@@ -16,32 +16,120 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 1,
"id": "obs_init_v1",
"metadata": {
"tags": [
"pipeline_observability_v1"
]
},
"outputs": [],
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"[observability] WARNING: using local stub implementations for pipeline_verification_bundle.*\n",
"[observability] run_id=990abc78e9 -> reports\\observability\n",
"[observability] logger, model_uuid_tracker, pipeline_manifest are now defined.\n"
]
}
],
"source": [
"# --- Pipeline Observability init (tag: pipeline_observability_v1) --------\n",
"# Auto-inserted by instrument_notebooks.py. Do not edit manually --\n",
"# re-run the injector to update.\n",
"from pathlib import Path\n",
"import sys\n",
"import hashlib\n",
"import time\n",
"from dataclasses import dataclass\n",
"\n",
"try:\n",
" from pipeline_verification_bundle.shared.pipeline_observability import (\n",
" PipelineObservability, PipelineLogger, ModelUUIDTracker,\n",
" PipelineManifest, CandidateTrace,\n",
" )\n",
"except ImportError:\n",
" import sys as _sys\n",
" _sys.path.insert(0, str(Path.cwd()))\n",
" from pipeline_verification_bundle.shared.pipeline_observability import (\n",
" PipelineObservability, PipelineLogger, ModelUUIDTracker,\n",
" PipelineManifest, CandidateTrace,\n",
" )\n",
"except Exception:\n",
" # Best-effort: try to make local package importable, then fallback to lightweight stubs\n",
" sys.path.insert(0, str(Path.cwd()))\n",
" try:\n",
" from pipeline_verification_bundle.shared.pipeline_observability import (\n",
" PipelineObservability, PipelineLogger, ModelUUIDTracker,\n",
" PipelineManifest, CandidateTrace,\n",
" )\n",
" except Exception:\n",
" # Minimal stubs to allow the notebook to run when the real package is unavailable.\n",
" class DummyLedger:\n",
" def __init__(self):\n",
" self._items = []\n",
"\n",
" def record(self, *args, **kwargs):\n",
" return\n",
"\n",
" def list_all(self):\n",
" return list(self._items)\n",
"\n",
" def add(self, item):\n",
" self._items.append(item)\n",
"\n",
" class DummyDecisions:\n",
" def __init__(self):\n",
" self._events = []\n",
"\n",
" def record(self, *args, **kwargs):\n",
" self._events.append((args, kwargs))\n",
"\n",
" class PipelineObservability:\n",
" def __init__(\n",
" self, output_dir, expected_session_by_tf=None,\n",
" material_survival_gap_pct=10.0, lost_trade_limit=100, lost_trade_tf=\"M15\"\n",
" ):\n",
" self.output_dir = Path(output_dir)\n",
" self.expected_session_by_tf = expected_session_by_tf or {}\n",
" self.material_survival_gap_pct = material_survival_gap_pct\n",
" self.lost_trade_limit = lost_trade_limit\n",
" self.lost_trade_tf = lost_trade_tf\n",
" self.ledger = DummyLedger()\n",
" self.decisions = DummyDecisions()\n",
" self.run_id = hashlib.sha256(f\"{time.time()}\".encode()).hexdigest()[:10]\n",
"\n",
" def import_from_profiler(self, profiler):\n",
" # best-effort hydration: no-op for stub\n",
" return {\"hydrated\": False}\n",
"\n",
" class PipelineLogger:\n",
" def __init__(self, ledger, decisions):\n",
" self.ledger = ledger\n",
" self.decisions = decisions\n",
"\n",
" def log(self, *args, **kwargs):\n",
" self.decisions.record(*args, **kwargs)\n",
"\n",
" class ModelUUIDTracker:\n",
" def __init__(self):\n",
" self._uuids = {}\n",
"\n",
" def register(self, tf, uuid):\n",
" self._uuids.setdefault(tf, set()).add(uuid)\n",
"\n",
" def verify_all(self):\n",
" return {}\n",
"\n",
" class PipelineManifest:\n",
" def __init__(self, path: Path):\n",
" self.path = Path(path)\n",
"\n",
" def write(self, **kwargs):\n",
" # returns the path for compatibility with calling code\n",
" return str(self.path)\n",
"\n",
" @staticmethod\n",
" def hash_object(obj):\n",
" s = str(obj).encode()\n",
" return hashlib.sha256(s).hexdigest()[:12]\n",
"\n",
" @dataclass\n",
" class CandidateTrace:\n",
" candidate_id: str\n",
" metadata: dict\n",
"\n",
" print(\"[observability] WARNING: using local stub implementations for pipeline_verification_bundle.*\")\n",
"\n",
"OBS_OUTPUT_DIR = Path(\"reports/observability\")\n",
"OBS_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)\n",
@@ -70,7 +158,7 @@
},
{
"cell_type": "code",
"execution_count": 1,
"execution_count": 2,
"id": "1bf0ecdc",
"metadata": {},
"outputs": [
@@ -215,7 +303,7 @@
},
{
"cell_type": "code",
"execution_count": 2,
"execution_count": 3,
"id": "530bb260",
"metadata": {},
"outputs": [
@@ -250,7 +338,7 @@
},
{
"cell_type": "code",
"execution_count": 3,
"execution_count": 4,
"id": "756bf414",
"metadata": {},
"outputs": [
@@ -621,7 +709,7 @@
},
{
"cell_type": "code",
"execution_count": 4,
"execution_count": 5,
"id": "ba12d976",
"metadata": {},
"outputs": [
@@ -643,7 +731,7 @@
},
{
"cell_type": "code",
"execution_count": 5,
"execution_count": 6,
"id": "1d5b1831",
"metadata": {},
"outputs": [],
@@ -764,7 +852,7 @@
},
{
"cell_type": "code",
"execution_count": 6,
"execution_count": 7,
"id": "a0f6fe15",
"metadata": {},
"outputs": [],
@@ -1175,7 +1263,7 @@
},
{
"cell_type": "code",
"execution_count": 7,
"execution_count": 8,
"id": "03e4cb76",
"metadata": {},
"outputs": [],
@@ -1441,7 +1529,7 @@
},
{
"cell_type": "code",
"execution_count": 8,
"execution_count": 9,
"id": "80085523",
"metadata": {},
"outputs": [
@@ -1478,12 +1566,12 @@
"output_type": "stream",
"text": [
"[Parallel(n_jobs=6)]: Using backend LokyBackend with 6 concurrent workers.\n",
"[Parallel(n_jobs=6)]: Done 1 tasks | elapsed: 1.7min\n",
"[Parallel(n_jobs=6)]: Done 6 tasks | elapsed: 6.3min\n",
"[Parallel(n_jobs=6)]: Done 12 out of 20 | elapsed: 8.3min remaining: 5.6min\n",
"[Parallel(n_jobs=6)]: Done 15 out of 20 | elapsed: 9.7min remaining: 3.2min\n",
"[Parallel(n_jobs=6)]: Done 18 out of 20 | elapsed: 16.0min remaining: 1.8min\n",
"[Parallel(n_jobs=6)]: Done 20 out of 20 | elapsed: 40.2min finished\n"
"[Parallel(n_jobs=6)]: Done 1 tasks | elapsed: 1.5min\n",
"[Parallel(n_jobs=6)]: Done 6 tasks | elapsed: 4.8min\n",
"[Parallel(n_jobs=6)]: Done 12 out of 20 | elapsed: 6.5min remaining: 4.3min\n",
"[Parallel(n_jobs=6)]: Done 15 out of 20 | elapsed: 7.6min remaining: 2.5min\n",
"[Parallel(n_jobs=6)]: Done 18 out of 20 | elapsed: 12.5min remaining: 1.4min\n",
"[Parallel(n_jobs=6)]: Done 20 out of 20 | elapsed: 28.8min finished\n"
]
},
{
@@ -2227,7 +2315,7 @@
},
{
"cell_type": "code",
"execution_count": 9,
"execution_count": 10,
"id": "513326a1",
"metadata": {},
"outputs": [
@@ -3781,7 +3869,7 @@
},
{
"cell_type": "code",
"execution_count": 10,
"execution_count": 11,
"id": "0347bafa",
"metadata": {},
"outputs": [
@@ -4206,7 +4294,7 @@
},
{
"cell_type": "code",
"execution_count": 11,
"execution_count": 12,
"id": "262adde2",
"metadata": {},
"outputs": [
@@ -4430,7 +4518,7 @@
},
{
"cell_type": "code",
"execution_count": 12,
"execution_count": 13,
"id": "06cb4bb4",
"metadata": {},
"outputs": [
@@ -4706,7 +4794,7 @@
},
{
"cell_type": "code",
"execution_count": 13,
"execution_count": 14,
"id": "669e1356",
"metadata": {},
"outputs": [
@@ -4778,7 +4866,7 @@
},
{
"cell_type": "code",
"execution_count": 14,
"execution_count": 15,
"id": "feab9afc",
"metadata": {},
"outputs": [
@@ -4874,7 +4962,7 @@
},
{
"cell_type": "code",
"execution_count": 15,
"execution_count": 16,
"id": "84fa1026",
"metadata": {},
"outputs": [
@@ -5042,7 +5130,7 @@
},
{
"cell_type": "code",
"execution_count": 16,
"execution_count": 17,
"id": "141a7ea5",
"metadata": {},
"outputs": [
@@ -5138,7 +5226,7 @@
},
{
"cell_type": "code",
"execution_count": 17,
"execution_count": 18,
"id": "d67cde11",
"metadata": {},
"outputs": [
@@ -5648,7 +5736,7 @@
},
{
"cell_type": "code",
"execution_count": 18,
"execution_count": 27,
"id": "cfd4fbc0",
"metadata": {},
"outputs": [
@@ -5701,7 +5789,7 @@
},
{
"cell_type": "code",
"execution_count": 19,
"execution_count": 28,
"id": "c5844b26",
"metadata": {},
"outputs": [
@@ -5740,7 +5828,7 @@
},
{
"cell_type": "code",
"execution_count": 20,
"execution_count": 29,
"id": "39361add",
"metadata": {},
"outputs": [
@@ -5810,7 +5898,7 @@
},
{
"cell_type": "code",
"execution_count": 21,
"execution_count": 22,
"id": "f534ac35",
"metadata": {},
"outputs": [
@@ -5860,7 +5948,7 @@
},
{
"cell_type": "code",
"execution_count": 22,
"execution_count": 30,
"id": "0a2bef50",
"metadata": {},
"outputs": [
@@ -5880,7 +5968,7 @@
"# ============================================================\n",
"if VERIFY_PIPELINE:\n",
" import pandas as pd\n",
" os.makedirs(\"reports\", exist_ok=True)\n",
" os.makedirs(\"pipeline_verification_bundle/reports\", exist_ok=True)\n",
" if candidates_path.exists():\n",
" # Extract the union of columns already used by exported candidates as the\n",
" # \"feature set\" fingerprint (kept generic - actual ML features are only\n",
@@ -5904,7 +5992,7 @@
},
{
"cell_type": "code",
"execution_count": 23,
"execution_count": 31,
"id": "60aed0a9",
"metadata": {},
"outputs": [
@@ -5947,14 +6035,27 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 32,
"id": "obs_finalize_v1",
"metadata": {
"tags": [
"pipeline_observability_v1"
]
},
"outputs": [],
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"[observability] hydrated from PipelineProfiler: {'hydrated': False}\n",
"[observability] WARNING: PipelineObservability.finalize() unavailable, using fallback result.\n",
"[manifest] wrote reports\\pipeline_manifest.json\n",
"\n",
"Observability artifacts written:\n",
" integrity_flags: {}\n"
]
}
],
"source": [
"# --- Pipeline Observability finalize (tag: pipeline_observability_v1) ----\n",
"# Hydrates from any existing PipelineProfiler traceability layer, then\n",
@@ -5988,7 +6089,15 @@
" all_pass = all(r['status'] == 'PASS' for r in _uuid_report.values())\n",
" _integrity['Model UUID Consistency'] = 'PASS' if all_pass else 'FAIL'\n",
"\n",
"_result = obs.finalize(integrity_flags=_integrity or None, verbose=True)\n",
"if hasattr(obs, \"finalize\"):\n",
" _result = obs.finalize(integrity_flags=_integrity or None, verbose=True)\n",
"else:\n",
" print(\"[observability] WARNING: PipelineObservability.finalize() unavailable, using fallback result.\")\n",
" _result = {\n",
" \"run_id\": getattr(obs, \"run_id\", None),\n",
" \"integrity_flags\": _integrity,\n",
" \"survival_gap_warnings\": None,\n",
" }\n",
"\n",
"# Phase 15 -- Strategy Tester writes the pipeline manifest for Explorer.\n",
"_hashes = {\n",
pipeline_verification_bundle/Strategy_Tester_fixed.orig.ipynb
-5916
View File
File diff suppressed because it is too large Load Diff
pipeline_verification_bundle/patches/patch_explorer_cell46.py
-260
View File
@@ -1,260 +0,0 @@
#!/usr/bin/env python3
"""
patch_explorer_cell46.py
Fixes the ModuleNotFoundError in the PIPELINE VERIFICATION import cell of
GoldRegimeX_Explorer.ipynb by making it auto-discover pipeline_verification.py
and shared/session_filter.py whether they sit in the project root or the
notebooks/ folder.
Usage:
python patch_explorer_cell46.py [path/to/GoldRegimeX_Explorer.ipynb]
- If no path is given, defaults to ./GoldRegimeX_Explorer.ipynb, then
./notebooks/GoldRegimeX_Explorer.ipynb.
- Writes a timestamped .bak next to the notebook before modifying.
- Locates the target cell by content signature (not by index), so it still
works even if you insert/remove cells above it.
- Idempotent: running again after a successful patch is a no-op.
- Validates every code cell with ast.parse before writing.
Do-not-modify list respected: no changes to Trend Pullback, TBM, CPCV,
HMM/XGBoost architecture, Risk Circuit Breaker, threshold optimisation,
or Grid Sensitivity Plateau logic. This only touches the verification
bootstrap cell.
"""
from __future__ import annotations
import argparse
import ast
import json
import shutil
import sys
from datetime import datetime
from pathlib import Path
# ---------------------------------------------------------------------------
# Target cell signature - substring that uniquely identifies the cell we edit.
# ---------------------------------------------------------------------------
CELL_SIGNATURE = "PIPELINE VERIFICATION (ADDED) - Global flag + imports"
# Marker used to detect an already-patched cell (idempotency).
PATCH_MARKER = "FIX (2026-07-07): auto-discover pipeline_verification.py"
# ---------------------------------------------------------------------------
# New cell source.
# ---------------------------------------------------------------------------
NEW_CELL_SRC = '''\
# ============================================================
# PIPELINE VERIFICATION (ADDED) - Global flag + imports
# Phases 1, 8. Purely additive - subsequent cells gated on
# VERIFY_PIPELINE. Does not modify CPCV, HMM/XGBoost architecture,
# Strategy logic, TBM, Grid Sensitivity Plateau, Risk Circuit Breaker,
# or threshold optimisation logic.
#
# FIX (2026-07-07): auto-discover pipeline_verification.py and
# shared/session_filter.py regardless of whether they sit in the
# project root or the notebooks/ folder. Previously this cell
# raised ModuleNotFoundError when the files were placed next to
# the .ipynb but Cell 1 had already chdir\'d to the project root.
# ============================================================
import sys, os, json, uuid, hashlib
from pathlib import Path
VERIFY_PIPELINE = True # set False to skip verification with zero runtime overhead
if VERIFY_PIPELINE:
# Candidate directories to search for pipeline_verification.py and shared/.
# Order matters: nearest to the notebook wins.
_cwd = Path.cwd().resolve()
_candidates = [
_cwd,
_cwd / "notebooks",
_cwd.parent,
_cwd.parent / "notebooks",
Path.home() / "GoldRegime_X",
Path.home() / "GoldRegime_X" / "notebooks",
]
_seen = set(); _search = []
for _p in _candidates:
_s = str(_p)
if _s not in _seen:
_seen.add(_s); _search.append(_p)
_pv_dir = None
_sf_dir = None
for _p in _search:
if _pv_dir is None and (_p / "pipeline_verification.py").is_file():
_pv_dir = _p
if _sf_dir is None and (_p / "shared" / "session_filter.py").is_file():
_sf_dir = _p
if _pv_dir is not None and _sf_dir is not None:
break
_missing = []
if _pv_dir is None:
_missing.append("pipeline_verification.py")
if _sf_dir is None:
_missing.append("shared/session_filter.py")
if _missing:
_tried = "\\n ".join(str(p) for p in _search)
raise ModuleNotFoundError(
"Could not locate: " + ", ".join(_missing) + ".\\n"
"Place them alongside the notebook (project root OR notebooks/ folder).\\n"
"Searched:\\n " + _tried
)
for _dir in {_pv_dir, _sf_dir}:
_s = str(_dir)
if _s not in sys.path:
sys.path.insert(0, _s)
# Ensure shared/ is importable as a package even if __init__.py is missing.
_shared_init = _sf_dir / "shared" / "__init__.py"
if not _shared_init.is_file():
try:
_shared_init.write_text("")
except Exception:
pass
from shared.session_filter import SessionFilter, run_session_filter_test_suite
import pipeline_verification as pv
from pipeline_verification import (
DatasetIntegrityVerifier,
FeatureLeakageVerifier,
EvaluationVerifier,
PredictionAlignmentVerifier,
SessionVerifier,
ThresholdVerifier,
CandidateIntegrityVerifier,
PipelineCertification,
RootCauseAnalyzerV2,
assign_model_uuid,
verify_model_identity,
assert_no_train_oos_leakage,
train_hash_of,
expected_calibration_error,
build_pipeline_waterfall,
build_manifest,
verify_manifest_match,
feature_set_hash,
candidate_ids_hash,
)
session_filter = SessionFilter()
certification = PipelineCertification()
print("VERIFY_PIPELINE=True.")
print(" pipeline_verification.py loaded from:", _pv_dir)
print(" shared/session_filter.py loaded from:", _sf_dir / "shared")
print(" SessionFilter hash:", session_filter.version_hash())
else:
print("VERIFY_PIPELINE=False. Skipping all verification cells.")
'''
def resolve_notebook_path(cli_arg: str | None) -> Path:
if cli_arg:
p = Path(cli_arg).expanduser().resolve()
if not p.is_file():
sys.exit(f"ERROR: notebook not found at {p}")
return p
for candidate in (
Path("GoldRegimeX_Explorer.ipynb"),
Path("notebooks/GoldRegimeX_Explorer.ipynb"),
):
if candidate.is_file():
return candidate.resolve()
sys.exit(
"ERROR: no notebook path given and none of the defaults exist:\n"
" ./GoldRegimeX_Explorer.ipynb\n"
" ./notebooks/GoldRegimeX_Explorer.ipynb"
)
def find_target_cell(nb: dict) -> int:
matches = []
for i, cell in enumerate(nb["cells"]):
if cell.get("cell_type") != "code":
continue
src = "".join(cell.get("source", []))
if CELL_SIGNATURE in src:
matches.append(i)
if not matches:
sys.exit(
"ERROR: could not find target cell. Signature not found:\n "
+ CELL_SIGNATURE
)
if len(matches) > 1:
sys.exit(
f"ERROR: signature matched {len(matches)} cells at indices {matches}. "
"Refusing to guess."
)
return matches[0]
def validate_all_code_cells(nb: dict) -> None:
failed = []
for i, cell in enumerate(nb["cells"]):
if cell.get("cell_type") != "code":
continue
src = "".join(cell.get("source", []))
try:
ast.parse(src)
except SyntaxError as e:
failed.append((i, str(e)))
if failed:
msg = "\n".join(f" cell {i}: {err}" for i, err in failed)
sys.exit(f"ERROR: syntax errors after patch:\n{msg}")
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("notebook", nargs="?", help="Path to Explorer notebook")
parser.add_argument(
"--no-backup", action="store_true", help="Skip writing a .bak file"
)
args = parser.parse_args()
nb_path = resolve_notebook_path(args.notebook)
print(f"Patching: {nb_path}")
with nb_path.open("r", encoding="utf-8") as f:
nb = json.load(f)
idx = find_target_cell(nb)
old_src = "".join(nb["cells"][idx].get("source", []))
if PATCH_MARKER in old_src:
print(f"Cell {idx} already patched (marker present). No-op.")
return
# Replace cell source.
new_lines = NEW_CELL_SRC.split("\n")
if new_lines and new_lines[-1] == "":
new_lines.pop()
nb["cells"][idx]["source"] = [line + "\n" for line in new_lines[:-1]] + [
new_lines[-1]
]
# Clear stale execution output so the diff is clean.
if "outputs" in nb["cells"][idx]:
nb["cells"][idx]["outputs"] = []
if "execution_count" in nb["cells"][idx]:
nb["cells"][idx]["execution_count"] = None
validate_all_code_cells(nb)
if not args.no_backup:
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
bak = nb_path.with_suffix(nb_path.suffix + f".bak_{ts}")
shutil.copy2(nb_path, bak)
print(f"Backup written: {bak}")
with nb_path.open("w", encoding="utf-8") as f:
json.dump(nb, f, indent=1)
f.write("\n")
print(f"Patched cell {idx} in place. {len(old_src)} chars -> {len(NEW_CELL_SRC)} chars.")
print("Reload the notebook in Jupyter (File -> Revert / close & reopen).")
if __name__ == "__main__":
main()
pipeline_verification_bundle/patches/patch_explorer_cell48.py
-184
View File
@@ -1,184 +0,0 @@
#!/usr/bin/env python3
"""
patch_explorer_cell48.py
Fixes NameError: name 'TIMEFRAMES' is not defined in Pipeline Verification
cell #3 (Phases 2 + 3: Dataset Integrity + Train/OOS leakage) of
GoldRegimeX_Explorer.ipynb.
Root cause:
That cell iterates `for tf in TIMEFRAMES:`, but TIMEFRAMES is defined in
an earlier setup cell (cell 2). If the user runs the verification cells
in isolation (kernel restart, then jumping straight to the verification
block), TIMEFRAMES is not in globals.
Fix:
Insert a tiny resolver at the top of the cell:
- If TIMEFRAMES is already defined -> use it (no-op path).
- Else if `pipeline` dict exists -> derive TIMEFRAMES from its keys
and expose it as a global so subsequent verification cells inherit
it (they all reference TIMEFRAMES too, so this fix propagates).
- Else raise a clear RuntimeError telling the user to run the
earlier pipeline-build cells first.
No other cells are touched. The do-not-modify list is respected.
Usage:
python patch_explorer_cell48.py [path/to/GoldRegimeX_Explorer.ipynb]
- If no path is given, defaults to ./GoldRegimeX_Explorer.ipynb, then
./notebooks/GoldRegimeX_Explorer.ipynb.
- Writes a timestamped .bak next to the notebook before modifying.
- Locates the target cell by content signature, not by index.
- Idempotent: running twice is a no-op the second time.
- Validates every code cell with ast.parse before writing.
"""
from __future__ import annotations
import argparse
import ast
import json
import shutil
import sys
from datetime import datetime
from pathlib import Path
CELL_SIGNATURE = "PIPELINE VERIFICATION - Phases 2 + 3: Dataset Integrity"
PATCH_MARKER = "FIX (2026-07-07-b): resolve TIMEFRAMES from pipeline"
# Anchor: the exact text we search for inside the cell.
OLD_ANCHOR = "if VERIFY_PIPELINE:\n integrity_verifier = DatasetIntegrityVerifier()"
# Replacement: same anchor with a resolver block injected between the `if`
# and `integrity_verifier = ...`. Indentation matches the existing cell.
NEW_BLOCK = (
"if VERIFY_PIPELINE:\n"
" # FIX (2026-07-07-b): resolve TIMEFRAMES from pipeline if not already\n"
" # defined. Lets verification cells run even after a kernel restart\n"
" # where the earlier constant-defining cells weren't re-executed.\n"
" if \"TIMEFRAMES\" not in globals():\n"
" if \"pipeline\" in globals() and isinstance(pipeline, dict) and pipeline:\n"
" TIMEFRAMES = list(pipeline.keys())\n"
" print(\"[verify] TIMEFRAMES inferred from pipeline:\", TIMEFRAMES)\n"
" else:\n"
" raise RuntimeError(\n"
" \"TIMEFRAMES is not defined and 'pipeline' dict is unavailable. \"\n"
" \"Run the earlier notebook cells that build the pipeline first.\"\n"
" )\n"
" integrity_verifier = DatasetIntegrityVerifier()"
)
def resolve_notebook_path(cli_arg):
if cli_arg:
p = Path(cli_arg).expanduser().resolve()
if not p.is_file():
sys.exit(f"ERROR: notebook not found at {p}")
return p
for candidate in (
Path("GoldRegimeX_Explorer.ipynb"),
Path("notebooks/GoldRegimeX_Explorer.ipynb"),
):
if candidate.is_file():
return candidate.resolve()
sys.exit(
"ERROR: no notebook path given and none of the defaults exist:\n"
" ./GoldRegimeX_Explorer.ipynb\n"
" ./notebooks/GoldRegimeX_Explorer.ipynb"
)
def find_target_cell(nb):
matches = []
for i, cell in enumerate(nb["cells"]):
if cell.get("cell_type") != "code":
continue
src = "".join(cell.get("source", []))
if CELL_SIGNATURE in src:
matches.append(i)
if not matches:
sys.exit(
"ERROR: could not find target cell. Signature not found:\n "
+ CELL_SIGNATURE
)
if len(matches) > 1:
sys.exit(
f"ERROR: signature matched {len(matches)} cells at indices {matches}. "
"Refusing to guess."
)
return matches[0]
def validate_all_code_cells(nb):
failed = []
for i, cell in enumerate(nb["cells"]):
if cell.get("cell_type") != "code":
continue
src = "".join(cell.get("source", []))
try:
ast.parse(src)
except SyntaxError as e:
failed.append((i, str(e)))
if failed:
msg = "\n".join(f" cell {i}: {err}" for i, err in failed)
sys.exit(f"ERROR: syntax errors after patch:\n{msg}")
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("notebook", nargs="?", help="Path to Explorer notebook")
parser.add_argument(
"--no-backup", action="store_true", help="Skip writing a .bak file"
)
args = parser.parse_args()
nb_path = resolve_notebook_path(args.notebook)
print(f"Patching: {nb_path}")
with nb_path.open("r", encoding="utf-8") as f:
nb = json.load(f)
idx = find_target_cell(nb)
old_src = "".join(nb["cells"][idx].get("source", []))
if PATCH_MARKER in old_src:
print(f"Cell {idx} already patched (marker present). No-op.")
return
if OLD_ANCHOR not in old_src:
sys.exit(
f"ERROR: expected anchor not found in cell {idx}. The cell may have\n"
"been hand-edited. Bailing out rather than guessing.\n"
"Expected anchor:\n" + OLD_ANCHOR
)
new_src = old_src.replace(OLD_ANCHOR, NEW_BLOCK, 1)
# Split back into ipynb source-lines format (each line keeps its trailing \n).
lines = new_src.split("\n")
if lines and lines[-1] == "":
lines.pop()
nb["cells"][idx]["source"] = [line + "\n" for line in lines[:-1]] + [lines[-1]]
if "outputs" in nb["cells"][idx]:
nb["cells"][idx]["outputs"] = []
if "execution_count" in nb["cells"][idx]:
nb["cells"][idx]["execution_count"] = None
validate_all_code_cells(nb)
if not args.no_backup:
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
bak = nb_path.with_suffix(nb_path.suffix + f".bak_{ts}")
shutil.copy2(nb_path, bak)
print(f"Backup written: {bak}")
with nb_path.open("w", encoding="utf-8") as f:
json.dump(nb, f, indent=1)
f.write("\n")
print(f"Patched cell {idx} in place. {len(old_src)} chars -> {len(new_src)} chars.")
print("Reload the notebook in Jupyter (File -> Revert / close & reopen).")
if __name__ == "__main__":
main()