5105221888

pending completion

Build # 5105221888

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge cccf7af1d into f14ac1458

Pull Request Pull Request #99: Refactor cache

Run Details

105 of 139 branches covered (75.54%)

Branch coverage included in aggregate %.

58 of 62 new or added lines in 5 files covered. (93.55%)

81 existing lines in 5 files now uncovered.

405 of 496 relevant lines covered (81.65%)

0.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

74.71

/src/stocktracer/cli.py

"""This is the CLI class for stocktracer."""
import importlib
import io
import logging
import warnings
from pathlib import Path
from typing import Literal, Optional, Union

import pandas as pd
from beartype import beartype
from beartype.typing import Sequence, Tuple

from stocktracer import cache
from stocktracer.interface import Analysis as AnalysisInterface
from stocktracer.interface import Options as CliOptions
from stocktracer.interface import ReportDate

logger = logging.getLogger(__name__)


@beartype
def get_analysis_instance(module_name: str, options: CliOptions) -> AnalysisInterface:
    """Dynamically import and load the Analysis class from a module.

    Args:
        module_name (str): full name of the module. For example, "my.module"
        options (CliOptions): options provided from the CLI

    Returns:
        AnalysisInterface: analysis instance
    """
    module = importlib.import_module(module_name)
    class_ = getattr(module, "Analysis")
    instance = class_(options)
    assert isinstance(instance, AnalysisInterface)
    return instance


ReportFormat = Literal["csv", "md", "json", "txt"]


@beartype
class Cli:
    """Tools for gathering resources, analyzing data, and publishing the results."""

    return_results: bool = True

    def analyze(  # pylint: disable=too-many-arguments
        self,
        tickers: Union[Sequence[str], str],
        analysis_plugin: str = "stocktracer.analysis.annual_reports",
        final_year: int = ReportDate().year,
        final_quarter: int = ReportDate().quarter,
        report_format: ReportFormat = "txt",
        report_file: Optional[Path | str] = None,
    ) -> Optional[pd.DataFrame]:
        """Perform stock analysis.

        Args:
            tickers (Union[Sequence[str], str]): tickers to include in the analysis
            analysis_plugin (str): module to load for analysis
            final_year (int): last year to consider for report collection
            final_quarter (int): last quarter to consider for report collection
            report_format (ReportFormat): Format of the report. Options include: csv, json, md (markdown)
            report_file (Optional[Path | str]): Where to store the report. Required if report_format is specified.

        Returns:
            Optional[pd.DataFrame]: results of analysis
        """
        if report_file:
            report_file = Path(report_file)
        if isinstance(tickers, str):
            tickers = list([tickers])
        else:
            tickers = list(tickers)

        tickers.sort()

        results, analysis_module = self._get_result(
            tickers=tickers,
            analysis_plugin=analysis_plugin,
            final_year=final_year,
            final_quarter=final_quarter,
        )

        self._generate_report(report_format, report_file, results)
        if analysis_module.under_development:
            warnings.warn(
                "This analysis module is under development and may be incorrect, incomplete, or may change."
            )
        if self.return_results:
            return results
        return None

    @classmethod
    def _generate_report(
        cls,
        report_format: ReportFormat,
        report_file: Path | io.StringIO | None,
        results: pd.DataFrame,
    ):
        if report_file is None:
            report_file = io.StringIO()
        results = results.transpose()
        match report_format.lower():
            case "csv":
                results.to_csv(report_file)
            case "md":
                results.to_markdown(report_file)
            case "json":
                results.to_json(report_file)
            case "txt":
                results.to_string(report_file)
        if isinstance(report_file, io.StringIO):
            print(report_file.getvalue())

    @cache.results.memoize(typed=True, expire=60 * 60 * 24 * 7, tag="results")
    def _get_result(
        self,
        tickers: list[str],
        analysis_plugin: str,
        final_year: int,
        final_quarter: int,
    ) -> Tuple[Optional[pd.DataFrame], AnalysisInterface]:
        """Gets the results.

        Args:
            tickers (list[str]): _description_
            analysis_plugin (str): _description_
            final_year (int): _description_
            final_quarter (int): _description_

        Raises:
            LookupError: no analysis results found

        Returns:
            Tuple[Optional[pd.DataFrame], AnalysisInterface]: _description_
        """
        analysis_module: AnalysisInterface = get_analysis_instance(
            analysis_plugin,
            CliOptions(
                tickers=frozenset(tickers),
                final_year=final_year,
                final_quarter=final_quarter,
            ),
        )

        # Call analysis plugin
        results = None
        results = analysis_module.analyze()
        if results is None or results.empty:
            raise LookupError("No analysis results available!")
        return results, analysis_module

