Vela → smart cache per time series e oltre

Nel fintech, ci troviamo spesso a dover gestire volumi piuttosto elevati di dati sui tassi di cambio delle valute. Riceviamo dati da diverse fonti, ognuna delle quali ha la propria interpretazione su come estrapolare i valori dei tassi per domani, dopodomani, il mese prossimo e persino nei prossimi tre anni. Se qualcuno fosse in grado di prevedere i tassi formulare correttamente, sarebbe tempo di chiudere l'attività e semplicemente scambiare denaro sconsideratamente. Alcune fonti sono più affidabili di altre, mentre alcune forniscono solo spazzatura, con rare occorrenze di valori quasi corretti, ma comunque per coppie esotiche. Il nostro compito è passare al setaccio queste decine di migliaia di valori al secondo e determinare cosa mostrare ai clienti. Dobbiamo filtrare il singolo valore corretto da tonnellate di sporcizia e fango, proprio come fanno i fenicotteri durante il loro pasto.

Una caratteristica distintiva dei fenicotteri è il loro grande becco ricurvo verso il basso, con cui filtrano il cibo dall'acqua o dal fango.
 — Wiki

Così è nata la libreria Vela, che memorizza la cache di stato per più valori in intervalli di tempo specificati. Sotto il cofano, filtra al volo dati errati e obsoleti, oltre a fornire accesso agli ultimi N valori convalidati per ogni chiave (coppie di valute, nel nostro caso).

Supponiamo di raccogliere i tassi per tre coppie di valute. La definizione più semplice Vela per memorizzare lo stato attuale apparirà più o meno così:

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

Aggiornamento dei valori

Vela.put/3 la funzione eseguirà sequenzialmente quanto segue:

• chiamerà il validatore sul valore, se definito (vedi il capitolo Validazione qui sotto);
• aggiungerà il valore a una serie di valori buoni, se la validazione è andata a buon fine, o a una serie di servizio :__errors__ in caso contrario;
• chiamerà il sorting se sorter è definito per questa chiave, o semplicemente aggiungerà il valore all'inizio dell'elenco (LIFO, vedi il capitolo Ordinamento qui sotto);
• taglierà la serie in base al parametro :limit passato alla creazione;
• restituirà la struttura aggiornata 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]}

Inoltre Vela implementa Access, quindi è possibile aggiornare i valori utilizzando una delle normali funzioni per l'aggiornamento profondo delle strutture del nostro arsenale Kernel: Kernel.get_in/2, Kernel.put_in/3, Kernel.update_in/3, Kernel.pop_in/2, e Kernel.get_and_update_in/3.

Validazione

Il validatore può essere definito come:

• una funzione esterna con un solo argomento (&MyMod.my_fun/1), riceverà solo il valore da validare;
• una funzione esterna con due argomenti, &MyMod.my_fun/2, riceverà la coppia serie, valore da validare;
• modulo che implementa Vela.Validator;
• parametro di configurazione threshold, e - facoltativamente - compare_by, vedi il capitolo Confronto qui sotto.

Se la validazione ha esito positivo, il valore viene aggiunto all'elenco sotto la chiave corrispondente, altrimenti la tupla {serie, valore} viene inviata a :__errors_.

Confronto

Valori memorizzati in queste righe possono essere qualsiasi cosa. Per insegnare Vela loro a confrontarsi, è necessario fornire compare_by un parametro nella definizione della riga (a meno che i valori non possano essere confrontati standardmente Kernel.</2); questo parametro deve avere tipo (Vela.value() -> numero()). Di default è semplicemente & &1.

Inoltre, nel definire una serie è possibile passare un parametro comparator per calcolare i valori delle differenze (min/max); ad esempio, passando Date.diff/2 come comparatore, si possono ottenere differenze corrette per le date.

Un altro modo utile di operare è passare un parametro threshold, che definisce il rapporto massimo consentito del nuovo valore rispetto a {min, max} all'intervallo. Poiché è specificato in percentuale, il controllo non utilizza comparator, ma utilizza ancora compare_by. Ad esempio, per specificare un valore soglia per il tempo delle date, è necessario indicare compare_by: &DateTime.to_unix/1 (per ottenere un valore intero) e threshold: 1, il che significa che i nuovi valori saranno consentiti solo se rientrano nel ±band intervallo dai valori attuali.

Infine, si può utilizzare Vela.equal?/2 per confrontare due cache. Se i valori definiscono la funzione equal?/2 o compare/2, allora queste funzioni saranno utilizzate per il confronto, altrimenti si utilizza semplicemente ==/2.

Ottenere valori

Il trattamento dello stato attuale inizia solitamente con la chiamata Vela.purge/1, che rimuove i valori obsoleti (se il validatore è legato a timestamps). Poi si può chiamare Vela.slice/1, che restituirà keyword con i nomi delle righe come chiavi e i valori attuali come primi.

Puoi anche utilizzare get_in/2/pop_in/2 per un accesso a basso livello ai valori in ogni riga.

Applicazione

Vela può rivelarsi estremamente utile come cache di serie temporali nello stato di un processo di tipo GenServer/Agente. Vogliamo mai utilizzare valori obsoleti delle tariffe, e per questo teniamo semplicemente un processo con uno stato, elaborato Vela, con il validatore mostrato di seguito.

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

e Vela.purge/1 che rimuove tranquillamente tutti i valori obsoleti ogni volta che abbiamo bisogno di dati. Per accedere ai valori attuali basta chiamare Vela.slice/1, e quando è necessaria una piccola storia del tasso (l'intera serie), semplicemente restituiamo essa - già ordinata - con i valori validati.

• GitHub
• Hexdocs

Buona cache delle serie temporali!

Fonte: habr.com