4909842897

Build Type

github

Committed by GitHub

Commit Message

Merge 4731bf138 into c0827eaa4

Pull Request Pull Request #36: Add support to generate reports in different formats supported by pandas

Run Details

82 of 108 branches covered (75.93%)

Branch coverage included in aggregate %.

162 of 182 new or added lines in 7 files covered. (89.01%)

22 existing lines in 1 file now uncovered.

293 of 338 relevant lines covered (86.69%)

1.73 hits per line

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

92.42

/src/stocktracer/cli.py

"""This is the CLI class for stocktracer."""
import importlib
import logging
import os
from pathlib import Path
from typing import 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"


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

    _default_analysis_module = "stocktracer.analysis.stub"

    def analyze(
        self,
        tickers: Union[Sequence[str], str],
        cache_path: Path = get_default_cache_path(),
        refresh: bool = False,
        analysis_plugin: str = _default_analysis_module,
    ) -> 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

        Raises:
            LookupError: no analysis results found

        Returns:
            Optional[DataFrame]: results of analysis
        """
        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)

        print(results.to_markdown())

    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 logging	2✔
4	import os	2✔
5	from pathlib import Path	2✔
6	from typing import Optional, Union	2✔
7
8	import pandas as pd	2✔
9	from beartype import beartype	2✔
10	from beartype.typing import Sequence, Tuple	2✔
11	from diskcache import Cache	2✔
12
13	from stocktracer.interface import Analysis as AnalysisInterface	2✔
14	from stocktracer.interface import Options as CliOptions	2✔
15
16	logger = logging.getLogger(__name__)	2✔
17
18
19	@beartype	2✔
20	def get_analysis_instance(module_name: str) -> AnalysisInterface:	2✔
21	"""Dynamically import and load the Analysis class from a module.
22
23	Args:
24	module_name (str): full name of the module. For example, "my.module"
25
26	Returns:
27	AnalysisInterface: analysis instance
28	"""
29	module = importlib.import_module(module_name)	2✔
30	class_ = getattr(module, "Analysis")	2✔
31	instance = class_()	2✔
32	assert isinstance(instance, AnalysisInterface)	2✔
33	return instance	2✔
34
35
36	@beartype	2✔
37	def get_default_cache_path() -> Path:	2✔
38	"""Get the default path for caching data.
39
40	Returns:
41	Path: path to cache data
42	"""
43	return Path(os.getcwd()) / ".ticker-cache"	2✔
44
45
46	@beartype	2✔
47	class Cli:	2✔
48	"""Tools for gathering resources, analyzing data, and publishing the results."""
49
50	_default_analysis_module = "stocktracer.analysis.stub"	2✔
51
52	def analyze(	2✔
53	self,
54	tickers: Union[Sequence[str], str],
55	cache_path: Path = get_default_cache_path(),
56	refresh: bool = False,
57	analysis_plugin: str = _default_analysis_module,
58	) -> Optional[pd.DataFrame]:
59	"""Perform stock analysis.
60
61	Args:
62	tickers (Union[Sequence[str], str]): tickers to include in the analysis
63	cache_path (Path): path where to cache data
64	refresh (bool): Whether to refresh the calculation or use the results from a prior one
65	analysis_plugin (str): module to load for analysis
66
67	Raises:
68	LookupError: no analysis results found
69
70	Returns:
71	Optional[DataFrame]: results of analysis
72	"""
73	if isinstance(tickers, str):	2✔
74	tickers = frozenset([tickers])	2✔
75	else:
76	tickers = frozenset(tickers)	2✔
77
78	analysis_module: AnalysisInterface = get_analysis_instance(analysis_plugin)	2✔
79	analysis_module.options = CliOptions(tickers=tickers, cache_path=cache_path)	2✔
80
81	cache, results_key, results = self._get_cached_results(	2✔
82	tickers, cache_path, analysis_plugin
83	)
84
85	if (	2!
86	refresh
87	or results is None
88	or not isinstance(results, pd.DataFrame)
89	or results.empty
90	):
91	# Call analysis plugin
92	results = analysis_module.analyze()	2✔
93	if results is None or results.empty:	2!
94	raise LookupError("No analysis results available!")	2✔
95
NEW 96	tickers = frozenset(tickers)	×
97
98	# Save one week expiry
NEW 99	cache.set(key=results_key, value=results, expire=3600 * 24 * 7)	×
100
NEW 101	print(results.to_markdown())	×
102
103	def _get_cached_results(	2✔
104	self, tickers, cache_path, analysis_plugin
105	) -> Tuple[Cache, str, Optional[pd.DataFrame]]:
106	assert isinstance(tickers, frozenset)	2✔
107	cache = Cache(directory=cache_path / "results")	2✔
108	results_key = get_cached_results_key(tickers, analysis_plugin)	2✔
109	results = cache.get(key=results_key, default=None)	2✔
110	return cache, results_key, results	2✔
111
112
113	@beartype	2✔
114	def get_cached_results_key(tickers: frozenset[str], analysis_module: str) -> str:	2✔
115	"""Get the key used for caching results from analyzed data.
116
117	>>> get_cached_results_key(frozenset({"aapl","msft"}),"my.analysis")
118	'my.analysis-aapl-msft'
119
120	Args:
121	tickers (frozenset[str]): tickers to check
122	analysis_module (str): name of analysis module
123
124	Returns:
125	str: string with the cached key
126	"""
127	sorted_tickers = list(tickers)	2✔
128	sorted_tickers.sort()	2✔
129	results_key = "-".join(sorted_tickers)	2✔
130	return "-".join((analysis_module, results_key))	2✔

gyund / fundamental-analysis / 4909842897

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