Overall Statistics
Total Orders
660
Average Win
0.15%
Average Loss
-0.03%
Compounding Annual Return
28.321%
Drawdown
6.100%
Expectancy
3.529
Start Equity
100000
End Equity
164774.42
Net Profit
64.774%
Sharpe Ratio
1.905
Sortino Ratio
2.006
Probabilistic Sharpe Ratio
90.702%
Loss Rate
17%
Win Rate
83%
Profit-Loss Ratio
4.47
Alpha
0.14
Beta
0.276
Annual Standard Deviation
0.1
Annual Variance
0.01
Information Ratio
0.067
Tracking Error
0.173
Treynor Ratio
0.686
Total Fees
$660.00
Estimated Strategy Capacity
$34000000.00
Lowest Capacity Asset
SXC TAWKXZJ50P5X
Portfolio Turnover
0.40%
# Filename: D_Volatility_Anomaly.py
from AlgorithmImports import *
import numpy as np

class DVolatilityAnomaly(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2020, 1, 1)
        self.SetEndDate(2022, 1, 1)
        self.SetCash(100000)
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddEquity("SPY", Resolution.Daily)
        self.SetBenchmark("SPY")

        self.vol_lookback = 252
        self.selection_count = 200  # 20% of 1000
        self.symbol_vol_dict = {}
        self.current_selection = set()
        self.rebalance_flag = False

        self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)

        self.Schedule.On(self.DateRules.MonthStart(), self.TimeRules.AfterMarketOpen("SPY", 30), self.SetRebalanceFlag)

    def SetRebalanceFlag(self):
        self.rebalance_flag = True

    def CoarseSelectionFunction(self, coarse):
        filtered = [x for x in coarse if x.HasFundamentalData and x.Price > 5 and x.Market == Market.USA]
        top = sorted(filtered, key=lambda x: x.MarketCap, reverse=True)[:1200]
        return [x.Symbol for x in top]

    def FineSelectionFunction(self, fine):
        sorted_fine = sorted(fine, key=lambda x: x.MarketCap, reverse=True)
        return [x.Symbol for x in sorted_fine[:1000]]

    def OnSecuritiesChanged(self, changes):
        for security in changes.RemovedSecurities:
            symbol = security.Symbol
            self.symbol_vol_dict.pop(symbol, None)
            self.current_selection.discard(symbol)

    def OnData(self, data):
        if not self.rebalance_flag:
            return
        self.rebalance_flag = False

        universe = [x for x in self.ActiveSecurities.Keys if x.SecurityType == SecurityType.Equity and x != Symbol.Create("SPY", SecurityType.Equity, Market.USA)]
        if len(universe) < self.selection_count:
            self.Debug(f"Universe too small: {len(universe)}")
            return

        symbol_vols = {}
        for symbol in universe:
            try:
                history = self.History(symbol, self.vol_lookback, Resolution.Daily)
                if history.empty or len(history['close']) < self.vol_lookback:
                    continue
                closes = history['close'].values
                returns = np.diff(np.log(closes))
                if len(returns) < 2:
                    continue
                vol = np.std(returns)
                symbol_vols[symbol] = vol
            except Exception as e:
                self.Debug(f"History error for {symbol}: {e}")
                continue

        if len(symbol_vols) < self.selection_count:
            self.Debug(f"Not enough stocks with volatility data: {len(symbol_vols)}")
            return

        sorted_syms = sorted(symbol_vols.items(), key=lambda x: x[1])
        selected = set([sym for sym, vol in sorted_syms[:self.selection_count]])

        invested = set([x.Symbol for x in self.Portfolio.Values if x.Invested and x.Symbol != Symbol.Create("SPY", SecurityType.Equity, Market.USA)])
        for symbol in invested:
            if symbol not in selected:
                self.Liquidate(symbol, "Not in low volatility quintile")

        weight = 1.0 / self.selection_count
        for symbol in selected:
            if symbol not in data or not data[symbol] or not self.Securities[symbol].IsTradable:
                continue
            self.SetHoldings(symbol, weight)

        self.current_selection = selected

    def OnEndOfDay(self):
        to_remove = set()
        for symbol in self.current_selection:
            if symbol not in self.Securities or not self.Securities[symbol].IsTradable:
                to_remove.add(symbol)
        self.current_selection -= to_remove

    def OnOrderEvent(self, orderEvent):
        if orderEvent.Status == OrderStatus.Filled:
            self.Debug(f"Order filled: {orderEvent.Symbol} {orderEvent.Direction} {orderEvent.FillQuantity} @ {orderEvent.FillPrice}")