# https://quantpedia.com/strategies/idiosyncratic-earnings-hedge-strategy/
# 
# The investment universe comprises common shares listed on one of the three major U.S. stock exchanges (NYSE, AMEX, or NASDAQ).
# (Data to estimate the market, industry, and firm-idiosyncratic components of profitability are collected from Compustat.)
# Calculation & Computation: The primary tools used in this strategy are the disaggregation of earnings into market, industry, and idiosyncratic components 
# and the calculation of IdiosROA. The methodology involves collecting earnings data, disaggregating it using the approach from Jackson et al. (2018), and 
# ranking firms based on IdiosROA.
# Rationale: Individual instruments are selected based on their idiosyncratic earnings, derived by disaggregating total earnings into market, industry, and 
# firm-specific components. Firms are then ranked annually based on the signed magnitude of their idiosyncratic Return on Assets (IdiosROA).
# Preparation & Sorting: Portfolios are formed annually by assigning firms to deciles based on the signed magnitude of the firm’s idiosyncratic earnings in year t.
# Execution: The hedge portfolio then consists of a long position in the highest-ranked portfolio and an offsetting short position in the lowest-ranked portfolio, 
# with the provided in parentheses. The top decile (10%) of firms with the highest IdiosROA is selected for long positions, while the bottom decile (10%) of firms
# with the lowest IdiosROA is selected for short positions. The buy rule is to take long positions in firms in the highest decile of IdiosROA. The sell rule is to 
# take short positions in firms in the lowest decile of IdiosROA.
# Risk Management and Positioning plus Rebalancing: The hedge portfolio is rebalanced annually at the start of the fourth month after the fiscal year-end to ensure
# all necessary information is available. The portfolio consists of an equal number of long and short positions, each equally weighted.
# 
# QC implementation changes:
#   - The investment universe consists of 3000 largest from NYSE, AMEX and NASDAQ.
#   - EPS is used as observed fundamental.

# region imports
from AlgorithmImports import *
from numpy import isnan
from functools import reduce
import statsmodels.api as sm
# endregion

class IdiosyncraticEarningsHedgeStrategy(QCAlgorithm):

    def initialize(self) -> None:
        self.set_start_date(2000, 1, 1)
        self.set_cash(100_000)

        self.tickers_to_ignore: List[str] = ['MVIS', 'BTX']
        self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']	
        fundamentals: List[str] = ['operation_ratios.roa.one_year', 'earning_reports.basic_eps.twelve_months']
        self.observed_fundamental: str = fundamentals[1]

        leverage: int = 20
        self.quantile: int = 10
        self.period: int = 10
        self.selection_month: int = 5

        self.data: Dict[Fundamental, RollingWindow] = {}
        self.long: List[Symbol] = []
        self.short: List[Symbol] = []

        market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol

        self.fundamental_count: int = 3_000
        self.fundamental_sorting_key = lambda x: x.market_cap
        self.selection_flag: bool = False
        self.universe_settings.leverage = leverage
        self.universe_settings.resolution = Resolution.DAILY
        self.add_universe(self.fundamental_selection_function)
        self.settings.daily_precise_end_time = False
        self.settings.minimum_order_margin_portfolio_percentage = 0.

        self.schedule.on(self.date_rules.month_start(market),
                        self.time_rules.after_market_open(market),
                        self.selection)

    def on_securities_changed(self, changes: SecurityChanges) -> None:
        for security in changes.added_securities:
            security.set_fee_model(CustomFeeModel())

    def fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]:
        # monthly selection
        if not self.selection_flag:
            return Universe.UNCHANGED

        for stock in fundamental:
            symbol: Symbol = stock.symbol
            if symbol in self.data:
                self.data[symbol].add(self.rgetattr(stock, self.observed_fundamental))

        selected: List[Fundamental] = [
            x for x in fundamental 
            if x.has_fundamental_data
            and x.market_cap != 0
            and x.market == 'usa'
            and not isnan(self.rgetattr(x, self.observed_fundamental)) and self.rgetattr(x, self.observed_fundamental) != 0
            and not isnan(x.asset_classification.morningstar_industry_group_code) and x.asset_classification.morningstar_industry_group_code != 0
            and x.security_reference.exchange_id in self.exchange_codes
            and x.symbol.value not in self.tickers_to_ignore
        ]
        
        if len(selected) > self.fundamental_count:
            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]

        # store stocks by industry code
        industries: Set[int] = set([x.asset_classification.morningstar_industry_group_code for x in selected])
        grouped_industries: Dict[int, List[Fundamental]] = {
            industry : [
                stock for stock in selected if stock.asset_classification.morningstar_industry_group_code == industry
            ] for industry in industries
        }

        symbols_for_regression: Dict[Fundamental, List[float]] = {}

        for stock in selected:
            symbol: Symbol = stock.symbol

            if symbol not in self.data:
                self.data[symbol] = RollingWindow[float](self.period)

            if self.data[symbol].is_ready:
                symbols_for_regression[stock] = self.data[symbol]

        if len(symbols_for_regression) == 0:
            return Universe.UNCHANGED

        residuals: Dict[Symbol, float] = {}

        # regression
        for stock, roll_window in symbols_for_regression.items():
            values: List[float] = list(roll_window)[::-1]
            market_values: List[float] = [sum(val) for val in zip(*[lst for x, lst in symbols_for_regression.items()])]
            industry_values: List[float] = [sum(val) for val in zip(*[lst for x, lst in symbols_for_regression.items() if x in grouped_industries[stock.asset_classification.morningstar_industry_group_code]])]
            x: List[Tuple[float]] = list(zip(market_values[:-1], industry_values[:-1]))
            y: List[float] = values[1:]

            model: RegressionResultWrapper = self.multiple_linear_regression(x, y)
            residuals[stock.symbol] = model.resid[-1]

        # sort and divide
        if len(residuals) > self.quantile:
            sorted_residuals: List[Symbol] = sorted(residuals, key=residuals.get, reverse=True)
            quantile: int = int(len(sorted_residuals) / self.quantile)
            self.long = sorted_residuals[:quantile]
            self.short = sorted_residuals[-quantile:]

        return self.long + self.short

    def on_data(self, slice: Slice) -> None:
        # order execution
        if not self.selection_flag:
            return
        self.selection_flag = False

        targets: List[PortfolioTarget] = []
        for i, portfolio in enumerate([self.long, self.short]):
            for symbol in portfolio:
                if slice.contains_key(symbol) and slice[symbol]:
                    targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
        
        self.SetHoldings(targets, True)
        self.long.clear()
        self.short.clear()

    def selection(self) -> None:
        if self.time.month == self.selection_month:
            self.selection_flag = True

    def rgetattr(self, obj, attr, *args):
        def _getattr(obj, attr):
            return getattr(obj, attr, *args)
        return reduce(_getattr, [obj] + attr.split('.'))

    def multiple_linear_regression(self, x: np.ndarray, y: np.ndarray):
        x = sm.add_constant(x, has_constant='add')
        result = sm.OLS(endog=y, exog=x).fit()
        return result

# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
        fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))