nubra_trap_detector

A market-trap detection library for Indian index derivatives (NIFTY, BANKNIFTY).
Identifies bull traps, bear traps, short squeezes, and long squeezes by combining price-action signals with live options-chain data from the Nubra SDK.

---

What is a "trap"?

A trap is a price move that looks like a breakout but has no genuine conviction behind it — so it reverses sharply, catching trend-followers on the wrong side.

Trap type	What happens	Who gets hurt
Bull trap	Price breaks above a resistance level on weak volume / no OI expansion	Breakout buyers
Bear trap	Price breaks below support on weak volume / no OI expansion	Short sellers
Short squeeze	Extreme put-loading + delta imbalance forces trapped shorts to cover	Short sellers
Long squeeze	Extreme call-loading + delta imbalance forces trapped longs to exit	Long holders

---

Installation

pip install nubra-trap-detector

Requires Python 3.10+ and the Nubra SDK (installed automatically):

pip install nubra-sdk>=0.3.8

---

Quick start

from nubra_python_sdk.start_sdk import InitNubraSdk, NubraEnv
from nubra_trap_detector import TrapEngine
from nubra_trap_detector.formatters import compact

nubra  = InitNubraSdk(NubraEnv.PROD, env_creds=True)
engine = TrapEngine(nubra_client=nubra, symbols=["NIFTY"], interval="5m")
engine.on_result(compact)
engine.start()
engine.wait()   # blocks; Ctrl+C to stop

Sample output (one line printed every 5 minutes at candle close):

13:20:02  NIFTY       5m  | P: 30.0%  S:  0.0% | [GRN]  16.5% bull_trap           down | bars=21 [09:15->13:20]
13:25:02  NIFTY       5m  | P:  0.0%  S: 65.0% | [ORG]  29.3% short_squeeze        up  | bars=22 [09:15->13:25]

---

Try without a live API (walkthrough mode)

The run_walkthrough() function takes raw candle / option-chain data and prints every formula and intermediate value — no API connection needed:

from nubra_trap_detector import run_walkthrough

result = run_walkthrough(
    symbol         = "NIFTY",
    candles        = [(open, high, low, close, volume), ...],   # 21 tuples
    ce_chain       = [(strike, oi, delta), ...],
    pe_chain       = [(strike, oi, delta), ...],
    prev_ce_oi     = [...],   # OI from previous chain snapshot (same order)
    prev_pe_oi     = [...],
    orderbook_asks = [{"price": int, "quantity": int}, ...],
    spot           = 24020,
    verbose        = True,    # False = return dict only, no printed output
)
print(result["total"]["trap_probability"])   # 0 – 100

Run the bundled example:

python calc_walkthrough.py

---

Architecture

Raw market data (Nubra WebSocket + REST)
        |
        v
  StateStore  (thread-safe ring buffers)
  |-- OHLCV candles   per symbol   (25-bar ring)
  |-- Option chain    per symbol   (refreshed every 60 s via REST)
  `-- Orderbook       per ref_id   (live WebSocket)
        |
        |---> PriceTrapDetector   (weight 0.55)
        |       Breakout check (20-bar high/low)
        |       Volume ratio
        |       Chain OI expansion
        |       Orderbook thinness
        |
        |---> SqueezeDetector     (weight 0.45)
        |       PCR (Put-Call Ratio)
        |       Delta-weighted OI imbalance
        |       OI unwind speed
        |
        v
  SignalAggregator
  trap_probability = (price x 0.55 + squeeze x 0.45) x 100   ->   0 – 100

---

How the score is calculated

PriceTrapDetector (weight: 0.55)

Fires when price breaks a key level but the move lacks conviction.

Sub-check	Formula	Score added
Breakout	close > max(high, last 20 bars) or close < min(low, last 20 bars)	Gate (required)
Volume ratio	vol_ratio = current_vol / mean(vol, last 20 bars) — fires when vol_ratio < 1.0	+0.30
Chain OI change	oi_growth = (curr_chain_OI - prev_chain_OI) / prev_chain_OI — fires when oi_growth < 0.01	+0.40
Orderbook thin	total_qty = sum(ask qty within 1% of breakout level) — fires when total_qty == 0	+0.20

Max score: 0.90 (capped at 1.0)

SqueezeDetector (weight: 0.45)

Fires when one side of the options market is dangerously overloaded.

Sub-check	Formula	Score added
PCR	PCR = put_OI / call_OI — fires when PCR > 1.2 (short squeeze) or PCR < 0.75 (long squeeze)	+0.35
Delta imbalance	`imbalance =	call_delta_OI - put_delta_OI
OI unwind	Top-3 OI strikes: drop = (prev_OI - curr_OI) / prev_OI — fires when any drop > 0.08	+0.25

Max score: 0.90 (capped at 1.0)

Final aggregation

trap_probability = (price_score x 0.55 + squeeze_score x 0.45) x 100

Clamped to [0, 100]. The detector with the highest score x weight product becomes the dominant detector and sets direction and dominant_trap_type.

Score interpretation

Range	Meaning	Suggested action
0 – 10	No trap	Trade normally
10 – 30	Low risk	Monitor
30 – 50	Moderate risk	Reduce size, tighten stop
50 – 75	High risk	Block new entries in trap direction
75 – 100	Very high conviction	Hard gate — avoid entry, consider hedge

---

API reference

TrapEngine

TrapEngine(
    nubra_client,                        # authenticated Nubra SDK client
    symbols          = ["NIFTY"],        # list of index symbols
    interval         = "5m",            # "1m" | "3m" | "5m" | "10m" | "15m" | "30m" | "1h" | "realtime"
    exchange         = "NSE",
    detector_weights = None,             # dict or None (uses config defaults)
)

