| Overall Statistics |
|
Total Orders 7563 Average Win 0.52% Average Loss -0.33% Compounding Annual Return 7.007% Drawdown 24.400% Expectancy 0.074 Start Equity 100000 End Equity 209213.13 Net Profit 109.213% Sharpe Ratio 0.262 Sortino Ratio 0.293 Probabilistic Sharpe Ratio 0.721% Loss Rate 59% Win Rate 41% Profit-Loss Ratio 1.60 Alpha 0.005 Beta 0.422 Annual Standard Deviation 0.15 Annual Variance 0.023 Information Ratio -0.258 Tracking Error 0.161 Treynor Ratio 0.093 Total Fees $0.00 Estimated Strategy Capacity $1100000.00 Lowest Capacity Asset SHY SGNKIKYGE9NP Portfolio Turnover 193.14% |
# region imports
from AlgorithmImports import *
# endregion
# Source: https://www.investopedia.com/articles/trading/04/091504.asp
class KellyCriterionSMACrossoverAlgorithm(QCAlgorithm):
def initialize(self):
self.set_start_date(2014, 1, 1)
# Remove fees to focus the research on the portfolio weighting, not the signal.
self.set_security_initializer(lambda s: s.set_fee_model(ConstantFeeModel(0)))
# Add the risky and risk-free assets.
self._risk_asset = self.add_equity('IBM', Resolution.HOUR, leverage=6)
self._rf_asset = self.add_equity('SHY', Resolution.HOUR, leverage=6)
# Add some strategy-specific indicators/variables.
self._risk_asset.short_sma = self.sma(self._risk_asset.symbol, 1)
self._risk_asset.long_sma = self.sma(self._risk_asset.symbol, 6)
# Create the KellyCriterion object.
self._risk_asset.signal = 0
self._kelly_criterion = KellyCriterion(1.5, 40)
# Add a warm-up period so we some historical performance of the strategy once we start trading.
self.set_warm_up(timedelta(365))
# Add a list and Scheduled Event to track the average exposure to the risky asset.
self._risky_weights = []
self.schedule.on(self.date_rules.every_day(self._risk_asset.symbol), self.time_rules.at(23, 59), self._sample_weight)
def on_data(self, data: Slice):
# Wait until the market is open.
if not data.bars or not self.is_market_open(self._risk_asset.symbol):
return
# Pass the latest signal to the KellyCriterion object.
if not self._risk_asset.signal and self._risk_asset.short_sma > self._risk_asset.long_sma:
self._risk_asset.signal = 1
self._kelly_criterion.update_signal(1, self._risk_asset.price)
elif self._risk_asset.signal and self._risk_asset.short_sma < self._risk_asset.long_sma:
self._risk_asset.signal = 0
self._kelly_criterion.update_signal(0, self._risk_asset.price)
# Wait until we can trade.
if self.is_warming_up or not self._kelly_criterion.is_ready:
return
# Update the portfolio holdings based on the signal.
if self._risk_asset.signal and not self._risk_asset.holdings.is_long:
# Cap the exposure at 575% to avoid errors.
weight = min(5.75, self._kelly_criterion.weight())
self.set_holdings(
[
PortfolioTarget(self._risk_asset.symbol, weight),
# If the target weight for the risky asset is <1, then raise the porfolio
# exposure to 100% with the risk-free asset.
PortfolioTarget(self._rf_asset.symbol, 0 if weight > 1 else 1-weight)
]
)
elif not self._risk_asset.signal and self._risk_asset.holdings.is_long:
# If the signal is 0, put 100% of the portfolio in the risk-free asset.
self.set_holdings([PortfolioTarget(self._rf_asset.symbol, 1)], True)
def _sample_weight(self):
self._risky_weights.append(self._risk_asset.holdings.holdings_value / self.portfolio.total_portfolio_value)
def on_end_of_algorithm(self):
self.log(f"Average weight: {sum(self._risky_weights) / len(self._risky_weights)}")
class KellyCriterion:
def __init__(self, factor, period):
self._factor = factor
self._period = period
self._trades = np.array([])
def update_signal(self, signal, price):
if signal: # Enter
self._entry_price = price
else: # Exit
self._trades = np.append(self._trades, [price - self._entry_price])[-self._period:]
def weight(self):
# Wait until there are enough trade samples.
if not self.is_ready:
return None
# Calculate the Kelly %.
wins = self._trades[self._trades > 0]
losses = self._trades[self._trades < 0]
if not losses.sum():
return self._factor
if not wins.sum():
return 0
win_loss_ratio = wins.mean() / losses.mean()
winning_probability = len(wins) / self._period
return self._factor*(winning_probability - (1-winning_probability)/win_loss_ratio)
@property
def is_ready(self):
return len(self._trades) == self._period