🏗️ Enterprise Flask Backend Refactoring - Production Ready for FastAPI Integration

.gitignore

@@ -1,5 +1,21 @@
+# Operating System / Editor specific
 .DS_Store
+.vscode/
+__pycache__/
+*.pyc
 .ipynb_checkpoints
+
+# Environment variables
+*.env
+frontend/.env
 backend/.env
 
+# Configuration files
 ./api/src/adapters/api_calls/api_client_config.py
+
+#Virtual environments
+.venv/
+venv/
+frontend/.venv/
+backend/.venv/
+pytest_output.txt

api/data/sp500/raw_data/sp500_stocks_wiki_info.csv

@@ -0,0 +1,4 @@
+,Ticker,Security,SEC filings,GICS Sector,GICS Sub-Industry,Employees
+0,MMM,3M Company,reports,Industrials,Industrial Conglomerates,95000
+1,MSFT,Microsoft Corp,reports,Information Technology,Software,181000
+2,ZTS,Zoetis,reports,Health Care,Pharmaceuticals,12000

backend/api/src/__init__.py

@@ -14,15 +14,15 @@
 from api.src.models.queries.sql_query_strings import companies_within_sub_sector_str, sub_sector_names_in_sector_query_str
 from settings import DB_HOST, DB_NAME, DB_PASSWORD, DB_USER, DEBUG, TESTING
 
