task_3_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. # безрисковая процентная ставка(в месяц)
MAXIMAL_VOLATILITY = .02

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 | cvx.Variable) -> float:
    return 1 - np.sum(weights) if isinstance(weights, np.ndarray) else cvx.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('AAPL.csv')
df_1['Close_AAPL'] = df_1.Close
df_2 = pd.read_csv('AMZN.csv')
df_2['Close_AMZN'] = df_2.Close
df_3 = pd.read_csv('IBM.csv')
df_3['Close_IBM'] = df_2.Close
df = df_1.merge(df_2.merge(df_3, on='Date', how='inner'), on='Date', how='inner')[['Date', 'Close_AAPL', 'Close_AMZN', 'Close_IBM']]
print(df_1.head(5))
print(df_2.head(5))
print(df_3.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, tmp3 = \
    train_df[['Date', 'Close_AAPL', 'Close_AMZN', 'Close_IBM']], \
    train_df[['Date', 'Close_AAPL', 'Close_AMZN', 'Close_IBM']], \
    train_df[['Date', 'Close_AAPL', 'Close_AMZN', 'Close_IBM']]

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

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

tmp3['Close'] = tmp3['Close_IBM']
tmp3 = tmp3[['Date', 'Close']]


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

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


# CAPM
print(f"{'CAPM':-^40}")
train_data = np.array([data_1, data_2, data_3])

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


optimal_weights = cvx.Variable(3)
portfolio_return = calc_portfolio(optimal_weights, expected_returns)
portfolio_variance = optimal_weights.T @ covariance_returns @ optimal_weights
optim_task = cvx.Maximize(portfolio_return)
constraints = [
    portfolio_variance <= MAXIMAL_VOLATILITY ** 2,
]

problem = cvx.Problem(optim_task, constraints)
problem.solve()
print(f"Оптимальные веса(Выпуклая оптимизация): {optimal_weights.value}")
print(f"{'Done':-^40}")


print(f"{'Результат':-^40}")
print(f"Норма доходности: { portfolio_return.value * 100 }%")
print(f"Веса вкладов")
print(f"\tAAPL: {optimal_weights.value[0]}")
print(f"\tAMZN: {optimal_weights.value[1]}")
print(f"\tIBM: {optimal_weights.value[2]}")
print(f"\tБезрисковый: {rf_weight(optimal_weights.value)}")
print(f"{'Done':-^40}")


print(f"{'Бектестинг':-^40}")
tmp1, tmp2, tmp3 = \
    test_df[['Date', 'Close_AAPL', 'Close_AMZN', 'Close_IBM']], \
    test_df[['Date', 'Close_AAPL', 'Close_AMZN', 'Close_IBM']], \
    test_df[['Date', 'Close_AAPL', 'Close_AMZN', 'Close_IBM']]

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

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

tmp3['Close'] = tmp3['Close_IBM']
tmp3 = tmp3[['Date', 'Close']]
test_data = np.array([data_1, data_2, data_3])

test_return_ratio = calc_portfolio(optimal_weights.value, test_data.mean(axis=1))
std = optimal_weights.value.T @ np.cov(test_data) @ optimal_weights.value
sharp_ratio = (test_return_ratio -  RISK_FREE_RATE_MONTH) / np.sqrt(std)
print(f"Ожидаемая норма доходности: { test_return_ratio * 100 }%")
print(f"Sharpratio: {sharp_ratio}")

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