uncertainty_post_march2026.py

import random
from copy import copy

import matplotlib
from typing import List, Union

import pandas as pd
import datetime
import numpy as np

from syscore.interactive.progress_bar import progressBar

matplotlib.rcParams.update({"font.size": 22})

def create_boring_corr_matrix(
    size: int, offdiag: float = 0.99, diag: float = 1.0
) -> pd.DataFrame:
    size_index = range(size)

    def _od(i, j, offdiag, diag):
        if i == j:
            return diag
        else:
            return offdiag

    corr_matrix_values_as_list = [
        [_od(i, j, offdiag, diag) for i in size_index] for j in size_index
    ]
    corr_matrix_values = np.array(corr_matrix_values_as_list)

    return pd.DataFrame(corr_matrix_values, index = range(size), columns=range(size))


def multivariate_random(Nlength: int, SR: Union[float, list],
                        corrmatrix: np.array,
                        annual_vol=.30,
                        date_start=datetime.datetime(1980, 1, 1)):

    if type(SR) is float:
        SRlist = [SR]*len(corrmatrix.columns)
    else:
        SRlist = copy(SR)
    periods_in_year = 256 ## fixed as we use business days

    daily_vol = annual_vol / (periods_in_year**.5)
    means = [x * annual_vol / (periods_in_year) for x in SRlist]
    stds = np.diagflat([daily_vol] * len(SRlist))

    covs = np.dot(stds, np.dot(corrmatrix, stds))

    array_of_returns = np.random.multivariate_normal(means, covs, Nlength).T
    dict_of_returns = dict([
        (asset_number,
         array_of_returns[asset_number]) for asset_number in range(len(SRlist))
    ])

    return create_arbitrary_pd_df(dict_of_returns,
                                 date_start=date_start,
                                  col_names=corrmatrix.columns)

def create_arbitrary_pd_df(
    data_list: dict, col_names: List[str], date_start=datetime.datetime(1980, 1, 1)
) -> pd.DataFrame:
    first_series = list(data_list.values())[0]
    date_index = pd.date_range(start=date_start, periods=len(first_series), freq="B")

    return pd.DataFrame(data_list, index=date_index, columns=col_names)


def portfolio_returns(returns_to_eval: pd.DataFrame, weights: pd.Series, risk_free: float):
    sum_weights = weights.sum()
    borrow_cost_or_interest = risk_free*(sum_weights -1)
    borrow_cost_or_interest_per_business_day = borrow_cost_or_interest/256
    portfolio = weights * returns_to_eval
    sum_portfolio = portfolio.sum(axis=1)

    return sum_portfolio - borrow_cost_or_interest_per_business_day

def calculate_weights_given_cor_matrix(cormatrix: pd.DataFrame,  per_asset_stdev: 0.3, per_asset_SR: 0.25, leverage_allowed: bool = False):
    weights =one_over_n_weights(cormatrix)
    idm = calculate_idm_including_risk_target_adjustment(weights=weights, cormatrix=cormatrix, per_asset_stdev=per_asset_stdev, per_asset_SR=per_asset_SR, leverage_allowed=leverage_allowed)

    return weights*idm

def one_over_n_weights(cormatrix: pd.DataFrame):
    avg_weight = 1.0 / len(cormatrix.columns)
    weights = pd.Series(dict([(i, avg_weight) for i in cormatrix.columns]))

    return weights

def calculate_idm_including_risk_target_adjustment(weights: pd.Series, cormatrix: pd.DataFrame,  per_asset_stdev: 0.3,
                                                   per_asset_SR: 0.25, leverage_allowed: bool = False):
    idm = calculate_div_multipilier(weights, cormatrix)
    expected_SR = idm*per_asset_SR
    target_risk = expected_SR ## change if not using full kelly
    expected_risk_without_idm_including_diversification_element = per_asset_stdev / idm

    risk_multiplier_to_use = target_risk / expected_risk_without_idm_including_diversification_element

    if leverage_allowed:
        return risk_multiplier_to_use
    else:
        return min([risk_multiplier_to_use, 1])


def calculate_div_multipilier(weights: pd.Series, corr_matrix:pd.DataFrame):

    weights_np = np.array(weights.values)
    corr_np= corr_matrix.values
    variance = weights_np.dot(corr_np).dot(weights_np)
    risk = variance ** 0.5

    return 1.0/risk

