cli for writing to json/csv

src/mwfin/__init__.py

@@ -0,0 +1 @@
+from .functions import get_company_financials

src/mwfin/__main__.py

@@ -0,0 +1,59 @@
+import asyncio
+import sys
+import csv
+import json
+from argparse import ArgumentParser
+from pathlib import Path
+
+from aiohttp import ClientSession
+
+from . import get_company_financials
+
+
+JSON_EXT, CSV_EXT = '.json', '.csv'
+
+
+def write_to_csv(data, file_obj) -> None:
+    writer = csv.writer(file_obj)
+    for statement in data.values():
+        for key, values in statement.items():
+            writer.writerow([key] + list(values))
+
+
+async def main() -> None:
+    parser = ArgumentParser(description="Scrape company financials")
+
+    parser.add_argument(
+        '-s', '--symbol',
+        type=str,
+        help="Stock ticker symbol of the company to be scraped the financials of"
+    )
+
+    parser.add_argument(
+        '-f', '--to-file',
+        type=Path,
+        help="Writes results to the specified destination file. If omitted results are printed to stdout."
+    )
+
+    args = parser.parse_args()
+    session = ClientSession()
+    try:
+        data = await get_company_financials(args.symbol, False, session)
+    finally:
+        await session.close()
+    path: Path = args.to_file
+    if path is None:
+        print(json.dumps(data, indent=2))
+        return
+    if path.suffix.lower() == JSON_EXT:
+        with open(path, 'w') as f:
+            json.dump(data, f, indent=2)
+    elif path.suffix.lower() == CSV_EXT:
+        with open(path, 'w') as f:
+            write_to_csv(data, f)
+    else:
+        print('unknown extension')
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

src/mwfin/functions.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Union, List
+from typing import Union, List, Dict
 
 from aiohttp.client import ClientSession
 from bs4 import BeautifulSoup
@@ -63,7 +63,7 @@ def extract_row_data(tr: Tag) -> tuple[str, tuple[int]]:
     item_name = str(tr.td.div.string).strip()
     data_div = tr.find_all('td')[-1].div.div
     values_str: str = data_div.attrs['data-chart-data']
-    values = tuple(int(float(s)) for s in values_str.split(','))
+    values = tuple(int(float(s if s != '' else 0)) for s in values_str.split(','))
     return item_name, values
 
 
@@ -115,11 +115,12 @@ async def get_cash_flow_statement(ticker_symbol: str, quarterly: bool = False,
 
 
 async def get_company_financials(ticker_symbol: str, quarterly: bool = False,
-                                 session: ClientSession = None) -> ResultDict:
+                                 session: ClientSession = None) -> Dict[str, ResultDict]:
     """
     Returns all fundamentals (balance sheet, income statement and cash flow statement) of the specified company.
     """
-    financials = await get_balance_sheet(ticker_symbol, quarterly, session)
+    return {
-    financials.update(await get_income_statement(ticker_symbol, quarterly, session))
+        constants.BS: await get_balance_sheet(ticker_symbol, quarterly, session),
-    financials.update(await get_cash_flow_statement(ticker_symbol, quarterly, session))
+        constants.IS: await get_income_statement(ticker_symbol, quarterly, session),
-    return financials
+        constants.CF: await get_cash_flow_statement(ticker_symbol, quarterly, session)
+    }

tests/test_functions.py

@@ -74,7 +74,7 @@ class FunctionsTestCase(IsolatedAsyncioTestCase):
 
     def test_extract_row_data(self):
         test_row = self.test_soup.find('div', attrs={'class': 'financials'}).tbody.tr
-        expected_output = ('Item_1', (11000000, -22000000))
+        expected_output = ('Cash & Short Term Investments', (11000000, -22000000))
         output = functions.extract_row_data(test_row)
         self.assertTupleEqual(expected_output, output)
 
@@ -155,10 +155,9 @@ class FunctionsTestCase(IsolatedAsyncioTestCase):
         mock_get_is.return_value = {END_DATE: mock_end_dates, 'b': (2, 3)}
         mock_get_cf.return_value = {END_DATE: mock_end_dates, 'c': (3, 4)}
         expected_output = {
-            END_DATE: mock_end_dates,
+            BS: {END_DATE: mock_end_dates, 'a': (1, 2)},
-            'a': (1, 2),
+            IS: {END_DATE: mock_end_dates, 'b': (2, 3)},
-            'b': (2, 3),
+            CF: {END_DATE: mock_end_dates, 'c': (3, 4)}
-            'c': (3, 4)
         }
         symbol, quarterly, mock_session = 'foo', False, MagicMock()
         output = await functions.get_company_financials(symbol, quarterly, mock_session)
@@ -171,12 +170,11 @@ class FunctionsTestCase(IsolatedAsyncioTestCase):
     async def test_integration_get_company_financials(self, mock_session_cls):
         mock_session_cls.return_value = mock_session_obj = self.get_mock_session(self.test_html)
         symbol = 'foo'
-        # Since we mock the web request and always receive the same HTML markup,
+        # Since the web request is mocked we always receive the same HTML markup.
-        # and the function essentially does 3 separate requests always updating the output dictionary with the same
-        # data, we expect it to remain unchanged and only having one item.
         expected_output = {
-            END_DATE: ('End_Date_1', 'End_Date_2'),
+            BS: {END_DATE: ('End_Date_1', 'End_Date_2'), 'Cash & Short Term Investments': (11000000, -22000000)},
-            'Cash & Short Term Investments': (11000000, -22000000),
+            IS: {END_DATE: ('End_Date_1', 'End_Date_2'), 'Cash & Short Term Investments': (11000000, -22000000)},
+            CF: {END_DATE: ('End_Date_1', 'End_Date_2'), 'Cash & Short Term Investments': (11000000, -22000000)}
         }
         output = await functions.get_company_financials(symbol, session=mock_session_obj)
         self.assertDictEqual(expected_output, output)