StockChartingDash/subroutines/security.py

import numpy as np
from numpy.fft import fft, ifft
import scipy.signal as sig

class Security:
    """
    This can be a list of stocks, bonds, or otherinvestment vehicles.
    price - Pandas DataFrame with datetime as index sorted to chronical order
    """
    def __init__(self, sym, price, volume=None, rfr: float = 0.01, sf: float = 252.0):
        """
        Parameters
        ----------
        price : TYPE pandas.DataFrame
            DESCRIPTION. historical adj. daily close prices of stocks under
            consideration
        volume : TYPE pandas.DataFrame
            DESCRIPTION. daily trading volume. The default is none.
        rfr : TYPE float, optional
            DESCRIPTION. annualized risk free rate. The default is 0.01.
        sf : TYPE sample frequency, optional
            DESCRIPTION. The default is 252 (daily). there are 252 trading
            days in a year. Monthly sampling frequency would be 12. And
            weekly sampling frequenc is 52.
        """
        self._symbol = sym
        self._price = price
        self._volume = volume
        # self._symbol = price.columns.values
        self._rfr = rfr
        self._sf = sf

    @property
    def symbol(self):
        return self._symbol
    @symbol.setter
    def symbol(self, value):
        raise AttributeError('security symbol is read only')

    @property
    def price(self):
        return self._price

    @price.setter
    def price(self, value):
        raise AttributeError('security price is read only')

    @property
    def volume(self):
        if self._volume is None:
            raise ValueError('trading volume information not available')
        return self._volume

    @volume.setter
    def volume(self, value):
        raise AttributeError('security volume is read only')

    def sma(self, window):
        return self.price.rolling(window).mean()

    def vwma(self, window):
        """
        Volume weighted moving average. When plotted against sma, it gives an
        early indicator when VWMA crosses SMA. When VWMA is above SMA, it
        indicates a strong upward trend and vice versa.
        """
        price_vol = self.price * self.volume
        return price_vol.rolling(window).sum() / self.volume.rolling(window).sum()

    def vosma(self, window):
        return self.volume.rolling(window).mean()

    def ema(self, window): # default to 14 day window
        # EMA pre-process the first point
        price = self.price
        temp = price.iloc[0:window].mean()
        price.iloc[window-1] = temp
        price.iloc[0:(window-1)] = np.nan

        # process the EMA
        avg = price.ewm(span=window, adjust=False).mean()
        return avg

    def voema(self, window): # default to 14 day window
        # EMA pre-process the first point
        vol = self.volume
        temp = vol.iloc[0:window].mean()
        vol.iloc[window-1] = temp
        vol.iloc[0:(window-1)] = np.nan

        # process the EMA
        avg = vol.ewm(span=window, adjust=False).mean()
        return avg

    def rsi(self, window = 14):
        """
        Traditional interpretation and usage of the RSI are that values of 70
        or above indicate that a security is becoming overbought or overvalued
        and may be primed for a trend reversal or corrective pullback in price.
        An RSI reading of 30 or below indicates an oversold or undervalued
        condition.
        """
        # use exponential averaging
        d_chg = self.price.diff()
        d_up, d_dn = d_chg.copy(), d_chg.copy()
        d_up[d_up < 0] = 0
        d_dn[d_dn > 0] = 0

        # EMA pre-process the first point
        temp = d_up.iloc[1:(window+1)].mean()
        d_up.iloc[window] = temp
        d_up.iloc[1:window] = np.nan
        temp = d_dn.iloc[1:(window+1)].mean()
        d_dn.iloc[window] = temp
        d_dn.iloc[1:window] = np.nan

        # process the EMA
        avg_up = d_up.ewm(span=window, adjust=False).mean()
        avg_dn = d_dn.ewm(span=window, adjust=False).mean()
        rs = avg_up / abs(avg_dn.values)
        exp_rsi = 100 - 100 / (1+rs)
        return exp_rsi


    def volume_rsi(self, window = 14):
        """
        The volume RSI (Relative Strength Index) is quite similar to the price
        based RSI with difference that up-volume and down-volume are used in
        the RSI formula instead changes in price. If price RSI shows relation
        between up-moves and down-moves within an analyzed period of time by
        revealing which moves are stronger, the volume RSI indicator shows the
        relation between volume traded during these price up-moves and
        down-moves respectfully by revealing whether up-volume (bullish money
        flow) or down-volume (bearish money flow) is stronger.

        The same as price RSI, volume RSI oscillates around 50% center-line in
        the range from 0 to 100%. In technical analysis this indicator could be
        used in the same way as well. The simplest way of using the volume RSI
        would be to generate trading signals on the crossovers of the indicator
        and 50% center-line around which it oscillates. Here you have to
        remember following:

        volume RSI reading above 50% are considered bullish as bullish volume
        dominates over bearish volume; volume RSI readings below 50% are
        considered bearish as bearish volume overcomes bullish volume.
        Respectfully, technical analysis would suggest to generate buy/sell
        signals by following rules:

        Buy when indicators moves above 50% line after being below it;
        Sell when indicator drops below 50% line after being above it.
        """
        # use exponential averaging
        volume = self.volume

        up_vol, dn_vol = volume.copy(), volume.copy()
        d_chg = self.price.diff()

        up_vol[d_chg < 0] = 0
        dn_vol[d_chg > 0] = 0
        up_vol.iloc[0] = np.nan
        dn_vol.iloc[0] = np.nan

        # EMA pre-process the first point
        temp = up_vol.iloc[1:(window+1)].mean()
        up_vol.iloc[window] = temp
        up_vol.iloc[1:window] = np.nan
        temp = dn_vol.iloc[1:(window+1)].mean()
        dn_vol.iloc[window] = temp
        dn_vol.iloc[1:window] = np.nan

        # EMA processing
        avg_up = up_vol.ewm(span=window, adjust=False).mean()
        avg_dn = dn_vol.ewm(span=window, adjust=False).mean()
        rs = avg_up / avg_dn.values
        exp_rsi = 100 - 100 / (1+rs)
        return exp_rsi

    def daily_returns(self):
        return self.price.pct_change()

    @property
    def annualized_return(self):
        dr = self.daily_returns()
        return self._sf * dr.mean()

    @property
    def annualized_stdev(self):
        dr = self.daily_returns()
        return np.sqrt(self._sf) * dr.std()

    @property
    def sharpe(self):
        return (self.annualized_return - self._rfr) / self.annualize_stdev

    def rolling_stdev(self, window):
        return self.price.rolling(window).std()

    def bollinger(self, window):
        """
        Parameters
        ----------
        window : TYPE int, optional
            DESCRIPTION - averaging window in days.

        Returns
        -------
        lower, upper : TYPE pandas.DataFrame
            DESCRIPTION - lower band (minus 2 sigma) and the upper band.
        """
        avg = self.sma(window)
        sdd2 = self.rolling_stdev(window).mul(2)
        lower = avg.sub(sdd2.values)
        upper = avg.add(sdd2.values)
        # low_up = lower.join(upper, lsuffix='_L', rsuffix='_U')

        return lower, upper

    def macd(self, short_wd = 12, long_wd = 26, sig_wd = 9):
        """
        MACD Line: (12-day EMA - 26-day EMA)
        Signal Line: 9-day EMA of MACD Line
        MACD Histogram: MACD Line - Signal Line

        MACD is calculated by subtracting the 26-period EMA from the 12-period
        EMA. MACD triggers technical signals when it crosses above (to buy) or
        below (to sell) its signal line. The speed of crossovers is also taken
        as a signal of a market is overbought or oversold. MACD helps investors
        understand whether the bullish or bearish movement in the price is
        strengthening or weakening

        MACD historgram represents signal line crossovers that are the most
        common MACD signals. The signal line is a 9-day EMA of the MACD line.
        As a moving average of the indicator, it trails the MACD and makes it
        easier to spot MACD turns. A bullish crossover occurs when the MACD
        turns up and crosses above the signal line. A bearish crossover occurs
        when the MACD turns down and crosses below the signal line. Crossovers
        can last a few days or a few weeks, depending on the strength of the
        move.
        """
        macd_short = self.ema(short_wd)
        macd_long = self.ema(long_wd)
        macd_line = macd_short - macd_long.values
        macd_sig = macd_line.ewm(span=sig_wd, adjust=False).mean()
        macd_hist = macd_line - macd_sig.values
        norm_hist = macd_hist.div(macd_long.values)
        return macd_line, macd_sig, macd_hist, norm_hist

