Skip to content

Reference

This module returns the most recent price of the selected company.

LivePriceDisplay

Returns the most recent price of the selected company.

Source code in src/live_price_display.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class LivePriceDisplay:
    """
    Returns the most recent price of the selected company.
    """

    @staticmethod
    def display_final_price_av(company_name: str) -> Union[str, dict, Any]:
        """
        Returns a the price using Alpha Vantage.

        Args:
            company_name: The ticker symbol of the company

        Returns:
            The most recent price in string
        """
        try:
            # Gets last available price by default
            price_params: dict = {
                "apikey": ALPHA_VANTAGE_API_KEY,
                "function": "TIME_SERIES_DAILY",
                "symbol": company_name,
            }
            price_response: requests.models.Response = requests.get(
                ALPHA_VANTAGE_ENDPOINT, params=price_params, timeout=20
            )
            if price_response.ok:
                response_data: dict = price_response.json()
                if "Time Series (Daily)" in response_data:
                    price_list: dict = response_data["Time Series (Daily)"]
                    most_recent_day: str = next(iter(price_list))
                    return price_list[most_recent_day]["4. close"]
                return response_data
            return price_response

        except (
            requests.exceptions.MissingSchema,
            requests.RequestException,
            KeyError,
            IndexError,
        ):
            return "Error fetching price"

    @staticmethod
    def display_final_price_yf(company_name: str) -> Union[float, str]:
        """
        Returns the price of the selected company using Yahoo Finance.

        Args:
            company_name: The ticker symbol of the company.

        Returns:
            The most recent price.
        """
        # Uncomment below for full company names in selection rather than ticker symbols.
        # conn = psycopg2.connect(database = "stocks", user='postgres', password='123456')
        # cursor = conn.cursor()
        # company_name = company_name.replace("\xa0", " ")
        # cursor.execute(f"SELECT ticker FROM companies WHERE company_name = '{company_name}';")
        # company_name = cursor.fetchall()[0]
        # conn.close()
        try:
            df: pd.DataFrame = yf.download(company_name)  # pylint: disable=C0103
            price: float = df.iloc[-1]["Close"]
            return round(price, 5)
        except IndexError:
            return "Error fetching price"

display_final_price_av(company_name) staticmethod

Returns a the price using Alpha Vantage.

Parameters:

Name Type Description Default
company_name str

The ticker symbol of the company

 required

Returns:

Type Description
Union[str, dict, Any]

The most recent price in string
Source code in src/live_price_display.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@staticmethod
def display_final_price_av(company_name: str) -> Union[str, dict, Any]:
    """
    Returns a the price using Alpha Vantage.

    Args:
        company_name: The ticker symbol of the company

    Returns:
        The most recent price in string
    """
    try:
        # Gets last available price by default
        price_params: dict = {
            "apikey": ALPHA_VANTAGE_API_KEY,
            "function": "TIME_SERIES_DAILY",
            "symbol": company_name,
        }
        price_response: requests.models.Response = requests.get(
            ALPHA_VANTAGE_ENDPOINT, params=price_params, timeout=20
        )
        if price_response.ok:
            response_data: dict = price_response.json()
            if "Time Series (Daily)" in response_data:
                price_list: dict = response_data["Time Series (Daily)"]
                most_recent_day: str = next(iter(price_list))
                return price_list[most_recent_day]["4. close"]
            return response_data
        return price_response

    except (
        requests.exceptions.MissingSchema,
        requests.RequestException,
        KeyError,
        IndexError,
    ):
        return "Error fetching price"

display_final_price_yf(company_name) staticmethod

Returns the price of the selected company using Yahoo Finance.

Parameters:

Name Type Description Default
company_name str

The ticker symbol of the company.

 required

Returns:

Type Description
Union[float, str]

