strategy rsi and diff based

Clone Algorithm

14

Loading...

There was an error loading this backtest.

Backtest from
to
with
initial capital

Cumulative performance:

Algorithm
Benchmark

Custom data:

Total Returns

--

Alpha

--

Beta

--

Sharpe

--

Sortino

--

Max Drawdown

--

Benchmark Returns

--

Volatility

--

Returns | 1 Month | 3 Month | 6 Month | 12 Month |

Alpha | 1 Month | 3 Month | 6 Month | 12 Month |

Beta | 1 Month | 3 Month | 6 Month | 12 Month |

Sharpe | 1 Month | 3 Month | 6 Month | 12 Month |

Sortino | 1 Month | 3 Month | 6 Month | 12 Month |

Volatility | 1 Month | 3 Month | 6 Month | 12 Month |

Max Drawdown | 1 Month | 3 Month | 6 Month | 12 Month |

from quantopian.algorithm import attach_pipeline, pipeline_output from quantopian.algorithm import order_optimal_portfolio from quantopian.pipeline import Pipeline from quantopian.pipeline.data.builtin import USEquityPricing from quantopian.pipeline.data import morningstar from quantopian.pipeline.data import Fundamentals from quantopian.pipeline.factors import AverageDollarVolume from quantopian.pipeline.filters import Q1500US from quantopian.pipeline.filters import Q500US from quantopian.pipeline.factors import SimpleMovingAverage, EWMA from quantopian.pipeline.factors import RSI from quantopian.pipeline.filters import QTradableStocksUS import talib as ta import numpy as np import math import pandas as pd import quantopian.optimize as opt def initialize(context): # Here define the commission and the slippage set_commission(commission.PerTrade(cost=0)) set_slippage(slippage.FixedSlippage(spread=0.00)) # Schedule balance function to run at the start of each week, when the market opens. schedule_function( balance, date_rules.every_day(), time_rules.market_open() ) # Record variables at the end of each day. schedule_function( record_vars, date_rules.every_day(), time_rules.market_close() ) # Create pipeline and attach it to algorithm. pipe = make_pipeline() attach_pipeline(pipe, 'pipeline') # Maximum leverage context.mxlv = 0 for i in range(1, 391): schedule_function(mxlv, date_rules.every_day(), time_rules.market_open(minutes=i)) def make_pipeline(): ''' Create pipeline. ''' # Number of stocks invested long and short at the same time num = 20 # Base universe set to the QTradableStocksUS. base_universe = sid(8554) # Create a dollar_volume factor using the top 2000 stocks of the base universe with the highest trading volume for the xx last trading days. -> Higher Liquidity highest_tradevolume = AverageDollarVolume(window_length=10, mask=base_universe).top(2000) # Choosing the stocks highest and lowest P/E ratio and Book Value Yield as additional filters (out of the stocks with the highest trading volume) per = morningstar.valuation_ratios.pe_ratio.latest bvy = morningstar.valuation_ratios.book_value_yield.latest # Choosing out of all stocks morningstar has data for selection_universe = bvy.top(250) and bvy.bottom(250) and per.top(250) and per.bottom(250) # Choosing out of stocks from defined filter #selection_universe = bvy.top(250, mask=highest_tradevolume) and bvy.bottom(250, mask=highest_tradevolume) and per.top(250, mask=highest_tradevolume) and per.bottom(250, mask=highest_tradevolume) # 10-day close exponential price average. mean_12 = EWMA( inputs=[USEquityPricing.close], window_length=12, decay_rate=(float, 0.8), mask=selection_universe) # 30-day close exponential price average. mean_18 = EWMA( inputs=[USEquityPricing.close], window_length=18, decay_rate=(float, 0.8), mask=selection_universe) # Calculate the percentage difference of the average prices and the ranking for the stocks of the selection universe. Stocks with the highest negative difference get the best ranking per_diff = (mean_12 - mean_18) / mean_18 ranking_per_diff = per_diff.rank(mask=selection_universe, ascending=True) # Calculate the RSI indicator and the ranking of the RSI for the stocks of the selection universe. Stocks with the lowest RSI's get the best ranking rsi = RSI(inputs=[USEquityPricing.close], window_length=14, mask=selection_universe) ranking_rsi = rsi.rank(mask=selection_universe, ascending=True) oversold = rsi < 25 and rsi.bottom(num) overbought = rsi > 75 and rsi.top(num) # Combine rankings of rsi and per_diff (sum of both together) ranking_combined = ranking_per_diff + ranking_rsi # Filter to select securities to short (num stocks with the highest sum and overbought) shorts = ranking_combined.top(num) and overbought # Filter to select securities to long (num stocks with the lowest sum and oversold) longs = ranking_combined.bottom(num) and oversold # Filter for all securities that want to trade (fulfilling the criteria) securities_to_trade = (shorts | longs) return Pipeline( columns={ 'rsi': rsi, 'longs': longs, 'shorts': shorts}, screen=(securities_to_trade)) def compute_target_weights(context, data): ''' Compute ordering weights. ''' # Weights proportional to rsi longs = context.out[context.out.longs ].rsi shrts = context.out[context.out.shorts].rsi longs *= .5 / longs.sum() if len(longs) else 1.0 # normalize shrts *= -.5 / shrts.sum() if len(shrts) else 1.0 weights = pd.concat([longs, shrts]) return weights # Initialize empty target weights dictionary. This will map securities to their target weight. #weights = {} # If there are securities in longs and shorts lists, compute even target weights for each security. #if context.longs and context.shorts: # long_weight = 0.5 / len(context.longs) # short_weight = -0.5 / len(context.shorts) # else: # return weights # Exit positions in portfolio if they are not in longs or shorts lists. for security in context.portfolio.positions: if security not in longs and security not in shrts and data.can_trade(security): weights[security] = 0 #for security in context.longs: weights[security] = longs #for security in context.shorts: weights[security] = shrts return weights def before_trading_start(context, data): ''' Get pipeline results. ''' # Gets pipeline output every day. context.out = pipe_results = pipeline_output('pipeline') # Go long in securities for which the 'longs' value is True and check if they can be traded. context.longs = [] for sec in pipe_results[pipe_results['longs']].index.tolist(): if data.can_trade(sec): context.longs.append(sec) # Go short in securities for which the 'shorts' value is True and check if they can be traded. context.shorts = [] for sec in pipe_results[pipe_results['shorts']].index.tolist(): if data.can_trade(sec): context.shorts.append(sec) def balance(context, data): ''' balance weekly. ''' # Calculate target weights to balance target_weights = compute_target_weights(context, data) # If have target weights, balance portfolio if len(target_weights): order_optimal_portfolio( objective=opt.TargetWeights(target_weights), constraints=[]) def record_vars(context, data): ''' Record variables at the end of each day. ''' longs = shorts = 0 for position in context.portfolio.positions.itervalues(): if position.amount > 0: longs += 1 elif position.amount < 0: shorts += 1 # Record variables. record( leverage=context.account.leverage, # shows leverage long_count=longs, # number of stocks invested long short_count=shorts) # number of stocks invested short def mxlv(context, data): if context.account.leverage > context.mxlv: context.mxlv = context.account.leverage record(MxLv = context.mxlv)