refactor remaining modules

fin_maestro_kin/main.py

@@ -1,7 +1,7 @@
 from fastapi import FastAPI, Request
 from starlette.responses import RedirectResponse
-from modules.sentiment_analysis.sentiment_analysis import router as sentiment_analysis_router
-from modules.trend_detector.trend_detector import router as trend_detector_router
+from modules.sentiment_analysis.sentiment_analysis import SentimentAnalyzer
+from modules.trend_detector.trend_detector import TrendDetector
 from modules.data_toolkit.nse.nse_operations import  NSEIndices, NSEEquities
 from modules.data_toolkit.screener.screener_equities import ScreenerEquities
 
@@ -15,9 +15,6 @@ async def apidog_docs_redirect(request: Request, call_next):
         return RedirectResponse(url=custom_docs_url, status_code=307)
     return await call_next(request)
 
-app.include_router(sentiment_analysis_router)
-app.include_router(trend_detector_router)
-
 screener_eq = ScreenerEquities()
 screener_eq.register_routes(app)
 
@@ -27,3 +24,8 @@ async def apidog_docs_redirect(request: Request, call_next):
 nse_indices = NSEIndices()
 nse_indices.register_routes(app)
 
+sentiment = SentimentAnalyzer()
+sentiment.register_routes(app)
+
+trend = TrendDetector()
+trend.register_routes(app)

fin_maestro_kin/modules/sentiment_analysis/sentiment_analysis.py

@@ -1,62 +1,97 @@
+import requests
+import json
 from fastapi import APIRouter, Query
-from .pcr_data import pcr_indice_scraper, pcr_stocks_scraper
 
-router = APIRouter(tags=["Sentiment"])
 
-#Example usage - http://127.0.0.1:8000/sentiment/pcr-indice-analysis
-@router.get("/sentiment/pcr-indice-analysis",tags=["Sentiment"])
-def analyze_indices():
-    try:
-        pcr_anal_result = pcr_indice_analysis()
-        return pcr_anal_result
-    except Exception as e:
-        return {"error": f"An error occurred during PCR analysis: {e}"}
+class PCR():
+    headers = {
+        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36',
+        'accept-encoding': 'gzip, deflate, br',
+        'accept-language': 'en-US,en;q=0.9'
+    }
+
+    @staticmethod
+    def pcr_indice_scraper(symbol):
+        url = 'https://www.nseindia.com/api/option-chain-indices?symbol=' + symbol
+        request = requests.get("https://www.nseindia.com", timeout=10, headers=PCR.headers)
+        cookies = dict(request.cookies)
+        response = requests.get(url, headers=PCR.headers, cookies=cookies).content
+        data = json.loads(response.decode('utf-8'))
+        totCE = data['filtered']['CE']['totOI']
+        totPE = data['filtered']['PE']['totOI']
+        pcr = totPE / totCE
+        return round(pcr, 3)
+
+    @staticmethod
+    def pcr_stocks_scraper(symbol):
+        url = 'https://www.nseindia.com/api/option-chain-equities?symbol=' + symbol
+        request = requests.get("https://www.nseindia.com", timeout=10, headers=PCR.headers)
+        cookies = dict(request.cookies)
+        response = requests.get(url, headers=PCR.headers, cookies=cookies).content
+        data = json.loads(response.decode('utf-8'))
+        totCE = data['filtered']['CE']['totOI']
+        totPE = data['filtered']['PE']['totOI']
+        pcr = totPE / totCE
+        return round(pcr, 3)
+
+
+class SentimentAnalyzer(PCR):
+    def __init__(self):
+        self.router = APIRouter(tags=["Sentiment"])
+
+    def register_routes(self, app):
+        self.router.add_api_route("/sentiment/pcr-indice-analysis", self.analyze_indices, methods=["GET"])
+        self.router.add_api_route("/sentiment/pcr-stocks-analysis", self.analyze_stock, methods=["GET"])
+        app.include_router(self.router)
+
+    def analyze_indices(self):
+        try:
+            pcr_anal_result = self.pcr_indice_analysis()
+            return pcr_anal_result
+        except Exception as e:
+            return {"error": f"An error occurred during PCR analysis: {e}"}
 
