diff --git a/agent.py b/agent.py new file mode 100644 index 0000000..576d0b7 --- /dev/null +++ b/agent.py @@ -0,0 +1,42 @@ +import numpy as np + + +def buy_BTFD(name, data, range_days, cand_window, marker_key='close_n', thresh_max=-10, thresh_min=1, buy_callback=None): + N = len(data['index']) + key = 'BTFD' + buy_list = {'index' : np.array([None]*N), key : np.array([None]*N)} + cand = None + do_buy = False + for n in range(N-range_days, N): + vmax = data['Qmax'][n] + vmin = data['Qmin'][n] + trend = data['macd_fdd'][n] + index = data['index'][n] + value = data[marker_key][n] + if vmin <= thresh_min: + if vmax <= thresh_max: + cand = n + if trend >= 0: + do_buy = True + print ("{}: Buy on {} at {:0.2f}".format(name, index, value)) + cand = None + + if cand is not None: + if n - cand <= cand_window: + if trend >= 0: + do_buy = True + former = data[marker_key][cand] + print("{}: Delayed buy on {} at {:0.2f} ({:0.2f})".format(name, index, value, former-value)) + cand = None + else: + cand = None + + if do_buy: + do_buy = False + buy_list['index'][n] = index + buy_list[key][n] = value + if buy_callback is not None: + buy_callback({'name': name, 'item': {'value': value, 'date': index}}) + + return buy_list + diff --git a/functions.py b/functions.py new file mode 100644 index 0000000..be2e883 --- /dev/null +++ b/functions.py @@ -0,0 +1,137 @@ +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 + diff --git a/pandas_eval.py b/pandas_eval.py index 8f1bce3..5e8bb9b 100644 --- a/pandas_eval.py +++ b/pandas_eval.py @@ -1,8 +1,6 @@ import pandas as pd import pandas_datareader.data as web from pandas_datareader._utils import RemoteDataError -import numpy as np -import math import matplotlib.pyplot as plt from scipy.interpolate import UnivariateSpline import datetime as dt @@ -10,6 +8,8 @@ import os import time import matplotlib as mpl +import agent +from functions import * # Stock Investors Financial Math # https://www.fmlabs.com/reference/default.htm?url=SimpleMA.htm @@ -130,177 +130,6 @@ symbols = { 'INTC' : {'name' : 'Intel Corporation', 'currency' : '$'}, 'I' : {'name': 'IntelSat', 'currency' : '$'}} -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 - -def agent_buy(name, data, range_days, cand_window, marker_key='close_n', thresh_max=-10, thresh_min=1, buy_callback=None): - N = len(data['index']) - buy_list = {'index' : np.array([None]*N), 'Buy_dip' : np.array([None]*N)} - cand = None - do_buy = False - for n in range(N-range_days, N): - vmax = data['Qmax'][n] - vmin = data['Qmin'][n] - trend = data['macd_fdd'][n] - index = data['index'][n] - value = data[marker_key][n] - if vmin <= thresh_min: - if vmax <= thresh_max: - cand = n - if trend >= 0: - do_buy = True - print ("{}: Buy on {} at {:0.2f}".format(name, index, value)) - cand = None - - if cand is not None: - if n - cand <= cand_window: - if trend >= 0: - do_buy = True - former = data[marker_key][cand] - print("{}: Delayed buy on {} at {:0.2f} ({:0.2f})".format(name, index, value, former-value)) - cand = None - else: - cand = None - - if do_buy: - do_buy = False - buy_list['index'][n] = index - buy_list['Buy_dip'][n] = value - if buy_callback is not None: - buy_callback({'name': name, 'item': {'value': value, 'date': index}}) - - return buy_list - class Title(object): def __init__(self, symbol, params): self.symbol = symbol @@ -398,7 +227,7 @@ class Title(object): self.data['macd_fdd'] = np.transpose(yf_dd) def analyze(self, buy_callback, range_days): - return agent_buy(self.symbol, self.data, marker_key='close_n', cand_window=5, range_days=range_days, buy_callback=buy_callback) + return agent.buy_BTFD(self.symbol, self.data, marker_key='close_n', cand_window=5, range_days=range_days, buy_callback=buy_callback) @staticmethod def has_candidate(data, key): @@ -447,24 +276,24 @@ class Title(object): self.fig = plt.figure(figNum) self.ax = plt.subplot(100*num_subplots + 10 + 1) self.__plot(self.data, ['min', 'max', 'close_n']) - self.__ind(indicators, ['Buy_dip']) + self.__ind(indicators, ['BTFD']) plt.title('{} ({})'.format(self.name, self.symbol)) plt.legend() plt.grid() - self.ax = plt.subplot(100*num_subplots + 10 + 2) + plt.subplot(100*num_subplots + 10 + 2) # self.__plot(self.boll, ['lower', 'mid', 'upper']) self.__plot(self.data, ['Qmin', 'Qmax']) plt.legend() plt.grid() - self.ax = plt.subplot(100*num_subplots + 10 + 3) + plt.subplot(100*num_subplots + 10 + 3) self.__plot(self.data, ['macd', 'macd_f']) plt.legend() plt.grid() - self.ax = plt.subplot(100*num_subplots + 10 + 4) + plt.subplot(100*num_subplots + 10 + 4) self.__plot(self.data, ['macd_fd', 'macd_fdd']) plt.legend() plt.grid() @@ -497,7 +326,7 @@ else: title.fetch() title.statistics() ind = title.analyze(buy_callback, range_days=show_range_days) - if Title.has_candidate(ind, 'Buy_dip'): + if Title.has_candidate(ind, 'BTFD'): print('-----------------------------------------------') title.show(ind, figNum=figNum) figNum += 1