Bancos de dados de séries temporais

Dados que chegam com um relógio

Ticks, cotações, números de risco e métricas de servidor têm a mesma forma: um carimbo de tempo, alguns rótulos, alguns valores medidos e um fluxo sem fim. Um banco de dados de séries temporais (InfluxDB, TimescaleDB, QuestDB, kdb+) é feito para isso. As escritas são quase sempre acréscimos no tempo mais recente, então são rápidas; os dados ficam em blocos ordenados no tempo, de modo que «a última hora» toca pouco; o detalhe antigo pode ser resumido em dados mais grossos e por fim apagado. O modelo de dados do InfluxDB é típico: uma medição (como uma tabela), tags (rótulos indexados, como o símbolo ou o mercado), campos (os valores medidos) e um carimbo de tempo.

# One point in InfluxDB line protocol: measurement,tags fields timestamp
point = "trades,symbol=AAPL,venue=XNAS price=184.25,qty=100i 1700000000000000000"
head, fields, ts = point.split(" ")
measurement, *tag_pairs = head.split(",")
tags = dict(pair.split("=") for pair in tag_pairs)
print(measurement, tags)                             # trades {'symbol': 'AAPL', 'venue': 'XNAS'}
print(fields, int(ts))                               # price=184.25,qty=100i 1700000000000000000

Reamostragem: de ticks a barras

Guardar cada tick para sempre é caro e raramente necessário. A reamostragem agrupa os pontos em baldes de tempo fixos e guarda um resumo de cada um. Para preços o resumo padrão é o OHLCV: abertura (primeiro preço do balde), máxima, mínima, fechamento (último preço) e volume. Um tick no instante ts pertence ao balde que começa em ts - ts % bucket. Os bancos fazem isso continuamente e gravam o resultado numa tabela separada bem menor (um agregado contínuo), que é o que os gráficos e a maioria dos backtests leem, enquanto os ticks brutos são guardados por pouco tempo.

# Rolling ticks (ts, price, size) into 60-second OHLCV bars
def ohlcv_bars(ticks, bucket):
    bars = []
    for ts, price, size in ticks:
        start = ts - ts % bucket
        if bars and bars[-1]["start"] == start:
            bar = bars[-1]
            bar["high"], bar["low"] = max(bar["high"], price), min(bar["low"], price)
            bar["close"] = price
            bar["volume"] += size
        else:
            bars.append({"start": start, "open": price, "high": price, "low": price, "close": price, "volume": size})
    return bars

ticks = [(0, 100.0, 10), (30, 101.0, 5), (61, 99.5, 20), (65, 100.5, 1)]
for bar in ohlcv_bars(ticks, 60):
    print(bar)
# {'start': 0, 'open': 100.0, 'high': 101.0, 'low': 100.0, 'close': 101.0, 'volume': 15}
# {'start': 60, 'open': 99.5, 'high': 100.5, 'low': 99.5, 'close': 100.5, 'volume': 21}

Junções as-of e retenção

As séries raramente marcam nos mesmos instantes, então juntá-las por carimbos iguais quase não encontra nada. A pergunta real é «qual era a cotação quando este negócio ocorreu?»: para cada negócio, pegue a última cotação no seu instante ou antes. Essa junção as-of (pandas.merge_asof, ASOF JOIN no kdb+ e no DuckDB) deve olhar só para trás; usar uma cotação posterior ao negócio vaza o futuro para a análise. Como as cotações estão ordenadas por tempo, uma busca binária (bisect_right) acha a certa em tempo logarítmico. Por fim, uma política de retenção descarta pontos mais antigos que um limite para manter o armazenamento limitado: guarde 30 dias de ticks, um ano de barras de minuto e barras diárias para sempre.

# For each trade, the latest quote at or before it, found by binary search
from bisect import bisect_right

def asof_join(trades, quotes):
    times = [quote[0] for quote in quotes]
    joined = []
    for ts, price in trades:
        position = bisect_right(times, ts)
        joined.append((ts, price, *quotes[position - 1][1:]) if position else (ts, price, None, None))
    return joined

quotes = [(10, 99.9, 100.1), (40, 100.5, 100.7)]
print(asof_join([(5, 100.0), (10, 100.0), (50, 101.0)], quotes))
# [(5, 100.0, None, None), (10, 100.0, 99.9, 100.1), (50, 101.0, 100.5, 100.7)]

# Retention: keep only the points younger than the limit
def apply_retention(points, now, ttl):
    return [(ts, value) for ts, value in points if ts > now - ttl]

points = [(10, 1.0), (40, 2.0), (50, 3.0), (90, 4.0)]
print(apply_retention(points, 100, 60))              # [(50, 3.0), (90, 4.0)]   the point at 40 is exactly 60 s old: dropped
print(apply_retention(points, 100, 1000))            # everything is younger than the limit

• Um ponto de série temporal é uma medição, tags indexadas, campos medidos e um carimbo de tempo; as escritas são acréscimos e os dados ficam em ordem temporal.
• Reamostre os ticks em barras OHLCV (ts - ts % bucket) e leia as barras; guarde os ticks brutos só por pouco tempo.
• Una séries com uma junção as-of que só olha para trás (busca binária), aplique retenção e controle a cardinalidade das tags e as unidades de tempo.

A seguir: empacotar uma biblioteca Python, transformar seu código em algo que outros possam instalar.

Lab: o que é avaliado

• Agrupar ticks em barras com ohlcv_bars25 pts
• Achar a cotação de cada negócio com asof_join25 pts
• Escrever um ponto com line_protocol25 pts
• Descartar dados expirados com apply_retention25 pts