iondv / aib Goto Github PK

Реализовать приложение ION предосталяющее GUI управления агентами шины. Приложение реализовать на основе модуля registry
Функционал:

1. Хранение списка агентов в виде обьектов с человекочитаемыми именами и описаниями, идентификатором и путем до модуля предоставляющего функцию-конструктор, либо именем метода компонента в di возвращающего функцию-конструктор.
2. Добавление агента в список
3. Удаление агента из списка
4. Редактирование агента
5. Регистрация и выключение агента на шине
6. Запуск/Остановка шины
7. Отображение статуса агента

В рамках данной задачи нужно реализовать:

1. Компонент шины AIB, представляющий собой обьект с состоянием предоставляющий следующий интерфейс программного взаимодействия:

class Bus {
  handle(Signal signal)
  wait(code, sender, thread): Promise
  register(Agent agent)
  eject(id)
  status(id)
  setStatus(id, data)
  serve()
  stop()
}

Описание функционала методов:

handle(Signal signal) - обработка входящего сигнала. Объект сигнала помещается в очередь для дальнейшей обработки. Очередь сообщений представлена внутренним компонентом реализующим интерфейс взаимодействия Queue:

class Queue {
  enqueue(Signal signal)
  dequeue(): Signal
}

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

Спецификация объекта класса Signal:
class Signal {
code: string
sender: string
thread: string
data: mixed
}

Где code - уникальный идентификатор типа сигнала, sender - уникальный идентификатор отправителя сигнала (может отсутствовать), thread - уникальный идентификатор потока обработки сигнала, который присваивается шиной в методе handle если не указан явно, data - данные произвольного формата, определяемого типом сигнала.
Таким образом метод handle просто помещает объект типа Signal в очередь сигналов, присваивая ему идентификатор потока обработки.

wait(code, sender, thread, timeout): Promise - ожидание сигнала. Создает и регистрирует во внутреннем хранилище ожидание события идентифицируемого переданными параметрами. Если метод вызван без параметров выполняется ожидание ЛЮБОГО следующего события. Также последним аргументом можно передать время ожидания сигнала (значение по умолчанию должно настраиваться на уровне всей шины), по истечении которого, если сигнал не был получен должно выбрасываться соответствующее исключение.

register(Agent agent, Signal trigger) - регистрирует на шине агента, который будет запускаться при поступлении сигнала описываемого аргументом trigger. Спецификация объекта Agent:

class Agent {
  id: string
  constructor: function
}

Где id - уникальный идентификатор агента, constructor - функция выполняющая инициализацию агента на шине. Эта функция будет вызвана при регистрации агента с передачей объекта шины в качестве единственного аргумента. Если результатом функции-конструктора является функция, то в случае если
передан параметр trigger, она становится обработчиком этого события на шине, если trigger не передан, то функция будет вызвана ОДИН раз при поступлении ЛЮБОГО следующего события (посредством wait).

eject(id) - отключает агента от шины, очищает его состояние, исключает его обработчик из цикла обработки сигнала.

setStatus(id, data) - устанавливает во внутреннем хранилище шины состояние агента идентифицируемого аргументом id. Состояние описывается аргументом data произвольного формата. Внутреннее хранилище шины - реализация интерфейса KeyValueRepository ядра ION, объект задаваемый через внедрение зависимостей.

status(id) - возвращает статус агента идентифицируемого аргументом id, если агент не зарегистрирован выбрасывает исключение.

serve() - запускает шину. метод выбирает сигнал из очереди сообщений и запускает его обработчики, после чего посредством process.setImmediate планирует свой следующий рекурсивный запуск.

stop() - останавливает шину. Устанавливает внутренний флаг прерывающий рекурсию метода serve()

Пример конструктора агента шины сообщений:

const AGENT_ID = 'cool-agent';
const extApi = {method: (data) => {return 'ololo'}}

(bus) => async ({code, sender, thread, data}) => {
  switch (signal.code) {
    case 'sig1':
      const sig2 = await bus.wait('sig2')
      bus.handle({code: 'sig3', sender: AGENT_ID, thread, data: data.a + sig2.data.b})
    break;
    case 'sig3':
      if (sender !== AGENT_ID) {
        await extApi.method(data)
      }
    break;
  }
}        

Шина получает входные сигналы, например из таска MODAIB-2: Метаданные агента и сигнала или из MODAIB-4: Каркас генератора сигналов из истории

