import pandas as pd import pandas_datareader.data as web import numpy as np import math import matplotlib.pyplot as plt from scipy.interpolate import UnivariateSpline 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 = 20 show_symbols = ['ITMPF', 'PLUG', 'MOR.DE', 'CSCO', 'ERCA.DE', 'AVGO', 'DIS'] #show_symbols = [] k_euro = 1 / 1.11 N_poly = 7 ema_alpha = 0.75 sma_days = 10 symbols = { 'WDI.DE' : {'name' : 'WireCard', 'currency' : '€'}, 'AAPL' : {'name' : 'Apple', 'currency' : '$'}, 'XLNX' : {'name' : 'Xilinx', 'currency' : '$'}, 'QCOM' : {'name' : 'Qualcomm', 'currency' : '$'}, 'DPW.DE' : {'name' : 'Deutsche Post', 'currency' : '€'}, 'CSCO' : {'name' : 'Cisco', 'currency' : '$'}, 'AIR' : {'name' : 'Airbus', 'currency' : '$'}, 'BA' : {'name' : 'Boeing', 'currency' : '$'}, 'NVDA' : {'name' : 'Nvidia', 'currency' : '$'}, 'MSFT' : {'name' : 'Microsoft', 'currency' : '$'}, 'DIS' : {'name' : 'Disney', 'currency' : '$'}, 'NFLX' : {'name' : 'Netflix', 'currency' : '$'}, 'OHB.DE' : {'name' : 'OHB', 'currency' : '€'}, 'ERCA.DE' : {'name' : 'Ericsson', 'currency' : '€'}, 'VAR1.DE' : {'name' : 'Varta', 'currency' : '€'}, 'HD' : {'name' : 'Home Depot', 'currency' : '$'}, 'AMZN' : {'name' : 'Amazon', 'currency' : '$'}, 'GOOGL' : {'name' : 'Google', 'currency' : '$'}, 'FB2A.DE' : {'currency' : '€'}, 'ZIL2.DE' : {'currency' : '€'}, 'SIS.DE' : {'currency' : '€'}, 'SIE.DE' : {'currency' : '€'}, 'GFT.DE' : {'currency' : '€'}, 'AMD.DE' : {'currency' : '€'}, 'CAP.DE' : {'currency' : '€'}, 'ADBE' : {'currency' : '$'}, 'PANW' : {'currency' : '$'}, 'AMAT' : {'currency' : '$'}, 'RIB.DE' : {'currency' : '€'}, 'WAF.DE' : {'currency' : '€'}, 'EVT.DE' : {'currency' : '€'}, 'VOW.DE' : {'currency' : '€'}, 'BMW.DE' : {'currency' : '€'}, 'NSU.DE' : {'currency' : '€'}, 'DAI.DE' : {'currency' : '€'}, 'SHA.DE' : {'currency' : '€'}, 'CON.DE' : {'currency' : '€'}, 'DHER.DE' : {'name' : 'Delivery Hero', 'currency' : '€'}, 'BSL.DE' : {'currency' : '€'}, 'D6H.DE' : {'currency' : '€'}, 'TC1.DE' : {'currency' : '€'}, 'VODI.DE' : {'currency' : '€'}, 'ATVI' : {'currency' : '$'}, 'GME' : {'currency' : '$'}, 'NTO.F' : {'currency' : '$'}, 'UBSFF' : {'currency' : '$'}, 'NXPRF' : {'currency' : '$'}, 'IMPUF' : {'currency' : '$'}, 'NMPNF' : {'currency' : '$'}, 'ALSMY' : {'currency' : '$'}, 'ARRD.F' : {'currency' : '$'}, 'PHG' : {'currency' : '$'}, 'KEYS' : {'currency' : '$'}, 'ORCL' : {'currency' : '$'}, 'UBER' : {'currency' : '$'}, 'VMW' : {'currency' : '$'}, 'AVGO' : {'name' : 'Avago', 'currency' : '$'}, 'CY' : {'currency' : '$'}, 'QABSY' : {'currency' : '$'}, 'BLDP' : {'currency' : '$'}, 'D7G.F' : {'currency' : '$'}, 'ITMPF' : {'currency' : '$'}, 'PLUG' : {'currency' : '$'}, 'PCELF' : {'currency' : '$'}, 'SU.PA' : {'currency' : '$'}, # 'SON1.DE' : {'currency' : '€'}, 'STM.DE' : {'currency' : '€'}, 'BC8.DE' : {'currency' : '€'}, 'MOR.DE' : {'name' : 'Morphosys', 'currency' : '€'}, 'BAYN.DE' : {'currency' : '€'}, 'BEI.DE' : {'currency' : '€'}, 'EUZ.DE' : {'currency' : '€'}, 'SLAB' : {'currency' : '$'}, 'SPLK' : {'name' : 'Splunk', 'currency' : '$'}, 'IAG' : {'name' : 'IAG', 'currency' : '$'}, 'INTC' : {'currency' : '$'}, 'I' : {'currency' : '$'}} 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 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 def fetch(self): filename = self.symbol.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.symbol)) # Get stock price via data reader start = dt.datetime(today.year,1,1) end = dt.date.today() data = web.DataReader(self.symbol, "av-daily", start, end, api_key=key) hdf = pd.HDFStore(filename) hdf[self.symbol] = data else: hdf = pd.HDFStore(filename, 'r') self.data_full = hdf[self.symbol] * self.currency_corr 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 = np.linspace(0, len(self._macd_data), len(self._macd_data)) y_r = self._macd_data.to_numpy() spl = UnivariateSpline(x_r, y_r) spl.set_smoothing_factor(0.5) yf = spl(x_r) yf_d = spl.derivative()(x_r) 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_symbols) > 0: for symbol in show_symbols: if symbol in symbols: title = Title(symbol, symbols[symbol]) title.fetch() print(title.analyze()) title.show(figNum) figNum += 1 else: for symbol in symbols: title = Title(symbol, symbols[symbol]) title.fetch() result = title.analyze() if result['macd_f'] > thresh_macd and result['macd_fd'] > 0: title.show(figNum) figNum += 1 plt.show()