The most recent price.
Source code in src/live_price_display.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@staticmethod
def display_final_price_yf(company_name: str) -> Union[float, str]:
    """
    Returns the price of the selected company using Yahoo Finance.

    Args:
        company_name: The ticker symbol of the company.

    Returns:
        The most recent price.
    """
    # Uncomment below for full company names in selection rather than ticker symbols.
    # conn = psycopg2.connect(database = "stocks", user='postgres', password='123456')
    # cursor = conn.cursor()
    # company_name = company_name.replace("\xa0", " ")
    # cursor.execute(f"SELECT ticker FROM companies WHERE company_name = '{company_name}';")
    # company_name = cursor.fetchall()[0]
    # conn.close()
    try:
        df: pd.DataFrame = yf.download(company_name)  # pylint: disable=C0103
        price: float = df.iloc[-1]["Close"]
        return round(price, 5)
    except IndexError:
        return "Error fetching price"

This module displays the most recent news of the selected company if available

NewsDisplay

Returns the most recent news of the selected company, if any.

Source code in src/news_display.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class NewsDisplay:
    """
    Returns the most recent news of the selected company, if any.
    """

    @staticmethod
    def _collect_news(company_name: str) -> list:
        """
        Collect recent news articles related to the selected company and format them.
        Args:
            company_name: The ticker symbol of the company

        Returns:
            five_article: The most recent five articles
        """
        news_params: dict = {"apiKey": NEWS_API_KEY, "qInTitle": company_name}

        news_response: requests.models.Response = requests.get(
            NEWS_ENDPOINT, params=news_params, timeout=20
        )
        articles: list = news_response.json()["articles"]
        five_articles: list = articles[:5]
        return five_articles

    def format_news_pyqt(self, company_name: str) -> list:
        """
        Formats the collected news to suit different PyQt5 UI.
        Args:
            company_name: The ticker symbol of the company

        Returns:
            five_article: The most recent five articles
        """
        news: list = self._collect_news(company_name)
        return [
            f"{article['title']}: '<a href=\"{article['url']}\">'{article['url']}'</a>'"
            for article in news
        ]

    def format_news_django(self, company_name: str) -> list:
        """
        Formats the collected news to suit different django UI.
        Args:
            company_name: The ticker symbol of the company

        Returns:
            five_article: The most recent five articles
        """
        news: list = self._collect_news(company_name)
        return [{"title": article["title"], "url": article["url"]} for article in news]

format_news_django(company_name)

Formats the collected news to suit different django UI. Args: company_name: The ticker symbol of the company

Returns:

Name Type Description
five_article list

The most recent five articles
Source code in src/news_display.py 
49
50
51
52
53
54
55
56
57
58
59
def format_news_django(self, company_name: str) -> list:
    """
    Formats the collected news to suit different django UI.
    Args:
        company_name: The ticker symbol of the company

    Returns:
        five_article: The most recent five articles
    """
    news: list = self._collect_news(company_name)
    return [{"title": article["title"], "url": article["url"]} for article in news]

format_news_pyqt(company_name)

Formats the collected news to suit different PyQt5 UI. Args: company_name: The ticker symbol of the company

Returns:

Name Type Description
five_article list

The most recent five articles
Source code in src/news_display.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def format_news_pyqt(self, company_name: str) -> list:
    """
    Formats the collected news to suit different PyQt5 UI.
    Args:
        company_name: The ticker symbol of the company

    Returns:
        five_article: The most recent five articles
    """
    news: list = self._collect_news(company_name)
    return [
        f"{article['title']}: '<a href=\"{article['url']}\">'{article['url']}'</a>'"
        for article in news
    ]

This module processes raw data and returns processed data in required format.

Model

Processes data and returns data in required format

