-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathCAPM_fn.py
217 lines (169 loc) · 7.39 KB
/
CAPM_fn.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
import numpy as np
import pandas as pd
import datetime as dt
from pandas_datareader import data as web
import yfinance as yf
import warnings
warnings.filterwarnings('ignore')
# interval = 'm'
# tickers = ['AAPL']
# risk_free = 0.0402
# end = dt.datetime.now()
# start = end - dt.timedelta(weeks=52*5+8)
#start = dt.datetime(2017,4,29)
#end = dt.datetime(2022,8,6)
### Import price data
# Get adjusted close price
# Use pandas_datareader
def get_data(tickers, start, end, interval, OHLC='Adj Close', market=True):
if market == True:
tickers.insert(0,'^GSPC')
# compounding frequency per annum
frequency = {'d':252, 'w':52, 'm': 12}
# monthly frequency
frequency = frequency[interval]
df = pd.DataFrame()
for t in tickers:
df[t] = web.get_data_yahoo(t, start, end, interval=interval)[OHLC]
df.dropna(inplace=True)
return df
# Use yfinance
def get_data_yf(tickers, start, end, interval, OHLC='Adj Close', market=True):
if market == True:
tickers.insert(0,'^GSPC')
# compounding frequency per annum
frequency = {'1d':252, '1wk':52, '1mo': 12}
# monthly frequency
frequency = frequency[interval]
df = pd.DataFrame()
for t in tickers:
df[t] = yf.download(t, start, end, interval=interval)[OHLC]
df.dropna(inplace=True)
return df
# Get total price with OHLC using pandas_datareader
def get_OHLC(tickers, start, end, interval,OHLC='Adj Close'):
tickers.insert(0,'^GSPC')
# compounding frequency per annum
frequency = {'d':252, 'w':52, 'm': 12}
# monthly frequency
frequency = frequency[interval]
open_df = get_data(tickers, start, end, interval,OHLC='Open')
high_df = get_data(tickers, start, end, interval,OHLC='High')
low_df = get_data(tickers, start, end, interval,OHLC='Low')
adjc_df = get_data(tickers, start, end, interval)
OHLC = pd.concat([open_df,
high_df,
low_df,
adjc_df], join="inner").sort_values(by='Date')
return OHLC
# Get total price with OHLC using yfinance
def get_OHLC_yf(tickers, start, end, interval,OHLC='Adj Close', market=True):
if market == True:
tickers.insert(0,'^GSPC')
# compounding frequency per annum
frequency = {'1d':252, '1wk':52, '1mo': 12}
# monthly frequency
frequency = frequency[interval]
open_df = get_data(tickers, start, end, interval,OHLC='Open')
high_df = get_data(tickers, start, end, interval,OHLC='High')
low_df = get_data(tickers, start, end, interval,OHLC='Low')
adjc_df = get_data(tickers, start, end, interval)
OHLC = pd.concat([open_df,
high_df,
low_df,
adjc_df], join="inner").sort_values(by='Date')
return OHLC
### Two returns calculation methods
# Calculate Simple Returns of the stocks
# (Missing first observation)
def simp_ret(stocks_price):
simp_ret = (stocks_price/stocks_price.shift(1) - 1)[1:]
return simp_ret
# Calculate Log Returns of the stocks
def log_ret(stocks_price):
log_return = np.log(stocks_price/stocks_price.shift(1))[1:]
return log_return
### Annualizing returns above
# Annualized simple return
def annual_simpret(simp_return, frequency):
# Gross returns
grossret = simp_return + 1
# Periodic geomean return
geomret = np.prod(grossret)**(1/len(grossret))
# Periodic geomean return compounded to frequency per annum
annual_simpret = geomret**frequency - 1
return annual_simpret
# Annualized log return
def annual_logret(log_return, frequency):
# Arithmetic mean of log returns compounded to annualize
annual_logret = log_return.mean()*frequency
return annual_logret
### Beta of stocks
# risk-free rate of annualized risk-free rate
# Difference from including risk_free is negigible, using risk-free=0 is fine
def get_beta(stock_returns, risk_free=0):
# De-annualize risk-free rate
periodic_rf = (1+risk_free)**(1/frequency)-1
tickers = stock_returns.columns
Beta_df = pd.DataFrame(columns = ['ticker','Beta'])
stock_returns = stock_returns - periodic_rf
for stock in range(len(tickers)):
stock_mkt_cov = stock_returns.cov().loc['^GSPC',tickers[stock]]
mkt_var = stock_returns['^GSPC'].var()
Beta = stock_mkt_cov/mkt_var
Beta_df = Beta_df.append({'ticker':tickers[stock],'Beta':Beta },ignore_index=True)
return Beta_df
# Mkt_ret_simp = annual_simpret(simpret['^GSPC'], frequency)
# actual_simpret = annual_simpret(simpret, frequency)
# risk-free, Mkt, actual stock returns must be annualized returns
def CAPM(beta_df, risk_free, Mkt_ret, actual_ret): # expected returns in annualized terms
stocks = list(beta_df['ticker'])[1:] # tickers except for S&P500 index
ER_df = pd.DataFrame(columns = ['ticker','CAPM ER'])
for stock in stocks:
beta = float(beta_df[beta_df['ticker']==stock]['Beta'])
expected_return = risk_free + beta*(Mkt_ret - risk_free)
actual_return = actual_ret[stock]
return_gap = actual_return - expected_return
ER_df = ER_df.append({'ticker':stock,
'Beta': beta,
'CAPM ER':expected_return,
'Actual Return': actual_return,
'Return Gap': return_gap},
ignore_index=True)
# if CAPM Expected Returns < actual returns, undervalued
ER_df['Undervalued'] = ER_df['CAPM ER'] < ER_df['Actual Return']
ER_df = ER_df.sort_values(by=['Return Gap'], ascending=False)
return ER_df
def rolling_beta(stock_df, ticker, beta_window, ma_window):
ticker = ticker
stock_P = get_data(['^GSPC',ticker], start, end, interval)[:-1]
stock_returns = simp_ret(stock_P[['^GSPC',ticker]]).rolling(beta_window)
stock_mkt_cov = stock_returns.cov().iloc[1::2, :].drop(columns=[ticker])#.loc['^GSPC','AAPL']
stock_mkt_cov.index = stock_mkt_cov.index.droplevel(level=1)
stock_mkt_cov = stock_mkt_cov.rename(columns={'^GSPC':f'{ticker} Beta'})
mkt_var = stock_returns.cov().iloc[::2, :].drop(columns=[ticker])
mkt_var.index =mkt_var.index.droplevel(level=1)
Beta = stock_mkt_cov.iloc[:,0]/mkt_var.iloc[:,0]
Beta_ma = Beta.rolling(ma_window).mean()
return Beta, Beta_ma
# Get Historical Beta, one datapoint per each stock
def get_beta_yf(stock_returns):
tickers = stock_returns.columns
Beta_df = pd.DataFrame(columns = ['ticker','Beta'])
for stock in range(len(tickers)):
stock_mkt_cov = stock_returns.cov().loc['^GSPC',tickers[stock]]
mkt_var = stock_returns['^GSPC'].var()
Beta = stock_mkt_cov/mkt_var
Beta_df = Beta_df.append({'ticker':tickers[stock],'Beta':Beta },ignore_index=True)
return Beta_df
# Get Time series of rolling Betas per each stock
def rolling_beta_yf(stock_df, ticker, beta_window, ma_window): # takes in single ticker at a time
stock_returns = simp_ret(stock_df[['^GSPC',ticker]]).rolling(beta_window)
stock_mkt_cov = stock_returns.cov().iloc[1::2].drop(columns=[ticker]).dropna()
stock_mkt_cov.index = stock_mkt_cov.index.droplevel(level=1)
mkt_var = stock_returns.cov().iloc[::2].drop(columns=[ticker]).dropna()
mkt_var.index =mkt_var.index.droplevel(level=1)
Beta = stock_mkt_cov/mkt_var
Beta = Beta.rename(columns={'^GSPC':ticker})
Beta_ma = Beta.rolling(ma_window).mean().dropna()
return Beta, Beta_ma