Coverage for src/scrilla/main.py: 46%

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 

16""" 

17# CLI Entrypoint  

18This script acts as the entrypoint for the CLI application and contains the majority of the control structures for the program. It parses the arguments supplied through the command line, delegates them to the appropriate application function and then passes the results to the `scrilla.util.outputter` module functions for formatting and printing to screen. 

19 

20The arguments are parsed in such a way that arguments which are not supplied are set to None. All application functions are set up to accept None as a value for their optional arguments. This makes passing arguments to application functions easier as the `main.py` script doesn't have to worry about their values. In other words, `main.py` always passes all arguments to application functions, even if they aren't supplied through the command line; it just sets the ones which aren't supplied to None. 

21""" 

22 

23import time 

24from datetime import date 

25from typing import Callable, Dict, List, Union 

26 

27from scrilla import settings, files, cache 

28from scrilla.static import definitions 

29from scrilla.util.errors import InputValidationError 

30from scrilla.static import formats 

31from scrilla.util.outputter import Logger 

32 

33# TODO: conditional imports based on value of ANALYSIS_MODE 

34 

35logger = Logger('main', settings.LOG_LEVEL) 

36 

37 

38def validate_function_usage(selection: str, args: List[str], wrapper_function: Callable, required_length: int = 1, exact: bool = False) -> None: 

39 """ 

40 Parameters 

41 ---------- 

42 1. **selection** : ``str`` 

43 2. **args**: ``list`` 

44 List of ticker/statistic symbols whose length is be to validated. 

45 3. **wrapper_function** : ``func()`` 

46 4. **required_length** : ``int`` 

47 The number of tickers required for the function. 

48 5. **exact** : ``bool`` 

49 *Optional*. If the required length constraint is an equality, set to `True`. If the constraint is an inequality, set to `False`. Defaults to `False`.  

50 """ 

51 

52 start_time = time.time() 

53 if(not exact and (len(args) > (required_length-1))) or (exact and (len(args) == required_length)): 53 ↛ 55line 53 didn't jump to line 55, because the condition on line 53 was never false

54 wrapper_function() 

55 elif exact: 

56 raise InputValidationError( 

57 f'Invalid number of arguments for \'{selection}\' function. Function requires {required_length} arguments.') 

58 else: 

59 raise InputValidationError( 

60 f'Invalid number of arguments for \'{selection}\' function. Function requires more than {required_length} arguments.') 

61 end_time = time.time() 

62 logger.info( 

63 f'Total execution time: {end_time - start_time}s', 'validate_function_usage') 

64 

65 

66def print_format_to_screen(args: Dict[str, Union[str, date, float, None, bool]]) -> bool: 

67 """ 

68 Checks if the inputted optional arguments allow printing pretty formatted text to screen. 

69 

70 Parameters 

71 ---------- 

72 1. **args** : ``dict`` 

73 Formatted dictionary containing optional arguments. Result of a call to `helper.format_xtra_args_dict`. 

74 

75 Returns 

76 ------- 

77 ``bool`` 

78 """ 

79 return not args['json'] and not args['suppress_output'] 

80 

81 

82def print_json_to_screen(args: Dict[str, Union[str, date, float, None, bool]]) -> bool: 

83 """ 

84 Checks if inputted optional arguments allow printing json formatted text to screen. 

85 

86 Parameters 

87 ---------- 

88 1. **args** : ``dict`` 

89 Formatted dictionary containing optional arguments. Result of a call to `helper.format_xtra_args_dict`. 

90 

91 Returns 

92 ------- 

93 ``bool`` 

94 """ 

95 return args['json'] and not args['suppress_output'] 

96 

97 

98def do_program(cli_args: List[str]) -> None: 

99 """ 

100 Parses command line arguments and passes the formatted arguments to appropriate function from the library. 

101 """ 

102 files.init_static_data() 

103 cache.init_cache() 

104 

105 args = formats.format_args(cli_args, settings.ESTIMATION_METHOD) 

106 exact, selected_function = False, None 

107 

108 # START CLI FUNCTION DEFINITIONS 

109 

110 # NO ARGUMENT FUNCTIONS 

111 # FUNCTION: Help Message 

112 if args['function_arg'] in definitions.FUNC_DICT["help"]['values']: 

113 def cli_help(): 

114 from scrilla.util.outputter import help_msg 

