From fb81111cdb692619cc9b79730ce9f06bda15866c Mon Sep 17 00:00:00 2001 From: Evan Parra Date: Tue, 7 Jul 2026 14:13:10 +0000 Subject: [PATCH] Priority 4 (MCP roadmap): option_minute_paths substrate + labeler outage fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Minute-path companion table (must-fix #6g / RM-002 / TF-14): - NEW profit_scout.option_minute_paths — per-minute option bars over each pool candidate's 3-trading-day excursion window. DDL + one-shot backfill (executed 2026-07-07: 144,448 bars / 2,675 contract-days back to 2026-04-13) + daily top-up via NEW forward-paper-trader POST /persist_minute_paths (reconciles the last 3 scan_dates each evening, DELETE-then-LOAD with the explicit schema, token-gated with POOL_LIQ_REFRESH_TOKEN, fully walled from the trader/labeler paths). - Unlocks MCP replay_contract + estimate_exit_rule exact first-crossing (heuristic_share 0.1045 -> 0.0) + trailing-rule scoring (gammarips-mcp PR #9). Classification: REALIZED TAPE — never a feature, never read by selection or the live trader (review-verified). PRODUCTION FIX folded in (discovered during this work): the labeler (/label_enriched_pool) had been failing since scan 2026-07-02 — the 07-02 enrichment autodetect landmine recurring in _write_shadow_records' staging load (all-NULL recommended_spread_pct autodetects to STRING vs the staging table's FLOAT). Load now binds the staging table's EXPLICIT schema; schema drift handled by Python-typed ADD COLUMNs instead of autodetect. RECOVERED: scans 07-01 (50 labeled) + 07-02 (50 labeled) re-run clean; substrate current again. See DECISIONS/2026-07-07-labeler-staging-autodetect-outage.md. gammarips-review: SHIP-WITH-FIXES, all applied (incl. incremental audit of the labeler fix: SHIP; atomicity preserved, existing columns type-pinned). Co-Authored-By: Claude Fable 5 --- docs/DATA-CONTRACTS.md | 6 + ...07-07-labeler-staging-autodetect-outage.md | 60 +++++ .../2026-07-07-option-minute-paths.md | 86 +++++++ forward-paper-trader/deploy.sh | 8 +- forward-paper-trader/main.py | 134 ++++++++++- forward-paper-trader/minute_paths.py | 225 ++++++++++++++++++ .../backfill_option_minute_paths.py | 200 ++++++++++++++++ .../create_option_minute_paths.py | 85 +++++++ 8 files changed, 795 insertions(+), 9 deletions(-) create mode 100644 docs/DECISIONS/2026-07-07-labeler-staging-autodetect-outage.md create mode 100644 docs/DECISIONS/2026-07-07-option-minute-paths.md create mode 100644 forward-paper-trader/minute_paths.py create mode 100644 scripts/ledger_and_tracking/backfill_option_minute_paths.py create mode 100644 scripts/ledger_and_tracking/create_option_minute_paths.py diff --git a/docs/DATA-CONTRACTS.md b/docs/DATA-CONTRACTS.md index 482a33f..a8d0c53 100644 --- a/docs/DATA-CONTRACTS.md +++ b/docs/DATA-CONTRACTS.md @@ -163,6 +163,12 @@ Every column belongs to exactly one group. The classification is written into th **Known data-quality caveats:** ~145 duplicate rows (a real finding, not a build blocker; `stg_enriched_option_outcomes` dedups to latest by `labeled_at`); ~27.7% of rows are `INVALID_LIQUIDITY` NULL-label (non-random illiquid tail — document the exclusion in any screen); pre-2026-06-11 daily counts are uneven. +## Minute paths — `profitscout-fida8.profit_scout.option_minute_paths` (added 2026-07-07) + +Per-minute option-premium OHLCV bars over each enriched-pool candidate's **3-trading-day excursion window** (`[entry_day .. exit_day_3d]`), one row per `(contract, entry_day, ts)` with `bar_date` (ET session) and `day_index` (1–3). This is the first-crossing substrate behind the MCP's `replay_contract` and `estimate_exit_rule`'s exact-resolution + trailing-rule scoring (must-fix #6g / RM-002 / TF-14). Writers: one-shot backfill (`scripts/ledger_and_tracking/backfill_option_minute_paths.py`, executed 2026-07-07) + `forward-paper-trader/minute_paths.py` via `POST /persist_minute_paths` (daily evening cron, reconciles the last 3 scan_dates DELETE-then-LOAD). **LOAD JOBS ONLY with the explicit schema — no streaming (the reconcile DELETE would hit buffers), no autodetect.** Partition `entry_day`, cluster `contract`. Schema source of truth: `scripts/ledger_and_tracking/create_option_minute_paths.py`. + +**Classification: REALIZED TAPE — never a feature.** Same class as `opp_*`: research/label substrate for CLOSED windows only. Never joined into `enriched_features_v1` or any as-of ≤ scan_date surface; never read by selection or the live trader. See `docs/DECISIONS/2026-07-07-option-minute-paths.md`. + ## Pool-liquidity telemetry — `profitscout-fida8.profit_scout.pool_liquidity_snapshot` (added 2026-07-07) Interval liquidity re-read of the **current enriched pool** (~50 contracts), one row per `(contract, as_of)`, written every ~10 minutes during RTH (plus one pre-open pass, `is_preopen=true`) by `signal-notifier/pool_liquidity.py` via `POST /refresh_pool_liquidity` (Cloud Scheduler `pool-liquidity-refresh`, `*/10 9-16 * * 1-5` ET). Consumed CACHE-FIRST by the gammarips-mcp `get_contract_snapshot` / `get_pool_liquidity` tools so an agent shortlist refreshes in one call at the ~10:00 ET decision window. See `docs/DECISIONS/2026-07-07-pool-liquidity-snapshot.md`. diff --git a/docs/DECISIONS/2026-07-07-labeler-staging-autodetect-outage.md b/docs/DECISIONS/2026-07-07-labeler-staging-autodetect-outage.md new file mode 100644 index 0000000..ce5481f --- /dev/null +++ b/docs/DECISIONS/2026-07-07-labeler-staging-autodetect-outage.md @@ -0,0 +1,60 @@ +# 2026-07-07 — Labeler staging autodetect outage (07-02→07-06) and fix + +## Status +Fixed 2026-07-07 (`forward-paper-trader/main.py:_write_shadow_records` step 2) +and RECOVERED: scans 2026-07-01 (50 labeled, 8W/37L) and 2026-07-02 +(50 labeled, 22W/26L) re-run locally through the fixed path; substrate current +again. Incremental gammarips-review: SHIP. + +## What happened +The daily `/label_enriched_pool` cron failed every evening from 2026-07-02 +through 2026-07-06 with: + +``` +400 Provided Schema does not match Table ..._stg_enriched_option_outcomes_... +Field recommended_spread_pct has changed type from FLOAT to STRING +``` + +`enriched_option_outcomes` silently stalled at scan 2026-06-30 — the paid +substrate's labels stopped accruing for four scan days (the failure DID +return 500s to the cron per must-fix #3a, but nobody was watching the pager). + +## Root cause — the 2026-07-02 landmine class, second location +This is the same autodetect-mistypes-all-NULL-columns bug that broke +enrichment on 2026-07-02 (`2026-07-02` decision / `3a04bee`), recurring in +the OTHER staged-write path. The atomic shadow writer (2026-07-01) created +its staging table correctly (`CREATE TABLE LIKE` the target → FLOAT), but the +load used `autodetect=True` + `ALLOW_FIELD_ADDITION`. When the 07-02 +enrichment fix made `recommended_spread_pct` properly all-NULL, every staged +JSONL load carried only `null`s for that field — BigQuery autodetect infers +an all-null field as STRING, proposed a FLOAT→STRING relaxation against the +typed staging table, and the load 500'd. The two fixes were siblings: the +07-02 fix removed autodetect from enrichment's staging load; this one removes +it from the labeler's. + +## Fix +`_write_shadow_records` step 2 now: +1. Diffs the row dicts' keys against the staging schema (which is LIKE-cloned + from the target, so existing columns keep live types). +2. For genuinely NEW keys, `ALTER TABLE ADD COLUMN` on STAGING with the type + derived from the PYTHON values (bool→BOOL before int→INT64 — bool is an + int subclass — float→FLOAT64, else STRING). Deterministic typing from code + we control, not BQ inference over serialized JSON. +3. Loads with `schema=` explicitly. No autodetect, no + ALLOW_FIELD_ADDITION. + +All atomicity guarantees unchanged (staging-only until the verified +transactional swap; failure = loud stall, target untouched). Applies to all +three `_write_shadow_records` callers (enriched outcomes, topscore shadow, +shadow intraday). + +## Rule (now enforced at both known locations) +**NEVER autodetect on a staged load, anywhere.** An all-NULL column — which +`recommended_spread_pct` permanently is on this data plan — autodetects to +STRING and poisons the load. If a third staged-write path is ever added, it +inherits this rule. + +## Watch item +The evening cron will label scan 2026-07-06 tonight (17:00 ET) — it needs the +FIXED code deployed on forward-paper-trader before then, or one more manual +re-run tomorrow. diff --git a/docs/DECISIONS/2026-07-07-option-minute-paths.md b/docs/DECISIONS/2026-07-07-option-minute-paths.md new file mode 100644 index 0000000..76aa695 --- /dev/null +++ b/docs/DECISIONS/2026-07-07-option-minute-paths.md @@ -0,0 +1,86 @@ +# 2026-07-07 — `option_minute_paths` companion table (must-fix #6g / MCP RM-002 + TF-14) + +## Status +Shipped 2026-07-07. Table created + FULL history backfilled (144,448 bars, +2,675 contract-days, entry days 2026-04-13 → 2026-07-01; 419 contract-days +legitimately empty — the no-print illiquid tail) + topped up through +2026-07-06. Daily top-up endpoint on forward-paper-trader +(`POST /persist_minute_paths`) + Cloud Scheduler `option-minute-paths-refresh`. + +## Context +The excursion substrate stored only MFE/MAE **extremes** (`opp_*`), so when a +bracket's target AND stop were both crossed inside the window, first-crossing +order was unrecoverable — `estimate_exit_rule` resolved those rows by a +minutes-to-extreme heuristic (~10.5% of classifications), and trailing rules +could not be scored at all. This was engine must-fix #6g (known-deferred) and +the trader harness's GAP-002. + +## Decision +Persist the minute tape the labeler already replays from: + +1. **`profit_scout.option_minute_paths`** — one row per (contract, entry_day, + ts): per-minute OHLCV premium bars over each enriched-pool candidate's + 3-trading-day excursion window. Partition `entry_day`, cluster `contract`. + DDL: `scripts/ledger_and_tracking/create_option_minute_paths.py`. + **LOAD JOBS ONLY with the explicit schema** (daily reconcile DELETEs by + scan_date — streaming buffers would block that; autodetect banned per the + 2026-07-02 outage rule). +2. **Backfill** — `scripts/ledger_and_tracking/backfill_option_minute_paths.py` + (executed 2026-07-07): one Polygon minute-aggs range call per labeled + (contract, entry_day) from `enriched_option_outcomes`. +3. **Daily top-up** — `forward-paper-trader/minute_paths.py` behind + `POST /persist_minute_paths`: each evening reconciles the last 3 + scan_dates' pools (window `[entry_day .. min(exit_day_3d, today)]`, + DELETE-then-LOAD per scan_date, idempotent). Pool comes from + `overnight_signals_enriched` so it is independent of labeler lag. Fully + walled: called only from its own endpoint; the trader/labeler paths never + import it. Token-gated with the same `POOL_LIQ_REFRESH_TOKEN` secret as + the notifier's pool-liquidity endpoint (scan_dates override refused + without the token). +4. **Cloud Scheduler `option-minute-paths-refresh`** — daily ~17:40 ET + weekdays (after the 17:00 labeler + 17:20 outcomes refresh), with the + `X-Refresh-Token` header. + +## What it unlocks (MCP) +- **`replay_contract(contract, date, target_pct?, stop_pct?)`** — per-session + minute path served table-first (upstream fallback for non-pool contracts), + with an optional exact first-crossing readout off the 10:00 ET anchor. +- **`estimate_exit_rule` exact resolution** — both-levels-crossed rows now + resolve by exact first touch (same-bar → STOP-first, the labeler's + pessimistic rule). Measured on the +40/−30 3d bracket: heuristic share + 0.1045 → **0.0** (212/212 ambiguous rows resolved; 90 TARGET / 122 STOP). +- **`estimate_exit_rule(rule='trailing')` (TF-14)** — bar-by-bar SQL replay + of a hard-stop + trailing-giveback (+optional activation) rule, horizons + same_day (day-1 bars ≤15:45 ET) and 3d. Conservative bar mechanics: trail + rides the PRIOR-bar peak; touch fills at min(bar open, level). The entry + anchor (`opp_entry_price`) already embeds the engine's entry-slippage fill; + no EXIT slippage is applied (results are entry-net, exit-gross). + +## Leakage classification +**REALIZED TAPE — never a feature.** Same class as the `opp_*` opportunity +surface: research/label substrate for CLOSED windows. Never joined into +`enriched_features_v1` or any as-of ≤ scan_date surface; never read by the +selection path or the live trader. Serving closed-window bars to agents is +data-vendor product (the historical tape, explicitly timestamped), consistent +with the opportunity-surface posture. + +## Recovery path +If a reconcile dies between its DELETE and LOAD on a window's LAST (day-3) +pass, that scan_date drops out of the next evening's last-3 set and stays +empty until manually re-run. Recovery: POST /persist_minute_paths with the +`X-Refresh-Token` header and `{"scan_dates": ["YYYY-MM-DD"]}` (token-gated +override, ≤10 dates). `--max-instances=1` + the daily cadence make the +overlap/double-run case remote. + +## Cost / blast radius +Backfill ≈ 3.1k Polygon calls (one-time). Daily ≈ 150 calls + ~6k rows. +Failure mode = stale minute paths (MCP replay falls back upstream; exit-rule +scoring reports coverage via `n_excluded_no_bars` / heuristic share) — never +a missed pick, never a broken label run. Kill switch = pause the Scheduler job. + +## Revisit when +- The pool cap or window length changes (row volume scales linearly). +- A real quote feed lands (bars could then carry NBBO context — new columns, + explicit schema change, never autodetect). +- Same-day arbitrary-bracket scoring is requested (the minute data now + supports it; deliberately not built in this pass). diff --git a/forward-paper-trader/deploy.sh b/forward-paper-trader/deploy.sh index 63e25d1..760e029 100755 --- a/forward-paper-trader/deploy.sh +++ b/forward-paper-trader/deploy.sh @@ -8,6 +8,12 @@ SERVICE_NAME="forward-paper-trader" echo "Deploying $SERVICE_NAME to Cloud Run in project $PROJECT_ID..." +# POOL_LIQ_REFRESH_TOKEN (2026-07-07): shared refresh token for the +# POST /persist_minute_paths research endpoint (X-Refresh-Token header; same +# secret the notifier's /refresh_pool_liquidity uses). MUST stay in +# --set-secrets — the flag REPLACES the secret set, so dropping it silently +# strips the mount on the next deploy. + gcloud run deploy $SERVICE_NAME \ --project=$PROJECT_ID \ --region=$REGION \ @@ -20,7 +26,7 @@ gcloud run deploy $SERVICE_NAME \ --min-instances=0 \ --max-instances=1 \ --set-env-vars="GCP_PROJECT_ID=$PROJECT_ID" \ - --set-secrets="POLYGON_API_KEY=POLYGON_API_KEY:latest" \ + --set-secrets="POLYGON_API_KEY=POLYGON_API_KEY:latest,POOL_LIQ_REFRESH_TOKEN=POOL_LIQ_REFRESH_TOKEN:latest" \ --service-account="firebase-adminsdk-fbsvc@$PROJECT_ID.iam.gserviceaccount.com" echo "Done!" \ No newline at end of file diff --git a/forward-paper-trader/main.py b/forward-paper-trader/main.py index f811f64..6f93444 100644 --- a/forward-paper-trader/main.py +++ b/forward-paper-trader/main.py @@ -1584,10 +1584,15 @@ def _write_shadow_records( committed — silently wiping that scan_date's rows with nothing to reload. New rows are now STAGED first, verified, then the target scan_date is replaced inside a single transaction that rolls back on any error, so the original rows - always survive a failure. autodetect=True keeps a newly-added feature column - from 500-ing the load; any new column is propagated onto the target BEFORE the - swap. When no new columns are present this is byte-for-byte equivalent to the - old delete-then-overwrite (scan_date fully replaced, no dups). + always survive a failure. Schema drift is handled WITHOUT autodetect + (2026-07-07 — autodetect inferred the all-NULL recommended_spread_pct as + STRING and 500'd every load, labeler down 07-02..07-06): new row keys are + diffed against the staging schema and ADDed as columns typed from the + Python values, then the load binds the staging table's explicit schema; + any new column is propagated onto the target BEFORE the swap. When no new + columns are present this is byte-for-byte equivalent to the old + delete-then-overwrite (scan_date fully replaced, no dups). NEVER re-add + autodetect here — see docs/DECISIONS/2026-07-07-labeler-staging-autodetect-outage.md. Mirrors _write_ledger_records' streaming-avoiding load-job pattern but takes an explicit ``table`` — deliberately NOT reusing _write_ledger_records (which @@ -1631,8 +1636,46 @@ def _write_shadow_records( ).result() try: - # 2) Load new rows into staging. autodetect=True + ALLOW_FIELD_ADDITION - # means a genuinely NEW field is ADDED to staging instead of 500-ing. + # 2) Load new rows into staging with the staging table's EXPLICIT + # schema — NEVER autodetect (2026-07-07 fix): with autodetect=True, + # an all-NULL column (recommended_spread_pct went permanently NULL + # with the 2026-07-02 enrichment fix) is inferred as STRING, which + # conflicts with the staging table's FLOAT and 500s EVERY load — + # the labeler was down 07-02..07-06 on exactly this (the same + # landmine class as the 07-02 enrichment outage, second location). + # Schema-drift tolerance (the reason autodetect was added 07-01) is + # preserved by diffing row keys against the staging schema and + # ADDing columns typed from the PYTHON values before the load — + # deterministic typing from code we control, not BQ inference over + # serialized JSON. + staging_schema = client.get_table(staging).schema + known_names = {f.name for f in staging_schema} + new_keys = sorted({k for r in rows for k in r} - known_names) + if new_keys: + + def _bq_type_of(key: str) -> str: + for r in rows: + v = r.get(key) + if v is None: + continue + if isinstance(v, bool): + return "BOOL" + if isinstance(v, int): + return "INT64" + if isinstance(v, float): + return "FLOAT64" + return "STRING" # str/date/datetime/etc — safe default + return "STRING" + + adds = ", ".join( + f"ADD COLUMN IF NOT EXISTS `{k}` {_bq_type_of(k)}" for k in new_keys + ) + client.query(f"ALTER TABLE `{staging}` {adds}").result() + staging_schema = client.get_table(staging).schema + logger.info( + f"shadow: staged {len(new_keys)} new drift column(s): {new_keys}" + ) + jsonl = "\n".join(json.dumps(r, default=str) for r in rows) load_job = client.load_table_from_file( io.BytesIO(jsonl.encode("utf-8")), @@ -1640,8 +1683,7 @@ def _write_shadow_records( job_config=bigquery.LoadJobConfig( write_disposition=bigquery.WriteDisposition.WRITE_APPEND, source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON, - schema_update_options=[bigquery.SchemaUpdateOption.ALLOW_FIELD_ADDITION], - autodetect=True, + schema=staging_schema, ), ) load_job.result() # raises on load failure -> target still untouched @@ -2640,6 +2682,82 @@ def trigger_paper_trading(): return jsonify({"status": "error", "message": str(e)}), 500 +@app.route("/persist_minute_paths", methods=["POST"]) +def trigger_persist_minute_paths(): + """RESEARCH-ONLY minute-path top-up (MCP Priority-4 / must-fix #6g). + + Reconciles `option_minute_paths` for the last 3 scan_dates' enriched + pools (each window [entry_day .. min(exit_day_3d, today)]); by a window's + third session it is complete and drops out of the set. Hit by the daily + `option-minute-paths-refresh` Cloud Scheduler cron in the evening, after + the labeler. Fully independent of the trader/labeler paths — fetch + + write logic lives in minute_paths.py behind a hard leakage wall (realized + tape; never a feature; never read by selection or the live trader). + + Token-gated like the notifier's pool-liquidity endpoint: when + POOL_LIQ_REFRESH_TOKEN is set, every call must send X-Refresh-Token, and + the scan_dates override is refused without it (fail-closed while services + are public; see docs/DECISIONS/2026-07-02-service-auth-hardening.md). + Body (optional, token-gated): {"scan_dates": ["YYYY-MM-DD", ...]}. + """ + import hmac as _hmac + + import minute_paths + + expected_token = os.environ.get("POOL_LIQ_REFRESH_TOKEN", "").strip() + provided_token = request.headers.get("X-Refresh-Token", "") + token_authed = bool(expected_token) and _hmac.compare_digest( + provided_token, expected_token + ) + if expected_token and not token_authed: + return jsonify({"status": "denied"}), 403 + + req = request.get_json(silent=True) or {} + if req.get("scan_dates") and not token_authed: + return jsonify( + {"status": "denied", "reason": "scan_dates override requires X-Refresh-Token"} + ), 403 + + try: + today_et = datetime.now(est).date() + if req.get("scan_dates"): + try: + scan_dates = [ + datetime.strptime(str(sd), "%Y-%m-%d").date() + for sd in req["scan_dates"] + ][:10] + except ValueError: + return jsonify({"status": "error", "reason": "scan_dates must be YYYY-MM-DD"}), 400 + else: + # the last 3 trading days strictly before today — the scan_dates + # whose 3-session excursion windows are still accreting bars + scan_dates = [] + d = today_et - timedelta(days=1) + while len(scan_dates) < 3 and d > today_et - timedelta(days=15): + if is_trading_day(d): + scan_dates.append(d) + d -= timedelta(days=1) + + windows = [] + for sd in scan_dates: + entry_day = get_next_trading_day(sd) + if entry_day > today_et: + continue + window_end = min(get_nth_next_trading_day(entry_day, 2), today_et) + windows.append( + {"scan_date": sd, "entry_day": entry_day, "window_end": window_end} + ) + if not windows: + return jsonify({"status": "skipped", "reason": "no closed windows to persist"}), 200 + + summary = minute_paths.persist_minute_paths(windows) + code = 200 if summary.get("status") in ("success", "partial") else 500 + return jsonify(summary), code + except Exception as e: + logger.error(f"Error in persist_minute_paths endpoint: {e}") + return jsonify({"status": "error", "message": str(e)}), 500 + + @app.route("/label_enriched_pool", methods=["POST", "GET"]) def trigger_label_enriched_pool(): """RESEARCH-ONLY counterfactual labeling endpoint. diff --git a/forward-paper-trader/minute_paths.py b/forward-paper-trader/minute_paths.py new file mode 100644 index 0000000..69699f1 --- /dev/null +++ b/forward-paper-trader/minute_paths.py @@ -0,0 +1,225 @@ +"""Daily option_minute_paths top-up (MCP Priority-4 / RM-002 / must-fix #6g). + +Each evening (after the labeler), POST /persist_minute_paths reconciles the +minute-bar excursion paths for the last 3 scan_dates' enriched pools into +`profit_scout.option_minute_paths`: one Polygon minute-aggs range call per +(contract, scan_date), covering [entry_day .. min(exit_day_3d, today)]. By a +window's third session it is complete and drops out of the reconcile set. + +The pool comes from overnight_signals_enriched (available the same evening, +independent of labeler lag). Writes are idempotent per scan_date: +DELETE-then-LOAD via EXPLICIT-SCHEMA load jobs — never streaming (the DELETE +would hit streaming buffers), never autodetect (2026-07-02 outage rule). + +LEAKAGE WALL: these are realized post-entry bars (the excursion tape) — same +class as the opp_* opportunity surface. Research/label substrate ONLY: never +a feature, never joined into enriched_features_v1, never read by the +selection path or the live trader. This module is called ONLY from its own +endpoint; run_forward_paper_trading / run_label_enriched_pool never import +it. History back to 2026-04-13 was loaded by +scripts/ledger_and_tracking/backfill_option_minute_paths.py (2026-07-07). +""" + +import logging +import os +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import UTC, date, datetime +from zoneinfo import ZoneInfo + +import requests +from google.cloud import bigquery + +logger = logging.getLogger(__name__) + +PROJECT_ID = "profitscout-fida8" +TABLE_ID = f"{PROJECT_ID}.profit_scout.option_minute_paths" + +FETCH_TIMEOUT_S = int(os.environ.get("MINUTE_PATHS_FETCH_TIMEOUT_S", "15")) +MAX_WORKERS = int(os.environ.get("MINUTE_PATHS_MAX_WORKERS", "16")) + +_ET = ZoneInfo("America/New_York") + +SCHEMA = [ + bigquery.SchemaField("scan_date", "DATE", mode="REQUIRED"), + bigquery.SchemaField("entry_day", "DATE", mode="REQUIRED"), + bigquery.SchemaField("contract", "STRING", mode="REQUIRED"), + bigquery.SchemaField("ticker", "STRING"), + bigquery.SchemaField("ts", "TIMESTAMP", mode="REQUIRED"), + bigquery.SchemaField("bar_date", "DATE"), + bigquery.SchemaField("day_index", "INTEGER"), + bigquery.SchemaField("open", "FLOAT"), + bigquery.SchemaField("high", "FLOAT"), + bigquery.SchemaField("low", "FLOAT"), + bigquery.SchemaField("close", "FLOAT"), + bigquery.SchemaField("volume", "INTEGER"), + bigquery.SchemaField("vwap", "FLOAT"), + bigquery.SchemaField("transactions", "INTEGER"), + bigquery.SchemaField("source", "STRING"), + bigquery.SchemaField("ingested_at", "TIMESTAMP"), +] + + +def _fetch_minute_bars(contract: str, start: str, end: str, api_key: str) -> list[dict]: + """One range fetch, bearer-header key (never in the URL), 429-aware, + fail-soft to [].""" + url = f"https://api.polygon.io/v2/aggs/ticker/{contract}/range/1/minute/{start}/{end}" + params = {"adjusted": "true", "sort": "asc", "limit": 50000} + for attempt in range(3): + try: + resp = requests.get( + url, params=params, + headers={"Authorization": f"Bearer {api_key}"}, + timeout=FETCH_TIMEOUT_S, + ) + if resp.status_code == 429: + time.sleep(2 * (attempt + 1)) + continue + resp.raise_for_status() + return (resp.json() or {}).get("results") or [] + except Exception as e: # noqa: BLE001 + logger.warning(f"minute-paths fetch {contract} attempt {attempt+1}: {e}") + time.sleep(1) + return [] + + +def _bars_to_rows( + scan_date: date, entry_day: date, contract: str, ticker: str, + bars: list[dict], ingested_at: str, +) -> list[dict]: + """Keep bars from the first 3 distinct ET session dates on/after entry_day + (the excursion window), tagging each with day_index 1..3.""" + rows: list[dict] = [] + session_dates: list[str] = [] + entry_s = entry_day.isoformat() + for b in bars: + try: + dt = datetime.fromtimestamp(int(b["t"]) / 1e3, tz=UTC) + except (KeyError, TypeError, ValueError, OSError): + continue + bar_date = dt.astimezone(_ET).date().isoformat() + if bar_date < entry_s: + continue + if bar_date not in session_dates: + if len(session_dates) >= 3: + continue + session_dates.append(bar_date) + rows.append( + { + "scan_date": scan_date.isoformat(), + "entry_day": entry_s, + "contract": contract, + "ticker": ticker, + "ts": dt.isoformat(), + "bar_date": bar_date, + "day_index": session_dates.index(bar_date) + 1, + "open": b.get("o"), + "high": b.get("h"), + "low": b.get("l"), + "close": b.get("c"), + "volume": int(b["v"]) if b.get("v") is not None else None, + "vwap": b.get("vw"), + "transactions": b.get("n"), + "source": "polygon minute aggs (daily persist)", + "ingested_at": ingested_at, + } + ) + return rows + + +def _fetch_pool(client: bigquery.Client, scan_date: date) -> list[tuple[str, str]]: + q = f""" + SELECT DISTINCT ticker, recommended_contract + FROM `{PROJECT_ID}.profit_scout.overnight_signals_enriched` + WHERE DATE(scan_date) = @sd AND recommended_contract IS NOT NULL + """ + job = client.query( + q, + job_config=bigquery.QueryJobConfig( + query_parameters=[bigquery.ScalarQueryParameter("sd", "DATE", scan_date)] + ), + ) + return [(str(r.ticker), str(r.recommended_contract)) for r in job.result()] + + +def persist_minute_paths(windows: list[dict]) -> dict: + """Reconcile minute paths for the given windows. Each window dict: + {"scan_date": date, "entry_day": date, "window_end": date} (window_end + already clamped to today by the caller). Returns a summary dict; never + raises past the top-level try.""" + api_key = (os.environ.get("POLYGON_API_KEY") or "").strip() + if not api_key: + return {"status": "error", "reason": "POLYGON_API_KEY not set"} + + client = bigquery.Client(project=PROJECT_ID) + ingested_at = datetime.now(UTC).isoformat() + summary: dict = {"status": "success", "windows": []} + + for w in windows: + scan_date: date = w["scan_date"] + entry_day: date = w["entry_day"] + window_end: date = w["window_end"] + try: + pool = _fetch_pool(client, scan_date) + if not pool: + summary["windows"].append( + {"scan_date": scan_date.isoformat(), "skipped": "empty pool"} + ) + continue + + rows: list[dict] = [] + with ThreadPoolExecutor(max_workers=max(1, min(MAX_WORKERS, len(pool)))) as ex: + futs = { + ex.submit( + _fetch_minute_bars, c, + entry_day.isoformat(), window_end.isoformat(), api_key, + ): (t, c) + for (t, c) in pool + } + for fut in as_completed(futs): + t, c = futs[fut] + try: + bars = fut.result() + except Exception as e: # noqa: BLE001 + logger.warning(f"minute-paths future {c}: {e}") + bars = [] + rows.extend(_bars_to_rows(scan_date, entry_day, c, t, bars, ingested_at)) + + # Idempotent reconcile: replace this scan_date's rows wholesale. + client.query( + f"DELETE FROM `{TABLE_ID}` WHERE scan_date = @sd", + job_config=bigquery.QueryJobConfig( + query_parameters=[ + bigquery.ScalarQueryParameter("sd", "DATE", scan_date) + ] + ), + ).result() + if rows: + job_config = bigquery.LoadJobConfig( + schema=SCHEMA, + write_disposition=bigquery.WriteDisposition.WRITE_APPEND, + source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON, + ) + client.load_table_from_json(rows, TABLE_ID, job_config=job_config).result() + + summary["windows"].append( + { + "scan_date": scan_date.isoformat(), + "entry_day": entry_day.isoformat(), + "window_end": window_end.isoformat(), + "contracts": len(pool), + "bars": len(rows), + } + ) + logger.info( + f"minute-paths: {scan_date} reconciled — {len(pool)} contracts, " + f"{len(rows)} bars through {window_end}" + ) + except Exception as e: # noqa: BLE001 + logger.error(f"minute-paths window {scan_date} failed: {e}") + summary["status"] = "partial" + summary["windows"].append( + {"scan_date": scan_date.isoformat(), "error": str(e)[:200]} + ) + + return summary diff --git a/scripts/ledger_and_tracking/backfill_option_minute_paths.py b/scripts/ledger_and_tracking/backfill_option_minute_paths.py new file mode 100644 index 0000000..5b0624f --- /dev/null +++ b/scripts/ledger_and_tracking/backfill_option_minute_paths.py @@ -0,0 +1,200 @@ +"""One-shot backfill of option_minute_paths from enriched_option_outcomes. + +For every labeled (contract, entry_day) in the outcomes substrate, fetches +the per-minute option bars over the 3-trading-day excursion window +[entry_day .. exit_day_3d] from Polygon minute aggregates (one range call +per contract-day) and loads them into +`profit_scout.option_minute_paths` via EXPLICIT-SCHEMA load jobs (no +streaming, no autodetect). Executed 2026-07-07 (~3.1k contract-days). + +Idempotent per scan_date batch: each batch DELETEs its scan_dates' rows +before loading, so a re-run cannot duplicate. Safe to re-run for a date +range with --from/--to. + +Usage: + PROJECT_ID=profitscout-fida8 POLYGON_API_KEY=... \ + python scripts/ledger_and_tracking/backfill_option_minute_paths.py [--from YYYY-MM-DD] [--to YYYY-MM-DD] +""" + +import argparse +import os +import sys +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import UTC, datetime, timedelta +from zoneinfo import ZoneInfo + +import requests +from google.cloud import bigquery + +PROJECT_ID = "profitscout-fida8" +TABLE_ID = f"{PROJECT_ID}.profit_scout.option_minute_paths" +OUTCOMES = f"`{PROJECT_ID}.profit_scout.enriched_option_outcomes`" +ET = ZoneInfo("America/New_York") +API_KEY = (os.environ.get("POLYGON_API_KEY") or "").strip() +WORKERS = int(os.environ.get("BACKFILL_WORKERS", "16")) + +SCHEMA = [ + bigquery.SchemaField("scan_date", "DATE", mode="REQUIRED"), + bigquery.SchemaField("entry_day", "DATE", mode="REQUIRED"), + bigquery.SchemaField("contract", "STRING", mode="REQUIRED"), + bigquery.SchemaField("ticker", "STRING"), + bigquery.SchemaField("ts", "TIMESTAMP", mode="REQUIRED"), + bigquery.SchemaField("bar_date", "DATE"), + bigquery.SchemaField("day_index", "INTEGER"), + bigquery.SchemaField("open", "FLOAT"), + bigquery.SchemaField("high", "FLOAT"), + bigquery.SchemaField("low", "FLOAT"), + bigquery.SchemaField("close", "FLOAT"), + bigquery.SchemaField("volume", "INTEGER"), + bigquery.SchemaField("vwap", "FLOAT"), + bigquery.SchemaField("transactions", "INTEGER"), + bigquery.SchemaField("source", "STRING"), + bigquery.SchemaField("ingested_at", "TIMESTAMP"), +] + + +def fetch_minute_bars(contract: str, start: str, end: str) -> list[dict]: + url = ( + f"https://api.polygon.io/v2/aggs/ticker/{contract}" + f"/range/1/minute/{start}/{end}" + ) + params = {"adjusted": "true", "sort": "asc", "limit": 50000} + for attempt in range(3): + try: + resp = requests.get( + url, params=params, + headers={"Authorization": f"Bearer {API_KEY}"}, timeout=15, + ) + if resp.status_code == 429: + time.sleep(2 * (attempt + 1)) + continue + resp.raise_for_status() + return (resp.json() or {}).get("results") or [] + except Exception as e: # noqa: BLE001 + print(f" warn: {contract} fetch attempt {attempt+1}: {e}", file=sys.stderr) + time.sleep(1) + return [] + + +def bars_to_rows(task: dict, bars: list[dict], ingested_at: str) -> list[dict]: + """Keep bars from the first 3 distinct ET session dates on/after entry_day.""" + rows = [] + session_dates: list[str] = [] + for b in bars: + try: + dt = datetime.fromtimestamp(int(b["t"]) / 1e3, tz=UTC) + except (KeyError, TypeError, ValueError, OSError): + continue + bar_date = dt.astimezone(ET).date().isoformat() + if bar_date < task["entry_day"]: + continue + if bar_date not in session_dates: + if len(session_dates) >= 3: + continue + session_dates.append(bar_date) + rows.append( + { + "scan_date": task["scan_date"], + "entry_day": task["entry_day"], + "contract": task["contract"], + "ticker": task["ticker"], + "ts": dt.isoformat(), + "bar_date": bar_date, + "day_index": session_dates.index(bar_date) + 1, + "open": b.get("o"), + "high": b.get("h"), + "low": b.get("l"), + "close": b.get("c"), + "volume": int(b["v"]) if b.get("v") is not None else None, + "vwap": b.get("vw"), + "transactions": b.get("n"), + "source": "polygon minute aggs (backfill 2026-07-07)", + "ingested_at": ingested_at, + } + ) + return rows + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--from", dest="dfrom", default=None) + ap.add_argument("--to", dest="dto", default=None) + args = ap.parse_args() + + if not API_KEY: + sys.exit("POLYGON_API_KEY not set") + + client = bigquery.Client(project=PROJECT_ID) + where = ["recommended_contract IS NOT NULL", "entry_day IS NOT NULL"] + if args.dfrom: + where.append(f"scan_date >= '{args.dfrom}'") + if args.dto: + where.append(f"scan_date <= '{args.dto}'") + q = f""" + SELECT DISTINCT CAST(scan_date AS STRING) AS scan_date, + CAST(entry_day AS STRING) AS entry_day, + recommended_contract AS contract, ticker, + CAST(COALESCE(exit_day_3d, + DATE_ADD(entry_day, INTERVAL 6 DAY)) AS STRING) AS window_end + FROM {OUTCOMES} + WHERE {' AND '.join(where)} + ORDER BY scan_date, contract + """ + tasks = [dict(r) for r in client.query(q).result()] + print(f"{len(tasks)} contract-days to backfill") + + ingested_at = datetime.now(UTC).isoformat() + all_rows: list[dict] = [] + n_empty = 0 + with ThreadPoolExecutor(max_workers=WORKERS) as ex: + futs = { + ex.submit( + fetch_minute_bars, t["contract"], t["entry_day"], t["window_end"] + ): t + for t in tasks + } + done = 0 + for fut in as_completed(futs): + t = futs[fut] + try: + bars = fut.result() + except Exception as e: # noqa: BLE001 + print(f" warn: {t['contract']} future: {e}", file=sys.stderr) + bars = [] + rows = bars_to_rows(t, bars, ingested_at) + if not rows: + n_empty += 1 + all_rows.extend(rows) + done += 1 + if done % 250 == 0: + print(f" fetched {done}/{len(tasks)} ({len(all_rows)} bars so far)") + + print(f"fetch complete: {len(all_rows)} bars, {n_empty} empty contract-days") + + # Idempotency: clear the scan_dates we are about to load. + scan_dates = sorted({t["scan_date"] for t in tasks}) + client.query( + f"DELETE FROM `{TABLE_ID}` WHERE scan_date IN UNNEST(@sds)", + job_config=bigquery.QueryJobConfig( + query_parameters=[bigquery.ArrayQueryParameter("sds", "DATE", scan_dates)] + ), + ).result() + + job_config = bigquery.LoadJobConfig( + schema=SCHEMA, + write_disposition=bigquery.WriteDisposition.WRITE_APPEND, + source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON, + ) + CHUNK = 100_000 + for i in range(0, len(all_rows), CHUNK): + chunk = all_rows[i : i + CHUNK] + job = client.load_table_from_json(chunk, TABLE_ID, job_config=job_config) + job.result() + print(f" loaded rows {i}..{i + len(chunk)}") + + print("DONE") + + +if __name__ == "__main__": + main() diff --git a/scripts/ledger_and_tracking/create_option_minute_paths.py b/scripts/ledger_and_tracking/create_option_minute_paths.py new file mode 100644 index 0000000..743f9fd --- /dev/null +++ b/scripts/ledger_and_tracking/create_option_minute_paths.py @@ -0,0 +1,85 @@ +"""Create the option_minute_paths companion table (MCP Priority-4 / RM-002 / +engine must-fix #6g — the known-deferred minute-path table). + +One row per (contract, entry_day, ts): the per-minute option-premium bars +over each enriched-pool candidate's 3-trading-day excursion window +[entry_day .. exit_day_3d]. This is the FIRST-CROSSING substrate the +distributional exit tools were missing — with it, `estimate_exit_rule` can +resolve both-levels-crossed rows by EXACT first touch instead of the +minutes-to-extreme heuristic, and can score TRAILING rules (TF-14); the MCP +`replay_contract` primitive serves the same rows for per-contract replay. + +Writers: + * scripts/ledger_and_tracking/backfill_option_minute_paths.py (one-shot + history backfill from enriched_option_outcomes; executed 2026-07-07) + * forward-paper-trader POST /persist_minute_paths (daily top-up cron; each + run reconciles the last 3 scan_dates so day-2/day-3 bars land as their + sessions close) +Both write via LOAD JOBS with this EXPLICIT schema — never streaming into +this table (daily reconcile DELETEs by scan_date, which streaming buffers +block), never autodetect (2026-07-02 outage rule). + +LEAKAGE CLASSIFICATION — REALIZED TAPE, NEVER A FEATURE +-------------------------------------------------------- +Every bar here is realized post-entry market data (the excursion window +starts at the 10:00 ET entry). Same class as the opp_* opportunity surface: +research/label substrate for CLOSED windows. It must never be joined into +enriched_features_v1 or any as-of <= scan_date feature surface, and never +consulted by the selection path. Serving CLOSED-window bars to agents is +fine (it is the historical tape, explicitly timestamped). + +Partitioned by entry_day (window-scoped reads + age-out), clustered by +contract (the replay lookup). Idempotent (exists_ok=True). + +Run once (safe isolated infra, NOT a deploy): + PROJECT_ID=profitscout-fida8 python scripts/ledger_and_tracking/create_option_minute_paths.py +""" + +from google.cloud import bigquery + +PROJECT_ID = "profitscout-fida8" +DATASET_ID = "profit_scout" +TABLE_ID = "option_minute_paths" + +client = bigquery.Client(project=PROJECT_ID) +table_ref = f"{PROJECT_ID}.{DATASET_ID}.{TABLE_ID}" + +schema = [ + # -- identity / join keys --------------------------------------------- + bigquery.SchemaField("scan_date", "DATE", mode="REQUIRED"), + bigquery.SchemaField("entry_day", "DATE", mode="REQUIRED"), + bigquery.SchemaField("contract", "STRING", mode="REQUIRED"), + bigquery.SchemaField("ticker", "STRING", mode="NULLABLE"), + # -- the bar ------------------------------------------------------------ + bigquery.SchemaField("ts", "TIMESTAMP", mode="REQUIRED"), # bar start (UTC) + bigquery.SchemaField("bar_date", "DATE", mode="NULLABLE"), # ET session date of the bar + bigquery.SchemaField("day_index", "INTEGER", mode="NULLABLE"), # 1..3 within the excursion window + bigquery.SchemaField("open", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("high", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("low", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("close", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("volume", "INTEGER", mode="NULLABLE"), + bigquery.SchemaField("vwap", "FLOAT", mode="NULLABLE"), + bigquery.SchemaField("transactions", "INTEGER", mode="NULLABLE"), + # -- provenance ----------------------------------------------------------- + bigquery.SchemaField("source", "STRING", mode="NULLABLE"), + bigquery.SchemaField("ingested_at", "TIMESTAMP", mode="NULLABLE"), +] + +table = bigquery.Table(table_ref, schema=schema) +table.time_partitioning = bigquery.TimePartitioning( + type_=bigquery.TimePartitioningType.DAY, field="entry_day" +) +table.clustering_fields = ["contract"] +table.description = ( + "Per-minute option-premium bars over each enriched-pool candidate's " + "3-trading-day excursion window (engine must-fix #6g / MCP RM-002). " + "REALIZED TAPE — research/label substrate for CLOSED windows only; " + "NEVER a feature, never joined into enriched_features_v1, never read by " + "the selection path. Load-jobs only (no streaming, no autodetect). See " + "docs/DECISIONS/2026-07-07-option-minute-paths.md." +) + +created = client.create_table(table, exists_ok=True) +print(f"OK: {created.full_table_id} (partition=entry_day, cluster=contract)") +print(f"Columns: {[f.name for f in created.schema]}")