Skip to content

Instantly share code, notes, and snippets.

@theankushjain

theankushjain/gist:f30f6e0793bbd10cf9d96e98cc8aa089

Last active March 12, 2026 07:53
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save theankushjain/f30f6e0793bbd10cf9d96e98cc8aa089 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save theankushjain/f30f6e0793bbd10cf9d96e98cc8aa089 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
gistfile1.txt
import os,time,threading as th,datetime as dt,requests as rq,pytz,pandas as pd,pandas_ta_classic as ta,upstox_client as uc
from flask import Flask,request as req,jsonify,redirect,session
from functools import wraps
AK,AS="137d58dd-52f4-4856-a0b9-3b177e3b0274","pe5hsplmgq"
TM,PIN,SK=False,"261295","super_secret_trading_key_123"
LEV,WM,IGW=5,2000.0,False
trk,pos,hist,inst={},{},[],{}
TF,LF,IST="upstox_token.txt","trading.log",pytz.timezone('Asia/Kolkata')
app=Flask(__name__);app.secret_key=SK;api,hapi=None,None
def login_required(f):
@wraps(f)
def d(*a,**kw):return f(*a,**kw) if session.get('logged_in') else redirect('/login')
return d
def init_upstox():
global api,hapi;t=open(TF).read().strip() if os.path.exists(TF) else ""
if t:c=uc.Configuration();c.access_token=t;cl=uc.ApiClient(c);api,hapi=uc.OrderApi(cl),uc.HistoryApi(cl)
def fetch_instruments():
global inst
try:d=pd.read_csv("https://assets.upstox.com/market-quote/instruments/exchange/NSE.csv.gz");inst={str(r['tradingsymbol']).upper():r['instrument_key'] for _,r in d[d['instrument_type'].isin(['EQ','EQUITY'])].iterrows()}
except:pass
def get_live_data(sym,i='1minute'):
k=inst.get(sym.upper());d=pd.DataFrame()
if not k:return d
try:
c1=[];c2=[]
try:r=hapi.get_intra_day_candle_data(k,i,"2.0");c1=r.data.candles if r.data and r.data.candles else []
except:pass
try:td=dt.datetime.now(IST);r2=hapi.get_historical_candle_data1(k,i,(td-dt.timedelta(days=1)).strftime('%Y-%m-%d'),(td-dt.timedelta(days=10)).strftime('%Y-%m-%d'),"2.0");c2=r2.data.candles if r2.data and r2.data.candles else []
except:pass
if c1 or c2:
d=pd.DataFrame(c2+c1,columns=['timestamp','Open','High','Low','Close','Volume','OI']);d['timestamp']=pd.to_datetime(d['timestamp']);return d.drop_duplicates(subset=['timestamp']).set_index('timestamp').sort_index()
except:pass
return d
def p_order(sym,qty,side):
k=inst.get(sym.upper())
if not k:return False
if TM:return "TEST_ORDER"
try:return api.place_order(uc.PlaceOrderRequest(quantity=int(qty),product="I",validity="DAY",price=0.0,tag="algo",instrument_token=k,order_type="MARKET",transaction_type=side,disclosed_quantity=0,trigger_price=0.0,is_amo=False),"2.0").data.order_id
except:return False
def add_at(df):
if len(df)<15:df['AT_Sell']=False;return df
df['TR']=ta.true_range(df['High'],df['Low'],df['Close']);df['ATR']=df['TR'].rolling(14).mean()
df['MFI_14']=ta.mfi(df['High'],df['Low'],df['Close'],df['Volume'],length=14)
u=df['Low']-df['ATR'];d=df['High']+df['ATR'];m=df['MFI_14'];at=[0.0]*len(df);uv=u.values;dv=d.values;mv=m.values
for i in range(14,len(df)):
if at[i-1]==0.0:at[i]=uv[i] if mv[i]>=50 else dv[i];continue
p=at[i-1];at[i]=(p if uv[i]<p else uv[i]) if mv[i]>=50 else (p if dv[i]>p else dv[i])
df['AT']=at;df['AT_Sell']=(df['AT']<df['AT'].shift(2))&(df['AT'].shift(1)>=df['AT'].shift(3));return df
@app.route('/login',methods=['GET','POST'])
def login():
if req.method=='POST' and req.form.get('pin')==PIN:session['logged_in']=True;return redirect('/')
return '<form method="post"><input type="password" name="pin"><button>Login</button></form>'
@app.route('/data')
@login_required
def data():
wp=mp=0.0;ar=hr=""
for s,p in pos.items():
d=get_live_data(s,'1minute');ltp=d.iloc[-1]['Close'] if not d.empty else p['buy_price'];pnl=(ltp-p['buy_price'])*p['qty']
if p['source']=='webhook':wp+=pnl
else:mp+=pnl
ar+=f"<tr><td>{s}</td><td>{p['qty']}</td><td>{p['buy_price']:.2f}</td><td>{ltp:.2f}</td><td>{pnl:.2f}</td><td><form action='/close' method='post'><input type='hidden' name='sym' value='{s}'><button>Exit</button></form></td></tr>"
for s in trk:
d=get_live_data(s,'1minute');st="Track"
if not d.empty and len(d)>=15:d.ta.ema(length=9,append=True);r=d.iloc[-1];st=f"LTP:{r['Close']:.2f}|9EMA:{r['EMA_9']:.2f}"
ar+=f"<tr><td><b>{s}</b></td><td colspan='5' style='color:#f39c12'>{st}</td></tr>"
for h in hist[::-1]:
if h.get('source')=='webhook':wp+=h['pnl']
else:mp+=h['pnl']
hr+=f"<tr><td>{h['sym']}</td><td>{h['qty']}</td><td>{h['buy_price']:.2f}</td><td>{h['sold_at']}</td><td>{h['pnl']:.2f}</td></tr>"
lg="<br>".join(open(LF).readlines()[-10:]) if os.path.exists(LF) else "No Logs"
return jsonify({'wp':wp,'mp':mp,'ar':ar,'hr':hr,'lg':lg})
@app.route('/')
@login_required
def dash():
s1="TEST" if TM else "LIVE";s2="NO TIME LMT" if IGW else "9:15-11"
return f"""<style>body{{font-family:sans-serif;background:#f4f7f6;padding:20px;color:#333}} .card{{background:#fff;padding:20px;border-radius:10px;box-shadow:0 3px 8px rgba(0,0,0,0.1);max-width:900px;margin:auto;margin-bottom:20px}} table{{width:100%;border-collapse:collapse;margin-top:10px}} th,td{{padding:12px;text-align:left;border-bottom:1px solid #eee}} th{{background:#f8f9fa}} button{{padding:8px 15px;background:#007bff;color:#fff;border:none;border-radius:4px;cursor:pointer}} input{{padding:8px;border:1px solid #ccc;border-radius:4px}} .flex{{display:flex;gap:10px;align-items:center}} .mb{{margin-bottom:15px}}</style><script>setInterval(()=>{{fetch('/data').then(r=>r.json()).then(d=>{{document.getElementById('wp').innerHTML='₹'+d.wp.toFixed(2);document.getElementById('mp').innerHTML='₹'+d.mp.toFixed(2);document.getElementById('tb').innerHTML='<tr><th>Sym</th><th>Qty</th><th>Buy</th><th>LTP</th><th>PnL</th><th>Act</th></tr>'+d.ar;document.getElementById('lg').innerHTML=d.lg}})}},1500)</script><div class="card"><div class="flex mb" style="background:#333;color:#fff;padding:15px;border-radius:8px;justify-content:space-between"><b>SYSTEM: {s1} | TIME: {s2}</b><div class="flex"><form action="/set_wm" method="post" style="margin:0" class="flex"><input name="wm" value="{WM}" style="width:70px" title="WH Amount"><button style="background:#28a745">Set WH</button></form><form action="/tm" method="post" style="margin:0"><button style="background:#ffc107;color:#000">Mode</button></form> <form action="/tw" method="post" style="margin:0"><button style="background:#17a2b8">Time</button></form></div></div><div class="flex mb"><div style="flex:1;background:#e3f2fd;padding:15px;border-radius:8px">Webhook PnL: <b id="wp">₹0.00</b></div><div style="flex:1;background:#f1f8e9;padding:15px;border-radius:8px">Manual PnL: <b id="mp">₹0.00</b></div></div><div class="flex mb"><form action="/manual" method="post" class="flex"><input name="sym" placeholder="Symbol"><input name="m" value="5000" style="width:100px"><button>Track</button></form><form action="/token" method="post" class="flex" style="margin-left:auto"><a href="https://api.upstox.com/v2/login/authorization/dialog?response_type=code&client_id={AK}&redirect_uri=https://127.0.0.1" target="_blank" style="font-size:12px">Code -></a><input name="code" placeholder="Auth Code"><button style="background:#28a745">Connect</button></form></div><table id="tb"><tr><th>Sym</th><th>Qty</th><th>Buy</th><th>LTP</th><th>PnL</th><th>Act</th></tr></table><div id="lg" style="background:#1e1e1e;color:#0f0;padding:15px;margin-top:15px;height:120px;overflow-y:auto;font-family:monospace;border-radius:8px;font-size:12px">Loading...</div></div><div class="card"><h3>Backtest Simulator</h3><form action="/backtest" method="post" enctype="multipart/form-data" class="flex"><input type="file" name="csv_file" accept=".csv" required><input type="date" name="sim_date"><span style="font-size:11px;color:#555">Blank=All Dates</span><button style="background:#6f42c1">Check P&L</button></form></div>"""
@app.route('/set_wm',methods=['POST'])
@login_required
def set_wm():global WM;WM=float(req.form.get('wm',WM));return redirect('/')
@app.route('/tm',methods=['POST'])
@login_required
def tm():global TM;TM=not TM;return redirect('/')
@app.route('/tw',methods=['POST'])
@login_required
def tw():global IGW;IGW=not IGW;return redirect('/')
@app.route('/token',methods=['POST'])
@login_required
def up_token():
c=req.form.get('code','').strip();c=c.split("code=")[1].split("&")[0] if "code=" in c else c;r=rq.post("https://api.upstox.com/v2/login/authorization/token",data={'code':c,'client_id':AK,'client_secret':AS,'redirect_uri':'https://127.0.0.1','grant_type':'authorization_code'})
if r.status_code==200:open(TF,'w').write(r.json().get('access_token'));init_upstox()
return redirect('/')
@app.route('/manual',methods=['POST'])
@login_required
def manual():
s=req.form.get('sym','').upper().strip()
if s in inst:trk[s]={'source':'manual','margin':float(req.form.get('m',5000))};open(LF,"a").write(f"[{dt.datetime.now(IST).strftime('%H:%M:%S')}] TRACKING: {s}\n")
else:open(LF,"a").write(f"[{dt.datetime.now(IST).strftime('%H:%M:%S')}] INV SYM: {s}\n")
return redirect('/')
@app.route('/close',methods=['POST'])
@login_required
def close_t():
s=req.form.get('sym')
if s in pos and p_order(s,pos[s]['qty'],"SELL"):
d=get_live_data(s,'1minute');ep=d.iloc[-1]['Close'] if not d.empty else pos[s]['buy_price'];hist.append({**pos[s],'sym':s,'sold_at':'Manual','pnl':(ep-pos[s]['buy_price'])*pos[s]['qty']});del pos[s]
return redirect('/')
@app.route('/webhook',methods=['POST'])
def wh():
n=dt.datetime.now(IST).time()
if 'stocks' in req.json and (TM or IGW or (dt.time(9,15)<=n<=dt.time(11,0))):
for s in req.json['stocks'].split(','):
s=s.strip().upper()
if s in inst and s not in pos:trk[s]={'source':'webhook','margin':WM};open(LF,"a").write(f"[{dt.datetime.now(IST).strftime('%H:%M:%S')}] WH ADDED: {s}\n")
return jsonify({"s":"success"})
@app.route('/backtest',methods=['POST'])
@login_required
def backtest():
f=req.files.get('csv_file');ds=req.form.get('sim_date')
if not f:return "Miss inputs",400
df=pd.read_csv(f).dropna(subset=['date'])
try:df['dt']=pd.to_datetime(df['date'],format='%d-%m-%Y %H:%M')
except:return "Inv Date Fmt",400
if ds:
td=pd.to_datetime(ds).date();dcsv=df[df['dt'].dt.date==td]
if dcsv.empty:return f"<h3>No alerts {td}</h3><a href='/'>Back</a>"
res,tpnl=run_backtest(dcsv,ds)
if res=="TOKEN_ERROR":return "<body style='font-family:sans-serif;padding:20px;text-align:center;background:#fff3f3;color:#d9534f;border:2px solid #d9534f;border-radius:8px;max-width:500px;margin:50px auto'><h2>⚠️ Upstox API Token Expired!</h2><p>Please connect Upstox.</p><a href='/' style='padding:10px;background:#007bff;color:#fff;border-radius:4px'>Back</a></body>"
c="green" if tpnl>=0 else "red";h=f"<body style='font-family:sans-serif;padding:20px;background:#f4f7f6'><div style='background:#fff;padding:20px;border-radius:10px;max-width:900px;margin:auto'><h2>Results {ds}</h2><h3 style='color:{c}'>Total PnL: ₹{tpnl:.2f}</h3><table style='width:100%;text-align:left' border='1'><tr><th>Sym</th><th>Buy Time</th><th>Buy</th><th>Sell Time</th><th>Sell</th><th>Qty</th><th>PnL</th><th>PnL %</th></tr>"
for r in sorted(res,key=lambda x:x['buy_time']):
pct=(r['pnl']/(r['buy_price']*r['qty']))*100 if r['qty'] else 0;c="green" if r['pnl']>0 else "red";h+=f"<tr><td>{r['sym']}</td><td>{r['buy_time']}</td><td>₹{r['buy_price']:.2f}</td><td>{r['sold_at']}</td><td>₹{r['sell_price']:.2f}</td><td>{r['qty']}</td><td style='color:{c}'><b>₹{r['pnl']:.2f}</b></td><td style='color:{c}'><b>{pct:.2f}%</b></td></tr>"
return h+"</table><br><a href='/'>Back</a></div></body>"
else:
uds=[d for d in df['dt'].dt.date.unique() if pd.notna(d)];gt=0.0;h="<body style='font-family:sans-serif;padding:20px;background:#f4f7f6'><div style='background:#fff;padding:20px;border-radius:10px;max-width:900px;margin:auto'><h2>All Dates Summary</h2><table style='width:100%;text-align:left' border='1'><tr><th>Date</th><th>Trades</th><th>PnL</th></tr>";th="<br><h2>All Trades Details</h2><table style='width:100%;text-align:left;font-size:14px' border='1'><tr><th>Date</th><th>Sym</th><th>Buy Time</th><th>Buy</th><th>Sell Time</th><th>Sell</th><th>Qty</th><th>PnL</th><th>PnL %</th></tr>";atr=[]
for ud in sorted(uds):
dstr=ud.strftime('%Y-%m-%d');dcsv=df[df['dt'].dt.date==ud];res,tpnl=run_backtest(dcsv,dstr)
if res=="TOKEN_ERROR":return "<h2>Token Expired!</h2><a href='/'>Back</a>"
gt+=tpnl;c="green" if tpnl>=0 else "red";h+=f"<tr><td>{dstr}</td><td>{len(res)}</td><td style='color:{c}'><b>₹{tpnl:.2f}</b></td></tr>"
for r in res:atr.append({'d':dstr,**r})
for r in sorted(atr,key=lambda x:x['d']+' '+x['buy_time']):
pct=(r['pnl']/(r['buy_price']*r['qty']))*100 if r['qty'] else 0;c="green" if r['pnl']>0 else "red";th+=f"<tr><td>{r['d']}</td><td>{r['sym']}</td><td>{r['buy_time']}</td><td>₹{r['buy_price']:.2f}</td><td>{r['sold_at']}</td><td>₹{r['sell_price']:.2f}</td><td>{r['qty']}</td><td style='color:{c}'><b>₹{r['pnl']:.2f}</b></td><td style='color:{c}'><b>{pct:.2f}%</b></td></tr>"
gc="green" if gt>=0 else "red";h+=f"</table><h3 style='color:{gc}'>Grand Total PnL: ₹{gt:.2f}</h3>{th}</table><br><a href='/'>Back</a></div></body>"
return h
def run_backtest(dcsv,ds):
init_upstox();bt,bp,bh={},{},[]
if not hapi:return "TOKEN_ERROR",0
sd=(pd.to_datetime(ds)-pd.Timedelta(days=10)).strftime('%Y-%m-%d');token_valid=False;d1m={}
for s in dcsv['symbol'].unique():
if pd.isna(s):continue
k=inst.get(str(s).upper())
if not k:continue
try:
r=hapi.get_historical_candle_data1(k,'1minute',ds,sd,"2.0");token_valid=True
if r.data and r.data.candles:d=pd.DataFrame(r.data.candles,columns=['timestamp','Open','High','Low','Close','Volume','OI']);d['timestamp']=pd.to_datetime(d['timestamp']).dt.tz_localize(None);d1m[str(s).upper()]=add_at(d.set_index('timestamp').sort_index())
time.sleep(0.05)
except:pass
if not token_valid and len(dcsv['symbol'].unique())>0:return "TOKEN_ERROR",0
for t in pd.date_range(f"{ds} 09:15:00",f"{ds} 15:30:00",freq='1min'):
if t.time()>=dt.time(15,12):
for s,p in list(bp.items()):
d=d1m.get(s);ep=p['buy_price']
if d is not None and not d[d.index<=t].empty:ep=d[d.index<=t].iloc[-1]['Close']
bh.append({'sym':s,'buy_time':p['buy_time'],'buy_price':p['buy_price'],'sold_at':'15:12(AutoSq)','sell_price':ep,'qty':p['qty'],'pnl':(ep-p['buy_price'])*p['qty']})
bp.clear();bt.clear();break
if t.time()<=dt.time(11,0):
for _,r in dcsv[dcsv['dt']==t].iterrows():
s=str(r['symbol']).upper()
if s not in bp and s in d1m:bt[s]={'margin':WM}
for s,m in list(bt.items()):
d=d1m.get(s)
if d is None or len(d[d.index<=t])<15:continue
dc=d[d.index<=t].copy();dc.ta.ema(length=9,append=True);dc['VW']=((dc['High']+dc['Low']+dc['Close'])/3*dc['Volume']).groupby(dc.index.date).cumsum()/dc['Volume'].groupby(dc.index.date).cumsum();r=dc.iloc[-1];l=r['EMA_9'];u=l*1.0015
if t.time()<=dt.time(11,0) and (r['Low']<=u) and (r['Close']>l) and (r['Close']>r['VW']) and (dc.iloc[-2]['Close']>l):
q=int((m['margin']*LEV)/r['Close'])
if q>0:del bt[s];bp[s]={'qty':q,'buy_price':r['Close'],'buy_time':t.strftime('%H:%M')}
for s,p in list(bp.items()):
d1t=d1m.get(s);d1t=d1t[d1t.index<=t] if d1t is not None else pd.DataFrame()
if d1t.empty:continue
row=d1t.iloc[-1]
if row['AT_Sell']==True:
ep=row['Close'];bh.append({'sym':s,'buy_time':p['buy_time'],'buy_price':p['buy_price'],'sold_at':t.strftime('%H:%M')+'(AT)','sell_price':ep,'qty':p['qty'],'pnl':(ep-p['buy_price'])*p['qty']});del bp[s]
return bh,sum(x['pnl'] for x in bh)
def s_loop():
init_upstox();fetch_instruments()
while True:
try:
n=dt.datetime.now(IST).time()
if dt.time(15,12)<=n<dt.time(15,30):
for s,p in list(pos.items()):
if p_order(s,p['qty'],"SELL"):hist.append({**p,'sym':s,'sold_at':'15:12(AutoSq)','pnl':0.0});del pos[s]
trk.clear();time.sleep(60);continue
for s,m in list(trk.items()):
d=get_live_data(s,'1minute')
if d.empty or len(d)<15:continue
d.ta.ema(length=9,append=True);d['VW']=((d['High']+d['Low']+d['Close'])/3*d['Volume']).groupby(d.index.date).cumsum()/d['Volume'].groupby(d.index.date).cumsum();r=d.iloc[-1]
if TM or ((r['Low']<=r['EMA_9']*1.0015) and (r['Close']>r['EMA_9']) and (r['Close']>r['VW']) and (d.iloc[-2]['Close']>r['EMA_9'])):
q=int((m['margin']*LEV)/r['Close'])
if q>0:
del trk[s]
if p_order(s,q,"BUY"):pos[s]={**m,'qty':q,'buy_price':r['Close']}
for s,p in list(pos.items()):
d=add_at(get_live_data(s,'1minute'))
if d.empty or len(d)<15:continue
curr=d.iloc[-1]
if curr['AT_Sell']==True and p_order(s,p['qty'],"SELL"):
hist.append({**p,'sym':s,'sold_at':'AT_Sell','pnl':(curr['Close']-p['buy_price'])*p['qty']});del pos[s]
except Exception as e:
open(LF,"a").write(f"[{dt.datetime.now(IST).strftime('%H:%M:%S')}] LOOP ERR: {str(e)}\n");pass
time.sleep(10)
if __name__=='__main__':th.Thread(target=s_loop,daemon=True).start();app.run(host='0.0.0.0',port=80)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment