03 · Construire le footprint¶

Le code de ce chapitre est testé et exécutable. Il ne dépend d'aucune bibliothèque externe — uniquement la bibliothèque standard Python. Le fichier complet avec sa suite de tests figure en annexe.

La classification de l'agresseur¶

def aggressor(price, bid, ask):
    """1 = acheteur agressif (trade >= ask), -1 = vendeur agressif (trade <= bid),
    0 = indéterminé (trade dans le spread)."""
    if ask is not None and price >= ask:
        return 1
    if bid is not None and price <= bid:
        return -1
    return 0

Le bid et l'ask passés doivent être ceux antérieurs à la transaction, comme établi au chapitre Le carnet d'ordres.

Le cas 0 — transaction dans le spread — se produit notamment sur les transactions de bloc et les combinaisons. L'ignorer plutôt que de l'attribuer arbitrairement à un côté évite d'introduire un biais systématique dans le delta.

La structure Footprint¶

from collections import defaultdict

class Footprint:
    """Agrège les trades d'une barre en volume bid / volume ask par prix."""

    def __init__(self, tick_size):
        self.tick = tick_size
        self.levels = defaultdict(lambda: [0, 0])  # prix -> [vol_bid, vol_ask]

    def add(self, price, size, side):
        key = round(price / self.tick)
        if side > 0:
            self.levels[key][1] += size
        elif side < 0:
            self.levels[key][0] += size

    def price(self, key):
        return key * self.tick

    @property
    def delta(self):
        return sum(a - b for b, a in self.levels.values())

    @property
    def volume(self):
        return sum(a + b for b, a in self.levels.values())

    def poc(self):
        """Point of control : prix du plus gros volume total de la barre."""
        if not self.levels:
            return None
        key = max(self.levels, key=lambda k: sum(self.levels[k]))
        return self.price(key)

Pourquoi indexer par entier plutôt que par prix flottant

round(price / tick) transforme 5000,25 en un entier. Les prix en virgule flottante ne se comparent pas de façon fiable : 5000.25 - 5000.00 ne vaut pas exactement 0.25 en binaire, et deux transactions au même prix peuvent créer deux clés distinctes dans le dictionnaire.

Une ligne, et toute une catégorie de bugs silencieux disparaît. La conversion inverse (key * tick) ne se fait qu'à l'affichage.

La détection des imbalances¶

    def imbalances(self, ratio=3.0, min_volume=10):
        """Imbalances diagonales.

        Convention standard : on compare le volume ASK au prix P au volume BID
        au prix P - 1 tick. Retourne [(prix, 'buy'|'sell', ratio)].
        """
        out = []
        for key in sorted(self.levels):
            ask_here = self.levels[key][1]
            bid_below = self.levels.get(key - 1, [0, 0])[0]
            if ask_here >= min_volume and bid_below * ratio <= ask_here:
                r = ask_here / bid_below if bid_below else float("inf")
                out.append((self.price(key), "buy", r))
            bid_here = self.levels[key][0]
            ask_above = self.levels.get(key + 1, [0, 0])[1]
            if bid_here >= min_volume and ask_above * ratio <= bid_here:
                r = bid_here / ask_above if ask_above else float("inf")
                out.append((self.price(key), "sell", r))
        return out

Deux détails qui comptent :

bid_below * ratio <= ask_here plutôt que ask_here / bid_below >= ratio : la forme multiplicative évite la division par zéro, cas fréquent quand un niveau n'a aucun volume d'un côté.
min_volume est appliqué avant tout : sans lui, 3 × 1 contrat déclenche une imbalance dépourvue de sens.

Les imbalances empilées¶

    def stacked(self, n=3, **kw):
        """Séquences de n imbalances de même sens sur des ticks consécutifs."""
        found = self.imbalances(**kw)
        by_side = defaultdict(list)
        for price, side, _ in found:
            by_side[side].append(round(price / self.tick))
        out = []
        for side, keys in by_side.items():
            keys.sort()
            run = [keys[0]]
            for k in keys[1:]:
                if k == run[-1] + 1:
                    run.append(k)
                else:
                    if len(run) >= n:
                        out.append((side, self.price(run[0]), self.price(run[-1])))
                    run = [k]
            if len(run) >= n:
                out.append((side, self.price(run[0]), self.price(run[-1])))
        return out

