17 июл 2019 Александр Мыльцев Все авторы
Работая с финансовыми данными, есть масса способов допустить ошибку. Многие из нас работали с большими таблицами временных рядов в EXCEL. Самой типичной, пожалуй, является работа с длинными цепочками данных цен закрытия какой-нибудь ценной бумаги. Надо ли говорить, что простейшие операции (например, вычисление доходности в другой валюте) требуют внимательности? Когда речь идет об одном – двух вычислениях, всё нормально. Можно заставить себя всё проверить. Но при повторяющихся многоходовых вычислениях ошибка просто неизбежна.
В этой статье мы расскажем, как контролировать правильность финансовых вычислений, связанных с временными рядами (Time Series), при помощи нового типа данных в Python.
Начнём с создания данных для двух временных рядов:
- значения индекса S&P с января 2015 года по март 2016
- значения инфляции США (CPI) с января 2015 года по март 2016
Для расчетов мы будем использовать функции финансовой библиотеки yapo
и numpy
для работы с временными рядами.
import yapo as yp # https://github.com/okama-io/yapo
import numpy as np
snp_asset = yp.portfolio_asset(name='us/SPY',
start_period='2015-1', end_period='2016-3', currency='usd')
snp_values = snp_asset.close().values
snp_values
infl_usd = yp.inflation(currency='usd', kind='values', start_period='2015-1', end_period='2016-3')
infl_usd_values = infl_usd.values
infl_usd_values
Примеры ошибок¶
При финансовых вычислениях есть много вариантов допустить ошибку. Рассмотрим наиболее распространенные из них.
snp_values # значение индекса S&P 500
infl_usd_values # значение инфляции
Первый тип ошибок¶
Начнем с рассчетов реального значения доходности для индекса S&P500.
(snp_values + 1.) / (infl_usd_values + 1.) - 1.
Несмотря на успешность выполнения операции, в этом подсчёте нет никакого смысла. Реальные значения доходности рассчитываются от доходности индекса, а не от его абсолютных значений. Ни numpy.array
, ни встроенные средства Python
не подсказывают об ошибке.
Но этот тип ошибок относительно безобиден, так как полученне значения выглядят неправдоподобно, и это легко заметить.
В целом этот тип ошибки можно отнести к операциям с несовместимыми данными. Так мы должны делать операции скорости со скоростью, а ускорения с ускорением. Здесь мы попытались использовать в одной формуле диснацию и скорость или нулевую производную с первой производной.
ВНИМАНИЕ: здесь и далее определение производной неалгебраическое, т.к. мы имеем дело со временными рядами, а не с непрерывными функциями.
Второй тип ошибки¶
Как говорилось в предыдущей статье, накопленная доходность считается так:
snp_ror = np.diff(snp_values) / snp_values[:-1]
np.testing.assert_equal(snp_asset.get_return().values, snp_ror)
snp_ror
Посчитаем реальные значения накопленной доходности.
Какой из следующих двух вариантов подсчёта реальной доходности правильный?
snp_ror_real_1 = (snp_ror + 1.) / (infl_usd_values[1:] + 1.) - 1.
snp_ror_real_1
snp_ror_real_2 = (snp_ror + 1.) / (infl_usd_values[:-1] + 1.) - 1.
snp_ror_real_2
# check
np.testing.assert_equal(
snp_asset.get_return(real=True).values,
snp_ror_real_1
)
Правильный ответ: первый.
Во втором варианте проблема была в неправильной проекции infl_usd_values
- временной ряд доходности с февраля 2015 по март 2016 был поделен на временной ряд инфляции с января 2015 по февраль 2016 года.
Теперь посчитаем реальное значение среднегодовой доходности (CAGR) за последний год:
years_ago = 1
snp_cagr = (snp_ror[-years_ago * 12:] + 1.).prod() ** (1 / years_ago) - 1.
# check:
assert snp_asset.cagr(years_ago=1).value == snp_cagr
snp_cagr
infl_usd_values[-years_ago*12:]
# check
assert snp_ror[-years_ago * 12:].shape == infl_usd_values[-years_ago * 12:].shape
infl_usd_accumulated = (infl_usd_values[-years_ago * 12 + 1:] + 1.).prod() - 1.
snp_cagr_real = (snp_cagr + 1.) / (infl_usd_accumulated + 1.) - 1.
snp_cagr_real
В канве статьи, здесь также есть ошибка. Оказывается, по аналогии с расчётом snp_ror_real
программист решил взять инфляцию без первого значения, т.е. за 11 месяцев вместо 12, затем посчитал аккумулированное значение, а по сути разделил одно число, равное произведению 12 чисел, на другое число, равное произведению 11 чисел!
Правильный расчёт:
years_ago = 1
snp_cagr = (snp_ror[-years_ago * 12:] + 1.).prod() ** (1 / years_ago) - 1.
# check:
assert snp_asset.cagr(years_ago=1).value == snp_cagr
infl_usd_accumulated = (infl_usd_values[-years_ago * 12:] + 1.).prod() - 1.
snp_cagr_real = (snp_cagr + 1.) / (infl_usd_accumulated + 1.) - 1.
# check:
assert snp_asset.cagr(years_ago=1, real=True).value == snp_cagr_real
snp_cagr_real
Второй тип ошибки, как можно заметить из примеров, связан с ситуациями, когда осуществляются математические операции с несовместимыми рядами, охватывающими разные отрезки времени. Этот тип ошибок еще более распространен в финасовых вычислениях, и гораздо сложнее выявляется, приводя зачастую к значительным погрешностям или неправильным результатам.
Как бороться с ошибками в финансовых вычислениях¶
Для отлова таких ошибок в Python необходима "расширенная" версия np.array
, которая бы:
- имела мета-информацию над значениями
np.array
:- начало и конец периода (дата и время)
- уровень частичных значений, чтобы отличать значения от производной и второй производной (и более высоких подярков, если такое понадобиться)
- иметь такой же интерфейс методов как
np.array
, включая арифметические операции, достаточное для текущих задач множество, но расширяемое по мере необходимости - валидировать каждый вызов метода: соответствующие начальный период, конечный период, и уровень частичных значений должны совпадать
Примером уровеней частичных значений для индекса S&P 500 может быть такой: значения индекса - уровень 0, доходность индекса - уровень 1. Инфляция (рост потребительских цен) и накопленная доходность - это другие примеры временных рядов уровня 1.
Мне известны два принципиальных способа профилактики описанных ошибок:
- вести реестр (хеш-таблица), в которой записывать всю мета-информацию, но всё равно придётся перегружать стандартные алгебраические операции, чтобы провалидировать данные
- расширить класс
np.array
через композицию или наследование. Расширение через наследование, согласно numpy API, особых преимуществ не даёт, а напротив может создать проблемы в случае измененияnumpy API
.
Обычные возможности np.array
будем расширять через композицию:
import pandas as pd
class TimeSeries:
def __init__(self, values, start_period: pd.Period, end_period: pd.Period, diff_level):
if not isinstance(values, np.ndarray):
raise ValueError('values should be numpy array')
if len(values) != (end_period - start_period).n + 1:
raise ValueError('values and period range has different lengths')
self.values = values
self.start_period = start_period
self.end_period = end_period
self.diff_level = diff_level
def __validate(self, time_series):
if self.start_period != time_series.start_period:
raise ValueError('start periods are incompatible')
if self.end_period != time_series.end_period:
raise ValueError('end periods are incompatible')
if self.diff_level != time_series.diff_level:
raise ValueError('diff levels are incompatible')
def apply(self, fun, *args):
'''
Обобщённый метод для применения произвольной функции `fun` с аргументами `args`
к текущему экземпляру `TimeSeries`
'''
# Сейчас TimeSeries поддерживает функции с 0 и 1 аргументом
# Пример функции без аргументов: np.array([2, 4]).cumprod() ~> np.array([2, 8])
if len(args) == 0:
ts = TimeSeries(values=fun(self.values),
start_period=self.start_period, end_period=self.end_period,
diff_level=self.diff_level)
return ts
# Сейчас TimeSeries в качестве второго аргумента поддерживает только TimeSeries или скаляр
else:
other = args[0]
if isinstance(other, TimeSeries):
self.__validate(other) # проверим, что TimeSeries совместимы
# для совместимых просто посчитаем функцию от значений
# мета-информация никак не меняется
ts = TimeSeries(values=fun(self.values, other.values),
start_period=self.start_period, end_period=self.end_period,
diff_level=self.diff_level)
return ts
# скаляры применяются к значениям безусловно, при этом мета-информация никак не меняется
elif isinstance(other, (int, float)):
ts = TimeSeries(fun(self.values, other),
start_period=self.start_period, end_period=self.end_period,
diff_level=self.diff_level)
return ts
else:
raise ValueError('argument has incompatible type')
# Все необходимые операции выражаются через apply
def __add__(self, other):
return self.apply(lambda x, y: x + y, other)
def __sub__(self, other):
return self.apply(lambda x, y: x - y, other)
def __truediv__(self, other):
return self.apply(lambda x, y: x / y, other)
def cumprod(self):
return self.apply(lambda x: x.cumprod())
def __repr__(self):
return 'TimeSeries(start_period={}, end_period={}, diff_level={}, values={}'.format(
self.start_period, self.end_period, self.diff_level, self.values
)
Проверяем TimeSeries¶
Имея такую надстройку, следующая попытка запуска расчетов с несовместимыми временными периодами ожидаемо привдёт к ошибке (start periods are incompatible
):
x = TimeSeries(values=np.array([4, 2]),
start_period=pd.Period('2015-1', freq='M'),
end_period=pd.Period('2015-2', freq='M'),
diff_level=1)
y = TimeSeries(values=np.array([1, 2]),
start_period=pd.Period('2015-2', freq='M'),
end_period=pd.Period('2015-3', freq='M'),
diff_level=1)
try:
print(x / y)
except ValueError as ve:
print(ve)
print('Деление `x / y` приводит к ошибке во время исполнения программы, т.к. начальные периоды несовместимы')
Теперь посчитаем реальную доходность для индекса S&P500 с помощью TimeSeries
:
snp_ror_ts = TimeSeries(
# Для простоты, выше не были введены операции `diff` и адресации массива, что позволило сделать так:
# snp_ts = TimeSeries(...)
# snp_ror_ts = snp_ts.diff() / snp_ts[:-1]
# Это можно проделать в качестве упражнения
values=np.diff(snp_values) / snp_values[:-1],
start_period=pd.Period('2015-2', freq='M'),
end_period=pd.Period('2016-3', freq='M'),
diff_level=1,
)
infl_usd_ts = TimeSeries(
values=infl_usd_values[1:],
start_period=pd.Period('2015-2', freq='M'),
end_period=pd.Period('2016-3', freq='M'),
diff_level=1,
)
snp_ror_real_ts = (snp_ror_ts + 1.) / (infl_usd_ts + 1.) - 1.
snp_ror_real_ts
# check:
np.testing.assert_equal(
snp_asset.get_return(real=True).values,
snp_ror_real_ts.values
)
Другие классы финансовых данных¶
Возможны случаи, когда временной ряд после применения операции преобразуется в одно число. Например, среднегодовая доходность (CAGR
) временного ряда является скалярной величиной. CAGR
не не вписывается в понятие временного ряда TimeSeries
. Поэтому лучше ввести дополнительный класс TimeValue
, с идентичной метаинформацией, а далее при необходимости расширить список типов аргументов для второго параметра в функции TimeSeries.apply
.
Реализация финансовых типов данных в библиотеке yapo¶
В статье мы рассмотрели на примере простой способ решения задачи о совместимости временных рядов. Как всё устроено в реальной библиотеке, можно изучить в исходном коде по ссылке. И каждый метод библиотеки, возвращающий временной ряд, использует TimeSeries
для работы с результатом:
snp_asset.close()
snp_asset.get_return()
snp_asset.get_return(real='True')
snp_asset.get_return(kind='cumulative')
snp_asset.cagr()
yp.inflation(currency='usd', kind='values', start_period='2015-1', end_period='2016-3')

Комментарии (3)