pandas_ta-0.1.66b.tar.gz

# -*- coding: utf-8 -*- import time from functools import wraps import pandas as pd from pandas.core.base import PandasObject from pandas_ta.candles import * from pandas_ta.momentum import * from pandas_ta.overlap import * from pandas_ta.performance import * from pandas_ta.statistics import * from pandas_ta.trend import * from pandas_ta.volatility import * from pandas_ta.volume import * from pandas_ta.utils import * version = ".".join(("0", "1", "66b")) def finalize(method): @wraps(method) def _wrapper(*class_methods, **method_kwargs): cm = class_methods[0] result = method(cm, **method_kwargs) cm._add_prefix_suffix(result, **method_kwargs) cm._append(result, **method_kwargs) return result return _wrapper class BasePandasObject(PandasObject): """Simple PandasObject Extension Ensures the DataFrame is not empty and has columns. It would be a sad Panda otherwise. Args: df (pd.DataFrame): Extends Pandas DataFrame """ def __init__(self, df, **kwargs): if df.empty: return if len(df.columns) > 0: self._df = df else: raise AttributeError(f" [X] No columns!") def __call__(self, kind, *args, **kwargs): raise NotImplementedError() @pd.api.extensions.register_dataframe_accessor('ta') class AnalysisIndicators(BasePandasObject): """AnalysisIndicators is a class that extends the Pandas DataFrame via Pandas @pd.api.extensions.register_dataframe_accessor('name') decorator. This Pandas Extension is named 'ta' for Technical Analysis that allows us to apply technical indicators by extension. Even though 'ta' is a Pandas DataFrame Extension, you can still call the Indicators individually. Use help() if needed. By default the 'ta' extension uses lower case column names: open, high, low, close, and volume. You can override the defaults by providing the it's replacement name when calling the indicator. For example, to call the indicator hl2(). With 'default' columns: open, high, low, close, and volume. >>> df.ta.hl2() >>> df.ta(kind='hl2') With DataFrame columns: Open, High, Low, Close, and Volume. >>> df.ta.hl2(high='High', low='Low') >>> df.ta(kind='hl2', high='High', low='Low') If you do not want to use a DataFrame Extension, just call it normally. >>> sma10 = ta.sma(df['Close']) # Default length=10 >>> sma50 = ta.sma(df['Close'], length=50) >>> ichimoku, span = ta.ichimoku(df['High'], df['Low'], df['Close']) Args: kind (str, optional): Default: None. Kind is the 'name' of the indicator. It converts kind to lowercase before calling. timed (bool, optional): Default: False. Curious about the execution speed? kwargs: Extension specific modifiers. append (bool, optional): Default: False. When True, it appends the resultant column(s) to the DataFrame. Returns: Most Indicators will return a Pandas Series. Others like MACD, BBANDS, KC, et al will return a Pandas DataFrame. Ichimoku on the other hand will return two DataFrames, the Ichimoku DataFrame for the known period and a Span DataFrame for the future of the Span values. Let's get started! 1. Loading the 'ta' module: >>> import pandas as pd >>> import ta as ta 2. Load some data: >>> df = pd.read_csv('AAPL.csv', index_col='date', parse_dates=True) 3. Help! 3a. General Help: >>> help(df.ta) >>> df.ta() 3b. Indicator Help: >>> help(ta.apo) 3c. Indicator Extension Help: >>> help(df.ta.apo) 4. Ways of calling an indicator. 4a. Calling just the MACD indicator without 'ta' DataFrame extension. >>> ta.apo(df['close']) 4b. Calling just the MACD indicator with 'ta' DataFrame extension. >>> df.ta.apo() 4c. Calling using kind. >>> df.ta(kind='apo') 5. Working with kwargs 5a. Append the result to the working df. >>> df.ta.apo(append=True) 5b. Timing an indicator. >>> apo = df.ta(kind='apo', timed=True) >>> print(apo.timed) """ _adjusted = None def __call__(self, kind=None, alias=None, timed=False, **kwargs): try: if isinstance(kind, str): kind = kind.lower() fn = getattr(self, kind) if timed: stime = time.time() # Run the indicator indicator = fn(**kwargs) if timed: time_diff = time.time() - stime ms = time_diff * 1000 indicator.timed = f"{ms:2.3f} ms ({time_diff:2.3f} s)" # print(f"execution time: {indicator.timed}") self._df.timed = indicator.timed # Add an alias if passed if alias: indicator.alias = f"{alias}" return indicator else: self.help() except: self.help() @property def adjusted(self) -> str: """property: df.ta.adjusted""" return self._adjusted @adjusted.setter def adjusted(self, value:str) -> None: """property: df.ta.adjusted = 'adj_close'""" if value is not None and isinstance(value, str): self._adjusted = value else: self._adjusted = None @property def datetime_ordered(self) -> bool: """Returns True if the index is a datetime and ordered.""" return is_datetime_ordered(self._df) @property def reverse(self) -> pd.DataFrame: """Reverses the DataFrame. Simply: df.iloc[::-1]""" return self._df.iloc[::-1] @property def version(self) -> str: """Returns the version.""" return version def _append(self, result=None, **kwargs): """Appends a Pandas Series or DataFrame columns to self._df.""" if 'append' in kwargs and kwargs['append']: df = self._df if df is None or result is None: return else: if isinstance(result, pd.DataFrame): for i, column in enumerate(result.columns): df[column] = result.iloc[:,i] else: df[result.name] = result def _add_prefix_suffix(self, result=None, **kwargs): """Add prefix and/or suffix to the result columns""" if result is None: return else: prefix = suffix = "" # delimiter = kwargs.pop("delimiter", "_") if "prefix" in kwargs: prefix = f"{kwargs['prefix']}_" if "suffix" in kwargs: suffix = f"_{kwargs['suffix']}" if isinstance(result, pd.Series): result.name = prefix + result.name + suffix else: result.columns = [prefix + column + suffix for column in result.columns] def _get_column(self, series, default): """Attempts to get the correct series or 'column' and return it.""" df = self._df if df is None: return # Explicitly passing a pd.Series to override default. if isinstance(series, pd.Series): return series # Apply default if no series nor a default. elif series is None or default is None: return df[self.adjusted] if self.adjusted is not None else df[default] # Ok. So it's a str. elif isinstance(series, str): # Return the df column since it's in there. if series in df.columns: return df[series] else: # Attempt to match the 'series' because it was likely misspelled. matches = df.columns.str.match(series, case=False) match = [i for i, x in enumerate(matches) if x] # If found, awesome. Retur