Portfolio Optimizer

This application performs portfolio optimization given a set of stock time series. Data flows through a chain of pn.rx expressions, and the expensive random sampling and efficient frontier are only recomputed when Run Analysis is clicked.

from io import BytesIO

import holoviews as hv
import hvplot.pandas  # noqa: F401
import numpy as np
import pandas as pd
import panel as pn

from scipy.optimize import minimize

pn.extension('tabulator', loading_indicator=True)

SEED = 42

Load data

@pn.cache
def get_stocks(data):
    if data is None:
        stock_file = 'https://datasets.holoviz.org/stocks/v1/stocks.csv'
    else:
        stock_file = BytesIO(data)
    return pd.read_csv(stock_file, index_col='Date', parse_dates=True)

def select_columns(df, cols):
    return df[cols] if cols else df

Business logic

These are plain functions of DataFrames, kept separate from the UI so the reactive pipeline below only wires them together.

def compute_random_allocations(log_return, num_ports=15000, seed=SEED):
    _, ncols = log_return.shape

    # Compute log and mean return
    mean_return = np.nanmean(log_return, axis=0)

    # Allocate normalized weights
    weights = np.random.default_rng(seed).random((num_ports, ncols))
    normed_weights = (weights.T / np.sum(weights, axis=1)).T
    data = dict(zip(log_return.columns, normed_weights.T))

    # Compute expected return and volatility of random portfolios
    data['Return'] = expected_return = np.sum((mean_return * normed_weights) * 252, axis=1)
    return_covariance = np.cov(log_return[1:], rowvar=False) * 252
    if not return_covariance.shape:
        return_covariance = np.array([[252.]])
    data['Volatility'] = volatility = np.sqrt(
        (normed_weights * np.tensordot(return_covariance, normed_weights.T, axes=1).T).sum(axis=1)
    )
    data['Sharpe'] = expected_return/volatility

    df = pd.DataFrame(data)
    df.attrs['mean_return'] = mean_return
    df.attrs['log_return'] = log_return
    return df

def check_sum(weights):
    return np.sum(weights) - 1

def get_return(mean_ret, weights):
    return np.sum(mean_ret * weights) * 252

def get_volatility(log_ret, weights):
    return np.sqrt(np.dot(weights.T, np.dot(np.cov(log_ret[1:], rowvar=False) * 252, weights)))

def compute_frontier(df, n=30):
    frontier_ret = np.linspace(df.Return.min(), df.Return.max(), n)
    frontier_volatility = []

    cols = len(df.columns) - 3
    bounds = tuple((0, 1) for i in range(cols))
    init_guess = [1./cols for i in range(cols)]
    for possible_return in frontier_ret:
        cons = (
            {'type':'eq', 'fun': check_sum},
            {'type':'eq', 'fun': lambda w, r=possible_return: get_return(df.attrs['mean_return'], w) - r}
        )
        result = minimize(lambda w: get_volatility(df.attrs['log_return'], w), init_guess, bounds=bounds, constraints=cons)
        frontier_volatility.append(result['fun'])
    return pd.DataFrame({'Volatility': frontier_volatility, 'Return': frontier_ret})

def minimize_difference(weights, des_vol, des_ret, log_ret, mean_ret):
    ret = get_return(mean_ret, weights)
    vol = get_volatility(log_ret, weights)
    return abs(des_ret-ret) + abs(des_vol-vol)

@pn.cache
def find_best_allocation(log_return, vol, ret):
    cols = log_return.shape[1]
    vol = vol or 0
    ret = ret or 0
    mean_return = np.nanmean(log_return, axis=0)
    bounds = tuple((0, 1) for i in range(cols))
    init_guess = [1./cols for i in range(cols)]
    cons = (
        {'type':'eq','fun': check_sum},
        {'type':'eq','fun': lambda w: get_return(mean_return, w) - ret},
        {'type':'eq','fun': lambda w: get_volatility(log_return, w) - vol}
    )
    opt = minimize(
        minimize_difference, init_guess, args=(vol, ret, log_return, mean_return),
        bounds=bounds, constraints=cons
    )
    ret = get_return(mean_return, opt.x)
    vol = get_volatility(log_return, opt.x)
    return pd.Series(list(opt.x)+[ret, vol], index=list(log_return.columns)+['Return', 'Volatility'], name='Weight')

def compute_log_return(stocks):
    return np.log(stocks/stocks.shift(1))

def max_sharpe_row(df):
    return df[df.Sharpe == df.Sharpe.max()]

def log_return_of(allocations):
    return allocations.attrs['log_return']

Declare UI components

The stock selector options and the date range follow the loaded data through bound references.

file_input = pn.ui.FileInput(accept='.csv', sizing_mode='stretch_width')
stocks = pn.rx(get_stocks)(file_input.param.value)

selector = pn.ui.MultiSelect(
    label='Select stocks', sizing_mode='stretch_width',
    options=stocks.rx.pipe(lambda df: df.columns.to_list())
)
n_samples = pn.ui.IntSlider(
    label='Random samples', value=10_000, start=1000, end=20_000, step=1000, sizing_mode='stretch_width'
)
button = pn.ui.Button(label='Run Analysis', sizing_mode='stretch_width')
investment = pn.ui.IntInput(label='Investment Value in $', value=5000, step=1000, start=1000, end=100000)
year = pn.ui.DateRangeSlider(
    label='Year',
    start=stocks.rx.pipe(lambda df: df.index.min()),
    end=stocks.rx.pipe(lambda df: df.index.max()),
    value=stocks.rx.pipe(lambda df: (df.index.min(), df.index.max())),
)
posxy = hv.streams.Tap(x=None, y=None)

