add bhav copy feature functions, exceptions & tests

bsedata/bhavcopy.py

@@ -0,0 +1,57 @@
+import os
+import io
+import csv
+import requests
+import tempfile
+import datetime
+from zipfile import ZipFile
+from bsedata.exceptions import BhavCopyNotFound
+from bsedata.helpers import COMMON_REQUEST_HEADERS
+
+
+def loadBhavCopyData(statsDate: datetime.date) -> list:
+    tempDir = os.path.join(tempfile.gettempdir(), "bsedata")
+    zipfileName = f"EQ{statsDate.strftime('%d%m%y')}_CSV.ZIP"
+    r = requests.get(
+        f"https://www.bseindia.com/download/BhavCopy/Equity/{zipfileName}",
+        headers=COMMON_REQUEST_HEADERS,
+    )
+
+    if r.status_code != 200:
+        raise BhavCopyNotFound()
+
+    try:
+        os.makedirs(tempDir)
+    except FileExistsError:
+        pass
+
+    f_zip = open(os.path.join(tempDir, zipfileName), "wb+")
+    f_zip.write(r.content)
+    f_zip.close()
+
+    output = []
+
+    with ZipFile(os.path.join(tempDir, zipfileName)) as bhavCopyZip:
+        with bhavCopyZip.open(f"EQ{statsDate.strftime('%d%m%y')}.CSV") as bhavCopyFile:
+            reader = csv.DictReader(io.TextIOWrapper(bhavCopyFile))
+            for row in reader:
+                output.append(mapBhavCopyRowToDict(row))
+
+    return output
+
+
+def mapBhavCopyRowToDict(row: dict) -> dict:
+    SC_TYPE_MAP = {"B": "bond", "Q": "equity", "D": "debenture", "P": "preference"}
+    return {
+        "scrip_code": row["SC_CODE"],
+        "open": row["OPEN"],
+        "high": row["HIGH"],
+        "low": row["LOW"],
+        "close": row["CLOSE"],
+        "last": row["LAST"],
+        "prev_close": row["PREVCLOSE"],
+        "total_trades": row["NO_TRADES"],
+        "total_shares_traded": row["NO_OF_SHRS"],
+        "net_turnover": row["NET_TURNOV"],
+        "scrip_type": SC_TYPE_MAP[row["SC_TYPE"]],
+    }

bsedata/bse.py

@@ -24,7 +24,8 @@
 
 """
 
-from . import losers, gainers, quote, indices
+from . import losers, gainers, quote, indices, bhavcopy
+import datetime
 import requests
 import json
 
@@ -78,6 +79,49 @@ def updateScripCodes(self):
         f_stk.write(json.dumps(r.json()))
         f_stk.close()
         return
+
+    def getBhavCopyData(self, statsDate: datetime.date):
+        """
+        Get historical OHLCV data from Bhav Copy released by BSE everyday after market closing.
+        The columns available in the data and their description is as given below.
+
+        .. list-table::
+            :widths: 25 75
+            :header-rows: 1
+
+            * - Dictionary Field
+              - Description
+            * - scrip_code
+              - Unique code assigned to a scrip of a company by BSE
+            * - open
+              - The price at which the security first trades on a given trading day
+            * - high
+              - The highest intra-day price of a stock
+            * - low
+              - The lowest intra-day price of a stock
+            * - close
+              - The final price at which a security is traded on a given trading day
+            * - last
+              - The last trade price of the stock
+            * - prev_close
+              - The closing price of the stock for the previous trading day
+            * - total_trades
+              - The total number of trades of a scrip
+            * - total_shares_traded
+              - The total number of shares transacted of a scrip
+            * - net_turnover
+              - Total turnover of a scrip
+            * - scrip_type
+              - Scrip category: Equity, Preference, Debenture or Bond
+
+        The Bhav Copy files have been mapped to the above mentioned custom fields. The complete documentation for Bhav Copy can be found here: https://www.bseindia.com/markets/MarketInfo/BhavCopy.aspx.
+
+        
+        :param statsDate: A `datetime.date` object for the for which you want to fetch the data
+        :returns: A list of dictionaries which contains OHLCV data for that day for all scrip codes active on that day
+        :raises BhavCopyNotFound: Raised when Bhav Copy file is not found on BSE
+        """
+        return bhavcopy.loadBhavCopyData(statsDate)
 
     def getScripCodes(self):
         """