115 # NOTE: in this case, the arguments are function names, not tickers. 

116 # it may be behoove the application to rejigger the argparse 

117 # just so names are consistent with what is represented. 

118 help_msg(function_filter=args['tickers']) 

119 selected_function, required_length = cli_help, 0 

120 

121 # FUNCTION: Clear Cache 

122 elif args['function_arg'] in definitions.FUNC_DICT["clear_cache"]['values']: 

123 def cli_clear_cache(): 

124 logger.info(f'Clearing {settings.CACHE_DIR}', 'do_program') 

125 files.clear_cache() 

126 selected_function, required_length = cli_clear_cache, 0 

127 

128 # FUNCTION: Clear Static 

129 elif args['function_arg'] in definitions.FUNC_DICT["clear_static"]['values']: 

130 def cli_clear_static(): 

131 logger.info(f'Clearing {settings.STATIC_DIR}', 'do_program') 

132 files.clear_directory(directory=settings.STATIC_DIR, retain=True) 

133 selected_function, required_length = cli_clear_static, 0 

134 

135 # FUNCTION: Clear Common 

136 elif args['function_arg'] in definitions.FUNC_DICT["clear_common"]['values']: 

137 def cli_clear_common(): 

138 logger.info(f'Clearing {settings.COMMON_DIR}', 'do_program') 

139 files.clear_directory(directory=settings.COMMON_DIR, retain=True) 

140 selected_function, required_length = cli_clear_common, 0 

141 

142 # FUNCTION: Print Stock Watchlist 

143 elif args['function_arg'] in definitions.FUNC_DICT['list_watchlist']['values']: 143 ↛ 144line 143 didn't jump to line 144, because the condition on line 143 was never true

144 def cli_watchlist(): 

145 from scrilla.util.outputter import title_line, print_list 

146 tickers = files.get_watchlist() 

147 title_line("Watchlist") 

148 print_list(tickers) 

149 selected_function, required_length = cli_watchlist, 0 

150 

151 # FUNCTION: Purge Data Directories 

152 elif args['function_arg'] in definitions.FUNC_DICT["purge"]['values']: 

153 def cli_purge(): 

154 from scrilla.files import clear_directory, clear_cache 

155 logger.info( 

156 f'Clearing {settings.STATIC_DIR}, {settings.CACHE_DIR} and {settings.COMMON_DIR}', 'do_program') 

157 files.clear_directory(directory=settings.STATIC_DIR, retain=True) 

158 files.clear_directory(directory=settings.COMMON_DIR, retain=True) 

159 files.clear_cache() 

160 selected_function, required_length = cli_purge, 0 

161 

162 # FUNCTION: Display Version 

163 elif args['function_arg'] in definitions.FUNC_DICT["version"]['values']: 

164 def cli_version(): 

165 from scrilla.settings import APP_DIR 

166 from os.path import join 

167 version_file = join(APP_DIR, 'version.txt') 

168 with open(version_file, 'r') as f: 

169 print(f.read()) 

170 selected_function, required_length = cli_version, 0 

171 

172 # FUNCTION: Yield Curve 

173 elif args['function_arg'] in definitions.FUNC_DICT['yield_curve']['values']: 

174 def cli_yield_curve(): 

175 from scrilla.static.keys import keys 

176 from scrilla.services import get_daily_interest_history 

177 yield_curve = {} 

178 # TODO: this is inefficient. get_daily_interest_history should be modified 

179 # to return all maturities if no maturity is specified. otherwise, this is 

180 # duplicating a ton of operations 

181 for maturity in keys['YIELD_CURVE']: 

182 curve_rate = get_daily_interest_history(maturity=maturity, 

183 start_date=args['start_date'], 

184 end_date=args['start_date']) 

185 

186 yield_curve[maturity] = curve_rate[list( 

187 curve_rate.keys())[0]]/100 

188 

189 if print_format_to_screen(args): 189 ↛ 190line 189 didn't jump to line 190, because the condition on line 189 was never true

190 from scrilla.util.outputter import scalar_result 

191 scalar_result(calculation=maturity, 

192 result=yield_curve[maturity], currency=False) 

193 

194 if print_json_to_screen(args): 194 ↛ 198line 194 didn't jump to line 198, because the condition on line 194 was never false

195 from json import dumps 

196 print(dumps(yield_curve)) 

197 

