Thursday, July 22, 2021

AlgoTrading - Python code for automated login in Fyers

 In this post sharing for Fyers Automated login for AlgoTrading


I have used Fyers API v2

Note: Code has been adjusted to adopt latest changes from Fyers in API


import json
import sys
from hashlib import sha256
from urllib import parse

import pyotp
import requests
from fyers_api import fyersModel

app_secret = "api_secret"
redirect_uri = "redirect_uri"
fyers_id = "fyers_id"
pin = "pin"
client_id = "client_id"
app_id = "app_id"
TOTP_KEY = "TOTP_KEY"  # TOTP secret is generated when we enable 2Factor TOTP from myaccount portal

APP_ID_TYPE = "2"  # Keep default as 2, It denotes web login
APP_TYPE = "100"
APP_ID_HASH = sha256((client_id + ":" + app_secret).encode('utf-8')).hexdigest()

SUCCESS = 1
ERROR = -1

# API endpoints
BASE_URL = "https://api-t2.fyers.in/vagator/v2"
BASE_URL_2 = "https://api.fyers.in/api/v2"
URL_SEND_LOGIN_OTP = BASE_URL + "/send_login_otp"
URL_VERIFY_TOTP = BASE_URL + "/verify_otp"
URL_VERIFY_PIN = BASE_URL + "/verify_pin"
URL_TOKEN = BASE_URL_2 + "/token"
URL_VALIDATE_AUTH_CODE = BASE_URL_2 + "/validate-authcode"


def send_login_otp(fy_id, app_id):
    try:
        payload = {
            "fy_id": fy_id,
            "app_id": app_id
        }

        result_string = requests.post(url=URL_SEND_LOGIN_OTP, json=payload)
        if result_string.status_code != 200:
            return [ERROR, result_string.text]

        result = json.loads(result_string.text)
        request_key = result["request_key"]

        return [SUCCESS, request_key]

    except Exception as e:
        return [ERROR, e]


def generate_totp(secret):
    try:
        generated_totp = pyotp.TOTP(secret).now()
        return [SUCCESS, generated_totp]

    except Exception as e:
        return [ERROR, e]


def verify_totp(request_key, totp):
    try:
        payload = {
            "request_key": request_key,
            "otp": totp
        }

        result_string = requests.post(url=URL_VERIFY_TOTP, json=payload)
        if result_string.status_code != 200:
            return [ERROR, result_string.text]

        result = json.loads(result_string.text)
        request_key = result["request_key"]

        return [SUCCESS, request_key]

    except Exception as e:
        return [ERROR, e]


def verify_PIN(request_key, pin):
    try:
        payload = {
            "request_key": request_key,
            "identity_type": "pin",
            "identifier": pin
        }

        result_string = requests.post(url=URL_VERIFY_PIN, json=payload)
        if result_string.status_code != 200:
            return [ERROR, result_string.text]

        result = json.loads(result_string.text)
        access_token = result["data"]["access_token"]

        return [SUCCESS, access_token]

    except Exception as e:
        return [ERROR, e]


def token(fy_id, app_id, redirect_uri, app_type, access_token):
    try:
        payload = {
            "fyers_id": fy_id,
            "app_id": app_id,
            "redirect_uri": redirect_uri,
            "appType": app_type,
            "code_challenge": "",
            "state": "sample_state",
            "scope": "",
            "nonce": "",
            "response_type": "code",
            "create_cookie": True
        }
        headers = {'Authorization': f'Bearer {access_token}'}

        result_string = requests.post(
            url=URL_TOKEN, json=payload, headers=headers
        )

        if result_string.status_code != 308:
            return [ERROR, result_string.text]

        result = json.loads(result_string.text)
        url = result["Url"]
        auth_code = parse.parse_qs(parse.urlparse(url).query)['auth_code'][0]

        return [SUCCESS, auth_code]

    except Exception as e:
        return [ERROR, e]


def validate_authcode(app_id_hash, auth_code):
    try:
        payload = {
            "grant_type": "authorization_code",
            "appIdHash": app_id_hash,
            "code": auth_code,
        }

        result_string = requests.post(url=URL_VALIDATE_AUTH_CODE, json=payload)
        if result_string.status_code != 200:
            return [ERROR, result_string.text]

        result = json.loads(result_string.text)
        access_token = result["access_token"]

        return [SUCCESS, access_token]

    except Exception as e:
        return [ERROR, e]


def check_login_status():
    user_config = get_user_config()

    token_date_json = user_config['fyers']["token_date"]
    access_token = user_config['fyers']["access_token"]
    # Convert date from json to date object
    token_date = datetime.strptime(token_date_json, "%d-%m-%Y")
    if token_date.date() != date.today():
        access_token = None

    return access_token


