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
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)
financials.update(await get_income_statement(ticker_symbol, quarterly, session))
financials.update(await get_cash_flow_statement(ticker_symbol, quarterly, session))
return financials
return {
constants.BS: await get_balance_sheet(ticker_symbol, quarterly, session),
constants.IS: await get_income_statement(ticker_symbol, quarterly, session),
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):
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,
'a': (1, 2),
'b': (2, 3),
'c': (3, 4)
BS: {END_DATE: mock_end_dates, 'a': (1, 2)},
IS: {END_DATE: mock_end_dates, 'b': (2, 3)},
CF: {END_DATE: mock_end_dates, '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,
# 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.
# Since the web request is mocked we always receive the same HTML markup.
expected_output = {
END_DATE: ('End_Date_1', 'End_Date_2'),
'Cash & Short Term Investments': (11000000, -22000000),
BS: {END_DATE: ('End_Date_1', 'End_Date_2'), '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)