add download_expiry_dates_nfo

core/brokers/base/base.py

@@ -2,6 +2,8 @@
 
 import base64
 from datetime import datetime, timedelta
+from os import popen
+from time import sleep
 from pandas.errors import OutOfBoundsDatetime
 from json import JSONDecodeError, dumps, loads
 from ssl import SSLError
@@ -67,7 +69,7 @@ def _create_session(cls) -> req_session:
             requests.Session: A new requests session object.
         """
         session = req_session()
-        session._session.mount("https://", HTTPAdapter(max_retries=RETRY_STRATEGY))
+        session.mount("https://", HTTPAdapter(max_retries=RETRY_STRATEGY))
         return session
 
     @classmethod
@@ -516,3 +518,60 @@ def filter_future_dates(data: Union[List[str], Series]) -> List[str]:
 
         # Format and return the result as a sorted list of strings
         return sorted(future_dates.dt.strftime("%Y-%m-%d").tolist())
+
+    @classmethod
+    def download_expiry_dates_nfo(
+        cls,
+        root,
+    ):
+        temp_session = req_session()
+
+        for _ in range(5):
+            try:
+                headers = {
+                    "accept": "*/*",
+                    "accept-language": "en-GB,en-US;q=0.9,en;q=0.8,hi;q=0.7",
+                    "dnt": "1",
+                    "referer": "https://www.nseindia.com/option-chain",
+                    "sec-ch-ua": '"Google Chrome";v="123", "Not:A-Brand";v="8", "Chromium";v="123"',
+                    "sec-ch-ua-mobile": "?0",
+                    "sec-ch-ua-platform": '"Windows"',
+                    "sec-fetch-dest": "empty",
+                    "sec-fetch-mode": "cors",
+                    "sec-fetch-site": "same-origin",
+                    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
+                }
+
+                params = {
+                    "symbol": f"{root}",
+                }
+
+                response = temp_session.request(
+                    method="GET",
+                    url=cls.nfo_url,
+                    params=params,
+                    cookies=cls.cookies,
+                    headers=headers,
+                    timeout=10,
+                )
+                data = response.json()
+                expiry_dates = data["records"]["expiryDates"]
+                cls.expiry_dates[root] = cls.filter_future_dates(expiry_dates)
+                return None
+
+            except Exception as exc:
+                print(f"Error: {exc}")
+
+            try:
+                response = popen(
+                    f'curl "{cls.nfo_url}?symbol={root}" -H "authority: beta.nseindia.com" -H "cache-control: max-age=0" -H "dnt: 1" -H "upgrade-insecure-requests: 1" -H "user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.117 Safari/537.36" -H "sec-fetch-user: ?1" -H "accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9" -H "sec-fetch-site: none" -H "sec-fetch-mode: navigate" -H "accept-encoding: gzip, deflate, br" -H "accept-language: en-US,en;q=0.9,hi;q=0.8" --compressed'
+                ).read()
+                data = loads(response)
+                expiry_dates = data["records"]["expiryDates"]
+                cls.expiry_dates[root] = cls.filter_future_dates(expiry_dates)
+                return None
+
+            except Exception as exc:
+                print(f"Error: {exc}")
+
+            sleep(5)

tests/core/brokers/base/test_base_class.py

@@ -2,9 +2,9 @@
 from json import JSONDecodeError, dumps
 import random
 from ssl import SSLError
-from pandas import DataFrame, DateOffset, Series, Timestamp
+from pandas import DataFrame, DateOffset, Series, Timedelta, Timestamp
 import pytest
-from unittest.mock import patch, Mock
+from unittest.mock import MagicMock, patch, Mock
 from requests.models import Response
 from requests.exceptions import (
     Timeout,
@@ -1330,23 +1330,53 @@ def test_filter_future_dates_with_list():
     This test ensures that the `filter_future_dates` method correctly filters a list of date strings, keeping only the future dates and removing the past dates.
     """
     # Test with a list of date strings
-    data = ["2024-08-12", "2024-07-30", "2024-08-15"]
-    expected = ["2024-08-12", "2024-08-15"]
+    # Create a Series with dates that are dynamically relative to the current date
+    current_date = datetime.now().date()
+    data = Series(
+        [
+            (current_date - Timedelta(days=5)).strftime("%Y-%m-%d"),
+            (current_date + Timedelta(days=3)).strftime("%Y-%m-%d"),
+            (current_date - Timedelta(days=10)).strftime("%Y-%m-%d"),
+            (current_date + Timedelta(days=15)).strftime("%Y-%m-%d"),
+        ]
+    )
+    # Expected result should include only future dates
+    expected = [
+        (current_date + Timedelta(days=3)).strftime("%Y-%m-%d"),
+        (current_date + Timedelta(days=15)).strftime("%Y-%m-%d"),
+    ]
     result = Broker.filter_future_dates(data)
     assert result == expected
 
 
 def test_filter_future_dates_with_series():
-    # Test with a pandas Series of date strings
     """
-    Create a pandas Series of date strings for testing the `filter_future_dates` method.
+    Test the `filter_future_dates` method with a dynamic pandas Series of date strings.
 
-    This Series contains a mix of future and past dates, which can be used to test the behavior of the `filter_future_dates` method.
+    This Series contains a mix of future and past dates relative to the current date.
     """
