import pandas as pd import pandas_datareader.data as web import numpy as np import math import matplotlib.pyplot as plt import datetime as dt import os # Stock Investors Financial Math # https://www.fmlabs.com/reference/default.htm?url=SimpleMA.htm # Alpha Vantage # https://www.alphavantage.co/documentation/ # Pandas # https://pandas.pydata.org/pandas-docs/stable/index.html # Knowledge # https://www.investopedia.com/articles/technical/02/050602.asp # https://ntguardian.wordpress.com/2016/09/19/introduction-stock-market-data-python-1 key = '0UO7Z2MVZ2YSQSVE' thresh_macd = 1.5 show_range_days = 40 show_title = ['QCOM', 'VAR1.DE', 'MOR.DE', 'QIA.DE', 'MDG1.DE'] k_euro = 1 / 1.11 N_poly = 7 ema_alpha = 0.75 sma_days = 10 titles = [ 'AAPL', 'XLNX', 'QCOM', 'DPW.DE', 'CSCO', 'AIR', 'BA', 'NVDA', 'MSFT', 'DIS', 'NFLX', 'OHB.DE', 'ERCA.DE', 'VAR1.DE', 'HD', 'AMZN', 'GOOGL', 'FB2A.DE', 'ZIL2.DE', 'SIS.DE', 'SIE.DE', 'GFT.DE', 'AMD.DE', 'CAP.DE', 'ADBE', 'PANW', 'AMAT', 'RIB.DE', 'WAF.DE', 'EVT.DE', 'VOW.DE', 'BMW.DE', 'NSU.DE', 'DAI.DE', 'SHA.DE', 'CON.DE', 'DHER.DE', 'BSL.DE', 'D6H.DE', 'TC1.DE', 'VODI.DE', 'ATVI', 'GME', 'NTO.F', 'UBSFF', 'NXPRF', 'IMPUF', 'NMPNF', 'ALSMY', 'ARRD.F', 'PHG', 'KEYS', 'ORCL', 'UBER', 'VMW', 'AVGO', 'CY', 'QABSY', 'BLDP', 'D7G.F', 'ITMPF', 'PLUG', 'PCELF', 'SU.PA', # 'SON1.DE', 'STM.DE', 'BC8.DE', 'MOR.DE', 'BAYN.DE', 'BEI.DE', 'EUZ.DE', 'SLAB', 'SPLK', 'IAG', 'INTC', 'I'] def ema(data, key, alpha=0.5, ic=None): result = [] if ic is None: r = data[key][0] else: r = ic for v in data[key]: r = alpha*r + (1-alpha)*v result.append(r) return np.array(result) def sma(data, key, window_days, ic=None): mem = [0] * window_days result = [] k=0 cumsum = 0 for v in data[key]: cumsum += (v - mem[k]) mem[k] = v k += 1 if k >= window_days: k=0 result.append(cumsum/window_days) return result def sd(data, key, window_days, ic=None): mem = [0] * window_days result = [] k=0 cumsum = 0 data_mean = data[key] - sma(data, key, window_days, ic) for v in data_mean: v2 = v * v cumsum += (v2 - mem[k]) mem[k] = v2 k += 1 if k >= window_days: k=0 var = max(0, cumsum) / window_days result.append(math.sqrt(var)) return result def colsum(data, keys): result = np.zeros(data.shape[0]) for key in keys: result += data[key] return result def macd(data, key): ema_short = ema(data, key, alpha=0.85) ema_long = ema(data, key, alpha=0.925) return ema_short - ema_long def bollinger(data, window_days, f=2): tp_ser = colsum(data, keys=['high', 'low', 'close']) / 3 tp_df = tp_ser.to_frame(name='tp') stddev = np.array(sd(tp_df, 'tp', window_days)) mid = np.array(sma(tp_df, key='tp', window_days=window_days)) upper = mid + f * stddev lower = mid - f * stddev result = pd.DataFrame(upper, index=data.index, columns=['upper']) result['lower'] = lower result['mid'] = mid return result class Symbol(object): def __init__(self, name): self.name = name self.ax = None self.fig = None def fetch(self): filename = self.name.replace('.','_') + '.h5' fetch = False today = dt.date.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 = True except: fetch = True if fetch: print ("Fetching \"{}\"".format(self.name)) # Get stock price via data reader start = dt.datetime(today.year,1,1) end = dt.date.today() data = web.DataReader(self.name, "av-daily", start, end, api_key=key) hdf = pd.HDFStore(filename) hdf[self.name] = data else: hdf = pd.HDFStore(filename, 'r') self.data_full = hdf[self.name]*k_euro hdf.close() def __plot(self, d, keys): dates = d.index for key in keys: data = d[key] data.plot() def hover(event): if event.inaxes == self.ax: x_idx = int(event.xdata) self.ax.format_xdata = lambda x: dates[x_idx] self.ax.format_ydata = lambda y: '€{:.2f}'.format(data[x_idx]) self.fig.canvas.mpl_connect("motion_notify_event", hover) def analyze(self): self.data_full['ema'] = ema(self.data_full, key='close', alpha=ema_alpha) self.data_full['macd'] = macd(self.data_full, key='close') self.data_full['sma'] = sma(self.data_full, key='close', window_days=sma_days) self.boll_full = bollinger(self.data_full, window_days=30) _macd_curr = self.data_full['macd'][len(self.data_full)-1] self._macd_data = self.data_full['macd'][len(self.data_full) - show_range_days:] x_r = list(range(0, len(self._macd_data)+1)) y_r = self._macd_data.to_numpy() y_r = np.append(y_r, y_r[len(y_r)-1]) poly = np.polyfit(x_r, y_r, N_poly) poly_d = np.polyder(poly) yf = np.polyval(poly, x_r[:len(x_r)-1]) yf_d = np.polyval(poly_d, x_r[:len(x_r)-1]) self._macd_fitted = pd.DataFrame(np.transpose([yf,yf_d]), index=self._macd_data.index, columns=['macd_fitted', 'macd_fitted_d']) _macd_f_curr = self._macd_fitted['macd_fitted'][len(self._macd_fitted)-1] _macd_fd_curr = self._macd_fitted['macd_fitted_d'][len(self._macd_fitted)-1] return {'macd' : _macd_curr, 'macd_f' : _macd_f_curr, 'macd_fd' : _macd_fd_curr} def show(self, figNum=1): data = self.data_full[len(self.data_full) - show_range_days:] boll = self.boll_full[len(self.boll_full) - show_range_days:] self.fig = plt.figure(figNum) self.ax = plt.subplot(311) self.__plot(data, ['ema', 'sma', 'close']) plt.title(self.name) plt.legend() plt.grid() plt.subplot(312) boll['upper'].plot() boll['mid'].plot() boll['lower'].plot() plt.legend() plt.grid() plt.subplot(313) self._macd_data.plot() self._macd_fitted['macd_fitted'].plot() self._macd_fitted['macd_fitted_d'].plot() plt.legend() plt.grid() figNum = 1 if len(show_title) > 0: for title in show_title: sym = Symbol(title) sym.fetch() print(sym.analyze()) sym.show(figNum) figNum += 1 else: for title in titles: sym = Symbol(title) sym.fetch() result = sym.analyze() if result['macd_f'] > thresh_macd and result['macd_fd'] > 0: sym.show(figNum) figNum += 1 plt.show()