Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- 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}")
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement