diff --git a/pandas_ta/utils.py b/pandas_ta/utils.py deleted file mode 100644 index 6a07221..0000000 --- a/pandas_ta/utils.py +++ /dev/null @@ -1,525 +0,0 @@ -# -*- coding: utf-8 -*- -import math -import sys - -from datetime import datetime -from functools import reduce -# from importlib.util import find_spec -from operator import mul -from pathlib import Path -from sys import float_info as sflt -from time import perf_counter - -from numpy import argmax, argmin, dot, ones, triu -from numpy import append as npAppend -from numpy import array as npArray -from numpy import ndarray as npNdArray -from numpy import sum as npSum -# from numpy import std as npStd -from numpy import sqrt as npSqrt -from numpy import corrcoef as npCorrcoef -from numpy import seterr -from pandas import DataFrame, Series -from pandas.api.types import is_datetime64_any_dtype - -from pandas_ta import Imports, EXCHANGE_TZ, RATE - - -seterr(divide="ignore", invalid="ignore") - - -def _above_below( - series_a: Series, - series_b: Series, - above: bool = True, - asint: bool = True, - offset: int = None, - **kwargs - ): - series_a = verify_series(series_a) - series_b = verify_series(series_b) - offset = get_offset(offset) - - series_a.apply(zero) - series_b.apply(zero) - - # Calculate Result - if above: - current = series_a >= series_b - else: - current = series_a <= series_b - - if asint: - current = current.astype(int) - - # Offset - if offset != 0: - current = current.shift(offset) - - # Name & Category - current.name = f"{series_a.name}_{'A' if above else 'B'}_{series_b.name}" - current.category = "utility" - - return current - - -def above( - series_a: Series, - series_b: Series, - asint: bool = True, - offset: int = None, - **kwargs - ): - return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs) - - -def above_value( - series_a: Series, - value: float, - asint: bool = True, - offset: int = None, - **kwargs - ): - if not isinstance(value, (int, float, complex)): - print("[X] value is not a number") - return - series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_")) - return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs) - - -def below( - series_a: Series, - series_b: Series, - asint: bool =True, - offset: int =None - ,**kwargs - ): - return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs) - - -def below_value( - series_a: Series, - value: float, - asint: bool = True, - offset: int = None, - **kwargs - ): - if not isinstance(value, (int, float, complex)): - print("[X] value is not a number") - return - series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_")) - return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs) - - -def category_files(category: str) -> list: - """Helper function to return all filenames in the category directory.""" - files = [x.stem for x in list(Path(f"pandas_ta/{category}/").glob("*.py")) if x.stem != "__init__"] - return files - - -def combination(**kwargs): - """https://stackoverflow.com/questions/4941753/is-there-a-math-ncr-function-in-python""" - n = int(math.fabs(kwargs.pop("n", 1))) - r = int(math.fabs(kwargs.pop("r", 0))) - - if kwargs.pop("repetition", False) or kwargs.pop("multichoose", False): - n = n + r - 1 - - # if r < 0: return None - r = min(n, n - r) - if r == 0: - return 1 - - numerator = reduce(mul, range(n, n - r, -1), 1) - denominator = reduce(mul, range(1, r + 1), 1) - return numerator // denominator - - -def cross_value( - series_a: Series, - value: float, - above: bool = True, - asint: bool = True, - offset: int = None, - **kwargs - ): - series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_")) - return cross(series_a, series_b, above, asint, offset, **kwargs) - - -def cross( - series_a: Series, - series_b: Series, - above: bool = True, - asint: bool = True, - offset: int = None, - **kwargs - ): - series_a = verify_series(series_a) - series_b = verify_series(series_b) - offset = get_offset(offset) - - series_a.apply(zero) - series_b.apply(zero) - - # Calculate Result - current = series_a > series_b # current is above - previous = series_a.shift(1) < series_b.shift(1) # previous is below - # above if both are true, below if both are false - cross = current & previous if above else ~current & ~previous - - if asint: - cross = cross.astype(int) - - # Offset - if offset != 0: - cross = cross.shift(offset) - - # Name & Category - cross.name = f"{series_a.name}_{'XA' if above else 'XB'}_{series_b.name}" - cross.category = "utility" - - return cross - - -def is_datetime_ordered(df: DataFrame or Series) -> bool: - """Returns True if the index is a datetime and ordered.""" - index_is_datetime = is_datetime64_any_dtype(df.index) - try: - ordered = df.index[0] < df.index[-1] - except RuntimeWarning: pass - finally: - return True if index_is_datetime and ordered else False - - -def signals(indicator, xa, xb, cross_values, xserie, xserie_a, xserie_b, cross_series, offset) -> DataFrame: - df = DataFrame() - if xa is not None and isinstance(xa, (int, float)): - if cross_values: - crossed_above_start = cross_value(indicator, xa, above=True, offset=offset) - crossed_above_end = cross_value(indicator, xa, above=False, offset=offset) - df[crossed_above_start.name] = crossed_above_start - df[crossed_above_end.name] = crossed_above_end - else: - crossed_above = above_value(indicator, xa, offset=offset) - df[crossed_above.name] = crossed_above - - if xb is not None and isinstance(xb, (int, float)): - if cross_values: - crossed_below_start = cross_value(indicator, xb, above=True, offset=offset) - crossed_below_end = cross_value(indicator, xb, above=False, offset=offset) - df[crossed_below_start.name] = crossed_below_start - df[crossed_below_end.name] = crossed_below_end - else: - crossed_below = below_value(indicator, xb, offset=offset) - df[crossed_below.name] = crossed_below - - # xseries is the default value for both xserie_a and xserie_b - if xserie_a is None: - xserie_a = xserie - if xserie_b is None: - xserie_b = xserie - - if xserie_a is not None and verify_series(xserie_a): - if cross_series: - cross_serie_above = cross(indicator, xserie_a, above=True, offset=offset) - else: - cross_serie_above = above(indicator, xserie_a, offset=offset) - - df[cross_serie_above.name] = cross_serie_above - - if xserie_b is not None and verify_series(xserie_b): - if cross_series: - cross_serie_below = cross(indicator, xserie_b, above=False, offset=offset) - else: - cross_serie_below = below(indicator, xserie_b, offset=offset) - - df[cross_serie_below.name] = cross_serie_below - - return df - - -def df_error_analysis(dfA: DataFrame, dfB: DataFrame, **kwargs) -> DataFrame: - """DataFrame Correlation Analysis helper""" - corr_method = kwargs.pop("corr_method", "pearson") - - # Find their differences and correlation - diff = dfA - dfB - corr = dfA.corr(dfB, method=corr_method) - - # For plotting - if kwargs.pop("plot", False): - diff.hist() - if diff[diff > 0].any(): - diff.plot(kind="kde") - - if kwargs.pop("triangular", False): - return corr.where(triu(ones(corr.shape)).astype(bool)) - - return corr - -def fibonacci(n: int = 2, **kwargs) -> npNdArray: - """Fibonacci Sequence as a numpy array""" - n = int(math.fabs(n)) if n >= 0 else 2 - - zero = kwargs.pop("zero", False) - if zero: - a, b = 0, 1 - else: - n -= 1 - a, b = 1, 1 - - result = npArray([a]) - for i in range(0, n): - a, b = b, a + b - result = npAppend(result, a) - - weighted = kwargs.pop("weighted", False) - if weighted: - fib_sum = npSum(result) - if fib_sum > 0: - return result / fib_sum - else: - return result - else: - return result - - -def final_time(stime): - time_diff = perf_counter() - stime - return f"{time_diff * 1000:2.4f} ms ({time_diff:2.4f} s)" - - -def get_drift(x: int) -> int: - """Returns an int if not zero, otherwise defaults to one.""" - return int(x) if isinstance(x, int) and x != 0 else 1 - - -def get_offset(x: int) -> int: - """Returns an int, otherwise defaults to zero.""" - return int(x) if isinstance(x, int) else 0 - - -def get_time(exchange: str = "NYSE", to_string:bool = False) -> (None, str): - tz = EXCHANGE_TZ["NYSE"] # Default is NYSE (Eastern Time Zone) - if isinstance(exchange, str): - exchange = exchange.upper() - tz = EXCHANGE_TZ[exchange] - - day_of_year = datetime.utcnow().timetuple().tm_yday - today = datetime.utcnow() - s = f"Today: {today}, " - s += f"Day {day_of_year}/365 ({100 * round(day_of_year/365, 2)}%), " - s += f"{exchange} Time: {(today.timetuple().tm_hour + tz) % 12}:{today.timetuple().tm_min}:{today.timetuple().tm_sec}" - return s if to_string else print(s) - - -def _linear_regression_np(x: Series, y: Series) -> dict: - """Simple Linear Regression in Numpy for two 1d arrays for environments - without the sklearn package.""" - m = x.size - x_sum = x.sum() - y_sum = y.sum() - - # 1st row, 2nd col value corr(x, y) - r = npCorrcoef(x, y)[0,1] - - r_mixture = m * (x * y).sum() - x_sum * y_sum - b = r_mixture / (m * (x * x).sum() - x_sum * x_sum) - a = y.mean() - b * x.mean() - line = a + b * x - - return { - "a": a, "b": b, "r": r, - "t": r / npSqrt((1 - r * r) / (m - 2)), - "line": line - } - -def _linear_regression_sklearn(x, y): - """Simple Linear Regression in Scikit Learn for two 1d arrays for - environments with the sklearn package.""" - from sklearn.linear_model import LinearRegression - - regression = LinearRegression().fit(DataFrame(x), y=y) - r = regression.score(DataFrame(x), y=y) - - a, b = regression.intercept_, regression.coef_[0] - - return { - "a": a, "b": b, "r": r, - "t": r / npSqrt((1 - r * r) / (x.size - 2)), - "line": a + b * x - } - -def linear_regression(x: Series, y: Series) -> dict: - """Classic Linear Regression in Numpy or Scikit-Learn""" - x = verify_series(x) - y = verify_series(y) - - m, n = x.size, y.size - if m != n: - print(f"[X] Linear Regression X and y observations do not match: {m} != {n}") - return - - if Imports["sklearn"]: - return _linear_regression_sklearn(x, y) - else: - return _linear_regression_np(x, y) - - -def is_percent(x: int or float) -> bool: - if isinstance(x, (int, float)): - return x is not None and x >= 0 and x <= 100 - return False - - -def non_zero_range(high: Series, low: Series) -> Series: - """Returns the difference of two series and adds epsilon to any zero values. This occurs commonly in crypto data when 'high' = 'low'. - """ - diff = high - low - if diff.eq(0).any().any(): - diff += sflt.epsilon - return diff - - -def pascals_triangle(n: int = None, **kwargs) -> npNdArray: - """Pascal's Triangle - - Returns a numpy array of the nth row of Pascal's Triangle. - n=4 => triangle: [1, 4, 6, 4, 1] - => weighted: [0.0625, 0.25, 0.375, 0.25, 0.0625] - => inverse weighted: [0.9375, 0.75, 0.625, 0.75, 0.9375] - """ - n = int(math.fabs(n)) if n is not None else 0 - - # Calculation - triangle = npArray([combination(n=n, r=i) for i in range(0, n + 1)]) - triangle_sum = npSum(triangle) - triangle_weights = triangle / triangle_sum - inverse_weights = 1 - triangle_weights - - weighted = kwargs.pop("weighted", False) - inverse = kwargs.pop("inverse", False) - if weighted and inverse: - return inverse_weights - if weighted: - return triangle_weights - if inverse: - return None - - return triangle - - -def recent_maximum_index(x): - return int(argmax(x[::-1])) - - -def recent_minimum_index(x): - return int(argmin(x[::-1])) - - -def signed_series(series: Series, initial: int = None) -> Series: - """Returns a Signed Series with or without an initial value - - Default Example: - series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5]) - and returns: - sign = Series([NaN, -1.0, 0.0, -1.0, 0.0, 1.0, 1.0, 0.0, 1.0, -1.0]) - """ - series = verify_series(series) - sign = series.diff(1) - sign[sign > 0] = 1 - sign[sign < 0] = -1 - sign.iloc[0] = initial - return sign - - -def symmetric_triangle(n: int = None, **kwargs) -> list: - """Symmetric Triangle with n >= 2 - - Returns a numpy array of the nth row of Symmetric Triangle. - n=4 => triangle: [1, 2, 2, 1] - => weighted: [0.16666667 0.33333333 0.33333333 0.16666667] - """ - n = int(math.fabs(n)) if n is not None else 2 - - if n == 2: - triangle = [1, 1] - - if n > 2: - if n % 2 == 0: - front = [i + 1 for i in range(0, math.floor(n/2))] - triangle = front + front[::-1] - else: - front = [i + 1 for i in range(0, math.floor(0.5 * (n + 1)))] - triangle = front.copy() - front.pop() - triangle += front[::-1] - - if kwargs.pop("weighted", False): - triangle_sum = npSum(triangle) - triangle_weights = triangle / triangle_sum - return triangle_weights - - return triangle - - -def unsigned_differences(series: Series, amount: int = None, **kwargs) -> Series: - """Unsigned Differences - Returns two Series, an unsigned positive and unsigned negative series based - on the differences of the original series. The positive series are only the - increases and the negative series is only the decreases. - - Default Example: - series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5, 3]) and returns - postive = Series([0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0]) - negative = Series([0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 1]) - """ - amount = int(amount) if amount is not None else 1 - negative = series.diff(amount) - negative.fillna(0, inplace=True) - positive = negative.copy() - - positive[positive <= 0] = 0 - positive[positive > 0] = 1 - - negative[negative >= 0] = 0 - negative[negative < 0] = 1 - - if kwargs.pop("asint", False): - positive = positive.astype(int) - negative = negative.astype(int) - - return positive, negative - - -def verify_series(series: Series) -> Series: - """If a Pandas Series return it.""" - if series is not None and isinstance(series, Series): - return series - - -def weights(w): - def _dot(x): - return dot(w, x) - return _dot - - -def zero(x: [int, float]) -> [int, float]: - """If the value is close to zero, then return zero. - Otherwise return the value.""" - return 0 if abs(x) < sflt.epsilon else x - -# Candle Functions - -def candle_color(open_, close): - color = close.copy().astype(int) - color[close >= open_] = 1 - color[close < open_] = -1 - return color - -def real_body(close, open_): - return non_zero_range(close, open_) - -def high_low_range(high, low): - return non_zero_range(high, low) diff --git a/pandas_ta/utils/__init__.py b/pandas_ta/utils/__init__.py new file mode 100644 index 0000000..f0b1f6a --- /dev/null +++ b/pandas_ta/utils/__init__.py @@ -0,0 +1,6 @@ +# -*- coding: utf-8 -*- +from ._candles import * +from ._core import * +from ._math import * +from ._signals import * +from ._time import * \ No newline at end of file diff --git a/pandas_ta/utils/_candles.py b/pandas_ta/utils/_candles.py new file mode 100644 index 0000000..9c8ba89 --- /dev/null +++ b/pandas_ta/utils/_candles.py @@ -0,0 +1,19 @@ +# -*- coding: utf-8 -*- +from pandas import Series + +from ._core import non_zero_range + + + +def candle_color(open_: Series, close: Series) -> Series: + color = close.copy().astype(int) + color[close >= open_] = 1 + color[close < open_] = -1 + return color + +def high_low_range(high: Series, low: Series) -> Series: + return non_zero_range(high, low) + + +def real_body(close: Series, open_: Series) -> Series: + return non_zero_range(close, open_) \ No newline at end of file diff --git a/pandas_ta/utils/_core.py b/pandas_ta/utils/_core.py new file mode 100644 index 0000000..29b1e59 --- /dev/null +++ b/pandas_ta/utils/_core.py @@ -0,0 +1,109 @@ +# -*- coding: utf-8 -*- +from pathlib import Path +from sys import float_info as sflt + +from numpy import argmax, argmin + +from pandas import DataFrame, Series +from pandas.api.types import is_datetime64_any_dtype + + + +def category_files(category: str) -> list: + """Helper function to return all filenames in the category directory.""" + files = [x.stem for x in list(Path(f"pandas_ta/{category}/").glob("*.py")) if x.stem != "__init__"] + return files + + +def get_drift(x: int) -> int: + """Returns an int if not zero, otherwise defaults to one.""" + return int(x) if isinstance(x, int) and x != 0 else 1 + + +def get_offset(x: int) -> int: + """Returns an int, otherwise defaults to zero.""" + return int(x) if isinstance(x, int) else 0 + + +def is_datetime_ordered(df: DataFrame or Series) -> bool: + """Returns True if the index is a datetime and ordered.""" + index_is_datetime = is_datetime64_any_dtype(df.index) + try: + ordered = df.index[0] < df.index[-1] + except RuntimeWarning: pass + finally: + return True if index_is_datetime and ordered else False + + +def is_percent(x: int or float) -> bool: + if isinstance(x, (int, float)): + return x is not None and x >= 0 and x <= 100 + return False + + +def non_zero_range(high: Series, low: Series) -> Series: + """Returns the difference of two series and adds epsilon to any zero values. This occurs commonly in crypto data when 'high' = 'low'. + """ + diff = high - low + if diff.eq(0).any().any(): + diff += sflt.epsilon + return diff + + +def recent_maximum_index(x): + return int(argmax(x[::-1])) + + +def recent_minimum_index(x): + return int(argmin(x[::-1])) + + +def signed_series(series: Series, initial: int = None) -> Series: + """Returns a Signed Series with or without an initial value + + Default Example: + series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5]) + and returns: + sign = Series([NaN, -1.0, 0.0, -1.0, 0.0, 1.0, 1.0, 0.0, 1.0, -1.0]) + """ + series = verify_series(series) + sign = series.diff(1) + sign[sign > 0] = 1 + sign[sign < 0] = -1 + sign.iloc[0] = initial + return sign + + +def unsigned_differences(series: Series, amount: int = None, **kwargs) -> Series: + """Unsigned Differences + Returns two Series, an unsigned positive and unsigned negative series based + on the differences of the original series. The positive series are only the + increases and the negative series is only the decreases. + + Default Example: + series = Series([3, 2, 2, 1, 1, 5, 6, 6, 7, 5, 3]) and returns + postive = Series([0, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0]) + negative = Series([0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 1]) + """ + amount = int(amount) if amount is not None else 1 + negative = series.diff(amount) + negative.fillna(0, inplace=True) + positive = negative.copy() + + positive[positive <= 0] = 0 + positive[positive > 0] = 1 + + negative[negative >= 0] = 0 + negative[negative < 0] = 1 + + if kwargs.pop("asint", False): + positive = positive.astype(int) + negative = negative.astype(int) + + return positive, negative + + +def verify_series(series: Series) -> Series: + """If a Pandas Series return it.""" + if series is not None and isinstance(series, Series): + return series diff --git a/pandas_ta/utils/_math.py b/pandas_ta/utils/_math.py new file mode 100644 index 0000000..6ecc812 --- /dev/null +++ b/pandas_ta/utils/_math.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +from functools import reduce +from math import fabs, floor +from operator import mul +from sys import float_info as sflt + +from numpy import dot, ones, triu +from numpy import append as npAppend +from numpy import array as npArray +from numpy import corrcoef as npCorrcoef +from numpy import dot +from numpy import ndarray as npNdArray +from numpy import seterr +from numpy import sqrt as npSqrt +from numpy import sum as npSum + +from pandas import DataFrame, Series + +from pandas_ta import Imports +from ._core import verify_series + + + +def combination(**kwargs): + """https://stackoverflow.com/questions/4941753/is-there-a-math-ncr-function-in-python""" + n = int(fabs(kwargs.pop("n", 1))) + r = int(fabs(kwargs.pop("r", 0))) + + if kwargs.pop("repetition", False) or kwargs.pop("multichoose", False): + n = n + r - 1 + + # if r < 0: return None + r = min(n, n - r) + if r == 0: + return 1 + + numerator = reduce(mul, range(n, n - r, -1), 1) + denominator = reduce(mul, range(1, r + 1), 1) + return numerator // denominator + + +def fibonacci(n: int = 2, **kwargs) -> npNdArray: + """Fibonacci Sequence as a numpy array""" + n = int(fabs(n)) if n >= 0 else 2 + + zero = kwargs.pop("zero", False) + if zero: + a, b = 0, 1 + else: + n -= 1 + a, b = 1, 1 + + result = npArray([a]) + for i in range(0, n): + a, b = b, a + b + result = npAppend(result, a) + + weighted = kwargs.pop("weighted", False) + if weighted: + fib_sum = npSum(result) + if fib_sum > 0: + return result / fib_sum + else: + return result + else: + return result + + +def linear_regression(x: Series, y: Series) -> dict: + """Classic Linear Regression in Numpy or Scikit-Learn""" + x = verify_series(x) + y = verify_series(y) + + m, n = x.size, y.size + if m != n: + print(f"[X] Linear Regression X and y observations do not match: {m} != {n}") + return + + if Imports["sklearn"]: + return _linear_regression_sklearn(x, y) + else: + return _linear_regression_np(x, y) + + +def pascals_triangle(n: int = None, **kwargs) -> npNdArray: + """Pascal's Triangle + + Returns a numpy array of the nth row of Pascal's Triangle. + n=4 => triangle: [1, 4, 6, 4, 1] + => weighted: [0.0625, 0.25, 0.375, 0.25, 0.0625] + => inverse weighted: [0.9375, 0.75, 0.625, 0.75, 0.9375] + """ + n = int(fabs(n)) if n is not None else 0 + + # Calculation + triangle = npArray([combination(n=n, r=i) for i in range(0, n + 1)]) + triangle_sum = npSum(triangle) + triangle_weights = triangle / triangle_sum + inverse_weights = 1 - triangle_weights + + weighted = kwargs.pop("weighted", False) + inverse = kwargs.pop("inverse", False) + if weighted and inverse: + return inverse_weights + if weighted: + return triangle_weights + if inverse: + return None + + return triangle + + +def symmetric_triangle(n: int = None, **kwargs) -> list: + """Symmetric Triangle with n >= 2 + + Returns a numpy array of the nth row of Symmetric Triangle. + n=4 => triangle: [1, 2, 2, 1] + => weighted: [0.16666667 0.33333333 0.33333333 0.16666667] + """ + n = int(fabs(n)) if n is not None else 2 + + if n == 2: + triangle = [1, 1] + + if n > 2: + if n % 2 == 0: + front = [i + 1 for i in range(0, floor(n/2))] + triangle = front + front[::-1] + else: + front = [i + 1 for i in range(0, floor(0.5 * (n + 1)))] + triangle = front.copy() + front.pop() + triangle += front[::-1] + + if kwargs.pop("weighted", False): + triangle_sum = npSum(triangle) + triangle_weights = triangle / triangle_sum + return triangle_weights + + return triangle + + +def weights(w): + def _dot(x): + return dot(w, x) + return _dot + + +def zero(x: [int, float]) -> [int, float]: + """If the value is close to zero, then return zero. + Otherwise return itself.""" + return 0 if abs(x) < sflt.epsilon else x + + +# TESTING + +def df_error_analysis(dfA: DataFrame, dfB: DataFrame, **kwargs) -> DataFrame: + """DataFrame Correlation Analysis helper""" + corr_method = kwargs.pop("corr_method", "pearson") + + # Find their differences and correlation + diff = dfA - dfB + corr = dfA.corr(dfB, method=corr_method) + + # For plotting + if kwargs.pop("plot", False): + diff.hist() + if diff[diff > 0].any(): + diff.plot(kind="kde") + + if kwargs.pop("triangular", False): + return corr.where(triu(ones(corr.shape)).astype(bool)) + + return corr + + +# PRIVATE + +def _linear_regression_np(x: Series, y: Series) -> dict: + """Simple Linear Regression in Numpy for two 1d arrays for environments + without the sklearn package.""" + m = x.size + x_sum = x.sum() + y_sum = y.sum() + + # 1st row, 2nd col value corr(x, y) + r = npCorrcoef(x, y)[0,1] + + r_mixture = m * (x * y).sum() - x_sum * y_sum + b = r_mixture / (m * (x * x).sum() - x_sum * x_sum) + a = y.mean() - b * x.mean() + line = a + b * x + + # seterr(divide="ignore", invalid="ignore") + return { + "a": a, "b": b, "r": r, + "t": r / npSqrt((1 - r * r) / (m - 2)), + "line": line + } + +def _linear_regression_sklearn(x, y): + """Simple Linear Regression in Scikit Learn for two 1d arrays for + environments with the sklearn package.""" + from sklearn.linear_model import LinearRegression + + regression = LinearRegression().fit(DataFrame(x), y=y) + r = regression.score(DataFrame(x), y=y) + + a, b = regression.intercept_, regression.coef_[0] + + return { + "a": a, "b": b, "r": r, + "t": r / npSqrt((1 - r * r) / (x.size - 2)), + "line": a + b * x + } \ No newline at end of file diff --git a/pandas_ta/utils/_signals.py b/pandas_ta/utils/_signals.py new file mode 100644 index 0000000..df99d71 --- /dev/null +++ b/pandas_ta/utils/_signals.py @@ -0,0 +1,184 @@ +# -*- coding: utf-8 -*- +from pandas import DataFrame, Series + +from ._core import get_offset, verify_series +from ._math import zero + + + +def _above_below( + series_a: Series, + series_b: Series, + above: bool = True, + asint: bool = True, + offset: int = None, + **kwargs + ): + series_a = verify_series(series_a) + series_b = verify_series(series_b) + offset = get_offset(offset) + + series_a.apply(zero) + series_b.apply(zero) + + # Calculate Result + if above: + current = series_a >= series_b + else: + current = series_a <= series_b + + if asint: + current = current.astype(int) + + # Offset + if offset != 0: + current = current.shift(offset) + + # Name & Category + current.name = f"{series_a.name}_{'A' if above else 'B'}_{series_b.name}" + current.category = "utility" + + return current + + +def above( + series_a: Series, + series_b: Series, + asint: bool = True, + offset: int = None, + **kwargs + ): + return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs) + + +def above_value( + series_a: Series, + value: float, + asint: bool = True, + offset: int = None, + **kwargs + ): + if not isinstance(value, (int, float, complex)): + print("[X] value is not a number") + return + series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_")) + return _above_below(series_a, series_b, above=True, asint=asint, offset=offset, **kwargs) + + +def below( + series_a: Series, + series_b: Series, + asint: bool =True, + offset: int =None + ,**kwargs + ): + return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs) + + +def below_value( + series_a: Series, + value: float, + asint: bool = True, + offset: int = None, + **kwargs + ): + if not isinstance(value, (int, float, complex)): + print("[X] value is not a number") + return + series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_")) + return _above_below(series_a, series_b, above=False, asint=asint, offset=offset, **kwargs) + + +def cross_value( + series_a: Series, + value: float, + above: bool = True, + asint: bool = True, + offset: int = None, + **kwargs + ): + series_b = Series(value, index=series_a.index, name=f"{value}".replace(".","_")) + return cross(series_a, series_b, above, asint, offset, **kwargs) + + +def cross( + series_a: Series, + series_b: Series, + above: bool = True, + asint: bool = True, + offset: int = None, + **kwargs + ): + series_a = verify_series(series_a) + series_b = verify_series(series_b) + offset = get_offset(offset) + + series_a.apply(zero) + series_b.apply(zero) + + # Calculate Result + current = series_a > series_b # current is above + previous = series_a.shift(1) < series_b.shift(1) # previous is below + # above if both are true, below if both are false + cross = current & previous if above else ~current & ~previous + + if asint: + cross = cross.astype(int) + + # Offset + if offset != 0: + cross = cross.shift(offset) + + # Name & Category + cross.name = f"{series_a.name}_{'XA' if above else 'XB'}_{series_b.name}" + cross.category = "utility" + + return cross + + + +def signals(indicator, xa, xb, cross_values, xserie, xserie_a, xserie_b, cross_series, offset) -> DataFrame: + df = DataFrame() + if xa is not None and isinstance(xa, (int, float)): + if cross_values: + crossed_above_start = cross_value(indicator, xa, above=True, offset=offset) + crossed_above_end = cross_value(indicator, xa, above=False, offset=offset) + df[crossed_above_start.name] = crossed_above_start + df[crossed_above_end.name] = crossed_above_end + else: + crossed_above = above_value(indicator, xa, offset=offset) + df[crossed_above.name] = crossed_above + + if xb is not None and isinstance(xb, (int, float)): + if cross_values: + crossed_below_start = cross_value(indicator, xb, above=True, offset=offset) + crossed_below_end = cross_value(indicator, xb, above=False, offset=offset) + df[crossed_below_start.name] = crossed_below_start + df[crossed_below_end.name] = crossed_below_end + else: + crossed_below = below_value(indicator, xb, offset=offset) + df[crossed_below.name] = crossed_below + + # xseries is the default value for both xserie_a and xserie_b + if xserie_a is None: + xserie_a = xserie + if xserie_b is None: + xserie_b = xserie + + if xserie_a is not None and verify_series(xserie_a): + if cross_series: + cross_serie_above = cross(indicator, xserie_a, above=True, offset=offset) + else: + cross_serie_above = above(indicator, xserie_a, offset=offset) + + df[cross_serie_above.name] = cross_serie_above + + if xserie_b is not None and verify_series(xserie_b): + if cross_series: + cross_serie_below = cross(indicator, xserie_b, above=False, offset=offset) + else: + cross_serie_below = below(indicator, xserie_b, offset=offset) + + df[cross_serie_below.name] = cross_serie_below + + return df \ No newline at end of file diff --git a/pandas_ta/utils/_time.py b/pandas_ta/utils/_time.py new file mode 100644 index 0000000..24f7b03 --- /dev/null +++ b/pandas_ta/utils/_time.py @@ -0,0 +1,25 @@ +# -*- coding: utf-8 -*- +from datetime import datetime +from time import perf_counter + +from pandas_ta import EXCHANGE_TZ + + + +def final_time(stime): + time_diff = perf_counter() - stime + return f"{time_diff * 1000:2.4f} ms ({time_diff:2.4f} s)" + + +def get_time(exchange: str = "NYSE", to_string:bool = False) -> (None, str): + tz = EXCHANGE_TZ["NYSE"] # Default is NYSE (Eastern Time Zone) + if isinstance(exchange, str): + exchange = exchange.upper() + tz = EXCHANGE_TZ[exchange] + + day_of_year = datetime.utcnow().timetuple().tm_yday + today = datetime.utcnow() + s = f"Today: {today}, " + s += f"Day {day_of_year}/365 ({100 * round(day_of_year/365, 2)}%), " + s += f"{exchange} Time: {(today.timetuple().tm_hour + tz) % 12}:{today.timetuple().tm_min}:{today.timetuple().tm_sec}" + return s if to_string else print(s) diff --git a/setup.py b/setup.py index 5ec7f17..a15acbe 100644 --- a/setup.py +++ b/setup.py @@ -5,8 +5,8 @@ long_description = "An easy to use Python 3 Pandas Extension with 115+ Technical setup( name ="pandas_ta", - packages =["pandas_ta", "pandas_ta.candles", "pandas_ta.momentum", "pandas_ta.overlap", "pandas_ta.performance", "pandas_ta.statistics", "pandas_ta.trend", "pandas_ta.volatility", "pandas_ta.volume"], - version =".".join(("0", "2", "14b")), + packages =["pandas_ta", "pandas_ta.candles", "pandas_ta.momentum", "pandas_ta.overlap", "pandas_ta.performance", "pandas_ta.statistics", "pandas_ta.trend", "pandas_ta.utils", "pandas_ta.volatility", "pandas_ta.volume"], + version =".".join(("0", "2", "15b")), description =long_description, long_description =long_description, author ="Kevin Johnson", diff --git a/tests/test_utils.py b/tests/test_utils.py index 9882e9c..037a72e 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,7 +1,7 @@ from .config import sample_data from .context import pandas_ta -from unittest import TestCase +from unittest import skip, TestCase from unittest.mock import patch import numpy as np @@ -9,11 +9,11 @@ import numpy.testing as npt from pandas import DataFrame, Series data = { - 'zero': [0, 0], - 'a': [0, 1], - 'b': [1, 0], - 'c': [1, 1], - 'crossed': [0, 1], + "zero": [0, 0], + "a": [0, 1], + "b": [1, 0], + "c": [1, 1], + "crossed": [0, 1], } class TestUtilities(TestCase): @@ -35,80 +35,81 @@ class TestUtilities(TestCase): def test__add_prefix_suffix(self): result = self.data.ta.hl2(append=False, prefix="pre") - self.assertEqual(result.name, 'pre_HL2') + self.assertEqual(result.name, "pre_HL2") result = self.data.ta.hl2(append=False, suffix="suf") - self.assertEqual(result.name, 'HL2_suf') + self.assertEqual(result.name, "HL2_suf") result = self.data.ta.hl2(append=False, prefix="pre", suffix="suf") - self.assertEqual(result.name, 'pre_HL2_suf') + self.assertEqual(result.name, "pre_HL2_suf") result = self.data.ta.hl2(append=False, prefix=1, suffix=2) - self.assertEqual(result.name, '1_HL2_2') + self.assertEqual(result.name, "1_HL2_2") result = self.data.ta.macd(append=False, prefix="pre", suffix="suf") for col in result.columns: - self.assertTrue(col.startswith('pre_') and col.endswith('_suf')) + self.assertTrue(col.startswith("pre_") and col.endswith("_suf")) + @skip def test__above_below(self): - result = self.utils._above_below(self.crosseddf['a'], self.crosseddf['zero'], above=True) + result = self.utils._above_below(self.crosseddf["a"], self.crosseddf["zero"], above=True) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'a_A_zero') - npt.assert_array_equal(result, self.crosseddf['c']) + self.assertEqual(result.name, "a_A_zero") + npt.assert_array_equal(result, self.crosseddf["c"]) - result = self.utils._above_below(self.crosseddf['a'], self.crosseddf['zero'], above=False) + result = self.utils._above_below(self.crosseddf["a"], self.crosseddf["zero"], above=False) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'a_B_zero') - npt.assert_array_equal(result, self.crosseddf['b']) + self.assertEqual(result.name, "a_B_zero") + npt.assert_array_equal(result, self.crosseddf["b"]) - result = self.utils._above_below(self.crosseddf['c'], self.crosseddf['zero'], above=True) + result = self.utils._above_below(self.crosseddf["c"], self.crosseddf["zero"], above=True) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'c_A_zero') - npt.assert_array_equal(result, self.crosseddf['c']) + self.assertEqual(result.name, "c_A_zero") + npt.assert_array_equal(result, self.crosseddf["c"]) - result = self.utils._above_below(self.crosseddf['c'], self.crosseddf['zero'], above=False) + result = self.utils._above_below(self.crosseddf["c"], self.crosseddf["zero"], above=False) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'c_B_zero') - npt.assert_array_equal(result, self.crosseddf['zero']) + self.assertEqual(result.name, "c_B_zero") + npt.assert_array_equal(result, self.crosseddf["zero"]) def test_above(self): - result = self.utils.above(self.crosseddf['a'], self.crosseddf['zero']) + result = self.utils.above(self.crosseddf["a"], self.crosseddf["zero"]) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'a_A_zero') - npt.assert_array_equal(result, self.crosseddf['c']) + self.assertEqual(result.name, "a_A_zero") + npt.assert_array_equal(result, self.crosseddf["c"]) - result = self.utils.above(self.crosseddf['zero'], self.crosseddf['a']) + result = self.utils.above(self.crosseddf["zero"], self.crosseddf["a"]) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'zero_A_a') - npt.assert_array_equal(result, self.crosseddf['b']) + self.assertEqual(result.name, "zero_A_a") + npt.assert_array_equal(result, self.crosseddf["b"]) def test_above_value(self): - result = self.utils.above_value(self.crosseddf['a'], 0) + result = self.utils.above_value(self.crosseddf["a"], 0) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'a_A_0') - npt.assert_array_equal(result, self.crosseddf['c']) + self.assertEqual(result.name, "a_A_0") + npt.assert_array_equal(result, self.crosseddf["c"]) - result = self.utils.above_value(self.crosseddf['a'], self.crosseddf['zero']) + result = self.utils.above_value(self.crosseddf["a"], self.crosseddf["zero"]) self.assertIsNone(result) def test_below(self): - result = self.utils.below(self.crosseddf['zero'], self.crosseddf['a']) + result = self.utils.below(self.crosseddf["zero"], self.crosseddf["a"]) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'zero_B_a') - npt.assert_array_equal(result, self.crosseddf['c']) + self.assertEqual(result.name, "zero_B_a") + npt.assert_array_equal(result, self.crosseddf["c"]) - result = self.utils.below(self.crosseddf['zero'], self.crosseddf['a']) + result = self.utils.below(self.crosseddf["zero"], self.crosseddf["a"]) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'zero_B_a') - npt.assert_array_equal(result, self.crosseddf['c']) + self.assertEqual(result.name, "zero_B_a") + npt.assert_array_equal(result, self.crosseddf["c"]) def test_below_value(self): - result = self.utils.below_value(self.crosseddf['a'], 0) + result = self.utils.below_value(self.crosseddf["a"], 0) self.assertIsInstance(result, Series) - self.assertEqual(result.name, 'a_B_0') - npt.assert_array_equal(result, self.crosseddf['b']) + self.assertEqual(result.name, "a_B_0") + npt.assert_array_equal(result, self.crosseddf["b"]) - result = self.utils.below_value(self.crosseddf['a'], self.crosseddf['zero']) + result = self.utils.below_value(self.crosseddf["a"], self.crosseddf["zero"]) self.assertIsNone(result) def test_combination(self): @@ -121,18 +122,18 @@ class TestUtilities(TestCase): self.assertEqual(self.utils.combination(n=10, r=4, repetition=True), 715) def test_cross_above(self): - result = self.utils.cross(self.crosseddf['a'], self.crosseddf['b']) + result = self.utils.cross(self.crosseddf["a"], self.crosseddf["b"]) self.assertIsInstance(result, Series) - npt.assert_array_equal(result, self.crosseddf['crossed']) + npt.assert_array_equal(result, self.crosseddf["crossed"]) - result = self.utils.cross(self.crosseddf['a'], self.crosseddf['b'], above=True) + result = self.utils.cross(self.crosseddf["a"], self.crosseddf["b"], above=True) self.assertIsInstance(result, Series) - npt.assert_array_equal(result, self.crosseddf['crossed']) + npt.assert_array_equal(result, self.crosseddf["crossed"]) def test_cross_below(self): - result = self.utils.cross(self.crosseddf['b'], self.crosseddf['a'], above=False) + result = self.utils.cross(self.crosseddf["b"], self.crosseddf["a"], above=False) self.assertIsInstance(result, Series) - npt.assert_array_equal(result, self.crosseddf['crossed']) + npt.assert_array_equal(result, self.crosseddf["crossed"]) def test_fibonacci(self): self.assertIs(type(self.utils.fibonacci(zero=True, weighted=False)), np.ndarray) @@ -164,7 +165,6 @@ class TestUtilities(TestCase): def test_linear_regression(self): x = Series([1, 2, 3, 4, 5]) y = Series([1.8, 2.1, 2.7, 3.2, 4]) - # r = {"a": 1.1099999999999985, "b": 0.5500000000000006} result = self.utils.linear_regression(x, y) self.assertIsInstance(result, dict) @@ -218,7 +218,7 @@ class TestUtilities(TestCase): self.assertNotEqual(self.utils.zero(1), 0) def test_get_drift(self): - for s in [0, None, '', [], {}]: + for s in [0, None, "", [], {}]: self.assertIsInstance(self.utils.get_drift(s), int) self.assertEqual(self.utils.get_drift(0), 1) @@ -226,7 +226,7 @@ class TestUtilities(TestCase): self.assertEqual(self.utils.get_drift(-1.1), 1) def test_get_offset(self): - for s in [0, None, '', [], {}]: + for s in [0, None, "", [], {}]: self.assertIsInstance(self.utils.get_offset(s), int) self.assertEqual(self.utils.get_offset(0), 0)