198 if args['save_file'] is not None: 198 ↛ 199line 198 didn't jump to line 199, because the condition on line 198 was never true

199 from scrilla.files import save_file 

200 save_file(file_to_save=yield_curve, 

201 file_name=args['save_file']) 

202 selected_function, required_length = cli_yield_curve, 0 

203 

204 # ARGUMENT FUNCTIONS 

205 # FUNCTION: Asset Grouping 

206 

207 elif args['function_arg'] in definitions.FUNC_DICT['asset_type']['values']: 

208 def cli_asset_type(): 

209 from scrilla.files import get_asset_type 

210 from scrilla.util.outputter import string_result 

211 for arg in args['tickers']: 

212 asset_type = get_asset_type(arg) 

213 string_result(f'asset_type({arg})', asset_type) 

214 selected_function, required_length = cli_asset_type, 1 

215 

216 # FUNCTION: Black-Scholes Value At Risk 

217 elif args['function_arg'] in definitions.FUNC_DICT['var']['values']: 

218 def cli_var(): 

219 from scrilla.analysis.models.geometric.statistics import calculate_risk_return 

220 from scrilla.analysis.models.geometric.probability import percentile 

221 from scrilla.services import get_daily_price_history 

222 from scrilla.util.helper import get_first_json_key 

223 from scrilla.static.keys import keys 

224 

225 all_vars = {} 

226 for arg in args['tickers']: 

227 prices = get_daily_price_history(ticker=arg, 

228 start_date=args['start_date'], 

229 end_date=args['end_date']) 

230 latest_price = prices[get_first_json_key( 

231 prices)][keys['PRICES']['CLOSE']] 

232 profile = calculate_risk_return(ticker=arg, 

233 sample_prices=prices, 

234 method=args['estimation_method']) 

235 valueatrisk = percentile(S0=latest_price, 

236 vol=profile['annual_volatility'], 

237 ret=profile['annual_return'], 

238 expiry=args['expiry'], 

239 prob=args['probability']) 

240 

241 all_vars[arg] = {keys['STATISTICS']['VAR']: valueatrisk} 

242 

243 if print_format_to_screen(args): 243 ↛ 244line 243 didn't jump to line 244, because the condition on line 243 was never true

244 from scrilla.util.outputter import scalar_result 

245 scalar_result(f'{arg}_VaR', valueatrisk) 

246 

247 if print_json_to_screen(args): 247 ↛ 251line 247 didn't jump to line 251, because the condition on line 247 was never false

248 from json import dumps 

249 print(dumps(all_vars)) 

250 

251 if args['save_file'] is not None: 

252 from scrilla.files import save_file 

253 save_file(file_to_save=all_vars, file_name=args['save_file']) 

254 

255 selected_function, required_length = cli_var, 1 

256 

257 # FUNCTION: Black-Scholes Conditional Value At Risk 

258 elif args['function_arg'] in definitions.FUNC_DICT['cvar']['values']: 

259 def cli_cvar(): 

260 from scrilla.services import get_daily_price_history 

261 from scrilla.static.keys import keys 

262 from scrilla.analysis.models.geometric.statistics import calculate_risk_return 

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

264 from scrilla.util.helper import get_first_json_key 

265 all_cvars = {} 

266 for arg in args['tickers']: 

267 prices = get_daily_price_history(ticker=arg, 

268 start_date=args['start_date'], 

269 end_date=args['end_date']) 

270 latest_price = prices[get_first_json_key( 

271 prices)][keys['PRICES']['CLOSE']] 

272 profile = calculate_risk_return(ticker=arg, 

273 sample_prices=prices, 

274 method=args['estimation_method']) 

275 valueatrisk = percentile(S0=latest_price, 

276 vol=profile['annual_volatility'], 

277 ret=profile['annual_return'], 

278 expiry=args['expiry'], 

279 prob=args['probability']) 

280 cvar = conditional_expected_value(S0=latest_price, 

281 vol=profile['annual_volatility'], 

282 ret=profile['annual_return'], 

283 expiry=args['expiry'], 

284 conditional_value=valueatrisk) 

285 all_cvars[arg] = {keys['STATISTICS']['CVAR']: cvar} 

286 

287 if print_format_to_screen(args): 287 ↛ 288line 287 didn't jump to line 288, because the condition on line 287 was never true

288 from scrilla.util.outputter import scalar_result 

289 scalar_result( 

290 f'{arg}_conditional_VaR', cvar) 

291 

292 if print_json_to_screen(args): 292 ↛ 296line 292 didn't jump to line 296, because the condition on line 292 was never false

293 from json import dumps 

294 print(dumps(all_cvars)) 

295 

296 if args['save_file'] is not None: 

297 from scrilla.files import save_file 

298 save_file(file_to_save=all_cvars, file_name=args['save_file']) 

299 

300 selected_function, required_length = cli_cvar, 1 

301 

302 # FUNCTION: Capital Asset Pricing Model Cost of Equity 

303 elif args['function_arg'] in definitions.FUNC_DICT['capm_equity_cost']['values']: 

304 def cli_capm_equity_cost(): 

305 from scrilla.analysis.markets import cost_of_equity 

306 from scrilla.static.keys import keys 

307 all_costs = {} 

308 for arg in args['tickers']: 

309 equity_cost = cost_of_equity(ticker=arg, 

310 start_date=args['start_date'], 

311 end_date=args['end_date'], 

312 method=args['estimation_method']) 

313 all_costs[arg] = {keys['STATISTICS']['EQUITY']: equity_cost} 

314 

315 if print_format_to_screen(args): 315 ↛ 316line 315 didn't jump to line 316, because the condition on line 315 was never true

316 from scrilla.util.outputter import scalar_result 

317 scalar_result(f'{arg}_equity_cost', 

318 equity_cost, currency=False) 

319 

320 if print_json_to_screen(args): 320 ↛ 324line 320 didn't jump to line 324, because the condition on line 320 was never false

321 from json import dumps 

322 print(dumps(all_costs)) 

323 

324 if args['save_file'] is not None: 

325 from scrilla.files import save_file 

326 save_file(file_to_save=all_costs, file_name=args['save_file']) 

327 

328 selected_function, required_length = cli_capm_equity_cost, 1 

329 

330 # FUNCTION: Capital Asset Pricing Model Beta 

331 elif args['function_arg'] in definitions.FUNC_DICT['capm_beta']['values']: 

332 def cli_capm_beta(): 

333 from scrilla.analysis.markets import market_beta 

334 from scrilla.static.keys import keys 

335 all_betas = {} 

336 for arg in args['tickers']: 

337 beta = market_beta(ticker=arg, 

338 start_date=args['start_date'], 

339 end_date=args['end_date'], 

340 method=args['estimation_method']) 

341 all_betas[arg] = {keys['STATISTICS']['BETA']: beta} 

342 

343 if print_format_to_screen(args): 343 ↛ 344line 343 didn't jump to line 344, because the condition on line 343 was never true

344 from scrilla.util.outputter import scalar_result 

345 scalar_result(f'{arg}_beta', beta, currency=False) 

346 

347 if print_json_to_screen(args): 347 ↛ 351line 347 didn't jump to line 351, because the condition on line 347 was never false

348 from json import dumps 

349 print(dumps(all_betas)) 

350 

351 if args['save_file'] is not None: 

352 from scrilla.files import save_file 

353 save_file(file_to_save=all_betas, file_name=args['save_file']) 

354 selected_function, required_length = cli_capm_beta, 1 

355 

356 # FUNCTION: Last Close Price 

357 elif args['function_arg'] in definitions.FUNC_DICT["close"]['values']: 

358 def cli_close(): 

359 from scrilla.services import get_daily_price_latest 

360 

361 all_prices = {} 

362 for arg in args['tickers']: 

363 price = get_daily_price_latest(arg) 

364 all_prices[arg] = price 

365 

366 if print_format_to_screen(args): 366 ↛ 367line 366 didn't jump to line 367, because the condition on line 366 was never true

367 from scrilla.util.outputter import scalar_result 

368 scalar_result( 

369 calculation=f'Last {arg} close price', result=float(price)) 

370 

371 if print_json_to_screen(args): 371 ↛ 375line 371 didn't jump to line 375, because the condition on line 371 was never false

372 from json import dumps 

373 print(dumps(all_prices)) 

374 

375 if args['save_file'] is not None: 

376 from scrilla.files import save_file 

377 save_file(file_to_save=all_prices, file_name=args['save_file']) 

378 selected_function, required_length = cli_close, 1 

379 

380 # FUNCTION: Correlation Matrix 

381 elif args['function_arg'] in definitions.FUNC_DICT["correlation"]['values']: 

382 def cli_correlation(): 

383 from scrilla.static.keys import keys 

384 from scrilla.analysis.models.geometric.statistics import correlation_matrix 

385 

386 if args['estimation_method'] == keys['ESTIMATION']['LIKE']: 

387 logger.info( 

388 'This calculation takes a while, strap in...', 'do_program') 

389 

390 matrix = correlation_matrix(tickers=args['tickers'], 

391 start_date=args['start_date'], 

392 end_date=args['end_date'], 

393 method=args['estimation_method']) 

394 

395 if print_format_to_screen(args): 

396 from scrilla.util.outputter import correlation_matrix as correlation_output 

397 correlation_output( 

398 tickers=args['tickers'], correl_matrix=matrix) 

399 

400 elif print_json_to_screen(args): 400 ↛ 404line 400 didn't jump to line 404, because the condition on line 400 was never false

401 from json import dumps 

402 print(dumps({'correlation_matrix': matrix})) 

403 

404 if args['save_file'] is not None: 

405 from scrilla.files import save_file 

406 save_file(file_to_save=matrix, file_name=args['save_file']) 

407 

408 selected_function, required_length = cli_correlation, 2 

409 

410 # FUNCTION: Correlation Time Series 

411 elif args['function_arg'] in definitions.FUNC_DICT['correlation_time_series']['values']: 411 ↛ 412line 411 didn't jump to line 412, because the condition on line 411 was never true

412 def cli_correlation_series(): 

413 from scrilla.analysis.models.geometric.statistics import calculate_moment_correlation_series 

414 if print_format_to_screen(args): 

415 from scrilla.util.outputter import scalar_result 

416 

417 logger.info( 

418 'This calculation takes a while, strap in...', 'do_program') 

419 ticker_1, ticker_2 = args['tickers'][0], args['tickers'][1] 

420 result = calculate_moment_correlation_series(ticker_1=ticker_1, 

421 ticker_2=ticker_2, 

422 start_date=args['start_date'], 

423 end_date=args['end_date']) 

424 if print_format_to_screen(args): 

425 for this_date in result: 

426 scalar_result(calculation=f'{this_date}_{ticker_1}_{ticker_2}_correlation', 

427 result=float(result[this_date]), currency=False) 

428 elif print_json_to_screen(args): 

429 from json import dumps 

430 print(dumps(result)) 

431 

432 if args['save_file'] is not None: 

433 from scrilla.files import save_file 

434 save_file(file_to_save=result, file_name=args['save_file']) 

435 selected_function, required_length, exact = cli_correlation_series, 2, True 

436 

437 # FUNCTION: Discount Dividend Model 

438 elif args['function_arg'] in definitions.FUNC_DICT["discount_dividend"]['values']: 

439 def cli_discount_dividend(): 

440 from scrilla.services import get_dividend_history 

441 from scrilla.analysis.objects.cashflow import Cashflow 

442 from scrilla.static.keys import keys 

443 model_results = {} 

444 

445 for arg in args['tickers']: 

446 dividends = get_dividend_history(arg) 

447 if args['discount'] is None: 

448 from scrilla.analysis.markets import cost_of_equity 

449 discount = cost_of_equity( 

450 ticker=arg, method=args['estimation_method']) 

451 else: 

452 discount = args['discount'] 

453 result = Cashflow(sample=dividends, 

454 discount_rate=discount).calculate_net_present_value() 

455 model_results[arg] = {keys['MODELS']['DDM']: result} 

456 

457 if print_format_to_screen(args): 457 ↛ 458line 457 didn't jump to line 458, because the condition on line 457 was never true

458 from scrilla.util.outputter import scalar_result 

459 scalar_result(f'Net Present Value ({arg} dividends)', 

460 model_results[arg][keys['MODELS']['DDM']]) 

461 

462 if print_json_to_screen(args): 462 ↛ 466line 462 didn't jump to line 466, because the condition on line 462 was never false

463 from json import dumps 

464 print(dumps(model_results)) 

465 

466 if args['save_file'] is not None: 466 ↛ 467line 466 didn't jump to line 467, because the condition on line 466 was never true

467 from scrilla.files import save_file 

468 save_file(file_to_save=model_results, 

469 file_name=args['save_file']) 

470 selected_function, required_length = cli_discount_dividend, 1 

471 

472 elif args['function_arg'] in definitions.FUNC_DICT['dividends']['values']: 

473 def cli_dividends(): 

474 from scrilla.services import get_dividend_history 

475 if print_format_to_screen(args): 475 ↛ 476line 475 didn't jump to line 476, because the condition on line 475 was never true

476 from scrilla.util.outputter import scalar_result 

477 

478 all_dividends = {} 

479 for arg in args['tickers']: 

480 dividends = get_dividend_history(arg) 

481 all_dividends[arg] = dividends 

482 

483 if print_format_to_screen(args): 483 ↛ 484line 483 didn't jump to line 484, because the condition on line 483 was never true

484 for this_date in dividends: 

485 scalar_result( 

486 calculation=f'{arg}_dividend({this_date})', result=dividends[this_date]) 

487 

488 if print_json_to_screen(args): 488 ↛ exitline 488 didn't return from function 'cli_dividends', because the condition on line 488 was never false

489 from json import dumps 

490 print(dumps(all_dividends)) 

491 

492 selected_function, required_length = cli_dividends, 1 

493 

494 # FUNCTION: Efficient Frontier 

495 elif args['function_arg'] in definitions.FUNC_DICT['efficient_frontier']['values']: 

496 def cli_efficient_frontier(): 

497 from scrilla.analysis.objects.portfolio import Portfolio 

498 from scrilla.analysis.optimizer import calculate_efficient_frontier 

499 

500 portfolio = Portfolio(tickers=args['tickers'], 

501 start_date=args['start_date'], 

502 end_date=args['end_date'], 

503 method=args['estimation_method']) 

504 frontier = calculate_efficient_frontier(portfolio=portfolio, 

505 steps=args['steps']) 

506 

507 if args['investment'] is not None: 507 ↛ 508line 507 didn't jump to line 508, because the condition on line 507 was never true

508 from scrilla.services import get_daily_prices_latest 

509 prices = get_daily_prices_latest( 

510 tickers=args['tickers']) 

511 else: 

512 prices = None 

513 

514 if print_format_to_screen(args): 514 ↛ 515line 514 didn't jump to line 515, because the condition on line 514 was never true

515 from scrilla.util.outputter import efficient_frontier as frontier_output 

516 frontier_output(portfolio=portfolio, 

517 frontier=frontier, 

518 investment=args['investment'], 

519 latest_prices=prices) 

520 if print_json_to_screen(args): 520 ↛ 527line 520 didn't jump to line 527, because the condition on line 520 was never false

521 from json import dumps 

522 print(dumps(formats.format_frontier(portfolio=portfolio, 

523 frontier=frontier, 

524 investment=args['investment'], 

525 latest_prices=prices))) 

526 

527 if args['save_file'] is not None: 527 ↛ 528line 527 didn't jump to line 528, because the condition on line 527 was never true

528 from scrilla.files import save_frontier 

529 save_frontier(portfolio=portfolio, 

530 frontier=frontier, 

531 investment=args['investment'], 

532 file_name=args['save_file'], 

533 latest_prices=prices) 

534 selected_function, required_length = cli_efficient_frontier, 2 

535 

536 # FUNCTION: Maximize Portfolio Return 

537 elif args['function_arg'] in definitions.FUNC_DICT['maximize_return']['values']: 

538 def cli_maximize_return(): 

539 from scrilla.analysis.objects.portfolio import Portfolio 

540 from scrilla.analysis.optimizer import maximize_portfolio_return 

541 

542 portfolio = Portfolio(tickers=args['tickers'], 

543 start_date=args['start_date'], 

544 end_date=args['end_date'], 

545 method=args['estimation_method']) 

546 

547 allocation = maximize_portfolio_return( 

548 portfolio=portfolio) 

549 

550 if args['investment'] is not None: 550 ↛ 551line 550 didn't jump to line 551, because the condition on line 550 was never true

551 from scrilla.services import get_daily_prices_latest 

552 prices = get_daily_prices_latest( 

553 tickers=args['tickers']) 

554 else: 

555 prices = None 

556 

557 if print_format_to_screen(args): 557 ↛ 558line 557 didn't jump to line 558, because the condition on line 557 was never true

558 from scrilla.util.outputter import optimal_result 

559 optimal_result(portfolio=portfolio, 

560 allocation=allocation, 

561 investment=args['investment'], 

562 latest_prices=prices) 

563 

564 if print_json_to_screen(args): 564 ↛ 571line 564 didn't jump to line 571, because the condition on line 564 was never false

565 from json import dumps 

566 print(dumps(formats.format_allocation(allocation=allocation, 

567 portfolio=portfolio, 

568 investment=args['investment'], 

569 latest_prices=prices))) 

570 

571 if args['save_file'] is not None: 571 ↛ 572line 571 didn't jump to line 572, because the condition on line 571 was never true

572 from scrilla.files import save_allocation 

573 save_allocation(allocation=allocation, 

574 portfolio=portfolio, 

575 file_name=args['save_file'], 

576 investment=args['investment'], 

577 latest_prices=prices) 

578 selected_function, required_length = cli_maximize_return, 2 

579 

580 # FUNCTION: Moving Averages of Logarithmic Returns 

581 elif args['function_arg'] in definitions.FUNC_DICT['moving_averages']['values']: 581 ↛ 582line 581 didn't jump to line 582, because the condition on line 581 was never true

582 def cli_moving_averages(): 

583 from scrilla.analysis.models.geometric.statistics import calculate_moving_averages 

584 # TODO: moving averages with estimation techniques 

585 

586 moving_averages = calculate_moving_averages(ticker=args['tickers'][0], 

587 start_date=args['start_date'], 

588 end_date=args['end_date'], 

589 method=args['estimation_method']) 

590 

591 if print_format_to_screen(args): 

592 from scrilla.util.outputter import moving_average_result 

593 moving_average_result( 

594 ticker=args['tickers'][0], averages=moving_averages) 

595 

596 if print_json_to_screen(args): 

597 from json import dumps 

598 print(dumps(moving_averages)) 

599 

600 if args['save_file'] is not None: 

601 from scrilla.files import save_file 

602 save_file(file_to_save=moving_averages, 

603 file_name=args['save_file']) 

604 

605 selected_function, required_length = cli_moving_averages, 1 

606 

607 # FUNCTION: Optimize Portfolio Variance/Volatility 

608 elif args['function_arg'] in definitions.FUNC_DICT['optimize_portfolio']['values']: 608 ↛ 609line 608 didn't jump to line 609, because the condition on line 608 was never true

609 def cli_optimize_portfolio_variance(): 

610 from scrilla.analysis.objects.portfolio import Portfolio 

611 

612 portfolio = Portfolio(tickers=args['tickers'], 

613 start_date=args['start_date'], 

614 end_date=args['end_date'], 

615 method=args['estimation_method']) 

616 

617 if args['optimize_sharpe']: 

618 from scrilla.analysis.optimizer import maximize_sharpe_ratio 

619 allocation = maximize_sharpe_ratio( 

620 portfolio=portfolio, target_return=args['target']) 

621 else: 

622 from scrilla.analysis.optimizer import optimize_portfolio_variance 

623 allocation = optimize_portfolio_variance( 

624 portfolio=portfolio, target_return=args['target']) 

625 

626 if args['investment'] is not None: 

627 from scrilla.services import get_daily_prices_latest 

628 prices = get_daily_prices_latest( 

629 tickers=args['tickers']) 

630 else: 

631 prices = None 

632 

633 if print_format_to_screen(args): 

634 from scrilla.util.outputter import optimal_result 

635 optimal_result( 

636 portfolio=portfolio, 

637 allocation=allocation, 

638 investment=args['investment'], 

639 latest_prices=prices) 

640 

641 if print_json_to_screen(args): 

642 from json import dumps 

643 print(dumps(formats.format_allocation(allocation=allocation, 

644 portfolio=portfolio, 

645 investment=args['investment'], 

646 latest_prices=prices))) 

647 

648 if args['save_file'] is not None: 

649 from scrilla.files import save_allocation 

650 save_allocation(allocation=allocation, 

651 portfolio=portfolio, 

652 file_name=args['save_file'], 

653 investment=args['investment'], 

654 latest_prices=prices) 

655 

656 selected_function, required_length = cli_optimize_portfolio_variance, 2 

657 

658 # FUNCTION: Optimize Portfolio Conditional Value At Risk 

659 elif args['function_arg'] in definitions.FUNC_DICT['optimize_portfolio_conditional_var']['values']: 659 ↛ 660line 659 didn't jump to line 660, because the condition on line 659 was never true

