4904155797

Build Type

github

Committed by GitHub

Commit Message

Merge e833a57cd into 4b6424b61

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

Run Details

60 of 95 branches covered (63.16%)

Branch coverage included in aggregate %.

103 of 137 new or added lines in 7 files covered. (75.18%)

63 existing lines in 2 files now uncovered.

239 of 340 relevant lines covered (70.29%)

1.41 hits per line

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

84.29

/src/stocktracer/cli.py

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 pandas.core.groupby.generic import DataFrameGroupBy

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

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
class Cli:

    """Tools for gathering resources, analyzing data, and publishing the results."""

    _default_analysis_module = "stocktracer.analysis.stub"

    def _getDefaultCachePath() -> Path:
        """Get the default path for caching data

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

    def analyze(
        self,
        tickers: Union[Sequence[str], str],
        cache_path: Path = _getDefaultCachePath(),
        refresh: bool = False,
        analysis_plugin: str = _default_analysis_module,
    ) -> None:
        """Perform stock analysis

        Args:
            tickers (list[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
        """
        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._getCachedResults(
            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 not isinstance(results, pd.DataFrame):
                raise ValueError("analysis modules must return a pandas.DataFrame")
            if 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)

        if isinstance(results, pd.DataFrame):
            print(results.to_markdown())
        else:
            print(results)

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

    def _get_results_key(tickers: frozenset, analysis_module: str) -> str:
        """
        >>> Cli._get_results_key(frozenset({"aapl","msft"}),"my.analysis")
        'my.analysis-aapl-msft'
        """
        sorted_tickers = list(tickers)
        sorted_tickers.sort()
        results_key = "-".join(sorted_tickers)
        return "-".join((analysis_module, results_key))

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

gyund / fundamental-analysis / 4904155797

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