Method	Returns	Description
engine.start()	None	Discover expiries, start feed, begin polling. Non-blocking.
engine.wait()	None	Block until Ctrl+C. Call after start().
engine.stop()	None	Gracefully stop all threads.
engine.on_result(fn)	None	Register a (dict) -> None callback, called on every poll.
engine.get_result(symbol)	dict	Run detectors now and return full result dict.
engine.get_score(symbol)	TrapScore	Lightweight: return TrapScore dataclass only.
engine.refresh_chain_snapshot(symbol)	None	Manually trigger a REST chain refresh.

---

get_result() — result dict schema

result = engine.get_result("NIFTY")

{
  "timestamp":    "2026-05-20T13:20:02+0530",
  "symbol":       "NIFTY",
  "interval":     "5m",
  "spot":         24020.0,
  "bars":         21,
  "candle_range": ["09:15", "13:20"],

  "price_trap": {
    "score":           30.0,        # 0–100
    "close":           24020.0,
    "high_n":          24010.0,     # 20-bar high (bull breakout level)
    "low_n":           23480.0,     # 20-bar low  (bear breakout level)
    "breakout":        "bull",      # "bull" | "bear" | "none"
    "vol_ratio":       0.740,       # current vol / 20-bar avg vol
    "chain_oi_change": 0.046,       # fractional chain OI change vs prior snapshot
    "orderbook_thin":  False,       # True = no liquidity near breakout level
    "factors": ["..."],             # plain-English reasons
  },

  "squeeze_trap": {
    "score":           0.0,         # 0–100
    "pcr":             1.047,       # put_OI / call_OI
    "delta_imbalance": 0.015,       # |call_dOI - put_dOI| / total_dOI
    "dominant_side":   "call",      # "call" | "put"
    "oi_unwinding":    False,       # True = top-3 strikes losing OI fast
    "factors": [],
  },

  "total": {
    "trap_probability": 16.5,       # 0–100 — the main signal
    "direction":        "bearish_trap",   # "bearish_trap" | "bullish_trap" | "neutral"
    "dominant_type":    "bull_trap",      # "bull_trap" | "bear_trap" | "short_squeeze" | "long_squeeze" | "none"
    "weights":          {"price_trap": 0.55, "squeeze_trap": 0.45},
  },
}

---

TrapScore dataclass

Returned by engine.get_score().

@dataclass
class TrapScore:
    trap_probability:  float        # 0.0 – 100.0
    dominant_trap_type: str         # "bull_trap" | "short_squeeze" | ...
    direction:         str          # "bearish_trap" | "bullish_trap" | "neutral"
    breakdown:         dict         # {"price_trap": 0.30, "squeeze_trap": 0.0}
    factors:           list[str]
    timestamp:         datetime

---

run_walkthrough()

Step-by-step formula trace — no API needed.

from nubra_trap_detector import run_walkthrough

result = run_walkthrough(
    symbol         = "NIFTY",
    candles        = [(open, high, low, close, volume), ...],  # min 21 tuples
    ce_chain       = [(strike, oi, delta), ...],
    pe_chain       = [(strike, oi, delta), ...],
    prev_ce_oi     = [int, ...],     # previous OI snapshot, same order as ce_chain
    prev_pe_oi     = [int, ...],
    orderbook_asks = [{"price": int, "quantity": int}, ...],
    spot           = 24020,
    weights        = None,           # optional override (default from config)
    verbose        = True,           # False = silent, return dict only
)

---

Formatters

Three built-in formatters ship with the library. Register them with engine.on_result().

from nubra_trap_detector.formatters import compact, table, json_lines

Formatter	Output	Best for
compact	One line per symbol per poll	Live monitoring terminal
table	Multi-line box with all metrics	Debugging
json_lines	Full result as a single JSON line	Log files, Grafana, dashboards

Write your own:

def my_formatter(result: dict) -> None:
    prob = result["total"]["trap_probability"]
    if prob > 50:
        send_telegram(f"Trap alert: {result['symbol']} {prob:.1f}%")

engine.on_result(my_formatter)

---

Configuration

All thresholds live in nubra_trap_detector/config.py. Modify the file to tune sensitivity for different market conditions.

# Price trap
BREAKOUT_BARS             = 20     # N-bar high/low lookback
VOLUME_RATIO_THRESHOLD    = 1.0    # below-average volume threshold
CHAIN_OI_EXPAND_THRESHOLD = 0.01   # 1% minimum OI growth to count as "expanding"
ORDERBOOK_THIN_PCT        = 0.01   # "near breakout" = within 1% of level

# Squeeze trap
PCR_UPPER                 = 1.2    # PCR above this -> short squeeze signal
PCR_LOWER                 = 0.75   # PCR below this -> long squeeze signal
DELTA_IMBALANCE_THRESHOLD = 0.20   # 20% imbalance threshold
OI_UNWIND_THRESHOLD       = 0.08   # 8% OI drop = rapid unwind

# Ensemble weights (must sum to 1.0)
DETECTOR_WEIGHTS = {
    "price_trap":   0.55,
    "squeeze_trap": 0.45,
}

---

Environment variables

Variable	Description
NUBRA_API_KEY	Your Nubra API key
NUBRA_API_SECRET	Your Nubra API secret

The SDK reads these automatically when you pass env_creds=True to InitNubraSdk.

---

Logging

The library uses Python's standard logging module under the nubra_trap_detector namespace.

import logging
logging.basicConfig(level=logging.DEBUG)   # see all internal messages
logging.getLogger("nubra_trap_detector").setLevel(logging.WARNING)   # suppress

---

License

MIT