from math import log, sqrt, exp, erf, pi
def ncdf(x):
return 0.5 * (1.0 + erf(x / sqrt(2.0)))
def npdf(x):
return exp(-x * x / 2.0) / sqrt(2.0 * pi)
def greeks(kind, S, K, T, r, vol):
"""Black-Scholes europeen, sans dividende."""
if T <= 0 or vol <= 0:
raise ValueError("T et vol doivent etre > 0")
sq = vol * sqrt(T)
d1 = (log(S / K) + (r + vol * vol / 2) * T) / sq
d2 = d1 - sq
disc = exp(-r * T)
if kind == "C":
price = S * ncdf(d1) - K * disc * ncdf(d2)
delta = ncdf(d1)
theta = (-S * npdf(d1) * vol / (2 * sqrt(T))
- r * K * disc * ncdf(d2)) / 365
else:
price = K * disc * ncdf(-d2) - S * ncdf(-d1)
delta = ncdf(d1) - 1.0
theta = (-S * npdf(d1) * vol / (2 * sqrt(T))
+ r * K * disc * ncdf(-d2)) / 365
gamma = npdf(d1) / (S * sq)
vega = S * npdf(d1) * sqrt(T) / 100
return dict(price=price, delta=delta, gamma=gamma,
vega=vega, theta=theta)
def payoff(legs, S):
"""P&L a l'echeance, par unite de sous-jacent."""
total = 0.0
for leg in legs:
if leg.kind == "U":
intrinsic = S
elif leg.kind == "C":
intrinsic = max(S - leg.strike, 0.0)
else:
intrinsic = max(leg.strike - S, 0.0)
total += leg.qty * leg.way * (intrinsic - leg.premium)
return total
def metrics(legs, hi, n=1600):
"""Les extremes tombent sur un strike ou un bord."""
grid = [0.0, hi] + [l.strike for l in legs if l.kind != "U"]
values = [payoff(legs, s) for s in grid if 0 <= s <= hi]
step = hi / n
slope = payoff(legs, hi) - payoff(legs, hi - step)
return dict(gain_max=max(values),
perte_max=min(values),
gain_illimite=slope > 1e-9,
perte_illimitee=slope < -1e-9)
def breakevens(legs, lo, hi, n=2000):
out, prev = [], None
for i in range(n + 1):
S = lo + (hi - lo) * i / n
v = payoff(legs, S)
if prev and (prev[1] > 0) != (v > 0):
w = abs(prev[1]) / (abs(prev[1]) + abs(v))
out.append(prev[0] + (S - prev[0]) * w)
prev = (S, v)
return outfrom math import log, sqrt, exp, erf, pi
def ncdf(x):
return 0.5 * (1.0 + erf(x / sqrt(2.0)))
def npdf(x):
return exp(-x * x / 2.0) / sqrt(2.0 * pi)
def greeks(kind, S, K, T, r, vol):
"""Black-Scholes europeen, sans dividende."""
if T <= 0 or vol <= 0:
raise ValueError("T et vol doivent etre > 0")
sq = vol * sqrt(T)
d1 = (log(S / K) + (r + vol * vol / 2) * T) / sq
d2 = d1 - sq
disc = exp(-r * T)
if kind == "C":
price = S * ncdf(d1) - K * disc * ncdf(d2)
delta = ncdf(d1)
theta = (-S * npdf(d1) * vol / (2 * sqrt(T))
- r * K * disc * ncdf(d2)) / 365
else:
price = K * disc * ncdf(-d2) - S * ncdf(-d1)
delta = ncdf(d1) - 1.0
theta = (-S * npdf(d1) * vol / (2 * sqrt(T))
+ r * K * disc * ncdf(-d2)) / 365
gamma = npdf(d1) / (S * sq)
vega = S * npdf(d1) * sqrt(T) / 100
return dict(price=price, delta=delta, gamma=gamma,
vega=vega, theta=theta)
def payoff(legs, S):
"""P&L a l'echeance, par unite de sous-jacent."""
total = 0.0
for leg in legs:
if leg.kind == "U":
intrinsic = S
elif leg.kind == "C":
intrinsic = max(S - leg.strike, 0.0)
else:
intrinsic = max(leg.strike - S, 0.0)
total += leg.qty * leg.way * (intrinsic - leg.premium)
return total
def metrics(legs, hi, n=1600):
"""Les extremes tombent sur un strike ou un bord."""
grid = [0.0, hi] + [l.strike for l in legs if l.kind != "U"]
values = [payoff(legs, s) for s in grid if 0 <= s <= hi]
step = hi / n
slope = payoff(legs, hi) - payoff(legs, hi - step)
return dict(gain_max=max(values),
perte_max=min(values),
gain_illimite=slope > 1e-9,
perte_illimitee=slope < -1e-9)
def breakevens(legs, lo, hi, n=2000):
out, prev = [], None
for i in range(n + 1):
S = lo + (hi - lo) * i / n
v = payoff(legs, S)
if prev and (prev[1] > 0) != (v > 0):
w = abs(prev[1]) / (abs(prev[1]) + abs(v))
out.append(prev[0] + (S - prev[0]) * w)
prev = (S, v)
return out
import numpy as np
import statsmodels.api as sm
from statsmodels.tsa.stattools import adfuller
def hedge_ratio(a, b):
"""Beta par moindres carres : a = alpha + beta * b."""
X = sm.add_constant(b)
return sm.OLS(a, X).fit().params[1]
def spread(a, b):
beta = hedge_ratio(a, b)
return a - beta * b, beta
def is_cointegrated(a, b, seuil=0.05):
ecart, beta = spread(a, b)
stat, pvalue, *_ = adfuller(ecart, maxlag=1)
return pvalue < seuil, stat, pvalue, beta
def zscore(ecart, fenetre=60):
mu = ecart.rolling(fenetre).mean()
sigma = ecart.rolling(fenetre).std(ddof=1)
return (ecart - mu) / sigma
def demi_vie(ecart):
"""Vitesse de retour a la moyenne, AR(1)."""
lag = ecart.shift(1).dropna()
delta = ecart.diff().dropna()
phi = sm.OLS(delta, sm.add_constant(lag)).fit().params[1]
if phi >= 0:
return np.inf
return -np.log(2) / np.log(1 + phi)
def signaux(z, entree=2.0, sortie=0.0, stop=3.0):
position = 0
for t, valeur in z.items():
if position == 0 and abs(valeur) >= entree:
position = -1 if valeur > 0 else 1
yield t, "ENTREE", position
elif position and abs(valeur) >= stop:
position = 0
yield t, "STOP", 0
elif position and valeur * position >= sortie:
position = 0
yield t, "SORTIE", 0
def taille_position(capital, risque_pct, perte_max_unitaire):
"""La taille vient du stop, jamais de la prime."""
if perte_max_unitaire <= 0:
raise ValueError("perte maximale inconnue")
budget = capital * risque_pct / 100.0
return int(budget // perte_max_unitaire)
def envoyer_alerte(url, payload, timeout=5):
import requests
reponse = requests.post(url, json=payload, timeout=timeout)
reponse.raise_for_status()
journal.append(dict(horodatage=now(), **payload))
return reponse.status_codeimport numpy as np
import statsmodels.api as sm
from statsmodels.tsa.stattools import adfuller
def hedge_ratio(a, b):
"""Beta par moindres carres : a = alpha + beta * b."""
X = sm.add_constant(b)
return sm.OLS(a, X).fit().params[1]
def spread(a, b):
beta = hedge_ratio(a, b)
return a - beta * b, beta
def is_cointegrated(a, b, seuil=0.05):
ecart, beta = spread(a, b)
stat, pvalue, *_ = adfuller(ecart, maxlag=1)
return pvalue < seuil, stat, pvalue, beta
def zscore(ecart, fenetre=60):
mu = ecart.rolling(fenetre).mean()
sigma = ecart.rolling(fenetre).std(ddof=1)
return (ecart - mu) / sigma
def demi_vie(ecart):
"""Vitesse de retour a la moyenne, AR(1)."""
lag = ecart.shift(1).dropna()
delta = ecart.diff().dropna()
phi = sm.OLS(delta, sm.add_constant(lag)).fit().params[1]
if phi >= 0:
return np.inf
return -np.log(2) / np.log(1 + phi)
def signaux(z, entree=2.0, sortie=0.0, stop=3.0):
position = 0
for t, valeur in z.items():
if position == 0 and abs(valeur) >= entree:
position = -1 if valeur > 0 else 1
yield t, "ENTREE", position
elif position and abs(valeur) >= stop:
position = 0
yield t, "STOP", 0
elif position and valeur * position >= sortie:
position = 0
yield t, "SORTIE", 0
def taille_position(capital, risque_pct, perte_max_unitaire):
"""La taille vient du stop, jamais de la prime."""
if perte_max_unitaire <= 0:
raise ValueError("perte maximale inconnue")
budget = capital * risque_pct / 100.0
return int(budget // perte_max_unitaire)
def envoyer_alerte(url, payload, timeout=5):
import requests
reponse = requests.post(url, json=payload, timeout=timeout)
reponse.raise_for_status()
journal.append(dict(horodatage=now(), **payload))
return reponse.status_code