text = """
#  Portfolio optimization

This application performs portfolio optimization given a set of stock time series.

To optimize your portfolio:

1. Upload a CSV of the daily stock time series for the stocks you are considering
2. Select the stocks to be included.
3. Run the Analysis
4. Click on the Return/Volatility plot to select the desired risk/reward profile

Upload a CSV containing stock data:
"""

explanation = """
The code for this app was taken from [this excellent introduction to Python for Finance](https://github.com/PrateekKumarSingh/Python/tree/master/Python%20for%20Finance/Python-for-Finance-Repo-master).
To learn some of the background and theory about portfolio optimization see [this notebook](https://github.com/PrateekKumarSingh/Python/blob/master/Python%20for%20Finance/Python-for-Finance-Repo-master/09-Python-Finance-Fundamentals/02-Portfolio-Optimization.ipynb).
"""

sidebar = pn.ui.Column(
    pn.ui.Markdown(text, margin=(0, 10)),
    file_input,
    selector,
    n_samples,
    button,
    pn.ui.Markdown(explanation),
    sizing_mode='stretch_width'
)

sidebar

Data pipeline

rx.when(button) holds the random allocations, and everything derived from them, until the button is clicked. The closest allocation additionally follows the Tap stream on the scatter plot.

selected_stocks = pn.rx(select_columns)(stocks, selector.param.value)
log_return = selected_stocks.rx.pipe(compute_log_return)
random_allocations = pn.rx(compute_random_allocations)(log_return, n_samples.param.value).rx.when(button)
efficient_frontier = random_allocations.rx.pipe(compute_frontier)
max_sharpe = random_allocations.rx.pipe(max_sharpe_row)
closest_allocation = pn.rx(find_best_allocation)(
    random_allocations.rx.pipe(log_return_of), posxy.param.x, posxy.param.y
)

Plot

Portfolio optimization plot

The scatter plot is the source of the Tap stream, so it is a DynamicMap that stays the same object while its data updates.

OPTS = {'x': 'Volatility', 'y': 'Return', 'responsive': True}

def allocations_plot(df):
    return df.hvplot.scatter(alpha=0.1, color='Sharpe', cmap='plasma', **OPTS).opts(tools=[])

def frontier_plot(df):
    return df.hvplot(line_dash='dashed', color='green', **OPTS)

def max_sharpe_plot(df):
    return df.hvplot.scatter(line_color='black', size=50, **OPTS)

def closest_plot(allocation):
    return allocation.to_frame().T.hvplot.scatter(color='green', line_color='black', size=50, **OPTS)

def describe(p):
    return (
        f"The selected portfolio has a volatility of {p.Volatility:.2f}, a return of {p.Return:.2f} "
        f"and Sharpe ratio of {p.Return/p.Volatility:.2f}."
    )

def weights_table(allocation):
    return allocation.to_frame().iloc[:-2]

allocations_scatter = hv.DynamicMap(pn.bind(allocations_plot, random_allocations))
posxy.source = allocations_scatter

plot = pn.ui.HoloViews(
    (
        allocations_scatter *
        hv.DynamicMap(pn.bind(frontier_plot, efficient_frontier)) *
        hv.DynamicMap(pn.bind(max_sharpe_plot, max_sharpe)) *
        hv.DynamicMap(pn.bind(closest_plot, closest_allocation))
    ).opts(min_height=400, show_grid=True),
    sizing_mode='stretch_both',
)
summary = pn.ui.Markdown(pn.bind(describe, closest_allocation), width=250)
table = pn.ui.Tabulator(closest_allocation.rx.pipe(weights_table))

analysis = pn.ui.Row(plot, pn.ui.Column(summary, table), sizing_mode='stretch_both')

analysis

Portfolio Performance plot

def performance_plot(stocks, allocation, investment, date_range):
    start, end = date_range
    weights = allocation.iloc[:-2] * investment
    value = (stocks[start:end] * weights / stocks[start:].iloc[0]).sum(axis=1).rename()
    return value.hvplot.line(
        ylabel='Total Value ($)', title='Portfolio performance', responsive=True, min_height=400
    )

performance = pn.ui.Column(
    pn.ui.Row(year, investment),
    pn.ui.HoloViews(
        pn.bind(performance_plot, selected_stocks, closest_allocation, investment, year),
        sizing_mode='stretch_both',
    ),
    sizing_mode='stretch_both'
)

performance

Plot stock prices

def timeseries_plot(stocks):
    return stocks.hvplot.line(
        'Date', group_label='Stock', value_label='Stock Price ($)', title='Daily Stock Price',
        min_height=300, responsive=True, grid=True, legend='top_left'
    )

timeseries = pn.ui.HoloViews(pn.bind(timeseries_plot, selected_stocks), sizing_mode='stretch_both')

timeseries

Log return plots

def log_return_hists(log_return):
    return log_return.hvplot.hist(
        min_height=300, min_width=400, responsive=True, bins=100, subplots=True, group_label='Stock'
    ).cols(2).opts(sizing_mode='stretch_both')

log_ret_hists = pn.ui.HoloViews(pn.bind(log_return_hists, log_return), sizing_mode='stretch_both')

log_ret_hists

Overall layout

main = pn.ui.Tabs(
    ('Analysis', pn.ui.Column(analysis, performance, sizing_mode='stretch_both')),
    ('Timeseries', timeseries),
    ('Log Return', pn.ui.Column(
        pn.ui.Markdown('## Daily normalized log returns'),
        pn.ui.Markdown('Width of distribution indicates volatility and center of distribution the mean daily return.'),
        log_ret_hists,
        sizing_mode='stretch_both'
    )),
    sizing_mode='stretch_both', min_height=1000
)

page = pn.ui.Page(main=[main], sidebar=[sidebar], title='Portfolio Optimizer')

page

Served App

if pn.state.served:
    page.servable()