Compare commits

..

2 Commits

4 changed files with 76 additions and 17 deletions

View File

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

View File

@ -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())

View File

@ -1,5 +1,5 @@
import logging import logging
from typing import Union, List from typing import Union, List, Dict
from aiohttp.client import ClientSession from aiohttp.client import ClientSession
from bs4 import BeautifulSoup 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() item_name = str(tr.td.div.string).strip()
data_div = tr.find_all('td')[-1].div.div data_div = tr.find_all('td')[-1].div.div
values_str: str = data_div.attrs['data-chart-data'] 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 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, 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. 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)
}

View File

@ -74,7 +74,7 @@ class FunctionsTestCase(IsolatedAsyncioTestCase):
def test_extract_row_data(self): def test_extract_row_data(self):
test_row = self.test_soup.find('div', attrs={'class': 'financials'}).tbody.tr 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) output = functions.extract_row_data(test_row)
self.assertTupleEqual(expected_output, output) 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_is.return_value = {END_DATE: mock_end_dates, 'b': (2, 3)}
mock_get_cf.return_value = {END_DATE: mock_end_dates, 'c': (3, 4)} mock_get_cf.return_value = {END_DATE: mock_end_dates, 'c': (3, 4)}
expected_output = { 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() symbol, quarterly, mock_session = 'foo', False, MagicMock()
output = await functions.get_company_financials(symbol, quarterly, mock_session) 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): 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) mock_session_cls.return_value = mock_session_obj = self.get_mock_session(self.test_html)
symbol = 'foo' 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 = { 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) output = await functions.get_company_financials(symbol, session=mock_session_obj)
self.assertDictEqual(expected_output, output) self.assertDictEqual(expected_output, output)