import numpy as np import math def exponential_moving_average(data, key, alpha=0.5, ic=None): result = np.zeros_like(data[key]) if ic is None: r = data[key][0] else: r = ic n = 0 for v in data[key]: r = alpha*r + (1-alpha)*v result[n] = r n += 1 return np.array(result) def moving_average(data, key, window_days, ic=0): mem = [ic] * window_days result = np.zeros_like(data[key]) k=0 cumsum = ic n = 0 for v in data[key]: cumsum += (v - mem[k]) mem[k] = v k += 1 if k >= window_days: k=0 result[n] = cumsum/window_days n += 1 return result def moving_variance(data, key, window_days, ic=0): mem = [ic] * window_days result = np.zeros_like(data[key]) k=0 cumsum = ic data_mean = data[key] - moving_average(data, key, window_days, ic) n = 0 for v in data_mean: try: v2 = v * v except: print ('v', v) cumsum += (v2 - mem[k]) mem[k] = v2 k += 1 if k >= window_days: k=0 var = max(0, cumsum) / window_days result[n] = math.sqrt(var) n += 1 return result def moving_max(data, key, window_days, ic=-1e9): mem = [ic] * window_days result = np.zeros_like(data[key]) k=0 n = 0 for v in data[key]: mem[k] = v k += 1 if k >= window_days: k=0 result[n] = np.max(mem) n += 1 return result def moving_min(data, key, window_days, ic=1e9): mem = [ic] * window_days result = np.zeros_like(data[key]) k=0 n = 0 for v in data[key]: mem[k] = v k += 1 if k >= window_days: k=0 result[n] = np.min(mem) n += 1 return result def normalize(data, key, ic=None): result = np.zeros_like(data[key]) cumsum = 0 if ic is None: prev = data[key][0] else: prev = ic n = 0 for v in data[key]: if prev == 0: print (prev) cumsum += 100*(v - prev)/(prev) result[n] = cumsum if v > 0: prev = v n += 1 return result def colsum(data, keys): result = np.zeros(data['index'].shape) for key in keys: result += data[key] return result def macd(data, key): ema_short = exponential_moving_average(data, key, alpha=0.85) ema_long = exponential_moving_average(data, key, alpha=0.925) return ema_short - ema_long def bollinger(data, window_days, f=2): tp = colsum(data, keys=['high', 'low', 'close']) / 3 tp_dict = {'tp': tp} stddev = np.array(moving_variance(tp_dict, 'tp', window_days)) mid = np.array(moving_average(tp_dict, key='tp', window_days=window_days)) upper = mid + f * stddev lower = mid - f * stddev result = {'index' : data['index'], 'upper' : upper, 'mid' : mid, 'lower' : lower} return result