| Overall Statistics |
|
Total Orders 3256 Average Win 0.50% Average Loss -0.28% Compounding Annual Return 9.811% Drawdown 10.900% Expectancy 0.108 Start Equity 100000 End Equity 159703.05 Net Profit 59.703% Sharpe Ratio 0.388 Sortino Ratio 0.508 Probabilistic Sharpe Ratio 34.415% Loss Rate 60% Win Rate 40% Profit-Loss Ratio 1.78 Alpha 0.03 Beta 0.008 Annual Standard Deviation 0.079 Annual Variance 0.006 Information Ratio -0.123 Tracking Error 0.161 Treynor Ratio 4.063 Total Fees $6395.18 Estimated Strategy Capacity $150000000.00 Lowest Capacity Asset SPY R735QTJ8XC9X Portfolio Turnover 245.33% Drawdown Recovery 202 |
# region imports
from AlgorithmImports import *
# endregion
class FormalRedPelican(QCAlgorithm):
def initialize(self):
self.set_start_date(self.end_date - timedelta(5*365))
self.set_cash(100_000)
self.settings.seed_initial_prices = True
self._spy = self.add_equity("SPY")
self._spy.session.size = 2
self._target_volatility = 0.02
self._max_leverage = 1.5
self._leverage_buffer = 0.95
# Attach VWAP, return, volatility, and mean absolute deviation indicators.
self._spy.vwap = self.vwap(self._spy)
self._spy.roc = self.rocp(self._spy, 1, Resolution.DAILY)
self._spy.vol = IndicatorExtensions.of(StandardDeviation(14), self._spy.roc)
self._spy.deviation = MeanAbsoluteDeviation(14)
# Consolidate intraday data into 30-minute bars.
self.consolidate(self._spy, timedelta(minutes=30), self._consolidation_handler)
self._signals = [Signal(1), Signal(-1)]
# Liquidate one minute before market close each day.
self.schedule.on(
self.date_rules.every_day(self._spy),
self.time_rules.before_market_close(self._spy, 1),
self.liquidate
)
self.set_warm_up(timedelta(30))
def _consolidation_handler(self, bar):
if self.is_warming_up or not (self._spy.session.is_ready and Extensions.is_market_open(self._spy, False)):
return
open_price = self._spy.session.open
intraday_return = bar.close / open_price - 1
self._spy.deviation.update(bar.end_time, intraday_return)
if not (self._spy.vol.is_ready and self._spy.vwap.is_ready and self._spy.deviation.is_ready):
return
vwap_price = self._spy.vwap.current.value
# Evaluate long and short signal state against VWAP and deviation bands.
for signal in self._signals:
signal.evaluate(
self._spy.holdings, self._spy.session[1].close, open_price, bar.close, self._spy.deviation.current.value, vwap_price
)
# Exit any active position immediately after stop-out.
if any(signal.stopped_out for signal in self._signals):
self.liquidate()
return
# Enter the side that has a fresh signal and no existing position.
entry_sides = [signal.direction for signal in self._signals if signal.has_entry and not signal.in_position]
if not entry_sides:
return
# Scale leverage to target volatility with a conservative hard cap.
leverage = min(self._max_leverage, self._target_volatility / (self._spy.vol.current.value / 100)) * self._leverage_buffer
self.set_holdings(self._spy, entry_sides[0] * leverage)
class Signal:
def __init__(self, direction):
# Use +1 for long and -1 for short.
self.direction = direction
self.in_position = False
self.stopped_out = False
self.has_entry = False
def evaluate(self, holdings, previous_close, open_price, close_price, deviation, vwap_price):
# Update entry and stop state from the latest consolidated bar.
entry_anchor = self._extreme(open_price, previous_close)
entry_band = entry_anchor * (1 + self.direction * deviation)
stop_price = self._extreme(vwap_price, entry_band)
self.in_position = holdings.is_long if self.direction == 1 else holdings.is_short
self.stopped_out = self.in_position and self.direction * (close_price - stop_price) <= 0
self.has_entry = self.direction * (close_price - entry_band) >= 0
def _extreme(self, first_value, second_value):
# Return the direction-specific extreme value between two prices.
midpoint = (first_value + second_value) / 2
half_spread = abs(first_value - second_value) / 2
return midpoint + self.direction * half_spread