| Overall Statistics |
|
Total Orders 1115 Average Win 0.64% Average Loss -0.67% Compounding Annual Return 8.717% Drawdown 20.200% Expectancy 0.229 Start Equity 100000 End Equity 302055.28 Net Profit 202.055% Sharpe Ratio 0.489 Sortino Ratio 0.426 Probabilistic Sharpe Ratio 7.934% Loss Rate 37% Win Rate 63% Profit-Loss Ratio 0.96 Alpha 0 Beta 0 Annual Standard Deviation 0.091 Annual Variance 0.008 Information Ratio 0.7 Tracking Error 0.091 Treynor Ratio 0 Total Fees $2812.37 Estimated Strategy Capacity $100000000.00 Lowest Capacity Asset SPY R735QTJ8XC9X Portfolio Turnover 15.56% |
#region imports from AlgorithmImports import * from strategy_calendar import StrategyCalendar, FOMC from pandas.core.frame import DataFrame #endregion class SeasonalityComposite(QCAlgorithm): _trade_TOM: bool = True _trade_FOMC: bool = True _trade_expiration: bool = True _trade_payday: bool = True _before_tom_d_offset: int = 5 _after_tom_sell_days: List[int] = [4,5,6,9] _min_offset: int = 5 _option_exp_week_holding_period: int = 5 _traded_weight: float = 1 def initialize(self) -> None: self.set_start_date(2012, 1, 1) self._init_cash: int = 100_000 self.set_cash(self._init_cash) self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN) self.settings.minimum_order_margin_portfolio_percentage = 0 self.settings.daily_precise_end_time = True assert max(self._after_tom_sell_days) <= 10, 'maximum day offset is 10' assert all(np.diff(self._after_tom_sell_days) > 0), 'exit day series (self._after_tom_sell_days) should be ascending and number should be repeated' self._sell_day_count: float = len(self._after_tom_sell_days) self._quantity_portion_to_sell: int = 0 ticker: str = 'SPY' self._market: Symbol = self.add_equity(ticker, Resolution.MINUTE).symbol # benchmark self._benchmark_values: List[float] = [] self._FOMC_obj = FOMC(self) self._liquidation_time: Optional[datetime] = None self._after_open_flag: bool = False self.schedule.on( self.date_rules.every_day(self._market), self.time_rules.after_market_open(self._market, self._min_offset), self._after_open ) self.schedule.on( self.date_rules.month_end(self._market, self._before_tom_d_offset - 1), self.time_rules.before_market_close(self._market, self._min_offset), self._open_TOM ) for d_offset in self._after_tom_sell_days: self.schedule.on( self.date_rules.month_start(self._market, d_offset - 1), self.time_rules.before_market_close(self._market, self._min_offset), self._close_TOM ) def _open_TOM(self) -> None: # open new trade if self._trade_TOM: if not self.portfolio[self._market].invested: self._quantity_portion_to_sell = 0 q_to_trade: int = self.calculate_order_quantity(self._market, 1.) adjusted_q: int = (q_to_trade // self._sell_day_count) * self._sell_day_count # adjust to nearest lower divisable quantity self.market_order(self._market, adjusted_q, tag='entry_TOM') def _close_TOM(self) -> None: # close trade partially if self.portfolio[self._market].invested: self.market_order( self._market, -self._quantity_portion_to_sell, tag=f'portional exit weight: {1. / self._sell_day_count} of Q ({self._sell_day_count * self._quantity_portion_to_sell})' ) def on_order_event(self, order_event: OrderEvent) -> None: order = self.transactions.get_order_by_id(order_event.order_id) if order_event.status == OrderStatus.FILLED: if 'entry_TOM' in order.tag: self._quantity_portion_to_sell = order.quantity / self._sell_day_count def on_data(self, slice: Slice) -> None: # try to liquidate if self._liquidation_time is not None: if self.time >= self._liquidation_time: self._liquidation_time = None self.liquidate(self._market) if not self._after_open_flag: return self._after_open_flag = False # open new trade if not self.portfolio.invested: now: datetime.date = self.time.date() fomc_signal: bool = StrategyCalendar.is_FOMC(now, self._FOMC_obj.dates) if self._trade_FOMC else False expiration_signal: bool = StrategyCalendar.is_option_expiration_week(now) if self._trade_expiration else False payday_signal: bool = StrategyCalendar.is_payday(now) if self._trade_payday else False # join two strategy signals; option expiration week and TOM is separate signal joined_signal: List[bool] = [ s for s in [ fomc_signal, payday_signal ] ] holding_period_days: int = -1 # expiration week signal has priority if expiration_signal: holding_period_days: int = self._option_exp_week_holding_period elif any(s == True for s in joined_signal): # execute only on one signal if multiple signals occured holding_period_days: int = 0 if holding_period_days != -1: liquidation_time: int = self.securities[self._market].exchange.hours.get_next_market_close( self.time + timedelta(days=holding_period_days), extended_market_hours=False ) - timedelta(minutes=self._min_offset) self._liquidation_time = liquidation_time # open new position self.set_holdings(self._market, self._traded_weight) def _after_open(self) -> None: self._after_open_flag = True def on_end_of_day(self) -> None: # wait for available data if not (self.securities.contains_key(self._market) and self.securities[self._market].get_last_data()): return # print benchmark in main equity plot mkt_price_df: DataFrame = self.history(self._market, 2, Resolution.DAILY) if not mkt_price_df.empty: benchmark_price: float = mkt_price_df['close'].unstack(level= 0).iloc[-1] if len(self._benchmark_values) == 2: self._benchmark_values[-1] = benchmark_price benchmark_perf: float = self._init_cash * (self._benchmark_values[-1] / self._benchmark_values[0]) self.plot('Strategy Equity', self._market, benchmark_perf) else: self._benchmark_values.append(benchmark_price)
#region imports from AlgorithmImports import * from pandas.tseries.offsets import BDay, BMonthEnd, BMonthBegin from dateutil.relativedelta import relativedelta, WE, FR, SA #endregion class FOMC(): def __init__(self, algorithm: QCAlgorithm) -> None: csv_string_file = algorithm.Download( 'data.quantpedia.com/backtesting_data/economic/fed_days.csv' ) dates = csv_string_file.split('\r\n') self._dates: List[datetime.date] = [ (datetime.strptime(x, "%Y-%m-%d")).date() for x in dates ] @property def dates(self) -> List[datetime.date]: return self._dates class StrategyCalendar(): @staticmethod def _get_friday_before_options_expiration(now: datetime.date) -> datetime.date: # find thrid saturday and get friday before that return (date(now.year, now.month, 1) + relativedelta(weekday=SA(3))) - timedelta(days=1) @staticmethod def _get_payday(now: datetime.date) -> datetime.date: payday = date(now.year, now.month, 1) + relativedelta(day=15) if payday.weekday() == 5: # saturday payday = payday - timedelta(days=1) elif payday.weekday() == 6: # sunday payday = payday - timedelta(days=2) return payday @staticmethod def is_TOM(now: datetime.date) -> bool: offset = BMonthEnd() return now == offset.rollforward(now).date() - timedelta(days=1) @staticmethod def is_FOMC(now: datetime.date, fomc_days: List[datetime.date]) -> bool: return now in fomc_days @staticmethod def is_option_expiration_week(now: datetime.date) -> bool: second_friday_of_month: datetime.date = StrategyCalendar._get_friday_before_options_expiration(now) return now == second_friday_of_month @staticmethod def is_payday(now: datetime.date) -> bool: paydate: datetime.date = StrategyCalendar._get_payday(now) return now == paydate - timedelta(days=1)
