Zeitreihendatenbanken

Daten, die mit einer Uhr eintreffen

Ticks, Quotes, Risikozahlen und Servermetriken haben dieselbe Form: ein Zeitstempel, einige Labels, einige Messwerte und ein endloser Strom davon. Eine Zeitreihendatenbank (InfluxDB, TimescaleDB, QuestDB, kdb+) ist genau dafür gebaut. Schreibvorgänge sind fast immer Anhänge an der neuesten Zeit und daher schnell; Daten liegen in zeitlich geordneten Blöcken, sodass „die letzte Stunde“ wenig berührt; alter Detailgrad lässt sich zu gröberen Daten zusammenfassen und schließlich löschen. Das Datenmodell von InfluxDB ist typisch: eine Measurement (wie eine Tabelle), Tags (indizierte Labels wie Symbol oder Handelsplatz), Fields (die Messwerte) und ein Zeitstempel.

# 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

Downsampling: von Ticks zu Bars

Jeden Tick ewig aufzubewahren ist teuer und selten nötig. Downsampling gruppiert Punkte in feste Zeitfenster und behält von jedem eine Zusammenfassung. Für Preise ist die übliche Zusammenfassung OHLCV: Open (erster Preis im Fenster), High, Low, Close (letzter Preis) und Volumen. Ein Tick zur Zeit ts gehört zum Fenster, das bei ts - ts % bucket beginnt. Datenbanken tun das fortlaufend und schreiben das Ergebnis in eine eigene, viel kleinere Tabelle (ein kontinuierliches Aggregat), die Charts und die meisten Backtests lesen, während die Roh-Ticks nur kurz aufbewahrt werden.

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

As-of-Joins und Aufbewahrung

Zeitreihen ticken selten zu denselben Zeitpunkten, daher findet ein Join über gleiche Zeitstempel fast nichts. Die eigentliche Frage lautet „Wie lautete die Quote, als dieser Trade stattfand?“: Nehmen Sie für jeden Trade die letzte Quote zu seiner Zeit oder davor. Dieser As-of-Join (pandas.merge_asof, ASOF JOIN in kdb+ und DuckDB) darf nur zurückschauen; eine Quote nach dem Trade zu verwenden, lässt die Zukunft in die Analyse sickern. Da Quotes nach Zeit sortiert sind, findet eine binäre Suche (bisect_right) die richtige in logarithmischer Zeit. Schließlich verwirft eine Aufbewahrungsrichtlinie Punkte, die älter als ein Limit sind, damit der Speicher begrenzt bleibt: 30 Tage Ticks, ein Jahr Minutenbars und Tagesbars für immer.

# 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

• Ein Zeitreihenpunkt besteht aus Measurement, indizierten Tags, Messfeldern und Zeitstempel; Schreibvorgänge sind Anhänge, die Daten liegen in Zeitreihenfolge.
• Verdichten Sie Ticks zu OHLCV-Bars (ts - ts % bucket) und lesen Sie die Bars; Roh-Ticks nur kurz aufbewahren.
• Verbinden Sie Serien mit einem nur rückwärts schauenden As-of-Join (binäre Suche), wenden Sie Aufbewahrung an und halten Sie Tag-Kardinalität und Zeiteinheiten im Griff.

Als Nächstes: eine Python-Bibliothek paketieren, Ihren Code in etwas verwandeln, das andere installieren können.

Lab: Was bewertet wird

• Mit ohlcv_bars Ticks zu Bars verdichten25 Pkt.
• Mit asof_join die Quote zu jedem Trade finden25 Pkt.
• Mit line_protocol einen Punkt schreiben25 Pkt.
• Mit apply_retention abgelaufene Daten verwerfen25 Pkt.