Coverage for src/scrilla/analysis/objects/portfolio.py: 70%

1# This file is part of scrilla: https://github.com/chinchalinchin/scrilla. 

2 

3# scrilla is free software: you can redistribute it and/or modify 

4# it under the terms of the GNU General Public License version 3 

5# as published by the Free Software Foundation. 

6 

7# scrilla is distributed in the hope that it will be useful, 

8# but WITHOUT ANY WARRANTY; without even the implied warranty of 

9# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

10# GNU General Public License for more details. 

11 

12# You should have received a copy of the GNU General Public License 

13# along with scrilla. If not, see <https://www.gnu.org/licenses/> 

14# or <https://github.com/chinchalinchin/scrilla/blob/develop/main/LICENSE>. 

15 

16from datetime import date 

17from math import trunc, sqrt 

18from decimal import Decimal 

19from itertools import groupby 

20from typing import Callable, Dict, List, Union 

21 

22# TODO: get rid of numpy functions. 

23# dot, multiply and transpose should be easy to replicate 

24# and it removes a big dependency from the package... 

25from numpy import dot, multiply, transpose 

26 

27from scrilla import settings 

28from scrilla.static import keys 

29from scrilla.util import errors, outputter 

30# TODO: conditional import module based on analysis_mode, i.e. geometric versus mean reverting. 

31from scrilla.analysis.models.geometric.statistics import calculate_risk_return, correlation_matrix 

32from scrilla.analysis.models.geometric.probability import percentile, conditional_expected_value 

33 

34logger = outputter.Logger( 

35 "scrilla.analysis.objects.portfolio", settings.LOG_LEVEL) 

36 

37# TODO: allow user to specify bounds for equities, i.e. min and max allocations. 

38 

39 

40class Portfolio: 

41 r""" 

42 A class that represents a portfolio of assets defined by the supplied list of ticker symbols in the `tickers` array. 

43 

44 The portfolio can be initialized with historical prices using the `start_date` and `end_date` parameters or the `sample_prices` parameter. If `start_date` and `end_date` are provided, the class will pass the dates to the PriceManager to query an external service for the required prices. If `sample_prices` is provided, the `start_date` and `end_date` are ignored and the `sample_prices` are used in lieu of an external query. 

45 

46 The `return_function` and `volatility_function` methods accept an allocation of percentage weights corresponding to each ticker in the `tickers` array and return the overall portfolio return and volatility. The return is the dot product of the weight and the individual asset returns. The `volatility_function` is the result of applying matrix multiplication to the transposed weight allocations, the correlation matrix and the untransposed weight allocations. These formulations are consistent with Modern Portfolio Theory. 

47 

48 Parameters 

49 ---------- 

50 1. **tickers**: ``List[str]`` 

51 An array of ticker symbols that decribe the assets in a portfolio. 

52 2. **start_date**: ``Union[date, None]`` 

53 *Optional*. The start date for the range of historical prices over which the portfolio will be optimized.  

54 3. **end_date**: ``Union[date, None]`` 

55 *Optional*. The end date for the range of historical prices over which the portfolio will be optimized. 

56 4. **sample_prices**: ``Union[Dict[str, Dict[str, float]], None]`` 

57 *Optional*. A list representing a sample of historical data over a time range. The list must be ordered in descending order, i.e. from latest to earliest. Must be formatted as: `{ 'ticker_1': { 'date' : { 'open': value, 'close': value},... }} 

58 5. **risk_profile** : ``Union[Dict[str, Dict[str, float]], None]`` 

59 Optional: Rather than use sample statistics calculated from historical data, this argument can override the calculated values. Must be formatted as: `{ ticker: { 'annual_return': float, 'annual_volatility': float }}` 

60 6. **correlation_matrix**: ``Union[List[List[float]], None]`` 

61 Optional: Rather than use correlations calculated from historical data, this argument can override the calculated vlaues. 

62 7. **asset_return_functions**: ``Union[List[Callable], None]`` 

63 *Optional*. An array of function that describes the expected logarithmic rate of return of each asset in the portfolio with respect to time. The order between `asset_return_functions` and `tickers` be must be preserved, i.e. the index of tickers must correspond to the symbol described by the function with same index in `asset_return_functions`. 

64 8. **asset_volatility_funtions**: ``Union[List[Callable], None]`` 

65 *Optional*. An array of functions that describe the mean volatility of each asset in the portfolio with respect to time. The order between `asset_volatility_functions` and `tickers` be must be preserved, i.e. the index of tickers must correspond to the symbol described by the function with the same index in `asset_volatility_functions`. 

66 

67 Attributes 

68 ---------- 

69 All parameters are exposed as properties on the class. With the exception of `asset_return_functions` and `asset_volatility_functions`, if optional parameters are not provided in the class constructor, they will be calculated based on available information. See *notes* for more details on how this class is constructed. 

70 

71 .. notes:: 

72 * While `start_date`, `end_date`, `sample_prices` are all by themselves optional, the `scrilla.analysis.objects.Portfolio` class must be initialized in one of three ways:  

73 1. *Portfolio*(`start_date`, `end_date`) -> `start_date` and `end_date` are passed to service for external query request. 

74 2. *Portfolio*(`sample_prices`) -> `start_date` and `end_date` are ignored and `sample_prices` are used for statistical calculations. 

75 3. *Portfolio*(`risk_profile`) -> `start_date`, `end_date` and `sample_prices` are ignored and the statistics in `risk_profile` are used istead of manual calculations. 

76 *The priority hierarchy is as follows : `risk_profile` > `sample_prices` > (`start_date`, `end_date`). If no arguments are provided to the constructor at all, the portfolio will default to the `scrilla.settings.DEFAULT_ANALYSIS_PERIOD` variable configured by the corresponding environment variable. If the environment variable is not set, this value will default to the last 100 trading days. 

77 * The `asset_return_functions` and `asset_volatility_functions` can be understood as the drift and noise functions for a random stochastic process, 

78 

79 $$ \frac{dX(t)}{X(t)} = \mu(t) \cdot dt + \sigma(t) \cdot dB(t) $$ 

80 

81 where B(t) ~ \\(N(0, \Delta \cdot t)\\). 

82 """ 