Что нужно. На этот сигнал, назовем его OnQuote - если мы получили - мы проверяем, нужно сохранять или нет из свойств сигнала и после этого вызываем все подписанные на сигнал агентов.

Агента мы регистрируем в шине по

MODAIB-2: Метаданные агента и сигнала - т.е. в регистри он создан. Возможно как-то инициализирован, если инициализация для агента важна (пакеты зависимостей устанавливать наверное не надо, но параметры сигналов входны и выходных можно регистрировать).

После запуска агента с тиком(значением) сигнала мы в соответствии с параметрами агента - ожидаем или не ожидаем ответа от него (сигнала)

Если есть ответ от агента - мы его раскидываем во все агенты которые на него подписаны. Если тип сигнала тербует сохранения в классе - сохраняем.

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

Второе формируем ли мы очередь сигалов - т.е. гарантированная доставка сигналов ожидания перед каждым из агентов, пока они их не обработают? Возможно нужно для агента указывать параметр для сигнала:

• формирование очереди пока не обработает (размер очереди надо где-то задавать)
• передаче всей очереди пакетом в агента,
• пропуск сигналов пока нет ответа от агента о готовности принять сигнал,
• подача сигнала без учета ответа (подал и забыл)

Размер максимальной очереди - надо выности в параметры модуля в deploy

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

• результатам инициализации агента
• по результатам обработки каждого сигнала
• по счетчику сохраненных сигналов
• по параметру агента о необходимости сохранить значение
• периодически по времени (нужен интервал времени)
• по выходу(остановке) системы или выгрузке агента

Ещё нужен режим обучения/тестирования/демо - параметр для агентов и шины в которой она не сохраняет полученные сигналы в классах и передаем агентам, что это тестовый режим - это для таска

Что делаем - у нас есть данные, например в таске

MODAIB-3: Каркас шиныTo Do - но эти данные - это сигналы. Эти сигналы входные будут храниться в классе уже.

По идее нам нужны сценарии обучения/тестирования - возможно это тоже какой-то объект класса сценариев, которые мы создаем. А потом множество раз прогоняем.

Кейс - система запускает обучение на выбранных данных из класса. Выбор:

выибраем сигнал(ы) по которым проводим обучение как входные сигналы

по интервалу времени для отбора сигналов, если он есть

по каким-то параметрам сигналов

Сигналы подаются на вход шины. Далее шина обрабатывает. Вопрос в уплотнении времени - т.к. система может быть нелинейной из-за того что часть агентов будут требовать времени на обработку (там при превышении очереди сломается) - по идее нам нужн параметр скорости - мультипликатор - в 2/4/100 и максимально возможной и задавать его в сценарии обучения/тестирования

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

 куда мы подаем ответ шины от подачи сигнала

или вызываем после подачи сигнала 

или вызываем его раз в какой-то период с учетом нашего мультипликатора времени

После окончания обучения - мы должны забрать параметры у укаждого из подключенных агентов и где-то их сохранть. Скорее всего в сценарии для каждого из агентов.

Шина получает входные сигналы, например из таска ёMODAIB-2: Метаданные агента и сигналаё или из ёMODAIB-4: Каркас генератора сигналов из историиё

Что нужно. На этот сигнал, назовем его OnQuote - если мы получили - мы проверяем, нужно сохранять или нет из свойств сигнала и после этого вызываем все подписанные на сигнал агентов.

Агента мы регистрируем в шине по

MODAIB-2: Метаданные агента и сигнала - т.е. в регистри он создан. Возможно как-то инициализирован, если инициализация для агента важна (пакеты зависимостей устанавливать наверное не надо, но параметры сигналов входны и выходных можно регистрировать).

После запуска агента с тиком(значением) сигнала мы в соответствии с параметрами агента - ожидаем или не ожидаем ответа от него (сигнала)

Если есть ответ от агента - мы его раскидываем во все агенты которые на него подписаны. Если тип сигнала тербует сохранения в классе - сохраняем.

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

Второе формируем ли мы очередь сигалов - т.е. гарантированная доставка сигналов ожидания перед каждым из агентов, пока они их не обработают? Возможно нужно для агента указывать параметр для сигнала:

• формирование очереди пока не обработает (размер очереди надо где-то задавать)
• передаче всей очереди пакетом в агента,
• пропуск сигналов пока нет ответа от агента о готовности принять сигнал,
• подача сигнала без учета ответа (подал и забыл)

