Enhancement: Add New Endpoints for Enhanced Data Retrieval from Screener.in

.gitignore

@@ -1,5 +1,6 @@
 # Ignore compiled Python files
 __pycache__/
+dist/
 *.pyc
 *.pyo
 *.pyd

fin_maestro_kin/constants.py

@@ -23,5 +23,10 @@
     '/sentiment/pcr-indice-analysis',
     '/sentiment/pcr-stocks-analysis?symbol=INFY',
     '/generate_plot?ticker=BSE.NS',
-    '/screener-equities/quarterly-result?symbol=vedl'
+    '/screener-equities/quarterly-result?symbol=vedl',
+    '/screener-equities/shareholding-pattern?symbol=vedl',
+    '/screener-equities/cash-flow?symbol=vedl',
+    '/screener-equities/balance-sheet?symbol=vedl',
+    '/screener-equities/annual-profit-loss?symbol=vedl',
+    '/screener-equities/ratios?symbol=vedl'
 ]

fin_maestro_kin/modules/data_toolkit/screener/helper.py

@@ -10,3 +10,63 @@ def process_quarterly_reports_data(data):
             formatted_reports[formatted_label] = {key: value for key, value in report.items() if key != ''}
 
     return {'symbol': stock_name, 'quarterly_reports': formatted_reports}
+
+
+def process_shareholding_pattern_data(data):
+    stock_name = data.get('symbol', '')
+    shareholding_pattern = data.get('shareholding_pattern', [])
+    formatted_reports = {}
+    for report in shareholding_pattern:
+        label = report.get("")
+        if label:
+            formatted_label = label.replace('+', '').strip()
+            formatted_reports[formatted_label] = {key: value for key, value in report.items() if key != ""}
+    return {'symbol': stock_name, 'shareholding_pattern': formatted_reports}
+
+
+def process_cash_flow_data(data):
+    stock_name = data.get('symbol', '')
+    cash_flow = data.get('cash_flow_statement', [])
+    formatted_reports = {}
+    for report in cash_flow:
+        label = report.get("")
+        if label:
+            formatted_label = label.replace('+', '').strip()
+            formatted_reports[formatted_label] = {key: value for key, value in report.items() if key != ""}
+    return {'symbol': stock_name, 'cash_flow_statement': formatted_reports}
+
+
+def process_balance_sheet_data(data):
+    stock_name = data.get('symbol', '')
+    cash_flow = data.get('balance_sheet', [])
+    formatted_reports = {}
+    for report in cash_flow:
+        label = report.get("")
+        if label:
+            formatted_label = label.replace('+', '').strip()
+            formatted_reports[formatted_label] = {key: value for key, value in report.items() if key != ""}
+    return {'symbol': stock_name, 'balance_sheet': formatted_reports}
+
+
+def process_profit_loss_data(data):
+    stock_name = data.get('symbol', '')
+    profit_loss_statement = data.get('profit_loss_statement', [])
+    formatted_reports = {}
+    for report in profit_loss_statement:
+        label = report.get("")
+        if label:
+            formatted_label = label.replace('+', '').strip()
+            formatted_reports[formatted_label] = {key: value for key, value in report.items() if key != ""}
+    return {'symbol': stock_name, 'profit_loss_statement': formatted_reports}
+
+
+def process_ratios_data(data):
+    stock_name = data.get('symbol', '')
+    ratios = data.get('ratios', [])
+    formatted_reports = {}
+    for report in ratios:
+        label = report.get("")
+        if label:
+            formatted_label = label.replace('+', '').strip()
+            formatted_reports[formatted_label] = {key: value for key, value in report.items() if key != ""}
+    return {'symbol': stock_name, 'ratios': formatted_reports}

fin_maestro_kin/modules/data_toolkit/screener/screener_equities.py

