@PravinJ @Dhan
I’m not a professional programmer, I created a trading strategy python code from YouTube tutorials and want to test it with real market data. Sandbox code test but full strategy code doesn’t test.
Buying a the ₹499 + GST one-month subscription doesn’t sense right now, because in one month 2 or 3 day data use rest of time spend most of it’s debugging my strategy code. So I request 2–3 days of API data access for testing my strategy code on live data.
If my coding trial succeeds, then I go for a monthly plan. So asking 2–3 days of data API subscription plan if possible provide
here attached code i want to test its ok or not help me
`import pdb
from Dhan_Tradehull import Tradehull
import pandas as pd
import talib
import numpy as np
import pandas_ta as ta
import requests
import datetime
import time
import json
BASE_URL = “https://api.dhan.co/v2”
— Load Credentials from config.json —
try:
with open(‘config.json’, ‘r’) as f:
config = json.load(f)
client_id = config[‘client_id’]
access_token = config[‘access_token’]
except FileNotFoundError:
print(“Error: config.json not found. Please create it with your client_id and access_token.”)
exit()
tsl = Tradehull(client_id, access_token)
stocks = [‘CIPLA’, ‘SBIN’, ‘MARUTI’, ‘BAJAJ-AUTO’, ‘DRREDDY’, ‘AXISBANK’, ‘EICHERMOT’,‘ADANIPORTS’, ‘WIPRO’, ‘ONGC’, ‘NESTLEIND’, ‘NTPC’, ‘POWERGRID’, ‘ETERNAL’,‘SUNPHARMA’, ‘BEL’, ‘ULTRACEMCO’, ‘ITC’, ‘TATACONSUM’, ‘HDFCBANK’, ‘HCLTECH’,‘TRENT’, ‘HINDUNILVR’, ‘M&M’, ‘ICICIBANK’, ‘JIOFIN’, ‘LT’, ‘RELIANCE’,‘INDIGO’, ‘COALINDIA’, ‘INFY’, ‘BAJFINANCE’, ‘ASIANPAINT’, ‘KOTAKBANK’,‘GRASIM’, ‘ADANIENT’, ‘MAXHEALTH’, ‘APOLLOHOSP’, ‘SBILIFE’, ‘BHARTIARTL’,‘HINDALCO’, ‘TATAMOTORS’, ‘BAJAJFINSV’, ‘TECHM’, ‘TITAN’, ‘SHRIRAMFIN’,‘JSWSTEEL’, ‘HDFCLIFE’, ‘TCS’, ‘TATASTEEL’, ‘ABB’, ‘ADANIENSOL’,‘ADANIGREEN’, ‘ADANIPOWER’, ‘AMBUJACEM’, ‘BAJAJHFL’, ‘BAJAJHLDNG’,‘BANKBARODA’, ‘BOSCHLTD’, ‘BPCL’, ‘BRITANNIA’, ‘CANBK’, ‘CGPOWER’,‘CHOLAFIN’, ‘DIVISLAB’, ‘DLF’, ‘DMART’, ‘ENRIN’, ‘GAIL’, ‘GODREJCP’, ‘HAL’,‘HAVELLS’, ‘HINDZINC’, ‘HYUNDAI’, ‘ICICIGI’, ‘INDHOTEL’, ‘IOC’, ‘IRFC’,‘JINDALSTEL’, ‘JSWENERGY’, ‘LICI’, ‘LODHA’, ‘LTIM’, ‘MAZDOCK’, ‘MOTHERSON’,‘NAUKRI’, ‘NIFTY 100’, ‘PFC’, ‘PIDILITIND’, ‘PNB’, ‘RECLTD’, ‘SHREECEM’,‘SIEMENS’, ‘SOLARINDS’, ‘TATAPOWER’, ‘TORNTPHARM’, ‘TVSMOTOR’, ‘UNITDSPR’,‘VBL’, ‘VEDL’, ‘ZYDUSLIFE’, ‘360ONE’, ‘ABCAPITAL’, ‘ACC’, ‘ALKEM’,‘APLAPOLLO’, ‘ASHOKLEY’, ‘ASTRAL’, ‘ATGL’, ‘AUBANK’, ‘AUROPHARMA’,‘BANKINDIA’, ‘BDL’, ‘BHARATFORG’, ‘BHARTIHEXA’, ‘BHEL’, ‘BIOCON’,‘BLUESTARCO’, ‘BSE’, ‘COCHINSHIP’, ‘COFORGE’, ‘COLPAL’, ‘CONCOR’,‘COROMANDEL’, ‘CUMMINSIND’, ‘DABUR’, ‘DIXON’, ‘EXIDEIND’, ‘FEDERALBNK’,‘FORTIS’, ‘GLENMARK’, ‘GMRAIRPORT’, ‘GODFRYPHLP’, ‘GODREJPROP’, ‘HDFCAMC’,‘HEROMOTOCO’, ‘HINDPETRO’, ‘HUDCO’, ‘IDEA’, ‘IDFCFIRSTB’, ‘IGL’, ‘INDIANB’,‘INDUSINDBK’, ‘INDUSTOWER’, ‘IRB’, ‘IRCTC’, ‘IREDA’, ‘ITCHOTELS’,‘JUBLFOOD’, ‘KALYANKJIL’, ‘KEI’, ‘KPITTECH’, ‘LICHSGFIN’, ‘LTF’, ‘LUPIN’,‘M&MFIN’, ‘MANKIND’, ‘MARICO’, ‘MFSL’, ‘MOTILALOFS’, ‘MPHASIS’, ‘MRF’,‘MUTHOOTFIN’, ‘NATIONALUM’, ‘NHPC’, ‘NIFTY 200’, ‘NMDC’, ‘NTPCGREEN’,‘NYKAA’, ‘OBEROIRLTY’, ‘OFSS’, ‘OIL’, ‘PAGEIND’, ‘PATANJALI’, ‘PAYTM’,‘PERSISTENT’, ‘PHOENIXLTD’, ‘PIIND’, ‘POLICYBZR’, ‘POLYCAB’, ‘POWERINDIA’,‘PREMIERENE’, ‘PRESTIGE’, ‘RVNL’, ‘SAIL’, ‘SBICARD’, ‘SONACOMS’, ‘SRF’,‘SUPREMEIND’, ‘SUZLON’, ‘SWIGGY’, ‘TATACOMM’, ‘TATAELXSI’, ‘TATATECH’,‘TIINDIA’, ‘TORNTPOWER’, ‘UNIONBANK’, ‘UPL’, ‘VMM’, ‘VOLTAS’, ‘WAAREEENER’,‘YESBANK’]
— Strategy Configuration —
CAPITAL_ALLOCATION_PERCENT = 0.50 # Use 50% of available capital per trade
RISK_REWARD_RATIO = 2.0 # 2:1 Reward to Risk
ATR_PERIOD = 14 # Period for ATR calculation for stop-loss
ATR_MULTIPLIER = 1.5 # Multiplier for ATR to set stop-loss distance
MAX_DAILY_LOSS = -2000 # Max loss in INR before activating kill switch
MAX_DAILY_TRADES = 6 # Max new trades per day
RISK_PER_TRADE_PERCENT = 0.02 # 2% of capital
MAX_RISK_PER_TRADE_RUPEES = 500 # Max risk in INR
def get_price_filtered_watchlist(full_watchlist):
“”"
Fetches LTP for the entire watchlist and filters it based on a price range.
This is more efficient than fetching historical data for each stock.
“”"
print(“— Pre-filtering watchlist by price (₹25 - ₹5000) —”)
filtered_stocks =
try:
ltp_data = tsl.get_ltp_data(full_watchlist)
if not ltp_data:
print(“Could not fetch LTP data for watchlist. Using full list.”)
return full_watchlist
for stock, price in ltp_data.items():
if 25 < price < 5000:
filtered_stocks.append(stock)
print(f"Price filter reduced watchlist from {len(full_watchlist)} to {len(filtered_stocks)} stocks.")
return filtered_stocks
except Exception as e:
print(f"Error during price filtering: {e}. Using full list.")
return full_watchlist
def get_daily_uptrend_stocks(stock_list):
“”"
Filters a list of stocks to find those in a daily uptrend based on Supertrend.
This should be run once per day.
“”"
print(“— Filtering watchlist based on Daily Supertrend (Price > Supertrend) —”)
uptrend_stocks =
for stock_symbol in stock_list:
try:
daily_data = tsl.get_historical_data(stock_symbol, ‘NSE’, ‘day’)
if daily_data is None or len(daily_data) < ATR_PERIOD + 2:
continue
daily_data.ta.supertrend(length=ATR_PERIOD, multiplier=3.0, append=True)
st_col = f'SUPERT_{ATR_PERIOD}_3.0'
daily_data.rename(columns={st_col: 'supertrend'}, inplace=True)
if daily_data['close'].iloc[-1] > daily_data['supertrend'].iloc[-1]:
uptrend_stocks.append(stock_symbol)
except Exception as e:
print(f"Could not check Supertrend for {stock_symbol}: {e}")
print(f"Found {len(uptrend_stocks)} stocks in a daily uptrend.")
return uptrend_stocks
def run_macd_strategy():
“”"
Runs a MACD crossover strategy on the specified watchlist.
- Buys when MACD line crosses above the signal line.
- Sells when MACD line crosses below the signal line.
“”"
print(“— Starting MACD Crossover Strategy —”)
# --- Initial Daily Scan (run once before the main loop) ---
print("\n--- Performing Initial Daily Scan ---")
price_filtered_stocks = get_price_filtered_watchlist(stocks)
uptrend_stocks = get_daily_uptrend_stocks(price_filtered_stocks)
print("--- Initial Daily Scan Complete. Starting trading loop. ---")
if not uptrend_stocks:
print("No stocks in uptrend found. The script will wait for the next day.")
# Daily state variables
kill_switch_activated = False
trades_today_count = 0
current_day = datetime.date.today()
# --- Fetch open positions at the start to avoid duplicates ---
try:
positions_df = tsl.get_positions()
open_positions = {} # Ensure it's always a dictionary
if positions_df is not None and not positions_df.empty:
open_positions_df = positions_df[positions_df['netQty'] != 0]
# Fetch order book to correctly map open positions to their entry orders and SL
order_book_df = tsl.get_orderbook()
if order_book_df is not None and not order_book_df.empty:
# We are using Cover Orders, so filter for 'CO'
co_orders = order_book_df[(order_book_df['productType'] == 'CO') & (order_book_df['orderStatus'] == 'TRADED')]
for index, row in open_positions_df.iterrows():
# Find the parent CO order for the open position
parent_order = co_orders[co_orders['tradingSymbol'] == row['tradingSymbol']].iloc[0] if not co_orders[co_orders['tradingSymbol'] == row['tradingSymbol']].empty else None
if parent_order is not None:
open_positions[row['tradingSymbol']] = {
'qty': row['netQty'], 'orderId': parent_order['orderId'],
'sl': parent_order['stopLoss'], 'direction': parent_order['transactionType']
}
if open_positions:
print(f"Found existing open positions: {list(open_positions.keys())}")
except Exception as e:
print(f"Warning: Could not fetch initial positions: {e}")
open_positions = {} # Default to an empty dictionary on error
while True:
try:
# --- Daily Reset Logic ---
if datetime.date.today() != current_day:
print("New day detected. Resetting daily limits and kill switch.")
current_day = datetime.date.today()
kill_switch_activated = False
trades_today_count = 0
# Re-run the daily scans for the new day
print("\n--- Performing Initial Daily Scan for new day---")
price_filtered_stocks = get_price_filtered_watchlist(stocks)
uptrend_stocks = get_daily_uptrend_stocks(price_filtered_stocks)
print("--- Initial Daily Scan Complete. Starting trading loop. ---")
# We do not clear open_positions to allow for swing trades
# --- Kill Switch Check ---
if kill_switch_activated:
print(f"Kill switch is active. No new trades will be placed today. Waiting for next day.")
time.sleep(60 * 5) # Sleep for 5 minutes before checking for a new day
continue
# Fetch available balance at the start of each scan
available_balance = tsl.get_balance()
if not available_balance or available_balance == 0:
print("Could not fetch balance or balance is zero. Waiting...")
time.sleep(60)
continue
# --- Dynamic Risk Calculation ---
risk_from_capital = available_balance * RISK_PER_TRADE_PERCENT
risk_amount = min(risk_from_capital, MAX_RISK_PER_TRADE_RUPEES)
# --- P&L Based Kill Switch ---
live_pnl = tsl.get_live_pnl()
print(f"Current Day P&L: ₹{live_pnl:.2f}")
if live_pnl is not None and live_pnl < MAX_DAILY_LOSS:
print(f"!!! MAX DAILY LOSS of ₹{MAX_DAILY_LOSS} REACHED. ACTIVATING KILL SWITCH. !!!")
print("--- Closing all open MIS/BO positions. ---")
tsl.cancel_all_orders(trade_type='CO') # This will square off open CO positions
kill_switch_activated = True
continue
# --- MANAGE OPEN POSITIONS (TRAILING SL & EXITS) ---
if open_positions:
print("--- Managing open positions ---")
for symbol, trade_info in list(open_positions.items()):
# --- 1. Check for MACD Exit Signal ---
hist_data_15min = tsl.get_intraday_data(symbol, 'NSE', 15)
if hist_data_15min is None or len(hist_data_15min) < 30: continue
macd, macdsignal, _ = talib.MACD(hist_data_15min['close'], fastperiod=12, slowperiod=26, signalperiod=9)
previous_macd = macd.iloc[-2]
latest_macd = macd.iloc[-1]
previous_signal = macdsignal.iloc[-2]
latest_signal = macdsignal.iloc[-1]
if trade_info['direction'] == 'BUY' and previous_macd > previous_signal and latest_macd < latest_signal:
print(f"EXIT (MACD Crossover): Closing BUY on {symbol} at {hist_data_15min['close'].iloc[-1]}")
tsl.order_placement(
tradingsymbol=symbol, exchange='NSE', quantity=trade_info['qty'], price=0,
trigger_price=0, order_type='MARKET', transaction_type='SELL', trade_type='MIS'
)
# Note: The original SL order from the CO will need to be cancelled manually or will be cancelled by Dhan upon position closure.
del open_positions[symbol]
continue # Move to next open position
# --- 2. Trail Stop-Loss (In-Memory for CO) ---
try:
hist_data = tsl.get_intraday_data(symbol, 'NSE', 15)
if hist_data is None or hist_data.empty:
continue
atr = talib.ATR(hist_data['high'], hist_data['low'], hist_data['close'], timeperiod=ATR_PERIOD)
latest_close = hist_data['close'].iloc[-1]
latest_atr = atr.iloc[-1]
if pd.isna(latest_atr):
continue
current_sl = trade_info['sl']
if trade_info['direction'] == 'BUY': # Trailing for BUY positions
new_sl = latest_close - (latest_atr * ATR_MULTIPLIER)
if new_sl > current_sl:
print(f"Trailing SL (in memory) for BUY {symbol} from {current_sl:.2f} to {new_sl:.2f}")
open_positions[symbol]['sl'] = new_sl
# Check if the new in-memory SL was breached by the latest candle's low
if latest_close < new_sl:
print(f"EXIT (Trailing SL Hit): Closing BUY on {symbol} at {latest_close}")
tsl.order_placement(
tradingsymbol=symbol, exchange='NSE', quantity=trade_info['qty'], price=0,
trigger_price=0, order_type='MARKET', transaction_type='SELL', trade_type='MIS'
)
del open_positions[symbol]
except Exception as e:
print(f"Error trailing SL for {symbol}: {e}")
# --- SCAN FOR NEW OPPORTUNITIES (only if balance > 0) ---
if available_balance <= 0:
print("Balance is zero or less. Skipping scan for new opportunities.")
time.sleep(15 * 60) # Wait for the next candle before checking again
continue
print(f"\nScanning for new trades at {datetime.datetime.now().strftime('%H:%M:%S')}...")
for stock_symbol in uptrend_stocks:
# --- Max Trades & Position Check ---
if trades_today_count >= MAX_DAILY_TRADES:
print("Max daily trades limit reached. No more new trades today.")
break # Stop scanning for new trades
if stock_symbol in open_positions.keys():
continue # Skip if position is already open
try:
# 1. Fetch 15-minute historical data
# Using tsl object from Dhan_Tradehull for robust data fetching
hist_data = tsl.get_intraday_data(stock_symbol, 'NSE', 15)
if hist_data is None or len(hist_data) < 30: # Need enough data for MACD and ATR
print(f"Not enough historical data for {stock_symbol}")
continue
# 2. Calculate MACD
macd, macdsignal, macdhist = talib.MACD(hist_data['close'], fastperiod=12, slowperiod=26, signalperiod=9)
# Calculate ATR for stop-loss
atr = talib.ATR(hist_data['high'], hist_data['low'], hist_data['close'], timeperiod=ATR_PERIOD) # Removed duplicate
# Calculate RSI
rsi = talib.RSI(hist_data['close'], timeperiod=14)
# Get the last two values to check for a crossover
latest_close = hist_data['close'].iloc[-1]
previous_close = hist_data['close'].iloc[-2]
latest_atr = atr.iloc[-1]
latest_rsi = rsi.iloc[-1]
# --- Price Filter ---
if pd.isna(latest_atr):
continue # Skip if ATR is not available
if pd.isna(latest_rsi):
continue # Skip if RSI is not available
latest_macd = macd.iloc[-1]
previous_macd = macd.iloc[-2]
latest_signal = macdsignal.iloc[-1]
previous_signal = macdsignal.iloc[-2]
# 3. Check for Crossover Signals
# Bullish Crossover: MACD crosses above Signal
if previous_macd < previous_signal and latest_macd > latest_signal and latest_close > previous_close and (58 < latest_rsi < 70):
print(f"BUY signal for {stock_symbol} at {latest_close}")
# --- Risk & Capital Management for BUY ---
stop_loss_price = latest_close - (latest_atr * ATR_MULTIPLIER)
risk_per_share = latest_close - stop_loss_price
if risk_per_share <= 0: continue
quantity = int(risk_amount / risk_per_share)
if quantity == 0:
print(f"Skipping {stock_symbol}, not enough capital for 1 share.")
continue
# Place a Cover Order (no fixed target)
order_id = tsl.order_placement(
tradingsymbol=stock_symbol, exchange='NSE', quantity=quantity, price=0,
trigger_price=round(stop_loss_price, 1), # SL for the Cover Order
order_type='MARKET', transaction_type='BUY', trade_type='CO'
)
if order_id:
print(f"Placed BUY Cover Order for {stock_symbol} | Qty: {quantity} | SL: {stop_loss_price:.2f}")
trades_today_count += 1
print(f"Trades taken today: {trades_today_count}/{MAX_DAILY_TRADES}")
open_positions[stock_symbol] = {
'qty': quantity, 'orderId': order_id,
'sl': stop_loss_price, 'direction': 'BUY'
}
except Exception as e:
print(f"Error processing {stock_symbol}: {e}")
print("--- Scan complete. Waiting for the next 15-minute candle... ---")
time.sleep(15 * 60) # Sleep for 15 minutes
except KeyboardInterrupt:
print("Strategy stopped by user.")
break
except Exception as e:
print(f"An unexpected error occurred in the main loop: {e}")
time.sleep(60) # Wait a minute before retrying
if name == “main”:
# You can choose which strategy to run here
run_macd_strategy()`