Bases de données de séries temporelles

Des données qui arrivent avec une horloge

Les ticks, cotations, chiffres de risque et métriques de serveur ont la même forme : un horodatage, quelques étiquettes, quelques valeurs mesurées, et un flux sans fin. Une base de séries temporelles (InfluxDB, TimescaleDB, QuestDB, kdb+) est conçue pour cela. Les écritures sont presque toujours des ajouts au temps le plus récent, donc rapides ; les données sont stockées en blocs ordonnés dans le temps, si bien que « la dernière heure » touche peu de choses ; le vieux détail peut être résumé en données plus grossières puis supprimé. Le modèle de données d'InfluxDB est typique : une mesure (comme une table), des tags (étiquettes indexées comme le symbole ou la place), des champs (les valeurs mesurées) et un horodatage.

# 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

Sous-échantillonnage : des ticks aux barres

Garder chaque tick pour toujours coûte cher et sert rarement. Le sous-échantillonnage regroupe les points en tranches de temps fixes et garde un résumé de chacune. Pour les prix, le résumé standard est OHLCV : ouverture (premier prix de la tranche), plus haut, plus bas, clôture (dernier prix) et volume. Un tick à l'instant ts appartient à la tranche qui commence à ts - ts % bucket. Les bases le font en continu et écrivent le résultat dans une table séparée bien plus petite (un agrégat continu), que lisent les graphiques et la plupart des backtests, tandis que les ticks bruts ne sont gardés que peu de temps.

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

Jointures as-of et rétention

Les séries ne battent presque jamais aux mêmes instants : les joindre sur des horodatages égaux ne trouve presque rien. La vraie question est « quelle était la cotation quand cette transaction a eu lieu ? » : pour chaque transaction, prenez la dernière cotation à son heure ou avant. Cette jointure as-of (pandas.merge_asof, ASOF JOIN dans kdb+ et DuckDB) ne doit regarder qu'en arrière ; utiliser une cotation postérieure à la transaction fait fuir le futur dans l'analyse. Comme les cotations sont triées par temps, une recherche binaire (bisect_right) trouve la bonne en temps logarithmique. Enfin, une politique de rétention supprime les points plus vieux qu'une limite pour borner le stockage : gardez 30 jours de ticks, un an de barres à la minute et des barres journalières pour toujours.

# 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 point de série temporelle est une mesure, des tags indexés, des champs mesurés et un horodatage ; les écritures sont des ajouts et les données sont stockées dans l'ordre du temps.
• Sous-échantillonnez les ticks en barres OHLCV (ts - ts % bucket) et lisez les barres ; ne gardez les ticks bruts que peu de temps.
• Joignez les séries par une jointure as-of qui ne regarde qu'en arrière (recherche binaire), appliquez une rétention, et maîtrisez cardinalité des tags et unités d'horodatage.

À suivre : empaqueter une bibliothèque Python, transformer votre code en quelque chose que d'autres peuvent installer.

Lab : ce qui est noté

• Agréger les ticks en barres avec ohlcv_bars25 pts
• Trouver la cotation de chaque transaction avec asof_join25 pts
• Écrire un point avec line_protocol25 pts
• Supprimer les données expirées avec apply_retention25 pts