Еще с третьей версии в Python появились аннотации типов, которые можно использовать в качестве комментариев к аргументам функций, для статического анализа и поиска ошибок или даже перегрузки методов в зависимости от типов аргументов. Помимо данных применений авторы Python оставили разработчикам возможность реализации своих сценариев. В этом туториале мы разработаем мини-фреймворк для автоматического построения цепочки вызовов, что позволит уменьшить объем интерфейсной части кода и упростить его масштабирование на дальнейших этапах.

Введение

Для начала напомним зачем нужны аннотации. Наиболее популярный вариант применения - это аннотация типов, которая позволяет обозначить типы передаваемых аргументов и возвращаемых значений, а за одно выявлять ошибки в коде:

# Все нормально (результат функции - строка)
def greeting(name: str) -> str:
    return 'Hello, ' + name

# Ошибка (некорректный тип результата - число, а ожидается строка)
def greeting(name: str) -> str:
    return 123

Менее известный вариант применения - это перегрузка операторов на основе передаваемых аргументов. В примере ниже вызывается либо первая, либо вторая функция в зависимости от типа аргумента name.

from multimethod import multimethod

@multimethod
def greeting(name: str) -> str:
    return 'Hello, ' + name

@multimethod
def greeting(name: list) -> str:
    return 'Hello, ' + ', '.join(name)

Кроме того, аннотации можно использовать для указания человеко-читаемых комментариев к аргументам без какой-либо синтаксической нагрузки для интерпретатора:

def greeting(name: 'Имя пользователя') -> str:
    return 'Hello, ' + name

Существует множество других вариантов, но мы сразу перейдем к нашему подходу.

Основной подход

В нашем туториале мы предлагаем использовать аннотации не просто как идентификатор типа аргумента (напр. число или строка), а в качестве идентификатора полезной нагрузки (напр. цена акции или название компании). Таким образом можно неявным образом связать между собой функции и классы, выполняющие логически связанную последовательность действий (например, выгрузка исторических котировок, преобразование, вычисление рисковых показателей и визуализация конечного результата), а с помощью фреймворка объединить эти действия в одну команду.

В дальнейшем с развитием кода реализация внутренних методов может изменяться и даже появляться новые, но фреймворк позволит автоматически перестраивать цепочки вызовов, опираясь на заданные нами аннотации. Таким образом, разработчик может сконцентрироваться на разработке логики прототипа и отложить решение некоторых архитектурных вопросов на более поздние итерации.

Образ результата

Прежде чем приступить к реализации давайте рассмотрим описанный выше подход на тривиальном примере вычисления средней цены акций за некоторый промежуток времени:

prices.py

import random

def get_prices(length=10) -> 'price_list':
    return [random.random() for _ in range(length)]

def calc_avg_price(p: 'price_list'):
    return sum(p) / len(p)

main.py (без фреймворка)

from prices import *
price_list = get_prices(10)
avg_price = calc_avg_price(price_list)

Обратите внимание, что идентификатор 'price_list' входного аргумента функции calc_avg_price() совпадает с идентификатором результата функции get_prices(), что позволяет объединить оба метода в один, например так:

main.py (с фреймворком)

from prices import *
from polimer import prices
avg_price = prices.calc_avg_price()

В данном примере отсутствует вызов функции get_prices(), т.к. он происходит автоматически, а результат подается на вход функции calc_avg_price(). Аналогичным образом можно упростить и более сложные цепочки вызовов, но для начала давайте реализуем описанный подход.

Реализация

Давайте перечислим что необходимо сделать для реализации описанной выше идеи на практике:

• Выгрузить список всех доступных методов и их аннотаций - список методов находится в служебном реестре sys.modules, а их аннотации из __annotations__

• Составить дерево/граф зависимостей - для этого будем использовать графовую структуру в виде списка смежностей, где узлами являются функции, а ребрами их зависимости по аргументам (например, когда результат работы одной функции является аргументом для другой, то между ними устанавливается ребро на графе)

• Провести топологическую сортировку узлов графа - чтобы вызов каждого последующего метода происходил строго после того, как были проинициализированы все требуемые аргументы

• Реализовать мета-функцию, которая будет строить и вызывать цепочки получившихся зависимостей исходя из того, какой конечный метод необходимо запустить (prices.calc_avg_price() в примере выше)

• Пробросить мета-функцию в составе публичного модуля (from polimer import prices в примере выше)

А теперь давайте детально рассмотрим реализацию каждого из описанных выше шагов. Начнем с выгрузки списка методов и аннотаций:

import sys, inspect

def load_functions():
    functions = {}
    for module_name in sys.modules:
        for item in vars(sys.modules[module_name]).values():
            if (inspect.isfunction(item) == True) and (len(item.__annotations__) > 0):
                f_id = item.__module__ + "." + item.__name__
                functions[f_id] = item
    return functions

Далее, составляем дерево зависимостей:

