04 · Le moteur de signaux¶
Le moteur de signaux transforme un état de marché en intentions de trading. Il est le cœur intellectuel du bot — et la partie la plus simple du code, si l'architecture est correcte.
Le contrat¶
def signaux(etat: EtatMarche, ctx: Contexte, params: Params) -> list[Intent]:
"""Fonction PURE. Pas d'I/O, pas d'horloge, pas d'aléatoire, pas d'état
global. Les mêmes entrées produisent toujours les mêmes sorties."""
Le test qui valide l'architecture
assert signaux(e, c, p) == signaux(e, c, p)
Si cette assertion échoue, votre moteur n'est pas pur : il lit l'heure courante, un compteur global, ou une source externe. Il est alors impossible à backtester correctement et ses bugs ne seront pas reproductibles.
Cette assertion d'une ligne vaut plus que n'importe quel test unitaire de stratégie.
Le contexte¶
Le Contexte porte tout ce qui est relatif : distributions récentes,
percentiles, zones, régime. C'est ce qui permet aux seuils de ne pas être des
constantes en dur.
from collections import deque
from statistics import median
class Contexte:
"""Statistiques glissantes servant de référence aux seuils."""
def __init__(self, taille=60):
self.deltas = deque(maxlen=taille)
self.volumes = deque(maxlen=taille)
self.prix = deque(maxlen=taille)
def observer(self, fenetre):
self.deltas.append(abs(fenetre.delta))
self.volumes.append(fenetre.volume)
self.prix.append(fenetre.cloture)
def percentile(self, serie, q):
"""Percentile empirique, interpolation par plus proche rang."""
if len(serie) < 20:
return None # pas assez d'historique
s = sorted(serie)
return s[min(len(s) - 1, int(q * len(s)))]
def seuil_delta(self, q=0.90):
return self.percentile(self.deltas, q)
def volume_relatif(self, fenetre):
m = median(self.volumes) if self.volumes else 0
return fenetre.volume / m if m else 0.0
def efficience(self):
"""Ratio d'efficience de Kaufman sur la fenêtre glissante."""
if len(self.prix) < 10:
return None
deplacement = abs(self.prix[-1] - self.prix[0])
chemin = sum(abs(self.prix[i] - self.prix[i - 1])
for i in range(1, len(self.prix)))
return deplacement / chemin if chemin else 0.0
Le retour None n'est pas un détail
Tant que l'historique est insuffisant, percentile et efficience
retournent None, et le moteur doit refuser de trader. Sans cela, les
premières minutes après chaque démarrage produisent des signaux calculés sur
trois observations.
C'est un bug particulièrement pernicieux car il ne se manifeste jamais en backtest — où l'historique est déjà chargé — et systématiquement en production, à chaque redémarrage.
Implémentation d'une stratégie¶
def signal_absorption(etat, ctx, p):
"""Absorption à un niveau clé. Voir strategies/02 pour la spécification."""
w = etat.fenetre_courante
if w is None:
return []
# --- couche 1 : régime -------------------------------------------------
if not p.creneau.contient(etat.heure_locale):
return []
eff = ctx.efficience()
if eff is None or eff > p.efficience_max:
return []
if ctx.volume_relatif(w) < p.volume_relatif_min:
return []
# --- couche 2 : zone ---------------------------------------------------
zone = etat.zones.plus_proche(w.prix_extreme, p.tolerance_ticks)
if zone is None:
return []
# --- couche 3 : déclencheur --------------------------------------------
seuil = ctx.seuil_delta(p.quantile_delta)
if seuil is None:
return []
absorption_vendeuse = (
w.delta >= seuil
and w.amplitude_ticks <= p.amplitude_max_ticks
and w.cloture <= w.ouverture + etat.tick
)
absorption_acheteuse = (
w.delta <= -seuil
and w.amplitude_ticks <= p.amplitude_max_ticks
and w.cloture >= w.ouverture - etat.tick
)
if not (absorption_vendeuse or absorption_acheteuse):
return []
# --- confirmation ------------------------------------------------------
if not etat.confirmation_disponible(p.delai_confirmation_s):
return [] # on attend, on n'entre pas
if absorption_vendeuse and not etat.delta_retourne_negatif():
return []
if absorption_acheteuse and not etat.delta_retourne_positif():
return []
# --- construction de l'intent ------------------------------------------
side = Side.SELL if absorption_vendeuse else Side.BUY
stop = (w.prix_max + 2 * etat.tick if side is Side.SELL
else w.prix_min - 2 * etat.tick)
distance = abs(etat.dernier_prix - stop)
cible = (etat.dernier_prix - p.ratio_cible * distance if side is Side.SELL
else etat.dernier_prix + p.ratio_cible * distance)
return [Intent(
symbol=etat.symbole,
side=side,
size=0, # dimensionné par la couche risque
stop_price=stop,
target_price=cible,
reason=f"absorption:{zone.nom}:delta={w.delta}:seuil={seuil}",
max_hold_seconds=p.duree_max_s,
)]
Le champ reason n'est pas cosmétique
Il est écrit dans le journal avec chaque intent. Trois mois plus tard, quand vous chercherez pourquoi la performance s'est dégradée, il permettra de ventiler les résultats par zone, par niveau de delta, par seuil déclencheur.
Sans lui, vous aurez une courbe de capital et aucune explication. Le champ coûte une f-string.
Les paramètres¶
Externalisés, jamais en dur :
from dataclasses import dataclass
@dataclass(frozen=True)
class ParamsAbsorption:
quantile_delta: float = 0.90
amplitude_max_ticks: int = 3
tolerance_ticks: int = 3
efficience_max: float = 0.40
volume_relatif_min: float = 0.80
ratio_cible: float = 2.5
delai_confirmation_s: int = 60
duree_max_s: int = 180
Le nombre de paramètres est un budget de surajustement
Huit paramètres sur 200 trades, c'est 25 observations par paramètre. C'est la limite basse du raisonnable. Chaque paramètre ajouté augmente la probabilité de trouver une combinaison qui fonctionne par hasard sur l'échantillon.
Règle pratique : au moins 30 trades par paramètre optimisé. Si vous ne les avez pas, fixez le paramètre à une valeur raisonnable et ne l'optimisez pas.
La composition de plusieurs stratégies¶
def moteur(etat, ctx, config):
intents = []
for strategie, params in config.strategies:
intents.extend(strategie(etat, ctx, params))
return arbitrer(intents)
def arbitrer(intents):
"""Résout les conflits. En cas de désaccord, on ne fait rien."""
if not intents:
return []
sides = {i.side for i in intents}
if len(sides) > 1:
journal.write({"type": "conflit_signaux",
"raisons": [i.reason for i in intents]})
return [] # signaux contradictoires -> abstention
return [max(intents, key=lambda i: i.confiance)]
L'abstention est la bonne réponse au conflit
Deux stratégies en désaccord signalent une situation ambiguë. Choisir la « meilleure » revient à parier sur laquelle des deux est fiable — pari pour lequel vous n'avez aucune information.
Ne rien faire coûte un trade manqué. Choisir au hasard coûte la moitié des trades ambigus. Un conflit fréquent est par ailleurs le symptôme d'un filtre de régime absent (cf. Imbalances empilées) : c'est le problème à corriger, pas l'arbitrage.
Tester le moteur¶
Le moteur étant pur, les tests sont des assertions sur des états construits à la main :
def test_absorption_refuse_en_tendance():
etat = etat_avec(delta=5000, amplitude_ticks=2, dans_zone=True)
ctx = contexte_avec(efficience=0.8) # marché très directionnel
assert signal_absorption(etat, ctx, ParamsAbsorption()) == []
def test_absorption_refuse_sans_confirmation():
etat = etat_avec(delta=5000, amplitude_ticks=2, dans_zone=True,
confirmation=False)
ctx = contexte_avec(efficience=0.2)
assert signal_absorption(etat, ctx, ParamsAbsorption()) == []
def test_absorption_declenche():
etat = etat_avec(delta=5000, amplitude_ticks=2, dans_zone=True,
confirmation=True, delta_suivant=-800)
ctx = contexte_avec(efficience=0.2, seuil_delta=3000)
out = signal_absorption(etat, ctx, ParamsAbsorption())
assert len(out) == 1 and out[0].side is Side.SELL
Trois tests : un cas de rejet par régime, un cas de rejet par confirmation, un cas de déclenchement. C'est le minimum, et c'est suffisant pour attraper les régressions qui comptent.
Résumé¶
- Le moteur est une fonction pure ;
signaux(e,c,p) == signaux(e,c,p)est le test d'architecture. - Le contexte porte les statistiques glissantes : aucun seuil en dur.
- Retourner
Nonetant que l'historique est insuffisant — bug invisible en backtest, systématique en production. - Le champ
reasonpermet l'analyse post-mortem. - Au moins 30 trades par paramètre optimisé.
- En cas de conflit entre stratégies : abstention.
Chapitre précédent : Construire le footprint · Chapitre suivant : Exécution et gestion des ordres