Overall Statistics Total Trades 25 Average Win 2.33% Average Loss -4.62% Compounding Annual Return 6.525% Drawdown 50.300% Expectancy -0.179 Net Profit 346.918% Sharpe Ratio 0.386 Probabilistic Sharpe Ratio 0.014% Loss Rate 45% Win Rate 55% Profit-Loss Ratio 0.50 Alpha 0.008 Beta 0.793 Annual Standard Deviation 0.144 Annual Variance 0.021 Information Ratio -0.061 Tracking Error 0.074 Treynor Ratio 0.07 Total Fees \$105.24 Estimated Strategy Capacity \$740000000.00 Lowest Capacity Asset SPY R735QTJ8XC9X Portfolio Turnover 0.20%
```# https://quantpedia.com/strategies/fed-model/
#
# Each month, the investor conducts a one-month predictive regression (using all available data up to that date) predicting excess stock market
# returns using the yield gap as an independent variable. The “Yield gap” is calculated as YG = EY − y, with earnings yield EY ≡ ln (1 ++ E/P)
# and y = ln (1 ++ Y) is the log 10 year Treasury bond yield. Then, the strategy allocates 100% in the risky asset if the forecasted excess
# returns are positive, and otherwise, it invests 100% in the risk-free rate.

from collections import deque
from AlgorithmImports import *
from typing import List, Tuple, Deque
import numpy as np
from scipy import stats

class FEDModel(QCAlgorithm):

def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)

self.period:int = 12 * 21
self.max_missing_days:int = 31
self.SetWarmUp(self.period)

self.market_data:Deque[Tuple[float,float]] = deque()

# risk free rate

# 10Y bond yield symbol

# SP500 earnings yield data

self.yield_gap:Deque[float] = deque()

self.recent_month:int = -1

def OnData(self, data:Slice) -> None:
rebalance_flag:bool = False

if self.sp_earnings_yield in data and data[self.sp_earnings_yield]:
if self.Time.month != self.recent_month:
self.recent_month = self.Time.month
rebalance_flag = True

if not rebalance_flag:
# earnings yield data is no longer comming in
if self.Securities[self.sp_earnings_yield].GetLastData():
if (self.Time.date() - self.Securities[self.sp_earnings_yield].GetLastData().Time.date()).days > self.max_missing_days:
self.Liquidate()
return

# update market price data
if self.market in data and self.bond_yield in data:
if data[self.market] and data[self.bond_yield] and self.Securities[self.risk_free_rate].GetLastData():
market_price:float = data[self.market].Value
rf_rate:float = self.Securities[self.risk_free_rate].Price / 100
bond_yield:float = data[self.bond_yield].Value / 100
sp_ey:float = data[self.sp_earnings_yield].Value / 100
if market_price != 0 and rf_rate != 0 and bond_yield != 0 and sp_ey != 0:
self.market_data.append((market_price, rf_rate))

yield_gap:float = np.log(sp_ey) - np.log(bond_yield)
self.yield_gap.append(yield_gap)
rebalance_flag = True

# ensure minimum data points to calculate regression
min_count:int = 6
if len(self.market_data) >= min_count:
market_closes:np.ndarray = np.array([x[0] for x in self.market_data])
market_returns:np.ndarray = (market_closes[1:] - market_closes[:-1]) / market_closes[:-1]
rf_rates:np.ndarray = np.array([x[1] for x in self.market_data][1:])
excess_returns:np.ndarray = market_returns - rf_rates

yield_gaps:List[float] = list(self.yield_gap)

# linear regression
# Y = α + (β ∗ X)
# intercept = alpha
# slope = beta
beta, alpha, r_value, p_value, std_err = stats.linregress(yield_gaps[1:-1], market_returns[1:])

# predicted market return
Y:float = alpha + (beta * yield_gaps[-1])

if Y > 0:
if self.Portfolio[self.cash].Invested:
self.Liquidate(self.cash)
self.SetHoldings(self.market, 1)
else:
if self.Portfolio[self.market].Invested:
self.Liquidate(self.market)
self.SetHoldings(self.cash, 1)

# Quantpedia bond yield data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaBondYield(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/bond_yield/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)

def Reader(self, config, line, date, isLiveMode):
data = QuantpediaBondYield()
data.Symbol = config.Symbol

if not line[0].isdigit(): return None
split = line.split(',')

data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
data['yield'] = float(split[1])
data.Value = float(split[1])

return data

# Quandl "value" data
class QuandlValue(PythonQuandl):
def __init__(self):
self.ValueColumnName = 'Value'```