From 8889b6c77cd25fbfe100ee2a177ccfc521dc0740 Mon Sep 17 00:00:00 2001 From: Jens Ahrensfeld Date: Wed, 25 Dec 2019 19:18:55 +0000 Subject: [PATCH] - refactored git-svn-id: http://moon:8086/svn/projects/Stock@328 fda53097-d464-4ada-af97-ba876c37ca34 --- robot.py | 213 +++++-------------------------------------------------- stock.py | 182 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 200 insertions(+), 195 deletions(-) create mode 100644 stock.py diff --git a/robot.py b/robot.py index 5e8bb9b..d699de2 100644 --- a/robot.py +++ b/robot.py @@ -1,15 +1,3 @@ -import pandas as pd -import pandas_datareader.data as web -from pandas_datareader._utils import RemoteDataError -import matplotlib.pyplot as plt -from scipy.interpolate import UnivariateSpline -import datetime as dt -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 @@ -23,8 +11,19 @@ from functions import * # https://www.investopedia.com/articles/technical/02/050602.asp # https://ntguardian.wordpress.com/2016/09/19/introduction-stock-market-data-python-1 +import matplotlib.pyplot as plt +from stock import Stock + key = '0UO7Z2MVZ2YSQSVE' -show_range_days = 5 +params = { + 'show_range_days' : 5, + 'k_euro' : 1 / 1.11, + 'ema_alpha' : 0.75, + 'sma_days' : 10, + 'q_days' : 10, + 'fetch_on_outdated' : True +} + show_symbols = ['OHB.DE', 'ITMPF', 'PLUG', 'MOR.DE', 'CSCO', 'ERCA.DE', 'AVGO', 'DIS', 'UBSFF', 'DHER.DE', 'AIR'] show_symbols = ['CSCO'] #show_symbols = ['OHB.DE'] @@ -33,13 +32,6 @@ show_symbols = ['DHER.DE'] #show_symbols = ['WDI.DE'] #show_symbols = ['EVT.DE'] show_symbols = [] -k_euro = 1 / 1.11 -ema_alpha = 0.75 -sma_days = 10 -q_days = 10 -fetch_on_outdated = True - -mpl.rc('figure', max_open_warning = 0) symbols = { @@ -130,175 +122,6 @@ symbols = { 'INTC' : {'name' : 'Intel Corporation', 'currency' : '$'}, 'I' : {'name': 'IntelSat', 'currency' : '$'}} -class Title(object): - def __init__(self, symbol, params): - self.symbol = symbol - self.params = params - try: - self.name = params['name'] - except: - self.name = symbol - self.ax = None - self.fig = None - self.currency = params['currency'] - self.currency_corr = 1 - if '$' in self.currency: - self.currency_corr = k_euro - - self.data = {} - self.boll = {} - self.indicators = {} - - def fetch(self): - filename = self.symbol.replace('.', '_') + '.h5' - fetch = False - today = dt.date.today() - start = dt.datetime(today.year, 1, 1) - end = today - - try: - hdf = pd.HDFStore(filename, 'r') - hdf.close() - except: - fetch = True - - try: - lastmodified = dt.datetime.fromtimestamp(os.stat(filename).st_mtime).date() - if lastmodified != today: - fetch = fetch_on_outdated - except: - fetch = True - - if fetch: - # Get stock price via data reader - while(True): - try: - print("Fetching \"{}\"".format(self.symbol)) - data = web.DataReader(self.symbol, "av-daily", start, end, api_key=key) - break - except RemoteDataError: - for timeout in reversed(range(0, 60)): - print ("Try again in {} s".format(timeout)) - time.sleep(1) - - hdf = pd.HDFStore(filename) - hdf[self.symbol] = data - else: - hdf = pd.HDFStore(filename, 'r') - - data = hdf[self.symbol] * self.currency_corr - N = len(data.index) - self.data['index'] = np.array(data.index[0:N]) - self.data['close'] = np.array(data['close'][0:N]) - self.data['high'] = np.array(data['high'][0:N]) - self.data['low'] = np.array(data['low'][0:N]) - - hdf.close() - - def get_latest(self, key): - N = len(self.data['index']) - return self.data[key][N-1] - - def statistics(self): - N = len(self.data['index']) - self.data['ema'] = exponential_moving_average(self.data, key='close', alpha=ema_alpha) - self.data['sma'] = moving_average(self.data, key='close', window_days=sma_days) - self.data['close_n'] = normalize(self.data, key='close') - self.data['macd'] = macd(self.data, key='close_n') - self.data['min'] = moving_min(self.data, key='close_n', window_days=q_days) - self.data['max'] = moving_max(self.data, key='close_n', window_days=q_days) - - self.data['Qmin'] = self.data['close_n'] - self.data['min'] - self.data['Qmax'] = self.data['close_n'] - self.data['max'] - self.boll = bollinger(self.data, window_days=30) - - x_r = np.linspace(0, N, N) - y_r = self.data['macd'] - spl = UnivariateSpline(x_r, y_r) - spl.set_smoothing_factor(0.25) - spl_d = spl.derivative() - spl_dd = spl_d.derivative() - yf = spl(x_r) - yf_d = spl_d(x_r) - yf_dd = spl_dd(x_r) - - self.data['macd_f'] = np.transpose(yf) - self.data['macd_fd'] = np.transpose(yf_d) - self.data['macd_fdd'] = np.transpose(yf_dd) - - def analyze(self, buy_callback, range_days): - 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): - result = np.count_nonzero(data[key] != None) - return result > 0 - - @staticmethod - def slice(data, start, stop): - result = {} - for key in iter(data): - result[key] = data[key][start:stop] - - return result - - def __plot(self, data, keys): - N = len(data['index']) - start = max(0, N - show_range_days) - stop = N - xr = list(range(start, stop)) - sliced = Title.slice(data, start, stop) - - for key in keys: - plt.plot(xr, sliced[key], label=key) - - - def hover(event): - if event.inaxes == self.ax: - x_idx = min(N-1, int(event.xdata)) - self.ax.format_xdata = lambda x: self.data['index'][x_idx] - self.ax.format_ydata = lambda y: '{:.2f}%'.format(self.data['close_n'][x_idx]) - - self.fig.canvas.mpl_connect("motion_notify_event", hover) - - def __ind(self, data, keys): - N = len(data['index']) - start = max(0, N - show_range_days) - stop = N - xr = list(range(start, stop)) - sliced = Title.slice(data, start, stop) - - for key in keys: - plt.plot(xr, sliced[key], 'go', label=key) - - def show(self, indicators, figNum=1): - num_subplots = 4 - 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, ['BTFD']) - plt.title('{} ({})'.format(self.name, self.symbol)) - plt.legend() - plt.grid() - - plt.subplot(100*num_subplots + 10 + 2) -# self.__plot(self.boll, ['lower', 'mid', 'upper']) - self.__plot(self.data, ['Qmin', 'Qmax']) - plt.legend() - plt.grid() - - - plt.subplot(100*num_subplots + 10 + 3) - self.__plot(self.data, ['macd', 'macd_f']) - plt.legend() - plt.grid() - - plt.subplot(100*num_subplots + 10 + 4) - self.__plot(self.data, ['macd_fd', 'macd_fdd']) - plt.legend() - plt.grid() - - buy_list = {} def buy_callback(data): name = data['name'] @@ -314,19 +137,19 @@ figNum = 1 if len(show_symbols) > 0: for symbol in show_symbols: if symbol in symbols: - title = Title(symbol, symbols[symbol]) + title = Stock(symbol, symbols[symbol]) title.fetch() title.statistics() - ind = title.analyze(buy_callback, range_days=show_range_days) + ind = title.analyze(buy_callback, range_days=params['show_range_days']) title.show(ind, figNum=figNum) figNum += 1 else: for symbol in symbols: - title = Title(symbol, symbols[symbol]) + title = Stock(params, symbol, symbols[symbol]) title.fetch() title.statistics() - ind = title.analyze(buy_callback, range_days=show_range_days) - if Title.has_candidate(ind, 'BTFD'): + ind = title.analyze(buy_callback, range_days=params['show_range_days']) + if Stock.has_candidate(ind, 'BTFD'): print('-----------------------------------------------') title.show(ind, figNum=figNum) figNum += 1 @@ -335,7 +158,7 @@ gain_accum = 0 num_stocks = 0 for symbol in buy_list: items = buy_list[symbol]['items'] - title = Title(symbol, symbols[symbol]) + title = Stock(params, symbol, symbols[symbol]) title.fetch() title.statistics() gain = title.get_latest('close_n') - items[0]['value'] diff --git a/stock.py b/stock.py new file mode 100644 index 0000000..825aed7 --- /dev/null +++ b/stock.py @@ -0,0 +1,182 @@ +import pandas as pd +import pandas_datareader.data as web +from pandas_datareader._utils import RemoteDataError +from scipy.interpolate import UnivariateSpline +import datetime as dt +import os +import time +import agent +from functions import * +import matplotlib.pyplot as plt +import matplotlib as mpl + +mpl.rc('figure', max_open_warning = 0) + +class Stock(object): + def __init__(self, params2, symbol, params): + self.params = params + self.symbol = symbol + self.params2 = params2 + try: + self.name = params['name'] + except: + self.name = symbol + self.ax = None + self.fig = None + self.currency = params['currency'] + self.currency_corr = 1 + if '$' in self.currency: + self.currency_corr = self.params2['k_euro'] + + self.data = {} + self.boll = {} + self.indicators = {} + + def fetch(self): + filename = self.symbol.replace('.', '_') + '.h5' + fetch = False + today = dt.date.today() + start = dt.datetime(today.year, 1, 1) + end = today + + try: + hdf = pd.HDFStore(filename, 'r') + hdf.close() + except: + fetch = True + + try: + lastmodified = dt.datetime.fromtimestamp(os.stat(filename).st_mtime).date() + if lastmodified != today: + fetch = fetch_on_outdated + except: + fetch = True + + if fetch: + # Get stock price via data reader + while(True): + try: + print("Fetching \"{}\"".format(self.symbol)) + data = web.DataReader(self.symbol, "av-daily", start, end, api_key=key) + break + except RemoteDataError: + for timeout in reversed(range(0, 60)): + print ("Try again in {} s".format(timeout)) + time.sleep(1) + + hdf = pd.HDFStore(filename) + hdf[self.symbol] = data + else: + hdf = pd.HDFStore(filename, 'r') + + data = hdf[self.symbol] * self.currency_corr + N = len(data.index) + self.data['index'] = np.array(data.index[0:N]) + self.data['close'] = np.array(data['close'][0:N]) + self.data['high'] = np.array(data['high'][0:N]) + self.data['low'] = np.array(data['low'][0:N]) + + hdf.close() + + def get_latest(self, key): + N = len(self.data['index']) + return self.data[key][N-1] + + def statistics(self): + N = len(self.data['index']) + self.data['ema'] = exponential_moving_average(self.data, key='close', alpha=self.params2['ema_alpha']) + self.data['sma'] = moving_average(self.data, key='close', window_days=self.params2['sma_days']) + self.data['close_n'] = normalize(self.data, key='close') + self.data['macd'] = macd(self.data, key='close_n') + self.data['min'] = moving_min(self.data, key='close_n', window_days=self.params2['q_days']) + self.data['max'] = moving_max(self.data, key='close_n', window_days=self.params2['q_days']) + + self.data['Qmin'] = self.data['close_n'] - self.data['min'] + self.data['Qmax'] = self.data['close_n'] - self.data['max'] + self.boll = bollinger(self.data, window_days=30) + + x_r = np.linspace(0, N, N) + y_r = self.data['macd'] + spl = UnivariateSpline(x_r, y_r) + spl.set_smoothing_factor(0.25) + spl_d = spl.derivative() + spl_dd = spl_d.derivative() + yf = spl(x_r) + yf_d = spl_d(x_r) + yf_dd = spl_dd(x_r) + + self.data['macd_f'] = np.transpose(yf) + self.data['macd_fd'] = np.transpose(yf_d) + self.data['macd_fdd'] = np.transpose(yf_dd) + + def analyze(self, buy_callback, range_days): + 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): + result = np.count_nonzero(data[key] != None) + return result > 0 + + @staticmethod + def slice(data, start, stop): + result = {} + for key in iter(data): + result[key] = data[key][start:stop] + + return result + + def __plot(self, data, keys): + N = len(data['index']) + start = max(0, N - self.params2['show_range_days']) + stop = N + xr = list(range(start, stop)) + sliced = Stock.slice(data, start, stop) + + for key in keys: + plt.plot(xr, sliced[key], label=key) + + + def hover(event): + if event.inaxes == self.ax: + x_idx = min(N-1, int(event.xdata)) + self.ax.format_xdata = lambda x: self.data['index'][x_idx] + self.ax.format_ydata = lambda y: '{:.2f}%'.format(self.data['close_n'][x_idx]) + + self.fig.canvas.mpl_connect("motion_notify_event", hover) + + def __ind(self, data, keys): + N = len(data['index']) + start = max(0, N - self.params2['show_range_days']) + stop = N + xr = list(range(start, stop)) + sliced = Stock.slice(data, start, stop) + + for key in keys: + plt.plot(xr, sliced[key], 'go', label=key) + + def show(self, indicators, figNum=1): + num_subplots = 4 + 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, ['BTFD']) + plt.title('{} ({})'.format(self.name, self.symbol)) + plt.legend() + plt.grid() + + plt.subplot(100*num_subplots + 10 + 2) +# self.__plot(self.boll, ['lower', 'mid', 'upper']) + self.__plot(self.data, ['Qmin', 'Qmax']) + plt.legend() + plt.grid() + + + plt.subplot(100*num_subplots + 10 + 3) + self.__plot(self.data, ['macd', 'macd_f']) + plt.legend() + plt.grid() + + plt.subplot(100*num_subplots + 10 + 4) + self.__plot(self.data, ['macd_fd', 'macd_fdd']) + plt.legend() + plt.grid()