83 

84 def __init__(self, tickers: List[str], start_date=Union[date, None], end_date=Union[date, None], sample_prices: Union[Dict[str, Dict[str, float]], None] = None, correl_matrix: Union[List[List[int]], None] = None, risk_profiles: Union[Dict[str, Dict[str, float]]] = None, risk_free_rate: Union[float, None] = None, asset_return_functions: Union[List[Callable], None] = None, asset_volatility_functions: Union[List[Callable], None] = None, method: str = settings.ESTIMATION_METHOD): 

85 self.estimation_method = method 

86 self.sample_prices = sample_prices 

87 self.tickers = tickers 

88 self.correl_matrix = correl_matrix 

89 self.asset_volatility_functions = asset_volatility_functions 

90 self.asset_return_functions = asset_return_functions 

91 self.risk_profiles = risk_profiles 

92 self.target_return = None 

93 

94 if self.sample_prices is None: 94 ↛ 98line 94 didn't jump to line 98, because the condition on line 94 was never false

95 self.start_date = start_date 

96 self.end_date = end_date 

97 else: 

98 self.start_date = list(self.sample_prices.keys())[-1] 

99 self.end_date = list(self.sample_prices.keys())[0] 

100 

101 if risk_free_rate is not None: 101 ↛ 102line 101 didn't jump to line 102, because the condition on line 101 was never true

102 self.risk_free_rate = risk_free_rate 

103 else: 

104 self.risk_free_rate = 0 

105 

106 self._init_asset_types() 

107 self._init_dates() 

108 self._init_stats() 

109 

110 def _init_asset_types(self): 

111 self.asset_types = [] 

112 for ticker in self.tickers: 

113 self.asset_types.append(errors.validate_asset_type(ticker)) 

114 

115 self.asset_groups = 0 

116 for _ in groupby(sorted(self.asset_types)): 

117 self.asset_groups += 1 

118 

119 def _init_dates(self): 

120 if self.asset_groups == 1 and self.asset_types[0] == keys.keys['ASSETS']['CRYPTO']: 

121 self.start_date, self.end_date = errors.validate_dates(self.start_date, 

122 self.end_date, 

123 keys.keys['ASSETS']['CRYPTO']) 

124 self.weekends = 1 

125 else: 

126 self.start_date, self.end_date = errors.validate_dates(self.start_date, 

127 self.end_date, 

128 keys.keys['ASSETS']['EQUITY']) 

129 self.weekends = 0 

130 

131 def _init_stats(self): 

132 self.mean_return = [] 

133 self.sample_vol = [] 

134 

135 # priority hierarchy: asset_functions -> risk_profiles -> sample_prices -> statistics.py calls 

136 if self.asset_volatility_functions is not None and self.asset_return_functions is not None: 136 ↛ 143line 136 didn't jump to line 143, because the condition on line 136 was never true