Размер максимальной очереди - надо выности в параметры модуля в deploy

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

• результатам инициализации агента
• по результатам обработки каждого сигнала
• по счетчику сохраненных сигналов
• по параметру агента о необходимости сохранить значение
• периодически по времени (нужен интервал времени)
• по выходу(остановке) системы или выгрузке агента

Ещё нужен режим обучения/тестирования/демо - параметр для агентов и шины в которой она не сохраняет полученные сигналы в классах и передаем агентам, что это тестовый режим - это для таска

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

Назначение - агента можно создавать в регистри, или вызовом специальной утилиты (но в регистри сразу будет работать модель безопасности, а это важно)

Структура данных класса агента:

• [id] id агента (автоматический)
• [code] код (агента)
• [version] версия агента
• [name] наименование агента
• [description] описание агента
• вид агента:
  • import - импортирует сигналы из внешней системы
  • export - экспортируе сигналы во внешнюю систему
  • gibrid - гибридный он и выдает сигналы и получает их для внешней системы
  • internal - внутренний.
• [type] тип агента (пока по идее достаточно только подключаемого js) - тип агента по идее какой-то расширяемы справочник регистрируемых типов - которые можно подключать отдельно (возможно нужна реинициализация модуля)
  • js
  • python (пока можно не делать, по идее можно в js запихать, но может можно как-то вызывать)
  • tensorflow (по идее можно веншюю модуль запуска, иначе надо библиотеку ставить,
  • веб-сервис
• [path] путь к исполняемому файлу с кодом агента (пока основное) - или скорее папке с агентом. Т.к. по идее агенты это отдельные проекты, которые могут содержать зависимости библиотек, устанавливаемых как npm install . Вероятней всего это имя приложения агнета??? из таска MODAIB-5
• [run] запустить/остановить (логическое поле или БП на логическое поле потом повесим)
• [runResult] результат запуска (обновляется при каждом запуске)
• [dateCreate] дата создания
• [dateRun] дата последнего запуска (не передачи сигналов к нему, а именно инициализации запуска когда обнаружено значение запустить)
• [runResult] текст последнего результатов запуска (логи с ошибкой например при инициализации)
• [signalParamWaitFinish]требуется ли дожидаться окончания выполнения агента (безсигнальные агенты по идее могут ничего не выдавать) -но в любом случае он должен сообщить, что закончил. Иначе его надо вырубать по таймауту
• [signalParamWaitResult]требуется ли дожидаться появления сгиналов для выполенения для передачи следующего
• [signalParamQueue] формировать ли очередь входных сигналов, если требуется ожидания выполнения следующего экземпляра агента.
• [input] сигналы входные
  • нужно ли в описании агента или мы их автоматически регистрируем при инициализации- из примера это поле "callback":"OnQuote".
  • нужно ли кроме сигнала ещё какие-то значения парамеров указывать для сигнала - например ниже из пример поле "secCode":"SBER" - т.е. только на акцию сбербанка агент заточен. Можем ли мы регистрировать одного и того же агента на одни и те же сигналы, но разные параметры. Или пока параметры лучше отложить.
• [output] сигналы выходные
• [memoryType] агент с памятью/без памяти и параметры сохранения
• [none] нет сохраняемой памяти
• [stop] сохранять память при остановке системы
• сохранять или нет периодически память и значение периода в секнудах(микросекундах) для сохранения (не сохранять, по 1 раз в секнуду (т.е. когда больше секунды прошло)
• [count] сохранять или нет по счетчику передачи сиганлов (в т.ч. по каждому сигналу - т.е. 0 не сохранятт, 1 по каждому, по 2 через один и т.д.)
• [memoryTypeValue] значение для типа памяти - время или счетчик
• [memoryValue] поле хранения памяти (json объект), если сохраняем память при остановке системы или по периоду => ПОСТАВИЛ ТИП Custom type - Но по идее нужен JSON ⚠️
• [runInit] нужна инициализация или нет при старте - не понятно нужно или нет, по идее контрактом агента должна быть эта функция, но он просто её может сразу возвращать.
• нужен ли деплой или нет (например если агент содержит свои классы для хранения обработанных сигналов - их храним в нейспейсе агента - т.е. агент это простейшее специализированное приложение/метаданные иона?) - не стал ставить атрибут - по идее если агент это приложение - мы его инициализировали.

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

Ещё вопрос по сигналам - по идее часть сигналов - это объекты класса. Т.е. при получении мы их должны сохранить в системе. А часть это просто информация для других агентов и сохранения не требует.

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

И тогда эти классы можно поключать к агенту - как входные и выходные сигналы. Сигналы с данными - будут иметь структуру класса - это удобно.

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

Пример тика (тик значение, объект) по сигналу по трейдингу (в трейдинге тик - это факт изменения сигнала - он может быть периодическим - т.е. тупо раз в секнуду или минуту мы подаем, или апериодическим - т.е. по факту изменения) - по идее это тоже свойство сигнала. Для периодического ещё период надо указывать, если в период не поступило - передавать ошибку отсуствия сигнала агентам.

{"datetime":1591880033,
"secCode":"SBER",
"value":
{"bid_count":"10.000000",
"offer_count":"10.000000",
"bid":[
{"price":"206.33","quantity":"633"},
{"price":"206.34","quantity":"2982"},
{"price":"206.35","quantity":"809"},
{"price":"206.36","quantity":"134"},
{"price":"206.37","quantity":"259"},
{"price":"206.38","quantity":"110"},
{"price":"206.39","quantity":"184"},
{"price":"206.41","quantity":"504"},
{"price":"206.42","quantity":"56"},
{"price":"206.43","quantity":"740"}],
"offer":[{"price":"206.50","quantity":"248"},
{"price":"206.51","quantity":"7"},
{"price":"206.52","quantity":"102"},
{"price":"206.53","quantity":"214"},
{"price":"206.54","quantity":"1944"},
{"price":"206.55","quantity":"283"},
{"price":"206.56","quantity":"150"},
{"price":"206.57","quantity":"343"},
{"price":"206.58","quantity":"733"},
{"price":"206.59","quantity":"706"}]},
"classCode":"TQBR",
"callback":"OnQuote",
"stockMarket":"MOEX"
}

"callback":"OnQuote" - это по сути имя сигнала - в данном случае это таблица данных по инструменту secCode - можно стандартизировать имя сигналов в json, можно делать отдельные сервисы для приема разных сигналов и для трансформации их в нужный на вход в шину. Т.е. нужен ещё какая то структура сигнала. Поля bid и offer по идее могут быть либо коллекциями, либо полем с типом object - мы такой обсуждали, но не ввели в ion. Возможно (и скорее всего) такое поле нужно, когда нет смысла хранить связанные объекты отдельно от основного

Возможная стурктура класса регистрации сигнала:

• [code] код сигнала
• [ver] версия структуры сигнала(?)
• коллекция имен параметров сигнала и их типы(?) насколько нужно?
• [source] источник сигнала
  • внешний,
  • другой агент
• [type] тип сигнала
  • периодический - поле параметра периода в секундах
  • апериодический
• [save] сигнал отражается в данных или нет - логический
• [saveClassName] если сигнал отражается - коллекция куда сохраняются данные
• надо ли проставлять метку времени получения сигнала - по идее в самом сигнале может быть служебное поле __busRegistrationDate или что-то такое и его можно и так создавать.

Где сигналы описаны? это какой-то класс? он в агенте лежит или отдельно? по идее версия сигналов может быть по модели semver - и тогда у агентов, можно указывать допустимые версии сигналов по модели как версии пакетов в npm - мажорная совместимость и т.п.

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

Но для начала условно говоря, это то приложение, в котором указан модуль AIB
MODAIB-3: Каркас шины и агенты

MODAIB-4: Каркас генератора сигналов из истории

Возможно приложение содержит в себе объекты класса agents и signals из

MODAIB-2: Метаданные агента и сигнала для импорта - т.е. данные.

Пример делаем на трейдинговой системе. Входные сигналы будут идти из файла в
MODAIB-3: Каркас шины
Какого-то прсотейшего агента по трейдинговым линиям мы напишем в

MODAIB-4: Каркас генератора сигналов из истории

Репо github https://github.com/iondv/trading

Нужна реализация простейшего агента. Агент далеко не всегда приложение iondv - это лишь один из типов агентов.

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

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

Это похоже на шину с гарантированной доставкой данных.

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

Они могут ставиться как обычные зависимости в package.json какого-то приложения AIB

Т.е. конкретное приложение с AIB - это о приложение IONDV в котором есть модуль AIB и агенты. И возможно какая-то отдельная логика по визуализации сигналов, результатов работы агентов - т.к. в приложении можно получить доступ например к выходным сигналом агентов или просто их сохраняемым данным.