Source code in src/model.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class Model:
    """Processes data and returns data in required format"""

    def __init__(self) -> None:
        """
        Constructs all the necessary attributes to make the necessary connection to the
        database.

        Args:
            None

        Returns:
            None
        """
        self.path: str = "../individual_stocks_5yr/"
        self.params: dict = {"host":"stocks-postgres",
                             "database":"stocks",
                             "user":"postgres",
                             "password":"123456",
                             "port":"5432"}

    def generate_company_list(self) -> Tuple[list, list]:
        """
        Returns a list of companies.

        Args:
            None

        Returns:
            A list of companies.
        """
        conn: psycopg2.extensions.connection = psycopg2.connect(**self.params)
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM companies;")
        records = cursor.fetchall()
        ticker_list: list = []
        companies_list: list = []
        for row in records:
            ticker: str
            company: str
            (_, ticker, company) = row
            company = company.replace("\xa0", " ")
            ticker_list.append(ticker)
            companies_list.append(company)
        conn.close()
        return ticker_list, companies_list

    def check_headers_and_data(self, file: str, expected_headers: list) -> bool:
        """
        Checks if each csv file has the expected headers and at least one data point for each header

        Args:
            file: The name of the file being checked
            expected_headers: The list of headers required

        Returns:
            The results of the file
        """
        has_expected_headers: bool = False
        has_data: bool = False
        try:
            parse_dates: list = ["date"]
            df: pd.DataFrame = pd.read_csv(  # pylint: disable=C0103
                self.path + file,
                skip_blank_lines=True,
                dtype={"date": "string", "close": "float64"},
                parse_dates=parse_dates,
            )
            headers: set = set(df.columns.to_list())
            expected_headers_copy: list = expected_headers[:]
            # Two conditions the while loop should break:
            # 1. No more headers in expected_headers_copy (all are met)
            # 2. At least one header is not met
            while expected_headers_copy:
                if expected_headers_copy[0] in headers:
                    expected_headers_copy.pop(0)
                else:
                    break
            if not expected_headers_copy:
                has_expected_headers = True
            else:
                return False
        except pd.errors.EmptyDataError:
            return False
        try:
            df.iloc[[0]]  # pylint: disable=E1101,W0104
            has_data = True
        except (ValueError, IndexError, NameError):
            return False
        return has_expected_headers and has_data

    def process_data(self) -> Union[pd.DataFrame, str]:
        """
        Slices the data as required.

        Args:
            None

        Returns:
            A DataFrame containing required information of all companies.
        """
        companies_list: Tuple[list, list] = self.generate_company_list()
        companies_data: dict = {}
        conn: psycopg2.extensions.connection = psycopg2.connect(**self.params)
        query: str = "SELECT company_id, trade_date, close FROM stock_prices_main \
        GROUP BY company_id, trade_date, close ORDER BY trade_date ASC;"
        all_data: pd.DataFrame = pd.read_sql(query, conn)
        grouped_data = all_data.groupby('company_id')[["trade_date", "close"]]
        for company_id, group_data in grouped_data:
            company_df: pd.DataFrame = group_data
            company_df["trade_date"] = pd.to_datetime(company_df["trade_date"])
            company_df["trade_date"] = company_df["trade_date"].dt.strftime("%Y-%m-%d")
            company_df["close"] = pd.to_numeric(company_df["close"])
            modified_data: dict = company_df.to_dict("list")
            assert isinstance(company_id, int)
            curr_company_ticker: str = companies_list[0][int(company_id) - 1]
            companies_data[curr_company_ticker] = modified_data
            # Uncomment below for full company names in selection rather than ticker symbols.
            # curr_company_name = companies_list[1][company_id-1]
            # companies_data[curr_company_name] = modified_data
        all_companies_data: pd.DataFrame = pd.DataFrame(companies_data)
        conn.close()
        return all_companies_data

__init__()

Constructs all the necessary attributes to make the necessary connection to the database.

Returns:

Type Description
None

None
Source code in src/model.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def __init__(self) -> None:
    """
    Constructs all the necessary attributes to make the necessary connection to the
    database.

    Args:
        None

    Returns:
        None
    """
    self.path: str = "../individual_stocks_5yr/"
    self.params: dict = {"host":"stocks-postgres",
                         "database":"stocks",
                         "user":"postgres",
                         "password":"123456",
                         "port":"5432"}

