Bases de datos de series temporales

Datos que llegan con un reloj

Los ticks, cotizaciones, cifras de riesgo y métricas de servidor comparten forma: una marca de tiempo, algunas etiquetas, algunos valores medidos y un flujo interminable. Una base de datos de series temporales (InfluxDB, TimescaleDB, QuestDB, kdb+) se construye para esto. Las escrituras son casi siempre añadidos en el tiempo más reciente, así que son rápidas; los datos se guardan en bloques ordenados por tiempo, de modo que «la última hora» toca poco; el detalle viejo puede resumirse en datos más gruesos y por fin borrarse. El modelo de datos de InfluxDB es típico: una medición (como una tabla), tags (etiquetas indexadas como el símbolo o el mercado), campos (los valores medidos) y una marca de tiempo.

# 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

Submuestreo: de ticks a barras

Guardar cada tick para siempre es caro y rara vez necesario. El submuestreo agrupa los puntos en cubos de tiempo fijos y guarda un resumen de cada uno. Para precios el resumen estándar es OHLCV: apertura (primer precio del cubo), máximo, mínimo, cierre (último precio) y volumen. Un tick en el instante ts pertenece al cubo que empieza en ts - ts % bucket. Las bases lo hacen continuamente y escriben el resultado en una tabla aparte mucho más pequeña (un agregado continuo), que es lo que leen los gráficos y la mayoría de los backtests, mientras los ticks en bruto se conservan poco tiempo.

# 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}

Uniones as-of y retención

Las series rara vez marcan en los mismos instantes, así que unirlas por marcas de tiempo iguales casi no encuentra nada. La pregunta real es «¿cuál era la cotización cuando ocurrió esta operación?»: para cada operación, toma la última cotización en su instante o antes. Esta unión as-of (pandas.merge_asof, ASOF JOIN en kdb+ y DuckDB) debe mirar solo hacia atrás; usar una cotización posterior a la operación filtra el futuro al análisis. Como las cotizaciones están ordenadas por tiempo, una búsqueda binaria (bisect_right) encuentra la correcta en tiempo logarítmico. Por último, una política de retención descarta los puntos más viejos que un límite para acotar el almacenamiento: conserva 30 días de ticks, un año de barras de minuto y barras diarias para siempre.

# 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

• Un punto de serie temporal es una medición, tags indexados, campos medidos y una marca de tiempo; las escrituras son añadidos y los datos se guardan en orden temporal.
• Submuestrea los ticks en barras OHLCV (ts - ts % bucket) y lee las barras; conserva los ticks en bruto solo poco tiempo.
• Une las series con una unión as-of que solo mire hacia atrás (búsqueda binaria), aplica una retención y controla la cardinalidad de los tags y las unidades de tiempo.

A continuación: empaquetar una biblioteca de Python, convertir tu código en algo que otros puedan instalar.

Lab: qué se califica

• Agrupar ticks en barras con ohlcv_bars25 pts
• Hallar la cotización de cada operación con asof_join25 pts
• Escribir un punto con line_protocol25 pts
• Descartar datos caducados con apply_retention25 pts