DhanHQ on Google Colab with a ServLoci static IP¶

import os
import sys

SMOKE_TEST = "--smoke" in sys.argv  # used only by the automated Colab CLI test
PROXY_KEYS = (
    "ALL_PROXY", "all_proxy",
    "HTTP_PROXY", "http_proxy",
    "HTTPS_PROXY", "https_proxy",
)

# A rerun after trading-static-ip 0.3.0 may retain both monkey patches and
# proxy variables. Undo them before pip contacts pypi.org/simple/*.
if "servloci" in sys.modules:
    try:
        sys.modules["servloci"].unconfigure()
    except Exception:
        pass

for key in PROXY_KEYS:
    if "comm.servloci.in:1080" in os.environ.get(key, ""):
        os.environ.pop(key, None)

assert not any("comm.servloci.in:1080" in os.environ.get(k, "") for k in PROXY_KEYS)
print("Clean start: package installation will use Colab's direct connection.")

import subprocess

subprocess.check_call([
    sys.executable, "-m", "pip", "install", "-q", "--upgrade",
    "trading-static-ip==0.3.1", "dhanhq==2.2.0",
])
print("Dependencies installed.")

import importlib
import servloci

# Handles a deliberate top-to-bottom rerun without requiring another runtime restart.
servloci = importlib.reload(servloci)
assert servloci.__version__ == "0.3.1", servloci.__version__

if SMOKE_TEST:
    STATIC_IP_TOKEN = "sl_live_colab_cli_smoke_test"
    DHAN_CLIENT_ID = "1000000000"
    DHAN_ACCESS_TOKEN = "dhan_cli_smoke_test"
else:
    from google.colab import userdata

    def required_secret(name):
        value = userdata.get(name)
        if not value:
            raise ValueError(f"Missing Colab secret: {name}")
        return value.strip()

    STATIC_IP_TOKEN = required_secret("STATIC_IP_TOKEN")
    DHAN_CLIENT_ID = required_secret("DHAN_CLIENT_ID")
    DHAN_ACCESS_TOKEN = required_secret("DHAN_ACCESS_TOKEN")
    if not STATIC_IP_TOKEN.startswith("sl_live_"):
        raise ValueError("STATIC_IP_TOKEN must be the latest sl_live_... portal token")

print("Secrets loaded without displaying their values.")

import requests

direct_ip = None
if not SMOKE_TEST:
    direct_ip = requests.get("https://api.ipify.org", timeout=20).text.strip()
    print("Direct Colab IP:", direct_ip)

route = servloci.configure(
    token=STATIC_IP_TOKEN,
    broker="dhan",
)

# 0.3.1 patches requests/HTTPX without leaking SOCKS settings to pip subprocesses.
assert not any("comm.servloci.in:1080" in os.environ.get(k, "") for k in PROXY_KEYS)
print("ServLoci transport enabled for Dhan; pip remains direct.")

import ipaddress

if SMOKE_TEST:
    print("Colab CLI smoke mode: live token/IP check skipped.")
else:
    static_ip = requests.get("https://api.ipify.org", timeout=20).text.strip()
    parsed = ipaddress.ip_address(static_ip)
    assert parsed.version == 6, f"Expected the assigned IPv6, received {static_ip}"
    assert static_ip != direct_ip, "ServLoci route did not change the observed IP"
    print("ServLoci static IPv6:", static_ip)

# DhanHQ 2.2 uses DhanContext; older keyword-only examples no longer apply.
from dhanhq import DhanContext, dhanhq

dhan_context = DhanContext(DHAN_CLIENT_ID, DHAN_ACCESS_TOKEN)
dhan = dhanhq(dhan_context)

# DhanHQ created its own requests.Session after configure(), so its internal
# HTTPS calls are transparently routed through the ServLoci static IP.
assert dhan.dhan_http.session is not None
print("DhanHQ client ready through ServLoci.")

if SMOKE_TEST:
    print("Colab CLI smoke mode: authenticated Dhan call skipped.")
else:
    result = dhan.get_fund_limits()
    status = result.get("status") if isinstance(result, dict) else None
    if status != "success":
        raise RuntimeError(f"Dhan read-only call failed: {result.get('remarks', result)}")
    print("Dhan read-only call succeeded. Response data is intentionally hidden.")

order = {
    "security_id": "1333",       # example only; verify the current instrument ID
    "exchange_segment": dhan.NSE,
    "transaction_type": dhan.BUY,
    "quantity": 1,
    "order_type": dhan.MARKET,
    "product_type": dhan.INTRA,
    "price": 0,
}

DRY_RUN = True
if DRY_RUN:
    print("DRY RUN — no order sent:", order)
else:
    raise RuntimeError("Review risk, symbol, quantity, and market status before enabling live orders.")