"""GalaxyGrid stock indicators: hard stop at worksheet row 1000 (999 data rows). Load inside GalaxyGrid: Automation > Python > Load script from file > Run Python. Uses the selected stock worksheet. Requires a fresh project with no formulas. Writes J:U; source A:I is preserved. Native automation clears Undo history. Rolling periods include extended hours; VWAP is rolling, not session-reset. Use a single ticker in chronological order. No external packages required. A unique summary JSON is written to Data/BenchmarkResults (or project folder). """ ROW_LIMIT = 1000 if "book" not in globals() or "sid" not in globals(): raise RuntimeError("Load and run this file inside GalaxyGrid's Automation > Python editor.") from terragrid import __version__ from terragrid.formulas import address from terragrid.benchmark_log import stage from pathlib import Path from time import perf_counter from datetime import datetime, timezone import json, os, platform, uuid def install_stock_indicators(book, sid, progress): script_started = perf_counter() def report(message): print(message, flush=True) progress(message) for name in ("visible_value", "_meta", "_view_sql", "_rules", "fill_visible", "set_cell"): if not callable(getattr(book, name, None)): raise RuntimeError("This script needs GalaxyGrid's current native workbook API (tested on 0.14.1 and 0.14.2).") expected = {0: "timestamp", 1: "symbol", 5: "high", 6: "low", 7: "close", 8: "volume"} for col, label in expected.items(): actual = str(book.visible_value(sid, 1, col) or "").strip().casefold() if actual != label: raise ValueError("Select the stock sheet first: expected %s in %s1; found %r." % (label, chr(65 + col), actual)) if any(book._rules(sheet["id"]) for sheet in book.sheets()): raise ValueError("Use a fresh project/import with no existing formulas for each test. Do not run both variants on the same populated test project.") if ROW_LIMIT is None: report("Finding the final stock timestamp in column A...") meta = book._meta(sid) source = book._view_sql(sid, columns=[meta["columns"][0]]) last = int(book.db.execute( "SELECT coalesce(max(rid), 0) FROM (" + source + ") stock " "WHERE rid >= 2 AND c0 IS NOT NULL AND trim(c0) <> ''" ).fetchall()[0][0]) if last < 51: raise ValueError("At least 50 stock-data rows are needed, through worksheet row 51.") else: last = ROW_LIMIT if not str(book.visible_value(sid, last, 0) or "").strip(): raise ValueError("The source must contain data through worksheet row 1000. This variant stops at row 1000, including the header.") folder = Path(os.environ.get("TERRAGRID_DATA_ROOT", str(book.path))) / "BenchmarkResults" folder.mkdir(parents=True, exist_ok=True) variant = "full" if ROW_LIMIT is None else "row1000" stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:8] log_path = folder / ("stock-indicators-" + variant + "-" + stamp + ".json") result = dict(application="GalaxyGrid", version=__version__, variant=variant, started_utc=datetime.now(timezone.utc).isoformat(), platform=platform.platform(), python=platform.python_version(), worksheet=book.sheet(sid)["name"], last_worksheet_row=last, source_data_rows=last-1, row_limit=ROW_LIMIT, history_enabled=bool(getattr(book, "_history_enabled", False)), atomic_script=False, detailed_diagnostics_enabled=bool(getattr(book,"benchmark_log",None)), processor=getattr(book,"_processor_status",None), memory_budget=getattr(book,"_memory_budget_status",None), status="running", indicators=[]) def save_summary(): temporary = log_path.with_suffix(".json.tmp") temporary.write_text(json.dumps(result, indent=2), encoding="utf-8") temporary.replace(log_path) save_summary() report("Summary: " + str(log_path)) report("Installing 12 indicators through worksheet row %s (%s source rows)." % (f"{last:,}", f"{last-1:,}")) try: # Use the same automatic-calculation semantics in both variants. book.set_calculation_mode("automatic") result["calculation_mode"] = book.calculation_mode() headers = ["Minute Return", "Typical Price", "Price x Volume", "SMA 20", "SMA 50", "EMA 20", "True Range", "ATR 20 - Wilder", "Average Volume 20", "Rolling VWAP 20", "Bollinger Upper 20", "Bollinger Lower 20"] starts = {"J":3,"K":2,"L":2,"M":21,"N":51,"O":21,"P":3,"Q":22, "R":21,"S":21,"T":21,"U":21} book.ensure_columns(sid, 21, spare=True) for col, label in enumerate(headers, 9): book.set_cell(sid, 1, col, label) for letter, first in starts.items(): if first > 2: col = address(letter + "1")[1] book.clear_visible(sid, 2, first-1, col, col) jobs = [ ("True Range", "P3", "=MAX(F3-G3,ABS(F3-H2),ABS(G3-H2))", None), ("Minute Return", "J3", "=H3/H2-1", None), ("Typical Price", "K2", "=(F2+G2+H2)/3", None), ("Price x Volume", "L2", "=K2*I2", None), ("SMA 20", "M21", "=AVERAGE(H2:H21)", None), ("SMA 50", "N51", "=AVERAGE(H2:H51)", None), ("EMA 20", "O22", "=H22*(2/21)+O21*(19/21)", ("O21", "=AVERAGE(H2:H21)")), ("ATR 20 - Wilder", "Q23", "=(Q22*19+P23)/20", ("Q22", "=AVERAGE(P3:P22)")), ("Average Volume 20", "R21", "=AVERAGE(I2:I21)", None), ("Rolling VWAP 20", "S21", "=SUM(L2:L21)/SUM(I2:I21)", None), ("Bollinger Upper 20", "T21", "=M21+2*STDEV.P(H2:H21)", None), ("Bollinger Lower 20", "U21", "=M21-2*STDEV.P(H2:H21)", None), ] total_start = perf_counter() for index, (name, cell, formula, seed) in enumerate(jobs, 1): first, col = address(cell) report("[%d/%d] %s: %s through row %s..." % (index, len(jobs), name, cell, f"{last:,}")) result["active_indicator"] = dict(name=name, first_cell=cell, last_row=last) save_summary() started = perf_counter() with stage(book, "stock_indicator_test: " + name): if seed: seed_row, seed_col = address(seed[0]) book.set_cell(sid, seed_row, seed_col, seed[1]) book.fill_visible(sid, first, last, col, formula) elapsed = perf_counter() - started result["indicators"].append(dict(name=name, first_cell=cell, formula=formula, seed=seed, last_row=last, fill_rows=last-first+1, seconds=elapsed)) save_summary() report("Completed %s in %.3f seconds." % (name, elapsed)) result["indicator_fills_seconds"] = perf_counter() - total_start result.pop("active_indicator", None) result["status"] = "success" report("Finished all 12 indicators through row %s in %.3f seconds. Formulas are retained." % (f"{last:,}", result["indicator_fills_seconds"])) except BaseException as error: result["status"] = "cancelled" if type(error).__name__ == "OperationCancelled" else "error" result["error"] = type(error).__name__ + ": " + str(error) raise finally: result["script_seconds"] = perf_counter() - script_started result["finished_utc"] = datetime.now(timezone.utc).isoformat() result["processor_final"] = getattr(book,"_processor_status",None) result["memory_budget_final"] = getattr(book,"_memory_budget_status",None) save_summary() print("Summary saved: " + str(log_path), flush=True) install_stock_indicators(book, sid, globals().get("progress", lambda message: None))