#!/usr/bin/env python3 """ orb_volume_surge_scout.py — opening-range breakout on volume scout (SCOUT SCRIPT CONTRACT v1, KIND: orb_volume_surge). Brief: Watch liquid NIFTY-100 F&O names for opening-range breakouts confirmed by volume running at least 2x the average, checked every 15 minutes during the first hour of the session (09:15-11:00 IST). Wake the owner when a clean breakout with volume confirmation fires. Wire routing (per sensor-catalog.md): the TradingView scanner (Wire 1) has no true "opening range" field — it is a cross-sectional snapshot endpoint, not an intraday time-series one. The opening range (the high/low of the first 09:15-09:30 IST bar) can only be built from real intraday candles, which is Wire 3 (Dhan charts intraday). This scout therefore: 1. Resolves each configured NSE equity symbol to its Dhan security id via the instrument master (Wire 4) — NSE_EQ / EQ rows, cached once per run. 2. Pulls today's 15-minute intraday bars for each symbol (Wire 3, /v2/charts/intraday) from session open through "now". 3. Opening range = high/low of the FIRST bar of the session (09:15-09:30 IST). A later bar's close breaking that range, on that bar's volume >= vol_multiplier x the trailing average of the opening bars seen so far, is the fire condition. Auth: reads DHAN_ACCESS_TOKEN / DHAN_CLIENT_ID from env, same as Wires 2/3/5. A scout with no eyes is SILENT, not crashed: on missing/expired token this exits 0 with fired=false and state noting 'no_credentials'. Config dials (SCOUT_CONFIG_JSON): universe (default a liquid NIFTY-100 F&O basket, see DEF_UNIVERSE) — list of NSE equity trading symbols to watch. vol_multiplier (default 2.0) — breakout bar's volume must be >= this many times the average volume of the opening-range bars seen so far this session. session_start (default "09:15") — IST HH:MM opening-range bar start. session_end_check (default "11:00") — IST HH:MM last time this scout should still be actively checking (informational; the cadence itself is owned by the scheduler, not this script — this is only used to skip quietly if called outside the window). interval (default "15") — Dhan intraday bar interval in minutes. Fires once per (symbol, direction, day) — state dedup so a breakout that is still standing on the next 15-min check does not re-wake the owner all session; a fresh breakout on a *different* name, or the *other* direction on the same name, can still fire independently the same day. """ import csv import io import json import os import sys from datetime import datetime, timedelta, timezone import requests # Dhan intraday bar timestamps are IST-based (same gotcha as daily bars) — # always convert through a fixed +05:30 offset, never bare UTC. IST = timezone(timedelta(hours=5, minutes=30)) SCRIP_MASTER_URL = "https://images.dhan.co/api-data/api-scrip-master.csv" INTRADAY_URL = "https://api.dhan.co/v2/charts/intraday" # A liquid NIFTY-100 F&O basket — config-overridable, never hardcoded logic, # just a sane default universe (catalog gotcha: no reliable server-side # "is F&O" filter exists, so an explicit list is the honest approach). DEF_UNIVERSE = [ "RELIANCE", "HDFCBANK", "ICICIBANK", "INFY", "TCS", "SBIN", "AXISBANK", "KOTAKBANK", "LT", "ITC", "BHARTIARTL", "HINDUNILVR", "BAJFINANCE", "MARUTI", "TATAMOTORS", "TATASTEEL", "SUNPHARMA", "ULTRACEMCO", "TITAN", "ADANIENT", ] DEF_VOL_MULTIPLIER = 2.0 DEF_SESSION_START = "09:15" DEF_INTERVAL = "15" def load_json_env(name): raw = (os.environ.get(name, "") or "").strip() if not raw: return {} try: val = json.loads(raw) return val if isinstance(val, dict) else {} except (ValueError, TypeError): return {} def emit(fired, headline, evidence, state): print(json.dumps({ "fired": bool(fired), "headline": headline, "evidence": evidence, "state": state, })) def is_auth_failure(resp): if resp.status_code in (401, 403): return True try: body = resp.json() except ValueError: return False et = str(body.get("errorType", "")).lower() ec = str(body.get("errorCode", "")).lower() if "auth" in et or "token" in et or ec in ("dh-901", "dh-902", "807"): return True return False def resolve_equity_security_ids(symbols, headers_unused=None): """Resolve NSE equity symbols to Dhan security ids from the instrument master (Wire 4). Verified live (2026-07-08 forge, orb_volume_surge): NSE cash equities are SEM_EXM_EXCH_ID=='NSE' with SEM_EXCH_INSTRUMENT_TYPE=='ES' and SEM_INSTRUMENT_NAME=='EQUITY' — NOT 'EQ'. Same family of instrument-type/instrument-name split trap as the documented FUTSTK/FUTIDX gotchas, now confirmed for the cash-equity segment too. Matched against SEM_TRADING_SYMBOL exactly (cash equities trade under their bare symbol, no '-' expiry suffix like futures).""" resp = requests.get(SCRIP_MASTER_URL, timeout=30) resp.raise_for_status() reader = csv.DictReader(io.StringIO(resp.content.decode("utf-8", errors="replace"))) by_symbol = {} wanted = set(symbols) for row in reader: if row.get("SEM_EXM_EXCH_ID") != "NSE": continue if row.get("SEM_EXCH_INSTRUMENT_TYPE") != "ES": continue if row.get("SEM_INSTRUMENT_NAME") != "EQUITY": continue sym = row.get("SEM_TRADING_SYMBOL", "") if sym in wanted and sym not in by_symbol: by_symbol[sym] = row.get("SEM_SMST_SECURITY_ID") return by_symbol def fetch_intraday_bars(security_id, headers, interval, to_dt): """Gotcha (verified 2026-07-08 forge, orb_volume_surge): a fromDate/toDate window scoped exactly to "today only" can come back with zero bars even when today's bars genuinely exist (observed live: a same-day-only query returned 0 rows for a session that, on a padded multi-day query, showed full 09:15-15:15 IST coverage). Mirrors the catalog's daily-bar guidance ("ask for more calendar days than trading days you need") — pad the window back several calendar days and filter to today's bars in Python rather than trusting Dhan's exact single-day boundary handling.""" from_dt = to_dt - timedelta(days=5) payload = { "securityId": security_id, "exchangeSegment": "NSE_EQ", "instrument": "EQUITY", "interval": str(interval), "fromDate": from_dt.strftime("%Y-%m-%d %H:%M:%S"), "toDate": to_dt.strftime("%Y-%m-%d %H:%M:%S"), } return requests.post(INTRADAY_URL, headers=headers, json=payload, timeout=30) def parse_hhmm(s, default): try: h, m = str(s or default).split(":") return int(h), int(m) except (ValueError, AttributeError): h, m = default.split(":") return int(h), int(m) def main(): config = load_json_env("SCOUT_CONFIG_JSON") prior_state = load_json_env("SCOUT_STATE_JSON") token = os.environ.get("DHAN_ACCESS_TOKEN", "") client_id = os.environ.get("DHAN_CLIENT_ID", "") if not token or not client_id: emit(False, None, None, {"status": "no_credentials", "reason": "DHAN_ACCESS_TOKEN / DHAN_CLIENT_ID not in env"}) return 0 headers = { "access-token": token, "client-id": client_id, "Content-Type": "application/json", } universe = config.get("universe") or DEF_UNIVERSE universe = [str(s).strip().upper() for s in universe if str(s).strip()] vol_multiplier = float(config.get("vol_multiplier", DEF_VOL_MULTIPLIER)) session_start = str(config.get("session_start") or DEF_SESSION_START) interval = str(config.get("interval") or DEF_INTERVAL) now_ist = datetime.now(IST) today_str = now_ist.strftime("%Y-%m-%d") start_h, start_m = parse_hhmm(session_start, DEF_SESSION_START) session_open = now_ist.replace(hour=start_h, minute=start_m, second=0, microsecond=0) try: id_map = resolve_equity_security_ids(universe) except requests.exceptions.RequestException as e: print(f"ERROR: scrip-master fetch failed: {e}", file=sys.stderr) return 1 unresolved = [s for s in universe if s not in id_map] if len(unresolved) == len(universe): print(f"ERROR: resolved zero NSE EQ security ids for universe {universe} — " f"scrip master shape may have changed.", file=sys.stderr) return 1 # Reset per-day dedup state on a new trading day. fired_today = prior_state.get("fired_today") or [] if prior_state.get("last_date") != today_str: fired_today = [] results_by_symbol = prior_state.get("results_by_symbol") or {} new_fires = [] checked = [] for symbol in universe: security_id = id_map.get(symbol) if not security_id: continue try: resp = fetch_intraday_bars(security_id, headers, interval, now_ist) except requests.exceptions.RequestException as e: print(f"WARN: intraday fetch failed for {symbol}: {e}", file=sys.stderr) continue if is_auth_failure(resp): emit(False, None, None, {"status": "no_credentials", "reason": f"intraday auth failure ({resp.status_code})"}) return 0 try: resp.raise_for_status() except requests.exceptions.HTTPError as e: print(f"WARN: intraday HTTP error for {symbol}: {e}", file=sys.stderr) continue bars = resp.json() timestamps = bars.get("timestamp") or [] highs = bars.get("high") or [] lows = bars.get("low") or [] closes = bars.get("close") or [] volumes = bars.get("volume") or [] if not timestamps: continue all_rows = [ (t, h, l, c, v) for t, h, l, c, v in zip(timestamps, highs, lows, closes, volumes) if h is not None and l is not None and c is not None and v is not None ] # Filter to only today's bars (>= session_open) — the fetch window # is padded back several calendar days on purpose (see the fetch # gotcha above), so prior sessions' bars are mixed into the raw # response and must be excluded here. session_open_ts = session_open.timestamp() rows = [r for r in all_rows if r[0] >= session_open_ts] rows.sort(key=lambda x: x[0]) if len(rows) < 2: # Only the opening bar itself has printed so far — nothing to # compare it against yet. continue opening_bar = rows[0] _, or_high, or_low, _, _ = opening_bar later_bars = rows[1:] checked.append(symbol) avg_vol_so_far = sum(v for _, _, _, _, v in rows) / len(rows) last_ts, last_high, last_low, last_close, last_vol = later_bars[-1] vol_ratio = (last_vol / avg_vol_so_far) if avg_vol_so_far > 0 else 0.0 big_volume = vol_ratio >= vol_multiplier broke_up = last_close > or_high broke_down = last_close < or_low direction = None level = None if broke_up and big_volume: direction = "UP" level = or_high elif broke_down and big_volume: direction = "DOWN" level = or_low results_by_symbol[symbol] = { "or_high": round(float(or_high), 2), "or_low": round(float(or_low), 2), "last_close": round(float(last_close), 2), "vol_ratio": round(float(vol_ratio), 2), } if direction is None: continue dedup_key = f"{today_str}:{symbol}:{direction}" if dedup_key in fired_today: continue new_fires.append({ "symbol": symbol, "direction": direction, "or_high": round(float(or_high), 2), "or_low": round(float(or_low), 2), "breakout_level": round(float(level), 2), "last_close": round(float(last_close), 2), "last_bar_volume": round(float(last_vol), 2), "avg_bar_volume": round(float(avg_vol_so_far), 2), "vol_ratio": round(float(vol_ratio), 2), }) fired_today.append(dedup_key) state = { "last_date": today_str, "fired_today": fired_today, "results_by_symbol": results_by_symbol, "universe_checked": checked, "unresolved_symbols": unresolved, } if not new_fires: emit(False, None, None, state) return 0 new_fires.sort(key=lambda f: f["vol_ratio"], reverse=True) top = new_fires[0] if len(new_fires) == 1: headline = ( f"{top['symbol']} opening-range breakout {top['direction']} " f"at {top['last_close']:.2f} (OR {top['or_low']:.2f}-{top['or_high']:.2f}) " f"on {top['vol_ratio']:.2f}x volume" ) else: headline = ( f"{len(new_fires)} opening-range breakouts on volume — led by " f"{top['symbol']} {top['direction']} at {top['last_close']:.2f} " f"({top['vol_ratio']:.2f}x volume)" ) evidence = { "match_count": len(new_fires), "universe_scanned": len(checked), "vol_multiplier_required": vol_multiplier, "matches": new_fires[:15], } emit(True, headline, evidence, state) return 0 if __name__ == "__main__": sys.exit(main())