def build_dep_tree(functions):
    f_ids = {} # artifact_id -> function_id map
    dep_tree = {}
    for f_id in functions:
        artifact_id = functions[f_id].__annotations__.get("return", None)
        if artifact_id != None: f_ids[artifact_id] = f_id
    for f_id in functions:
        dep_tree[f_id] = []
        annotations = functions[f_id].__annotations__
        for argument in annotations:
            dep_artifact_id = annotations[argument]
            dep_f_id = f_ids.get(dep_artifact_id, None)
            if (argument != "return") and (dep_f_id != None):
                dep_tree[f_id].append(dep_f_id)
    return dep_tree

Для удобства понимания давайте представим дерево зависимостей на примере тривиального фрагмента кода:

import random

def get_range() -> 'num_days':
    return 30

def get_prices(l: 'num_days') -> 'price_list':
    return [random.random() for _ in range(l)]

def calc_avg(p: 'price_list', l: 'num_days') -> 'average_price':
    return sum(p) / l

Для такого фрагмента дерево зависимостей будет выглядеть следующим образом:

Дерево зависимостей

Допустим, мы хотим вызвать функцию calc_avg() - как сформировать цепочку вызовов, в которой все зависимые функции расположены после функций, от которых они зависят? Для этого достаточно произвести топологическую сортировку:

from collections import deque

def topology_sort(dep_tree, start_f_id):
    res_deque = deque()
    visited = set()
    stack = [[start_f_id]]
    while stack:
        for f_id in stack[-1]:
            if (f_id in visited) and (f_id not in res_deque):
                res_deque.appendleft(f_id)
            if f_id not in visited:
                visited.add(f_id)
                stack.append(dep_tree[f_id])
                break
        else:
            stack.pop()
    result = list(res_deque)
    result.reverse()
    return result

Дерево зависимостей (после топологической сортировки)

Как мы видим, первой следует вызывать функцию get_range(), после нее get_prices(), а затем calc_avg().

Далее нужно написать мета-функцию, которая будет строить цепочку вызовов исходя из заданной точки входа и последовательно вызывать эту цепочку, сохраняя промежуточные результаты на каждом шаге.

def run_chain(chain, functions):
    result = None
    artifacts = {}
    for f_id in chain:
        res = functions[f_id]()
        artifact_id = functions[f_id].__annotations__.get("return", None)
        if artifact_id != None: artifacts[artifact_id] = res
        result = res
    return result

def run(f_id):
    functions = load_functions()
    dep_tree = build_dep_tree(functions)
    chain = topology_sort(dep_tree, f_id)
    return run_chain(chain, functions)

Мета-функция run() производит описанные выше шаги - инициализирует список доступных методов, строит дерево зависимостей, осуществляет топологическую сортировку и вызывает сформированную цепочку. Для вызова цепочки реализован отдельный метод run_chain(), внутри него заводится словарь artifacts, хранящий промежуточные результаты.

Теперь нам необходимо пробросить интерфейс мета-функций наружу для удобства пользователя. Для этого заведем виртуальные подмодули (с помощью типа ModuleType) в реестре globals(), куда и пропишем мета-функции. Напомним, что виртуальные подмодули нужны, чтобы исключить возможные коллизии в названиях методов:

from types import ModuleType
from copy import deepcopy

def get_func(f_id):
    def func(**kwargs):
        return run(f_id, kwargs)
    return func

functions = load_functions()
__all__ = []

for f_id in functions:
    module_name, function_name = f_id.split(".", 1)
    if module_name not in globals():
        globals()[module_name] = ModuleType(module_name)
        __all__.append(module_name)
    setattr(globals()[module_name], function_name, deepcopy(get_func(f_id)))

__all__ = tuple(__all__)

Наконец, осталось объединить код нашего мини-фреймворка воедино, сформировать дистрибутив и загрузить его в репозитарий PyPI, чтобы его можно было устанавливать через pip install. Эти подробности мы не будем описывать, но вы можете посмотреть готовый результат в конце статьи. Теперь любой метод можно импортировать из нашего фреймворка следующим образом:

prices.py

def get_range():
   #...

def calc_avg():
   #...

main.py

from prices import *
from polimer import prices

prices.calc_avg()

Как видите, мы сократили цепочку вызовов с двух до одного метода (т.е. в 2 раза). Уже неплохо, но давайте посмотрим что будет, если применить фреймворк в более реалистичной задаче из области финтеха. Финтех был выбран исключительно для наглядности и в будущем мы можем с одинаковым успехом рассмотреть любую другую область, например телекоммуникации, искусственный интеллект или даже квантовые вычисления.

Прикладной пример (финтех)

Для наглядности мы рассмотрим одну из актуальных задач в области финтеха - определение рыночного режима (от англ. market regime detection) на основе наблюдений исторических биржевых котировок. Понимание текущего рыночного режима позволяет инвесторам условно разделять временные интервалы на периоды стабильности и кризисов, высокой и низкой волатильности, высокого и низкого уровня риска, а также принимать решения на основе такой классификации.

На иллюстрации ниже вы можете увидеть один из вариантов классификации состояний экономики за периоды с 1971 года по 2021-ый год. Примечательным является то, что периоды финансового кризиса 2008-го, а также ковидного 2019-го годов отмечены фиолетовым цветом. В нашем примере мы также будем рассчитывать, что кризисные периоды будут отмечены отдельным классом, что подтвердит корректность работы алгоритма в связке с фреймворком polimer.

