Hi,
Using indicator in indicator is quite normal. I tried to build such a customer "indicator in indicator" built-in-factor but fialt. I got error as I run it:
"
Runtime exception: AttributeError: type object 'MyTalibRSI' has no attribute 'rsi'
"
Below is my code
"""
This is a template algorithm on Quantopian for you to adapt and fill in.
"""
from quantopian.algorithm import attach_pipeline, pipeline_output
from quantopian.pipeline import Pipeline
from quantopian.pipeline.data.builtin import USEquityPricing
from quantopian.pipeline.factors import AverageDollarVolume
from quantopian.pipeline.filters.morningstar import Q1500US
from quantopian.pipeline.factors import CustomFactor
from collections import defaultdict
from quantopian.pipeline.filters import StaticAssets
import numpy as np
import pandas as pd
import math
import talib
class MyTalibRSI(CustomFactor):
inputs = (USEquityPricing.close,)
params = {'rsi_len' : 14,}
window_length = 60 # allow at least 4 times RSI len for proper smoothing
def compute(self, today, assets, out, close, rsi_len):
# TODO: figure out how this can be coded without a loop
rsi = []
for col in close.T:
try:
rsi.append(talib.RSI(col, timeperiod=rsi_len)[-1])
except:
rsi.append(np.nan)
out[:] = rsi
class MyDoubleRSI(CustomFactor):
inputs = (MyTalibRSI.rsi,)
outputs = ['rsi']
params = {'rsi_len' : 14,}
window_length = 60 # allow at least 4 times RSI len for proper smoothing
def compute(self, today, assets, out, rsi, rsi_len):
# TODO: figure out how this can be coded without a loop
rsi = []
for col in rsi.T:
try:
rsi.append(talib.RSI(col, timeperiod=rsi_len)[-1])
except:
rsi.append(np.nan)
out.rsi[:] = rsi
def initialize(context):
"""
Called once at the start of the algorithm.
"""
context.stock = sid(8554) # SPY
# Create our dynamic stock selector.
attach_pipeline(make_pipeline(context), 'my_pipeline')
def make_pipeline(context):
"""
A function to create our dynamic stock selector (pipeline). Documentation on
pipeline can be found here: https://www.quantopian.com/help#pipeline-title
"""
sec_mask = StaticAssets([context.stock])
my_mask = sec_mask
# Factor of yesterday's close price.
yesterday_close = USEquityPricing.close.latest
rsi = MyTalibRSI(rsi_len=14, window_length = 80)
double_rsi = MyDoubleRSI(inputs=[MyTalibRSI.rsi], rsi_len=14, window_length = 80)
pipe = Pipeline(
screen = my_mask,
columns = {
'close': yesterday_close,
'rsi': rsi,
'double_rsi': double_rsi,
}
)
return pipe
def before_trading_start(context, data):
"""
Called every day before market open.
"""
context.output = pipeline_output('my_pipeline').dropna()
# log.info("Length of DF:%d" %len(context.output))
log.info("Original DF:\n%s" %context.output)
# These are the securities that we are interested in trading each day.
context.security_list = context.output.index