Python talib 模块，STOCHF 实例源码

def KDJ_CN(high, low, close, fastk_period=9, slowk_period=3, fastd_period=3) :
        len1 = len(high)
        len2 = len(low)
        len3 = len(close)
        if len1 != len2 or len1 != len3:
            print ("KDJ_CN input invalid for len:%s %s %s " %(str(len1),str(len2),str(len3)))
            return np.array(np.nan),np.array(np.nan),np.array(np.nan)
        kValue, dValue = tl.STOCHF(high, low, close, fastk_period, fastd_period=fastd_period, fastd_matype=0)

        kValue = np.array(map(lambda x : SecurityDataSrcBase.SMA_CN(kValue[:x], slowk_period), range(1, len(kValue) + 1)))
        dValue = np.array(map(lambda x : SecurityDataSrcBase.SMA_CN(kValue[:x], fastd_period), range(1, len(kValue) + 1)))

        jValue = 3 * kValue - 2 * dValue

        func = lambda arr : np.array([0 if x < 0 else (100 if x > 100 else x) for x in arr])

        kValue = func(kValue)
        dValue = func(dValue)
        jValue = func(jValue)
        return kValue, dValue, jValue

def STOCH(df, n = 14, slowk_period = 3, slowd_period = 3):
    fastk, fastd = talib.STOCHF(df['high'].values, df['low'].values, df['close'].values, fastk_period = n, fastd_period=slowk_period)
    slowk, slowd = talib.STOCH(df['high'].values, df['low'].values, df['close'].values, fastk_period = n, slowk_period=slowk_period, slowd_period=slowd_period)
    fk = pd.Series(fastk, index = df.index, name = "STOCHFK_%s_%s_%s" % (str(n), str(slowk_period), str(slowd_period)))
    sk = pd.Series(slowk, index = df.index, name = "STOCHSK_%s_%s_%s" % (str(n), str(slowk_period), str(slowd_period)))
    sd = pd.Series(slowd, index = df.index, name = "STOCHSD_%s_%s_%s" % (str(n), str(slowk_period), str(slowd_period)))
    return pd.concat([fk, sk, sd], join='outer', axis=1)

def STOCHF(df, n = 14, fastd_period = 3):
    fastk, fastd = talib.STOCHF(df['high'].values, df['low'].values, df['close'].values, fastk_period = n, fastd_period=fastd_period)
    fk = pd.Series(fastk, index = df.index, name = "STOCFK_%s_%s" % (str(n), str(fastd_period)))
    sk = pd.Series(fastd, index = df.index, name = "STOCSK_%s_%s" % (str(n), str(fastd_period)))
    return pd.concat([fk, sk], join='outer', axis=1)

def add_STOCHF(self, fastk_period=5, fastd_period=3, fastd_matype=0,
               types=['line', 'line'],
               colors=['primary', 'tertiary'],
               **kwargs):
    """Fast Stochastic Oscillator.

    Note that the first argument of types and colors refers to Fast Stoch %K,
    while second argument refers to Fast Stoch %D
    (signal line of %K obtained by MA).

    """
    if not (self.has_high and self.has_low and self.has_close):
        raise Exception()

    utils.kwargs_check(kwargs, VALID_TA_KWARGS)
    if 'kind' in kwargs:
        kwargs['type'] = kwargs['kind']
    if 'kinds' in kwargs:
        types = kwargs['type']

    if 'type' in kwargs:
        types = [kwargs['type']] * 2
    if 'color' in kwargs:
        colors = [kwargs['color']] * 2

    name = 'STOCHF({},{})'.format(str(fastk_period),
                                  str(fastd_period))
    fastk = name + r'[%k]'
    fastd = name + r'[%d]'
    self.sec[fastk] = dict(type=types[0], color=colors[0])
    self.sec[fastd] = dict(type=types[1], color=colors[1], on=fastk)
    self.ind[fastk], self.ind[fastd] = talib.STOCHF(self.df[self.hi].values,
                                                    self.df[self.lo].values,
                                                    self.df[self.cl].values,
                                                    fastk_period, fastd_period,
                                                    fastd_matype)

def KDJ(security_list, fastk_period=5, slowk_period=3, fastd_period=3):
    def SMA_CN(close, timeperiod):
        close = np.nan_to_num(close)
        return reduce(lambda x, y: ((timeperiod - 1) * x + y) / timeperiod, close)

    # ????????????
    if isinstance(security_list, str):
        security_list = [security_list]
    # ?? KDJ
    n = max(fastk_period, slowk_period, fastd_period)
    k = {}
    d = {}
    j = {}
    for stock in security_list:

        security_data = attribute_history(
            stock, n * 2, '1d', fields=['high', 'low', 'close'], df=False)
        high = security_data['high']
        low = security_data['low']
        close = security_data['close']
        kValue, dValue = talib.STOCHF(
            high, low, close, fastk_period, fastd_period, fastd_matype=0)
        kValue = np.array(map(lambda x: SMA_CN(
            kValue[:x], slowk_period), range(1, len(kValue) + 1)))
        dValue = np.array(map(lambda x: SMA_CN(
            kValue[:x], fastd_period), range(1, len(kValue) + 1)))
        jValue = 3 * kValue - 2 * dValue

        def func(arr): return np.array(
            [0 if x < 0 else (100 if x > 100 else x) for x in arr])

        k[stock] = func(kValue)
        d[stock] = func(dValue)
        j[stock] = func(jValue)
    return k, d, j

# RSI

def test_fso_expected_with_talib(self, seed):
        """
        Test the output that is returned from the fast stochastic oscillator
        is the same as that from the ta-lib STOCHF function.
        """
        window_length = 14
        nassets = 6
        rng = np.random.RandomState(seed=seed)

        input_size = (window_length, nassets)

        # values from 9 to 12
        closes = 9.0 + (rng.random_sample(input_size) * 3.0)

        # Values from 13 to 15
        highs = 13.0 + (rng.random_sample(input_size) * 2.0)

        # Values from 6 to 8.
        lows = 6.0 + (rng.random_sample(input_size) * 2.0)

        expected_out_k = []
        for i in range(nassets):
            fastk, fastd = talib.STOCHF(
                high=highs[:, i],
                low=lows[:, i],
                close=closes[:, i],
                fastk_period=window_length,
                fastd_period=1,
            )

            expected_out_k.append(fastk[-1])
        expected_out_k = np.array(expected_out_k)

        today = pd.Timestamp('2015')
        out = np.empty(shape=(nassets,), dtype=np.float)
        assets = np.arange(nassets, dtype=np.float)

        fso = FastStochasticOscillator()
        fso.compute(
            today, assets, out, closes, lows, highs
        )

        assert_equal(out, expected_out_k, array_decimal=6)