137 # TODO: implement ito integration and calculate asset return and volatilities! 

138 # use return and volatility functions to integrate over time period [0, infinity] for each asset. don't forget to 

139 # discount! I(x) = discounted expected payoff 

140 # Integral(d ln S) = Integral(Mean dt) + Integral(Vol dZ) 

141 # Need methods to compute ito Integrals in...statistics.py? markets.py? Perhaps a new module. 

142 # https://math.stackexchange.com/questions/1780956/mean-and-variance-geometric-brownian-motion-with-not-constant-drift-and-volatili 

143 pass 

144 

145 else: 

146 

147 # TODO: there is a logical error here. if the portfolio is made up of mixed assets (crypto, equity), 

148 # then calculate_risk_return will calculate the risk profile over a different time period than 

149 # the correlation matrix. the reason is: risk_return is univariate, but correlation is bivariate, 

150 # so when the correlation of an equity and crypto is calculated, it truncates the sample to dates 

151 # where both assets trade, i.e. crypto prices on weekends get ignored. the risk_profile of the crypto 

152 # will be over a shorter date range because the analysis will include weekends, whereas the crypto 

153 # correlation will not include weekends if the asset types are mixed. the problem is further 

154 # compounded since the correlation method will retrieve the univariate profile to use in its calculation. 

155 # need a flag in the cache to tell the program the statistic includes/exclude weekend prices. 

156 

157 if self.risk_profiles is None: 157 ↛ 174line 157 didn't jump to line 174, because the condition on line 157 was never false

158 for ticker in self.tickers: 

159 if self.sample_prices is not None: 159 ↛ 160line 159 didn't jump to line 160, because the condition on line 159 was never true

160 stats = calculate_risk_return(ticker=ticker, 

161 sample_prices=self.sample_prices[ticker], 

162 method=self.estimation_method, 

163 weekends=self.weekends) 

164 else: 

165 stats = calculate_risk_return(ticker=ticker, 

166 start_date=self.start_date, 

167 end_date=self.end_date, 

168 method=self.estimation_method, 

169 weekends=self.weekends) 

170 

171 self.mean_return.append(stats['annual_return']) 

172 self.sample_vol.append(stats['annual_volatility']) 

173 else: 

174 for ticker in self.risk_profiles: 

175 self.mean_return.append( 

176 self.risk_profiles[ticker]['annual_return']) 

177 self.sample_vol.append( 

178 self.risk_profiles[ticker]['annual_volatility']) 

179 

180 if self.correl_matrix is None: 180 ↛ exitline 180 didn't return from function '_init_stats', because the condition on line 180 was never false

181 self.correl_matrix = correlation_matrix(tickers=self.tickers, 

182 start_date=self.start_date, 

183 end_date=self.end_date, 

184 sample_prices=self.sample_prices, 

185 weekends=self.weekends, 

186 method=self.estimation_method) 

187 

188 def return_function(self, x): 

189 """ 

190 Returns the portfolio return for a vector of allocations. It can be used as objective function input for `scipy.optimize`'s optimization methods.  

191 

192 Parameters 

193 ---------- 

194 1. **x**: ``list`` 

195 Vector representing the allocation of each asset in the portfolio. Must be preserve the order of `portfolio.tickers`, i.e. each element's index should map to each element of the `portfolio.tickers`'s list. 

196 

197 Returns 

198 ------- 

199 ``float`` 

200 The portfolio return on an annualized basis. 

201 """ 

202 return dot(x, self.mean_return) 

203 

204 def volatility_function(self, x): 

205 """ 

206 Returns the portfolio volatility for a vector of allocations. This function can be used as objective input function for `scipy.optimize`'s optimization or solver methods.\n\n 

207 

208 Parameters 

209 ---------- 

210 1. **x**: ``list`` 

211 Vector representing the allocation of each asset in the portfolio. Must be preserve the order of `portfolio.tickers`, i.e. each element's index should map to each element of the `portfolio.tickers`'s list. 

212 

213 Returns 

214 ------- 

215 ``float`` 

216 The portfolio volatility on an annualized basis. 

217 """ 

218 return sqrt(multiply(x, self.sample_vol).dot(self.correl_matrix).dot(transpose(multiply(x, self.sample_vol)))) 

219 

220 def sharpe_ratio_function(self, x): 