check_headers_and_data(file, expected_headers)

Checks if each csv file has the expected headers and at least one data point for each header

Parameters:

Name Type Description Default
file str

The name of the file being checked

 required
expected_headers list

The list of headers required

 required

Returns:

Type Description
bool

The results of the file
Source code in src/model.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def check_headers_and_data(self, file: str, expected_headers: list) -> bool:
    """
    Checks if each csv file has the expected headers and at least one data point for each header

    Args:
        file: The name of the file being checked
        expected_headers: The list of headers required

    Returns:
        The results of the file
    """
    has_expected_headers: bool = False
    has_data: bool = False
    try:
        parse_dates: list = ["date"]
        df: pd.DataFrame = pd.read_csv(  # pylint: disable=C0103
            self.path + file,
            skip_blank_lines=True,
            dtype={"date": "string", "close": "float64"},
            parse_dates=parse_dates,
        )
        headers: set = set(df.columns.to_list())
        expected_headers_copy: list = expected_headers[:]
        # Two conditions the while loop should break:
        # 1. No more headers in expected_headers_copy (all are met)
        # 2. At least one header is not met
        while expected_headers_copy:
            if expected_headers_copy[0] in headers:
                expected_headers_copy.pop(0)
            else:
                break
        if not expected_headers_copy:
            has_expected_headers = True
        else:
            return False
    except pd.errors.EmptyDataError:
        return False
    try:
        df.iloc[[0]]  # pylint: disable=E1101,W0104
        has_data = True
    except (ValueError, IndexError, NameError):
        return False
    return has_expected_headers and has_data

generate_company_list()

Returns a list of companies.

Returns:

Type Description
Tuple[list, list]

A list of companies.
Source code in src/model.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def generate_company_list(self) -> Tuple[list, list]:
    """
    Returns a list of companies.

    Args:
        None

    Returns:
        A list of companies.
    """
    conn: psycopg2.extensions.connection = psycopg2.connect(**self.params)
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM companies;")
    records = cursor.fetchall()
    ticker_list: list = []
    companies_list: list = []
    for row in records:
        ticker: str
        company: str
        (_, ticker, company) = row
        company = company.replace("\xa0", " ")
        ticker_list.append(ticker)
        companies_list.append(company)
    conn.close()
    return ticker_list, companies_list

process_data()

Slices the data as required.

Returns:

Type Description
Union[DataFrame, str]

A DataFrame containing required information of all companies.
Source code in src/model.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def process_data(self) -> Union[pd.DataFrame, str]:
    """
    Slices the data as required.

    Args:
        None

    Returns:
        A DataFrame containing required information of all companies.
    """
    companies_list: Tuple[list, list] = self.generate_company_list()
    companies_data: dict = {}
    conn: psycopg2.extensions.connection = psycopg2.connect(**self.params)
    query: str = "SELECT company_id, trade_date, close FROM stock_prices_main \
    GROUP BY company_id, trade_date, close ORDER BY trade_date ASC;"
    all_data: pd.DataFrame = pd.read_sql(query, conn)
    grouped_data = all_data.groupby('company_id')[["trade_date", "close"]]
    for company_id, group_data in grouped_data:
        company_df: pd.DataFrame = group_data
        company_df["trade_date"] = pd.to_datetime(company_df["trade_date"])
        company_df["trade_date"] = company_df["trade_date"].dt.strftime("%Y-%m-%d")
        company_df["close"] = pd.to_numeric(company_df["close"])
        modified_data: dict = company_df.to_dict("list")
        assert isinstance(company_id, int)
        curr_company_ticker: str = companies_list[0][int(company_id) - 1]
        companies_data[curr_company_ticker] = modified_data
        # Uncomment below for full company names in selection rather than ticker symbols.
        # curr_company_name = companies_list[1][company_id-1]
        # companies_data[curr_company_name] = modified_data
    all_companies_data: pd.DataFrame = pd.DataFrame(companies_data)
    conn.close()
    return all_companies_data