diff --git a/.gitignore b/.gitignore index d489181..154af3c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Ignore compiled Python files __pycache__/ +dist/ *.pyc *.pyo *.pyd diff --git a/fin_maestro_kin/constants.py b/fin_maestro_kin/constants.py index fb0f14f..494d02b 100644 --- a/fin_maestro_kin/constants.py +++ b/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' ] \ No newline at end of file diff --git a/fin_maestro_kin/modules/data_toolkit/screener/helper.py b/fin_maestro_kin/modules/data_toolkit/screener/helper.py index fe7077d..78ddf2f 100644 --- a/fin_maestro_kin/modules/data_toolkit/screener/helper.py +++ b/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} diff --git a/fin_maestro_kin/modules/data_toolkit/screener/screener_equities.py b/fin_maestro_kin/modules/data_toolkit/screener/screener_equities.py index fe6c3d7..c8e3e32 100644 --- a/fin_maestro_kin/modules/data_toolkit/screener/screener_equities.py +++ b/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}")