Un parcours linéaire des clés triées, avec détection de séquences consécutives. La dernière séquence est traitée après la boucle — oubli classique qui fait manquer les piles terminales, c'est-à-dire les plus récentes, donc les seules exploitables en temps réel.

L'OFI¶

def ofi_increment(prev, cur):
    """Order Flow Imbalance de Cont, Kukanov & Stoikov (2014), un événement.

    prev / cur : (bid_price, bid_size, ask_price, ask_size).
    """
    pb0, qb0, pa0, qa0 = prev
    pb1, qb1, pa1, qa1 = cur
    e = 0
    if pb1 >= pb0:
        e += qb1
    if pb1 <= pb0:
        e -= qb0
    if pa1 <= pa0:
        e -= qa1
    if pa1 >= pa0:
        e += qa0
    return e

Les quatre conditions ne sont pas exclusives : quand le prix ne change pas, les deux premières s'appliquent simultanément et l'incrément vaut \(Q_b^n - Q_b^{n-1}\), soit la variation de la file. C'est le comportement voulu, et c'est une source d'erreur classique d'implémentation (utiliser elif casse la formule).

Les tests¶

Le code non trivial doit laisser derrière lui une vérification exécutable.

def _test_imbalance():
    fp = Footprint(0.25)
    # ask 40 en 5000.25 contre bid 10 en 5000.00 -> ratio 4 -> buy imbalance
    fp.add(5000.00, 10, -1)
    fp.add(5000.25, 40, 1)
    imb = fp.imbalances(ratio=3.0, min_volume=10)
    assert any(abs(p - 5000.25) < 1e-9 and s == "buy" for p, s, _ in imb), imb
    # ratio 4 < 5 -> plus d'imbalance si on durcit le seuil
    assert fp.imbalances(ratio=5.0, min_volume=10) == []


def _test_ofi():
    prev = (5000.00, 100, 5000.25, 100)
    # bid renforcé de 50, ask inchangé
    assert ofi_increment(prev, (5000.00, 150, 5000.25, 100)) == 50
    # ask levé (le prix ask monte) : la file vendeuse disparaît
    assert ofi_increment(prev, (5000.00, 100, 5000.50, 80)) == 100
    # bid balayé : OFI négatif
    assert ofi_increment(prev, (4999.75, 90, 5000.25, 100)) < 0

Exécution :

$ python3 orderflow.py
ok  _test_absorption
ok  _test_aggressor
ok  _test_footprint
ok  _test_imbalance
ok  _test_ofi
ok  _test_stacked
tous les tests passent

Six assertions valent mieux qu'un framework

Pas de pytest, pas de fixtures, pas de mocks. Un if __name__ == "__main__" qui appelle toutes les fonctions commençant par _test_. Cela tient en cinq lignes, se lance en une commande, et attrape exactement les bugs qui comptent : inversion de signe, division par zéro, séquence terminale manquée.

La performance¶

Opération	Coût
`Footprint.add`	O(1)
`imbalances()`	O(niveaux log niveaux), typiquement < 50 niveaux
`ofi_increment`	O(1)

Sur une barre de volume de 1 000 contrats sur MES, imbalances() s'exécute en quelques dizaines de microsecondes en Python pur. Ce n'est pas le goulot d'étranglement.

Où se trouve le vrai goulot

Le coût dominant est la désérialisation protobuf et la traversée de la boucle d'événements asyncio, pas votre calcul de footprint. Profilez avant d'optimiser : l'intuition se trompe presque toujours sur ce point.

Résumé¶

Indexer les prix par entier (round(price/tick)) élimine les bugs de virgule flottante.
Forme multiplicative pour les ratios : pas de division par zéro.
Plancher de volume appliqué avant le ratio.
L'OFI utilise quatre if indépendants, jamais elif.
Six assertions dans un __main__ suffisent à couvrir les bugs réels.
Le footprint n'est pas le goulot d'étranglement : profilez avant d'optimiser.

Chapitre précédent : Connexion Rithmic en Python · Chapitre suivant : Moteur de signaux