Trading with DhanHQ means refreshing your API token every 24 hours—a tedious, time-sensitive task. This automation simplifies and secures the token refresh process, alerts you in advance when tokens expire, and lets you monitor and control your bot remotely via Telegram commands.
Prerequisites
- A valid Dhan trading account with API access.
- Telegram account for receiving bot alerts.
- Basic knowledge of Python programming.
- A device running Python 3.8 or newer
Step 1: Enable API and Generate Credentials
- Log in to your Dhan Web Portal.
- Enable TOTP authentication via API settings.
- Scan the provided QR code with your authenticator app to get the TOTP secret.
- Generate your API credentials — Client ID, API Key, and API Secret.
- Important: Store these credentials securely and never share them publicly.
Step 2: Set Up Your Telegram Bot
- Use Telegram’s BotFather to create a new bot and save the bot token.
- Message
@userinfoboton Telegram to retrieve your chat ID. - This bot token and chat ID enable your trading bot to notify you of critical updates and accept commands.
Step 3: Configure Your Bot
Create a JSON configuration file named dhan_config.json with the following structure (replace values with your credentials):
json
{
"client_id": "YOUR_CLIENT_ID",
"api_key": "YOUR_API_KEY",
"api_secret": "YOUR_API_SECRET",
"totp_secret": "YOUR_TOTP_SECRET",
"telegram_bot_token": "YOUR_TELEGRAM_BOT_TOKEN",
"telegram_chat_id": "YOUR_TELEGRAM_CHAT_ID",
"admin_chat_id": "YOUR_ADMIN_CHAT_ID"
}
Step 4: Running the Bot
- Install required Python packages like
pyotp,requests, andschedule. - Run the bot on your device. It will:
- Automatically handle token generation and refresh using TOTP.
- Monitor token validity and alert you via Telegram 2 hours before expiration.
- Accept Telegram commands such as
/start,/status,/refresh, and/accountfor real-time bot control and account overview.
Key Telegram Commands
/start— View bot and system status./status— Check API token validity and bot health./refresh— Manually trigger token refresh./account— Get a summary of your account holdings and funds./help— List available commands and usage instructions.
Security and Best Practices
- Securely store your credentials; never share the
dhan_config.jsoncontents publicly. - Use mobile browsers for TOTP token refresh login screens (desktop browsers may hang).
Configuration File Example (dhan_config.json):
json
{
"client_id": "YOUR_CLIENT_ID",
"api_key": "YOUR_API_KEY",
"api_secret": "YOUR_API_SECRET",
"totp_secret": "YOUR_TOTP_SECRET",
"telegram_bot_token": "YOUR_TELEGRAM_BOT_TOKEN",
"telegram_chat_id": "YOUR_TELEGRAM_CHAT_ID",
"admin_chat_id": "YOUR_ADMIN_CHAT_ID"
}
Bot Initialization Snippet
python
class DhanCompleteBot:
def __init__(self, config_file='dhan_config.json'):
self.load_config()
self.setup_token_management()
self.setup_telegram_bot()
self.bot_active = True
self.waiting_for_token = False
# further initialization...
Loading Configuration from JSON
python
def load_config(self):
with open(self.config_file, 'r') as f:
config = json.load(f)
self.client_id = config.get('client_id')
self.api_key = config.get('api_key')
self.api_secret = config.get('api_secret')
self.totp_secret = config.get('totp_secret')
self.bot_token = config.get('telegram_bot_token')
self.chat_id = config.get('telegram_chat_id')
Generating TOTP Code for Authentication
python
def generate_totp_code(self):
totp = pyotp.TOTP(self.totp_secret)
return totp.now()
Starting Token Refresh Example Message (Telegram)
python
login_url = f"https://auth.dhan.co/login/consentApp-login?consentAppId={consent_id}"
totp_code = self.generate_totp_code()
message = f"""
🔄 Token Refresh Started
Login URL:
{login_url}
TOTP Code: `{totp_code}`
Instructions:
1. Open URL on MOBILE browser
2. Login with Dhan credentials
3. Use TOTP code above
4. Copy the callback URL after login
5. Send it back to this bot
Bot accepts:
- Callback URL
- Login URL
- Direct tokenId
"""
self.send_message(message, chat_id)
Telegram Bot Commands Overview
text
/start - Show system status
/status - Check token & monitoring status
/account - View account balance & portfolio
/refresh - Force token refresh
/help - Show help message
python
#!/usr/bin/env python3
"""
Dhan API TOTP Authentication with Telegram Integration
=====================================================
Automates token management using TOTP for Dhan API key-secret authentication
with Telegram bot for notifications and remote commands.
Prerequisites:
- Config file dhan_config.json with:
{
"client_id": "YOUR_CLIENT_ID",
"api_key": "YOUR_API_KEY",
"api_secret": "YOUR_API_SECRET",
"totp_secret": "YOUR_TOTP_SECRET",
"telegram_bot_token": "YOUR_TELEGRAM_BOT_TOKEN",
"telegram_chat_id": "YOUR_TELEGRAM_CHAT_ID"
}
"""
import json
import logging
import threading
import time
from datetime import datetime, timedelta
import os
import requests
import pyotp
import schedule
from telegram import Bot, Update
from telegram.ext import Updater, CommandHandler, CallbackContext
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
class DhanTOTPBot:
def __init__(self, config_path="dhan_config.json"):
basedir = os.path.dirname(os.path.abspath(__file__))
self.config_path = os.path.join(basedir, config_path)
self.token_file = os.path.join(basedir, "dhan_token.json")
self.load_config()
self.load_token()
self.bot = Bot(token=self.telegram_bot_token)
self.updater = Updater(token=self.telegram_bot_token, use_context=True)
self.dispatcher = self.updater.dispatcher
self.register_handlers()
self.start_scheduler()
def load_config(self):
with open(self.config_path, "r") as f:
cfg = json.load(f)
self.client_id = cfg.get("client_id")
self.api_key = cfg.get("api_key")
self.api_secret = cfg.get("api_secret")
self.totp_secret = cfg.get("totp_secret")
self.telegram_bot_token = cfg.get("telegram_bot_token")
self.telegram_chat_id = cfg.get("telegram_chat_id")
required = [self.client_id, self.api_key, self.api_secret, self.totp_secret,
self.telegram_bot_token, self.telegram_chat_id]
if any(x is None for x in required):
raise ValueError("Missing required configuration values")
def load_token(self):
if os.path.exists(self.token_file):
with open(self.token_file, "r") as f:
data = json.load(f)
self.access_token = data.get("access_token")
expiry = data.get("expiry_time")
self.expiry_time = datetime.fromisoformat(expiry) if expiry else None
logging.info("Loaded saved token.")
else:
self.access_token = None
self.expiry_time = None
def save_token(self):
with open(self.token_file, "w") as f:
json.dump({
"access_token": self.access_token,
"expiry_time": self.expiry_time.isoformat() if self.expiry_time else None
}, f, indent=2)
logging.info("Access token saved.")
def is_token_valid(self):
if not self.access_token or not self.expiry_time:
return False
return datetime.now() + timedelta(minutes=5) < self.expiry_time
def generate_totp(self):
totp = pyotp.TOTP(self.totp_secret)
return totp.now()
def authenticate(self):
# Generate TOTP code
totp = self.generate_totp()
logging.info(f"Generated TOTP: {totp}")
url = "https://auth.dhan.co/app/generateApp-consent"
headers = {"Content-Type": "application/json"}
data = {
"clientId": self.client_id,
"apiKey": self.api_key,
"apiSecret": self.api_secret,
"totp": totp
}
try:
resp = requests.post(url, json=data, headers=headers, timeout=30)
resp.raise_for_status()
consent = resp.json()
consent_id = consent.get("consentAppId")
logging.info(f"Consent ID: {consent_id}")
return consent_id
except Exception as e:
logging.error(f"Consent generation failed: {e}")
return None
def get_access_token(self, consent_id):
# Call consume consent with consent id to get access token
if not consent_id:
return False
url = f"https://auth.dhan.co/app/consumeApp-consent?tokenId={consent_id}"
headers = {
"app_id": self.api_key,
"app_secret": self.api_secret
}
try:
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
data = resp.json()
self.access_token = data.get("accessToken", None)
exp_time_str = data.get("expiryTime", None)
if exp_time_str:
self.expiry_time = datetime.strptime(exp_time_str[:19], "%Y-%m-%d %H:%M:%S")
self.save_token()
logging.info(f"Access token obtained, expires at {self.expiry_time}")
return True
except Exception as e:
logging.error(f"Failed to get access token: {e}")
return False
def refresh_token(self):
# Full refresh flow: generate consent -> get token
consent_id = self.authenticate()
if not consent_id:
return False
return self.get_access_token(consent_id)
# Telegram command handlers
def start(self, update: Update, context: CallbackContext):
valid_status = "✅ Valid" if self.is_token_valid() else "❌ Expired"
context.bot.send_message(update.effective_chat.id,
f"Dhan TOTP Bot running.\nToken status: {valid_status}")
def status(self, update: Update, context: CallbackContext):
if self.is_token_valid():
remain = self.expiry_time - datetime.now()
hours = remain.total_seconds() / 3600
context.bot.send_message(update.effective_chat.id,
f"Access token valid for {hours:.2f} hours.\nExpires at: {self.expiry_time}")
else:
context.bot.send_message(update.effective_chat.id,
"Access token expired or invalid. Use /refresh to renew.")
def refresh(self, update: Update, context: CallbackContext):
success = self.refresh_token()
if success:
context.bot.send_message(update.effective_chat.id,
f"✅ Token refreshed successfully!\nExpires at: {self.expiry_time}")
else:
context.bot.send_message(update.effective_chat.id,
"❌ Failed to refresh token. Check logs for details.")
def help(self, update: Update, context: CallbackContext):
help_text = (
"/start - Bot status\n"
"/status - Token validity\n"
"/refresh - Refresh token using TOTP\n"
"/help - This message"
)
context.bot.send_message(update.effective_chat.id, help_text)
def register_handlers(self):
self.dispatcher.add_handler(CommandHandler("start", self.start))
self.dispatcher.add_handler(CommandHandler("status", self.status))
self.dispatcher.add_handler(CommandHandler("refresh", self.refresh))
self.dispatcher.add_handler(CommandHandler("help", self.help))
def start_scheduler(self):
def check_expiry():
if self.is_token_valid():
remaining = self.expiry_time - datetime.now()
hrs = remaining.total_seconds() / 3600
if hrs < 2:
self.bot.send_message(chat_id=self.telegram_chat_id,
text=f"⚠️ Token expiring soon in {hrs:.2f} hours! Use /refresh to renew.")
else:
self.bot.send_message(chat_id=self.telegram_chat_id,
text="❌ Token expired or invalid! Use /refresh to renew.")
schedule.every(30).minutes.do(check_expiry)
def run_scheduler():
while True:
schedule.run_pending()
time.sleep(60)
thread = threading.Thread(target=run_scheduler, daemon=True)
thread.start()
def run(self):
self.register_handlers()
logging.info("Starting Dhan TOTP Bot...")
self.updater.start_polling()
self.updater.idle()
if __name__ == "__main__":
bot = DhanTOTPBot()
bot.run()