bsedata/exceptions.py

@@ -38,3 +38,14 @@ def __init__(self, status: str = "Inactive stock"):
         else:
             self.status = status
         super().__init__(self.status)
+
+
+class BhavCopyNotFound(Exception):
+    """
+    Exception raised when the BhavCopy file is not found on BSE website.
+    """
+
+    def __init__(self):
+        super().__init__(
+            """The BhavCopy file was not found on the BSE website. You are probably trying to get data for a trading holiday."""
+        )

bsedata/helpers.py

@@ -1,3 +1,3 @@
 COMMON_REQUEST_HEADERS = {
-    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.97 Safari/537.36 Edg/83.0.478.45'
-}
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.97 Safari/537.36 Edg/83.0.478.45"
+}

docs/apiref.rst

@@ -5,4 +5,7 @@ API Reference
    :members:
 
 .. autoexception:: bsedata.exceptions.InvalidStockException
+   :members:
+
+.. autoexception:: bsedata.exceptions.BhavCopyNotFound
    :members:

docs/introduction.rst

@@ -36,11 +36,11 @@ bsedata is a library for collecting real-time data from Bombay Stock Exchange (I
    :target: https://codecov.io/gh/sdabhi23/bsedata
    :alt: Code Coverage
 
-.. |testsMaster| image:: https://github.com/sdabhi23/bsedata/actions/workflows/tests.yml/badge.svg?branch=master
+.. |testsMaster| image:: https://github.com/sdabhi23/bsedata/actions/workflows/dev-tests.yml/badge.svg?branch=master
    :target: https://github.com/sdabhi23/bsedata/actions/workflows/tests.yml
    :alt: Status of tests on master branch
 
-.. |testsDev| image:: https://github.com/sdabhi23/bsedata/actions/workflows/tests.yml/badge.svg?branch=dev
+.. |testsDev| image:: https://github.com/sdabhi23/bsedata/actions/workflows/dev-tests.yml/badge.svg?branch=dev
    :target: https://github.com/sdabhi23/bsedata/actions/workflows/tests.yml
    :alt: Status of tests on dev branch
 

test_bsedata.py

@@ -23,12 +23,12 @@
     SOFTWARE.
 
 """
-import datetime
 
+import time
 import pytest
-
+import datetime
 from bsedata.bse import BSE
-from bsedata.exceptions import InvalidStockException
+from bsedata.exceptions import InvalidStockException, BhavCopyNotFound
 
 b = BSE(update_codes=True)
 
@@ -102,3 +102,21 @@ def test_getIndices(category):
     indices = b.getIndices(category)
     datetime.datetime.strptime(indices["updatedOn"], "%d %b %Y")
     assert len(indices["indices"]) >= 1
+    time.sleep(1)
+
+
+def test_getBhavCopyData_on_trade_holiday():
+    with pytest.raises(BhavCopyNotFound):
+        b.getBhavCopyData(datetime.date(2024, 1, 26))
+
+
+def test_getBhavCopyData():
+    bhavCopy = b.getBhavCopyData(datetime.date(2024, 1, 25))
+
+    scripCodeTypes = {x["scrip_type"] for x in bhavCopy}
+
+    predefinedScripCodeTypes = {"equity", "debenture", "preference", "bond"}
+
+    assert scripCodeTypes == predefinedScripCodeTypes
+
+    assert len(bhavCopy) > 0