import streamlit as st
import pandas as pd
from src.data import fetch_pair_data
from src.cointegration import test_cointegration, compute_spread, compute_zscore
from src.signals import generate_signals
from src.backtest import run_backtest
from src.visualization import plot_normalized_prices, plot_spread_zscore, plot_equity_curve
# --- Configuration de la page ---
st.set_page_config(
page_title="Pairs Trading — Cointegration Dashboard",
layout="wide",
)
st.title("Pairs Trading — Cointegration Dashboard")
st.caption("Analyse de cointégration et backtest de stratégies de pairs trading")
# --- Initialisation du session_state pour les tickers ---
if "ticker_a" not in st.session_state:
st.session_state.ticker_a = "KO"
if "ticker_b" not in st.session_state:
st.session_state.ticker_b = "PEP"
# --- Sidebar : paramètres ---
with st.sidebar:
st.header("Paramètres")
periode = st.selectbox("Période de données", ["3y", "5y", "10y", "15y", "20y", "max"], index=1)
entry_threshold = st.slider(
"Z-score seuil d'entrée",
min_value=1.5, max_value=3.0, value=2.0, step=0.1,
)
exit_threshold = st.slider(
"Z-score seuil de sortie",
min_value=0.0, max_value=1.0, value=0.5, step=0.1,
)
rolling_window = st.slider(
"Fenêtre rolling (jours)",
min_value=20, max_value=120, value=60, step=5,
)
transaction_cost = st.slider(
"Coût de transaction (%)",
min_value=0.0, max_value=0.5, value=0.1, step=0.05,
)
# --- Sélection des tickers ---
# --- Paires suggérées ---
st.markdown("**Paires suggérées :**")
paires_suggerees = {
"V / MA": ("V", "MA"),
"MCD / YUM": ("MCD", "YUM"),
"PG / CL": ("PG", "CL"),
"XOM / CVX": ("XOM", "CVX"),
"KO / PEP": ("KO", "PEP"),
}
colonnes_paires = st.columns(len(paires_suggerees))
for i, (nom, (ta, tb)) in enumerate(paires_suggerees.items()):
with colonnes_paires[i]:
if st.button(nom, use_container_width=True):
st.session_state.ticker_a = ta
st.session_state.ticker_b = tb
st.rerun()
# --- Sélection des tickers ---
col_input_a, col_input_b, col_btn = st.columns([2, 2, 1])
with col_input_a:
ticker_a = st.text_input("Ticker A", key="ticker_a")
with col_input_b:
ticker_b = st.text_input("Ticker B", key="ticker_b")
with col_btn:
st.write("") # Espacement vertical
st.write("")
bouton_analyser = st.button("Analyser", type="primary", use_container_width=True)
# --- Analyse ---
if bouton_analyser and ticker_a and ticker_b:
st.divider()
# Récupération des données
with st.spinner(f"Récupération des données pour {ticker_a} / {ticker_b}..."):
try:
prix = fetch_pair_data(ticker_a, ticker_b, period=periode)
except ValueError as e:
st.error(f"Erreur : {e}")
st.stop()
# Test de cointégration
resultats_coint = test_cointegration(prix[ticker_a], prix[ticker_b])
# --- Résultats du test ---
st.header("Résultats du test de cointégration")
col_stat, col_pval, col_beta = st.columns(3)
with col_stat:
valeur_critique_5 = resultats_coint["critical_values"]["5%"]
st.metric(
"Statistique de test",
f"{resultats_coint['test_statistic']:.4f}",
delta=f"seuil 5% : {valeur_critique_5:.4f}",
delta_color="off",
)
with col_pval:
st.metric("P-value", f"{resultats_coint['p_value']:.6f}")
with col_beta:
st.metric("Beta (hedge ratio)", f"{resultats_coint['beta']:.4f}")
if resultats_coint["is_cointegrated"]:
st.success(
f"✅ **La paire {ticker_a}/{ticker_b} est cointégrée** (p-value = {resultats_coint['p_value']:.4f}). "
"Le spread montre un comportement de retour à la moyenne — les signaux de pairs trading sont exploitables."
)
else:
st.warning(
f"⚠️ **La paire {ticker_a}/{ticker_b} n'est PAS cointégrée** (p-value = {resultats_coint['p_value']:.4f}). "
"Le spread ne montre pas de comportement de retour à la moyenne. "
"Les signaux de pairs trading seraient peu fiables. Essayez une autre paire ou ajustez la période."
)
# Valeurs critiques
with st.expander("Valeurs critiques du test"):
for niveau, valeur in resultats_coint["critical_values"].items():
st.write(f"**{niveau}** : {valeur}")
# --- Graphique 1 : Prix normalisés ---
st.header("Prix normalisés (base 100)")
fig_prix = plot_normalized_prices(prix)
st.plotly_chart(fig_prix, use_container_width=True)
# --- Calcul du spread et z-score ---
beta = resultats_coint["beta"]
spread = compute_spread(prix[ticker_a], prix[ticker_b], beta)
zscore = compute_zscore(spread, window=rolling_window)
# --- Graphique 2 : Spread & Z-Score ---
st.header("Spread & Z-Score")
fig_spread = plot_spread_zscore(spread, zscore, entry_threshold, exit_threshold)
st.plotly_chart(fig_spread, use_container_width=True)
# --- Backtest (uniquement si cointégrée ou si l'utilisateur veut quand même) ---
if not resultats_coint["is_cointegrated"]:
afficher_backtest = st.checkbox(
"Afficher le backtest quand même (résultats peu fiables)", value=False
)
else:
afficher_backtest = True
if afficher_backtest:
# Génération des signaux
signals = generate_signals(zscore, entry_threshold, exit_threshold)
# Backtest
resultats_bt = run_backtest(
prix[ticker_a], prix[ticker_b],
signals, beta,
transaction_cost=transaction_cost / 100,
)
# --- Graphique 3 : Equity Curve ---
st.header("Courbe d'Equity (Backtest)")
fig_equity = plot_equity_curve(resultats_bt["equity_curve"])
st.plotly_chart(fig_equity, use_container_width=True)
# --- Métriques de performance ---
st.header("Métriques de performance")
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Rendement total", f"{resultats_bt['total_return_pct']:.1f}%")
st.metric("Rendement annualisé", f"{resultats_bt['annualized_return_pct']:.1f}%")
with col2:
st.metric("Sharpe Ratio", f"{resultats_bt['sharpe_ratio']:.2f}")
st.metric("Profit Factor", f"{resultats_bt['profit_factor']:.2f}")
with col3:
st.metric("Max Drawdown", f"{resultats_bt['max_drawdown']:.2f} $")
st.metric("Win Rate", f"{resultats_bt['win_rate']:.1f}%")
with col4:
st.metric("Nombre de trades", f"{resultats_bt['num_trades']}")
st.metric("Durée moyenne", f"{resultats_bt['avg_trade_duration']:.0f} jours")
# --- Log des trades ---
st.header("Log des trades")
if resultats_bt["trade_log"]:
df_trades = pd.DataFrame(resultats_bt["trade_log"])
df_trades["date_entree"] = pd.to_datetime(df_trades["date_entree"]).dt.date
df_trades["date_sortie"] = pd.to_datetime(df_trades["date_sortie"]).dt.date
# Coloration des lignes selon le P&L
st.dataframe(
df_trades.style.map(
lambda v: "color: #00d4aa" if isinstance(v, (int, float)) and v > 0
else ("color: #ff6b6b" if isinstance(v, (int, float)) and v < 0 else ""),
subset=["pnl"],
),
use_container_width=True,
hide_index=True,
)
else:
st.info("Aucun trade détecté sur cette période.")
# --- Footer ---
st.divider()
st.caption("© 2026 — Pairs Trading Dashboard | Développé par Celestin Guilhen | Données fournies par Yahoo Finance")