Моя «первая» модель CVaR или «Как балансировать между умением выдерживать шторм и способностью заработать на рынке РФ»

Привет! Меня зовут Андрей, 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:

$text{Price}_{t}=frac{text{Price}_{t+1}}{left(1+frac{text{REPO}_{t}}{100}right)^{frac{1}{252}}}$

Также это дало моделям возможность инвестиций в безрисковый инструмент на всем историческом горизонте.

2. Автоматический детектор технических сплитов и консолидаций

Сырая история цен закрытия (CLOSE, LOW, HIGH и пр.) на Мосбирже хранит исторические цены «как есть». Но мы знаем что сплиты существуют и если скормить эти данные оптимизатору, то в день сплита модель зафиксирует ложный катастрофический обвал акции на 99%, волатильность улетит в космос, и алгоритм навсегда забанит этот актив (в худшем случае якобы обнулив капитал).

Так модуль DataCleaner заимел проверку временных рядов на предмет аномальных cross-day скачков цены ( $frac{P_t}{P_{t-1}}≤0.4$ или ). При обнаружении аномалии алгоритм подбирает реальный коэффициент деноминации биржи и пропорционально масштабирует всю историческую кривую цен «влево» от события до самого начала истории, полностью сохраняя процентные доходности, но выравнивая масштаб цен под современные реалии.

3. Переход к волатильности Паркинсона

Очень быстро пришло осознание, что требуется мера риска актива, но классическое стандартное отклонение доходностей цен закрытия слишком мало несет информации, да и запаздывает, а значит требовался более чувствительный инструмент. Вместо цен закрытия в конвейер был внедрен оценщик Паркинсона, основанный на экстремумах торговой сессии (HIGH и LOW), Однако для защиты от ложных гэпов открытия на нашем волатильном рынке классическая формула была модифицирована добавлением цены закрытия (CLOSE) прошлого дня:

$sigma ^{2}=frac{1}{4ln 2}cdot left(ln frac{max (text{HIGH},text{CLOSE})}{min (text{LOW},text{CLOSE})}right)^{2}$

Дополнительно в конвейер заложена 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 прогонов Монте-Карло для самостоятельного анализа. Буду искренне рад конструктивной критике, идеям по улучшению робастности и советам от практикующих квантов (и даже алготрейдеров, тк и в этом направлении двигаться придется).

GitHub: https://github.com/markelovandrey5454-ship-it/multi-model-portfolio-backtester ^[1]

Автор: Andrey5454

Источник ^[2]

Сайт-источник PVSM.RU: https://www.pvsm.ru

Путь до страницы источника: https://www.pvsm.ru/python/458025

Ссылки в тексте:

[1] https://github.com/markelovandrey5454-ship-it/multi-model-portfolio-backtester: https://github.com/markelovandrey5454-ship-it/multi-model-portfolio-backtester

[2] Источник: https://habr.com/ru/articles/1082296/?utm_source=habrahabr&utm_medium=rss&utm_campaign=1082296

Нажмите здесь для печати.