1	"""This is the CLI class for stocktracer."""
2	import importlib	1✔
3	import io	1✔
4	import logging	1✔
5	import warnings	1✔
6	from pathlib import Path	1✔
7	from typing import Literal, Optional, Union	1✔
8
9	import pandas as pd	1✔
10	from beartype import beartype	1✔
11	from beartype.typing import Sequence, Tuple	1✔
12
13	from stocktracer import cache	1✔
14	from stocktracer.interface import Analysis as AnalysisInterface	1✔
15	from stocktracer.interface import Options as CliOptions	1✔
16	from stocktracer.interface import ReportDate	1✔
17
18	logger = logging.getLogger(__name__)	1✔
19
20
21	@beartype	1✔
22	def get_analysis_instance(module_name: str, options: CliOptions) -> AnalysisInterface:	1✔
23	"""Dynamically import and load the Analysis class from a module.
24
25	Args:
26	module_name (str): full name of the module. For example, "my.module"
27	options (CliOptions): options provided from the CLI
28
29	Returns:
30	AnalysisInterface: analysis instance
31	"""
32	module = importlib.import_module(module_name)	1✔
33	class_ = getattr(module, "Analysis")	1✔
34	instance = class_(options)	1✔
35	assert isinstance(instance, AnalysisInterface)	1✔
36	return instance	1✔
37
38
39	ReportFormat = Literal["csv", "md", "json", "txt"]	1✔
40
41
42	@beartype	1✔
43	class Cli:	1✔
44	"""Tools for gathering resources, analyzing data, and publishing the results."""
45
46	return_results: bool = True	1✔
47
48	def analyze( # pylint: disable=too-many-arguments	1✔
49	self,
50	tickers: Union[Sequence[str], str],
51	analysis_plugin: str = "stocktracer.analysis.annual_reports",
52	final_year: int = ReportDate().year,
53	final_quarter: int = ReportDate().quarter,
54	report_format: ReportFormat = "txt",
55	report_file: Optional[Path \| str] = None,
56	) -> Optional[pd.DataFrame]:
57	"""Perform stock analysis.
58
59	Args:
60	tickers (Union[Sequence[str], str]): tickers to include in the analysis
61	analysis_plugin (str): module to load for analysis
62	final_year (int): last year to consider for report collection
63	final_quarter (int): last quarter to consider for report collection
64	report_format (ReportFormat): Format of the report. Options include: csv, json, md (markdown)
65	report_file (Optional[Path \| str]): Where to store the report. Required if report_format is specified.
66
67	Returns:
68	Optional[pd.DataFrame]: results of analysis
69	"""
70	if report_file:	1!
UNCOV 71	report_file = Path(report_file)	×
72	if isinstance(tickers, str):	1✔
73	tickers = list([tickers])	1✔
74	else:
75	tickers = list(tickers)	1✔
76
77	tickers.sort()	1✔
78
79	results, analysis_module = self._get_result(	1✔
80	tickers=tickers,
81	analysis_plugin=analysis_plugin,
82	final_year=final_year,
83	final_quarter=final_quarter,
84	)
85
UNCOV 86	self._generate_report(report_format, report_file, results)	×
UNCOV 87	if analysis_module.under_development:	×
UNCOV 88	warnings.warn(	×
89	"This analysis module is under development and may be incorrect, incomplete, or may change."
90	)
UNCOV 91	if self.return_results:	×
UNCOV 92	return results	×
UNCOV 93	return None	×
94
95	@classmethod	1✔
96	def _generate_report(	1✔
97	cls,
98	report_format: ReportFormat,
99	report_file: Path \| io.StringIO \| None,
100	results: pd.DataFrame,
101	):
102	if report_file is None:	1!
UNCOV 103	report_file = io.StringIO()	×
104	results = results.transpose()	1✔
105	match report_format.lower():	1✔
106	case "csv":	1✔
107	results.to_csv(report_file)	1✔
108	case "md":	1✔
109	results.to_markdown(report_file)	1✔
110	case "json":	1!
111	results.to_json(report_file)	1✔
UNCOV 112	case "txt":	×
UNCOV 113	results.to_string(report_file)	×
114	if isinstance(report_file, io.StringIO):	1!
115	print(report_file.getvalue())	1✔
116
117	@cache.results.memoize(typed=True, expire=60 * 60 * 24 * 7, tag="results")	1✔
118	def _get_result(	1✔
119	self,
120	tickers: list[str],
121	analysis_plugin: str,
122	final_year: int,
123	final_quarter: int,
124	) -> Tuple[Optional[pd.DataFrame], AnalysisInterface]:
125	"""Gets the results.
126
127	Args:
128	tickers (list[str]): _description_
129	analysis_plugin (str): _description_
130	final_year (int): _description_
131	final_quarter (int): _description_
132
133	Raises:
134	LookupError: no analysis results found
135
136	Returns:
137	Tuple[Optional[pd.DataFrame], AnalysisInterface]: _description_
138	"""
139	analysis_module: AnalysisInterface = get_analysis_instance(	1✔
140	analysis_plugin,
141	CliOptions(
142	tickers=frozenset(tickers),
143	final_year=final_year,
144	final_quarter=final_quarter,
145	),
146	)
147
148	# Call analysis plugin
149	results = None	1✔
150	results = analysis_module.analyze()	1✔
151	if results is None or results.empty:	1!
152	raise LookupError("No analysis results available!")	1✔
NEW 153	return results, analysis_module	×

gyund / fundamental-analysis / 5105221888

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous