scripts/probe_native_broker_live.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Run redacted, read-only live probes for native broker adapters.
The script is intentionally interactive. It never accepts secrets on the command
line, never prints broker payloads, and performs no order writes.
"""
from __future__ import annotations
import argparse
import asyncio
import getpass
import re
import sys
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Awaitable, Callable
from zoneinfo import ZoneInfo
ROOT = Path(__file__).resolve().parents[1]
for rel in ("packages/core/core/src", "packages/integrations/gateway/src"):
path = str(ROOT / rel)
if path not in sys.path:
sys.path.insert(0, path)
from flinttrade_gateway.adapter import BROKER_CATALOG # noqa: E402
from flinttrade_gateway.brokers.delta import DeltaAdapter # noqa: E402
from flinttrade_gateway.brokers.dhan import DhanAdapter # noqa: E402
from flinttrade_gateway.brokers.groww import GrowwAdapter # noqa: E402
from flinttrade_gateway.brokers.indmoney import IndMoneyAdapter # noqa: E402
from flinttrade_gateway.brokers.kotakneo import KotakNeoAdapter # noqa: E402
from flinttrade_gateway.brokers.upstox import UpstoxAdapter # noqa: E402
from flinttrade_gateway.native_login import should_keep_session_after_probe_error # noqa: E402
AdapterFactory = Callable[[], Any]
ReadCall = Callable[[Any], Awaitable[Any]]
ADAPTER_FACTORIES: dict[str, AdapterFactory] = {
"deltaexchange": DeltaAdapter,
"dhan": DhanAdapter,
"groww": GrowwAdapter,
"indmoney": IndMoneyAdapter,
"kotakneo": KotakNeoAdapter,
"upstox": UpstoxAdapter,
}
COMMON_READ_CHOICES = ("profile", "funds", "positions", "holdings", "orders", "trades")
ORDER_DETAIL_READ_CHOICES = ("orderstatus", "orderhistory", "ordertrades")
COMMON_MARKET_READ_CHOICES = ("quotes", "depth", "margin", "history")
DHAN_MARKET_READ_CHOICES = ("quotes", "ltp", "ohlc", "quote_details", "margin", "history")
DHAN_QUOTE_PROJECTION_READS = frozenset({"ltp", "ohlc", "quote_details"})
DHAN_SECURITY_RESOLVER_READ_CHOICES = frozenset(DHAN_MARKET_READ_CHOICES)
GROWW_MARKET_READ_CHOICES = ("quotes", "ltp", "ohlc", "margin", "history", "expiry")
INDMONEY_MARKET_READ_CHOICES = ("quotes", "ltp", "depth", "margin", "history")
UPSTOX_MARKET_READ_CHOICES = COMMON_MARKET_READ_CHOICES + ("ltp", "ohlc")
DELTA_PROBE_SYMBOL = "BTCUSD"
DELTA_PROBE_UNDERLYING = "BTC"
DELTA_READ_CHOICES = COMMON_READ_CHOICES + ("quotes", "depth", "history", "optionchain", "orderstatus", "orderhistory")
PROBE_EXCHANGE = "NSE"
PROBE_SYMBOL = "RELIANCE"
PROBE_QUOTE_SYMBOL = f"{PROBE_EXCHANGE}:{PROBE_SYMBOL}"
PROBE_UNDERLYING = "NIFTY"
PROBE_INDEX_EXCHANGE = "NSE_INDEX"
PROBE_OPTION_SYMBOL = "NFO:NIFTY25000CE"
KOTAK_PROBE_EXCHANGE = PROBE_EXCHANGE
KOTAK_PROBE_SYMBOL = PROBE_SYMBOL
KOTAK_READ_CHOICES = (
"funds",
"limits",
"positions",
"holdings",
"orders",
"trades",
"orderhistory",
"ordertrades",
"margin",
"scrip_master",
"search_scrip",
"quotes",
"quote_details",
"market_depth",
)
READ_CHOICES_BY_BROKER: dict[str, tuple[str, ...]] = {
"dhan": COMMON_READ_CHOICES + DHAN_MARKET_READ_CHOICES + ("expiry", "optionchain"),
"groww": COMMON_READ_CHOICES + ("orderstatus", "ordertrades") + GROWW_MARKET_READ_CHOICES + ("optionchain",),
"indmoney": COMMON_READ_CHOICES + ("orderstatus", "ordertrades") + INDMONEY_MARKET_READ_CHOICES,
"kotakneo": KOTAK_READ_CHOICES,
"upstox": COMMON_READ_CHOICES + UPSTOX_MARKET_READ_CHOICES + (
*ORDER_DETAIL_READ_CHOICES,
"optiongreeks",
"expiry",
"optionchain",
"search",
"timings",
"holidays",
),
"deltaexchange": DELTA_READ_CHOICES,
}
READ_CHOICES = tuple(dict.fromkeys(read for choices in READ_CHOICES_BY_BROKER.values() for read in choices))
DEFAULT_READS: dict[str, tuple[str, ...]] = {
# One Dhan marketfeed call is enough for bundled probes (default and all):
# ltp/ohlc and quote_details project the same quote_data response and remain
# explicitly selectable so the 1 req/s quote limit is not hit by redundancy.
"dhan": COMMON_READ_CHOICES + ("quotes", "margin", "history"),
"groww": COMMON_READ_CHOICES + GROWW_MARKET_READ_CHOICES,
"indmoney": COMMON_READ_CHOICES + INDMONEY_MARKET_READ_CHOICES,
"kotakneo": KOTAK_READ_CHOICES,
"upstox": COMMON_READ_CHOICES + UPSTOX_MARKET_READ_CHOICES + ("search", "timings", "holidays"),
"deltaexchange": ("profile", "funds", "positions", "orders", "quotes", "history"),
}
DEFAULT_METHOD: dict[str, str] = {
"dhan": "access_token",
"groww": "api_key_secret",
"indmoney": "access_token",
"kotakneo": "totp_mpin",
"upstox": "access_token",
"deltaexchange": "api_key",
}
@dataclass(frozen=True)
class CredentialField:
name: str
label: str
required: bool = True
CREDENTIAL_FIELDS: dict[str, dict[str, tuple[CredentialField, ...]]] = {
"dhan": {
"access_token": (
CredentialField("client_id", "Dhan client ID"),
CredentialField("access_token", "Dhan access token"),
),
"pin_totp": (
CredentialField("client_id", "Dhan client ID"),
CredentialField("pin", "Dhan PIN"),
CredentialField("totp", "Current Dhan TOTP"),
),
"oauth_token_id": (
CredentialField("client_id", "Dhan client ID"),
CredentialField("app_id", "Dhan app ID"),
CredentialField("app_secret", "Dhan app secret"),
CredentialField("token_id", "Dhan OAuth tokenId"),
),
},
"groww": {
"api_key_totp": (
CredentialField("api_key", "Groww Trade API key"),
CredentialField("totp", "Current Groww TOTP"),
CredentialField("user_id", "Optional Groww user/account label", required=False),
),
"api_key_secret": (
CredentialField("api_key", "Groww Trade API key"),
CredentialField("api_secret", "Groww Trade API secret"),
CredentialField("user_id", "Optional Groww user/account label", required=False),
),
"access_token": (
CredentialField("access_token", "Groww Trade API access token"),
CredentialField("user_id", "Optional Groww user/account label", required=False),
),
},
"indmoney": {
"access_token": (
CredentialField("access_token", "INDstocks access token"),
CredentialField("user_id", "Optional INDstocks user/account label", required=False),
),
},
"kotakneo": {
"totp_mpin": (
CredentialField("access_token", "Kotak Neo Trade API access token"),
CredentialField("mobile_number", "Kotak Neo mobile number"),
CredentialField("ucc", "Kotak Neo UCC"),
CredentialField("totp", "Current Kotak Neo TOTP"),
CredentialField("mpin", "Kotak Neo MPIN"),
CredentialField("neo_fin_key", "Optional Kotak neo_fin_key", required=False),
),
},
"upstox": {
"access_token": (
CredentialField("access_token", "Upstox access token"),
CredentialField("client_id", "Optional Upstox account/client label", required=False),
),
"analytics_access_token": (
CredentialField("access_token", "Upstox Analytics access token"),
CredentialField("client_id", "Optional Upstox account/client label", required=False),
),
"oauth_code": (
CredentialField("code", "Upstox OAuth code"),
CredentialField("api_key", "Upstox API key"),
CredentialField("api_secret", "Upstox API secret"),
CredentialField("redirect_uri", "Registered redirect URI"),
CredentialField("client_id", "Optional Upstox account/client label", required=False),
),
},
"deltaexchange": {
"api_key": (
CredentialField("api_key", "Delta API key"),
CredentialField("api_secret", "Delta API secret"),
CredentialField("environment", "Venue (india_prod, india_testnet, global_prod, global_testnet)"),
CredentialField("user_id", "Optional Delta user id", required=False),
),
},
}
def _credential_defaults(broker: str, method: str) -> dict[str, str]:
"""Auth-method credential defaults from the broker catalogue.
The in-app connect flow merges the selected method's ``credential_defaults``
(e.g. the Upstox analytics token's ``read_only``/``token_scope`` flags) into
the captured credentials. The probe derives them from the SAME catalogue
entry so the two paths can never drift.
"""
info = BROKER_CATALOG.get(broker)
for entry in info.auth_methods if info else []:
if entry.id == method:
return {str(key): str(value) for key, value in entry.credential_defaults.items()}
return {}
SECRET_FIELD_NAMES = (
"access[_ -]?token",
"accessToken",
"api[_ -]?key",
"apiKey",
"api[_ -]?secret",
"apiSecret",
"app[_ -]?id",
"appId",
"app[_ -]?secret",
"appSecret",
"client[_ -]?id",
"clientId",
"client[_ -]?secret",
"clientSecret",
"code",
"consumer[_ -]?key",
"consumerKey",
"mobile[_ -]?number",
"mobileNumber",
"mpin",
"neo[_ -]?fin[_ -]?key",
"neoFinKey",
"pin",
"redirect[_ -]?uri",
"redirectUri",
"token[_ -]?id",
"tokenId",
"totp",
"ucc",
"user[_ -]?id",
"userId",
)
SECRET_FIELDS_RE = re.compile(
r"(?i)(['\"]?\b(?:"
+ "|".join(SECRET_FIELD_NAMES)
+ r")\b['\"]?\s*[:=]\s*)['\"]?[^,'\"\s)}\]]+['\"]?"
)
EMAIL_RE = re.compile(r"[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}", re.IGNORECASE)
MOBILE_RE = re.compile(r"\b(?:\+?91[-\s]?)?[6-9]\d{9}\b")
IP_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
URL_RE = re.compile(r"https?://[^\s,'\")\]}]+")
BEARER_RE = re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._~+/=-]+")
JWT_RE = re.compile(r"\beyJ[A-Za-z0-9_-]+(?:\.[A-Za-z0-9_-]+){1,2}\b")
LONG_TOKEN_RE = re.compile(r"\b[A-Za-z0-9_-]{18,}\b")
ACCOUNTISH_RE = re.compile(r"\b[A-Z0-9]{5,16}\b")
LONG_NUMBER_RE = re.compile(r"\b\d{4,}\b")
class PromptAborted(Exception):
"""Raised when the operator cancels a hidden-input prompt."""
class InconclusiveRead(Exception):
"""Raised when a read is supported but needs live state the account lacks."""
def broker_read_help(broker: str) -> str:
"""Return the exact read names accepted by the shared probe for one broker."""
choices = READ_CHOICES_BY_BROKER[broker]
return " ".join(choices)
def broker_method_help() -> str:
"""Return the exact credential methods accepted by the shared probe."""
lines = []
for broker in sorted(CREDENTIAL_FIELDS):
methods = ", ".join(sorted(CREDENTIAL_FIELDS[broker]))
default = DEFAULT_METHOD[broker]
lines.append(f"{broker}: {methods} (default: {default})")
return "\n".join(lines)
def broker_reads_help() -> str:
"""Return the exact read choices accepted by the shared probe."""
lines = []
for broker in sorted(READ_CHOICES_BY_BROKER):
lines.append(f"{broker}: {broker_read_help(broker)}")
return "\n".join(lines)
def redact(text: object) -> str:
"""Return a bounded string with broker/account identifiers removed."""
value = str(text)
value = SECRET_FIELDS_RE.sub(lambda match: f"{match.group(1)}[redacted]", value)
value = EMAIL_RE.sub("[email]", value)
value = MOBILE_RE.sub("[mobile]", value)
value = IP_RE.sub("[ip]", value)
value = URL_RE.sub("[url]", value)
value = BEARER_RE.sub("Bearer [redacted]", value)
value = JWT_RE.sub("[token]", value)
value = LONG_TOKEN_RE.sub("[token]", value)
value = ACCOUNTISH_RE.sub("[id]", value)
value = LONG_NUMBER_RE.sub("[number]", value)
return value[:500]
def summarise_payload(payload: Any) -> str:
"""Describe only the shape of a broker response, never its contents."""
if isinstance(payload, list):
return f"ok rows={len(payload)}"
if isinstance(payload, tuple):
return f"ok rows={len(payload)}"
if isinstance(payload, dict):
return f"ok object_keys={len(payload)}"
if payload is None:
return "ok empty"
return f"ok type={type(payload).__name__}"
def _secret_prompt(label: str, *, required: bool = True) -> str:
while True:
try:
value = getpass.getpass(f"{label}: ").strip()
except (EOFError, KeyboardInterrupt) as exc:
raise PromptAborted from exc
if value or not required:
return value
print(f"{label} is required.")
def _resolve_method(broker: str, method: str | None) -> str:
selected = method or DEFAULT_METHOD[broker]
if selected not in CREDENTIAL_FIELDS[broker]:
allowed = ", ".join(sorted(CREDENTIAL_FIELDS[broker]))
raise ValueError(f"{broker} method must be one of: {allowed}")
return selected
def collect_credentials(broker: str, method: str, environment: str) -> dict[str, str]:
"""Collect broker credentials without shell history or terminal echo."""
display = {
"dhan": "Dhan",
"groww": "Groww",
"indmoney": "INDstocks",
"kotakneo": "Kotak Neo",
"upstox": "Upstox",
"deltaexchange": "Delta Exchange",
}[broker]
print(f"Enter {display} values locally. They will not be printed or stored by this script.")
credentials: dict[str, str] = {}
if broker == "kotakneo":
credentials["environment"] = environment
for field in CREDENTIAL_FIELDS[broker][method]:
value = _secret_prompt(field.label, required=field.required)
if value:
credentials[field.name] = value
credentials.update(_credential_defaults(broker, method))
return credentials
def _bundled_reads(broker: str, requested: str) -> list[str]:
"""Expand ``default`` / ``all`` without repeating Dhan quote projections.
Dhan ``ltp``, ``ohlc``, and ``quote_details`` all dispatch through
``quotes()`` to the same ``quote_data`` endpoint (1 req/s). Bundled
inventories keep one representative quotes read; the projections stay
individually selectable.
"""
if requested == "default":
return list(DEFAULT_READS[broker])
reads = list(READ_CHOICES_BY_BROKER[broker])
if broker == "dhan":
return [read for read in reads if read not in DHAN_QUOTE_PROJECTION_READS]
return reads
def _resolve_reads(broker: str, requested: list[str] | None) -> list[str]:
allowed_reads = READ_CHOICES_BY_BROKER[broker]
if not requested or requested == ["default"]:
return _bundled_reads(broker, "default")
if requested == ["all"]:
return _bundled_reads(broker, "all")
reads: list[str] = []
for read in requested:
if read == "default":
reads.extend(_bundled_reads(broker, "default"))
elif read == "all":
reads.extend(_bundled_reads(broker, "all"))
else:
reads.append(read)
deduped: list[str] = []
for read in reads:
if read not in allowed_reads:
raise ValueError(f"unknown {broker} read {read!r}; choose from: {', '.join(allowed_reads)}")
if read not in deduped:
deduped.append(read)
return deduped
def _dhan_reads_need_security_resolver(reads: list[str]) -> bool:
return any(read in DHAN_SECURITY_RESOLVER_READ_CHOICES for read in reads)
async def _prepare_adapter_for_probe(broker: str, adapter: Any, reads: list[str]) -> None:
if broker != "dhan" or not _dhan_reads_need_security_resolver(reads):
return
if getattr(adapter, "_security_resolver", None) is not None:
return
fetch_security_list = getattr(adapter, "fetch_security_list", None)
if not callable(fetch_security_list):
return
try:
from flinttrade_gateway.brokers.dhan_mapping import build_security_resolver # noqa: PLC0415
rows = await fetch_security_list("compact")
except Exception as exc: # noqa: BLE001 - market reads will report the actual resolver miss
print(f"resolver: failed {type(exc).__name__}: {redact(exc)}")
return
adapter._security_resolver = build_security_resolver(rows)
print(f"resolver: ok dhan scrip_master rows={len(rows)}")
def _today_ist() -> datetime:
return datetime.now(ZoneInfo("Asia/Kolkata"))
def _history_request() -> dict[str, Any]:
end = _today_ist().date()
start = end - timedelta(days=7)
return {
"symbol": PROBE_SYMBOL,
"exchange": PROBE_EXCHANGE,
"interval": "1d",
"start_date": start.isoformat(),
"end_date": end.isoformat(),
}
def _sample_order() -> Any:
from flinttrade_core.models import Order # noqa: PLC0415
return Order(
symbol=PROBE_SYMBOL,
exchange=PROBE_EXCHANGE,
action="BUY",
product="MIS",
quantity="1",
pricetype="MARKET",
price="0",
trigger_price="0",
)
def _delta_history_request() -> dict[str, Any]:
end = _today_ist()
start = end - timedelta(days=7)
return {
"symbol": DELTA_PROBE_SYMBOL,
"exchange": "CRYPTO",
"interval": "1h",
"start_date": start.date().isoformat(),
"end_date": end.date().isoformat(),
}
def _delta_option_chain_request() -> dict[str, Any]:
expiry = (_today_ist().date() + timedelta(days=30)).isoformat()
return {
"symbol": DELTA_PROBE_UNDERLYING,
"underlying": DELTA_PROBE_UNDERLYING,
"exchange": "CRYPTO",
"expiry": expiry,
"expiry_date": expiry,
}
def _quote_symbol(broker: str) -> str:
return DELTA_PROBE_SYMBOL if broker == "deltaexchange" else PROBE_QUOTE_SYMBOL
def _option_chain_request() -> dict[str, Any]:
expiry = (_today_ist().date() + timedelta(days=30)).isoformat()
return {
"symbol": PROBE_UNDERLYING,
"underlying": PROBE_UNDERLYING,
"exchange": PROBE_INDEX_EXCHANGE,
"expiry": expiry,
"expiry_date": expiry,
}
def _payload_rows(payload: Any) -> list[Any]:
"""Return likely table rows from a broker read payload."""
if isinstance(payload, list):
return payload
if isinstance(payload, tuple):
return list(payload)
if hasattr(payload, "model_dump"):
return [payload.model_dump()]
if isinstance(payload, dict):
for key in ("data", "orders", "order_list", "items", "rows"):
value = payload.get(key)
if isinstance(value, list):
return value
return [payload]
return []
def _read_field(row: Any, *names: str) -> str:
if hasattr(row, "model_dump"):
row = row.model_dump()
if not isinstance(row, dict):
return ""
for name in names:
value = row.get(name)
if value not in (None, ""):
return str(value)
return ""
async def _sample_order_id(adapter: Any, session: Any) -> str:
"""Find a broker order id from order_book for per-order read probes."""
order_book = getattr(adapter, "order_book", None)
if not callable(order_book):
raise InconclusiveRead("adapter has no order book to sample an order id")
rows = _payload_rows(await order_book(session))
for row in rows:
order_id = _read_field(
row,
"orderid",
"order_id",
"groww_order_id",
"nOrdNo",
"orderId",
"id",
)
if order_id:
return order_id
raise InconclusiveRead("no sample order id available from order book")
def _read_call(adapter: Any, broker: str, name: str) -> ReadCall | None:
if name == "profile":
if broker == "dhan":
return adapter.user_profile
return getattr(adapter, "profile", None)
if name == "orders":
return adapter.order_book
if name == "trades":
return adapter.trade_book
if name in {"orderstatus", "orderhistory", "ordertrades"}:
method_by_name = {
"orderstatus": "order_details",
"orderhistory": "order_history",
"ordertrades": "order_trades",
}
method = getattr(adapter, method_by_name[name], None)
if not callable(method):
if name == "ordertrades" and broker == "upstox":
method = getattr(adapter, "trades_by_order", None)
if not callable(method):
return None
async def _call_order_read(session: Any) -> Any:
order_id = await _sample_order_id(adapter, session)
return await method(session, order_id)
return _call_order_read
if name == "quotes":
return lambda session: adapter.quotes(session, [_quote_symbol(broker)])
if name == "ltp":
call = getattr(adapter, "ltp", None) or getattr(adapter, "ltp_quotes", None)
return (lambda session: call(session, [PROBE_QUOTE_SYMBOL])) if callable(call) else None
if name == "ohlc":
call = getattr(adapter, "ohlc", None) or getattr(adapter, "ohlc_quotes", None)
return (
(lambda session: call(session, [PROBE_QUOTE_SYMBOL]))
if callable(call)
else None
)
if name == "quote_details":
call = getattr(adapter, "quote_details", None)
return (
(lambda session: call(session, [PROBE_QUOTE_SYMBOL], "ltp"))
if callable(call)
else None
)
if name in {"depth", "market_depth"}:
symbol = _quote_symbol(broker)
return (
(lambda session: adapter.market_depth(session, [symbol]))
if callable(getattr(adapter, "market_depth", None))
else None
)
if name == "margin":
return (
(lambda session: adapter.margin_calculator(session, _sample_order()))
if callable(getattr(adapter, "margin_calculator", None))
else None
)
if name == "history":
request = _delta_history_request() if broker == "deltaexchange" else _history_request()
return (
(lambda session: adapter.historical(session, request))
if callable(getattr(adapter, "historical", None))
else None
)
if name == "expiry":
return (
(lambda session: adapter.expiry_list(session, PROBE_UNDERLYING, PROBE_INDEX_EXCHANGE))
if callable(getattr(adapter, "expiry_list", None))
else None
)
if name == "optionchain":
request = _delta_option_chain_request() if broker == "deltaexchange" else _option_chain_request()
return (
(lambda session: adapter.option_chain(session, request))
if callable(getattr(adapter, "option_chain", None))
else None
)
if name == "optiongreeks":
return (
(lambda session: adapter.option_greeks(session, [PROBE_OPTION_SYMBOL]))
if callable(getattr(adapter, "option_greeks", None))
else None
)
if name == "search":
return (
(lambda session: adapter.search_instruments(session, PROBE_SYMBOL))
if callable(getattr(adapter, "search_instruments", None))
else None
)
if name == "timings":
return (
(lambda session: adapter.market_timings(session, _today_ist().date().isoformat()))
if callable(getattr(adapter, "market_timings", None))
else None
)
if name == "holidays":
return (
(lambda session: adapter.market_holidays(session))
if callable(getattr(adapter, "market_holidays", None))
else None
)
if broker == "kotakneo":
if name == "scrip_master":
return lambda session: adapter.scrip_master(session, KOTAK_PROBE_EXCHANGE)
if name == "search_scrip":
return lambda session: adapter.search_scrip(session, KOTAK_PROBE_SYMBOL, KOTAK_PROBE_EXCHANGE)
return getattr(adapter, name, None)
async def run_probe(
broker: str,
method: str | None = None,
read_names: list[str] | None = None,
environment: str = "prod",
logout: bool = False,
) -> int:
broker = broker.lower()
if broker not in ADAPTER_FACTORIES:
allowed = ", ".join(sorted(ADAPTER_FACTORIES))
print(f"probe: unsupported broker {broker!r}; choose from: {allowed}")
return 2
try:
selected_method = _resolve_method(broker, method)
reads = _resolve_reads(broker, read_names)
except ValueError as exc:
print(f"probe: {exc}")
return 2
adapter = ADAPTER_FACTORIES[broker]()
await _prepare_adapter_for_probe(broker, adapter, reads)
session = None
try:
session = await adapter.login(collect_credentials(broker, selected_method, environment))
except PromptAborted:
print("probe: cancelled before login")
return 130
except Exception as exc: # noqa: BLE001 - CLI boundary must report safely
print(f"login: failed {type(exc).__name__}: {redact(exc)}")
return 2
print("login: ok")
failures = 0
for name in reads:
call = _read_call(adapter, broker, name)
if call is None:
print(f"{name}: skipped unsupported")
continue
try:
payload = await call(session)
except InconclusiveRead as exc:
print(f"{name}: inconclusive {redact(exc)}")
continue
except Exception as exc: # noqa: BLE001 - continue to collect all read statuses
if should_keep_session_after_probe_error(exc):
print(f"{name}: inconclusive {type(exc).__name__}: {redact(exc)}")
continue
failures += 1
print(f"{name}: failed {type(exc).__name__}: {redact(exc)}")
continue
print(f"{name}: {summarise_payload(payload)}")
if logout:
try:
await adapter.logout(session)
except Exception as exc: # noqa: BLE001 - logout is best effort but must stay redacted
failures += 1
print(f"logout: failed {type(exc).__name__}: {redact(exc)}")
else:
print("logout: ok")
else:
print("logout: skipped (pass --logout to revoke/close the broker token)")
return 1 if failures else 0
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Read-only native broker adapter probe with redacted output.",
epilog=(
"Credential methods by broker:\n"
f"{broker_method_help()}\n\n"
"Read choices by broker:\n"
f"{broker_reads_help()}"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("broker", choices=sorted(ADAPTER_FACTORIES))
parser.add_argument(
"--method",
help=(
"Credential method. Defaults per broker: dhan/upstox/indmoney access_token, "
"groww api_key_secret, kotakneo totp_mpin, deltaexchange api_key."
),
)
parser.add_argument(
"--environment",
default="prod",
choices=("prod", "uat"),
help="Kotak Neo SDK environment to use.",
)
parser.add_argument(
"--reads",
nargs="+",
default=["default"],
help=(
"Read-only calls: default, all, or broker-supported names. Common names include "
"profile funds positions holdings orders trades quotes ltp depth margin history. "
"Broker extras include expiry optionchain optiongreeks search timings holidays "
"limits scrip_master search_scrip quote_details market_depth."
),
)
parser.add_argument(
"--logout",
action="store_true",
help="Call the broker logout endpoint after probing; this may revoke access tokens.",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(sys.argv[1:] if argv is None else argv)
try:
return asyncio.run(run_probe(args.broker, args.method, args.reads, args.environment, logout=args.logout))
except KeyboardInterrupt:
print("probe: cancelled")
return 130
if __name__ == "__main__":
raise SystemExit(main())