task_5_v2

import pandas as pd
import numpy as np
import cvxpy as cvx


RISK_FREE_RATE = 0.035 # безрисковая процентная ставка
RISK_FREE_RATE_MONTH = 0.035 / 12. # безрисковая процентная ставка(в месяц)

def filter_days(df: pd.DataFrame, timegate: str = '2023-01-01') -> tuple[pd.DataFrame, pd.DataFrame]:
    """ Удаляем выходные дни

    :param df: Датафрейм со столбцом Date

    :returns: Отфильтрованный DataFrame
    """
    mask = np.is_busday(df['Date'].to_numpy().tolist(), weekmask='1111100')
    df = df[mask]
    return df[df['Date'] < timegate], df[df['Date'] >= timegate]


def prepare_data(df: pd.DataFrame) -> np.ndarray:
    """ Получаем все необходимые данные из DataFram'а

    :param df: DataFrame c данными
    """
    returns = []

    cur_month = ''
    begin_pr = -1
    prev = None
    # YYYY-MM-DD
    # 0123456789
    for row in df.iterrows():
        data = row[1]
        if data['Date'][5:7] != cur_month or cur_month == '':
            if cur_month != '':
                returns.append((prev['Close'] - begin_pr) / begin_pr)
            cur_month = data['Date'][5:7]
            begin_pr = data['Close']
        prev = data

    returns.append((df.iloc[-1]['Close'] - begin_pr) / begin_pr)

    return np.array(returns)


def sample_var(data: np.array, percentile: float) -> np.float64:
    """ Получение VaR по выборке по формуле Var(X) = -X_(percentile * n).
    Где n длинна выборки

    :param data: выборка
    :param percentile: персентиль VaR'а

    :returns: VaR соответсвтующего персентиля
    """
    return -np.sort(data)[int(percentile * len(data))]


def gaussian_var(data: np.array, percentile: float) -> np.float64:
    """ Получение VaR в предположении гаусовости.
    Формула VaR = -(a + s * q). Где
    - a - оцененное среднее
    - s - оцененная дисперсия
    - q - минус квантиль нужного персентиля Гаусовского распределения

    :param data: выборка
    :param percentile: персентиль VaR'а

    :returns: VaR соответсвтующего персентиля
    """

    mean = np.mean(data)
    std_dev = np.std(data)
    return -mean + 1.65 * std_dev  # 1.65 - квантиль нормального распределения для 5%

def rf_weight(weights: np.ndarray) -> float:
    return 1 - np.sum(weights)

def calc_portfolio(weights: np.ndarray, expectaion: float) -> float:
    return weights.T @ expectaion + rf_weight(weights) * RISK_FREE_RATE_MONTH

def backtest(var: float, test_sample: np.ndarray, step: int=5) -> None:
    data = prepare_data(test_sample, step)
    exception_count = np.sum(data < -var)
    exception_freq = exception_count / len(data)
    print(f"Количество исключений {exception_count} из {len(data)}")
    print(f"Исключения в процентах {exception_freq * 100}%")



print(f"{'Читаем данные и объединяем':-^40}")
df_1 = pd.read_csv('MSFT.csv')
df_1['Close_MSFT'] = df_1.Close
df_2 = pd.read_csv('TSLA.csv')
df_2['Close_TSLA'] = df_2.Close
df = df_1.merge(df_2, on='Date', how='inner')[['Date', 'Close_MSFT', 'Close_TSLA']]
print(df_1.head(5))
print(df_2.head(5))
print(f"{'Итоговый':-^40}")
print(df.head(5))
print(f"{'Done':-^40}")

# Удаляем выходные дни
print(f"{'Удаляем выходные дни':-^40}")
train_df, test_df = filter_days(df)
print(train_df.head(5))
print(test_df.head(5))
print(f"{'Done':-^40}")


# Подготавливаем данные
print(f"{'Считаем месячные доходности акций':-^40}")
tmp1, tmp2 = train_df[['Date', 'Close_MSFT']], train_df[['Date', 'Close_TSLA']]

tmp1['Close'] = tmp1['Close_MSFT']
tmp1 = tmp1[['Date', 'Close']]

tmp2['Close'] = tmp2['Close_TSLA']
tmp2 = tmp2[['Date', 'Close']]

data_1 = prepare_data(tmp1)
data_2 = prepare_data(tmp2)

print(data_1[:5])
print(data_2[:5])
print(f"{'Done':-^40}")


# Келли
print(f"{'Келли':-^40}")
data = np.array([data_1, data_2])

expected_returns = data.mean(axis=1)
covariance_returns = np.cov(data)
print(f"Среднее : {expected_returns}")
print(f"Матрица ковариаций : {covariance_returns}")

optimal_weights_anal = np.linalg.inv(covariance_returns) @ (expected_returns - RISK_FREE_RATE_MONTH)

print(f"Оптимальные веса(Аналитическое решение): {optimal_weights_anal}")

optimal_weights_conv = cvx.Variable(2)
cvx.Problem(cvx.Maximize(
RISK_FREE_RATE_MONTH + optimal_weights_conv.T @ (expected_returns - RISK_FREE_RATE_MONTH)
- (1 / 2) * (optimal_weights_conv.T @ covariance_returns @ optimal_weights_conv)
)).solve()

print(f"Оптимальные веса(Выпуклая оптимизация): {optimal_weights_conv.value}")
print(f"{'Done':-^40}")

print(f"{'Результат':-^40}")
print(f"Норма доходности: { str(calc_portfolio(optimal_weights_anal, expected_returns) * 100)}%")
print(f"Веса вкладов")
print(f"\tMSFT: {optimal_weights_anal[0]}")
print(f"\tTSLA: {optimal_weights_anal[1]}")
print(f"Безрисковый: {rf_weight(optimal_weights_anal)}")
print(f"{'Done':-^40}")


# БЭКТЕСТ

print(f"{'Бэктест':=^50}")

print(f"{'Считаем месячные доходности акций':-^40}")
tmp1, tmp2 = test_df[['Date', 'Close_MSFT']], test_df[['Date', 'Close_TSLA']]

tmp1['Close'] = tmp1['Close_MSFT']
tmp1 = tmp1[['Date', 'Close']]

tmp2['Close'] = tmp2['Close_TSLA']
tmp2 = tmp2[['Date', 'Close']]

data_1 = prepare_data(tmp1)
data_2 = prepare_data(tmp2)

print(data_1[:5])
print(data_2[:5])
print(f"{'Done':-^40}")

test_data = np.array([data_1, data_2])

test_return_ratio = calc_portfolio(optimal_weights_anal, test_data.mean(axis=1))
print(f"Среднемесячная ожидаемая норма дохожности равна : {test_return_ratio * 100}%")


std = optimal_weights_anal.T @ np.cov(test_data) @ optimal_weights_anal
sharp_ratio = (test_return_ratio -  RISK_FREE_RATE_MONTH) / np.sqrt(std)
print(f"Sharp ratio : {sharp_ratio}")

print(f"{'Done':=^40}")