С точки зрения машинного обучения эта задача относится к классу методов обучения без учителя (unsupervised learning) и может решаться с помощью таких подходов как скрытые гауссовские Марковские модели, модели гауссовых смесей или даже обычным k-means подходом. Чтобы не сильно отходить от нашей основной темы мы не будем вдаваться в детали каждого подхода, а возьмем готовую реализацию метода скрытой гауссовской Марковской модели из библиотеки hmmlearn.

А теперь давайте рассмотрим по поэтапно, как осуществить классификацию рыночных режимов на основе исторических биржевых котировок, а также как можно упростить эту задачу помощью фреймворка polimer.

• Во-первых, необходимо выгрузить сами биржевые котировки за исторический период. Мы возьмем их из открытых источников с помощью утилиты yfinance (yahoo finance). В качестве данных будем использовать котировки траста SPDR S&P 500 ETF (старое название Standard & Poor's Depositary Receipts), которые торгуются на бирже NYSE Arca под кодом SPY (тикер)

• Далее, нам понадобится предобработать данные сырых котировок, чтобы посчитать два производных показателя - доходность и диапазон цены за сутки

• Следующим шагом мы обучим модель на основе скрытого гауссовского марковского процесса, а также проведем классификацию на базе обученной модели

• В заключении, проведем визуализацию полученного результата в виде графика с цветовой индикацией определенных нами рыночных режимов

Мы не будем подробно описывать каждый этап, а приведем код всех описанных выше шагов сразу. Обратите внимание на аннотации методов (через двоеточие), с первого взгляда они не несут существенной нагрузки, но чуть позже вы поймете насколько сильно это упростит наше взаимодействие с кодом, благодаря polimer-у.

pip install polimer yfinance hmmlearn pandas numpy matplotlib

market_regimes.py

import numpy as np
import pandas as pd
import yfinance as yf
from hmmlearn import hmm

def load_data(index="SPY") -> "data":
    data = yf.download(index)
    return data

def prepare_dataset(data: "data") -> "dataset":
    returns = np.log(data.Close / data.Close.shift(1))
    range = (data.High - data.Low)
    features = pd.concat([returns, range], axis=1).dropna()
    features.columns = ["returns", "range"]
    return features

def train_model(dataset: "dataset") -> "model":
    model = hmm.GaussianHMM(
        n_components=3,
        covariance_type="full",
        n_iter=1000,
    )
    model.fit(dataset)
    return model

def predict_states(data: "data", dataset: "dataset", model: "model") -> "states":
    states = pd.Series(model.predict(dataset), index=data.index[1:])
    states.name = "state"
    return states

def plot_regimes(data: "data", states: "states"):
    color_map = {
        0.0: "green",
        1.0: "orange",
        2.0: "red"
    }
    pd.concat([data["Close"], states], axis=1).dropna().set_index("state", append=True)["SPY"].\
        unstack("state").plot(color=color_map, figsize=[16, 12])

А теперь давайте посмотрим как можно получить конечный результат - график с рыночными режимами. Для наглядности приведем 2 листинга - один с помощью фреймворка polimer и второй без него.

from market_regimes import *
data = load_data(index="SPY")
dataset = prepare_dataset(data)
model = train_model(dataset)
states = predict_states(data, dataset, model)
plot_regimes(data, states)

from market_regimes import *
from polimer import market_regimes

market_regimes.plot_regimes(index = "SPY")

Как видите, код с полимером стал более лаконичным, а все основные шаги (выгрузка и предобработка данных, обучение модели и классификация) выполняются автоматически. Аналогичный эффект можно достичь не только при классификации рыночных режимов в финтехе, но и для любой другой задачи, в которой вычисления осуществляются в несколько этапов по типу конвейера.

Давайте запустим получившийся код и рассмотрим внимательно полученный результат. На графике ниже представлены исторические котировки индекса SPY с выделенными разными цветами рыночными состояниями, которые были классифицированы нашим алгоритмом. Примечательно, что кризисные периоды 2008 и 2019-х годов отмечены отдельным цветом, что в целом соответствует ожиданиям и подтверждает корректность работы алгоритма в составе фреймворка.

На этом наш туториал подошел к концу, а в заключение хотелось бы отметить, что успех прикладной ИТ-разработки все чаще зависит не только от качества решения целевой задачи, но и от своевременного развития структурного инструментария, упрощающего масштабирование стартового прототипа в промышленный продукт. А если вы занимаетесь экспериментальной разработкой и хотите подготовиться к успешному выводу своего ИТ-продукта на рынок в будущем, не стесняйтесь обращаться к нам в комментариях или личных сообщениях - мы с удовольствием рассмотрим ваш кейс и поможем спроектировать оптимальный ИТ-инструментарий.

Всех с наступающими праздниками и ярких открытый в Новом году!

Ссылки на материалы из статьи:

• Исходный код фреймворка

• Определение рыночных режимов

• Упаковка PyPI репозитариев