+    def analyze_stock(self, symbol: str = Query(..., title="Symbol", description="Stock symbol")):
+        try:
+            pcr_anal_result = self.pcr_stocks_analysis(symbol)
+            return pcr_anal_result
+        except Exception as e:
+            return {"error": f"An error occurred during PCR analysis for {symbol}: {e}"}
 
-#Example usage - http://127.0.0.1:8000/sentiment/pcr-stocks-analysis?symbol=INFY
-@router.get("/sentiment/pcr-stocks-analysis",tags=["Sentiment"])
-def analyze_stock(symbol: str = Query(..., title="Symbol", description="Stock symbol")):
-    try:
-        pcr_anal_result = pcr_stocks_analysis(symbol)
-        return pcr_anal_result
-    except Exception as e:
-        return {"error": f"An error occurred during PCR analysis for {symbol}: {e}"}
+    @staticmethod
+    def pcr_indice_analysis():
+        pcr_anal_result = {}
+        indices = ["NIFTY", "BANKNIFTY"]
 
+        for symbol in indices:
+            try:
+                pcr_value = PCR.pcr_indice_scraper(symbol)
+            except Exception as e:
+                print(f"Error fetching PCR for {symbol}: {e}")
+                return {"error": f"Failed to fetch PCR for {symbol}"}
 
-def pcr_indice_analysis():
-    pcr_anal_result = {}
-    indices = ["NIFTY", "BANKNIFTY"]
+            state = SentimentAnalyzer.get_state(pcr_value, [1.4, 1.19, 1, 0.91, 0.6])
+            pcr_anal_result[symbol] = [state, pcr_value]
+
+        return pcr_anal_result
 
-    for symbol in indices:
+    @staticmethod
+    def pcr_stocks_analysis(symbol):
         try:
-            pcr_value = pcr_indice_scraper(symbol)
+            pcr_anal_result = {}
+            pcr_value = PCR.pcr_stocks_scraper(symbol)
         except Exception as e:
             print(f"Error fetching PCR for {symbol}: {e}")
             return {"error": f"Failed to fetch PCR for {symbol}"}
 
-        state = get_state(pcr_value, [1.4, 1.19, 1, 0.91, 0.6])
+        state = SentimentAnalyzer.get_state(pcr_value, [1, 0.75, 0.50, 0.4])
         pcr_anal_result[symbol] = [state, pcr_value]
 
-    return pcr_anal_result
-
-
-def pcr_stocks_analysis(symbol):
-    try:
-        pcr_anal_result = {}
-        pcr_value = pcr_stocks_scraper(symbol)
-    except Exception as e:
-        print(f"Error fetching PCR for {symbol}: {e}")
-        return {"error": f"Failed to fetch PCR for {symbol}"}
-
-    state = get_state(pcr_value, [1, 0.75, 0.50, 0.4])
-    pcr_anal_result[symbol] = [state, pcr_value]
-
-    return pcr_anal_result
-
-
-def get_state(pcr_value, thresholds):
-    for threshold, label in zip(thresholds, ["Overbought", "Slightly overbought", "Neutral", "Slightly oversold"]):
-        if pcr_value >= threshold:
-            return label
+        return pcr_anal_result
 
-    return "Oversold"
+    @staticmethod
+    def get_state(pcr_value, thresholds):
+        for threshold, label in zip(thresholds, ["Overbought", "Slightly overbought", "Neutral", "Slightly oversold"]):
+            if pcr_value >= threshold:
+                return label
+        return "Oversold"

fin_maestro_kin/modules/trend_detector/trend_detector.py

@@ -8,51 +8,53 @@
 import warnings
 
 warnings.filterwarnings("ignore")
-initial, end = dt.date.today() - dt.timedelta(days=374), dt.date.today()
 
-router = APIRouter(tags=["Trend Detector"])
 
+class TrendDetector:
+    def __init__(self):
+        self.router = APIRouter(tags=["Trend Detector"])
+        matplotlib.use('Agg')
+        plt.style.use('dark_background')
 
