Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
sreenivasdoosa
GitHub Repository: sreenivasdoosa/sdoosa-algo-trade-python
 Path: blob/master/src/instruments/Instruments.py
296 views
1
import os

2
import logging

3
import json

4


5
from config.Config import getServerConfig, getTimestampsData, saveTimestampsData

6
from core.Controller import Controller

7
from utils.Utils import Utils

8


9
class Instruments:

10
  instrumentsList = None

11
  symbolToInstrumentMap = None

12
  tokenToInstrumentMap = None

13


14
  @staticmethod

15
  def shouldFetchFromServer():

16
    timestamps = getTimestampsData()

17
    if 'instrumentsLastSavedAt' not in timestamps:

18
      return True

19
    lastSavedTimestamp = timestamps['instrumentsLastSavedAt']

20
    nowEpoch = Utils.getEpoch()

21
    if nowEpoch - lastSavedTimestamp >= 24 * 60* 60:

22
      logging.info("Instruments: shouldFetchFromServer() returning True as its been 24 hours since last fetch.")

23
      return True

24
    return False

25


26
  @staticmethod

27
  def updateLastSavedTimestamp():

28
    timestamps = getTimestampsData()

29
    timestamps['instrumentsLastSavedAt'] = Utils.getEpoch()

30
    saveTimestampsData(timestamps)

31


32
  @staticmethod

33
  def loadInstruments():

34
    serverConfig = getServerConfig()

35
    instrumentsFilepath = os.path.join(serverConfig['deployDir'], 'instruments.json')

36
    if os.path.exists(instrumentsFilepath) == False:

37
      logging.warn('Instruments: instrumentsFilepath %s does not exist', instrumentsFilepath)

38
      return [] # returns empty list

39


40
    isdFile = open(instrumentsFilepath, 'r')

41
    instruments = json.loads(isdFile.read())

42
    logging.info('Instruments: loaded %d instruments from file %s', len(instruments), instrumentsFilepath)

43
    return instruments

44


45
  @staticmethod

46
  def saveInstruments(instruments = []):

47
    serverConfig = getServerConfig()

48
    instrumentsFilepath = os.path.join(serverConfig['deployDir'], 'instruments.json')

49
    with open(instrumentsFilepath, 'w') as isdFile:

50
      json.dump(instruments, isdFile, indent=2, default=str)

51
    logging.info('Instruments: Saved %d instruments to file %s', len(instruments), instrumentsFilepath)

52
    # Update last save timestamp

53
    Instruments.updateLastSavedTimestamp()

54


55
  @staticmethod

56
  def fetchInstrumentsFromServer():

57
    instrumentsList = []

58
    try:

59
      brokerHandle = Controller.getBrokerLogin().getBrokerHandle()

60
      logging.info('Going to fetch instruments from server...')

61
      instrumentsList = brokerHandle.instruments('NSE')

62
      instrumentsListFnO = brokerHandle.instruments('NFO')

63
      # Add FnO instrument list to the main list

64
      instrumentsList.extend(instrumentsListFnO)

65
      logging.info('Fetched %d instruments from server.', len(instrumentsList))

66
    except Exception as e:

67
      logging.exception("Exception while fetching instruments from server")

68
    return instrumentsList

69


70
  @staticmethod

71
  def fetchInstruments():

72
    if Instruments.instrumentsList:

73
      return Instruments.instrumentsList

74


75
    instrumentsList = Instruments.loadInstruments()

76
    if len(instrumentsList) == 0 or Instruments.shouldFetchFromServer() == True:

77
      instrumentsList = Instruments.fetchInstrumentsFromServer()

78
      # Save instruments to file locally

79
      if len(instrumentsList) > 0:

80
        Instruments.saveInstruments(instrumentsList)

81


82
    if len(instrumentsList) == 0:

83
      print("Could not fetch/load instruments data. Hence exiting the app.")

84
      logging.error("Could not fetch/load instruments data. Hence exiting the app.");

85
      exit(-2)

86
    

87
    Instruments.symbolToInstrumentMap = {}

88
    Instruments.tokenToInstrumentMap = {}

89
    for isd in instrumentsList:

90
      tradingSymbol = isd['tradingsymbol']

91
      instrumentToken = isd['instrument_token']

92
      # logging.info('%s = %d', tradingSymbol, instrumentToken)

93
      Instruments.symbolToInstrumentMap[tradingSymbol] = isd

94
      Instruments.tokenToInstrumentMap[instrumentToken] = isd

95
    

96
    logging.info('Fetching instruments done. Instruments count = %d', len(instrumentsList))

97
    Instruments.instrumentsList = instrumentsList # assign the list to static variable

98
    return instrumentsList

99


100
  @staticmethod

101
  def getInstrumentDataBySymbol(tradingSymbol):

102
    return Instruments.symbolToInstrumentMap[tradingSymbol]

103


104
  @staticmethod

105
  def getInstrumentDataByToken(instrumentToken):

106
    return Instruments.tokenToInstrumentMap[instrumentToken]

107
    

108