mirror of
https://github.com/wassname/pandas-ta.git
synced 2026-09-11 12:30:30 +08:00
ENH strategy method params and col_numbers kwargs MAINT test_strat refactor + new test csv
This commit is contained in:
Kevin Johnson
+5242
-702
File diff suppressed because it is too large
Load Diff
+10
-10
@@ -45,7 +45,7 @@
|
||||
"Numpy v1.18.3\n",
|
||||
"Pandas v1.1.0\n",
|
||||
"mplfinance v0.12.6a3\n",
|
||||
"Pandas TA v0.2.04b\n"
|
||||
"Pandas TA v0.2.08b\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
@@ -112,13 +112,13 @@
|
||||
"text": [
|
||||
"[!] Loading All: SPY, QQQ, AAPL, TSLA\n",
|
||||
"[i] Loaded['D']: SPY_D.csv\n",
|
||||
"[i] Runtime: 1832.0667 ms (1.8321 s)\n",
|
||||
"[i] Runtime: 812.7898 ms (0.8128 s)\n",
|
||||
"[i] Loaded['D']: QQQ_D.csv\n",
|
||||
"[i] Runtime: 1728.6331 ms (1.7286 s)\n",
|
||||
"[i] Runtime: 818.7984 ms (0.8188 s)\n",
|
||||
"[i] Loaded['D']: AAPL_D.csv\n",
|
||||
"[i] Runtime: 948.8466 ms (0.9488 s)\n",
|
||||
"[i] Runtime: 1227.1223 ms (1.2271 s)\n",
|
||||
"[i] Loaded['D']: TSLA_D.csv\n",
|
||||
"[i] Runtime: 1005.4914 ms (1.0055 s)\n"
|
||||
"[i] Runtime: 1858.9133 ms (1.8589 s)\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
@@ -147,7 +147,7 @@
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"AAPL (5241, 10)\n",
|
||||
"Columns: open, high, low, close, volume, SMA_10, SMA_20, SMA_50, SMA_200, VOL_VOL_SMA_20\n"
|
||||
"Columns: open, high, low, close, volume, SMA_10, SMA_20, SMA_50, SMA_200, VOL_SMA_20\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
@@ -174,7 +174,7 @@
|
||||
"output_type": "stream",
|
||||
"text": [
|
||||
"AAPL (252, 10)\n",
|
||||
"Columns: open, high, low, close, volume, SMA_10, SMA_20, SMA_50, SMA_200, VOL_VOL_SMA_20\n"
|
||||
"Columns: open, high, low, close, volume, SMA_10, SMA_20, SMA_50, SMA_200, VOL_SMA_20\n"
|
||||
]
|
||||
}
|
||||
],
|
||||
@@ -353,7 +353,7 @@
|
||||
{
|
||||
"data": {
|
||||
"text/plain": [
|
||||
"<matplotlib.axes._subplots.AxesSubplot at 0x10922b4c0>"
|
||||
"<matplotlib.axes._subplots.AxesSubplot at 0x11196a5b0>"
|
||||
]
|
||||
},
|
||||
"execution_count": 9,
|
||||
@@ -396,7 +396,7 @@
|
||||
{
|
||||
"data": {
|
||||
"text/plain": [
|
||||
"<matplotlib.axes._subplots.AxesSubplot at 0x115368a00>"
|
||||
"<matplotlib.axes._subplots.AxesSubplot at 0x11d8d3fd0>"
|
||||
]
|
||||
},
|
||||
"execution_count": 10,
|
||||
@@ -441,7 +441,7 @@
|
||||
{
|
||||
"data": {
|
||||
"text/plain": [
|
||||
"<matplotlib.axes._subplots.AxesSubplot at 0x115a36070>"
|
||||
"<matplotlib.axes._subplots.AxesSubplot at 0x11d9e70d0>"
|
||||
]
|
||||
},
|
||||
"execution_count": 11,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
+85
-85
File diff suppressed because one or more lines are too long
+9
-18
@@ -23,13 +23,13 @@ from pandas_ta.volatility import *
|
||||
from pandas_ta.volume import *
|
||||
from pandas_ta.utils import *
|
||||
|
||||
version = ".".join(("0", "2", "07b"))
|
||||
version = ".".join(("0", "2", "08b"))
|
||||
|
||||
|
||||
# Strategy (Data)Class
|
||||
# Strategy DataClass
|
||||
@dataclass
|
||||
class Strategy:
|
||||
"""Strategy (Data)Class
|
||||
"""Strategy DataClass
|
||||
A way to name and group your favorite indicators
|
||||
|
||||
Args:
|
||||
@@ -126,7 +126,7 @@ class BasePandasObject(PandasObject):
|
||||
# Preemptively drop the rows that are all NaNs
|
||||
# Might need to be moved to AnalysisIndicators.__call__() to be
|
||||
# toggleable via kwargs.
|
||||
df.dropna(axis=0, inplace=True)
|
||||
# df.dropna(axis=0, inplace=True)
|
||||
# Preemptively rename columns to lowercase
|
||||
df.rename(columns=common_names, errors="ignore", inplace=True)
|
||||
|
||||
@@ -398,16 +398,9 @@ class AnalysisIndicators(BasePandasObject):
|
||||
method, args, kwargs = arguments
|
||||
|
||||
if method != "ichimoku":
|
||||
try:
|
||||
result = getattr(self,method)(*args, **kwargs)
|
||||
self._add_prefix_suffix(result=result, **kwargs)
|
||||
self._append(result=result, **kwargs)
|
||||
return result
|
||||
except:
|
||||
print(f"[X] Multiprocessing Error: {method} has not been added to the dataframe.")
|
||||
return
|
||||
return getattr(self, method)(*args, **kwargs)
|
||||
else:
|
||||
return getattr(self,method)(*args, **kwargs)[0]
|
||||
return getattr(self, method)(*args, **kwargs)[0]
|
||||
|
||||
|
||||
def _post_process(self, result, **kwargs) -> (pd.Series, pd.DataFrame):
|
||||
@@ -600,18 +593,16 @@ class AnalysisIndicators(BasePandasObject):
|
||||
custom_ta = [(ind["kind"], ind["params"] if "params" in ind and isinstance(ind["params"], tuple) else (), {**ind, **kwargs}) for ind in ta]
|
||||
|
||||
# Custom multiprocessing pool. Must be ordered for Chained Strategies
|
||||
results = pool.imap(self._mp_worker, custom_ta)
|
||||
results = pool.imap(self._mp_worker, custom_ta, self.cores)#, cpus) # May fix this to cpus if Chaining/Composition if it remains inconsistent
|
||||
else:
|
||||
default_ta = [(ind, tuple(), kwargs) for ind in ta]
|
||||
# All and Categorical multiprocessing pool. Speed over Order.
|
||||
results = pool.imap_unordered(self._mp_worker, default_ta)
|
||||
results = pool.imap_unordered(self._mp_worker, default_ta, self.cores)
|
||||
pool.close()
|
||||
pool.join()
|
||||
|
||||
# Apply prefixes/suffixes and appends indicator results to the DataFrame
|
||||
for r in results:
|
||||
self._add_prefix_suffix(result=r, **kwargs)
|
||||
self._append(result=r, **kwargs)
|
||||
[self._post_process(r, **kwargs) for r in results]
|
||||
|
||||
if timed: ftime = final_time(stime)
|
||||
|
||||
|
||||
@@ -24,7 +24,7 @@ def vwap(high, low, close, volume, offset=None, **kwargs):
|
||||
|
||||
# Name & Category
|
||||
vwap.name = "VWAP"
|
||||
vwap.category = 'overlap'
|
||||
vwap.category = "overlap"
|
||||
|
||||
return vwap
|
||||
|
||||
|
||||
+5
-2
@@ -1,5 +1,5 @@
|
||||
import os
|
||||
from pandas import read_csv
|
||||
from pandas import DatetimeIndex, read_csv
|
||||
|
||||
VERBOSE = True
|
||||
|
||||
@@ -13,9 +13,12 @@ sample_data = read_csv(
|
||||
f"data/SPY_D.csv",
|
||||
index_col=0,
|
||||
parse_dates=True,
|
||||
infer_datetime_format=False,
|
||||
infer_datetime_format=True,
|
||||
keep_date_col=True
|
||||
)
|
||||
sample_data.set_index(DatetimeIndex(sample_data["date"]), inplace=True, drop=True)
|
||||
sample_data.drop("date", axis=1, inplace=True)
|
||||
|
||||
|
||||
def error_analysis(df, kind, msg, icon=INFO, newline=True):
|
||||
if VERBOSE:
|
||||
|
||||
+85
-209
@@ -11,114 +11,92 @@ from pandas import DataFrame
|
||||
|
||||
from pandas_ta.utils import final_time
|
||||
|
||||
_verbose = False
|
||||
_timed = True
|
||||
speed_table = False
|
||||
cores = 4
|
||||
cumulative = False
|
||||
cores = 2
|
||||
speed_table = False
|
||||
strategy_timed = False
|
||||
timed = True
|
||||
verbose = False
|
||||
|
||||
class TestStrategyMethods(TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.data = sample_data
|
||||
cls.data.ta.cores = cores
|
||||
print(f"[i] Testing Cores: {cls.data.ta.cores}")
|
||||
# cls.data.ta.cores = cores
|
||||
cls.speed_test = DataFrame()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
del cls.data
|
||||
cls.speed_test = cls.speed_test.T
|
||||
cls.speed_test.index.name = "Test"
|
||||
cls.speed_test.columns = ["Columns", "Seconds"]
|
||||
if cumulative: cls.speed_test["Cum. Seconds"] = cls.speed_test["Seconds"].cumsum()
|
||||
if speed_table: cls.speed_test.to_csv("tests/speed_test.csv")
|
||||
print(cls.speed_test)
|
||||
if timed:
|
||||
print(f"[i] Cores: {cls.data.ta.cores}")
|
||||
print(f"[i] Total Datapoints: {cls.data.shape[0]}")
|
||||
print(cls.speed_test)
|
||||
del cls.data
|
||||
|
||||
|
||||
|
||||
def setUp(self): pass
|
||||
def tearDown(self): pass
|
||||
def setUp(self):
|
||||
self.added_cols = 0
|
||||
self.category = ""
|
||||
self.init_cols = len(self.data.columns)
|
||||
self.time_diff = 0
|
||||
self.result = None
|
||||
if verbose: print()
|
||||
if timed: self.stime = perf_counter()
|
||||
|
||||
def tearDown(self):
|
||||
if timed: self.time_diff = perf_counter() - self.stime
|
||||
self.added_cols = len(self.data.columns) - self.init_cols
|
||||
self.assertGreaterEqual(self.added_cols, 1)
|
||||
|
||||
self.result = self.data[self.data.columns[-self.added_cols:]]
|
||||
self.assertIsInstance(self.result, DataFrame)
|
||||
self.data.drop(columns=self.result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[self.category] = [self.added_cols, self.time_diff]
|
||||
|
||||
|
||||
# @skip
|
||||
def test_all(self):
|
||||
if _verbose: print()
|
||||
category = "All"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
self.category = "All"
|
||||
self.data.ta.strategy(verbose=verbose, timed=strategy_timed)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
@skip
|
||||
def test_all_strategy(self):
|
||||
self.data.ta.strategy(pandas_ta.AllStrategy, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
|
||||
# def test_all_strategy(self):
|
||||
# if _verbose: print()
|
||||
# init_cols = len(self.data.columns)
|
||||
# self.data.ta.strategy(pandas_ta.AllStrategy, verbose=_verbose)
|
||||
# added_cols = len(self.data.columns) - init_cols
|
||||
# self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
# result = self.data[self.data.columns[-added_cols:]]
|
||||
# self.assertIsInstance(result, DataFrame)
|
||||
# self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
# def test_all_name_strategy(self):
|
||||
# if _verbose: print()
|
||||
# init_cols = len(self.data.columns)
|
||||
# self.data.ta.strategy("All", verbose=_verbose)
|
||||
# added_cols = len(self.data.columns) - init_cols
|
||||
# self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
# result = self.data[self.data.columns[-added_cols:]]
|
||||
# self.assertIsInstance(result, DataFrame)
|
||||
# self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
@skip
|
||||
def test_all_name_strategy(self):
|
||||
self.category = "All"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_candles_category(self):
|
||||
if _verbose: print()
|
||||
category = "Candles"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Candles"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_common(self):
|
||||
if _verbose: print()
|
||||
category = "Common"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(pandas_ta.CommonStrategy, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
self.category = "Common"
|
||||
self.data.ta.strategy(pandas_ta.CommonStrategy, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_custom_a(self):
|
||||
if _verbose: print()
|
||||
self.category = "Custom A"
|
||||
|
||||
momo_bands_sma_ta = [
|
||||
{"kind":"sma", "length": 50}, # 1
|
||||
{"kind":"sma", "length": 200}, # 1
|
||||
{"kind":"bbands", "length": 20}, # 3
|
||||
{"kind":"macd"}, # 3
|
||||
{"kind":"rsi"}, # 1
|
||||
{"kind":"log_return", "cumulative": True}, # 1
|
||||
{"kind":"sma", "close": "CUMLOGRET_1", "length": 5, "suffix": "CUMLOGRET"}, # 1
|
||||
{"kind": "rsi"}, # 1
|
||||
{"kind": "macd"}, # 3
|
||||
{"kind": "sma", "length": 50}, # 1
|
||||
{"kind": "sma", "length": 200}, # 1
|
||||
{"kind": "bbands", "length": 20}, # 3
|
||||
{"kind": "log_return", "cumulative": True}, # 1
|
||||
{"kind": "sma", "close": "CUMLOGRET_1", "length": 5, "suffix": "CUMLOGRET"}, # 1
|
||||
]
|
||||
|
||||
custom = pandas_ta.Strategy(
|
||||
@@ -126,158 +104,56 @@ class TestStrategyMethods(TestCase):
|
||||
momo_bands_sma_ta, # ta
|
||||
"MACD and RSI Momo with BBANDS and SMAs 50 & 200 and Cumulative Log Returns" # description
|
||||
)
|
||||
self.data.ta.strategy(custom, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
category = "Custom A"
|
||||
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(custom, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertEqual(added_cols, 11)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
|
||||
# @skip
|
||||
def test_custom_args_tuple(self):
|
||||
if _verbose: print()
|
||||
self.category = "Custom B"
|
||||
|
||||
custom_args_ta = [
|
||||
{"kind":"fisher", "params": (13, 7)},
|
||||
{"kind":"macd", "params": (9, 19, 7)},
|
||||
{"kind":"macd", "params": (9, 19, 7), "col_numbers": (1,)},
|
||||
{"kind":"ema", "params": (5,)},
|
||||
{"kind":"linreg", "close": "EMA_5", "length": 8, "prefix": "EMA_5"}
|
||||
{"kind":"linreg", "close": "EMA_5", "length": 8, "suffix": "EMA_5"}
|
||||
]
|
||||
|
||||
custom = pandas_ta.Strategy(
|
||||
"Custom Args Tuple", custom_args_ta,
|
||||
"Allow for easy filling in indicator arguments without naming them"
|
||||
)
|
||||
self.data.ta.strategy(custom, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
category = "Custom B"
|
||||
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(custom, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
|
||||
# @skip
|
||||
def test_momentum_category(self):
|
||||
if _verbose: print()
|
||||
category = "Momentum"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Momentum"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_overlap_category(self):
|
||||
if _verbose: print()
|
||||
category = "Overlap"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Overlap"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_performance_category(self):
|
||||
if _verbose: print()
|
||||
category = "Performance"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Performance"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_statistics_category(self):
|
||||
if _verbose: print()
|
||||
category = "Statistics"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Statistics"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_trend_category(self):
|
||||
if _verbose: print()
|
||||
category = "Trend"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Trend"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_volatility_category(self):
|
||||
if _verbose: print()
|
||||
category = "Volatility"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Volatility"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
|
||||
# @skip
|
||||
def test_volume_category(self):
|
||||
if _verbose: print()
|
||||
category = "Volume"
|
||||
init_cols = len(self.data.columns)
|
||||
if _timed: stime = perf_counter()
|
||||
self.data.ta.strategy(category, verbose=_verbose)
|
||||
if _timed: time_diff = perf_counter() - stime
|
||||
added_cols = len(self.data.columns) - init_cols
|
||||
self.assertGreaterEqual(added_cols, 1)
|
||||
|
||||
result = self.data[self.data.columns[-added_cols:]]
|
||||
self.assertIsInstance(result, DataFrame)
|
||||
self.data.drop(columns=result.columns, axis=1, inplace=True)
|
||||
|
||||
self.speed_test[category] = [added_cols, time_diff]
|
||||
self.category = "Volume"
|
||||
self.data.ta.strategy(self.category, verbose=verbose, timed=strategy_timed)
|
||||
Reference in New Issue
Block a user