def get_crossing(stocks):
    """
    Parameters
    ----------
    stocks : TYPE instance of class 'security'

    Returns
    -------
    cross : TYPE pandas DataFrame
        DESCRIPTION - +1 when 50 day moving average is above 200 day moving
        average. -1 when vice versa. transition days are of value +3 and -3
        respectively.
    """
    sma50 = stocks.sma(50)
    sma200 = stocks.sma(200)
    cross = np.sign(sma50.sub(sma200.values))
    cross_diff = cross.diff()
    cross = cross.add(cross_diff.values)
    cross.columns = stocks.price.columns

    return cross

def get_sma_slope(stocks, wd = 50):
    """
    Parameters
    ----------
    stocks : TYPE
        DESCRIPTION.
    wd : TYPE, optional
        DESCRIPTION. The default is 50.

    Returns
    -------
    slope : TYPE pandas DataFrame
        DESCRIPTION - +1 when n day moving average is positive. -1 when
        negative. transition days are of value +3 and -3 respectively.
    """
    sma = stocks.sma(wd)
    slope = np.sign(sma.diff())
    slope_diff = slope.diff()
    slope = slope.add(slope_diff.values)

    return slope

def fill_missing_data(df):
    df.ffill(inplace=True)
    df.bfilln(inplace=True)

def fft_convolve(signal, window):
    fft_signal = fft(signal)
    fft_window = fft(window)
    return ifft(fft_signal * fft_window)

def zero_pad(array, n):
    """Extends an array with zeros.

    array: numpy array
    n: length of result

    returns: new NumPy array
    """
    res = np.zeros(n)
    res[: len(array)] = array
    return res

def smooth(price, hsize=10, sigma=3):
    """
    Parameters
    ----------
    price : TYPE DataFrame.
        DESCRIPTION - with time index and no invalid values
    hsize : TYPE integer
        DESCRIPTION - this adds phase delay. similar to SMA window
    sigma : TYPE float
        DESCRIPTION - gaussian standard deviation affects smoothness

    Returns
    -------
    TYPE DataFrame
        DESCRIPTION - smoothed price
    Doesn't offer much benefit over sma. Only theoretical values. For future
    different smooth functiona experiments
    """
    data = price.copy()
    window = sig.gaussian(M=hsize, std=sigma)
    window /= window.sum()
    padded = zero_pad(window, data.shape[0])
    for col in data.columns:
        ys = data[col].values
        smoo = abs(fft_convolve(ys, padded))
        smoo[0:hsize-1] = np.nan
        data[col] = smoo

    return data