Skip to content

Instantly share code, notes, and snippets.

@mr-easy
Created May 25, 2024 09:28
Show Gist options
  • Save mr-easy/87bf5b2e2c02d3d94d40751c2d36c54b to your computer and use it in GitHub Desktop.
Save mr-easy/87bf5b2e2c02d3d94d40751c2d36c54b to your computer and use it in GitHub Desktop.
## WARNING: This is not meant to be used, this is just for demonstration purpose. Get official kite API access from Zerodha.
## USAGE:
# from kiteext_pro_open import KiteExt
# kite = KiteExt()
# kite.login_with_credentials(user_id, password, twofa_secret)
# print(kite.profile())
# print(kite.ltp(['NSE:TCS']))
# print(kite.quote(['NSE:TCS']))
# known issue: history api not working.
import pyotp
import requests
from urllib.parse import urlparse, parse_qs
from kiteconnect import KiteConnect
endpoints = {
'app_login': 'https://kite.zerodha.com/connect/login',
'login': 'https://kite.zerodha.com/api/login',
'twofa': 'https://kite.zerodha.com/api/twofa',
'app_authorize': 'https://kite.zerodha.com/connect/finish',
'streak_login': 'https://api-op.streak.tech/login/',
'streak_session': 'https://api-op.streak.tech/get_session_status/'
}
streak_api_key = '2b0bx36bnep2vslf'
class KiteExt(KiteConnect):
def __init__(self):
super().__init__(api_key=streak_api_key)
def login_with_credentials(self, user_id, password, twofa_secret):
s = requests.session()
s.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Referer': 'https://www.streak.tech/'
}
r1 = s.get(endpoints['app_login'], params={'api_key': streak_api_key})
sess_id = parse_qs(urlparse(r1.url).query)['sess_id']
r2 = s.post(
endpoints['login'],
data = {'user_id':user_id, 'password':password},
allow_redirects=True
)
request_id = r2.json()['data']['request_id']
r3 = s.post(
endpoints['twofa'],
data={'user_id':user_id, 'request_id':request_id, 'twofa_value':pyotp.TOTP(twofa_secret).now(), 'skip_session':True},
allow_redirects=True
)
r4 = s.get(
endpoints['app_login'],
params={'api_key':streak_api_key, 'sess_id':sess_id, 'skip_session':True},
allow_redirects=True
)
request_token = parse_qs(urlparse(r4.url).query)['request_token'][0]
r5 = s.post(
endpoints['streak_login'],
data = {'action':'login', 'type':'login', 'status':'success','request_token':request_token}
)
csrf = r5.json()['csrf']
sessionid = r5.json()['sessionid']
s.headers.update({'Authorization': f"csrftoken={csrf};csrfmiddlewaretoken={csrf};sessionid={sessionid}"})
r6 = s.get(endpoints['streak_session'])
access_token = r6.json()['access_token']
self.set_access_token(access_token)
return
@tradewithmadhan
Copy link

@mr-easy Where did you get streak_api_key ?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment