Coverage for src/scrilla/analysis/plotter.py: 0%
196 statements
« prev ^ index » next coverage.py v6.4.2, created at 2022-07-18 18:14 +0000
1import datetime
2from typing import Dict, List, Union
3import numpy
4import matplotlib
5from PIL import Image
7from matplotlib.figure import Figure
8from matplotlib import dates as mdates
9from matplotlib.ticker import PercentFormatter
11from scrilla import settings
12from scrilla.static import formats, keys
13from scrilla.analysis.objects.portfolio import Portfolio
14from scrilla.analysis.objects.cashflow import Cashflow
15from scrilla.util.errors import InputValidationError
16from scrilla.util import dater
18if settings.APP_ENV == 'local':
19 from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
20 matplotlib.use("Qt5Agg")
21elif settings.APP_ENV == 'container':
22 from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas
23 matplotlib.use("agg")
26def _show_or_save(canvas: FigureCanvas, show: bool = True, savefile: str = None) -> Union[FigureCanvas, None]:
27 if savefile is not None:
28 canvas.print_jpeg(filename_or_obj=savefile)
30 s, (width, height) = canvas.print_to_buffer()
31 im = Image.frombytes("RGBA", (width, height), s)
33 if show:
34 im.show()
35 return None
36 canvas.draw()
37 return canvas
40def plot_qq_series(ticker: str, qq_series: list, show: bool = True, savefile: str = None) -> Union[FigureCanvas, None]:
41 title = f'{ticker} Return Q-Q Plot'
42 normal_series = []
43 sample_series = []
44 for point in qq_series:
45 normal_series.append(point[0])
46 sample_series.append(point[1])
48 canvas = FigureCanvas(Figure())
49 axes = canvas.figure.subplots()
51 axes.plot(normal_series, sample_series,
52 linestyle="None", marker=".", markersize=10.0)
53 # axes.plot(normal_series, normal_series)
54 axes.grid()
55 axes.set_xlabel('Normal Percentiles')
56 axes.set_ylabel('Sample Percentiles')
57 axes.set_title(title)
59 return _show_or_save(canvas=canvas, show=show, savefile=savefile)
62def plot_correlation_series(tickers: list, series: dict, show: bool = True, savefile: str = None) -> Union[FigureCanvas, None]:
63 start, end = list(series.keys())[-1], list(series.keys())[0]
64 title = f'({tickers[0]}, {tickers[1]}) correlation time series'
65 subtitle = f'{start} to {end}, rolling {settings.DEFAULT_ANALYSIS_PERIOD}-day estimate'
67 canvas = FigureCanvas(Figure())
68 figure = canvas.figure
69 axes = figure.subplots()
71 locator = mdates.AutoDateLocator()
72 formatter = mdates.AutoDateFormatter(locator)
74 correl_history, dates = [], []
75 for date in series:
76 dates.append(dater.parse(date))
77 correl_history.append(series[date])
79 axes.plot(dates, correl_history)
80 axes.grid()
81 axes.xaxis.set_major_locator(locator)
82 axes.xaxis.set_major_formatter(formatter)
83 axes.set_ylabel('Correlation')
84 axes.set_xlabel('Dates')
85 axes.set_title(subtitle, fontsize=12)
86 figure.suptitle(title, fontsize=18)
87 figure.autofmt_xdate()
89 return _show_or_save(canvas=canvas, show=show, savefile=savefile)
92def plot_frontier(portfolio: Portfolio, frontier: list, show: bool = True, savefile: str = None) -> Union[FigureCanvas, None]:
93 title = " ("
94 for i, item in enumerate(portfolio.tickers):
95 if i != (len(portfolio.tickers) - 1):
96 title += item + ", "
97 else:
98 title += item + ") Efficient Frontier"
100 return_profile, risk_profile = [], []
101 for allocation in frontier:
102 return_profile.append(portfolio.return_function(allocation))
103 risk_profile.append(portfolio.volatility_function(allocation))
105 canvas = FigureCanvas(Figure())
106 axes = canvas.figure.subplots()
108 axes.plot(risk_profile, return_profile, linestyle='dashed')
109 axes.grid()
110 axes.set_xlabel('Volatility')
111 axes.set_ylabel('Return')
112 axes.set_title(title)
114 return _show_or_save(canvas=canvas, show=show, savefile=savefile)
117def plot_yield_curve(yield_curve: Dict[str, List[float]], show: bool = True, savefile: str = None) -> Union[FigureCanvas, None]:
118 title = f'US Treasury Yield Curve On {list(yield_curve.keys())[0]}'
120 canvas = FigureCanvas(Figure())
121 axes = canvas.figure.subplots()
123 maturities, rates = [], yield_curve[list(yield_curve.keys())[0]]
124 yield_map = keys.keys['SERVICES']['STATISTICS']['QUANDL']['MAP']['YIELD_CURVE']
125 for i in range(len(rates)):
126 maturities.append(yield_map[keys.keys['YIELD_CURVE'][i]])
128 axes.plot(maturities, rates, linestyle="dashed",
129 marker=".", markersize=10.0)
130 axes.grid()
131 axes.set_xlabel('Maturity')
132 axes.set_ylabel('Annual Yield %')
133 axes.set_title(title)
135 return _show_or_save(canvas=canvas, show=show, savefile=savefile)
138def plot_return_histogram(ticker: str, sample: List[float], show: bool = True, savefile: str = None) -> Union[FigureCanvas, None]:
139 canvas = FigureCanvas(Figure())
140 axes = canvas.figure.subplots()
142 axes.hist(x=sample, bins=formats.formats['BINS'], density=True)
143 axes.xaxis.set_major_formatter(PercentFormatter(xmax=1))
144 axes.set_title(f'Distribution of {ticker} Returns')
145 axes.set_xlabel('Daily Return')
147 return _show_or_save(canvas=canvas, show=show, savefile=savefile)
150def plot_profiles(symbols: List[str], profiles: Dict[str, Dict[str, float]], show: bool = True, savefile: str = None, subtitle: str = None) -> Union[FigureCanvas, None]:
151 canvas = FigureCanvas(Figure())
153 no_symbols = len(symbols)
154 axes = canvas.figure.subplots()
156 title = "("
157 for symbol in symbols:
158 if symbols.index(symbol) != (len(symbols)-1):
159 title += symbol + ", "
160 else:
161 title += symbol + ') Risk-Return Profile'
162 if subtitle is not None:
163 title += "\n" + subtitle
165 return_profile, risk_profile = [], []
166 for profile in profiles:
167 return_profile.append(profiles[profile]['annual_return'])
168 risk_profile.append(profiles[profile]['annual_volatility'])
170 axes.plot(risk_profile, return_profile,
171 linestyle='None', marker=".", markersize=10.0)
172 axes.grid()
173 axes.set_xlabel('Volatility')
174 axes.set_ylabel('Return')
175 axes.set_title(title)
177 for i in range(no_symbols):
178 axes.annotate(symbols[i], (risk_profile[i], return_profile[i]))
180 return _show_or_save(canvas=canvas, show=show, savefile=savefile)
182 # TODO: figure out date formatting for x-axis
185def plot_moving_averages(ticker: str, averages: Dict[str, Dict[str, float]], show: bool = False, savefile: str = None):
186 canvas = FigureCanvas(Figure())
187 axes = canvas.figure.subplots()
189 date_format = matplotlib.dates.DateFormatter('%m-%d')
191 ma1s, ma2s, ma3s, date_range = [], [], [], []
192 ma1_label, ma2_label, ma3_label = None, None, None
194 for this_date, these_averages in averages.items():
195 if ma1_label is None:
196 ma1_label = list(these_averages.keys())[0]
197 if ma2_label is None:
198 ma2_label = list(these_averages.keys())[1]
199 if ma3_label is None:
200 ma3_label = list(these_averages.keys())[2]
201 ma1s.append(list(these_averages.values())[0])
202 ma2s.append(list(these_averages.values())[1])
203 ma3s.append(list(these_averages.values())[2])
204 if len(averages) == 1:
205 date_range.append(this_date)
206 else:
207 date_range.append(datetime.datetime.strptime(
208 this_date, '%Y-%m-%d').toordinal())
210 if len(averages) == 1:
211 width = formats.formats['BAR_WIDTH']
212 x = numpy.arange(1)
213 axes.bar(x + width, ma1s, width, color="darkgreen", label=ma1_label)
214 axes.bar(x, ma2s, width, color="gold", label=ma2_label)
215 axes.bar(x - width, ma3s, width, color="orangered", label=ma3_label)
216 axes.set_xticks(x)
217 axes.set_xticklabels([str(date_range[0])])
219 else:
220 x = date_range
221 axes.plot(x, ma1s, linestyle="solid",
222 color="darkgreen", label=ma1_label)
223 axes.plot(x, ma2s, linestyle="dotted",
224 color="gold", label=ma2_label)
225 axes.plot(x, ma3s, linestyle="dashdot",
226 color="orangered", label=ma3_label)
227 axes.set_xticks(x)
228 date_locator = matplotlib.dates.WeekdayLocator(
229 byweekday=(matplotlib.dates.WE))
230 axes.xaxis.set_major_locator(date_locator)
231 axes.xaxis.set_major_formatter(date_format)
233 axes.set_ylabel('Annualized Logarthmic Return %')
234 axes.set_xlabel('Dates')
235 axes.set_title(
236 f'{ticker} Annualized Return Moving Averages')
237 axes.legend()
238 return _show_or_save(canvas=canvas, show=show, savefile=savefile)
241def plot_cashflow(ticker: str, cashflow: Cashflow, show: bool = True, savefile: str = None) -> Union[FigureCanvas, None]:
242 if not cashflow.beta or not cashflow.alpha or len(cashflow.sample) < 3:
243 raise InputValidationError(
244 "Cashflow model does not contain enough information to be plotted")
246 canvas = FigureCanvas(Figure())
247 figure = canvas.figure
248 axes = figure.subplots()
250 sup_title_str = f'{ticker} Dividend Linear Regression Model'
251 title_str = f'NPV(dividends | discount = {round(cashflow.discount_rate,4)}) = $ {round(cashflow.calculate_net_present_value(), 2)}'
253 dividend_history, ordinal_x, dates = [], [], []
254 for date in cashflow.sample:
255 dates.append(date)
256 ordinal_x.append(datetime.datetime.strptime(
257 date, '%Y-%m-%d').toordinal())
258 dividend_history.append(cashflow.sample[date])
260 dates.reverse()
262 model_map = list(map(lambda x: cashflow.alpha +
263 cashflow.beta*x, cashflow.time_series))
265 axes.scatter(ordinal_x, dividend_history, marker=".")
266 axes.plot(ordinal_x, model_map)
267 axes.grid()
268 # axes.xaxis.set_major_formatter(date_format)
269 axes.set_xticklabels(dates)
270 axes.set_ylabel('Dividend Amount')
271 axes.set_xlabel('Payment Date')
272 axes.set_title(title_str, fontsize=12)
273 figure.suptitle(sup_title_str, fontsize=18)
274 figure.autofmt_xdate()
276 return _show_or_save(canvas=canvas, show=show, savefile=savefile)