4954584384

Build Type

github

Committed by GitHub

Commit Message

Merge 436f7e159 into b165bae82

Pull Request Pull Request #45: Add support for transforming collected data into something useful

Run Details

109 of 144 branches covered (75.69%)

Branch coverage included in aggregate %.

39 of 54 new or added lines in 6 files covered. (72.22%)

38 existing lines in 2 files now uncovered.

336 of 394 relevant lines covered (85.28%)

1.71 hits per line

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

81.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) -> AnalysisInterface:
    """Dynamically import and load the Analysis class from a module.

    Args:
        module_name (str): full name of the module. For example, "my.module"

    Returns:
        AnalysisInterface: analysis instance
    """
    module = importlib.import_module(module_name)
    class_ = getattr(module, "Analysis")
    instance = class_()
    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"]


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

    return_results: bool = True

    def analyze(
        self,
        tickers: Union[Sequence[str], str],
        cache_path: str = str(get_default_cache_path()),
        refresh: bool = False,
        analysis_plugin: str = "stocktracer.analysis.stub",
        report_format: Optional[ReportFormat] = "md",
        report_file: Optional[str] = None,
    ) -> Optional[pd.DataFrame]:
        """Perform stock analysis.

        Args:
            tickers (Union[Sequence[str], str]): tickers to include in the analysis
            cache_path (str): 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
            report_format (Optional[ReportFormat]): Format of the report. Options include: csv, json, md (markdown)
            report_file (Optional[str]): 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)
        analysis_module.options = CliOptions(tickers=tickers, cache_path=cache_path)

        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 = analysis_module.analyze()
            if results is None or results.empty:
                raise LookupError("No analysis results available!")

            tickers = frozenset(tickers)

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

gyund / fundamental-analysis / 4954584384

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