@@ -38,3 +38,168 @@ def get_quarterly_results(
             return JSONResponse(content={"error": "Failed to fetch data from URL"}, status_code=500)
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Error fetching quarterly reports data: {e}")
+
+
+def shareholding_pattern(company_code):
+    url = f"https://www.screener.in/company/{company_code}/#shareholding"
+    response = requests.get(url)
+
+    if response.status_code == 200:
+        soup = BeautifulSoup(response.content, 'html.parser')
+        table = soup.find('section', {'id': 'shareholding'})
+        headers = [th.text.strip() for th in table.find('thead').find_all('th')]
+        rows = table.find('tbody').find_all('tr')
+        data = [{header: td.text.strip() for header, td in zip(headers, row.find_all('td'))} for row in rows]
+        return {'symbol': company_code, 'shareholding_pattern': data}
+    elif response.status_code == 404:
+        raise HTTPException(status_code=404, detail=f"No data found for the specified parameters.")
+    else:
+        raise HTTPException(status_code=500, detail=f"Failed to load the URL. Status code: {response.status_code}")
+
+
+#Example usage: http://localhost:8000/screener-equities/shareholding-pattern?symbol=vedl
+@router.get("/screener-equities/shareholding-pattern", tags=["Screener Equities"])
+def get_shareholding_pattern(
+    symbol: str = Query(..., title="Symbol", description="Stock Symbol")
+):
+    try:
+        data = shareholding_pattern(symbol)
+        if data is not None:
+            processed_data = process_shareholding_pattern_data(data)
+            return JSONResponse(content=processed_data)
+        else:
+            return JSONResponse(content={"error": "Failed to fetch data from URL"}, status_code=500)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error fetching shareholding pattern data: {e}")
+
+
+def cash_flow_statement(company_code):
+    url = f"https://www.screener.in/company/{company_code}/#cash-flow"
+    response = requests.get(url)
+
+    if response.status_code == 200:
+        soup = BeautifulSoup(response.content, 'html.parser')
+        table = soup.find('section', {'id': 'cash-flow'})
+        headers = [th.text.strip() for th in table.find('thead').find_all('th')]
+        rows = table.find('tbody').find_all('tr')
+        data = [{header: td.text.strip() for header, td in zip(headers, row.find_all('td'))} for row in rows]
+        return {'symbol': company_code, 'cash_flow_statement': data}
+    elif response.status_code == 404:
+        raise HTTPException(status_code=404, detail=f"No data found for the specified parameters.")
+    else:
+        raise HTTPException(status_code=500, detail=f"Failed to load the URL. Status code: {response.status_code}")
+
+
+#Example usage: http://localhost:8000/screener-equities/cash-flow?symbol=vedl
+@router.get("/screener-equities/cash-flow", tags=["Screener Equities"])
+def get_cash_flow_statement(
+    symbol: str = Query(..., title="Symbol", description="Stock Symbol")
+):
+    try:
+        data = cash_flow_statement(symbol)
+        if data is not None:
+            processed_data = process_cash_flow_data(data)
+            return JSONResponse(content=processed_data)
+        else:
+            return JSONResponse(content={"error": "Failed to fetch data from URL"}, status_code=500)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error fetching cash flow data: {e}")
+
+
+def balance_sheet(company_code):
+    url = f"https://www.screener.in/company/{company_code}/#balance-sheet"
+    response = requests.get(url)
+
+    if response.status_code == 200:
+        soup = BeautifulSoup(response.content, 'html.parser')
+        table = soup.find('section', {'id': 'balance-sheet'})
+        headers = [th.text.strip() for th in table.find('thead').find_all('th')]
+        rows = table.find('tbody').find_all('tr')
+        data = [{header: td.text.strip() for header, td in zip(headers, row.find_all('td'))} for row in rows]
+        return {'symbol': company_code, 'balance_sheet': data}
+    elif response.status_code == 404:
+        raise HTTPException(status_code=404, detail=f"No data found for the specified parameters.")
+    else:
+        raise HTTPException(status_code=500, detail=f"Failed to load the URL. Status code: {response.status_code}")
+
+
+#Example usage: http://localhost:8000/screener-equities/balance-sheet?symbol=vedl
+@router.get("/screener-equities/balance-sheet", tags=["Screener Equities"])
+def get_balance_sheet(
+    symbol: str = Query(..., title="Symbol", description="Stock Symbol")
+):
+    try:
+        data = balance_sheet(symbol)
+        if data is not None:
+            processed_data = process_balance_sheet_data(data)
+            return JSONResponse(content=processed_data)
+        else:
+            return JSONResponse(content={"error": "Failed to fetch data from URL"}, status_code=500)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error fetching balance sheet data: {e}")
+
+
+def annual_profit_and_loss_statement(company_code):
+    url = f"https://www.screener.in/company/{company_code}/#profit-loss"
+    response = requests.get(url)
+
+    if response.status_code == 200:
+        soup = BeautifulSoup(response.content, 'html.parser')
+        table = soup.find('section', {'id': 'profit-loss'})
+        headers = [th.text.strip() for th in table.find('thead').find_all('th')]
+        rows = table.find('tbody').find_all('tr')
+        data = [{header: td.text.strip() for header, td in zip(headers, row.find_all('td'))} for row in rows]
+        return {'symbol': company_code, 'profit_loss_statement': data}
+    elif response.status_code == 404:
+        raise HTTPException(status_code=404, detail=f"No data found for the specified parameters.")
+    else:
+        raise HTTPException(status_code=500, detail=f"Failed to load the URL. Status code: {response.status_code}")
+
+
+#Example usage: http://localhost:8000/screener-equities/annual-profit-loss?symbol=vedl
+@router.get("/screener-equities/annual-profit-loss", tags=["Screener Equities"])
+def get_annual_profit_and_loss_statement(
+    symbol: str = Query(..., title="Symbol", description="Stock Symbol")
+):
+    try:
+        data = annual_profit_and_loss_statement(symbol)
+        if data is not None:
+            processed_data = process_profit_loss_data(data)
+            return JSONResponse(content=processed_data)
+        else: 
+            return JSONResponse(content={"error": "Failed to fetch data from URL"}, status_code=500)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error fetching annual profit and loss data: {e}")
+
+
+def ratios(company_code):
+    url = f"https://www.screener.in/company/{company_code}/#ratios"
+    response = requests.get(url)
+
+    if response.status_code == 200:
+        soup = BeautifulSoup(response.content, 'html.parser')
+        table = soup.find('section', {'id': 'ratios'})
+        headers = [th.text.strip() for th in table.find('thead').find_all('th')]
+        rows = table.find('tbody').find_all('tr')
+        data = [{header: td.text.strip() for header, td in zip(headers, row.find_all('td'))} for row in rows]
+        return {'symbol': company_code, 'ratios': data}
+    elif response.status_code == 404:
+        raise HTTPException(status_code=404, detail=f"No data found for the specified parameters.")
+    else:
+        raise HTTPException(status_code=500, detail=f"Failed to load the URL. Status code: {response.status_code}")
+
+
+#Example usage: http://localhost:8000/screener-equities/ratios?symbol=vedl
+@router.get("/screener-equities/ratios", tags=["Screener Equities"])
+def get_ratios(
+    symbol: str = Query(..., title="Symbol", description="Stock Symbol")
+):
+    try:
+        data = ratios(symbol)
+        if data is not None:
+            processed_data = process_ratios_data(data)
+            return JSONResponse(content=processed_data)
+        else: 
+            return JSONResponse(content={"error": "Failed to fetch data from URL"}, status_code=500)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error fetching ratios data: {e}")