🥇Vela → интелигентен кеш за времеви серии и не само

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

Особен отличителен белег на фламингото е масивният му извит надолу клюн, с който филтрира храната от вода или тиня.
— Wiki

Така се роди библиотеката Vela, която съхранява кеш за състоянието на няколко стойности в зададени времеви интервали. Под капака тя в реално време отсява лоши и остарели данни, а също така предоставя достъп до последните N валидирани стойности за всеки ключ (валутни двойки, в нашия случай).

Да предположим, че събираме курсове за три валутни двойки. Най-простото дефиниране Vela на актуалното състояние ще изглежда по следния начин:

defmodule Pairs do
  use Vela,
    eurusd: [sorter: &Kernel. 0
end

Актуализиране на стойностите

Vela.put/3 функцията последователно ще направи следното:

ще извика валидатор на стойността, ако такъв е определен (вж. глава Валидация по-долу);
ще добави стойността или в реда на добри стойности, ако валидирането завърши успешно, или в служебния ред :__errors__ в противен случай;
ще извика сортиране, ако sorter е определен за този ключ, или просто ще постави стойността в началото на списъка (LIFO, виж. глава Сортиране по-долу);
ще отсече реда в съответствие с параметъра :limit предаден при създаването;
ще върне актуализираната структура Vela.

iex|1 > pairs = %Pairs{}
iex|2 > Vela.put(pairs, :eurcad, 1.0)
#⇒ %Pairs{..., eurcad: [1.0], ...}
iex|3 > Vela.put(pairs, :eurcad, -1.0)
#⇒ %Pairs{__errors__: [eurcad: -1.0], ...}
iex|4 > pairs |> Vela.put(:eurusd, 2.0) |> Vela.put(:eurusd, 1.0)
#⇒ %Pairs{... eurusd: [1.0, 2.0]}

Също Vela имплементира Достъп, така че можете да използвате всяка от стандартните функции за дълбоко обновление на структури от арсенала Kernel: Kernel.get_in/2, Kernel.put_in/3, Kernel.update_in/3, Kernel.pop_in/2, и Kernel.get_and_update_in/3.

Валидация

Валидаторът може да бъде дефиниран като:

външна функция с един аргумент (&MyMod.my_fun/1), тя ще получи само стойността за валидация;
външна функция с два аргумента, &MyMod.my_fun/2, тя ще получи двойка serie, value за валидация;
модул, имплементиращ Vela.Validator;
конфигурационен параметър праг, и – опционално – compare_by, виж. глава Сравнение по-долу.

Ако валидирането е успешно, стойността се добавя в списъка под съответния ключ, в противен случай кортеж {serie, value} се изпраща в :__errors_.

Сравнение

Стойностите, запазени в тези редове, могат да бъдат всякакви. За да обучите Vela ги да сравняват, е необходимо да предадете compare_by параметър в определението на реда (освен ако стойностите не могат да бъдат сравнени стандартно Kernel.</2); този параметър трябва да има тип (Vela.value() -> number()). По подразбиране това просто е & &1.

Също така, в определението на реда може да се предаде параметър comparator за изчисляване на стойностите на делтата (min/max); например, предавайки Date.diff/2 като компаратор, можете да получите правилни делти за дати.

Друг удобен начин за работа е предаване на параметър праг, който определя максимално допустимото отношение на новата стойност към {min, max} интервала. Тъй като е зададен в проценти, проверката не използва comparator, но все пак използва compare_by. Например, за да зададете прагова стойност за времена на дати, е необходимо да зададете compare_by: &DateTime.to_unix/1 (за получаване на цяло число) и праг: 1, в резултат на което новите стойности ще бъдат разрешени, само ако са в ±band интервала от текущите стойности.

Накрая, можете да използвате Vela.equal?/2 compare/2 , ако стойностите определят функцията или equal?/2, тогава тези функции ще бъдат използвани за сравнение, в противен случай просто ще използваме ==/2.

Получаване на стойности

Обработката на текущото състояние обикновено започва с извикването Vela.purge/1, което премахва остарелите стойности (ако валидатор е обвързано с timestampи). След това може да се извика Vela.slice/1, който ще върне ключова дума с имената на редовете като ключове и първите актуални стойности.

Може също да използвате get_in/2/pop_in/2 за нискоуреден достъп до стойностите в всеки ред.

Приложение

Vela може да се окаже изключително полезен като кеш на времеви серии в състоянието на процеса от тип GenServer/Agent. Искаме никога да не използваме остарели стойности на курсовете, и за това просто поддържаме процес със състояние, което се обработва Vela, с валидатор, показан по-долу.

@impl Vela.Validator
def valid?(_key, %Rate{} = rate),
  do: Rate.age(rate) < @death_age

и Vela.purge/1 спокойно премахва всички остарели стойности всеки път, когато ни трябват данни. За достъп до актуалните стойности просто извикваме Vela.slice/1, а когато ни е нужна малка история за курса (цялата серия), просто го връщаме — вече сортирано — с валидирани стойности.

Успех с кеширането на времевите серии!

Източник: habr.com