Привет, Хабр! Меня зовут Андрей, 27 лет, ныне студент 4 курса мехмата. Эта статья – не очередной манифест про «успешный алготрейдинг», а скорее срез моей текущей дипломной научно-исследовательской работы (которая возможно будет интересна и вам, а возможно и вы сможете подсказать новые интересные векторы развития). И, самое главное, напомнить (или показать), как разбиваются красивые «бумажные» университетские теории о грязную реальность биржевой микроструктуры.
Постановка задачи: Написание модели ежедневной торговли, которая в первую очередь способна системно не терять деньги в моменты жестких рыночных шоков (коих на российском рынке за последнее десятилетие хватало с избытком).
Если мы не можем верить ковариациям (добрый день оставшимся любителям теории Марковица), куда двигаться? Ответ индустрии – минимизация метрики CVaR, которая математически точно отвечает на вопрос: «Если всё пойдет совсем плохо и мы окажемся, например, в 5% худших рыночных сценариев, каков будет наш средний убыток?».
Для достижения цели казалось, что осталось всего ничего: спроектировать и реализовать на Python инвестиционную систему, которая способна адаптивно переключать рыночные режимы, рассчитывать индекс внутренней паники и контролировать хвостовые риски, не удушая при этом рыночную доходность (CAGR) стратегии.
На деле все куда проблемнее, но оставим нытье до главы 2. А пока:
Глава 1. Контур тотального стресс-тестирования
Чтобы убедиться, что модель – это не случайная подгонка кривой под исторический график, стоит устроить системе два типа проверок:
Из более очевидного – сквозной 11-летний исторический трек (2015–2026 гг.): Симуляция работы портфеля в трех жизненных парадигмах инвестора – разовый старт капитала (Lump-Sum), инфляционно-корректируемое ежемесячное накопление (DCA) и фаза ренты (FIRE) с регулярным изъятием средств на жизнь.
Из того, к чему пришел позже – краш-тест Монте-Карло: 250 независимых прогонов на случайных подмножествах выбранных из ~70 активов рынка РФ на случайных интервалах времени (3, 4 и 5 лет). Причем в пул активов были честно включены «трупы» исторических делистингов и банкротств, а в структуре портфеля обязательно присутствовали три типа защитных гаваней: денежный рынок (LQDT/РЕПО), государственные облигации (ОФЗ) и драгметаллы (золото/серебро) для хэджирования системных рисков.
В этой статье я пошагово раскрою всю изнанку проекта: от архитектуры ETL-конвейера очистки данных до результатов финального противостояния моделей на графиках эквити.
Глава 2. Инженерный ад подготовки данных и борьба за чистый бэктeст
Не знаю, как состоят дела у разработчиков, а именно имеется ли у них на руках идеальный датасет (желательно еще и минутных таймфреймов), но в моей реальности 70% времени ушло не на написание красивых моделей выпуклой оптимизации, а на ручную, тяжелую очистку «токсичных отходов», которые выдают биржевые API (не без помощи специально написанных для этого модулей).
Рынок РФ специфичен, концентрирован и полон инфраструктурных сюрпризов. Чтобы все возможные прогоны капитала имели хоть какой-то реальный смысл, я спроектировал инкрементальный конвейер обработки данных (ETL), состоящий из цепочки классов MoexISSClient → LocalCSVStorage → DataCleaner. Вот несколько из фундаментальных проблем данных, которые мне пришлось побеждать на уровне архитектуры кода:
1. Синтетический ретроспективный синтез Денежного рынка (LQDT + REPO)
Для построения относительных метрик риска критически необходим трек безрисковой ставки. На российском рынке идеальным прокси является фонд денежного рынка LQDT (основанный на операциях РЕПО с Центральным Контрагентом). Но проблема в том, что фонд LQDT физически был запущен только в 2022 году. Что делаем если защитного актива половину времени не существовало? Правильно. применяем метод обратной ретроспективной экстраполяции: алгоритм находит цену пая в первый официальный день торгов фонда LQDT, запрашивает исторические данные индекса MOEXREPO (ставки РЕПО с ЦК) назад до 2013 года (чтобы с запасом перекрыть стартовую точку симуляции в 2015 году получив еще и нужные исторические данные) и с помощью ежедневного сложного процента рассчитывает непрерывную синтетическую цену пая LQDT:
Также это дало моделям возможность инвестиций в безрисковый инструмент на всем историческом горизонте.
2. Автоматический детектор технических сплитов и консолидаций
Сырая история цен закрытия (CLOSE, LOW, HIGH и пр.) на Мосбирже хранит исторические цены «как есть». Но мы знаем что сплиты существуют и если скормить эти данные оптимизатору, то в день сплита модель зафиксирует ложный катастрофический обвал акции на 99%, волатильность улетит в космос, и алгоритм навсегда забанит этот актив (в худшем случае якобы обнулив капитал).
Так модуль DataCleaner заимел проверку временных рядов на предмет аномальных cross-day скачков цены (
или 
). При обнаружении аномалии алгоритм подбирает реальный коэффициент деноминации биржи и пропорционально масштабирует всю историческую кривую цен «влево» от события до самого начала истории, полностью сохраняя процентные доходности, но выравнивая масштаб цен под современные реалии.
3. Переход к волатильности Паркинсона
Очень быстро пришло осознание, что требуется мера риска актива, но классическое стандартное отклонение доходностей цен закрытия слишком мало несет информации, да и запаздывает, а значит требовался более чувствительный инструмент. Вместо цен закрытия в конвейер был внедрен оценщик Паркинсона, основанный на экстремумах торговой сессии (HIGH и LOW), Однако для защиты от ложных гэпов открытия на нашем волатильном рынке классическая формула была модифицирована добавлением цены закрытия (CLOSE) прошлого дня:
Дополнительно в конвейер заложена Point-in-Time очистка дивидендов (с автоматическим удержанием налога 13%, поиском даты отсечки на Т+2 и конвертацией валютных выплат по курсу ЦБ на дату реестра), а также принудительное списание капитала в дефолтных и делистингованных эмитентах в -100%.
Посмотреть исходный код ядра математической очистки DataCleaner
class DataCleaner: def __init__(self, storage: LocalCSVStorage): self.storage = storage def build_cleaned_market_data(self, assets_config: dict, target_start: str, delist_history: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: close_series_dict = {} high_series_dict = {} low_series_dict = {} for asset_name, stages in assets_config.items(): df = self.storage.get_market_data(asset_name=asset_name, stages=stages, target_start=target_start) if not df.empty: close_series_dict[asset_name] = df[f"{asset_name}_close"] high_series_dict[asset_name] = df[f"{asset_name}_high"] low_series_dict[asset_name] = df[f"{asset_name}_low"] price_matrix = pd.concat(close_series_dict.values(), axis=1,keys=close_series_dict.keys()).sort_index().replace(0.0, np.nan) high_matrix = pd.concat(high_series_dict.values(), axis=1, keys=high_series_dict.keys()).sort_index().replace(0.0, np.nan) low_matrix = pd.concat(low_series_dict.values(), axis=1, keys=low_series_dict.keys()).sort_index().replace(0.0,np.nan) if 'Денежный рынок(REPO)' in price_matrix.columns and 'Денежный рынок(LQDT)' in price_matrix.columns: price_matrix['Денежный рынок(REPO)'] = price_matrix['Денежный рынок(REPO)'].ffill() first_lqdt_date = price_matrix['Денежный рынок(LQDT)'].first_valid_index() t_ipo = price_matrix.index.get_loc(first_lqdt_date) base_lqdt_price = price_matrix.loc[first_lqdt_date, 'Денежный рынок(LQDT)'] repo_rates = price_matrix['Денежный рынок(REPO)'].to_numpy() synthetic_prices = np.zeros(len(price_matrix)) synthetic_prices[t_ipo] = base_lqdt_price for t in range(t_ipo - 1, -1, -1): daily_repo_rate = (1.0 + (repo_rates[t] / 100.0)) ** (1.0 / 252.0) synthetic_prices[t] = synthetic_prices[t + 1] / daily_repo_rate df_synthetic = pd.Series(synthetic_prices[:t_ipo], index=price_matrix.index[:t_ipo]) price_matrix['Денежный рынок(LQDT)'] = price_matrix['Денежный рынок(LQDT)'].combine_first(df_synthetic) high_matrix['Денежный рынок(LQDT)'] = high_matrix['Денежный рынок(LQDT)'].combine_first(df_synthetic) low_matrix['Денежный рынок(LQDT)'] = low_matrix['Денежный рынок(LQDT)'].combine_first(df_synthetic) for col in price_matrix.columns: f_idx = price_matrix[col].first_valid_index() if f_idx is not None: price_matrix.loc[f_idx:, col] = price_matrix.loc[f_idx:, col].ffill() high_matrix.loc[f_idx:, col] = high_matrix.loc[f_idx:, col].ffill() low_matrix.loc[f_idx:, col] = low_matrix.loc[f_idx:, col].ffill() for ticker in price_matrix.columns: if ticker in ['Денежный рынок(REPO)', 'Денежный рынок(LQDT)']: continue p = price_matrix[ticker].to_numpy() for t in range(1, len(p)): if pd.isna(p[t]) or pd.isna(p[t - 1]): continue if p[t] / p[t - 1] <= 0.4 or p[t] / p[t - 1] >= 2.5: for ratio in [1000.0, 100.0, 20.0, 10.0, 8.0, 3.0, 0.1, 0.01, 0.001, 0.0002]: if p[t - 1] * 0.8 <= p[t] * ratio <= p[t - 1] * 1.2: break else: ratio = p[t - 1] / p[t] price_matrix.iloc[:t, price_matrix.columns.get_loc(ticker)] /= ratio high_matrix.iloc[:t, high_matrix.columns.get_loc(ticker)] /= ratio low_matrix.iloc[:t, low_matrix.columns.get_loc(ticker)] /= ratio returns_matrix = (price_matrix - price_matrix.shift(1)) / price_matrix.shift(1) div_matrix = pd.DataFrame(0.0, index=price_matrix.index, columns=price_matrix.columns) translate_cur = {'USD':'Доллар', 'EUR':'Евро'} for asset_name, stages in assets_config.items(): if stages[0]['market'] != 'shares' or stages[0]['secid'] == 'LQDT': continue tickers = {st['secid'] for st in stages} df_div = self.storage.get_dividends_data(asset_name=asset_name, tickers=list(tickers)) if df_div.empty: continue for idx, row in df_div.iterrows(): if idx < price_matrix.index.min() or idx > price_matrix.index.max(): continue payout_date = div_matrix.index[div_matrix.index >= idx][0] if idx not in div_matrix.index else idx div_value, currency = float(row['value']), str(row['currency']).upper().strip() if currency in ['USD', 'EUR']: fx_date = price_matrix.index[price_matrix.index <= idx][-1] if idx not in price_matrix.index else idx rub_value = div_value * price_matrix.loc[fx_date, translate_cur[currency]] else: rub_value = div_value price_date = price_matrix.index[price_matrix.index < idx][-1] rub_value *= 0.87 high_matrix.loc[payout_date, asset_name] += rub_value low_matrix.loc[payout_date, asset_name] += rub_value div_matrix.loc[payout_date, asset_name] += rub_value / price_matrix.loc[price_date, asset_name] prev_close_matrix = price_matrix.shift(1) robust_high = np.maximum(high_matrix.to_numpy(), prev_close_matrix.to_numpy()) robust_low = np.minimum(low_matrix.to_numpy(), prev_close_matrix.to_numpy()) robust_low = np.where(robust_low == 0, 1e-8, robust_low) const_factor = 1.0 / (4.0 * np.log(2.0)) vol_numpy = np.sqrt(const_factor * (np.log(robust_high / robust_low) ** 2)) vol_matrix = pd.DataFrame(vol_numpy, index=price_matrix.index, columns=price_matrix.columns) for col in price_matrix.columns: first_valid_idx = price_matrix[col].first_valid_index() if first_valid_idx is not None: vol_matrix.loc[:first_valid_idx, col] = np.nan post_ipo_slice = vol_matrix.loc[first_valid_idx:, col].to_numpy() if col == 'Денежный рынок(LQDT)': post_ipo_slice = np.where(np.isnan(post_ipo_slice) | (post_ipo_slice < 0.0001), 0.0001, post_ipo_slice) else: post_ipo_slice = np.where(np.isnan(post_ipo_slice) | (post_ipo_slice == 0.0), 0.0005, post_ipo_slice) vol_matrix.loc[first_valid_idx:, col] = post_ipo_slice for m in [returns_matrix, div_matrix, high_matrix, low_matrix, vol_matrix]: m.drop(columns=['Денежный рынок(REPO)'], inplace=True) for delist_company in delist_history.index: delist_date = delist_history.loc[delist_company, "Date"] if delist_company in returns_matrix.columns: returns_matrix.loc[delist_date:, delist_company] = np.nan returns_matrix.loc[delist_date, delist_company] = -1.0 vol_matrix.loc[delist_date:, delist_company] = np.nan if delist_history.loc[delist_company, "Currency"] in ['USD', 'EUR']: fx_date = price_matrix.index[price_matrix.index <= delist_date][-1] delist_history.loc[delist_company, "Amount"] *= price_matrix.loc[fx_date, translate_cur[delist_history.loc[delist_company, "Currency"]]] price_date = price_matrix.index[price_matrix.index < delist_date][-1] delist_history.loc[delist_company, "Amount"] /= price_matrix.loc[price_date, delist_company] delist_history.drop(columns=['Currency'], inplace=True) delist_history.to_csv('data/matrix/global_delist_panel.csv') return returns_matrix, div_matrix, vol_matrix
Глава 3. Анатомия дискретного симулятора, получилось ли избежать ошибки джунов?
Самая частая ошибка при создании торговых систем о которой был наслышан – это Look-Ahead Bias (заглядывание в будущее), хотя она же интуитивно самая очевидная. Написать классный оптимизатор весов на исторических данных не так сложно, но заставить его работать в дискретных петлях времени так, как он бы функционировал на реальном биржевом терминале – это отдельная инженерная задача.
Мой симуляционный контур разделен на два независимых модуля: Оркестратор (PortfolioOrchestrator), который пошагово формирует инвестиционную вселенную, и Бэктестер (PortfolioBacktester), который обсчитывает движение капитала. Вот три архитектурных барьера, которые я внедрил в код, чтобы симуляция соответствовала суровой реальности:
1. Строгая Point-in-Time изоляция скользящего окна
На каждом шаге ребалансировки портфеля current_date оркестратор запрашивает исторический срез данных. Чтобы полностью исключить просачивание информации из будущего, внутри цикла реализована жесткая отсечка:
historical_slice = self.board_panel[self.asset_tickers].loc[:current_date] historical_slice = historical_slice.iloc[:-1]
Мы запрашиваем срез включая текущий день, но принудительно отрезаем его через .iloc[:-1]. Оптимизатор принимает решения утром текущего дня, видя историю строго до вчерашнего закрытия. Если какая-то бумага еще не вышла на IPO на этот день или уже прошла делистинг, маска active_mask динамически изолирует её из вектора оптимизации.
2. Динамический пересчет рыночного дрейфа весов
Рассчитывать транзакционные издержки, сравнивая новые веса с теми весами, которые модель выставила на прошлой ребалансировке – уже хорошо. Но за время между шагами цены активов изменились, а значит, и их реальные доли в портфеле «уплыли». Реализовать пересчет рыночного дрейфа долей относительно фактического текущего состояния портфеля перед списанием комиссий – отлично:
drifted_weights = prev_target_weights * (1.0 + prev_day_returns) if (spv := np.sum(drifted_weights)) > 0: drifted_weights /= spv
Мы берем целевые веса прошлого дня, умножаем их на реальную вчерашнюю доходность активов и ренормализуем вектор. Штраф за оборот капитала (turnover_penalty) считается оптимизатором от фактического состава портфеля на текущее утро, что гарантирует точный расчет комиссий брокера.
3. Моделирование Т+15 дивгэп-лага(а также при делистинге) и ежегодного НДФЛ 13%
Продолжение пути к реальным биржевым условиям, ведь дивиденды не падают на торговый счет в день отсечки – деньги идут через депозитарную цепочку от 10 до 25 рабочих дней. Бэктестер содержит очередь выплат (payout_queue) с жестким лагом для дивов в 15 торговых дней. Стратегия не может реинвестировать дивиденды мгновенно на дивгэпе: деньги «висят в воздухе» три недели, а на границе исторической выборки они честно дисконтируются назад к терминальной дате через безрисковую ставку LQDT.
Аналогичная логика применена и к инфраструктурным дефолтам: при наступлении делистинга компания не исчезает мгновенно — её остаточная ликвидационная стоимость возвращается на баланс строго с заложенным в матрицу временным лагом (Pay_lag), симулируя реальные сроки внебиржевого выкупа или расчетов. Кроме того, в конце каждого календарного года бэктестер фиксирует финансовый результат, рассчитывает налоговую базу и принудительно списывает НДФЛ 13%, формируя честную чистую кривую капитала, очищенную от фискальной нагрузки.
Посмотреть исходный код ядра симуляции BacktestEngine
class PortfolioOrchestrator: def __init__(self, board_panel: pd.DataFrame, volatility_panel: pd.DataFrame, strategies: list, commission: float = 0.0005): self.board_panel = board_panel.sort_index() self.strategies = strategies self.commission = commission self.asset_tickers = [col for col in self.board_panel.columns if not col.endswith('_div')] self.volatility_panel = volatility_panel.sort_index() def generate_weights_history(self, start_date: str = "2015-01-01", end_date: str = "2027-01-01", assets: str = None) -> dict[str, pd.DataFrame]: self.asset_tickers = [col for idx, col in enumerate(self.asset_tickers) if assets[idx] == '1'] if assets else self.asset_tickers test_returns = self.board_panel[self.asset_tickers].loc[start_date:end_date] sim_dates = test_returns.index weights_histories = { strat.name: pd.DataFrame(0.0, index=sim_dates, columns=self.asset_tickers) for strat in self.strategies } current_weights = {strat.name: np.zeros(len(self.asset_tickers)) for strat in self.strategies} for t_idx, current_date in enumerate(sim_dates): logging.info(f"Завершен рассчет на дату {current_date}") for strat in self.strategies: name = strat.name historical_slice = self.board_panel[self.asset_tickers].loc[:current_date] last_two_days = historical_slice.tail(2) historical_slice = historical_slice.iloc[:-1] active_mask = ~last_two_days.isna().any().to_numpy() live_cols = [col for idx, col in enumerate(self.asset_tickers) if active_mask[idx]] cleaned_slice = historical_slice[live_cols] vol_slice = self.volatility_panel[live_cols].loc[:current_date] vol_slice = vol_slice.iloc[:-1] div_cols = [f"{col}_div" for col in live_cols] div_slice = self.board_panel[div_cols].loc[:current_date].iloc[:-1] total_return_slice = cleaned_slice.to_numpy() + div_slice.to_numpy() historical_returns_adjusted = pd.DataFrame(total_return_slice, index=cleaned_slice.index, columns=live_cols) last_day_returns = np.nan_to_num(historical_slice.iloc[-1].to_numpy(), nan=0.0) live_prev_weights = (current_weights[name] * (1.0 + last_day_returns))[active_mask].copy() if np.sum(live_prev_weights) > 0: live_prev_weights = live_prev_weights / np.sum(live_prev_weights) else: live_prev_weights = np.zeros(len(live_cols)) try: live_new_weights = strat.optimize_weights(historical_returns_adjusted, live_prev_weights, vol_slice) live_new_weights = np.nan_to_num(live_new_weights, nan=0.0) except Exception as e: logging.error(f"Крах стратегии {name} на дату {current_date}: {e}") live_new_weights = live_prev_weights new_global_weights = np.zeros(len(self.asset_tickers)) global_live_indices = [self.asset_tickers.index(c) for c in live_cols] new_global_weights[global_live_indices] = live_new_weights current_weights[name] = new_global_weights weights_histories[name].iloc[t_idx] = current_weights[name] return weights_histories class PortfolioBacktester: def __init__(self, board_panel: pd.DataFrame, delist_history: pd.DataFrame, inflation_annual: float = 0.075, commission: float = 0.0005): self.board_panel = board_panel.sort_index() self.daily_inflation = (1.0 + inflation_annual) ** (1.0 / 252.0) - 1.0 self.commission = commission self.delist_history = delist_history self.lqdt = board_panel['Денежный рынок(LQDT)'] def _sim_core(self, weights_history: pd.DataFrame, initial_capital: float, extra_capital: float, scenario_type: str, wherewithal: float = 0.0, tax: float = 0.13) -> tuple[np.ndarray, np.ndarray]: sim_dates = weights_history.index prices_subset = self.board_panel.loc[sim_dates] lqdt_last = self.lqdt.index.get_loc(sim_dates[-1]) asset_tickers = [col for col in self.board_panel.columns if not col.endswith('_div')] div_tickers = [f"{ticker}_div" for ticker in asset_tickers] returns_matrix = prices_subset[asset_tickers] ticker_to_idx = {ticker: idx for idx, ticker in enumerate(returns_matrix)} returns_matrix = returns_matrix.to_numpy() div_yield_matrix = prices_subset[div_tickers].to_numpy() weights_matrix = weights_history[asset_tickers].to_numpy() T = len(sim_dates) portfolio_values = np.zeros(T) benchmark_values = np.zeros(T) portfolio_values[0] = initial_capital benchmark_values[0] = initial_capital if scenario_type != "DCA" else extra_capital prev_year_cap = initial_capital payout_queue = [] key_deposit = True for t in range(1, T): prev_capital = portfolio_values[t - 1] is_new_month = sim_dates[t].month != sim_dates[t - 1].month current_weights = weights_matrix[t] prev_target_weights = weights_matrix[t - 1] prev_day_returns = np.nan_to_num(returns_matrix[t - 1], nan=0.0) day_returns = returns_matrix[t] clean_day_returns = np.nan_to_num(day_returns, nan=0.0) drifted_weights = prev_target_weights * (1.0 + prev_day_returns) if (spv := np.sum(drifted_weights)) > 0: drifted_weights /= spv else: drifted_weights = np.zeros_like(current_weights) day_div_yields = np.nan_to_num(div_yield_matrix[t], nan=0.0) dividend_accrued = prev_capital * np.nansum(current_weights * day_div_yields) if dividend_accrued > 0: if t + 15 < T: payout_queue.append((t + 15, dividend_accrued)) elif (end_idx := lqdt_last + (t + 16 - T)) <= len(self.lqdt): payout_queue.append((T - 1, dividend_accrued / np.prod(1 + self.lqdt.iloc[lqdt_last + 1 : end_idx]))) else: payout_queue.append((T - 1, dividend_accrued / (np.mean(1 + self.lqdt.iloc[lqdt_last - 4 : lqdt_last + 1])) ** (t + 16 - T))) if sim_dates[t] in self.delist_history.index: delist_row = self.delist_history.loc[sim_dates[t]] if (comp := str(delist_row['Company'])) in self.board_panel.columns: if t + int(delist_row['Pay_lag']) < T: payout_queue.append((t + int(delist_row['Pay_lag']), prev_capital * current_weights[ticker_to_idx[comp]] * float(delist_row['Amount']))) elif (end_idx := lqdt_last + (t + 1 + int(delist_row['Pay_lag']) - T)) <= len(self.lqdt): payout_queue.append((T - 1, prev_capital * current_weights[ticker_to_idx[comp]] * float(delist_row['Amount']) / np.prod(1 + self.lqdt.iloc[lqdt_last + 1: end_idx]))) else: payout_queue.append((T - 1, prev_capital * current_weights[ticker_to_idx[comp]] * float(delist_row['Amount']) / (np.mean(1 + self.lqdt.iloc[lqdt_last - 4: lqdt_last + 1])) ** (t + 1 + int(delist_row['Pay_lag']) - T))) current_asset_values = prev_capital * drifted_weights if sim_dates[t].year != sim_dates[t - 1].year: tax_base = prev_capital - prev_year_cap if tax_base > 0: tax_amount = tax_base * tax prev_capital -= tax_amount prev_year_cap = prev_capital payout = 0.0 ready_2_pay = [item for item in payout_queue if item[0] <= t] payout_queue = [item for item in payout_queue if item[0] > t] for item in ready_2_pay: payout += item[1] prev_capital += payout cash_flow = 0.0 if scenario_type == "LUMPSUM": benchmark_values[t] = benchmark_values[t - 1] * (1.0 + self.lqdt[sim_dates[t]]) elif scenario_type == "DCA": extra_capital *= (1.0 + self.daily_inflation) month_replenishment = extra_capital if is_new_month else 0.0 cash_flow = month_replenishment benchmark_values[t] = benchmark_values[t - 1] * (1.0 + self.lqdt[sim_dates[t]]) + month_replenishment elif scenario_type == "FIRE": wherewithal *= (1.0 + self.daily_inflation) monthly_withdrawal = wherewithal if is_new_month else 0.0 cash_flow = -monthly_withdrawal if key_deposit and (benchmark_values[t - 1] * (1.0 + self.daily_inflation) - monthly_withdrawal > 0): benchmark_values[t] = benchmark_values[t - 1] * (1.0 + self.lqdt[sim_dates[t]]) - monthly_withdrawal else: benchmark_values[t] = 0.0 key_deposit = False prev_capital += cash_flow if prev_capital <= 0: portfolio_values[t] = 0.0 if scenario_type == "FIRE": break continue target_asset_values = prev_capital * current_weights turnover_rub = np.sum(np.abs(target_asset_values - current_asset_values)) transaction_cost = turnover_rub * self.commission prev_capital = max(0.0, prev_capital - transaction_cost) capital_growth = np.sum(current_weights * clean_day_returns) prev_capital *= (1.0 + capital_growth) portfolio_values[t] = prev_capital return portfolio_values, benchmark_values def run_lumpsum_simulation(self, strategy_weights_history: pd.DataFrame, initial_capital: float = 1_000_000.0) -> pd.DataFrame: p_val, b_val = self._sim_core(strategy_weights_history, initial_capital, 0.0, "LUMPSUM", 0.0, 0.13) p_val_taxless, _ = self._sim_core(strategy_weights_history, initial_capital, 0.0, "LUMPSUM", 0.0, 0.0) res = pd.DataFrame(index=strategy_weights_history.index) res['Nominal_Capital'] = p_val res['Nominal_Capital_Taxless'] = p_val_taxless res['Inflation_Benchmark'] = b_val return res def run_dca_simulation(self, strategy_weights_history: pd.DataFrame, extra_capital: float = 50_000.0) -> pd.DataFrame: p_val, b_val = self._sim_core(strategy_weights_history, extra_capital, extra_capital, "DCA") res = pd.DataFrame(index=strategy_weights_history.index) res['Nominal_Capital'] = p_val res['Inflation_Benchmark'] = b_val return res def run_fire_simulation(self, strategy_weights_history: pd.DataFrame, initial_capital: float = 6_000_000.0, wherewithal: float = 60_000.0) -> pd.DataFrame: p_val, b_val = self._sim_core(strategy_weights_history, initial_capital, 0.0, "FIRE", wherewithal) res = pd.DataFrame(index=strategy_weights_history.index) res['Nominal_Capital'] = p_val res['Inflation_Benchmark'] = b_val return res[(res['Nominal_Capital'] > 0) | (res['Inflation_Benchmark'] > 0)]
Глава 4. Великое противостояние моделей на исторических данных и Монте-Карло
После того как инфраструктурный конвейер очистки и петля бэктестера были построены, пришло время выпустить модели на арену. Как обсуждалось ранее проводим анализ в двух плоскостях: сперва сквозной исторический трек за 11 лет, а после масштабное стресс-тестирование методом Монте-Карло пока на 250 случайных выборках активов и таймлайнов.
Часть 1: С сквозной 11-летний трек (Парадокс канонических моделей)
Когда мы запускаем симуляцию на историческом промежутке с 2015 по 2026 год, на графиках эквити разворачивается удивительный сюжет. Мы наглядно видим, почему классические подходы проигрывают в реальном мире:
Классический не-робастный CVaR (Base CVaR): Пришлось помучиться, ведь чтобы модель послужила должным бенчмарком потребовалось незначительно модернизировать «эталонный» алгоритм (что в оригинале вел себя как «трусливый бот» сидя в LQDT на 99%) добавив максимальные лимиты, хотя и в этом случае оптимизатор Рокфеллера — Урьясева, напуганный регулярными шоками рынка РФ, забивается на все доступные лимиты (40%) в фонд денежного рынка LQDT и уходит в глухую оборону. Результат закономерный – модель практически не совершает глубоких просадок, но в моменты высоких ставок сливает даже инфляционному бенчмарку, фиксируя скромный CAGR.
Линейно-взвешенный Momentum-ротатор: Полная противоположность. На сильных, затяжных трендах Моментум летит вверх быстрее индекса Мосбиржи. Но в моменты внезапных шоков эта стратегия оказывается запертой в самых перегретых бумагах. Алгоритм ловит «черного лебедя» всем объемом капитала и падает камнем вниз под инфляционную линию.
Классический Марковиц: Из-за высокой чувствительности к шумам и техническим аномалиям ковариационной матрицы, Марковиц в моменты смены режимов рынка теряет ориентиры и уходит под инфляционный бенчмарк, показывая отрицательный коэффициент Сортино.
Эволюция робастных прототипов: Путь к флагману 0.1 (а с вашей помощью возможно и развитие)
Мои собственные модели на этом треке продемонстрировали четкую эволюционную цепочку:
Прототип с высоким риском (High-risk prototype): Агрессивно собирает рыночную доходность на растущих фазах, ловит мощные импульсы, но обладает повышенной волатильностью.
Прототип с контролируемым риском (Controlled-risk prototype): В моменты экстремального шторма 2022 года этот алгоритм по триггерам индекса паники мгновенно сворачивает рисковые позиции, уходит в LQDT/ОФЗ (и все прочее что считает выгодным) и теряет меньше, почти полностью защищая капитал от рыночной катастрофы (при этом имея возможность «недозаработать»).
Флагманская модель (Robust CVaR 0.1): «Сбалансированный Грааль» системы, что на деле пока первичная комбинация моделей выше. Заметны попытки сохранить защиту консервативной версии, но (не без помощи высокорисоковой) показывает более выдающийся CAGR на уровне 17.56% при коэффициенте Сортино 0.92 (не считая тестов без токсичных активов – лучший, на момент написания статьи, результат).
Часть 2: 250 прогонов Монте-Карло (Финальный вердикт)
Чтобы доказать, что исторический трек – это не случайное совпадение, симуляция polytest обсчитала 250 случайных Out-of-Sample окон. Статистика t-баллов relative-Сортино расставила всё по местам:
Против портфеля 1/N и Random Monkey модель Robust CVaR 0.1 выигрывает со статистической значимостью в 43–45% случаев, уходя в нейтральную ничью в 41–44% и уступая лишь в ~13% прогонов (причем в зонах поражений отставание по CAGR минимально и составляет всего около 2%).
В битве против Марковица и Base CVaR флагман демонстрирует тотальное доминирование – 50.0% и 56.4% чистых побед со средним t-баллом выборки до +1.94. Графики плотности распределения рисков (KDE) подтверждают: холм распределения максимальной просадки (Max Drawdown) у робастных прототипов является самым узким и высоким, у него полностью отсутствуют «тяжелые левые хвосты» глубоких убытков, которые размазаны у Марковица и Моментума до -40% и ниже.
Ультимативный критерий «1 vs ALL» (Один против всех): Дополнительным тестом стал интегральный зачет, где модель считалась проигравшей, если уступала хотя бы одному из четырех классических бенчмарков на случайном отрезке. Результаты разметки подтвердили триумф эволюции моделей: если консервативный прототип
Newиз-за параноидального риск-оффа уступал в 74.4% случаев, то у финального флагмана0.1доля поражений в режиме «один против всех» упала до исторического минимума. Модель0.1уже ‘научилась’ ультимативно забирать подиум, избегая уязвимостей, в которые ранние прототипы периодически проваливались (как на бычьих, так и на медвежьих рынках).
Распределение полной и относительной CAGR моделей 
Плотность вероятностей KDE для максимальных просадок и хвостовых рисков 
Сравнительное распределение долей зон устойчивости
Глава 5. Финал? Открытый бэклог и точки роста
Если вы дочитали до этой части то вам или сильно скучно или пытаетесь понять не впал ли я в иллюзию, что «хакнул биржу». Данная двухмасштабная робастная CVaR-система находится в статусе стабильного рабочего прототипа, однако хожу все еще с недовольным лицом (хотя текущие результаты после запуска торгов – позабавили, да и порадовали). Однако чтобы модель была готова к работе на реальном институциональном капитале, я выделил для себя пять ключевых точек роста, над кодом которых планирую работать в течение следующего года:
ADV-ограничения на ликвидность: На текущий момент модель не учитывает дневные объемы торгов активов внутри случайных выборок Монте-Карло. Логичный шаг для апгрейда – внедрение жесткого линейного ограничения в солвер cvxpy: вес любого актива в портфеле не может превышать X% от его реального среднедневного объема торгов за последние 20 сессий (до этого брался стандартный лимит концентрации в 10%). Это защитит модель от Market Impact при работе с неликвидными бумагами 2-3 эшелонов.
Эндогенный расчет проскальзывания: Сейчас транзакционные издержки заложены в виде плоского L1-штрафа брокерской комиссии (0.0005). В реальности крупные сделки всегда сдвигают стакан против инвестора. Я планирую переписать turnover_penalty в виде квадратичной L2-регуляризации, завязанной на текущую волатильность и спред конкретного тикера, чтобы симулировать проскальзывание при исполнении крупных ордеров (раз уж на руках будут объемы торгов – чего бы и нет).
Исследование возможности перехода на меньший таймфрейм: Достаточно ли будет ограничения на доли, наличия транзакционных издержек и самой устойчивости модели, чтобы не выполнять лишних/шумовых сделок (или, выполняя, зарабатывать на них не только брокеру)? То, что исследовать и интересно, и необходимо в качестве шага в сторону полноценной торговой системы.
Полноценно реализовать объединенную модель: А также в целом доработать то, что, к сожалению, обсуждать нет смысла из-за закрытого кода (однако я это пишу, ведь ориентируясь даже на результаты бэктестов – кто-то может заметить особенности моделей, которые я мог не заметить или не обращать внимания).
Ну и из самого веселого – продолжать стресс-тестирования: Добавить возможность шорта (как минимум индекса в качестве хэджа). Посмотреть, что было бы, если бы активы, которые нравятся модели, падали сильнее, а те, что она покупала меньше – росли чаще? Проверить, одержит ли итоговая модель полную победу хотя бы в 30% случаев против агрессивного ML-алгоритма?
Вместо заключения: Как пройти этот путь самостоятельно?
Весь инфраструктурный конвейер проекта – от ISS MOEX парсера с инкрементальным кэшированием до дискретной петли бэктестера с дивгэп-лагами и налогами – я выложил в открытый доступ на GitHub. Там же лежат preprocompiled CSV-матрицы 250 прогонов Монте-Карло для самостоятельного анализа. Буду искренне рад конструктивной критике, идеям по улучшению робастности и советам от практикующих квантов (и даже алготрейдеров, тк и в этом направлении двигаться придется).