def calculate_expected_CAGR(size: int, avg_corr: float=.5, risk_free: float = 0.0375, per_asset_stdev: float= 0.3, per_asset_SR:float= 0.25, leverage_allowed: bool = False):
    cormatrix = create_boring_corr_matrix(size=size, offdiag=avg_corr)
    weights = one_over_n_weights(cormatrix)
    risk_reduction = calculate_div_multipilier(weights, cormatrix)
    expected_risk = per_asset_stdev/risk_reduction
    expected_SR = per_asset_SR*risk_reduction

    risk_target = expected_SR ## Kelly optimal

    optimal_leverage = risk_target / expected_risk
    if not leverage_allowed:
        optimal_leverage = min([1, optimal_leverage])

    expected_portfolio_stdev = optimal_leverage * expected_risk
    expected_excess_return_per_asset = optimal_leverage *  per_asset_stdev * per_asset_SR
    expected_return_per_asset = expected_excess_return_per_asset + risk_free
    cagr = expected_return_per_asset - .5*(expected_portfolio_stdev**2)

    return cagr

def randomness(Nlength: int, size: int, bootstraps: int, avg_corr: float=.5, risk_free: float = 0.0375, per_asset_stdev: float= 0.3,
               per_asset_SR:float= 0.25, leverage_allowed: bool = False):

    cormatrix = create_boring_corr_matrix(size=size, offdiag=avg_corr)
    weights = calculate_weights_given_cor_matrix(cormatrix=cormatrix, leverage_allowed=leverage_allowed, per_asset_stdev=per_asset_stdev, per_asset_SR=per_asset_SR)
    p = progressBar(bootstraps)
    list_of_outcomes = [get_portfolio_returns(p=p,
        weights=weights,
        Nlength=Nlength,
        cormatrix=cormatrix,
        risk_free=risk_free,
        SR=per_asset_SR
    ) for __ in range(bootstraps)]

    return pd.DataFrame(list_of_outcomes).transpose()

def get_portfolio_returns(p: progressBar, Nlength: int, SR: Union[float, list[float]], weights: pd.Series, cormatrix: pd.DataFrame,  risk_free: float):
    p.iterate()
    some_returns = multivariate_random(
        Nlength=Nlength,
        SR=SR,
        corrmatrix=cormatrix
    )
    p_returns = portfolio_returns(weights=weights, returns_to_eval=some_returns, risk_free=risk_free)

    return p_returns

x=randomness(1*256,3,5000, per_asset_SR=.25, leverage_allowed=False)
avg=x.mean(axis=1)
x.cumsum().plot(color="grey")
avg.cumsum().plot(color="black")
matplotlib.pyplot.show(block=True)

def calculate_cagr(series: pd.Series):
    ## risk free already included
    one_plus = 1+series
    one_plus_compound = one_plus.cumprod()
    final_value = one_plus_compound[-1]
    len_series_years = len(series)/256

    return final_value**(1/len_series_years)-1

def hist_of_cagr(x: pd.DataFrame):
    return pd.Series([calculate_cagr(x[i]) for i in x.columns])

h=hist_of_cagr(x)
h.plot.hist(bins=50)
matplotlib.pyplot.show(block=True)


def generate_SR_list_given_some_skill(portfolio_size:int, skill_ratio: float= 1, pool_size: int = 48,
                                      number_of_high_SR_in_pool: int = 3, typical_SR: float = 0.25, high_SR: float = 0.5):
    if portfolio_size==pool_size:
        l1=[high_SR]*number_of_high_SR_in_pool
        l2=[typical_SR]*(pool_size-number_of_high_SR_in_pool)
        return l1+l2

    current_pool_size = copy(pool_size)
    high_SR_remaining = copy(number_of_high_SR_in_pool)
    list_of_SR = []
    for i in range(portfolio_size):
        if found_high_SR(current_pool_size, high_SR_remaining, skill_ratio=skill_ratio):
            list_of_SR.append(high_SR)
            high_SR_remaining+=-1
        else:
            list_of_SR.append(typical_SR)

        current_pool_size+=-1

    return list_of_SR

def found_high_SR(current_pool_size: int, high_SR_remaining: int, skill_ratio: float = 1):
    if high_SR_remaining==0:
        return False

    if high_SR_remaining==current_pool_size:
        return True

    prob_of_a_high_SR_with_no_skill = float(high_SR_remaining)/current_pool_size
    prob_of_a_high_SR_with_skill = min([1, prob_of_a_high_SR_with_no_skill*skill_ratio])

    return random.uniform(0,1)<prob_of_a_high_SR_with_skill

x = [generate_SR_list_given_some_skill(3, number_of_high_SR_in_pool=3, pool_size=48, high_SR=.5, typical_SR=.23) for __ in range(5000)]
print(pd.DataFrame(x).mean(axis=1).mean())

x = [generate_SR_list_given_some_skill(48, number_of_high_SR_in_pool=3, pool_size=48, high_SR=.5, typical_SR=.23) for __ in range(5000)]
print(pd.DataFrame(x).mean(axis=1).mean())


y=[.23]*45+[.5]*3 ## .246875
print(pd.Series(y).mean())


def randomness_with_varying_SR(Nlength: int, portfolio_size: int, bootstraps: int, avg_corr: float=.5,
                                        skill_ratio: float= 1, pool_size: int = 48,
                                      number_of_high_SR_in_pool: int = 3, typical_SR: float = 0.25, high_SR: float = 0.5,
                               risk_free: float = 0.0375, per_asset_stdev: float= 0.3,
                leverage_allowed: bool = False):


    p = progressBar(bootstraps)
    list_of_outcomes = [generate_SR_list_and_get_portfolio_returns(
        p=p,
        Nlength=Nlength,
        portfolio_size=portfolio_size,
        avg_corr=avg_corr,
        pool_size=pool_size,
        number_of_high_SR_in_pool=number_of_high_SR_in_pool,
        typical_SR=typical_SR,
        high_SR=high_SR,
        skill_ratio=skill_ratio,
        risk_free=risk_free,
        per_asset_stdev=per_asset_stdev,
        leverage_allowed=leverage_allowed

    ) for __ in range(bootstraps)]

    return pd.DataFrame(list_of_outcomes).transpose()


def generate_SR_list_and_get_portfolio_returns(p: progressBar, Nlength: int, portfolio_size: int,  avg_corr: float=.5,
                                        skill_ratio: float= 1, pool_size: int = 48,
                                      number_of_high_SR_in_pool: int = 3, typical_SR: float = 0.25, high_SR: float = 0.5,
                               risk_free: float = 0.0375, per_asset_stdev: float= 0.3,
                leverage_allowed: bool = False):
    sR_list = generate_SR_list_given_some_skill(portfolio_size=portfolio_size,
                                                pool_size=pool_size,
                                                number_of_high_SR_in_pool=number_of_high_SR_in_pool,
                                                skill_ratio=skill_ratio,
                                                typical_SR=typical_SR,
                                                high_SR=high_SR)
    cormatrix = create_boring_corr_matrix(size=portfolio_size, offdiag=avg_corr)
    weights = calculate_weights_given_cor_matrix(cormatrix=cormatrix, leverage_allowed=leverage_allowed, per_asset_stdev=per_asset_stdev, per_asset_SR=per_asset_SR)
    return get_portfolio_returns(p=p,
                          weights=weights,
                          Nlength=Nlength,
                          cormatrix=cormatrix,
                          risk_free=risk_free,
                          SR=sR_list
                          )

x_3=randomness_with_varying_SR(Nlength=256, portfolio_size=3, pool_size=48, number_of_high_SR_in_pool=3, skill_ratio=1, bootstraps=5000,
                             typical_SR=0.23, high_SR=0.5)

x_48 = randomness_with_varying_SR(Nlength=256, portfolio_size=48, pool_size=48, number_of_high_SR_in_pool=3, skill_ratio=1, bootstraps=5000,
                             typical_SR=0.23, high_SR=0.5)

h_3=hist_of_cagr(x_3)
h_48=hist_of_cagr(x_48)

x_3 = randomness_with_varying_SR(Nlength=256, portfolio_size=3, pool_size=48, number_of_high_SR_in_pool=3, skill_ratio=9, bootstraps=5000,
                             typical_SR=0.23, high_SR=0.5)
h_3=hist_of_cagr(x_3)
print(h_3.quantile(.25))