-#Example usage - http://127.0.0.1:8000/generate_plot?ticker=BSE.NS
-def signals_generator(ticker, num_of_signals):
-    matplotlib.use('Agg')
-    yf.pdr_override()
-    plt.style.use('dark_background')  # dark theme for plot
+    def register_routes(self, app):
+        self.router.add_api_route("/generate_plot", self.generate_plot, methods=["GET"])
+        app.include_router(self.router)
 
-    df = yf.download(ticker, initial, end)  # dataframe from yfinance for a specific ticker
+    @staticmethod
+    def signals_generator(ticker, num_of_signals):
+        yf.pdr_override()
+        initial, end = dt.date.today() - dt.timedelta(days=374), dt.date.today()
+        df = yf.download(ticker, initial, end) 
+        closes = sorted(df.Close.tolist())
+        low, high = closes[num_of_signals], closes[-num_of_signals]
 
-    closes = sorted(df.Close.tolist())
-    low, high = closes[num_of_signals], closes[-num_of_signals]
+        df['Signal'] = 0
 
-    df['Signal'] = 0
+        df.loc[df['Adj Close'] > high, 'Signal'] = -1
+        df.loc[df['Adj Close'] < low, 'Signal'] = 1
 
-    df.loc[df['Adj Close'] > high, 'Signal'] = -1
-    df.loc[df['Adj Close'] < low, 'Signal'] = 1
+        long = df.loc[df['Signal'] == 1]
+        short = df.loc[df['Signal'] == -1]
 
-    long = df.loc[df['Signal'] == 1]
-    short = df.loc[df['Signal'] == -1]
+        fig = plt.figure()
+        fig.set_figwidth(12)
+        plt.plot(df.index, df['Adj Close'], color='white', label='Close Price')
+        plt.plot(long.index, df.loc[long.index]['Adj Close'], '^', markersize=10, color='g', label='Long/Buy')
+        plt.plot(short.index, df.loc[short.index]['Adj Close'], 'v', markersize=10, color='r', label='Short/Sell')
+        plt.ylabel('Closing Price')
+        plt.xlabel('Date')
+        plt.title(ticker)
+        plt.legend(loc='best')
 
-    fig = plt.figure()
-    fig.set_figwidth(12)
-    plt.plot(df.index, df['Adj Close'], color='white', label='Close Price')
-    plt.plot(long.index, df.loc[long.index]['Adj Close'], '^', markersize=10, color='g', label='Long/Buy')
-    plt.plot(short.index, df.loc[short.index]['Adj Close'], 'v', markersize=10, color='r', label='Short/Sell')
-    plt.ylabel('Closing Price')
-    plt.xlabel('Date')
-    plt.title(ticker)
-    plt.legend(loc='best')
+        return fig
 
-    return fig
+    async def generate_plot(self, ticker: str = Query(..., title="Ticker", description="Stock ticker symbol"),
+                            num_of_signals: int = Query(10, title="Num of Signals",
+                                                        description="Number of buy/sell signals to be plotted")):
+        fig = self.signals_generator(ticker, num_of_signals)
 
+        image_stream = io.BytesIO()
+        fig.savefig(image_stream, format="png")
+        image_stream.seek(0)
 
-@router.get("/generate_plot",tags=["Trend Detector"])
-async def generate_plot(ticker: str = Query(..., title="Ticker", description="Stock ticker symbol"),
-                        num_of_signals: int = Query(10, title="Num of Signals", description="Number of buy/sell signals to be plotted")):
-    fig = signals_generator(ticker, num_of_signals)
-
-    # Convert the matplotlib figure to a PNG image
-    image_stream = io.BytesIO()
-    fig.savefig(image_stream, format="png")
-    image_stream.seek(0)
-
-    return StreamingResponse(content=image_stream, media_type="image/png")
+        return StreamingResponse(content=image_stream, media_type="image/png")

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "fin-maestro-kin"
-version = "0.2.5"
+version = "0.2.6"
 description = "Seamless Finance: Docker Deployed APIs for Smart Investments."
 authors = ["DEV_FINWIZ <devjuneja43@gmail.com>"]
 license = "MIT"