5034066773

Build Type

github

Committed by GitHub

Commit Message

Merge 44f05d339 into 7ddf8cae8

Pull Request Pull Request #78: Add f-score and custom analysis module

Run Details

126 of 157 branches covered (80.25%)

Branch coverage included in aggregate %.

56 of 75 new or added lines in 7 files covered. (74.67%)

40 existing lines in 4 files now uncovered.

377 of 440 relevant lines covered (85.68%)

1.71 hits per line

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

79.65

/src/stocktracer/cli.py

"""This is the CLI class for stocktracer."""
import importlib
import io
import logging
import os
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 diskcache import Cache

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

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


@beartype
def get_default_cache_path() -> Path:
    """Get the default path for caching data.

    Returns:
        Path: path to cache data
    """
    return Path(os.getcwd()) / ".ticker-cache"


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],
        cache_path: Path = get_default_cache_path(),
        refresh: bool = False,
        analysis_plugin: str = "stocktracer.analysis.annual_reports",
        final_year: Optional[int] = None,
        final_quarter: Optional[int] = None,
        report_format: ReportFormat = "txt",
        report_file: Optional[Path] = None,
    ) -> Optional[pd.DataFrame]:
        """Perform stock analysis.

        Args:
            tickers (Union[Sequence[str], str]): tickers to include in the analysis
            cache_path (Path): path where to cache data
            refresh (bool): Whether to refresh the calculation or use the results from a prior one
            analysis_plugin (str): module to load for analysis
            final_year (Optional[int]): last year to consider for report collection
            final_quarter (Optional[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]): Where to store the report. Required if report_format is specified.

        Raises:
            LookupError: no analysis results found

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

        analysis_module: AnalysisInterface = get_analysis_instance(
            analysis_plugin,
            CliOptions(
                tickers=tickers,
                cache_path=cache_path,
                final_year=final_year,
                final_quarter=final_quarter,
            ),
        )

        cache, results_key, results = self._get_cached_results(
            tickers, cache_path, analysis_plugin
        )

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

            # Save one week expiry
            cache.set(key=results_key, value=results, expire=3600 * 24 * 7)

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

    def _get_cached_results(
        self, tickers, cache_path, analysis_plugin
    ) -> Tuple[Cache, str, Optional[pd.DataFrame]]:
        assert isinstance(tickers, frozenset)
        cache = Cache(directory=cache_path / "results")
        results_key = get_cached_results_key(tickers, analysis_plugin)
        results = cache.get(key=results_key, default=None)
        return cache, results_key, results


@beartype
def get_cached_results_key(tickers: frozenset[str], analysis_module: str) -> str:
    """Get the key used for caching results from analyzed data.

    >>> get_cached_results_key(frozenset({"aapl","msft"}),"my.analysis")
    'my.analysis-aapl-msft'

    Args:
        tickers (frozenset[str]): tickers to check
        analysis_module (str): name of analysis module

    Returns:
        str: string with the cached key
    """
    sorted_tickers = list(tickers)
    sorted_tickers.sort()
    results_key = "-".join(sorted_tickers)
    return "-".join((analysis_module, results_key))

1	"""This is the CLI class for stocktracer."""
2	import importlib	2✔
3	import io	2✔
4	import logging	2✔
5	import os	2✔
6	import warnings	2✔
7	from pathlib import Path	2✔
8	from typing import Literal, Optional, Union	2✔
9
10	import pandas as pd	2✔
11	from beartype import beartype	2✔
12	from beartype.typing import Sequence, Tuple	2✔
13	from diskcache import Cache	2✔
14
15	from stocktracer.interface import Analysis as AnalysisInterface	2✔
16	from stocktracer.interface import Options as CliOptions	2✔
17
18	logger = logging.getLogger(__name__)	2✔
19
20
21	@beartype	2✔
22	def get_analysis_instance(module_name: str, options: CliOptions) -> AnalysisInterface:	2✔
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)	2✔
33	class_ = getattr(module, "Analysis")	2✔
34	instance = class_(options)	2✔
35	assert isinstance(instance, AnalysisInterface)	2✔
36	return instance	2✔
37
38
39	@beartype	2✔
40	def get_default_cache_path() -> Path:	2✔
41	"""Get the default path for caching data.
42
43	Returns:
44	Path: path to cache data
45	"""
46	return Path(os.getcwd()) / ".ticker-cache"	2✔
47
48
49	ReportFormat = Literal["csv", "md", "json", "txt"]	2✔
50
51
52	@beartype	2✔
53	class Cli:	2✔
54	"""Tools for gathering resources, analyzing data, and publishing the results."""
55
56	return_results: bool = True	2✔
57
58	def analyze( # pylint: disable=too-many-arguments	2✔
59	self,
60	tickers: Union[Sequence[str], str],
61	cache_path: Path = get_default_cache_path(),
62	refresh: bool = False,
63	analysis_plugin: str = "stocktracer.analysis.annual_reports",
64	final_year: Optional[int] = None,
65	final_quarter: Optional[int] = None,
66	report_format: ReportFormat = "txt",
67	report_file: Optional[Path] = None,
68	) -> Optional[pd.DataFrame]:
69	"""Perform stock analysis.
70
71	Args:
72	tickers (Union[Sequence[str], str]): tickers to include in the analysis
73	cache_path (Path): path where to cache data
74	refresh (bool): Whether to refresh the calculation or use the results from a prior one
75	analysis_plugin (str): module to load for analysis
76	final_year (Optional[int]): last year to consider for report collection
77	final_quarter (Optional[int]): last quarter to consider for report collection
78	report_format (ReportFormat): Format of the report. Options include: csv, json, md (markdown)
79	report_file (Optional[Path]): Where to store the report. Required if report_format is specified.
80
81	Raises:
82	LookupError: no analysis results found
83
84	Returns:
85	Optional[pd.DataFrame]: results of analysis
86	"""
87	cache_path = Path(cache_path)	2✔
88	if report_file:	2!
89	report_file = Path(report_file)	×
90	if isinstance(tickers, str):	2✔
91	tickers = frozenset([tickers])	2✔
92	else:
93	tickers = frozenset(tickers)	2✔
94
95	analysis_module: AnalysisInterface = get_analysis_instance(	2✔
96	analysis_plugin,
97	CliOptions(
98	tickers=tickers,
99	cache_path=cache_path,
100	final_year=final_year,
101	final_quarter=final_quarter,
102	),
103	)
104
105	cache, results_key, results = self._get_cached_results(	2✔
106	tickers, cache_path, analysis_plugin
107	)
108
109	if (	2!
110	refresh
111	or results is None
112	or not isinstance(results, pd.DataFrame)
113	or results.empty
114	):
115	# Call analysis plugin
116	results = None	2✔
117	results = analysis_module.analyze()	2✔
118	if results is None or results.empty:	2!
119	raise LookupError("No analysis results available!")	2✔
120
121	# Save one week expiry
UNCOV 122	cache.set(key=results_key, value=results, expire=3600 * 24 * 7)	×
123
UNCOV 124	self._generate_report(report_format, report_file, results)	×
UNCOV 125	if analysis_module.under_development:	×
UNCOV 126	warnings.warn(	×
127	"This analysis module is under development and may be incorrect, incomplete, or may change."
128	)
UNCOV 129	if self.return_results:	×
UNCOV 130	return results	×
UNCOV 131	return None	×
132
133	@classmethod	2✔
134	def _generate_report(	2✔
135	cls,
136	report_format: ReportFormat,
137	report_file: Path \| io.StringIO \| None,
138	results: pd.DataFrame,
139	):
140	if report_file is None:	2!
UNCOV 141	report_file = io.StringIO()	×
142	results = results.transpose()	2✔
143	match report_format.lower():	2✔
144	case "csv":	2✔
145	results.to_csv(report_file)	2✔
146	case "md":	2✔
147	results.to_markdown(report_file)	2✔
148	case "json":	2!
149	results.to_json(report_file)	2✔
UNCOV 150	case "txt":	×
UNCOV 151	results.to_string(report_file)	×
152	if isinstance(report_file, io.StringIO):	2!
153	print(report_file.getvalue())	2✔
154
155	def _get_cached_results(	2✔
156	self, tickers, cache_path, analysis_plugin
157	) -> Tuple[Cache, str, Optional[pd.DataFrame]]:
158	assert isinstance(tickers, frozenset)	2✔
159	cache = Cache(directory=cache_path / "results")	2✔
160	results_key = get_cached_results_key(tickers, analysis_plugin)	2✔
161	results = cache.get(key=results_key, default=None)	2✔
162	return cache, results_key, results	2✔
163
164
165	@beartype	2✔
166	def get_cached_results_key(tickers: frozenset[str], analysis_module: str) -> str:	2✔
167	"""Get the key used for caching results from analyzed data.
168
169	>>> get_cached_results_key(frozenset({"aapl","msft"}),"my.analysis")
170	'my.analysis-aapl-msft'
171
172	Args:
173	tickers (frozenset[str]): tickers to check
174	analysis_module (str): name of analysis module
175
176	Returns:
177	str: string with the cached key
178	"""
179	sorted_tickers = list(tickers)	2✔
180	sorted_tickers.sort()	2✔
181	results_key = "-".join(sorted_tickers)	2✔
182	return "-".join((analysis_module, results_key))	2✔

gyund / fundamental-analysis / 5034066773

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