-def create_app(database='investment_analysis', testing=False, debug=True):
+def create_app(db_name='investment_analysis', db_user='postgres', db_password='postgres', testing=False):
     """Create and configure an instance of the Flask application."""
     app = Flask(__name__)
 
     # connect to the local computer's Postgres
     app.config.from_mapping(
-        DB_USER = 'postgres',
-        DB_NAME = database,
-        DB_PASSWORD = 'postgres',
+        DB_USER = db_user,
+        DB_NAME = db_name,
+        DB_PASSWORD = db_password,
         DB_HOST = '127.0.0.1',
         DEBUG = DEBUG,
         TESTING = TESTING
@@ -42,93 +42,13 @@ def create_app(database='investment_analysis', testing=False, debug=True):
 
     @app.route('/')
     def root_url():
-        return 'Welcome to the Economic Analysis api, through the prism of the S&P 500 stocks performance.'
-
-    @app.route('/sectors/')
-    def sector_avg_financial_performance():
-        """
-        url parameter format: f'/sectors/?financial_indicator={financial_indicator_name}'
-        returns the quarterly average, over the most recent 8 quarters, of the financial indicator of each and every sector
-        """
-        conn, cursor, financial_indicator = financial_performance_query_tools()
-        historical_financials_json_dicts = get_historical_financials_json(financial_indicator, cursor)
-        return json.dumps(historical_financials_json_dicts, default = str)
-
-    def get_historical_financials_json(financial_indicator, cursor):
-        if financial_indicator in ['revenue', 'net_income', 'earnings_per_share', 'profit_margin']:
-            historical_financials_json_dicts = (models.SubIndustry.
-                                                        find_avg_quarterly_financials_by_sector(financial_indicator, cursor))
-        elif financial_indicator in ['closing_price', 'price_earnings_ratio']:
-            historical_financials_json_dicts = (models.SubIndustry.
-                                                        find_sector_avg_price_pe(financial_indicator, cursor))
-        # needs to handle dropdown menu selection of 'Done. Continue to the sub-Sector level.'
-        else:
-            historical_financials_json_dicts = 'Please enter the name of a financial_indicator, such as revenue, net_income.'
-        return historical_financials_json_dicts
-
-    @app.route('/sectors/search')
-    def sub_industries_within_sector():
-        """
-        url parameter format example: /sectors/search?sector_name=Energy&financial_indicator=revenue
-        returns the quarterly average, over the most recent 8 quarters, of the selected financial indicator and sector
-        """
-        conn, cursor, sector_name, financial_indicator = sub_sector_performance_query_tools()
-        if sector_name == 'all_sectors':        
-            conn = db.get_db()
-            cursor = conn.cursor()
-            sector_names = MixinSectorPricePE.get_all_sector_names(models.SubIndustry, cursor)
-            return {'all_sector_names': sector_names}
-        else:
-            if financial_indicator in ['revenue', 'net_income', 'earnings_per_share', 'profit_margin']:
-                historical_financials_json_dicts = (models.SubIndustry.
-                                                        find_avg_quarterly_financials_by_sub_industry(sector_name, financial_indicator, cursor))
-            elif financial_indicator in ['closing_price', 'price_earnings_ratio']:
-                historical_financials_json_dicts = (models.SubIndustry.
-                                                        find_sub_industry_avg_quarterly_price_pe(sector_name, financial_indicator, cursor))
-            else:
-                historical_financials_json_dicts = {'Please enter the name of a financial indicator.'}
-            return json.dumps(historical_financials_json_dicts, default = str)
-
-    @app.route('/sub_sectors/search')
-    def search_sub_sectors():
-        conn, cursor, sub_sector_name, financial_indicator = company_performance_query_tools()
-        if sub_sector_name == 'all_sub_sectors':
-            sector_name = financial_indicator
-            sub_sector_names = MixinSubSectorPricePE.get_sub_sector_names_of_sector(models.SubIndustry, sector_name, cursor)
-            return json.dumps({'sub_sector_names': sub_sector_names}, default=str)
-        else:
-            if financial_indicator in ['revenue', 'net_income', 'earnings_per_share', 'profit_margin']:
-                historical_financials_json_dicts = (models.Company.
-                                                            find_companies_quarterly_financials(sub_sector_name, financial_indicator, cursor))
-            elif financial_indicator in ['closing_price', 'price_earnings_ratio']:
-                historical_financials_json_dicts = (models.Company.find_company_quarterly_price_pe(sub_sector_name, financial_indicator, cursor))
-            else:
-                historical_financials_json_dicts = {'Please enter the name of a financial indicator.'}
-            return json.dumps(historical_financials_json_dicts, default = str)
-
-    @app.route('/sub_sectors/<sub_industry_name>')
-    def company_financial_performance(sub_industry_name):
-        conn, cursor, financial_indicator = financial_performance_query_tools()
-        if sub_industry_name == 'all_sub_industries':
-            sector_name = financial_indicator
-            sub_industry_names = MixinCompanyFinancialsPricePE.get_all_sub_sector_names_in_sector(models.Company, sector_name, cursor)
-            return json.dumps({'sub_industry_names': sub_industry_names}, default=str)
-        else:
-            if financial_indicator in ['revenue', 'net_income', 'earnings_per_share', 'profit_margin']:
-                historical_financials_json_dicts = (models.SubIndustry.
-                                                        find_companies_quarterly_financials(sub_sector_name, financial_indicator, cursor))
-            elif financial_indicator in ['closing_price', 'price_earnings_ratio']:
-                historical_financials_json_dicts = (models.SubIndustry.
-                                                        find_company_quarterly_price_pe(sector_name, financial_indicator, cursor))
-            else:
-                historical_financials_json_dicts = {'Please enter the name of a financial indicator.'}
-            return json.dumps(historical_financials_json_dicts, default = str)
-
-    @app.route('/sectors/<sector_name>')
-    def get_sub_sector_names_within_sector(sector_name):
-        conn = db.get_db()
-        cursor = conn.cursor()
-        sub_sector_names = MixinSubSectorPricePE.get_sub_sector_names_of_sector(models.SubIndustry, sector_name, cursor) 
-        return json.dumps({'sub_sector_names': sub_sector_names}, default=str)     
+        return {'message': 'API is running.'}
+
+    from .routes.sector_routes import sector_bp
+    from .routes.sub_sector_routes import sub_sector_bp
+    from .routes.company_routes import company_bp
+    app.register_blueprint(sector_bp)
+    app.register_blueprint(sub_sector_bp)
+    app.register_blueprint(company_bp)
 
     return app

backend/api/src/adapters/wiki_page_client.py

@@ -2,7 +2,9 @@
 # coding: utf-8
 
 
+import os
 import pandas as pd
+import requests
 
 def ingest_sp500_stocks_info():
     """
@@ -18,26 +20,34 @@ def ingest_sp500_stocks_info():
     """
 
     sp500_wiki_data_filepath = "./api/data/sp500/raw_data/sp500_stocks_wiki_info.csv"
-    with open(sp500_wiki_data_filepath) as existing_file:
-        if not existing_file:
-            sp500_df = get_sp500_wiki_info()
-            employees_total_df = get_employees_total()
-            sp500_incl_employees_df = merge_df(sp500_df, employees_total_df)
-            sp500_wiki_data_filepath = save_csv(sp500_incl_employees_df, sp500_wiki_data_filepath)
+    import os
+    if not os.path.exists(sp500_wiki_data_filepath):
+        sp500_df = get_sp500_wiki_info()
+        employees_total_df = get_employees_total()
+        sp500_incl_employees_df = merge_df(sp500_df, employees_total_df)
+        sp500_wiki_data_filepath = save_csv(sp500_incl_employees_df, sp500_wiki_data_filepath)
     return sp500_wiki_data_filepath
 
+import requests
+
 def get_sp500_wiki_info():
     """ingest each and every S&P 500 company's basic info from the Wikipedia web page"""
-    sp500_df = pd.read_html('https://en.wikipedia.org/wiki/List_of_S%26P_500_companies')[0]
+    headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
+    url = 'https://en.wikipedia.org/wiki/List_of_S%26P_500_companies'
+    response = requests.get(url, headers=headers)
+    sp500_df = pd.read_html(response.text)[0]
     column_names = list(sp500_df.columns)
     column_names[0] = 'Ticker'
     sp500_df.columns = column_names
     return sp500_df
 
 def get_employees_total():
     """ ingest each company's total number of employees """
-    returned_dataframes = pd.read_html('https://www.liberatedstocktrader.com/sp-500-companies-list-by-number-of-employees/')
-    employees_total = returned_dataframes[2]
+    headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
+    url = 'https://www.liberatedstocktrader.com/sp-500-companies-list-by-number-of-employees/'
+    response = requests.get(url, headers=headers)
+    returned_dataframes = pd.read_html(response.text)
+    employees_total = returned_dataframes[0]
     employees_total_df = employees_total.iloc[1:, 1:].copy()
     employees_total_df.columns = employees_total.iloc[0, 1:]
     return employees_total_df
@@ -51,10 +61,11 @@ def merge_df(sp500_df, employees_total_df):
     sp500_incl_employees_df = sp500_incl_employees_df[security_col_notna]
     return sp500_incl_employees_df
 
-def save_csv(sp500_incl_employees_df, sp500_wiki_data_filepath):
-    # save the merged dataframe in a csv file
-    sp500_incl_employees_df.to_csv(sp500_wiki_data_filepath)
-    return sp500_wiki_data_filepath
+def save_csv(sp500_incl_employees_df, filepath):
+    """Save the dataframe to a CSV file."""
+    os.makedirs(os.path.dirname(filepath), exist_ok=True)
+    sp500_incl_employees_df.to_csv(filepath)
+    return filepath
 
 if __name__ == "__main__":
     ingest_sp500_stocks_info() 

backend/api/src/db/db.py

@@ -1,24 +1,17 @@
-from flask import current_app
-from flask import g
+from flask import current_app, g
 import psycopg2
 from datetime import datetime, timedelta
-from settings import DB_USER, DB_NAME, DB_HOST, DB_PASSWORD, DEBUG, TESTING # backend/settings.py
-
-# Connecting to Postgres on local Mac (parameters hard-coded):
-conn = psycopg2.connect(database = 'investment_analysis', user = 'postgres', password = 'postgres')
-cursor = conn.cursor()
-
-def get_db():
-    if "db" not in g:
-        # connect to postgres on the local computer
-        g.db = psycopg2.connect(user = 'postgres', password = 'postgres',
-            dbname = current_app.config['DB_NAME']) # apply this to user, password in __init__.py (at the top of this script, already imported from SETTINGS)
-
-        """
-        # connect to postgres on the AWS RDS instance
-        g.db = psycopg2.connect(user = 'postgres', password = 'postgres',
-            dbname = current_app.config['DATABASE'])
-        """
+# from settings import DB_USER, DB_NAME, DB_HOST, DB_PASSWORD, DEBUG, TESTING # backend/settings.py
+
+def get_db(db_name=None, db_user=None, db_password=None):
+    if 'db' not in g:
+        if db_name is None:
+            db_name = current_app.config.get('DB_NAME', 'investment_analysis')
+        if db_user is None:
+            db_user = current_app.config.get('DB_USER', 'postgres')
+        if db_password is None:
+            db_password = current_app.config.get('DB_PASSWORD', 'postgres')
+        g.db = psycopg2.connect(host='localhost', database=db_name, user=db_user, password=db_password)
     return g.db
 
 """
@@ -41,8 +34,7 @@ def close_db(e=None):
 def build_from_record(Class, record):
     if not record: return None
     attr = dict(zip(Class.columns, record))
-    obj = Class()
-    obj.__dict__ = attr
+    obj = Class(**attr)
     return obj
 
 def build_from_records(Class, records):

backend/api/src/models/company.py

@@ -35,32 +35,30 @@ def find_by_company_id(self, company_id, cursor):
         record = cursor.fetchone()
         return db.build_from_record(models.Company, record)
 
-    @classmethod
-    def to_company_financials_history_json(self, sub_industry_name, cursor):
-        # return in json format the financials and stock price, price-earnings-ratios of all the companies in a sub_industry
-        company_names = MixinCompanyPricePE.get_all_company_names_in_sub_sector(sub_industry_name, cursor)
-        companies_quarterly_financials_dict = {}
-        for company_name in company_names:
-            companies_quarterly_financials_dict[company_name] = to_quarterly_financials_json(self, company_name, cursor)
-        return companies_quarterly_financials_dict
-
     @classmethod
     def to_quarterly_financials_json(self, company_name, cursor):
-        quarterly_financials_json = self.__dict__
-        quarterly_reports_obj = self.get_company_quarterly_financials(self, company_name, cursor)
-        quarterly_financials_json['Quarterly_financials'] = [report_obj.__dict__ for report_obj in quarterly_reports_obj]
-        prices_pe_obj = self.get_company_quarterly_prices_pe(self, company_name, cursor)
-        quarterly_financials_json['Closing_prices_and_P/E_ratio'] = [
-                                                    price_pe_obj.__dict__ for price_pe_obj in prices_pe_obj]
-        return quarterly_financials_json
+        quarterly_reports = self.get_company_quarterly_financials(company_name, cursor)
+        prices_pe = self.get_company_quarterly_prices_pe(company_name, cursor)
+
+        # Create a dictionary for prices_pe for easier lookup
+        prices_pe_dict = {(p.year, p.quarter): p for p in prices_pe}
+
+        merged_data = []
+        for report in quarterly_reports:
+            price_pe_data = prices_pe_dict.get((report.year, report.quarter))
+            if price_pe_data:
+                merged_record = {**report.__dict__, **price_pe_data.__dict__}
+                merged_data.append(merged_record)
+
+        return merged_data
 
     @classmethod
     def get_company_quarterly_financials(self, company_name, cursor):
         sql_str = f"""
                     SELECT quarterly_reports.* 
                     FROM quarterly_reports JOIN {self.__table__}
                     ON quarterly_reports.company_id = {self.__table__}.id
-                    WHERE {self.__table__}.company_name = %s;        
+                    WHERE {self.__table__}.name = %s;        
                     """
         cursor.execute(sql_str, (company_name,))
         records = cursor.fetchall()
@@ -76,7 +74,7 @@ def get_company_quarterly_prices_pe(self, company_name, cursor):
                     """
         cursor.execute(sql_str, (company_name,))
         records = cursor.fetchall()
-        return db.build_from_records(models.QuarterlyReport, records)
+        return db.build_from_records(models.PricePE, records)
 
     @classmethod
     def find_companies_quarterly_financials(self, sub_sector_name:str, financial_indicator:str, cursor):
@@ -89,31 +87,29 @@ def find_companies_quarterly_financials(self, sub_sector_name:str, financial_ind
         financial_indicator name, year, quarter], and their corresponding values stored in a list as 
         the dictionary value.
         """
-        companies_quarterly_financials_json = self.to_company_quarterly_financials_json(sub_sector_name, financial_indicator, cursor)
+        companies_quarterly_financials_json = self.to_all_companies_quarterly_financials_json(sub_sector_name, financial_indicator, cursor)
         single_financial_indicator_json = extract_single_financial_indicator(financial_indicator, companies_quarterly_financials_json)
         return single_financial_indicator_json
 
     @classmethod
-    def to_company_quarterly_financials_json(self, sub_sector_name, financial_indicator, cursor):
-        company_names = MixinCompanyPricePE.get_all_company_names_in_sub_sector(self, sub_sector_name, cursor)
+    def to_all_companies_quarterly_financials_json(self, sub_sector_name, financial_indicator, cursor):
+        company_names = self.get_all_company_names_in_sub_sector(sub_sector_name, cursor)
         avg_quarterly_financials_dict = {}
         for company_name in company_names:
-            avg_quarterly_financials_dict[company_name] = (MixinCompanyFinancials.
-                                                                    to_quarterly_financials_json(self, company_name, cursor))
+            avg_quarterly_financials_dict[company_name] = self.to_quarterly_financials_json(company_name, cursor)
         return avg_quarterly_financials_dict
 
     @classmethod
     def find_company_quarterly_price_pe(self, sub_sector_name:str, financial_indicator:str, cursor):
-        companies_quarterly_price_pe_json = self.to_company_quarterly_price_pe_json(sub_sector_name, financial_indicator, cursor)
+        companies_quarterly_price_pe_json = self.to_all_companies_quarterly_price_pe_json(sub_sector_name, financial_indicator, cursor)
         single_financial_indicator_json = extract_single_financial_indicator(financial_indicator, companies_quarterly_price_pe_json)
         return single_financial_indicator_json
 
     @classmethod
-    def to_company_quarterly_price_pe_json(self, sub_sector_name, financial_indicator, cursor):
-        company_names = MixinCompanyPricePE.get_all_company_names_in_sub_sector(self, sub_sector_name, cursor)
+    def to_all_companies_quarterly_price_pe_json(self, sub_sector_name, financial_indicator, cursor):
+        company_names = self.get_all_company_names_in_sub_sector(sub_sector_name, cursor)
         avg_quarterly_price_pe_dict = {}
         for company_name in company_names:
-            avg_quarterly_price_pe_dict[company_name] = (MixinCompanyPricePE.
-                                                                to_quarterly_price_pe_json(self, company_name, cursor))
+            avg_quarterly_price_pe_dict[company_name] = self.to_quarterly_price_pe_json(company_name, cursor)
         return avg_quarterly_price_pe_dict
 

backend/api/src/models/price_pe.py

@@ -3,19 +3,22 @@
 
 class PricePE:
     __table__ = 'prices_pe' 
-    columns = ['id', 'date', 'company_id', 'closing_price', 'price_earnings_ratio']
+    columns = ['id', 'date', 'company_id', 'closing_price', 'price_earnings_ratio', 'year', 'quarter']
 
     def __init__(self, **kwargs):
         for key in kwargs.keys():
             if key not in self.columns:
                 raise f"{key} not in {self.columns}"
         for k, v in kwargs.items():
             setattr(self, k, v)
+        if hasattr(self, 'date'):
+            self.year = self.date.year
+            self.quarter = (self.date.month - 1) // 3 + 1
 
     @classmethod
     def find_by_company_id(self, company_id, cursor):
         sql_str = f"""SELECT * FROM {self.__table__}
                         WHERE company_id = %s;"""
         cursor.execute(sql_str, (company_id,))
         records = cursor.fetchall()
-        return db.build_from_records(models.SubIndustry, records)
+        return db.build_from_records(self, records)