from dhanhq import dhanhq
import pandas as pd
import numpy as np
import logging
import datetime
import pytz
Set up logging
logging.basicConfig(level=logging.INFO)
Configuration
CLIENT_ID = # Your client ID
TRADING_API_KEY = “” # Replace with your API key
CAPITAL = 40000 # Starting capital
LOT_SIZE = 75
DAILY_TARGET = 0.05 # 5% daily target
MAX_RISK_PER_TRADE = 0.02 # 2% risk per trade
MAX_LOSS_PER_TRADE = 0.02 # 2% max loss per trade
DAILY_LOSS_LIMIT = 0.025 # 2.5% daily loss limit
TRAILING_STOP_LOSS_PERCENT = 0.01 # 1% trailing stop-loss
TIMEFRAMES = [“1m”, “5m”, “15m”] # Timeframes for intraday
EXCHANGE_SEGMENT = “NSE_FNO” # Segment for Nifty options
Initialize Dhan API
dhan = dhanhq(CLIENT_ID, TRADING_API_KEY)
Fetch Security ID for a Symbol
def fetch_security_id(symbol):
try:
filepath = “C:/Users/Surface Laptop Go 3/Desktop/New folder (2)/api-scrip-master.csv”
df = pd.read_csv(filepath, low_memory=False)
logging.info(f"Data fetched from {filepath}“)
security = df[df[‘SM_SYMBOL_NAME’] == symbol]
if not security.empty:
return int(security.iloc[0][‘SEM_SMST_SECURITY_ID’]) # Convert to standard integer
logging.error(f"Failed to find Security ID for {symbol}.”)
except Exception as e:
logging.exception(f"Error fetching Security ID for {symbol}: {e}")
return None
Fetch Expiry List
def fetch_expiry_list(security_id):
try:
response = dhan.get_option_expiry_list(
under_security_id=security_id,
under_exchange_segment=EXCHANGE_SEGMENT
)
if response.get(‘status’) == ‘success’:
return response[‘data’]
logging.error(f"Failed to fetch expiry list. Response: {response}“)
except Exception as e:
logging.exception(f"Error fetching expiry list: {e}”)
return None
Fetch Option Chain Data
def fetch_option_data(security_id, expiry):
try:
response = dhan.option_chain(
under_security_id=security_id,
under_exchange_segment=EXCHANGE_SEGMENT,
expiry=expiry
)
if response.get(‘status’) == ‘success’:
return response[‘data’]
logging.error(f"Failed to fetch option chain data. Response: {response}“)
except Exception as e:
logging.exception(f"Error fetching option data: {e}”)
return None
Calculate Indicators
def calculate_indicators(data):
df = pd.DataFrame(data)
df[‘Close’] = df[‘price’]
df[‘EMA9’] = df[‘Close’].ewm(span=9, adjust=False).mean()
df[‘EMA20’] = df[‘Close’].ewm(span=20, adjust=False).mean()
df[‘RSI’] = calculate_rsi(df[‘Close’], 14)
df[‘MACD’], df[‘MACD_Signal’], df[‘MACD_Hist’] = calculate_macd(df[‘Close’])
return df
RSI Calculation
def calculate_rsi(series, period):
delta = series.diff()
gain = delta.where(delta > 0, 0).rolling(window=period).mean()
loss = -delta.where(delta < 0, 0).rolling(window=period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
MACD Calculation
def calculate_macd(series):
ema12 = series.ewm(span=12, adjust=False).mean()
ema26 = series.ewm(span=26, adjust=False).mean()
macd = ema12 - ema26
signal = macd.ewm(span=9, adjust=False).mean()
return macd, signal, macd - signal
Fetch Market Data
def fetch_market_data(exchange_token, timeframe):
try:
response = dhan.get_market_data(
exchange_token=exchange_token,
exchange_segment=EXCHANGE_SEGMENT,
timeframe=timeframe
)
if response.get(‘status’) == ‘success’:
return response[‘data’]
logging.error(f"Failed to fetch market data. Response: {response}“)
except Exception as e:
logging.exception(f"Error fetching market data: {e}”)
return None
Execute Trade
def execute_trade(security_id, transaction_type, quantity):
try:
response = dhan.place_order(
security_id=security_id,
exchange_segment=EXCHANGE_SEGMENT,
transaction_type=transaction_type,
quantity=quantity,
order_type=“MARKET”,
product_type=“INTRA”,
price=0
)
logging.info(f"Trade executed: {response}“)
except Exception as e:
logging.exception(f"Error executing trade: {e}”)
Backtest Strategy
def backtest_strategy(data, security_id):
df = calculate_indicators(data)
df[‘Signal’] = “HOLD”
df.loc[(df[‘EMA9’] > df[‘EMA20’]) & (df[‘RSI’] < 30), ‘Signal’] = “BUY”
df.loc[(df[‘EMA9’] < df[‘EMA20’]) & (df[‘RSI’] > 70), ‘Signal’] = “SELL”
balance = CAPITAL
position = 0
for i in range(1, len(df)):
current_price = df['Close'][i]
signal = df['Signal'][i]
if signal == "BUY" and position == 0:
position = int((balance * MAX_RISK_PER_TRADE) / current_price)
execute_trade(security_id, "BUY", position)
elif signal == "SELL" and position > 0:
execute_trade(security_id, "SELL", position)
position = 0
logging.info(f"Final balance after backtesting: {balance}")
Main Execution
if name == “main”:
logging.info(“Algo Trading Script Started”)
# Fetch Nifty Security ID
nifty_security_id = fetch_security_id("NIFTY")
if nifty_security_id:
expiry_list = fetch_expiry_list(nifty_security_id)
if expiry_list:
expiry_date = expiry_list[0] # Example: use the nearest expiry
option_data = fetch_option_data(nifty_security_id, expiry_date)
if option_data:
market_data = fetch_market_data(option_data[0]['exchange_token'], "5m")
if market_data:
backtest_strategy(market_data, nifty_security_id)
else:
logging.error("No market data available for backtesting.")
else:
logging.error("Failed to fetch option data.")
else:
logging.error("Failed to fetch expiry list.")
else:
logging.error("Failed to fetch Nifty Security ID.")