Overall Statistics
Total Trades
95
Average Win
1.70%
Average Loss
-1.87%
Compounding Annual Return
-3.827%
Drawdown
15.900%
Expectancy
-0.033
Net Profit
-6.893%
Sharpe Ratio
-0.077
Probabilistic Sharpe Ratio
3.988%
Loss Rate
49%
Win Rate
51%
Profit-Loss Ratio
0.91
Alpha
-0.006
Beta
-0.391
Annual Standard Deviation
0.168
Annual Variance
0.028
Information Ratio
-0.112
Tracking Error
0.281
Treynor Ratio
0.033
Total Fees
$207.01
Estimated Strategy Capacity
$1200000000.00
Lowest Capacity Asset
SPY R735QTJ8XC9X
Portfolio Turnover
24.05%
# region imports
from AlgorithmImports import *
# endregion

class HipsterFluorescentPinkMule(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2021, 9, 1)  # Set Start Date
        self.SetEndDate(2023, 7, 1)
        self.SetCash(100000)  # Set Strategy Cash
        self.symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
        self.dataset_symbol = self.AddData(RegalyticsRegulatoryArticles, "REG").Symbol
        self.lookback_period = timedelta(days=90)
        
        self.Train(self.DateRules.Every(DayOfWeek.Sunday), self.TimeRules.At(0,0), self.train_model)
        self.train_model()


        #self.SetBenchmark(self.symbol)

    def train_model(self):
        spy_history = self.History(self.symbol, self.lookback_period).loc[self.symbol]
        daily_returns = spy_history['open'].pct_change()[1:]

        reg_history = self.History[RegalyticsRegulatoryArticles](self.dataset_symbol, self.lookback_period)

        returns_by_alert_type = {}
        for articles in reg_history:
            for article in articles:
                future_spy_returns = daily_returns.loc[daily_returns.index > article.Time] # Is this time the same time we get in backtests?
                if len(future_spy_returns) < 2:
                    continue
                if article.AlertType not in returns_by_alert_type:
                    returns_by_alert_type[article.AlertType] = []
                returns_by_alert_type[article.AlertType].append(future_spy_returns.iloc[1])
                
        self.expected_return_by_alert_type = {
            alert_type: sum(future_returns)/len(future_returns) 
                for alert_type, future_returns in returns_by_alert_type.items()
        }


    def OnData(self, slice: Slice) -> None:
        # Parse articles
        if not slice.ContainsKey(self.dataset_symbol):
            return
        articles = slice[self.dataset_symbol]
        expected_returns = []
        alert_types = set()
        for article in articles:
            alert_types.add(article.AlertType)
        for alert_type in alert_types:
            if alert_type in self.expected_return_by_alert_type:
                expected_returns.append(self.expected_return_by_alert_type[alert_type])
        if len(expected_returns) == 0:
            return
        expected_return = sum(expected_returns) / len(expected_returns)
        if expected_return > 0 and not self.Portfolio[self.symbol].IsLong:
            self.SetHoldings(self.symbol, 1)
        elif expected_return < 0 and not self.Portfolio[self.symbol].IsShort:
            self.SetHoldings(self.symbol, -1)