660 def cli_optimize_conditional_value_at_risk(): 

661 from scrilla.analysis.optimizer import optimize_conditional_value_at_risk 

662 from scrilla.analysis.objects.portfolio import Portfolio 

663 

664 portfolio = Portfolio(tickers=args['tickers'], 

665 start_date=args['start_date'], 

666 end_date=args['end_date'], 

667 method=args['estimation_method']) 

668 allocation = optimize_conditional_value_at_risk(portfolio=portfolio, 

669 prob=args['probability'], 

670 expiry=args['expiry'], 

671 target_return=args['target']) 

672 if args['investment'] is not None: 

673 from scrilla.services import get_daily_prices_latest 

674 prices = get_daily_prices_latest( 

675 tickers=args['tickers']) 

676 else: 

677 prices = None 

678 

679 if print_format_to_screen(args): 

680 from scrilla.util.outputter import optimal_result 

681 optimal_result(portfolio=portfolio, 

682 allocation=allocation, 

683 investment=args['investment'], 

684 latest_prices=prices) 

685 

686 if print_json_to_screen(args): 

687 from json import dumps 

688 print(dumps(formats.format_allocation(allocation=allocation, 

689 portfolio=portfolio, 

690 investment=args['investment'], 

691 latest_prices=prices))) 

692 

693 if args['save_file'] is not None: 

694 from scrilla.files import save_allocation 

695 save_allocation(allocation=allocation, 

696 portfolio=portfolio, 

697 file_name=args['save_file'], 

698 investment=args['investment'], 

699 latest_prices=prices) 

700 selected_function, required_length = cli_optimize_conditional_value_at_risk, 2 

701 

702 # FUNCTION: Plot Correlation Time Series 

703 elif args['function_arg'] in definitions.FUNC_DICT['plot_correlation']['values']: 703 ↛ 704line 703 didn't jump to line 704, because the condition on line 703 was never true

704 def cli_plot_correlation(): 

705 from scrilla.analysis.models.geometric.statistics import calculate_moment_correlation_series 

706 from scrilla.analysis.plotter import plot_correlation_series 

707 

708 logger.info( 

709 'This calculation takes a while, strap in...', 'do_program') 

710 

711 correlation_history = calculate_moment_correlation_series(ticker_1=args['tickers'][0], 

712 ticker_2=args['tickers'][1], 

713 start_date=args['start_date'], 

714 end_date=args['end_date']) 

715 plot_correlation_series(tickers=args['tickers'], 

716 series=correlation_history, 

717 savefile=args['save_file']) 

718 

719 selected_function, required_length, exact = cli_plot_correlation, 2, True 

720 

721 # FUNCTION: Plot Dividend History With Linear Regression Model 

722 elif args['function_arg'] in definitions.FUNC_DICT['plot_dividends']['values']: 722 ↛ 723line 722 didn't jump to line 723, because the condition on line 722 was never true

723 def cli_plot_dividends(): 

724 from scrilla.services import get_dividend_history 

725 from scrilla.analysis.objects.cashflow import Cashflow 

726 from scrilla.analysis.plotter import plot_cashflow 

727 

728 dividends = get_dividend_history(ticker=args['tickers'][0]) 

729 if args['discount'] is None: 

730 from scrilla.analysis.markets import cost_of_equity 

731 args['discount'] = cost_of_equity(ticker=args['tickers'][0], 

732 method=args['estimation_method']) 

733 div_cashflow = Cashflow(sample=dividends, 

734 discount_rate=args['discount']) 

735 plot_cashflow(ticker=args['tickers'][0], 

736 cashflow=div_cashflow, show=True, 

737 savefile=args['save_file']) 

738 

739 selected_function, required_length, exact = cli_plot_dividends, 1, True 

740 

741 # FUNCTION: Plot Efficient Frontier 

742 elif args['function_arg'] in definitions.FUNC_DICT['plot_frontier']['values']: 742 ↛ 743line 742 didn't jump to line 743, because the condition on line 742 was never true

743 def cli_plot_frontier(): 

744 from scrilla.analysis.objects.portfolio import Portfolio 

745 from scrilla.analysis.optimizer import calculate_efficient_frontier 

746 from scrilla.analysis.plotter import plot_frontier 

747 

748 portfolio = Portfolio(tickers=args['tickers'],