def get_auth_code():
    access_token = check_login_status()
    if access_token is not None:
        return access_token

    # Step 1 - Retrieve request_key from send_login_otp API
    send_otp_result = send_login_otp(fy_id=fyers_id, app_id=APP_ID_TYPE)
    if send_otp_result[0] != SUCCESS:
        print(f"send_login_otp failure - {send_otp_result[1]}")
        sys.exit()
    else:
        print("send_login_otp success")

    # Step 2 - Generate totp
    generate_totp_result = generate_totp(secret=TOTP_KEY)
    if generate_totp_result[0] != SUCCESS:
        print(f"generate_totp failure - {generate_totp_result[1]}")
        sys.exit()
    else:
        print("generate_totp success")

    # Step 3 - Verify totp and get request key from verify_otp API
    request_key = send_otp_result[1]
    totp = generate_totp_result[1]
    verify_totp_result = verify_totp(request_key=request_key, totp=totp)
    if verify_totp_result[0] != SUCCESS:
        print(f"verify_totp_result failure - {verify_totp_result[1]}")
        sys.exit()
    else:
        print("verify_totp_result success")

    # Step 4 - Verify pin and send back access token
    request_key_2 = verify_totp_result[1]
    verify_pin_result = verify_PIN(request_key=request_key_2, pin=pin)
    if verify_pin_result[0] != SUCCESS:
        print(f"verify_pin_result failure - {verify_pin_result[1]}")
        sys.exit()
    else:
        print("verify_pin_result success")

    # Step 5 - Get auth code for API V2 App from trade access token
    token_result = token(
        fy_id=fyers_id, app_id=app_id, redirect_uri=redirect_uri, app_type=APP_TYPE,
        access_token=verify_pin_result[1]
    )
    if token_result[0] != SUCCESS:
        print(f"token_result failure - {token_result[1]}")
        sys.exit()
    else:
        print("token_result success")

    # Step 6 - Get API V2 access token from validating auth code
    auth_code = token_result[1]
    validate_authcode_result = validate_authcode(
        app_id_hash=APP_ID_HASH, auth_code=auth_code
    )
    if validate_authcode_result[0] != SUCCESS:
        print(f"validate_authcode failure - {validate_authcode_result[1]}")
        sys.exit()
    else:
        print("validate_authcode success")

    access_token = validate_authcode_result[1]

    update_token(access_token)
    print(f"access_token - {access_token}")
    return access_token



if __name__ == '__main__':
    token = get_auth_code()
    fyers = fyersModel.FyersModel(client_id=client_id, token=token, log_path='/tmp/')
    print(fyers.orderbook())


This is explained in the below video, leave your feedback in comments.

 

Wednesday, April 7, 2021

Python Code for Straddle Algotrading using Zerodha KITE API

In this post I am sharing code of Short Straddle Order execution using Zerodha Kite API and Python Programming Language explained in Youtube Video 

Like and Subscribe channel for more such videos.

Code:

 


import logging
from datetime import datetime

from dateutil.relativedelta import relativedelta, TH
from kiteconnect import KiteConnect


def get_kite():
    kiteObj = KiteConnect(api_key='API_KEY')
    kiteObj.set_access_token('ACCESS_TOKEN')
    return kiteObj


kite = get_kite()
instrumentsList = None


def getCMP(tradingSymbol):
    quote = kite.quote(tradingSymbol)
    if quote:
        return quote[tradingSymbol]['last_price']
    else:
        return 0


def get_symbols(expiry, name, strike, ins_type):
    global instrumentsList

    if instrumentsList is None:
        instrumentsList = kite.instruments('NFO')

    lst_b = [num for num in instrumentsList if num['expiry'] == expiry and num['strike'] == strike
             and num['instrument_type'] == ins_type and num['name'] == name]
    return lst_b[0]['tradingsymbol']


def place_order(tradingSymbol, price, qty, direction, exchangeType, product, orderType):
    try:
        orderId = kite.place_order(
            variety=kite.VARIETY_REGULAR,
            exchange=exchangeType,
            tradingsymbol=tradingSymbol,
            transaction_type=direction,
            quantity=qty,
            price=price,
            product=product,
            order_type=orderType)

        logging.info('Order placed successfully, orderId = %s', orderId)
        return orderId
    except Exception as e:
        logging.info('Order placement failed: %s', e.message)


if __name__ == '__main__':
    # Find ATM Strike of Nifty
    atm_strike = round(getCMP('NSE:NIFTY 50'), -2)

    next_thursday_expiry = datetime.today() + relativedelta(weekday=TH(1))

    symbol_ce = get_symbols(next_thursday_expiry.date(), 'NIFTY', atm_strike, 'CE')
    symbol_pe = get_symbols(next_thursday_expiry.date(), 'NIFTY', atm_strike, 'PE')

    place_order(symbol_ce, 0, 75, kite.TRANSACTION_TYPE_SELL, KiteConnect.EXCHANGE_NFO, KiteConnect.PRODUCT_MIS,
                KiteConnect.ORDER_TYPE_MARKET)

    place_order(symbol_pe, 0, 75, kite.TRANSACTION_TYPE_SELL, KiteConnect.EXCHANGE_NFO, KiteConnect.PRODUCT_MIS,
                KiteConnect.ORDER_TYPE_MARKET) 

If you have any doubt then write it in comment box below.

Enjoy Coding and Algo Trading.