-    data = Series(["2024-08-12", "2024-07-30", "2024-08-15"])
-    expected = ["2024-08-12", "2024-08-15"]
+    # Create a Series with dates that are dynamically relative to the current date
+    current_date = datetime.now().date()
+    data = Series(
+        [
+            (current_date - Timedelta(days=5)).strftime("%Y-%m-%d"),
+            (current_date + Timedelta(days=3)).strftime("%Y-%m-%d"),
+            (current_date - Timedelta(days=10)).strftime("%Y-%m-%d"),
+            (current_date + Timedelta(days=15)).strftime("%Y-%m-%d"),
+        ]
+    )
+
+    # Expected result should include only future dates
+    expected = [
+        (current_date + Timedelta(days=3)).strftime("%Y-%m-%d"),
+        (current_date + Timedelta(days=15)).strftime("%Y-%m-%d"),
+    ]
+
+    # Call the method
     result = Broker.filter_future_dates(data)
-    assert result == expected
+
+    # Assert the result matches the expected future dates
+    assert result == expected, f"Expected {expected}, but got {result}"
 
 
 def test_filter_future_dates_with_empty_list():
@@ -1414,3 +1444,116 @@ def test_filter_future_dates_edge_case():
         result = Broker.filter_future_dates(data)
 
     assert result == expected
+
+
+@patch("core.brokers.base.base.req_session")
+def test_successful_request(mock_session):
+    """
+    Test the successful execution of the download_expiry_dates_nfo method.
+
+    This test ensures that when the request is successful, the method correctly
+    processes the response and stores the expiry dates.
+    """
+    mock_response = MagicMock()
+    mock_response.json.return_value = {
+        "records": {"expiryDates": ["2024-08-29", "2024-09-26", "2024-10-31"]}
+    }
+    mock_session.return_value.request.return_value = mock_response
+
+    Broker.expiry_dates = {}
+    Broker.nfo_url = "https://example.com/nfo"
+    Broker.cookies = {"cookie": "value"}
+
+    Broker.download_expiry_dates_nfo("NIFTY")
+
+    assert "NIFTY" in Broker.expiry_dates
+    assert len(Broker.expiry_dates["NIFTY"]) == 3
+
+
+@patch("core.brokers.base.base.req_session")
+@patch("core.brokers.base.base.popen")
+def test_fallback_to_curl(mock_popen, mock_session):
+    """
+    Test the fallback to curl when the initial request fails.
+
+    This test verifies that if the initial request fails, the method
+    falls back to using curl and still processes the response correctly.
+    """
+    mock_session.return_value.request.side_effect = Exception("Request failed")
+
+    mock_popen.return_value.read.return_value = dumps(
+        {"records": {"expiryDates": ["2024-08-29", "2024-09-26"]}}
+    )
+
+    Broker.expiry_dates = {}
+    Broker.nfo_url = "https://example.com/nfo"
+    Broker.cookies = {"cookie": "value"}
+
+    Broker.download_expiry_dates_nfo("BANKNIFTY")
+
+    assert "BANKNIFTY" in Broker.expiry_dates
+    assert len(Broker.expiry_dates["BANKNIFTY"]) == 2
+
+
+@patch("core.brokers.base.base.req_session")
+@patch("core.brokers.base.base.popen")
+@patch("core.brokers.base.base.sleep")
+def test_all_attempts_fail(mock_sleep, mock_popen, mock_session):
+    """
+    Test the behavior when all download attempts fail.
+
+    This test ensures that when both the initial request and the curl fallback
+    fail, the method handles the failure gracefully and does not add any
+    expiry dates for the symbol.
+    """
+    mock_session.return_value.request.side_effect = Exception("Request failed")
+    mock_popen.return_value.read.side_effect = Exception("Curl failed")
+
+    Broker.expiry_dates = {}
+    Broker.nfo_url = "https://example.com/nfo"
+    Broker.cookies = {"cookie": "value"}
+
+    Broker.download_expiry_dates_nfo("FAIL")
+
+    assert "FAIL" not in Broker.expiry_dates
+    assert mock_sleep.call_count == 5
+
+
+@patch("core.brokers.base.base.req_session")
+def test_filter_future_dates(mock_session):
+    """
+    Test the filtering of future dates in the download_expiry_dates_nfo method.
+
+    This test verifies that the method correctly filters out past dates and
+    only keeps future dates (including today) in the expiry_dates dictionary.
+    """
+    today = datetime.now().date()
+    past_date = (today - timedelta(days=1)).strftime("%Y-%m-%d")
+    future_date1 = (today + timedelta(days=1)).strftime("%Y-%m-%d")
+    future_date2 = (today + timedelta(days=30)).strftime("%Y-%m-%d")
+
+    mock_response = MagicMock()
+    mock_response.json.return_value = {
+        "records": {
+            "expiryDates": [
+                past_date,
+                today.strftime("%Y-%m-%d"),
+                future_date1,
+                future_date2,
+            ]
+        }
+    }
+    mock_session.return_value.request.return_value = mock_response
+
+    Broker.expiry_dates = {}
+    Broker.nfo_url = "https://example.com/nfo"
+    Broker.cookies = {"cookie": "value"}
+
+    Broker.download_expiry_dates_nfo("TEST")
+
+    assert "TEST" in Broker.expiry_dates
+    assert len(Broker.expiry_dates["TEST"]) == 2
+    assert past_date not in Broker.expiry_dates["TEST"]
+    assert today.strftime("%Y-%m-%d") not in Broker.expiry_dates["TEST"]
+    assert future_date1 in Broker.expiry_dates["TEST"]
+    assert future_date2 in Broker.expiry_dates["TEST"]