| Overall Statistics |
|
Total Orders 60 Average Win 4.56% Average Loss -3.70% Compounding Annual Return 2.063% Drawdown 10.300% Expectancy 0.154 Start Equity 100000 End Equity 110751.04 Net Profit 10.751% Sharpe Ratio -0.587 Sortino Ratio -0.464 Probabilistic Sharpe Ratio 3.720% Loss Rate 48% Win Rate 52% Profit-Loss Ratio 1.23 Alpha -0.02 Beta -0.115 Annual Standard Deviation 0.045 Annual Variance 0.002 Information Ratio -0.515 Tracking Error 0.164 Treynor Ratio 0.231 Total Fees $90.01 Estimated Strategy Capacity $340000000.00 Lowest Capacity Asset GOOCV VP83T1ZUHROL Portfolio Turnover 1.68% Drawdown Recovery 161 |
# region imports
from AlgorithmImports import *
from stat_arb import PairStatArb
# endregion
class SpectralStatArb(QCAlgorithm):
def initialize(self):
self.set_start_date(self.end_date - timedelta(5 * 365))
self.set_cash(100_000)
self.settings.automatic_indicator_warm_up = True
self.settings.seed_initial_prices = True
self._formation_date_rule = FuncDateRule("semiannual_formation", lambda start, end: [datetime(year, month, 1) for year in range(start.year, end.year + 1) for month in (1, 7) if start <= datetime(year, month, 1) <= end])
self.universe_settings.resolution = Resolution.DAILY
self.universe_settings.schedule.on(self._formation_date_rule)
self._universe = self.add_universe(lambda fundamental: [f.symbol for f in sorted([f for f in fundamental if f.has_fundamental_data and not np.isnan(f.market_cap) and f.market_cap > 5e9], key=lambda f: f.dollar_volume)[-100:]])
self.laplacian_lookback = 252
self.entry_z_score = 2.0
self.exit_z_score = 0.0
self._pair_spread = None
self.set_warm_up(timedelta(45))
def on_warmup_finished(self):
# Configure training and prediction schedule.
training_time_rule = self.time_rules.at(6, 0)
trading_time_rule = self.time_rules.at(8, 0)
self.train(self._formation_date_rule, training_time_rule, self._train)
self.schedule.on(self.date_rules.every_day("SPY"), trading_time_rule, self._rebalance)
# Rebalance today too.
if self.live_mode:
self._train()
self._rebalance()
else:
self.train(self.date_rules.today, training_time_rule, self._train)
self.schedule.on(self.date_rules.today, trading_time_rule, self._rebalance)
def on_securities_changed(self, changes):
for security in changes.added_securities:
# Track returns for clustering.
security.log_return = self.logr(security, 1)
security.log_return.window.size = self.laplacian_lookback
security.log_return.reset()
history = self.history[TradeBar](security, self.laplacian_lookback + 1, Resolution.DAILY)
for bar in history:
security.log_return.update(bar.end_time, bar.close)
for security in changes.removed_securities:
# Drop the active pair if one of its legs leaves the universe.
if self._pair_spread and security.symbol in self._pair_spread.symbols:
self._drop_pair()
self.deregister_indicator(security.log_return)
def _train(self):
# Train pair selection and warm the selected spread indicator.
pair_spread = PairStatArb(self.laplacian_lookback).train(self, self._universe)
self._drop_pair()
if pair_spread is None:
return
# Store the active spread indicator.
self._pair_spread = pair_spread
self.register_indicator(pair_spread.symbol_a, self._pair_spread, Resolution.DAILY)
self.register_indicator(pair_spread.symbol_b, self._pair_spread, Resolution.DAILY)
def _rebalance(self):
pair = self._pair_spread
if pair is None or not pair.is_ready:
return
asset_a, asset_b = pair.symbols
security_a = self.securities[asset_a]
security_b = self.securities[asset_b]
half_life = float(pair.half_life)
if not np.isfinite(half_life) or half_life > 126:
return
z_score = float(pair.z_score)
if not np.isfinite(z_score):
return
invested = security_a.invested or security_b.invested
if not invested and abs(z_score) > self.entry_z_score:
# Short the spread when z-score is high, and long it when z-score is low.
gross = 1 + abs(pair.hedge_ratio)
if not np.isfinite(gross) or gross <= 0:
return
direction = np.sign(z_score)
weight_a = -direction / gross
weight_b = direction * pair.hedge_ratio / gross
self.set_holdings([PortfolioTarget(asset_a, weight_a), PortfolioTarget(asset_b, weight_b)], False)
elif invested and (security_a.holdings.quantity < 0 and z_score <= self.exit_z_score or security_a.holdings.quantity > 0 and z_score >= -self.exit_z_score):
# Take profit once the spread reverts toward the mean.
self.liquidate(asset_a)
self.liquidate(asset_b)
def _drop_pair(self):
# Close the pair positions and release the active spread indicator.
if self._pair_spread:
for symbol in self._pair_spread.symbols:
self.liquidate(symbol)
self.deregister_indicator(self._pair_spread)
self._pair_spread = None
from AlgorithmImports import *
class PairSpread(PythonIndicator):
def __init__(self, symbol_a, symbol_b, hedge_ratio: float, price_scale_a: float, price_scale_b: float, spread_mean: float, spread_sd: float, half_life: float, distance: float, name: str = "PairSpread"):
super().__init__()
self.name = name
self.symbol_a = symbol_a
self.symbol_b = symbol_b
self.symbols = (symbol_a, symbol_b)
self.hedge_ratio = float(hedge_ratio)
self.distance = float(distance)
self.value = 0.0
self.spread = 0.0
self.z_score = 0.0
self.half_life = float(half_life)
self.time = datetime.min
self.current = IndicatorDataPoint(self.time, self.value)
# Store the formation-period normalizers and spread moments.
self._price_scale_by_symbol = {symbol_a: float(price_scale_a), symbol_b: float(price_scale_b)}
self._spread_mean = float(spread_mean)
self._spread_sd = float(spread_sd)
# Cache the latest price and timestamp for each leg.
self._prices = {}
self._last_time = None
self._is_ready = False
def update(self, input_: BaseData):
price = float(input_.value)
if input_.symbol not in self.symbols or price <= 0:
return self.is_ready
self._prices[input_.symbol] = (input_.end_time, price)
if self.symbol_a not in self._prices or self.symbol_b not in self._prices:
return self.is_ready
time_a, price_a = self._prices[self.symbol_a]
time_b, price_b = self._prices[self.symbol_b]
if time_a != time_b or time_a == self._last_time:
return self.is_ready
self._last_time = time_a
self.time = time_a
# Calculate the normalized-price spread using the selected hedge ratio.
self.value = float(price_a / self._price_scale_by_symbol[self.symbol_a] - self.hedge_ratio * price_b / self._price_scale_by_symbol[self.symbol_b])
self.spread = self.value
self.z_score = float((self.value - self._spread_mean) / self._spread_sd)
self.current = IndicatorDataPoint(self.time, self.value)
self._is_ready = True
return self.is_ready
@property
def is_ready(self) -> bool:
return self._is_ready
from AlgorithmImports import *
from pair_spread import PairSpread
from statsmodels.tsa.stattools import adfuller
class PairStatArb:
# Initialize the strategy parameters for graph clustering and statistical arbitrage lookbacks.
def __init__(self, laplacian_lookback: int):
self._laplacian_lookback = laplacian_lookback
def train(self, algorithm, universe):
ready_securities = sorted([algorithm.securities[s] for s in universe.selected if algorithm.securities[s].log_return.window.is_ready], key=lambda security: security.symbol.value)
# Ensure there are at least two valid assets available to form pairs.
if len(ready_securities) < 2:
return None
# Build the log-return matrix from each indicator's own window.
df_returns = pd.DataFrame({security.symbol.value: [float(point.value) for point in security.log_return.window] for security in ready_securities})
# Build the weighted correlation graph and its Laplacian.
weights = np.abs(df_returns.corr().fillna(0).values)
np.fill_diagonal(weights, 0)
laplacian = np.diag(np.sum(weights, axis=1)) - weights
# Compute the eigenvalues and eigenvectors to find the Fiedler vector for spectral clustering.
_, eigenvectors = np.linalg.eigh(laplacian)
fiedler_vector = eigenvectors[:, 1]
# Standardize the Fiedler vector orientation based on its maximum absolute value.
if fiedler_vector[np.argmax(np.abs(fiedler_vector))] < 0:
fiedler_vector = -fiedler_vector
# Search both spectral partitions for the closest normalized-price pair.
clusters = [[ready_securities[i].symbol for i in range(len(ready_securities)) if fiedler_vector[i] <= 0], [ready_securities[i].symbol for i in range(len(ready_securities)) if fiedler_vector[i] > 0]]
# Fetch the historical close prices for the identified candidates.
history = algorithm.history([security.symbol for security in ready_securities], self._laplacian_lookback, Resolution.DAILY)
if history.empty:
return None
prices = history["close"].unstack(level=0)
best_pair = None
best_mse = np.inf
# Iterate over all asset combinations within each partition to locate the best pair.
for cluster in clusters:
for i in range(len(cluster)):
for j in range(i + 1, len(cluster)):
pair_spread = self._try_create_pair(algorithm, prices, cluster[i], cluster[j])
if pair_spread and pair_spread.distance < best_mse:
best_pair = pair_spread
best_mse = pair_spread.distance
return best_pair
def _try_create_pair(self, algorithm, prices, asset_a, asset_b):
if asset_a not in prices or asset_b not in prices:
return None
aligned = pd.concat([prices[asset_a], prices[asset_b]], axis=1).dropna()
if len(aligned) < self._laplacian_lookback or (aligned <= 0).values.any():
return None
normalized = aligned / aligned.iloc[0]
best_pair = None
best_adf = np.inf
# Use normalized price distance for screening.
mse = float(np.mean(np.square(normalized.iloc[:, 0] - normalized.iloc[:, 1])))
for dependent_index, independent_index in ((0, 1), (1, 0)):
dependent = normalized.iloc[:, dependent_index]
independent = normalized.iloc[:, independent_index]
if np.all(independent == independent.iloc[0]):
continue
hedge_ratio = float(np.polyfit(independent, dependent, 1)[0])
spread = dependent - hedge_ratio * independent
if np.all(spread == spread.iloc[0]):
continue
spread_values = np.array(spread, dtype=float)
if not np.isfinite(spread_values).all():
continue
spread_sd = float(np.std(spread_values))
if spread_sd <= 0:
continue
lagged = spread_values[:-1]
if len(lagged) <= 1 or np.all(lagged == lagged[0]):
continue
slope = float(np.polyfit(lagged, spread_values[1:] - lagged, 1)[0])
if not np.isfinite(slope) or slope >= 0:
continue
adf = float(adfuller(spread)[0])
if adf < best_adf:
if dependent_index == 0:
best_pair = PairSpread(asset_a, asset_b, hedge_ratio, float(aligned.iloc[0, 0]), float(aligned.iloc[0, 1]), float(np.mean(spread_values)), spread_sd, float(-np.log(2) / slope), mse)
else:
best_pair = PairSpread(asset_b, asset_a, hedge_ratio, float(aligned.iloc[0, 1]), float(aligned.iloc[0, 0]), float(np.mean(spread_values)), spread_sd, float(-np.log(2) / slope), mse)
best_adf = adf
if best_pair is None:
return None
# Fetch historical bar data to seed the active pair price cache.
pair_history = algorithm.history[TradeBar]([best_pair.symbol_a, best_pair.symbol_b], 2, Resolution.DAILY)
for bars in pair_history:
if best_pair.symbol_a in bars and best_pair.symbol_b in bars:
best_pair.update(bars[best_pair.symbol_a])
best_pair.update(bars[best_pair.symbol_b])
return best_pair