Code complet et vérifié¶
Trois fichiers Python, sans aucune dépendance externe, exécutés et validés au moment de la rédaction de ce rapport. Copiez-les tels quels et lancez-les.
Ce qui est garanti
Ces deux fichiers ont été exécutés et leurs assertions passent. Les tableaux de la partie Économie sont la sortie réelle du second.
Les extraits de code des chapitres Connexion Rithmic, Moteur de signaux et Exécution sont en revanche illustratifs : ils dépendent d'APIs externes dont les signatures doivent être vérifiées dans leur documentation en vigueur.
orderflow.py¶
Primitives d'order flow : classification d'agresseur, footprint, imbalances, OFI, absorption.
"""Briques order flow en Python pur : classification d'agresseur, footprint,
imbalances diagonales, OFI de Cont-Kukanov-Stoikov, absorption.
Aucune dépendance externe. Lancer `python3 orderflow.py` pour les tests.
"""
from collections import defaultdict
# ---------------------------------------------------------------- 1. agresseur
def aggressor(price, bid, ask):
"""1 = acheteur agressif (trade >= ask), -1 = vendeur agressif (trade <= bid),
0 = indeterminé (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
# ---------------------------------------------------------------- 2. footprint
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)
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
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
# ---------------------------------------------------------------- 3. OFI (CKS)
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
def ofi(book_updates):
"""Somme des incréments sur une fenêtre de mises à jour du best bid/ask."""
return sum(ofi_increment(book_updates[i - 1], book_updates[i])
for i in range(1, len(book_updates)))
# ---------------------------------------------------------------- 4. absorption
def absorption(fp, price_range_ticks, min_volume, max_range_ticks):
"""Vrai si un gros volume s'échange sans que le prix parcoure de distance."""
return fp.volume >= min_volume and price_range_ticks <= max_range_ticks
# ------------------------------------------- 5. lecture du carnet et du tape
def thickening(cost_per_level):
"""Épaississement (Jigsaw) : le nombre de contrats nécessaires pour faire
céder chaque niveau successif croît-il ?
cost_per_level : contrats absorbés à chaque niveau avant qu'il ne cède,
dans l'ordre chronologique. Retourne la pente moyenne (>0 = épaississement).
"""
if len(cost_per_level) < 3:
return 0.0
deltas = [cost_per_level[i] - cost_per_level[i - 1]
for i in range(1, len(cost_per_level))]
return sum(deltas) / len(deltas)
def burst(trades, window_ns, volume_recent, min_volume, min_basis_ratio=0.10):
"""Détection de rafale (type scanner Power Trades).
trades : [(ts_ns, size, side)] déjà filtrés sur la fenêtre.
window_ns : durée de la fenêtre, en nanosecondes.
volume_recent : volume total des dernières minutes (dénominateur du ratio).
min_basis_ratio: part de la rafale dans le volume récent (~10 % élevé sur ES).
Retourne (est_rafale, volume, delta, basis_ratio).
"""
if not trades:
return False, 0, 0, 0.0
span = trades[-1][0] - trades[0][0]
vol = sum(s for _, s, _ in trades)
delta = sum(side * s for _, s, side in trades)
ratio = vol / volume_recent if volume_recent else 0.0
est = (span <= window_ns and vol >= min_volume and ratio >= min_basis_ratio)
return est, vol, delta, ratio
def iceberg_ratio(traded_volume, displayed_size):
"""Signature d'iceberg : beaucoup plus traité que ce qui était affiché,
sans que l'affichage diminue. Un ratio élevé (> 5) est suspect."""
if displayed_size <= 0:
return float("inf") if traded_volume > 0 else 0.0
return traded_volume / displayed_size
def replenishment_stopped(history, tolerance=0.3):
"""Le niveau cesse-t-il de se recharger ?
history : tailles affichées après chaque vague de consommation, dans
l'ordre. Vrai si la dernière est nettement inférieure à la médiane des
précédentes — c'est le moment de bascule avant la cassure.
"""
if len(history) < 3:
return False
passe = sorted(history[:-1])
mediane = passe[len(passe) // 2]
return mediane > 0 and history[-1] < tolerance * mediane
# ---------------------------------------------------------------------- tests
def _test_aggressor():
assert aggressor(5000.25, 5000.00, 5000.25) == 1
assert aggressor(5000.00, 5000.00, 5000.25) == -1
assert aggressor(5000.10, 5000.00, 5000.25) == 0
def _test_footprint():
fp = Footprint(0.25)
fp.add(5000.00, 10, -1)
fp.add(5000.25, 40, 1)
fp.add(5000.25, 5, -1)
assert fp.volume == 55
assert fp.delta == 40 - 15
assert fp.poc() == 5000.25
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_stacked():
fp = Footprint(0.25)
for i in range(4):
fp.add(5000.00 + i * 0.25, 10, -1)
fp.add(5000.25 + i * 0.25, 50, 1)
st = fp.stacked(n=3, ratio=3.0, min_volume=10)
assert any(side == "buy" for side, _, _ in st), st
def _test_ofi():
prev = (5000.00, 100, 5000.25, 100)
# bid renforcé de 50, ask inchangé : +150 -100 = +50
assert ofi_increment(prev, (5000.00, 150, 5000.25, 100)) == 50
# ask levé (prix ask monte) : la file vendeuse disparaît -> pression achat
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
def _test_absorption():
fp = Footprint(0.25)
fp.add(5000.00, 3000, -1)
assert absorption(fp, price_range_ticks=2, min_volume=2000, max_range_ticks=3)
assert not absorption(fp, price_range_ticks=8, min_volume=2000, max_range_ticks=3)
def _test_thickening():
# coût croissant par niveau -> épaississement
assert thickening([37, 45, 69, 72, 103]) > 0
# coût décroissant -> le plancher se dérobe
assert thickening([103, 72, 69, 45, 37]) < 0
# trop peu de niveaux pour conclure
assert thickening([37, 45]) == 0.0
def _test_burst():
ms = 1_000_000
trades = [(0, 200, -1), (50 * ms, 300, -1), (120 * ms, 300, -1)]
est, vol, delta, ratio = burst(trades, 200 * ms, volume_recent=5000,
min_volume=500)
assert est and vol == 800 and delta == -800
assert abs(ratio - 0.16) < 1e-9
# même rafale dans un marché dix fois plus actif : ratio trop faible
est2, _, _, _ = burst(trades, 200 * ms, volume_recent=50000, min_volume=500)
assert not est2
# trop lente pour être une rafale
lents = [(0, 400, -1), (900 * ms, 400, -1)]
est3, _, _, _ = burst(lents, 200 * ms, volume_recent=5000, min_volume=500)
assert not est3
def _test_iceberg():
assert iceberg_ratio(1800, 40) == 45.0
assert iceberg_ratio(120, 100) < 5
def _test_replenishment():
assert replenishment_stopped([250, 300, 280, 220, 30])
assert not replenishment_stopped([250, 300, 280, 220, 240])
assert not replenishment_stopped([250, 30])
if __name__ == "__main__":
for name, fn in sorted(globals().items()):
if name.startswith("_test_"):
fn()
print(f"ok {name}")
print("tous les tests passent")
Sortie :
$ python3 orderflow.py
ok _test_absorption
ok _test_aggressor
ok _test_burst
ok _test_footprint
ok _test_iceberg
ok _test_imbalance
ok _test_ofi
ok _test_replenishment
ok _test_stacked
ok _test_thickening
tous les tests passent
economie.py¶
Modèle de coûts, espérance et simulation de survie.
"""Modèle de coût / espérance / survie pour un bot order flow sur compte prop.
Python pur (stdlib uniquement). `python3 economie.py` imprime les tableaux
utilisés dans la partie « Économie » du rapport, et vérifie les invariants.
"""
import random
TICK_MES = 1.25 # $ par tick, Micro E-mini S&P 500
TICK_ES = 12.50 # $ par tick, E-mini S&P 500
RT_MES = 1.02 # commission aller-retour Rithmic chez Apex : 0.51 $/side
RT_ES = 3.98 # 1.99 $/side
def expectancy(win_rate, tp_ticks, sl_ticks, tick_value, rt_cost,
slippage_ticks=0.5):
"""Espérance nette par trade, en dollars par contrat."""
gross_win = tp_ticks * tick_value
gross_loss = sl_ticks * tick_value
friction = rt_cost + slippage_ticks * tick_value
return win_rate * gross_win - (1 - win_rate) * gross_loss - friction
def breakeven_win_rate(tp_ticks, sl_ticks, tick_value, rt_cost,
slippage_ticks=0.5):
"""Taux de réussite minimal pour une espérance nulle."""
friction = rt_cost + slippage_ticks * tick_value
num = sl_ticks * tick_value + friction
den = (tp_ticks + sl_ticks) * tick_value
return num / den
def simulate(win_rate, tp_ticks, sl_ticks, tick_value, rt_cost, contracts,
trades_per_day, days, trailing_dd, target, slippage_ticks=0.5,
seed=0, runs=20000):
"""Monte Carlo : proportion de comptes qui atteignent la cible avant de
toucher le drawdown trailing (drawdown depuis le plus haut)."""
rng = random.Random(seed)
friction = rt_cost + slippage_ticks * tick_value
win = tp_ticks * tick_value - friction
loss = -sl_ticks * tick_value - friction
passed = blown = 0
for _ in range(runs):
equity = 0.0
peak = 0.0
for _ in range(days * trades_per_day):
equity += (win if rng.random() < win_rate else loss) * contracts
peak = max(peak, equity)
if equity <= peak - trailing_dd:
blown += 1
break
if equity >= target:
passed += 1
break
return passed / runs, blown / runs
def _tests():
# Espérance nulle exactement au taux d'équilibre.
for tp, sl in [(8, 8), (12, 6), (4, 12)]:
wr = breakeven_win_rate(tp, sl, TICK_MES, RT_MES)
assert abs(expectancy(wr, tp, sl, TICK_MES, RT_MES)) < 1e-9
# La friction augmente strictement le taux d'équilibre.
a = breakeven_win_rate(8, 8, TICK_MES, RT_MES, slippage_ticks=0)
b = breakeven_win_rate(8, 8, TICK_MES, RT_MES, slippage_ticks=1)
assert b > a > 0.5
# Le micro subit une friction relative plus lourde que le mini.
assert (breakeven_win_rate(8, 8, TICK_MES, RT_MES)
> breakeven_win_rate(8, 8, TICK_ES, RT_ES))
# Un edge nul ne passe presque jamais une évaluation à drawdown serré.
p, _ = simulate(breakeven_win_rate(8, 8, TICK_MES, RT_MES), 8, 8, TICK_MES,
RT_MES, contracts=5, trades_per_day=10, days=30,
trailing_dd=2500, target=3000, runs=3000)
assert p < 0.55, p
if __name__ == "__main__":
_tests()
print("=== Taux de réussite d'équilibre (MES, 0.5 tick de slippage) ===")
print(f"{'TP/SL (ticks)':>16} | {'MES':>7} | {'ES':>7}")
for tp, sl in [(4, 4), (6, 6), (8, 8), (12, 12), (12, 6), (6, 12), (20, 10)]:
m = breakeven_win_rate(tp, sl, TICK_MES, RT_MES)
e = breakeven_win_rate(tp, sl, TICK_ES, RT_ES)
print(f"{tp:>7}/{sl:<8} | {m:>6.1%} | {e:>6.1%}")
print()
print("=== Espérance nette par contrat et par trade (MES, TP=SL=8 ticks) ===")
for wr in [0.50, 0.52, 0.55, 0.58, 0.60, 0.65]:
e = expectancy(wr, 8, 8, TICK_MES, RT_MES)
print(f" win rate {wr:.0%} -> {e:+6.2f} $/trade"
f" ({e * 10 * 5:+8.2f} $/jour à 10 trades x 5 contrats)")
print()
print("=== Scénario asymétrique TP 12 / SL 6 ticks, 10 MES, 6 trades/jour ===")
print(f"(équilibre à {breakeven_win_rate(12, 6, TICK_MES, RT_MES):.1%})")
for wr in [0.42, 0.45, 0.48, 0.50, 0.55]:
e = expectancy(wr, 12, 6, TICK_MES, RT_MES)
print(f" win rate {wr:.0%} -> {e:+5.2f} $/contrat/trade"
f" | {e * 10 * 6:+6.0f} $/jour | {e * 10 * 6 * 21:+7.0f} $/mois")
print()
print("=== Évaluation 50 K (cible 3 000 $, trailing 2 500 $), 40 jours ===")
print(f"{'win rate':>9} | {'passée':>7} | {'grillée':>8} | {'en cours':>9}")
for wr in [0.45, 0.48, 0.50, 0.55]:
p, b = simulate(wr, 12, 6, TICK_MES, RT_MES, contracts=10,
trades_per_day=6, days=40, trailing_dd=2500,
target=3000, seed=11, runs=8000)
print(f"{wr:>8.0%} | {p:>6.1%} | {b:>7.1%} | {1 - p - b:>8.1%}")
Sortie :
$ python3 economie.py
=== Taux de réussite d'équilibre (MES, 0.5 tick de slippage) ===
TP/SL (ticks) | MES | ES
4/4 | 66.5% | 60.2%
6/6 | 61.0% | 56.8%
8/8 | 58.2% | 55.1%
12/12 | 55.5% | 53.4%
12/6 | 40.6% | 37.9%
6/12 | 74.0% | 71.2%
20/10 | 37.7% | 36.1%
=== Espérance nette par contrat et par trade (MES, TP=SL=8 ticks) ===
win rate 50% -> -1.65 $/trade ( -82.25 $/jour à 10 trades x 5 contrats)
win rate 52% -> -1.24 $/trade ( -62.25 $/jour à 10 trades x 5 contrats)
win rate 55% -> -0.65 $/trade ( -32.25 $/jour à 10 trades x 5 contrats)
win rate 58% -> -0.05 $/trade ( -2.25 $/jour à 10 trades x 5 contrats)
win rate 60% -> +0.35 $/trade ( +17.75 $/jour à 10 trades x 5 contrats)
win rate 65% -> +1.35 $/trade ( +67.75 $/jour à 10 trades x 5 contrats)
=== Scénario asymétrique TP 12 / SL 6 ticks, 10 MES, 6 trades/jour ===
(équilibre à 40.6%)
win rate 42% -> +0.30 $/contrat/trade | +18 $/jour | +384 $/mois
win rate 45% -> +0.98 $/contrat/trade | +59 $/jour | +1235 $/mois
win rate 48% -> +1.65 $/contrat/trade | +99 $/jour | +2085 $/mois
win rate 50% -> +2.10 $/contrat/trade | +126 $/jour | +2652 $/mois
win rate 55% -> +3.23 $/contrat/trade | +194 $/jour | +4070 $/mois
=== Évaluation 50 K (cible 3 000 $, trailing 2 500 $), 40 jours ===
win rate | passée | grillée | en cours
45% | 45.8% | 3.5% | 50.7%
48% | 78.9% | 0.7% | 20.4%
50% | 92.3% | 0.2% | 7.4%
55% | 99.8% | 0.0% | 0.2%
Rappel sur la simulation
Les colonnes « grillée » sont des planchers. Le modèle suppose des trades indépendants et identiquement distribués, hypothèse violée par le regroupement des pertes en régime défavorable. Voir Espérance et dimensionnement.
metrics.py¶
Étiquetage en triple barrière, métriques d'edge, et Sharpe dégonflé. Support du chapitre Validation statistique et du chapitre Mesurer et surveiller l'edge.
"""Métriques d'edge et validation statistique, en Python pur.
Triple-barrière (López de Prado), R-multiples, profit factor, espérance, SQN,
MAE/MFE, Ulcer Index, Probabilistic & Deflated Sharpe Ratio, Minimum Backtest
Length.
Aucune dépendance externe. `python3 metrics.py` lance les tests.
"""
import math
# ------------------------------------------------------- 1. loi normale (stdlib)
def norm_cdf(x):
"""Fonction de répartition de la loi normale centrée réduite."""
return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))
def norm_ppf(p, lo=-40.0, hi=40.0, tol=1e-12):
"""Quantile de la loi normale, par bissection."""
if not 0.0 < p < 1.0:
raise ValueError("p doit être dans ]0,1[")
for _ in range(200):
mid = (lo + hi) / 2.0
if norm_cdf(mid) < p:
lo = mid
else:
hi = mid
if hi - lo < tol:
break
return (lo + hi) / 2.0
# ----------------------------------------------- 2. triple-barrière (labelling)
def triple_barrier(prices, entry_idx, side, tp_ticks, sl_ticks, max_bars,
tick_size):
"""Étiquette un trade par la PREMIÈRE barrière touchée.
prices : séquence de (high, low) par barre, en unités de prix.
side : +1 long, -1 short.
Retourne (label, bars_held, mfe_ticks, mae_ticks) avec label dans
{+1 gain, -1 perte, 0 sortie au temps}.
"""
entry = prices[entry_idx][0] if side > 0 else prices[entry_idx][1]
tp = entry + side * tp_ticks * tick_size
sl = entry - side * sl_ticks * tick_size
mfe = mae = 0.0
for i in range(entry_idx + 1, min(entry_idx + 1 + max_bars, len(prices))):
hi, lo = prices[i]
fav = (hi - entry) if side > 0 else (entry - lo)
adv = (entry - lo) if side > 0 else (hi - entry)
mfe = max(mfe, fav / tick_size)
mae = max(mae, adv / tick_size)
touched_sl = (lo <= sl) if side > 0 else (hi >= sl)
touched_tp = (hi >= tp) if side > 0 else (lo <= tp)
# Conservateur : si les deux barrières sont touchées dans la même barre,
# on suppose le stop d'abord. Sans données intrabarre, c'est la seule
# hypothèse honnête.
if touched_sl:
return -1, i - entry_idx, mfe, mae
if touched_tp:
return 1, i - entry_idx, mfe, mae
return 0, min(max_bars, len(prices) - 1 - entry_idx), mfe, mae
# --------------------------------------------------------- 3. métriques d'edge
def profit_factor(results):
"""Gains bruts / pertes brutes. 1.0 = seuil de rentabilité."""
gains = sum(r for r in results if r > 0)
pertes = -sum(r for r in results if r < 0)
if pertes == 0:
return float("inf") if gains > 0 else 0.0
return gains / pertes
def expectancy(results):
"""Résultat moyen par trade, dans l'unité des résultats."""
return sum(results) / len(results) if results else 0.0
def r_multiples(results, risk_per_trade):
"""Convertit des résultats en devise en multiples du risque initial."""
return [r / risk_per_trade for r in results]
def stdev(xs):
if len(xs) < 2:
return 0.0
m = sum(xs) / len(xs)
return math.sqrt(sum((x - m) ** 2 for x in xs) / (len(xs) - 1))
def sqn(r_list, cap=100):
"""System Quality Number (Van Tharp) : (moyenne R / écart-type R) * sqrt(n).
n est plafonné (100 par défaut) : sans plafond, la racine gonfle
mécaniquement le score des systèmes très actifs.
"""
if len(r_list) < 2:
return 0.0
sd = stdev(r_list)
if sd == 0:
return 0.0
n = min(len(r_list), cap)
return (sum(r_list) / len(r_list)) / sd * math.sqrt(n)
def ulcer_index(equity):
"""Racine de la moyenne des drawdowns relatifs au carré, en %."""
if not equity:
return 0.0
peak = equity[0]
carres = []
for v in equity:
peak = max(peak, v)
dd = 0.0 if peak == 0 else (v - peak) / peak * 100.0
carres.append(dd * dd)
return math.sqrt(sum(carres) / len(carres))
def max_drawdown(equity):
"""Drawdown maximal en valeur absolue (unité de l'equity)."""
peak = equity[0] if equity else 0.0
worst = 0.0
for v in equity:
peak = max(peak, v)
worst = min(worst, v - peak)
return -worst
# ------------------------------------- 4. Sharpe dégonflé (Bailey & LdP, 2014)
def _moments(returns):
n = len(returns)
m = sum(returns) / n
sd = stdev(returns)
if sd == 0:
return m, 0.0, 0.0, 3.0
skew = sum(((r - m) / sd) ** 3 for r in returns) / n
kurt = sum(((r - m) / sd) ** 4 for r in returns) / n
return m, sd, skew, kurt
def sharpe(returns):
m, sd, _, _ = _moments(returns)
return m / sd if sd else 0.0
def probabilistic_sharpe(returns, sr_benchmark=0.0):
"""PSR : probabilité que le vrai Sharpe dépasse `sr_benchmark`."""
n = len(returns)
if n < 3:
return 0.5
sr = sharpe(returns)
_, _, skew, kurt = _moments(returns)
denom = 1.0 - skew * sr + (kurt - 1.0) / 4.0 * sr * sr
if denom <= 0:
return 1.0 if sr > sr_benchmark else 0.0
return norm_cdf((sr - sr_benchmark) * math.sqrt(n - 1) / math.sqrt(denom))
EULER = 0.5772156649015329
def expected_max_sharpe(n_trials, var_sharpe):
"""Sharpe maximal attendu par pur hasard après `n_trials` essais."""
if n_trials < 2:
return 0.0
a = norm_ppf(1.0 - 1.0 / n_trials)
b = norm_ppf(1.0 - 1.0 / (n_trials * math.e))
return math.sqrt(var_sharpe) * ((1.0 - EULER) * a + EULER * b)
def deflated_sharpe(returns, n_trials, var_sharpe):
"""DSR : PSR dont le seuil monte avec le nombre d'essais de la recherche.
var_sharpe : variance des Sharpe observés sur l'ensemble des essais.
Compter TOUS les essais, brouillons compris — sinon le test ne sert à rien.
"""
return probabilistic_sharpe(returns, expected_max_sharpe(n_trials, var_sharpe))
def min_backtest_length(n_trials, target_sharpe_annual):
"""Longueur minimale d'historique (en années) pour qu'un Sharpe annuel
donné soit crédible après `n_trials` essais (Bailey & al.)."""
if target_sharpe_annual <= 0:
return float("inf")
e = expected_max_sharpe(n_trials, 1.0)
return (e / target_sharpe_annual) ** 2
# ----------------------------------------------------------------- tests
def _test_norm():
assert abs(norm_cdf(0.0) - 0.5) < 1e-12
assert abs(norm_cdf(1.96) - 0.975) < 1e-3
assert abs(norm_ppf(0.975) - 1.959964) < 1e-4
assert abs(norm_ppf(norm_cdf(0.7)) - 0.7) < 1e-6
def _test_triple_barrier():
tick = 0.25
# montée régulière : le TP (8 ticks = 2.00) est touché avant le SL
prices = [(100.0 + i * 0.25, 99.9 + i * 0.25) for i in range(20)]
label, bars, mfe, mae = triple_barrier(prices, 0, 1, 8, 8, 15, tick)
assert label == 1, (label, bars)
assert mfe >= 8
# baisse régulière : le stop part en premier
prices = [(100.0 - i * 0.25, 99.9 - i * 0.25) for i in range(20)]
label, _, _, mae = triple_barrier(prices, 0, 1, 8, 8, 15, tick)
assert label == -1
assert mae >= 8
# marché plat : sortie au temps
prices = [(100.05, 99.95)] * 20
label, bars, _, _ = triple_barrier(prices, 0, 1, 8, 8, 10, tick)
assert label == 0 and bars == 10
def _test_edge_metrics():
res = [100, -50, 100, -50, -50, 200]
assert abs(profit_factor(res) - 400 / 150) < 1e-9
assert abs(expectancy(res) - 250 / 6) < 1e-9
rs = r_multiples(res, 50)
assert rs[0] == 2.0 and rs[1] == -1.0
# système régulier -> SQN plus élevé qu'un système erratique de même moyenne
regulier = [0.25] * 50 + [0.35] * 50
erratique = [3.0] * 10 + [-0.006] * 90
assert abs(expectancy(regulier) - expectancy(erratique)) < 0.02
assert sqn(regulier) > sqn(erratique)
def _test_drawdown():
eq = [100, 110, 105, 120, 90, 130]
assert abs(max_drawdown(eq) - 30) < 1e-9
assert ulcer_index([100, 100, 100]) == 0.0
assert ulcer_index(eq) > 0.0
def _test_sharpe():
import random
rng = random.Random(11)
# 400 trades à edge faible mais réel (Sharpe par trade ~0.1)
bons = [rng.gauss(0.10, 1.0) for _ in range(400)]
psr = probabilistic_sharpe(bons)
assert psr > 0.9, psr
# série sans edge -> PSR proche de 0.5
plats = [0.01, -0.01] * 75
assert 0.3 < probabilistic_sharpe(plats) < 0.7
# le DSR est strictement plus sévère que le PSR dès qu'il y a des essais
dsr = deflated_sharpe(bons, 200, 0.01)
assert dsr < psr, (dsr, psr)
# et plus il y a d'essais, plus il est sévère
assert deflated_sharpe(bons, 1000, 0.01) < deflated_sharpe(bons, 10, 0.01)
# plus d'essais -> seuil plus haut
assert expected_max_sharpe(1000, 0.5) > expected_max_sharpe(10, 0.5)
# MinBTL croît avec le nombre d'essais et décroît avec le Sharpe visé
assert min_backtest_length(1000, 1.0) > min_backtest_length(10, 1.0)
assert min_backtest_length(100, 2.0) < min_backtest_length(100, 1.0)
if __name__ == "__main__":
for name, fn in sorted(globals().items()):
if name.startswith("_test_"):
fn()
print(f"ok {name}")
print("tous les tests passent")
print()
print("=== Sharpe maximal attendu par pur hasard (variance des essais = 0.5) ===")
for n in [10, 50, 100, 500, 1000, 5000]:
print(f" {n:>5} essais -> Sharpe attendu par hasard : "
f"{expected_max_sharpe(n, 0.5):.2f}")
print()
print("=== Historique minimal (années) pour croire à un Sharpe annuel ===")
print(f"{'essais':>8} | {'SR=1.0':>8} | {'SR=1.5':>8} | {'SR=2.0':>8}")
for n in [10, 50, 100, 500, 1000]:
print(f"{n:>8} | {min_backtest_length(n, 1.0):>7.1f}a | "
f"{min_backtest_length(n, 1.5):>7.1f}a | "
f"{min_backtest_length(n, 2.0):>7.1f}a")
Sortie :
$ python3 metrics.py
ok _test_drawdown
ok _test_edge_metrics
ok _test_norm
ok _test_sharpe
ok _test_triple_barrier
tous les tests passent
=== Sharpe maximal attendu par pur hasard (variance des essais = 0.5) ===
10 essais -> Sharpe attendu par hasard : 1.11
50 essais -> Sharpe attendu par hasard : 1.61
100 essais -> Sharpe attendu par hasard : 1.79
500 essais -> Sharpe attendu par hasard : 2.16
1000 essais -> Sharpe attendu par hasard : 2.30
5000 essais -> Sharpe attendu par hasard : 2.61
=== Historique minimal (années) pour croire à un Sharpe annuel ===
essais | SR=1.0 | SR=1.5 | SR=2.0
10 | 2.5a | 1.1a | 0.6a
50 | 5.2a | 2.3a | 1.3a
100 | 6.4a | 2.8a | 1.6a
500 | 9.3a | 4.1a | 2.3a
1000 | 10.6a | 4.7a | 2.6a
Le tableau à relire deux fois
Après 1 000 essais de paramétrage, un Sharpe de 2,30 est le résultat attendu du pur hasard. Et il faut 10,6 ans d'historique pour qu'un Sharpe annuel de 1,0 soit crédible après 1 000 essais.
Avec les quelques mois de données dont dispose un retail sur les futures, la contrainte est brutale : formuler peu d'hypothèses et les tester une fois. Le balayage de grille est structurellement incompatible avec la quantité de données disponible.
Comment adapter à votre cas¶
| Variable | À modifier si |
|---|---|
RT_MES / RT_ES |
Votre firme a d'autres commissions |
slippage_ticks |
Mesurez le vôtre : prix demandé vs prix obtenu |
trailing_dd / target |
Selon le compte visé |
trades_per_day |
Fréquence réelle mesurée en paper trading |
win_rate |
Le seul que vous ne pouvez pas choisir |
Retour aux annexes