221 """ 

222 Returns the portfolio sharpe ratio for a vector of allocations. This function can be used as objective input function for `scipy.optimize`'s optimization or solver methods.\n\n 

223 

224 Parameters 

225 ---------- 

226 1. **x**: ``list`` 

227 Vector representing the allocation of each asset in the portfolio. Must be preserve the order of `portfolio.tickers`, i.e. each element's index should map to each element of the `portfolio.tickers`'s list. 

228 

229 Returns 

230 ------- 

231 ``float`` 

232 The portfolio sharpe ratio on an annualized basis. 

233 """ 

234 return (dot(x, self.mean_return) - self.risk_free_rate) / (self.volatility_function(x)) 

235 

236 def percentile_function(self, x, time, prob): 

237 """ 

238 Returns the given percentile of the portfolio's assumed distribution.\n\n 

239 

240 Parameters 

241 ---------- 

242 1. **x**: ``list`` 

243 Vector representing the allocation of each asset in the portfolio. Must be preserve the order of `self.tickers`, i.e. each element's index should map to each element of the `self.tickers`'s list. 

244 2: **time**: ``float`` 

245 time horizon (in years) of the value at risk, i.e. the period of time into the future at which the value at risk is being calculated  

246 3. **prob**: ``float`` 

247 percentile desired 

248 """ 

249 portfolio_return = self.return_function(x) * time 

250 portfolio_volatility = self.volatility_function(x) * sqrt(time) 

251 

252 return percentile(S0=1, vol=portfolio_volatility, ret=portfolio_return, 

253 expiry=time, prob=prob) 

254 

255 def conditional_value_at_risk_function(self, x, time, prob): 

256 """ 

257 Calculates the conditional value at risk for a portfolio of stocks over a specified time horizon. The value will be given in percentage terms relative to the initial value of the portfolio at the beginning of the time horizon, i.e. a return value of 5% would mean 5% of your portfolio's initial value is at risk with probability `prob`. A negative value would indicate there is no value at risk, i.e. value would actually accrue. This function can be used as objective input function for `scipy.optimize`'s optimization or solver methods. 

258 

259 Parameters 

260 ---------- 

261 1. **x**: ``list`` 

262 an array of decimals representing percentage allocations of the portfolio. Must preserve order with `self.tickers`. 

263 2. **time**: ``float`` 

264 time horizon (in years) of the value at risk, i.e. the period of time into the future at which the value at risk is being calculated. 

265 3. **prob**: ``float`` 

266 desired probability of loss. 

267 """ 

268 portfolio_return = self.return_function(x) * time 

269 portfolio_volatility = self.volatility_function(x) * sqrt(time) 

270 value_at_risk = self.percentile_function(x=x, time=time, prob=prob) 

271 return (1 - conditional_expected_value(S0=1, vol=portfolio_volatility, ret=portfolio_return, 

272 expiry=time, conditional_value=value_at_risk)) 

273 

274 def get_init_guess(self): 

275 length = len(self.tickers) 

276 uniform_guess = 1/length 

277 guess = [uniform_guess for i in range(length)] 

278 return guess 

279 

280 @staticmethod 

281 def get_constraint(x): 

282 return sum(x) - 1 

283 

284 def get_default_bounds(self): 

285 return [[0, 1] for y in range(len(self.tickers))] 

286 

287 def set_target_return(self, target): 

288 self.target_return = target 

289 

290 def get_target_return_constraint(self, x): 

291 if self.target_return is not None: 291 ↛ 293line 291 didn't jump to line 293, because the condition on line 291 was never false

292 return (dot(x, self.mean_return) - self.target_return) 

293 return None 

294 

295 @staticmethod 

296 def calculate_approximate_shares(x, total, latest_prices: Dict[str, float]): 

297 """ 

298 

299 Parameters 

300 ---------- 

301 1. **x**:  

302 2. **total**: 

303 3. **latest_prices**: ``Dict[str, float]``.  

304 Dictionary of latest asset prices. Must preserve order with the allocation **x**, i.e. the *i*-th element of **x** must correspond to the same asset as the *i*-th element of the **latest_prices**. 

305 """ 

306 shares = [] 

307 for i, item in enumerate(x): 

308 price = list(latest_prices.values())[i] 

309 share = Decimal(item) * Decimal(total) / Decimal(price) 

310 shares.append(trunc(share)) 

311 return shares 

312 

313 @staticmethod 

314 def calculate_actual_total(x, total, latest_prices): 

315 actual_total = 0 

316 shares = Portfolio.calculate_approximate_shares( 

317 x=x, total=total, latest_prices=latest_prices) 

318 for i, item in enumerate(shares): 

319 price = list(latest_prices.values())[i] 

320 portion = Decimal(item) * Decimal(price) 

321 actual_total = actual_total + portion 

322 return actual_total