TRADING_ENV_DICT = {1: "Quantopian Algorithm", 2: "Quantopian Notebook", 3: "pylivetrader Paper", 4: "pylivetrader live", 5: "zipline trader"} TRADING_ENV = 3 MINUTES_PER_DAY = 450 import numpy as np import pandas as pd import datetime from talib import ATR, RSI ################# Refering Common (by all TRADING_ENV) Used zipline libraries ############## # refering to https://www.quantopian.com/algorithms/5ee42350bf88df0042c8a5ed # and https://www.quantopian.com/posts/guide-for-porting-your-algorithms-to-a-local-zipline-research-environment # StaticAsset refers to https://www.quantopian.com/posts/exclude-stocks-from-pipeline-by-sid # from zipline.api import * import zipline.algorithm as algo from zipline.pipeline import Pipeline from zipline.pipeline.filters import StaticAssets from zipline.finance.execution import StopOrder from zipline.utils.calendars import get_calendar from logbook import Logger, set_datetime_format from pytz import timezone ################# Refering Diffrent zipline libraries which different TRADING_ENV will use differently ############ set_datetime_format(lambda :datetime.datetime.now(tz=timezone('US/Pacific'))) log = Logger("Aegis") debug = log.debug info = log.info def null_print(s): pass if TRADING_ENV==5: from zipline.api import * from zipline.pipeline.data import USEquityPricing import matplotlib.pyplot as plt info = null_print debug = print else: from pylivetrader.api import order_target_percent, order_target, symbol, get_open_orders from pipeline_live.data.alpaca.pricing import USEquityPricing from pylivetrader.finance.execution import LimitOrder, StopOrder def get_this_day(): if TRADING_ENV in {1, 5}: return get_datetime() elif TRADING_ENV in {3, 4}: tz = timezone('PST8PDT') return pd.Timestamp.now(tz) ######################### Import Libraries Ends Here ####################### def update_ab_signal(context, data=None): ''' Update buy and sell symbol list from Pandas.dataframe -- context.ab_patterns ''' info(" update_ab_signal starts ") for security in list(context.security_list): this_day = get_this_day() this_day_date = this_day.date() symbol_pattern = context.ab_patterns[context.ab_patterns['symbol']==security.symbol] if not (security in context.ab_data): context.ab_data[security] = {} context.ab_data[security]["last_confirmed"] = \ get_symbol_pattern_information(\ symbol_pattern[symbol_pattern['confirmation_date']= this_day_date): context.ab_data[security]["latest"]["confirmation_level"] = np.nan context.ab_data[security]["latest"]["confirmation_date"] = np.nan context.ab_data[security]["latest"]["signal"] = np.nan info("Latest AB Data: {}".format(context.ab_data[security]["latest"])) info("Last Confirmed AB Data: {}".format(context.ab_data[security]["last_confirmed"])) # Reset Open Price and Previous Closing Price Everyday context.ab_data[security]["open_price"] = np.nan context.ab_data[security]["previous_closing_price"] = np.nan info(" update_ab_signal done ") def cal_mavg_short_long(data, security, context): if data==None: return short_data = data.history(security, 'price', bar_count=5, frequency="1d") long_data = data.history(security, 'price', bar_count=13, frequency="1d") short_mavg = short_data.mean() long_mavg = long_data.mean() context.short_mavg[security] = short_mavg context.long_mavg[security] = long_mavg def load_pattern(context): url = 'http://stock-data.fipy.me/ab_patterns_extend.csv' date_cols = ['date', 'confirmation_date'] context.ab_patterns = pd.read_csv(url, parse_dates=date_cols) info(context.ab_patterns.head()) def get_my_static_list(): return [symbol('TQQQ')] def initialize(context): """ Called once at the start of the algorithm. """ context.buy_today = set() context.sell_today = set() context.ab_patterns = None context.ab_data = {} context.ab_stops = {} context.short_mavg = {} context.long_mavg = {} context.atr_stoploss = {} context.num_trading_min = 0 context.params = {'start_trade_after_market_open_hours': 0.5, 'pct_init': 0.03, 'pct_trail': 0.04, 'days_wait_max': 4, 'atr_stop_days': 14, 'atr_times_sigma': 2.0, 'bar_counts': {'short': 5, 'long': 13}} # Record tracking variables at the end of each day. schedule_function( record_vars, date_rules.every_day(), time_rules.market_close(), ) # Load AB data load_pattern(context) context.security_list = get_my_static_list() for security in list(context.security_list): context.ab_data[security] = {} context.ab_data[security]['last_confirmed'] = {} context.ab_data[security]["open_price"] = np.nan context.ab_data[security]["previous_closing_price"] = np.nan update_ab_signal(context) def before_trading_start(context, data): """ Called every day before market open. """ info("Daily holding {}".format(context.portfolio.positions)) # context.output = pipeline_output('pipeline') context.short_mavg = {} context.long_mavg = {} context.atr_stoploss = {} context.num_trading_min = 0 # These are the securities that we are interested in trading each day. # context.security_list = context.output.index context.security_list = get_my_static_list() for security in list(context.security_list): context.ab_data[security]["open_price"] = np.nan context.ab_data[security]["previous_closing_price"] = np.nan load_pattern(context) update_ab_signal(context, data) def get_symbol_pattern_information(symbol_pattern, sort_column='date'): if symbol_pattern.empty or (len(symbol_pattern.values)==0): return {} latest_pattern = symbol_pattern.sort_values(sort_column, ascending=False).iloc[0, :] return { "date": latest_pattern['date'], "pattern_name": latest_pattern['pattern'], "buy_level": latest_pattern['buy_level'], "stop_loss": latest_pattern['stop_loss'], "sell_level": latest_pattern['sell_level'], "confirmation_level": latest_pattern['confirmation_level'], "confirmation_date": latest_pattern['confirmation_date'], "signal": latest_pattern['signal'], "type": latest_pattern['type'], "patterns_total": latest_pattern['patterns_total'], "confirmed_total": latest_pattern['confirmed_total'], "profitable_signals": latest_pattern['profitable_signals'], "confirmation": latest_pattern['confirmation'], "profitable_signal": latest_pattern['profitable_signal'], "profit": latest_pattern['profit'], "loss": latest_pattern['loss'], "total_net": latest_pattern['total_net'] } def record_vars(context, data): """ Plot variables at the end of each day. """ # info("{} positions {}".format(get_datetime(), context.portfolio)) update_ab_signal(context, data) def get_atr_trail_percent(data, security, context): ''' Return ATR (Average Trading Randing) Percent for Trailing Stop Loss ATR is using 35 moving prices pandas >= 0.13.0:Passing a Series directly to a cython function expecting an ndarray type will no long work directly, you must pass Series.values ''' highs = data.history(security, 'high', 35, '1d').dropna().values lows = data.history(security, 'low', 35, '1d').dropna().values closes = data.history(security, 'close', 35, '1d').dropna().values atr_stop_price_delta = ATR(highs, lows, closes, timeperiod=context.params['atr_stop_days'])[-1] context.atr_stoploss[security] = atr_stop_price_delta #context.params['atr_times_sigma'] * (atr_stop_price_delta / current_price) def get_n_minute_slope(data, security, context, minute = 15, bars = 5): ''' Return slope value (current moving average price - last moving average price) / last moving average price ''' minute_prices = data.history(security, 'price', minute + 2 * bars, '1m').dropna().values last_avg_price = np.mean(minute_prices[-(minute+bars):-bars]) current_avg_price = np.mean(minute_prices[-minute:]) return (current_avg_price - last_avg_price) / last_avg_price def get_rsi_value(data, security, timeperiod = 6): ''' Return current rsi value. Note that in Intraday, because of the change of current price, it will change on every different minute. ''' current_price = data.current(security, 'price') past_close_prices = data.history(security, 'price', 2 * timeperiod, '1d').dropna().values close_prices = past_close_prices + [current_price] rsi = RSI(close_prices, timeperiod = timeperiod) return rsi[-1] def get_n_minute_slope_noavg(data, security, context, minute = 15, bars = 5): current_price = data.current(security, 'price') n_min_ago_price_list = data.history(security, 'price', minute, '1m').dropna().values n_min_ago_price = n_min_ago_price_list[0] slope = (n_min_ago_price - current_price)/n_min_ago_price return slope def get_rounded_buy_share_count(context, data, security): ''' Use current security market price and cash available to caculate how many shares can buy, to avoid margin call. after this call, can put buy order use zipline.api.order_target(self, asset, target, limit_price=None, stop_price=None, style=None) to place order https://www.zipline.io/appendix.html#zipline.api.order_target ''' current_security_price = data.current(security, 'price') cash_available = context.portfolio.cash slope = 1.6 * abs(get_n_minute_slope_noavg(data, security, context)) cash_usage_ratio = (1 - slope) buy_share_ability_cash_based = int((cash_available * cash_usage_ratio) / current_security_price) price_bumpup_ratio = (1 + slope) buy_share_ability_price_based = int(cash_available / (current_security_price * price_bumpup_ratio)) debug("buy_share_ability_cash_based:{}, buy_share_ability_price_based:{}".format(buy_share_ability_cash_based,buy_share_ability_price_based)) return max(buy_share_ability_cash_based, buy_share_ability_price_based) def handle_data(context, data): """ Called every minute. """ def debug_prices(print_level=debug): print_level("~~~~~~~~ Technical Analysis ~~~~~~~~") print_level("Short MAVG 5: {:.3f}".format(short_mavg)) print_level("Short MAVG 13: {:.3f}".format(long_mavg)) print_level("ATR Stoploss Range: {:.3f}".format(atr_stoploss_range)) print_level("ATR Stoploss: {:.3f}".format(atr_stoploss)) print_level("Slope 15: {:.6f}".format(slope_15min)) print_level("RSI: {:.3f}".format(rsi)) def print_signal_check(print_level=debug): print_level("******** SIGNAL CHECK ********") print_level("~~~~~~~~ {} ~~~~~~~~".format(security.symbol)) print_level("~~~~~~~~ Latest Prices ~~~~~~~~") print_level("Current System Time: {}".format(current_system_time)) print_level("Current Price Real: {}".format(current_price_real)) print_level("Current Price (10 Min Mean): {}".format(current_price)) print_level("Open Price: {}".format(open_price)) print_level("Previous Closing Price: {}".format(previous_closing_price)) print_level("~~~~~~~~ Latest AB Signal ~~~~~~~~") print_level("Last AB Pattern: {}".format(last_ab_pattern)) print_level("Last AB Date: {}".format(last_ab_date)) print_level("Last AB Age: {}".format(last_ab_age)) print_level("Last AB Buy Level: {}".format(last_ab_buy_level)) print_level("Last AB Stop-Loss Level: {}".format(last_ab_stop_loss)) print_level("Last AB Sell Level: {}".format(last_ab_sell_level)) print_level("Last AB Signal: {}".format(last_ab_signal)) print_level("Last Confirmation Level: {}".format(last_ab_confirmation_level)) print_level("Last Confirmation Date: {}".format(last_ab_confirmation_date)) print_level("Last Confirmation Age: {}".format(last_ab_confirmation_age)) if bool(last_confirmed_ab_record): print_level("~~~~~~~~ Last Confirmed AB Signal ~~~~~~~~") print_level("Last Confirmed AB Signal: {}".format(last_confirmed_ab_signal)) print_level("Last Confirmed AB Confirmation Level: {}".format(last_confirmed_ab_confirmation_level)) print_level("Last Confirmed AB Confirmation Date: {}".format(last_confirmed_ab_confirmation_date)) debug_prices(print_level) # ----------------------------------------------------- # Do not trade the First 30 min of market open # - Record Open Price # ----------------------------------------------------- context.num_trading_min += 1 number_of_min_to_skip = 30 current_system_time = get_this_day() #info("Current System Time: {}".format(current_system_time)) #info("Current Price:{}".format(data.current(context.security_list, 'price'))) # ----------------------------------------------------- # Record Open Price # ----------------------------------------------------- if context.num_trading_min < number_of_min_to_skip: # ----------------------------------------------------- # Record Open Price # ----------------------------------------------------- for security in context.security_list: if np.isnan(context.ab_data[security]["open_price"]): open_price = data.current(security, 'price') previous_closing_price = data.history(security, 'price', bar_count=2, frequency="1d").dropna()[0] info("Current System Time: {}".format(current_system_time)) info("Open Price:{}".format(open_price)) #info("Previous Closing Price:{}".format(previous_closing_price)) context.ab_data[security]["open_price"] = open_price context.ab_data[security]["previous_closing_price"] = previous_closing_price return for security in context.security_list: # debug("Time: {}, Current Price:{}".format(get_datetime(), data.current(security, 'price'))) if security in context.ab_data: if (security not in context.short_mavg) or (security not in context.long_mavg): cal_mavg_short_long(data, security, context) get_atr_trail_percent(data, security, context) if not context.ab_data[security].get("used", False): if "latest" in context.ab_data[security]: num_min_avg = 10 # ----------------------------------------------------- # Buy / Sell Logic # ----------------------------------------------------- current_price_real = data.current(security, 'price') current_price = data.history(security, 'price',\ bar_count=num_min_avg, frequency='1m').dropna().median() open_price = context.ab_data[security]["open_price"] previous_closing_price = context.ab_data[security]["previous_closing_price"] last_ab_record = context.ab_data[security]["latest"] last_confirmed_ab_record = context.ab_data[security]["last_confirmed"] last_ab_pattern = last_ab_record['pattern_name'] last_ab_buy_level = float(last_ab_record['buy_level']) last_ab_stop_loss = float(last_ab_record['stop_loss']) last_ab_sell_level = float(last_ab_record['sell_level']) this_day_date = get_this_day().date() last_ab_date = last_ab_record['date'].date() last_ab_age = (this_day_date - last_ab_date).days # ----------------------------------------- # Current AB Signal Confirmation Info # ----------------------------------------- last_ab_signal = last_ab_record['signal'] last_ab_confirmation_level = \ float(last_ab_record['confirmation_level']) last_ab_confirmation_date = \ last_ab_record['confirmation_date'] last_ab_confirmation_age = 100 if (not pd.isnull(last_ab_confirmation_date)): last_ab_confirmation_date = last_ab_confirmation_date.date() last_ab_confirmation_age = (this_day_date - last_ab_confirmation_date).days # ----------------------------------------- # Current AB Signal Performance Stats # ----------------------------------------- last_ab_total_net_raw = last_ab_record['total_net'] if not pd.isnull(last_ab_total_net_raw): last_ab_total_net = int(last_ab_total_net_raw[0:len(last_ab_total_net_raw)-1]) else: last_ab_total_net = 0 # ----------------------------------------- # Last Confirmed AB Signal # ----------------------------------------- if bool(last_confirmed_ab_record): last_confirmed_ab_signal = last_confirmed_ab_record['signal'] last_confirmed_ab_confirmation_level = \ float(last_confirmed_ab_record['confirmation_level']) last_confirmed_ab_confirmation_date = \ last_confirmed_ab_record['confirmation_date'] # ----------------------------------------- # Technical Analysis # ----------------------------------------- #Moving Average short_mavg = context.short_mavg[security] long_mavg = context.long_mavg[security] # ATR StopLoss atr_stoploss_range = context.atr_stoploss[security] atr_stoploss = open_price - atr_stoploss_range # Last 15 Min Slope slope_15min = get_n_minute_slope(data, security, context) * 10000 rsi = get_rsi_value(data, security) debug_prices(print_level=info) # ----------------------------------------- # Orders and Holdings # ----------------------------------------- num_open_orders = len(get_open_orders()) position_amount = context.portfolio.positions[security].amount # debug("Time: {} Price: {:.6f} Slope 15 min: {:.6f} Slope_15min_no_avg: {:.6f} ". # format(current_system_time, current_price_real, slope_15min, slope_15min_no_avg)) # debug("Positon Amount: {} Num Open Orders: {}". # format(position_amount, num_open_orders)) record(TQQQ=current_price, short_mavg=short_mavg, long_mavg=long_mavg, atr_stoploss=atr_stoploss, slope_15min=slope_15min) print_signal_check(print_level=info) # +++++++++++++++++++++++++++++++++++++++++ # Trading Logic: Buy # +++++++++++++++++++++++++++++++++++++++++ if (context.portfolio.positions[security].amount <= 0 and num_open_orders == 0): perform_buy = False buy_reason = [] buy_check_mvg_cross = False buy_check_AB = False buy_check_atr_stoploss = False buy_check_slope_15min = False #---------------------- # Do not use signal #---------------------- #buy_check_mvg_cross = True #buy_check_slope_15min = True #buy_check_atr_stoploss = True # # Make Buy Decision # # Check if short_mavg > long_mavg #info("MAvg Short/Long: {} / {}".format(short_mavg, long_mavg)) if (short_mavg > long_mavg): buy_check_mvg_cross = True buy_reason.append("Short_Mavg > Long_Mavg") # # Auto Intraday Buy Confirmed # if (not buy_check_AB and "BULLISH" in last_ab_pattern and last_ab_total_net >= 7): if (current_price > last_ab_buy_level): buy_check_AB = True buy_reason.append("AB Intraday Buy Confirmed using Buy Level with Total Net >= 7") if (last_ab_buy_level is None and current_price > last_ab_stop_loss): buy_check_AB = True buy_reason.append("AB Intraday Buy Confirmed using BULLISH STOP LOSS with Total Net >= 7") # if (slope_15min > 35.0): # buy_check_AB = True # buy_reason.append("AB Intraday Buy Confirmed using Slope 15 > 35 with Total Net >= 7") # # AB Buy Confirmed # if (not buy_check_AB and last_ab_signal == 'BUY' and last_ab_total_net >= 3): buy_check_AB = True buy_reason.append("AB Buy Confirmed with Total Net >= 3") # # AB Buy Re-Confirmed # # if (not buy_check_AB and # last_confirmed_ab_signal == 'BUY' and slope_15min > 35.0 and mavg_5min > mavg_15min): # buy_check_AB = True # buy_reason.append("AB Buy Re-Confirmed") # # Check Within ATR-StopLoss OR Slope 15 > 35 # if (current_price > atr_stoploss or slope_15min > 35.0): buy_check_atr_stoploss = True buy_reason.append("Within ATR-StopLoss OR Slope 15 > 35") # # Slope Last 15 Min Up > -20 # if (not buy_check_slope_15min and slope_15min > -20): buy_check_slope_15min = True buy_reason.append("Slope Last 15 Min > -20") # ------------------------------------------------ # Check if All Buy conditions are met and # set perf_buy to True # # 1. Short_Mavg > Long_Mavg # 2. AB Buy Confirmed # 3. Within ATR-StopLoss # 4. Slope Last 15 Min Up # ------------------------------------------------ if (buy_check_mvg_cross and buy_check_AB and buy_check_atr_stoploss and buy_check_slope_15min): buy_reason.insert(0, "Buy Method 1: ") perform_buy = True # ------------------------------------------------ # Buy Check Method 2 # # 1. Short_Mavg > Long_Mavg # 2. Buy_check_atr_stoploss # 3. buy_check_slope_15min # ------------------------------------------------ if (buy_check_mvg_cross and slope_15min > 35 and current_price_real > open_price and not ("BEARISH" in last_ab_pattern and last_ab_age < 2) and not ((last_ab_signal == 'SELL' or last_ab_signal == 'SHORT') and last_ab_total_net >= 5 and last_ab_confirmation_age < 3) ): buy_reason.insert(0, "Buy Method 2: ") perform_buy = True # # Execute Buy # if (perform_buy): debug("++++++++ PERFORM BUY ++++++++") debug(last_ab_record) debug("Buy Reason: {}".format(buy_reason)) #info(context.portfolio.positions[security].amount) print_signal_check() #order_target_percent(security, 1.0) order_target_percent(security, 1.0, limit_price=current_price_real) # --------------------------------------------- # Trading Logic: SELL # --------------------------------------------- if (context.portfolio.positions[security].amount > 0 and num_open_orders == 0): perform_sell = False sell_reason = [] sell_check_mvg_cross = False sell_check_AB = False # # # # Auto Intraday Sell Confirmation # # Disable for now, as live trading shows lose trade on # # - 11/23/2020, 12/21/2020 # # # if ("BEARISH" in last_ab_pattern and \ # current_price < last_ab_sell_level and # current_price < open_price): # perform_sell = True # sell_reason = "AB Intraday Sell Confirmed" # # Sell Using ATR-StopLoss # if ((current_price_real < atr_stoploss)# or current_price_real < long_mavg) and slope_15min < -40 ): perform_sell = True sell_reason = "Sell Method 1: ATR-StopLoss Confirmed, Slope Last 15 Min < -40" # # AB Sell Confirmation # if ((last_ab_signal == 'SELL' or last_ab_signal == 'SHORT') and slope_15min < -40 and last_ab_total_net >= 5 and last_ab_confirmation_age < 3): perform_sell = True sell_reason = "Sell Method 2: AB Sell Confirmed, Slope Last 15 Min < -40, Total Net > 5, Confirmation Age < 3" # # Execute Sell # if (perform_sell): debug("-------- PERFORM SELL --------") debug(last_ab_record) debug(sell_reason) print_signal_check() order_target_percent(security, 0.0) ############################### END OF ALGO ########################