diff --git a/.github/workflows/daily_update.yml b/.github/workflows/daily_update.yml index 22b8c43..99e11bf 100644 --- a/.github/workflows/daily_update.yml +++ b/.github/workflows/daily_update.yml @@ -2,64 +2,52 @@ name: Daily Portfolio Analytics on: schedule: - # Lundi-Vendredi à 21:00 UTC pour s'assurer que les données API sont à jour - - cron: '0 21 * * 1-5' - workflow_dispatch: # Bouton manuel - -# permissions: -# contents: write -# actions: read + - cron: '0 21 * * 1-5' # 21h00, du lundi au vendredi + workflow_dispatch: jobs: portfolio-optimization: name: Run ML Pipeline & Sync to Hugging Face runs-on: ubuntu-latest timeout-minutes: 20 - + steps: - # ══════════════════════════════════════════════════════ - # 1. Setup Environment - # ══════════════════════════════════════════════════════ - name: 📥 Checkout code uses: actions/checkout@v4 with: fetch-depth: 0 - + - name: 🐍 Setup Python 3.11 uses: actions/setup-python@v5 with: python-version: '3.11' cache: 'pip' - + - name: 📦 Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt - pip install huggingFace_hub datasets pyarrow - - - # ══════════════════════════════════════════════════════ - # 2. Run Analytics Pipeline - # ══════════════════════════════════════════════════════ - - name: Execute Portfolio Optimization + + - name: Execute Daily Pipeline run: | - echo " Starting portfolio analytics pipeline..." - python daily_run.py - echo "✅ Pipeline completed successfully" + echo "Starting portfolio analytics pipeline..." + export PYTHONPATH=$PYTHONPATH:. + python src/pipeline/daily_run.py + echo "Pipeline completed successfully" env: PYTHONUNBUFFERED: 1 TZ: Europe/Paris - HF_TOKEN: ${{ secrets.HF_TOKEN }} - - - # ══════════════════════════════════════════════════════ - # 3. Summary jobs - # ══════════════════════════════════════════════════════ - - name: Generate job summary + HF_TOKEN: ${{ secrets.HF_TOKEN }} + MLFLOW_TRACKING_USERNAME: "SORADATA" + MLFLOW_TRACKING_PASSWORD: ${{ secrets.HF_TOKEN }} + + + - name: 📋 Generate job summary if: always() run: | - echo "## AlphaEdge Analytics Summary" >> $GITHUB_STEP_SUMMARY + echo "## AlphaEdge Daily Summary" >> $GITHUB_STEP_SUMMARY echo "- **Status**: ${{ job.status }}" >> $GITHUB_STEP_SUMMARY - echo "- **Data Provider**: Hugging Face (soradata/alphaedge-data)" >> $GITHUB_STEP_SUMMARY - echo "- **Market Timestamp**: $(date +'%Y-%m-%d %H:%M UTC')" >> $GITHUB_STEP_SUMMARY - \ No newline at end of file + echo "- **Run date**: $(date +'%Y-%m-%d %H:%M %Z')" >> $GITHUB_STEP_SUMMARY + echo "- **Market**: CAC40 & US_TECH" >> $GITHUB_STEP_SUMMARY + # Ajouter un lien vers le rapport si tu génères un artifact + echo "Check the logs above for detailed performance metrics." >> $GITHUB_STEP_SUMMARY \ No newline at end of file diff --git a/.github/workflows/ml_pipeline.yml b/.github/workflows/ml_pipeline.yml new file mode 100644 index 0000000..4023dad --- /dev/null +++ b/.github/workflows/ml_pipeline.yml @@ -0,0 +1,47 @@ +name: AlphaEdge Auto-Retrain Pipeline + +on: + schedule: + - cron: '0 20 * * 5' # Vendredi 20h UTC + workflow_dispatch: + +jobs: + train-and-log: + runs-on: ubuntu-latest + timeout-minutes: 45 + + steps: + - name: 📥 Checkout code + uses: actions/checkout@v4 + + - name: 🐍 Setup Python 3.11 + uses: actions/setup-python@v5 + with: + python-version: '3.11' + cache: 'pip' + + - name: 📦 Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + + - name: 🔄 Run ETL (Download & Prepare Data) + run: python src/pipeline/etl.py + env: + PYTHONPATH: . + + - name: 🤖 Retrain & Push to Hugging Face + run: python src/models/train.py + env: + PYTHONPATH: . + PYTHONUNBUFFERED: 1 + HF_TOKEN: ${{ secrets.HF_TOKEN }} + MLFLOW_TRACKING_USERNAME: "SORADATA" + MLFLOW_TRACKING_PASSWORD: ${{ secrets.HF_TOKEN }} + + - name: 📋 Generate job summary + if: always() + run: | + echo "## AlphaEdge Weekly Retrain" >> $GITHUB_STEP_SUMMARY + echo "- **Status**: ${{ job.status }}" >> $GITHUB_STEP_SUMMARY + echo "- **Run date**: $(date +'%Y-%m-%d %H:%M UTC')" >> $GITHUB_STEP_SUMMARY \ No newline at end of file diff --git a/__init__.py b/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/config/.gitkeep b/config/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/config/market_config.json b/config/market_config.json deleted file mode 100644 index 5db1ec3..0000000 --- a/config/market_config.json +++ /dev/null @@ -1,13 +0,0 @@ -{ - "market_name": "CAC 40 (France)", - "benchmark_ticker": "^FCHI", - "currency": "EUR", - "assets": [ - "AI.PA", "AIR.PA", "ALO.PA", "MT.AS", "ATO.PA", "CS.PA", "BNP.PA", - "EN.PA", "CAP.PA", "CA.PA", "DSY.PA", "EL.PA", "ENGI.PA", "ERF.PA", - "RMS.PA", "KER.PA", "OR.PA", "LR.PA", "MC.PA", "ML.PA", "ORA.PA", - "RI.PA", "PUB.PA", "RNO.PA", "SAF.PA", "SGO.PA", "SAN.PA", "SU.PA", - "GLE.PA", "STLAP.PA", "STMPA.PA", "TEP.PA", "HO.PA", "TTE.PA", - "URW.PA", "VIE.PA", "DG.PA", "VIV.PA", "WLN.PA", "FR.PA" - ] -} \ No newline at end of file diff --git a/config/markets/Nasdaq.json b/config/markets/Nasdaq.json new file mode 100644 index 0000000..3632551 --- /dev/null +++ b/config/markets/Nasdaq.json @@ -0,0 +1,5 @@ +{ + "market_name": "US_TECH", + "tickers": ["AAPL", "MSFT", "NVDA", "GOOGL"], + "ff_region": "F-F_Research_Data_5_Factors_2x3" +} \ No newline at end of file diff --git a/config/markets/cac40.json b/config/markets/cac40.json new file mode 100644 index 0000000..fb1f9da --- /dev/null +++ b/config/markets/cac40.json @@ -0,0 +1,12 @@ +{ + "market_name": "CAC40", + "tickers": [ + "AI.PA", "AIR.PA", "ALO.PA", "MT.AS", "ATO.PA", "CS.PA", "BNP.PA", + "EN.PA", "CAP.PA", "CA.PA", "DSY.PA", "EL.PA", "ENGI.PA", "ERF.PA", + "RMS.PA", "KER.PA", "OR.PA", "LR.PA", "MC.PA", "ML.PA", "ORA.PA", + "RI.PA", "PUB.PA", "RNO.PA", "SAF.PA", "SGO.PA", "SAN.PA", "SU.PA", + "GLE.PA", "STLAP.PA", "STMPA.PA", "TEP.PA", "HO.PA", "TTE.PA", + "URW.PA", "VIE.PA", "DG.PA", "VIV.PA", "WLN.PA", "FR.PA" + ], + "ff_region": "Europe_5_Factors" +} \ No newline at end of file diff --git a/const.py b/const.py index 55d4403..136b3ad 100644 --- a/const.py +++ b/const.py @@ -1,109 +1,110 @@ +# const.py + from pathlib import Path -# from pathlib import Path from dotenv import load_dotenv import os -import json - -# ============================================================================= -# PATHS -# ============================================================================= BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" -MODEL_DIR = BASE_DIR / 'src' / 'models' +MODEL_DIR = BASE_DIR / "src" / "models" LOG_DIR = BASE_DIR / "logs" -CONFIG_FILE = BASE_DIR / "config" / "market_config.json" +CONFIG_DIR = BASE_DIR / "config" load_dotenv(BASE_DIR / ".env") - -# Charge la config JSON pour récupérer huggingface.repo_id -with open(CONFIG_FILE, "r", encoding="utf-8") as f: - cfg = json.load(f) - -huggingface_cfg = cfg.get("huggingface", {}) - HF_TOKEN = os.getenv("HF_TOKEN") -REPO_ID = huggingface_cfg.get("repo_id", "soradata/alphaedge-data") - - -# ============================================================================= -# MARKET & RISK -# ============================================================================= TRADING_DAYS_YEAR: int = 252 RISK_FREE_RATE: float = 0.03 -# ============================================================================= -# ML SIGNAL -# ============================================================================= - -TARGET_CLUSTER: int = 3 -PROBA_THRESHOLD: float = 0.6 +TARGET_CLUSTER: int = 3 +PROBA_THRESHOLD: float = 0.51 +PROBA_MIN: float = 0.51 FEATURE_COLS: list[str] = [ - "rsi", - "macd", - "bb_low", - "bb_high", - "atr", - "return_2m", - "return_3m", - "return_6m", - "euro_volume_lag1", - "garman_klass_vol_lag1", - "Mkt-RF_lag1", - "SMB_lag1", - "HML_lag1", - "RMW_lag1", - "CMA_lag1", - "cluster", + "rsi_lag1", "macd_lag1", "bb_low_lag1", "bb_mid_lag1", + "bb_high_lag1", "atr_lag1", "cluster_lag1", + "return_1m", "return_2m", "return_3m", "return_6m", "return_9m", "return_12m", + "mom_12_1", "mom_6_1", + "realized_vol_3m", "realized_vol_12m", "vol_ratio", + "sharpe_3m", "sharpe_6m", "sortino_6m", + "return_skew_6m", "hist_var_5pct", "cvar_5pct", + "amihud_illiquidity", "volume_zscore", + "price_zscore_12", "nearness_52w_high", + "mom_12_1_rank", "sharpe_6m_rank", "realized_vol_3m_rank", "amihud_illiquidity_rank", + "Mkt-RF_lag1", "SMB_lag1", "HML_lag1", "RMW_lag1", "CMA_lag1", + "euro_volume_lag1", "garman_klass_vol_lag1", + "month_sin", "month_cos", "is_q_end", ] -# ============================================================================= -# TECHNICAL INDICATORS -# ============================================================================= - -RSI_WINDOW: int = 20 -BB_WINDOW: int = 20 -BB_STD: int = 2 -ATR_WINDOW: int = 14 -MACD_SLOW: int = 26 -MACD_FAST: int = 12 -MACD_SIGN: int = 9 - -# ============================================================================= -# FEATURE ENGINEERING -# ============================================================================= +RSI_WINDOW: int = 20 +BB_WINDOW: int = 20 +BB_STD: int = 2 +ATR_WINDOW: int = 14 +MACD_SLOW: int = 26 +MACD_FAST: int = 12 +MACD_SIGN: int = 9 -MIN_HISTORY_TA: int = 20 # minimum bars for RSI / Bollinger -MIN_HISTORY_FF: int = 24 # minimum bars for Fama-French rolling betas +MIN_HISTORY_TA: int = 20 +MIN_HISTORY_FF: int = 24 +WINSOR_CUTOFF: float = 0.005 MOMENTUM_LAGS: list[int] = [1, 2, 3, 6, 9, 12] -WINSOR_CUTOFF: float = 0.005 VARS_TO_LAG: list[str] = [ - "Mkt-RF", - "SMB", - "HML", - "RMW", - "CMA", + "Mkt-RF", "SMB", "HML", "RMW", "CMA", "euro_volume", "garman_klass_vol", ] FAMA_FRENCH_FACTORS: list[str] = ["Mkt-RF", "SMB", "HML", "RMW", "CMA"] -# ============================================================================= -# RESAMPLING -# ============================================================================= - RESAMPLE_MEAN_COLS: list[str] = ["euro_volume"] RESAMPLE_LAST_EXCLUDE: list[str] = [ - "euro_volume", - "volume", - "open", - "high", - "low", - "close", + "euro_volume", "volume", "open", "high", "low", "close", ] + +TRANSACTION_COST: float = 0.0010 +MIN_STOCKS_OPTIM: int = 3 +MAX_STOCKS_SELECT: int = 10 +WEIGHT_BOUNDS: tuple = (0.03, 0.20) + +SHARPE_THRESHOLD: float = 0.30 +MAX_DD_THRESHOLD: float = -0.40 + +BACKTEST_YEARS: int = 2 + +FEATURE_GROUPS = { + "momentum": [ + "return_1m", "return_2m", "return_3m", "return_6m", "return_9m", "return_12m", + "mom_12_1", "mom_6_1", "mom_3_1", "mom_12_1_rank", + ], + "volatility": [ + "realized_vol_3m", "realized_vol_12m", "vol_ratio", + "realized_vol_3m_rank", "garman_klass_vol_lag1", "idio_vol", + ], + "risk_adjusted": [ + "sharpe_3m", "sharpe_6m", "sortino_6m", "calmar_proxy", "sharpe_6m_rank", + ], + "tail_risk": [ + "return_skew_6m", "return_kurt_6m", "hist_var_5pct", "cvar_5pct", + ], + "technical": [ + "rsi_lag1", "macd_lag1", "bb_low_lag1", "bb_mid_lag1", + "bb_high_lag1", "atr_lag1", "cluster_lag1", + "bb_position", "rsi_divergence", "macd_sign", + ], + "liquidity": [ + "amihud_illiquidity", "volume_trend_3m", "volume_zscore", + "amihud_illiquidity_rank", "euro_volume_lag1", + ], + "mean_reversion": [ + "price_zscore_12", "nearness_52w_high", + ], + "macro": [ + "Mkt-RF_lag1", "SMB_lag1", "HML_lag1", "RMW_lag1", "CMA_lag1", + ], + "seasonality": [ + "month_sin", "month_cos", "is_q_end", "is_jan", + ], +} diff --git a/daily_run.py b/daily_run.py deleted file mode 100644 index bb97d0d..0000000 --- a/daily_run.py +++ /dev/null @@ -1,143 +0,0 @@ -import sys -import os -import json -import warnings -from datetime import datetime -from pathlib import Path - -from huggingface_hub import HfApi - -from const import ( - BASE_DIR, - TARGET_CLUSTER, - PROBA_THRESHOLD, - FEATURE_COLS -) -from src.utils.logger import setup_logger -from src.utils.config_loader import TICKERS, BENCHMARK_TICKER, MARKET_NAME -from src.utils.market_utils import build_export_df - -from src.pipeline.etl import get_data_pipeline, load_models -from src.pipeline.backtest import backtest_strategy_with_rebalancing, get_optimal_weights - - -# ============================================================================= -# INITIALIZATION -# ============================================================================= - -warnings.filterwarnings('ignore') -logger = setup_logger("DailyRun") - -HF_REPO_ID = os.getenv("HF_REPO_ID", "soradata/alphaedge-data") -HF_BRANCH = os.getenv("HF_DATA_BRANCH", "main") -HF_TOKEN = os.getenv("HF_TOKEN") -hf_api = HfApi() - - -# ============================================================================= -# HF UPLOAD HELPER -# ============================================================================= - -def upload_to_hf(local_path: Path, hf_filename: str): - """Upload un fichier local vers HuggingFace dans data/{HF_BRANCH}/""" - try: - hf_api.upload_file( - path_or_fileobj=str(local_path), - path_in_repo=f"data/{hf_filename}", - repo_id=HF_REPO_ID, - repo_type="dataset", - token=HF_TOKEN, - ) - logger.info(f"Uploaded to HF: data/{HF_BRANCH}/{hf_filename}") - except Exception as e: - logger.error(f"HF upload failed for {hf_filename}: {e}") - - -# ============================================================================= -# MAIN ORCHESTRATOR -# ============================================================================= - -def run_pipeline(): - start_time = datetime.now() - logger.info("-" * 60) - logger.info(f"STARTING DAILY PIPELINE | MARKET: {MARKET_NAME}") - logger.info(f"ASSETS: {len(TICKERS)} | BENCHMARK: {BENCHMARK_TICKER}") - logger.info(f"HF TARGET: {HF_REPO_ID}/data/{HF_BRANCH}/") - logger.info("-" * 60) - - try: - # 1. LOAD MODELS - xgb_model, kmeans_model = load_models() - if xgb_model is None: - raise RuntimeError("ML Models not found in src/models/") - - # 2. ETL & FEATURE ENGINEERING - df_daily, df_monthly = get_data_pipeline() - if df_daily is None or df_monthly is None: - raise RuntimeError("Data Pipeline failure.") - - # 3. GENERATE CURRENT SIGNALS - last_date = df_monthly.index.get_level_values('date').max() - logger.info(f"Generating signals for: {last_date.date()}") - - today_data = df_monthly.xs(last_date, level=0).copy() - today_data['cluster'] = kmeans_model.predict(today_data[['rsi']].fillna(50)) - today_data['proba_upside'] = xgb_model.predict_proba( - today_data[FEATURE_COLS].fillna(0))[:, 1] - selected = today_data[ - (today_data['cluster'] == TARGET_CLUSTER) & - (today_data['proba_upside'] > PROBA_THRESHOLD) - ] - final_alloc = {} - if not selected.empty: - tickers = selected.index.tolist() - prices_subset = df_daily['adj close'].unstack()[tickers].iloc[-252:].dropna(axis=1) - weights, success = get_optimal_weights(prices_subset) - final_alloc = weights if success else {t: 1.0/len(tickers) for t in tickers} - - # 4. EXPORT DAILY SIGNALS → HF - export_df = build_export_df(today_data, final_alloc) - signals_path = BASE_DIR / 'latest_signals.parquet' - export_df.to_parquet(signals_path, index=False) - upload_to_hf(signals_path, "latest_signals.parquet") - logger.info(f"Signals exported: {len(selected)} BUY signals.") - - # 5. BACKTESTING & MONITORING → HF - logger.info("Executing strategy backtest...") - hist_df, rebal_df = backtest_strategy_with_rebalancing( - df_daily, - df_monthly, - xgb_model, - kmeans_model, - get_optimal_weights - ) - hist_path = BASE_DIR / 'portfolio_history.parquet' - rebal_path = BASE_DIR / 'rebalance_history.parquet' - hist_df.to_parquet(hist_path) - rebal_df.to_parquet(rebal_path) - upload_to_hf(hist_path, "portfolio_history.parquet") - upload_to_hf(rebal_path, "rebalance_history.parquet") - - # 6. METADATA UPDATE → HF - metadata = { - 'market_name': MARKET_NAME, - 'last_update': datetime.now().isoformat(), - 'n_assets_tracked': len(TICKERS), - 'current_allocation': final_alloc, - 'hf_branch': HF_BRANCH, - } - meta_path = BASE_DIR / 'data_metadata.json' - with open(meta_path, 'w') as f: - json.dump(metadata, f, indent=4) - upload_to_hf(meta_path, "data_metadata.json") - - duration = (datetime.now() - start_time).total_seconds() - logger.info(f"PIPELINE COMPLETED SUCCESSFULLY in {duration:.1f}s") - - except Exception as e: - logger.critical(f"CRITICAL FAILURE: {e}", exc_info=True) - sys.exit(1) - - -if __name__ == "__main__": - run_pipeline() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 500f13e..44d2b9a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,40 +1,53 @@ # Data manipulation -pandas>=2.2.0,<3.0.0 -numpy>=1.26.0,<2.0.0 +pandas>=2.2.0,<3.0.0 +numpy>=1.26.0 + # Data collection yfinance>=0.2.32 -pandas-datareader==0.10.0 +pandas-datareader>=0.10.0 + # Machine Learning & Stats -scikit-learn>=1.3.2,<1.6.0 +scikit-learn>=1.5.0 xgboost>=2.0.2 statsmodels>=0.14.0 +optuna>=3.6.0 +lightgbm>=4.3.0 +mlflow>=2.13.0 +xgboost>=2.0.0 + # Portfolio optimization PyPortfolioOpt>=1.5.5 + # Visualization matplotlib>=3.8.2 seaborn>=0.13.0 plotly>=5.18.0 + # Technical indicators ta>=0.11.0 + # Utilities jupyter>=1.0.0 notebook>=7.0.6 tqdm>=4.66.1 + # Web App streamlit>=1.30.0 streamlit-autorefresh -# Hugging face + +# Hugging Face huggingface_hub>=0.20.0 datasets>=2.16.0 pyarrow>=14.0.0 -# environnement variables + +# Environment variables python-dotenv>=1.0.0 \ No newline at end of file diff --git a/src/extract/.gitkeep b/src/extract/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/src/extract/data_macro.py b/src/extract/data_macro.py deleted file mode 100644 index 79d512c..0000000 --- a/src/extract/data_macro.py +++ /dev/null @@ -1,24 +0,0 @@ -import pandas as pd -import yfinance as yf -from pathlib import Path - -PROJECT_ROOT = Path("/home/onyxia/work/Gestion-portefeuille/") - -# 1. TAUX BCE (Main Refinancing Operations) -print("📥 Téléchargement du taux BCE...") -# Le taux BCE n'est pas directement sur yfinance, on utilise l'Euribor 3M comme proxy -euribor = yf.download("EURIBOR3MD.HA", start="2015-01-01", end="2024-12-31") -euribor = euribor[['Close']].rename(columns={'Close': 'ECB_Rate'}) -euribor = euribor.reset_index() -euribor['Date'] = pd.to_datetime(euribor['Date']) -euribor.to_csv(PROJECT_ROOT / "data/raw/ECB_Rate.csv", index=False) -print(f"✅ Taux BCE sauvegardé : {len(euribor)} jours") - -# 2. V2X (Volatilité européenne) -print("\n📥 Téléchargement du V2X...") -v2x = yf.download("^V2X", start="2015-01-01", end="2024-12-31") -v2x = v2x[['Close']].rename(columns={'Close': 'V2X'}) -v2x = v2x.reset_index() -v2x['Date'] = pd.to_datetime(v2x['Date']) -v2x.to_csv(PROJECT_ROOT / "data/raw/V2X.csv", index=False) -print(f"✅ V2X sauvegardé : {len(v2x)} jours") diff --git a/src/extract/extractor.py b/src/extract/extractor.py new file mode 100644 index 0000000..b716352 --- /dev/null +++ b/src/extract/extractor.py @@ -0,0 +1,239 @@ +""" +MarketExtractor +================ +Gère le téléchargement des données boursières brutes via yfinance. + +Stratégie : + - Premier run : téléchargement complet (history_years) + - Runs suivants : delta uniquement (depuis last_date - 5 jours) + - Retry automatique (3 tentatives, 5s entre chaque) + - Validation des données avant sauvegarde +""" + +import time +from datetime import datetime +from pathlib import Path +from typing import Optional + +import pandas as pd +import yfinance as yf + +from const import DATA_DIR +from src.utils.logger import setup_logger + +logger = setup_logger("extractor") + +_DOWNLOAD_RETRIES = 3 +_DOWNLOAD_RETRY_WAIT = 5 +_DELTA_OVERLAP_DAYS = 5 + + +class MarketExtractor: + """ + Gère uniquement le téléchargement des données boursières brutes. + + Parameters + ---------- + market_name : str — nom du marché (ex: 'CAC40') + tickers : list[str] — liste des tickers yfinance + history_years : int — années d'historique pour le premier téléchargement + """ + + def __init__( + self, + market_name: str, + tickers: list[str], + history_years: int = 10, + ): + self.market_name = market_name + self.tickers = tickers + self.history_years = history_years + + self.raw_data_path = DATA_DIR / "raw" / self.market_name + self.raw_data_path.mkdir(parents=True, exist_ok=True) + self.file_path = self.raw_data_path / f"{self.market_name}_raw.csv" + + self.data = pd.DataFrame() + logger.info( + f"Extracteur initialisé : {self.market_name} ({len(tickers)} tickers)" + ) + + # Helpers privés + + def _load_existing(self) -> Optional[pd.DataFrame]: + """Charge le CSV existant avec typage strict.""" + if not self.file_path.exists(): + return None + try: + df = pd.read_csv( + self.file_path, + sep=";", + index_col=["date", "ticker"], + parse_dates=True, + date_format="%Y-%m-%d", + ) + return df + except Exception as e: + logger.warning(f"Impossible de lire le fichier existant ({e}). Full download.") + return None + + def _validate_download(self, df: pd.DataFrame) -> bool: + """ + Vérifie que le téléchargement est utilisable. + + Critères : + - Non vide + - Au moins 1 ticker valide (non entièrement NaN) + - Colonne 'adj close' présente + """ + if df.empty: + logger.warning("DataFrame téléchargé est vide.") + return False + + if "adj close" not in df.columns: + logger.warning( + f"Colonne 'adj close' absente. Colonnes disponibles : {df.columns.tolist()}" + ) + return False + + # Vérifier qu'au moins un ticker a des données réelles + valid_tickers = ( + df["adj close"] + .unstack("ticker") + .dropna(how="all", axis=1) + .columns + .tolist() + ) + if not valid_tickers: + logger.warning("Tous les tickers sont entièrement NaN.") + return False + + n_expected = len(self.tickers) + n_valid = len(valid_tickers) + if n_valid < n_expected: + logger.warning( + f" Only {n_valid}/{n_expected} tickers avec des données valides." + ) + + return True + + def _parse_yfinance(self, df: pd.DataFrame) -> pd.DataFrame: + """ + Normalise le DataFrame yfinance vers le format (date, ticker) MultiIndex. + + Gère les deux formats yfinance : + - MultiIndex colonnes (Price, Ticker) → stack level=1 + - Index simple (un seul ticker) → ajout colonne ticker + """ + if isinstance(df.columns, pd.MultiIndex): + df = df.stack(level=1, future_stack=True) + elif len(self.tickers) == 1: + df["ticker"] = self.tickers[0] + df = df.reset_index().set_index(["Date", "ticker"]) + + df.index.names = ["date", "ticker"] + df.columns = df.columns.str.lower() + + return df + + def _merge_with_existing( + self, + existing_df: pd.DataFrame, + new_df: pd.DataFrame, + ) -> pd.DataFrame: + """ + Fusionne les données existantes avec le delta téléchargé. + En cas de doublon sur (date, ticker), garde la valeur la plus récente (keep='last'). + """ + merged = pd.concat([existing_df, new_df]) + merged = merged[~merged.index.duplicated(keep="last")].sort_index() + return merged + + # Interface publique + + def fetch_market_data(self) -> Optional[pd.DataFrame]: + """ + Télécharge les données brutes. + - Delta si fichier existant (depuis last_date - 5 jours) + - Téléchargement complet sinon + + Returns + ------- + pd.DataFrame avec MultiIndex (date, ticker) ou None si échec. + """ + existing_df = self._load_existing() + + if existing_df is not None: + last_date = existing_df.index.get_level_values("date").max() + start_date = (last_date - pd.Timedelta(days=_DELTA_OVERLAP_DAYS)).strftime("%Y-%m-%d") + logger.info(f" [{self.market_name}] Mise à jour depuis le {last_date.date()}...") + else: + start_date = ( + datetime.today() - pd.DateOffset(days=365 * self.history_years) + ).strftime("%Y-%m-%d") + logger.info(f" [{self.market_name}] Premier téléchargement complet...") + + end_date = (datetime.today() + pd.DateOffset(days=1)).strftime("%Y-%m-%d") + + for attempt in range(1, _DOWNLOAD_RETRIES + 1): + try: + raw = yf.download( + self.tickers, + start=start_date, + end=end_date, + auto_adjust=False, # garde Adj Close séparé de Close + progress=False, + threads=True, + ) + + if raw.empty: + logger.warning( + f"Réponse vide de YFinance (tentative {attempt}/{_DOWNLOAD_RETRIES})" + ) + time.sleep(_DOWNLOAD_RETRY_WAIT) + continue + + # Normalisation du format yfinance + df = self._parse_yfinance(raw) + + # Validation avant fusion + if not self._validate_download(df): + logger.warning(f"Validation échouée (tentative {attempt}/{_DOWNLOAD_RETRIES})") + time.sleep(_DOWNLOAD_RETRY_WAIT) + continue + + # Fusion avec l'historique existant + if existing_df is not None: + self.data = self._merge_with_existing(existing_df, df) + else: + self.data = df + + # Sauvegarde + self.data.to_csv(self.file_path, sep=";") + logger.info( + f"[{self.market_name}] Base brute enregistrée : " + f"{self.data.shape[0]} lignes, " + f"{self.data.index.get_level_values('ticker').nunique()} tickers." + ) + + return self.data + + except Exception as exc: + logger.warning( + f"Erreur de téléchargement (tentative {attempt}/{_DOWNLOAD_RETRIES}): {exc}" + ) + if attempt < _DOWNLOAD_RETRIES: + time.sleep(_DOWNLOAD_RETRY_WAIT) + + logger.error( + f" Échec du téléchargement pour {self.market_name} " + f"après {_DOWNLOAD_RETRIES} tentatives." + ) + return None + + def __repr__(self) -> str: + n_rows = len(self.data) if not self.data.empty else 0 + return ( + f"MarketExtractor(market={self.market_name}, " + f"tickers={len(self.tickers)}, rows={n_rows})" + ) diff --git a/src/extract/yfinance_downloader.py b/src/extract/yfinance_downloader_test.py similarity index 100% rename from src/extract/yfinance_downloader.py rename to src/extract/yfinance_downloader_test.py diff --git a/src/features/alpha_features.py b/src/features/alpha_features.py new file mode 100644 index 0000000..6cb6d73 --- /dev/null +++ b/src/features/alpha_features.py @@ -0,0 +1,264 @@ +""" +Alpha Features - Production Grade +Sépare formellement les indicateurs journaliers (TA) de l'enrichissement mensuel (Alpha). +""" + +import numpy as np +import pandas as pd +import statsmodels.api as sm +from statsmodels.regression.rolling import RollingOLS +from ta.momentum import RSIIndicator +from ta.volatility import BollingerBands + +from const import ( + RSI_WINDOW, BB_WINDOW, BB_STD, + MIN_HISTORY_TA, MIN_HISTORY_FF, + MOMENTUM_LAGS, WINSOR_CUTOFF, + FAMA_FRENCH_FACTORS, +) +from src.utils.feature_utils import compute_atr, compute_macd +from src.utils.logger import setup_logger + +logger = setup_logger("alpha_features") + +# ══════════════════════════════════════════════════════════════════ +# UTILITAIRES +# ══════════════════════════════════════════════════════════════════ + + +def _safe_div(a: pd.Series, b: pd.Series) -> pd.Series: + return a.div(b.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) + + +def _rolling_sortino(returns: pd.Series, window: int = 6) -> pd.Series: + def _sortino_scalar(r: np.ndarray) -> float: + neg = r[r < 0] + if len(neg) == 0 or np.std(neg) == 0: + return np.nan + return (np.mean(r) / np.std(neg)) * np.sqrt(12) + return returns.rolling(window, min_periods=window // 2).apply(_sortino_scalar, raw=True) + + +def _rolling_maxdrawdown(returns: pd.Series, window: int = 12) -> pd.Series: + def _mdd(r: np.ndarray) -> float: + cumulative = np.cumprod(1 + r) + peak = np.maximum.accumulate(cumulative) + return ((cumulative - peak) / peak).min() + return returns.rolling(window, min_periods=window // 2).apply(_mdd, raw=True) + + +def add_rank_features(df: pd.DataFrame) -> pd.DataFrame: + features_to_rank = [ + "mom_12_1", "mom_6_1", "sharpe_6m", "sortino_6m", + "realized_vol_3m", "realized_vol_12m", + "amihud_illiquidity", "return_skew_6m", "hist_var_5pct", + ] + for feat in features_to_rank: + rank_col = f"{feat}_rank" + if feat in df.columns: + df[rank_col] = df.groupby(level="date")[feat].transform(lambda x: x.rank(pct=True)) + else: + df[rank_col] = 0.5 + return df.fillna({f"{f}_rank": 0.5 for f in features_to_rank}) + +# ══════════════════════════════════════════════════════════════════ +# 1. ÉTAPE JOURNALIÈRE (Appelé AVANT agrégation) +# ══════════════════════════════════════════════════════════════════ + + +def compute_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: + """Calcule les indicateurs TA stricts sur les données journalières.""" + logger.info("Computing daily technical indicators...") + + # Garman-Klass Volatility avec OHLC journalier + if all(col in df.columns for col in ["high", "low", "open", "adj close"]): + df["garman_klass_vol"] = ( + (np.log(df["high"]) - np.log(df["low"])) ** 2 / 2 + - (2 * np.log(2) - 1) * (np.log(df["adj close"]) - np.log(df["open"])) ** 2 + ) + else: + df["garman_klass_vol"] = 0.0 + + for ticker in df.index.get_level_values(1).unique(): + idx = (slice(None), ticker) + close = df.loc[idx, "adj close"] + if len(close) > MIN_HISTORY_TA: + df.loc[idx, "rsi"] = RSIIndicator(close=close, window=RSI_WINDOW).rsi().values + bb = BollingerBands(close=np.log1p(close), window=BB_WINDOW, window_dev=BB_STD) + df.loc[idx, "bb_low"] = bb.bollinger_lband().values + df.loc[idx, "bb_mid"] = bb.bollinger_mavg().values + df.loc[idx, "bb_high"] = bb.bollinger_hband().values + df.loc[idx, "bb_position"] = (np.log1p(close) - bb.bollinger_lband()) / (bb.bollinger_hband() - bb.bollinger_lband() + 1e-9) + + df["atr"] = df.groupby(level=1, group_keys=False).apply(compute_atr) + df["macd"] = df.groupby(level=1, group_keys=False).apply(compute_macd) + df["macd_sign"] = np.sign(df["macd"].fillna(0)) + + if "volume" in df.columns: + df["euro_volume"] = (df["adj close"] * df["volume"]) / 1e6 + else: + df["euro_volume"] = 0.0 + + return df + +# ══════════════════════════════════════════════════════════════════ +# 2. ÉTAPE MENSUELLE (Appelé APRÈS agrégation) +# ══════════════════════════════════════════════════════════════════ + + +def _add_momentum_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["return_1m"] = g["adj close"].transform(lambda x: x.pct_change(1)) + for lag in [2, 3, 6, 9, 12]: + df[f"return_{lag}m"] = g["adj close"].transform(lambda x: x.pct_change(lag)) + + pct_12 = g["adj close"].transform(lambda x: x.pct_change(12)) + pct_1 = g["adj close"].transform(lambda x: x.pct_change(1)) + pct_6 = g["adj close"].transform(lambda x: x.pct_change(6)) + + df["mom_12_1"] = pct_12.div(pct_1.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) + df["mom_6_1"] = pct_6.div(pct_1.replace(0, np.nan)).replace([np.inf, -np.inf], np.nan) + df["mom_3_1"] = g["adj close"].transform(lambda x: x.pct_change(3)) + return df + + +def calculate_returns(df: pd.DataFrame) -> pd.DataFrame: + """Wrapper pour processor.py (mensuel).""" + return _add_momentum_factors(df, df.groupby(level="ticker")) + + +def _add_mean_reversion_factors(df: pd.DataFrame, g) -> pd.DataFrame: + ma12 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).mean()) + std12 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).std()) + df["price_zscore_12"] = _safe_div(df["adj close"] - ma12, std12) + high52 = g["adj close"].transform(lambda x: x.rolling(12, min_periods=6).max()) + df["nearness_52w_high"] = _safe_div(df["adj close"], high52) + return df + + +def _add_volatility_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["realized_vol_3m"] = g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).std() * np.sqrt(12)) + df["realized_vol_12m"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).std() * np.sqrt(12)) + df["vol_ratio"] = df["realized_vol_3m"].div(df["realized_vol_12m"]).replace([np.inf, -np.inf], np.nan) + + excess = df["return_1m"] - df["Mkt-RF"].fillna(0) if "Mkt-RF" in df.columns else df["return_1m"] + df["idio_vol"] = excess.groupby(level=1).transform(lambda x: x.rolling(6, min_periods=3).std() * np.sqrt(12)) + return df + + +def _add_risk_adjusted_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["sharpe_3m"] = _safe_div( + g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).mean()), + g["return_1m"].transform(lambda x: x.rolling(3, min_periods=2).std()), + ) * np.sqrt(12) + df["sharpe_6m"] = _safe_div( + g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).mean()), + g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).std()), + ) * np.sqrt(12) + df["sortino_6m"] = g["return_1m"].transform(lambda x: _rolling_sortino(x, window=6)) + df["calmar_proxy"] = _safe_div( + g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).mean() * 12), + g["return_1m"].transform(lambda x: _rolling_maxdrawdown(x, window=12)).abs(), + ) + return df + + +def _add_tail_risk_factors(df: pd.DataFrame, g) -> pd.DataFrame: + df["return_skew_6m"] = g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).skew()) + df["return_kurt_6m"] = g["return_1m"].transform(lambda x: x.rolling(6, min_periods=3).kurt()) + df["hist_var_5pct"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).quantile(0.05)) + + def _cvar(r: np.ndarray) -> float: + t = np.quantile(r, 0.05) + tail = r[r <= t] + return tail.mean() if len(tail) > 0 else np.nan + + df["cvar_5pct"] = g["return_1m"].transform(lambda x: x.rolling(12, min_periods=6).apply(_cvar, raw=True)) + return df + + +def _add_technical_enrichment(df: pd.DataFrame, g) -> pd.DataFrame: + """Enrichit les données techniques sur la base mensuelle.""" + if "rsi" in df.columns: + df["rsi_divergence"] = g["adj close"].transform(lambda x: x.pct_change(3)) - g["rsi"].transform(lambda x: x.pct_change(3)) + + if "euro_volume" in df.columns: + df["amihud_illiquidity"] = _safe_div(df["return_1m"].abs(), df["euro_volume"]) + df["volume_trend_3m"] = g["euro_volume"].transform(lambda x: x.pct_change(3)) + df["volume_zscore"] = g["euro_volume"].transform(lambda x: _safe_div(x - x.rolling(12, min_periods=6).mean(), x.rolling(12, min_periods=6).std())) + else: + for col in ["amihud_illiquidity", "volume_trend_3m", "volume_zscore"]: + df[col] = 0.0 + return df + + +def _add_seasonality_features(df: pd.DataFrame) -> pd.DataFrame: + dates = df.index.get_level_values("date") + df["month_sin"] = np.sin(2 * np.pi * dates.month / 12) + df["month_cos"] = np.cos(2 * np.pi * dates.month / 12) + df["is_q_end"] = dates.month.isin([3, 6, 9, 12]).astype(int) + df["is_jan"] = (dates.month == 1).astype(int) + return df + + +def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: + """Récupère les facteurs Fama-French.""" + logger.info("Retrieving Fama-French factors...") + try: + import pandas_datareader.data as web + factor_data = web.DataReader("Europe_5_Factors", "famafrench", start="2010")[0].drop("RF", axis=1) + factor_data.index = pd.to_datetime(factor_data.index.to_timestamp()).tz_localize(None) + factor_data = factor_data.resample("BME").last().div(100) + factor_data.index.name = "date" + + if "return_1m" not in data.columns: + return data + + betas_list = [] + for ticker in data.index.get_level_values(1).unique(): + ticker_data = data.xs(ticker, level=1) + y = ticker_data["return_1m"].dropna() + if y.empty: continue + X = factor_data.loc[factor_data.index.intersection(y.index)] + y = y.loc[X.index] + if len(y) <= MIN_HISTORY_FF: continue + + params = RollingOLS(y, sm.add_constant(X[FAMA_FRENCH_FACTORS]), window=MIN_HISTORY_FF).fit().params.drop("const", axis=1) + params["ticker"] = ticker + betas_list.append(params) + + if betas_list: + betas_df = pd.concat(betas_list).set_index("ticker", append=True) + data = data.join(betas_df.groupby("ticker").shift()) + data[FAMA_FRENCH_FACTORS] = data.groupby(level="ticker", group_keys=False)[FAMA_FRENCH_FACTORS].transform(lambda x: x.fillna(x.mean())) + return data + except Exception as exc: + logger.warning(f"Fama-French retrieval failed ({exc}).") + + return data.assign(**{f: 0.0 for f in FAMA_FRENCH_FACTORS}) + + +def add_all_features(df: pd.DataFrame) -> pd.DataFrame: + """Master pipeline exécuté sur les données mensuelles.""" + if not isinstance(df.index, pd.MultiIndex): raise ValueError("MultiIndex requis.") + df = df.copy() + g = df.groupby(level="ticker") + + logger.info("Computing alpha features...") + df = _add_momentum_factors(df, g) + df = _add_mean_reversion_factors(df, g) + df = _add_volatility_factors(df, g) + df = _add_risk_adjusted_factors(df, g) + df = _add_tail_risk_factors(df, g) + df = _add_technical_enrichment(df, g) + df = _add_seasonality_features(df) + df = add_rank_features(df) + + cols_to_lag = ["rsi", "macd", "bb_low", "bb_mid", "bb_high", "atr", "garman_klass_vol", + "bb_position", "macd_sign", "Mkt-RF", "SMB", "HML", "RMW", "CMA"] + for col in cols_to_lag: + if col in df.columns: + df[f"{col}_lag1"] = df.groupby(level="ticker")[col].shift(1) + + df = df.fillna(0).replace([np.inf, -np.inf], 0) + logger.info(f" Features prêtes. Shape : {df.shape}") + return df diff --git a/src/models/.gitkeep b/src/models/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/src/models/CAC40/ensemble_model.pkl b/src/models/CAC40/ensemble_model.pkl new file mode 100644 index 0000000..ce34bf5 Binary files /dev/null and b/src/models/CAC40/ensemble_model.pkl differ diff --git a/src/models/CAC40/model_card.json b/src/models/CAC40/model_card.json new file mode 100644 index 0000000..ea8f6af --- /dev/null +++ b/src/models/CAC40/model_card.json @@ -0,0 +1,58 @@ +{ + "market": "CAC40", + "trained_at": "2026-06-27T17:51:54.550863", + "architecture": "XGBoost + LightGBM + Ridge \u2192 LogisticRegression", + "metrics_ml": { + "auc_test": 0.5517, + "apr_test": 0.6124, + "baseline_auc": 0.5, + "lift": 0.0517 + }, + "metrics_fin": { + "sharpe": 1.2591, + "max_drawdown": -0.0564, + "total_return": 0.1039 + }, + "model_weights": { + "xgb": 0.0681, + "lgb": 0.042, + "ridge": 0.8899 + }, + "walk_forward": [ + { + "window": 1, + "test_start": "2026-04-30", + "test_end": "2026-06-30", + "auc": 0.4362, + "apr": 0.633, + "n_test": 119 + }, + { + "window": 2, + "test_start": "2026-01-30", + "test_end": "2026-03-31", + "auc": 0.6503, + "apr": 0.5659, + "n_test": 120 + }, + { + "window": 3, + "test_start": "2025-10-31", + "test_end": "2025-12-31", + "auc": 0.5131, + "apr": 0.5949, + "n_test": 120 + }, + { + "window": 4, + "test_start": "2025-07-31", + "test_end": "2025-09-30", + "auc": 0.507, + "apr": 0.541, + "n_test": 120 + } + ], + "n_features": 50, + "train_size": 4440, + "test_size": 279 +} \ No newline at end of file diff --git a/src/models/US_TECH/ensemble_model.pkl b/src/models/US_TECH/ensemble_model.pkl new file mode 100644 index 0000000..2747f80 Binary files /dev/null and b/src/models/US_TECH/ensemble_model.pkl differ diff --git a/src/models/US_TECH/model_card.json b/src/models/US_TECH/model_card.json new file mode 100644 index 0000000..64e12f7 --- /dev/null +++ b/src/models/US_TECH/model_card.json @@ -0,0 +1,58 @@ +{ + "market": "US_TECH", + "trained_at": "2026-06-27T17:48:58.336014", + "architecture": "XGBoost + LightGBM + Ridge \u2192 LogisticRegression", + "metrics_ml": { + "auc_test": 0.8182, + "apr_test": 0.8102, + "baseline_auc": 0.5, + "lift": 0.3182 + }, + "metrics_fin": { + "sharpe": 0.6194, + "max_drawdown": -0.0876, + "total_return": 0.0847 + }, + "model_weights": { + "xgb": 0.1367, + "lgb": 0.1504, + "ridge": 0.7129 + }, + "walk_forward": [ + { + "window": 1, + "test_start": "2026-04-30", + "test_end": "2026-06-30", + "auc": 0.6667, + "apr": 0.7052, + "n_test": 11 + }, + { + "window": 2, + "test_start": "2026-01-30", + "test_end": "2026-03-31", + "auc": 0.75, + "apr": 0.6429, + "n_test": 12 + }, + { + "window": 3, + "test_start": "2025-10-31", + "test_end": "2025-12-31", + "auc": 0.6857, + "apr": 0.6393, + "n_test": 12 + }, + { + "window": 4, + "test_start": "2025-07-31", + "test_end": "2025-09-30", + "auc": 0.95, + "apr": 0.9909, + "n_test": 12 + } + ], + "n_features": 50, + "train_size": 453, + "test_size": 27 +} \ No newline at end of file diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..098de57 --- /dev/null +++ b/src/models/__init__.py @@ -0,0 +1,4 @@ +from src.models.ensemble import AlphaEdgeEnsemble, FEATURE_GROUPS +from src.models.cv import PurgedTimeSeriesSplit + +__all__ = ["AlphaEdgeEnsemble", "FEATURE_GROUPS", "PurgedTimeSeriesSplit"] \ No newline at end of file diff --git a/src/models/cv.py b/src/models/cv.py new file mode 100644 index 0000000..973a08a --- /dev/null +++ b/src/models/cv.py @@ -0,0 +1,47 @@ +""" +PurgedTimeSeriesSplit — López de Prado (2018), Advances in Financial ML. +Évite le leakage entre folds adjacents via purge + embargo. +""" + +import numpy as np +from sklearn.model_selection import BaseCrossValidator + + +class PurgedTimeSeriesSplit(BaseCrossValidator): + """ + Walk-forward CV avec purge et embargo pour séries financières. + + Purge : retire du train les observations dont le label chevauche la fenêtre test. + Embargo: retire les N observations immédiatement après le test (autocorrélation résiduelle). + + Parameters + ---------- + n_splits : nombre de folds + embargo_pct : fraction des données utilisée comme embargo après chaque fold test + """ + + def __init__(self, n_splits: int = 5, embargo_pct: float = 0.01): + self.n_splits = n_splits + self.embargo_pct = embargo_pct + + def split(self, X, y=None, groups=None): + n = len(X) + fold_size = n // (self.n_splits + 1) + embargo = int(n * self.embargo_pct) + + for i in range(1, self.n_splits + 1): + test_start = i * fold_size + test_end = test_start + fold_size + purge_start = max(0, test_start - embargo) + train_idx = np.concatenate([ + np.arange(0, purge_start), + np.arange(min(test_end + embargo, n), n), + ]) + test_idx = np.arange(test_start, min(test_end, n)) + + if len(train_idx) == 0 or len(test_idx) == 0: + continue + yield train_idx, test_idx + + def get_n_splits(self, X=None, y=None, groups=None): + return self.n_splits diff --git a/src/models/ensemble.py b/src/models/ensemble.py new file mode 100644 index 0000000..cf1df63 --- /dev/null +++ b/src/models/ensemble.py @@ -0,0 +1,253 @@ +""" +AlphaEdgeEnsemble +================== +XGBoost + LightGBM + Ridge → LogisticRegression (stacking). +Optimisation Bayésienne (Optuna) par modèle. +Interface sklearn : fit / predict_proba / get_model_weights. +""" + +import warnings +import numpy as np +import pandas as pd +import optuna +import pickle +from typing import Dict, List, Optional + +import xgboost as xgb +import lightgbm as lgb +from sklearn.linear_model import LogisticRegression, RidgeClassifier +from sklearn.calibration import CalibratedClassifierCV +from sklearn.preprocessing import StandardScaler +from sklearn.base import BaseEstimator, ClassifierMixin + +from src.models.cv import PurgedTimeSeriesSplit +from src.utils.logger import setup_logger +from const import FEATURE_GROUPS + +optuna.logging.set_verbosity(optuna.logging.WARNING) +warnings.filterwarnings("ignore") +logger = setup_logger("ensemble") + + +def _all_features() -> List[str]: + seen, result = set(), [] + for feats in FEATURE_GROUPS.values(): + for f in feats: + if f not in seen: + seen.add(f) + result.append(f) + return result + + +def _available(df: pd.DataFrame) -> List[str]: + all_f = _all_features() + return [f for f in all_f if f in df.columns] + + +def _prepare_X(df: pd.DataFrame, features: List[str]) -> pd.DataFrame: + return df[features].fillna(0).replace([np.inf, -np.inf], 0) + + +# ══════════════════════════════════════════════════════════════════ +# OPTUNA OBJECTIVES +# ══════════════════════════════════════════════════════════════════ + +def _xgb_objective(X: pd.DataFrame, y: pd.Series, n_trials: int) -> xgb.XGBClassifier: + from sklearn.metrics import roc_auc_score + cv = PurgedTimeSeriesSplit(n_splits=5) + + def objective(trial): + params = { + "n_estimators": trial.suggest_int("n_estimators", 100, 500, step=50), + "max_depth": trial.suggest_int("max_depth", 3, 6), + "learning_rate": trial.suggest_float("learning_rate", 0.01, 0.15, log=True), + "subsample": trial.suggest_float("subsample", 0.6, 1.0), + "colsample_bytree": trial.suggest_float("colsample_bytree", 0.5, 1.0), + "min_child_weight": trial.suggest_int("min_child_weight", 1, 10), + "reg_alpha": trial.suggest_float("reg_alpha", 1e-3, 10.0, log=True), + "reg_lambda": trial.suggest_float("reg_lambda", 1e-3, 10.0, log=True), + "gamma": trial.suggest_float("gamma", 0.0, 1.0), + "eval_metric": "auc", "random_state": 42, "n_jobs": -1, + } + scores = [] + for tr_idx, val_idx in cv.split(X): + m = xgb.XGBClassifier(**params) + m.fit(X.iloc[tr_idx], y.iloc[tr_idx], verbose=False) + p = m.predict_proba(X.iloc[val_idx])[:, 1] + if len(np.unique(y.iloc[val_idx])) > 1: + scores.append(roc_auc_score(y.iloc[val_idx], p)) + return np.mean(scores) if scores else 0.0 + + study = optuna.create_study( + direction="maximize", + sampler=optuna.samplers.TPESampler(seed=42), + pruner=optuna.pruners.MedianPruner(n_startup_trials=10), + ) + study.optimize(objective, n_trials=n_trials, show_progress_bar=False) + + best = xgb.XGBClassifier(**study.best_params, eval_metric="auc", random_state=42, n_jobs=-1) + best.fit(X, y, verbose=False) + logger.info(f" XGB best AUC (CV): {study.best_value:.4f} | params: {study.best_params}") + return best + + +def _lgb_objective(X: pd.DataFrame, y: pd.Series, n_trials: int) -> lgb.LGBMClassifier: + from sklearn.metrics import roc_auc_score + cv = PurgedTimeSeriesSplit(n_splits=5) + + def objective(trial): + params = { + "n_estimators": trial.suggest_int("n_estimators", 100, 500, step=50), + "max_depth": trial.suggest_int("max_depth", 3, 7), + "learning_rate": trial.suggest_float("learning_rate", 0.01, 0.15, log=True), + "subsample": trial.suggest_float("subsample", 0.6, 1.0), + "colsample_bytree": trial.suggest_float("colsample_bytree", 0.5, 1.0), + "min_child_samples":trial.suggest_int("min_child_samples", 5, 50), + "reg_alpha": trial.suggest_float("reg_alpha", 1e-3, 10.0, log=True), + "reg_lambda": trial.suggest_float("reg_lambda", 1e-3, 10.0, log=True), + "random_state": 42, "n_jobs": -1, "verbose": -1, + } + scores = [] + for tr_idx, val_idx in cv.split(X): + m = lgb.LGBMClassifier(**params) + m.fit(X.iloc[tr_idx], y.iloc[tr_idx]) + p = m.predict_proba(X.iloc[val_idx])[:, 1] + if len(np.unique(y.iloc[val_idx])) > 1: + scores.append(roc_auc_score(y.iloc[val_idx], p)) + return np.mean(scores) if scores else 0.0 + + study = optuna.create_study( + direction="maximize", + sampler=optuna.samplers.TPESampler(seed=0), + pruner=optuna.pruners.MedianPruner(n_startup_trials=10), + ) + study.optimize(objective, n_trials=n_trials, show_progress_bar=False) + + best = lgb.LGBMClassifier(**study.best_params, random_state=42, n_jobs=-1, verbose=-1) + best.fit(X, y) + logger.info(f" LGB best AUC (CV): {study.best_value:.4f} | params: {study.best_params}") + return best + + +# ══════════════════════════════════════════════════════════════════ +# ALPHAEDGE ENSEMBLE +# ══════════════════════════════════════════════════════════════════ + +class AlphaEdgeEnsemble(BaseEstimator, ClassifierMixin): + """ + Ensemble à 2 niveaux : + Niveau 0 : XGBoost + LightGBM + Ridge (calibré) + Niveau 1 : LogisticRegression (méta-learner) + + Le méta-learner est entraîné sur les probabilités out-of-fold + des modèles de base via PurgedTimeSeriesSplit. + + Parameters + ---------- + n_optuna_trials : int — nombre d'essais Optuna par modèle de base + """ + + def __init__(self, n_optuna_trials: int = 50): + self.n_optuna_trials = n_optuna_trials + self.xgb_model_: Optional[xgb.XGBClassifier] = None + self.lgb_model_: Optional[lgb.LGBMClassifier] = None + self.ridge_model_: Optional[CalibratedClassifierCV] = None + self.meta_model_: Optional[LogisticRegression] = None + self.scaler_: Optional[StandardScaler] = None + self.features_: Optional[List[str]] = None + self.classes_ = np.array([0, 1]) + + def _oof_probas( + self, + X: pd.DataFrame, + y: pd.Series, + model_cls, + fit_kwargs: dict, + ) -> np.ndarray: + """Génère les probabilités out-of-fold pour le méta-learner.""" + oof = np.zeros(len(X)) + cv = PurgedTimeSeriesSplit(n_splits=5) + + for tr_idx, val_idx in cv.split(X): + m = model_cls(**fit_kwargs) + m.fit(X.iloc[tr_idx], y.iloc[tr_idx]) + oof[val_idx] = m.predict_proba(X.iloc[val_idx])[:, 1] + + return oof + + def fit(self, df: pd.DataFrame, y: pd.Series) -> "AlphaEdgeEnsemble": + logger.info("Training AlphaEdgeEnsemble...") + + self.features_ = _available(df) + if not self.features_: + raise ValueError("Aucune feature disponible dans df.") + + X = _prepare_X(df, self.features_) + + trials_each = max(10, self.n_optuna_trials // 2) + + logger.info(" [1/3] XGBoost...") + self.xgb_model_ = _xgb_objective(X, y, trials_each) + + logger.info(" [2/3] LightGBM...") + self.lgb_model_ = _lgb_objective(X, y, trials_each) + + logger.info(" [3/3] Ridge (calibré)...") + self.scaler_ = StandardScaler() + X_scaled = pd.DataFrame(self.scaler_.fit_transform(X), columns=X.columns, index=X.index) + ridge_base = RidgeClassifier(alpha=1.0, random_state=42) + self.ridge_model_ = CalibratedClassifierCV(ridge_base, cv=5, method="sigmoid") + self.ridge_model_.fit(X_scaled, y) + + # ── Niveau 1 : méta-learner sur probas OOF + logger.info(" [Meta] Stacking LogisticRegression...") + oof_xgb = self._oof_probas(X, y, xgb.XGBClassifier, + {**self.xgb_model_.get_params(), "eval_metric": "auc"}) + oof_lgb = self._oof_probas(X, y, lgb.LGBMClassifier, + {**self.lgb_model_.get_params(), "verbose": -1}) + oof_ridge = self.ridge_model_.predict_proba(X_scaled)[:, 1] + + meta_X = np.column_stack([oof_xgb, oof_lgb, oof_ridge]) + self.meta_model_ = LogisticRegression(C=1.0, random_state=42, max_iter=500) + self.meta_model_.fit(meta_X, y) + + weights = self.get_model_weights() + logger.info( + f" Weights → XGB: {weights['xgb']:.3f} | " + f"LGB: {weights['lgb']:.3f} | Ridge: {weights['ridge']:.3f}" + ) + return self + + def predict_proba(self, df: pd.DataFrame) -> np.ndarray: + X = _prepare_X(df, self.features_) + X_scaled = pd.DataFrame(self.scaler_.transform(X), columns=X.columns, index=X.index) + + p_xgb = self.xgb_model_.predict_proba(X)[:, 1] + p_lgb = self.lgb_model_.predict_proba(X)[:, 1] + p_ridge = self.ridge_model_.predict_proba(X_scaled)[:, 1] + + meta_X = np.column_stack([p_xgb, p_lgb, p_ridge]) + return self.meta_model_.predict_proba(meta_X) + + def predict(self, df: pd.DataFrame) -> np.ndarray: + return (self.predict_proba(df)[:, 1] >= 0.5).astype(int) + + def get_model_weights(self) -> Dict[str, float]: + if self.meta_model_ is None: + return {"xgb": 1/3, "lgb": 1/3, "ridge": 1/3} + coefs = np.abs(self.meta_model_.coef_[0]) + total = coefs.sum() or 1.0 + return { + "xgb": round(coefs[0] / total, 4), + "lgb": round(coefs[1] / total, 4), + "ridge": round(coefs[2] / total, 4), + } + + def save(self, path) -> None: + with open(path, "wb") as f: + pickle.dump(self, f) + + @staticmethod + def load(path) -> "AlphaEdgeEnsemble": + with open(path, "rb") as f: + return pickle.load(f) \ No newline at end of file diff --git a/src/models/ensemble_model.pkl b/src/models/ensemble_model.pkl new file mode 100644 index 0000000..6d3c623 Binary files /dev/null and b/src/models/ensemble_model.pkl differ diff --git a/src/models/kmeans_model.pkl b/src/models/kmeans_model.pkl deleted file mode 100644 index 8425584..0000000 Binary files a/src/models/kmeans_model.pkl and /dev/null differ diff --git a/src/models/metrics.json b/src/models/metrics.json deleted file mode 100644 index 1dbcc96..0000000 --- a/src/models/metrics.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "accuracy": 0.562874251497006, - "precision": 0.5973333333333334, - "recall": 0.5114155251141552, - "f1_score": 0.5510455104551045, - "auc_score": 0.5903005417342395, - "confusion_matrix": [ - [ - 246, - 151 - ], - [ - 214, - 224 - ] - ], - "training_date": "2025-12-31 09:40:17", - "best_params": { - "colsample_bytree": 0.8, - "learning_rate": 0.01, - "max_depth": 3, - "n_estimators": 300, - "subsample": 0.8 - }, - "test_samples": 835, - "features_count": 16, - "top_feature": "macd" -} \ No newline at end of file diff --git a/src/models/model_card.json b/src/models/model_card.json new file mode 100644 index 0000000..18ca289 --- /dev/null +++ b/src/models/model_card.json @@ -0,0 +1,58 @@ +{ + "market": "CAC40", + "trained_at": "2026-06-27T17:42:46.440788", + "architecture": "XGBoost + LightGBM + Ridge \u2192 LogisticRegression", + "metrics_ml": { + "auc_test": 0.5816, + "apr_test": 0.6404, + "baseline_auc": 0.5, + "lift": 0.0816 + }, + "metrics_fin": { + "sharpe": 2.3468, + "max_drawdown": -0.0162, + "total_return": 0.1318 + }, + "model_weights": { + "xgb": 0.0367, + "lgb": 0.0068, + "ridge": 0.9565 + }, + "walk_forward": [ + { + "window": 1, + "test_start": "2026-04-30", + "test_end": "2026-06-30", + "auc": 0.4969, + "apr": 0.6741, + "n_test": 119 + }, + { + "window": 2, + "test_start": "2026-01-30", + "test_end": "2026-03-31", + "auc": 0.6557, + "apr": 0.5795, + "n_test": 120 + }, + { + "window": 3, + "test_start": "2025-10-31", + "test_end": "2025-12-31", + "auc": 0.4829, + "apr": 0.6012, + "n_test": 120 + }, + { + "window": 4, + "test_start": "2025-07-31", + "test_end": "2025-09-30", + "auc": 0.5132, + "apr": 0.5482, + "n_test": 120 + } + ], + "n_features": 50, + "train_size": 4440, + "test_size": 279 +} \ No newline at end of file diff --git a/src/models/model_loader.py b/src/models/model_loader.py new file mode 100644 index 0000000..fe4da1b --- /dev/null +++ b/src/models/model_loader.py @@ -0,0 +1,68 @@ +import os +import pickle +import mlflow +import mlflow.sklearn +from mlflow.tracking import MlflowClient +from pathlib import Path + +from const import MODEL_DIR +from src.utils.logger import setup_logger + +logger = setup_logger("model_loader") + + +def load_champion(market_name: str): + """ + Charge le modèle champion depuis MLflow registry pour un marché donné. + Fallback 1 : dernière version enregistrée si l'alias 'champion' est absent. + Fallback 2 : .pkl local si MLflow est totalement indisponible. + + Args: + market_name: Nom du marché tel que défini dans config/markets/*.json + (ex: "CAC40", "NASDAQ", "SP500") + """ + hf_token = os.getenv("HF_TOKEN") + registered_name = f"AlphaEdge_Ensemble_{market_name}" + + if hf_token: + os.environ["MLFLOW_TRACKING_USERNAME"] = "SORADATA" + os.environ["MLFLOW_TRACKING_PASSWORD"] = hf_token + mlflow.set_tracking_uri("https://soradata-alphaedge-registry.hf.space") + + # Tentative 1 : alias champion + try: + model_uri = f"models:/{registered_name}@champion" + model = mlflow.sklearn.load_model(model_uri) + logger.info(f"[{market_name}] Champion chargé depuis MLflow : {model_uri}") + return model + except Exception as e: + logger.warning(f"[{market_name}] Alias 'champion' introuvable ({e}), tentative sur dernière version...") + + # Tentative 2 : dernière version enregistrée + try: + client = MlflowClient() + versions = client.get_latest_versions(registered_name) + if versions: + latest = sorted(versions, key=lambda v: int(v.version))[-1] + model_uri = f"models:/{registered_name}/{latest.version}" + model = mlflow.sklearn.load_model(model_uri) + logger.warning( + f"[{market_name}] Pas de champion — version {latest.version} chargée : {model_uri}" + ) + return model + else: + logger.warning(f"[{market_name}] Aucune version enregistrée pour {registered_name}.") + except Exception as e: + logger.warning(f"[{market_name}] MLflow indisponible ({e}), fallback disque local.") + + # Fallback final : .pkl local par marché + pkl_path = MODEL_DIR / market_name / "ensemble_model.pkl" + if not pkl_path.exists(): + raise FileNotFoundError( + f"[{market_name}] Aucun modèle disponible — MLflow inaccessible et pas de .pkl local à {pkl_path}" + ) + + with open(pkl_path, "rb") as f: + model = pickle.load(f) + logger.info(f"[{market_name}] Champion chargé depuis disque : {pkl_path}") + return model \ No newline at end of file diff --git a/src/models/train.py b/src/models/train.py new file mode 100644 index 0000000..0df5693 --- /dev/null +++ b/src/models/train.py @@ -0,0 +1,218 @@ +import os +import json +import warnings +from pathlib import Path + +import pandas as pd +from dotenv import load_dotenv +from sklearn.dummy import DummyClassifier +from sklearn.metrics import roc_auc_score, average_precision_score +import mlflow +import mlflow.sklearn +from mlflow.tracking import MlflowClient + +from const import DATA_DIR, MODEL_DIR, CONFIG_DIR, SHARPE_THRESHOLD, MAX_DD_THRESHOLD +from src.utils.metrics import calculate_financial_metrics +from src.features.alpha_features import add_all_features +from src.models.ensemble import AlphaEdgeEnsemble, FEATURE_GROUPS +from src.utils.logger import setup_logger + +load_dotenv() +warnings.filterwarnings("ignore") +logger = setup_logger("train") + +HF_TOKEN = os.getenv("HF_TOKEN") +USE_MLFLOW = bool(HF_TOKEN) + +if USE_MLFLOW: + os.environ["MLFLOW_TRACKING_USERNAME"] = "SORADATA" + os.environ["MLFLOW_TRACKING_PASSWORD"] = HF_TOKEN + mlflow.set_tracking_uri("https://soradata-alphaedge-registry.hf.space") + mlflow.set_experiment("AlphaEdge_Ensemble_Production") + + +# ============================================================================= +# WALK-FORWARD VALIDATION +# ============================================================================= +def walk_forward_eval( + df: pd.DataFrame, + n_windows: int = 4, + test_months: int = 3, + n_optuna_trials: int = 20, +) -> pd.DataFrame: + dates = df.index.get_level_values("date").unique().sort_values() + results = [] + + for i in range(n_windows): + test_end = dates[-(i * test_months + 1)] + test_start = dates[-(i * test_months + test_months)] + train_end = test_start - pd.DateOffset(months=1) + + df_tr = df[df.index.get_level_values("date") <= train_end] + df_te = df[ + (df.index.get_level_values("date") >= test_start) + & (df.index.get_level_values("date") <= test_end) + ] + + if len(df_tr) < 50 or len(df_te) < 5: + continue + if len(df_te["target"].unique()) < 2: + continue + + model = AlphaEdgeEnsemble(n_optuna_trials=n_optuna_trials) + model.fit(df_tr, df_tr["target"]) + proba = model.predict_proba(df_te)[:, 1] + + results.append({ + "window": i + 1, + "test_start": str(test_start.date()), + "test_end": str(test_end.date()), + "auc": round(roc_auc_score(df_te["target"], proba), 4), + "apr": round(average_precision_score(df_te["target"], proba), 4), + "n_test": len(df_te), + }) + logger.info( + f"Window {i+1} | AUC: {results[-1]['auc']:.4f} | " + f"APR: {results[-1]['apr']:.4f} | n={results[-1]['n_test']}" + ) + + return pd.DataFrame(results) + + +# ============================================================================= +# PIPELINE D'ENTRAÎNEMENT +# ============================================================================= +def train_pipeline(market_name: str) -> tuple[AlphaEdgeEnsemble, dict]: + logger.info(f"Début du cycle d'entraînement AlphaEdge — {market_name}") + + data_path = DATA_DIR / "processed" / market_name / "monthly_features.parquet" + if not data_path.exists(): + raise FileNotFoundError(f"Fichier source introuvable : {data_path}") + + df = pd.read_parquet(data_path) + + # Eviter le leakage data + df["future_return"] = df.groupby(level="ticker")["adj close"].pct_change(1).shift(-1) + df["target"] = df["future_return"].gt(0).astype(int) + df = df.dropna(subset=["target", "future_return"]) + + dates = df.index.get_level_values("date") + split_date = dates.max() - pd.DateOffset(months=6) + + # Split temporel AVANT add_all_features pour éviter le leakage global + df_train = df[dates <= split_date].copy() + df_test = df[dates > split_date].copy() + + df_train = add_all_features(df_train) + df_test = add_all_features(df_test) + + if len(df_train) < 100: + raise ValueError(f"Volume de données insuffisant : {len(df_train)}") + + # Baseline + all_feats = [f for g in FEATURE_GROUPS.values() for f in g] + available = [f for f in all_feats if f in df_train.columns] + dummy = DummyClassifier(strategy="most_frequent").fit(df_train[available], df_train["target"]) + baseline_auc = roc_auc_score(df_test["target"], dummy.predict_proba(df_test[available])[:, 1]) + + # Entraînement + model = AlphaEdgeEnsemble(n_optuna_trials=50) + model.fit(df_train, df_train["target"]) + + proba = model.predict_proba(df_test)[:, 1] + final_auc = roc_auc_score(df_test["target"], proba) + final_apr = average_precision_score(df_test["target"], proba) + lift = final_auc - baseline_auc + fin_metrics = calculate_financial_metrics(df_test, probas=proba, threshold=0.5) + + logger.info(f"ML -> AUC : {final_auc:.4f} | APR : {final_apr:.4f} | Lift : +{lift:.4f}") + logger.info(f"Finances -> Sharpe : {fin_metrics['sharpe']} | Max DD : {fin_metrics['max_drawdown']}") + + # Walk-forward sur df entier (target déjà calculé, pas de leakage supplémentaire) + df_full = pd.concat([df_train, df_test]) + wf_results = walk_forward_eval(df_full, n_windows=4, test_months=3, n_optuna_trials=20) + + # Sauvegarde locale par marché + market_model_dir = MODEL_DIR / market_name + market_model_dir.mkdir(parents=True, exist_ok=True) + model.save(market_model_dir / "ensemble_model.pkl") + logger.info(f"Modèle sauvegardé : {market_model_dir / 'ensemble_model.pkl'}") + + model_card = { + "market": market_name, + "trained_at": pd.Timestamp.now().isoformat(), + "architecture": "XGBoost + LightGBM + Ridge → LogisticRegression", + "metrics_ml": { + "auc_test": round(final_auc, 4), + "apr_test": round(final_apr, 4), + "baseline_auc": round(baseline_auc, 4), + "lift": round(lift, 4), + }, + "metrics_fin": fin_metrics, + "model_weights": model.get_model_weights(), + "walk_forward": wf_results.to_dict(orient="records") if not wf_results.empty else [], + "n_features": len(model.features_), + "train_size": len(df_train), + "test_size": len(df_test), + } + with open(market_model_dir / "model_card.json", "w") as f: + json.dump(model_card, f, indent=2) + + # MLflow + if USE_MLFLOW: + registered_model_name = f"AlphaEdge_Ensemble_{market_name}" + try: + with mlflow.start_run(run_name=f"Ensemble_{market_name}") as run: + mlflow.log_params({ + "architecture": "XGB+LGB+Ridge->LR", + "market": market_name, + "n_features": len(model.features_), + }) + mlflow.log_metrics({ + "AUC_Test": final_auc, + "APR_Test": final_apr, + "Sharpe_Ratio": fin_metrics["sharpe"], + "Max_Drawdown": fin_metrics["max_drawdown"], + "Total_Return": fin_metrics["total_return"], + "WF_AUC_mean": wf_results["auc"].mean() if not wf_results.empty else 0.0, + }) + mlflow.log_dict(model_card, "model_card.json") + mlflow.sklearn.log_model(model, name="ensemble_model") + mv = mlflow.register_model( + model_uri=f"runs:/{run.info.run_id}/ensemble_model", + name=registered_model_name, + ) + client = MlflowClient() + if ( + fin_metrics["sharpe"] >= SHARPE_THRESHOLD + and fin_metrics["max_drawdown"] >= MAX_DD_THRESHOLD + ): + client.set_registered_model_alias(registered_model_name, "champion", mv.version) + logger.info(f"[{market_name}] Promotion : Version {mv.version} → 'champion'.") + else: + logger.warning( + f"[{market_name}] Promotion refusée : seuils non atteints. " + "Ancien champion maintenu." + ) + except Exception as e: + logger.error(f"[{market_name}] Échec MLflow ({e}) — modèle local sauvegardé uniquement.") + + return model, model_card + + +# ============================================================================= +# ORCHESTRATEUR +# ============================================================================= +if __name__ == "__main__": + config_dir = CONFIG_DIR / "markets" + if not config_dir.exists(): + raise FileNotFoundError(f"Dossier de configs introuvable : {config_dir}") + + for config_file in sorted(config_dir.glob("*.json")): + with open(config_file) as f: + market_cfg = json.load(f) + market = market_cfg.get("market_name") + if not market: + logger.warning(f"Clé 'market_name' absente dans {config_file.name}, ignoré.") + continue + train_pipeline(market) diff --git a/src/models/xgboost_model.pkl b/src/models/xgboost_model.pkl deleted file mode 100644 index 1e4000c..0000000 Binary files a/src/models/xgboost_model.pkl and /dev/null differ diff --git a/src/pipeline/backtest.py b/src/pipeline/backtest.py index 481ec39..ce9c4da 100644 --- a/src/pipeline/backtest.py +++ b/src/pipeline/backtest.py @@ -1,88 +1,216 @@ -from typing import Any, Dict, Tuple +from typing import Any, Dict, List, Tuple import numpy as np import pandas as pd -from pypfopt import EfficientFrontier, risk_models, expected_returns +from pypfopt import EfficientFrontier, risk_models, expected_returns, objective_functions + from const import ( - TARGET_CLUSTER, - PROBA_THRESHOLD, - FEATURE_COLS, TRADING_DAYS_YEAR, - RISK_FREE_RATE + RISK_FREE_RATE, + TRANSACTION_COST, + MIN_STOCKS_OPTIM, + MAX_STOCKS_SELECT, + PROBA_MIN, + WEIGHT_BOUNDS ) from src.utils.logger import setup_logger -from src.utils.config_loader import BENCHMARK_TICKER from src.utils.market_utils import get_benchmark_returns logger = setup_logger("backtest") -def get_optimal_weights(prices_df: pd.DataFrame) -> Tuple[Dict[str, float], bool]: +def get_optimal_weights(prices_df: pd.DataFrame, risk_free_rate: float = RISK_FREE_RATE) -> Tuple[Dict[str, float], str]: + if prices_df.shape[1] < MIN_STOCKS_OPTIM: + n = prices_df.shape[1] + return {t: 1.0 / n for t in prices_df.columns}, "equal_weight" + try: - mu = expected_returns.mean_historical_return(prices_df, frequency=TRADING_DAYS_YEAR) + mu = expected_returns.ema_historical_return(prices_df, frequency=TRADING_DAYS_YEAR, span=252) S = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() - ef = EfficientFrontier(mu, S, weight_bounds=(0.02, 0.25)) - ef.max_sharpe(risk_free_rate=RISK_FREE_RATE) - - return dict(ef.clean_weights()), True - except Exception as e: - logger.warning(f"Optimization failed: {e}") - return {}, False + ef = EfficientFrontier(mu, S, weight_bounds=WEIGHT_BOUNDS) + ef.add_objective(objective_functions.L2_reg, gamma=0.1) + ef.max_sharpe(risk_free_rate=risk_free_rate) + return dict(ef.clean_weights()), "max_sharpe" -def _generate_monthly_signals( - month_data: pd.DataFrame, - xgb_model: Any, - kmeans_model: Any, -) -> pd.DataFrame: - if "rsi" in month_data.columns: - month_data = month_data.copy() - month_data["cluster"] = kmeans_model.predict(month_data[["rsi"]].fillna(50)) + except Exception as e1: + logger.warning(f"Max Sharpe failed ({e1}) -> min_volatility fallback.") + try: + mu = expected_returns.ema_historical_return(prices_df, frequency=TRADING_DAYS_YEAR, span=252) + S = risk_models.CovarianceShrinkage(prices_df, frequency=TRADING_DAYS_YEAR).ledoit_wolf() + + ef2 = EfficientFrontier(mu, S, weight_bounds=WEIGHT_BOUNDS) + ef2.min_volatility() + + return dict(ef2.clean_weights()), "min_vol" - if not all(c in month_data.columns for c in FEATURE_COLS): + except Exception as e2: + logger.warning(f"Min vol failed ({e2}) -> equal_weight fallback.") + n = prices_df.shape[1] + return {t: 1.0 / n for t in prices_df.columns}, "equal_weight" + + +def _generate_monthly_signals(month_data: pd.DataFrame, model: Any) -> pd.DataFrame: + """ + Génère les signaux ML pour un mois donné. + Le modèle AlphaEdgeEnsemble extrait lui-même les features dont il a besoin. + """ + if month_data.empty: + return pd.DataFrame() + + try: + month_data = month_data.copy() + month_data["proba_upside"] = model.predict_proba(month_data)[:, 1] + except Exception as e: + logger.error(f"predict_proba failed: {e}") return pd.DataFrame() - month_data["proba_upside"] = xgb_model.predict_proba(month_data[FEATURE_COLS].fillna(0))[:, 1] return month_data +def _select_tickers(month_data: pd.DataFrame, proba_min: float = PROBA_MIN, max_stocks: int = MAX_STOCKS_SELECT) -> List[str]: + if "proba_upside" not in month_data.columns: + return [] + + selected = month_data[month_data["proba_upside"] >= proba_min] + + if selected.empty: + logger.info("Alerte Marché : Aucun signal au-dessus du seuil. Passage en 100% Cash.") + return [] + + return selected.nlargest(max_stocks, "proba_upside").index.tolist() + + +def _compute_turnover(new_alloc: Dict[str, float], old_alloc: Dict[str, float]) -> float: + all_tickers = set(new_alloc) | set(old_alloc) + return sum(abs(new_alloc.get(t, 0.0) - old_alloc.get(t, 0.0)) for t in all_tickers) / 2.0 + + def _simulate_daily_returns( allocation: Dict[str, float], + drifted_allocation: Dict[str, float], trading_days: pd.DatetimeIndex, daily_returns: pd.DataFrame, benchmark_returns: pd.Series, portfolio_value: float, benchmark_value: float, -) -> Tuple[list, float, float]: +) -> Tuple[list, float, float, Dict[str, float]]: records = [] + is_first_day = True + stock_values = {t: portfolio_value * w for t, w in allocation.items()} if allocation else {} + for date in trading_days: bench_ret = benchmark_returns.get(date, 0.0) - strat_ret = 0.0 + if is_first_day: + turnover = _compute_turnover(allocation, drifted_allocation) + transaction_fees = portfolio_value * turnover * TRANSACTION_COST + portfolio_value -= transaction_fees + stock_values = {t: portfolio_value * w for t, w in allocation.items()} + is_first_day = False - if allocation: - weights = list(allocation.values()) - rets = [ - daily_returns.loc[date, t] if t in daily_returns.columns and date in daily_returns.index else 0.0 - for t in allocation - ] - strat_ret = float(np.average(pd.Series(rets).fillna(0), weights=weights)) + daily_portfolio_pnl = 0.0 + if stock_values: + for t in list(stock_values.keys()): + ret = daily_returns.loc[date, t] if (t in daily_returns.columns and date in daily_returns.index) else 0.0 + pnl = stock_values[t] * ret + stock_values[t] += pnl + daily_portfolio_pnl += pnl - portfolio_value *= (1 + strat_ret) + portfolio_value += daily_portfolio_pnl benchmark_value *= (1 + bench_ret) - records.append({"Date": date, "Strategy": portfolio_value, "Benchmark": benchmark_value}) - return records, portfolio_value, benchmark_value + records.append({ + "Date": date, + "Strategy": portfolio_value, + "Benchmark": benchmark_value, + "N_Stocks": len(stock_values), + }) + + new_drifted_allocation = {} + if portfolio_value > 0 and stock_values: + new_drifted_allocation = {t: val / portfolio_value for t, val in stock_values.items()} + + return records, portfolio_value, benchmark_value, new_drifted_allocation + + +def compute_performance_metrics(results_df: pd.DataFrame, rebalance_log: pd.DataFrame, risk_free_rate: float = RISK_FREE_RATE) -> Dict[str, float]: + strat = results_df["Strategy"] + bench = results_df["Benchmark"] + + strat_ret = strat.pct_change().dropna() + bench_ret = bench.pct_change().dropna() + + n_years = len(strat_ret) / TRADING_DAYS_YEAR + cagr = (strat.iloc[-1] / strat.iloc[0]) ** (1 / n_years) - 1 if n_years > 0 else 0.0 + + vol = strat_ret.std() * np.sqrt(TRADING_DAYS_YEAR) + excess = strat_ret - risk_free_rate / TRADING_DAYS_YEAR + sharpe = (excess.mean() / strat_ret.std()) * np.sqrt(TRADING_DAYS_YEAR) if strat_ret.std() != 0 else 0.0 + + downside = strat_ret[strat_ret < 0].std() * np.sqrt(TRADING_DAYS_YEAR) + sortino = (cagr - risk_free_rate) / downside if downside > 0 else np.nan + + rolling_max = strat.cummax() + drawdown = (strat - rolling_max) / rolling_max + max_dd = drawdown.min() + + calmar = cagr / abs(max_dd) if max_dd != 0 else np.nan + + align_df = pd.concat([strat_ret, bench_ret], axis=1).dropna() + if len(align_df) > 1: + cov_matrix = np.cov(align_df.iloc[:, 0], align_df.iloc[:, 1]) + beta = cov_matrix[0, 1] / cov_matrix[1, 1] if cov_matrix[1, 1] != 0 else np.nan + else: + beta = np.nan + + alpha = (cagr - risk_free_rate) - beta * ( + (bench.iloc[-1] / bench.iloc[0]) ** (1 / n_years) - 1 - risk_free_rate + ) + + monthly_strat = strat.resample("BME").last().pct_change().dropna() + hit_rate = (monthly_strat > 0).mean() + + avg_stocks = rebalance_log["N_Stocks"].mean() if "N_Stocks" in rebalance_log.columns else np.nan + + metrics = { + "CAGR": round(cagr, 4), + "Volatility": round(vol, 4), + "Sharpe": round(sharpe, 4), + "Sortino": round(sortino, 4), + "Calmar": round(calmar, 4), + "Max_Drawdown": round(max_dd, 4), + "Alpha": round(alpha, 4), + "Beta": round(beta, 4), + "Hit_Rate": round(hit_rate, 4), + "Avg_N_Stocks": round(avg_stocks, 1), + "Final_Value": round(strat.iloc[-1], 2), + } + + logger.info("=" * 50) + logger.info("📊 PERFORMANCE METRICS") + logger.info(f" CAGR : {cagr:.2%}") + logger.info(f" Sharpe Ratio : {sharpe:.3f}") + logger.info(f" Sortino Ratio : {sortino:.3f}") + logger.info(f" Calmar Ratio : {calmar:.3f}") + logger.info(f" Max Drawdown : {max_dd:.2%}") + logger.info(f" Alpha : {alpha:.2%}") + logger.info(f" Beta : {beta:.3f}") + logger.info(f" Hit Rate : {hit_rate:.1%}") + logger.info("=" * 50) + + return metrics def backtest_strategy_with_rebalancing( df_daily: pd.DataFrame, df_monthly: pd.DataFrame, - xgb_model: Any, - kmeans_model: Any, - get_optimal_weights_fn: Any, -) -> Tuple[pd.DataFrame, pd.DataFrame]: - logger.info("Starting backtest...") + model: Any, + benchmark_ticker: str = "^FCHI", + proba_min: float = PROBA_MIN, + max_stocks: int = MAX_STOCKS_SELECT, +) -> Tuple[pd.DataFrame, pd.DataFrame, Dict[str, float]]: + logger.info(f"Starting backtest | Benchmark: {benchmark_ticker}") daily_prices = df_daily["adj close"].unstack().ffill() daily_returns = daily_prices.pct_change().fillna(0) @@ -90,45 +218,66 @@ def backtest_strategy_with_rebalancing( date_min = df_daily.index.get_level_values("date").min() date_max = df_daily.index.get_level_values("date").max() - benchmark_returns = get_benchmark_returns( - BENCHMARK_TICKER, date_min, date_max, daily_prices.index - ) + benchmark_returns = get_benchmark_returns(benchmark_ticker, date_min, date_max, daily_prices.index) - portfolio_value, benchmark_value = 100.0, 100.0 - all_records, rebalance_log = [], [] + portfolio_value = 100.0 + benchmark_value = 100.0 + all_records = [] + rebalance_log = [] + drifted_allocation: Dict[str, float] = {} monthly_dates = df_monthly.index.get_level_values("date").unique().sort_values() for i, month_date in enumerate(monthly_dates[:-1]): month_data = _generate_monthly_signals( df_monthly.xs(month_date, level="date").copy(), - xgb_model, - kmeans_model, + model, ) allocation: Dict[str, float] = {} + optim_method = "no_signal" + if not month_data.empty: - selected = month_data[ - (month_data["cluster"] == TARGET_CLUSTER) & - (month_data["proba_upside"] > PROBA_THRESHOLD) - ] + tickers = _select_tickers(month_data, proba_min, max_stocks) - if not selected.empty: - tickers = selected.index.tolist() - prices_subset = daily_prices[tickers].iloc[-TRADING_DAYS_YEAR:].dropna(axis=1) + if tickers: + available_tickers = [t for t in tickers if t in daily_prices.columns] + prices_subset = ( + daily_prices[available_tickers] + .loc[:month_date] + .iloc[-TRADING_DAYS_YEAR:] + .dropna(axis=1, thresh=int(TRADING_DAYS_YEAR * 0.8)) + ) - if not prices_subset.empty and len(prices_subset.columns) >= 3: - weights, success = get_optimal_weights_fn(prices_subset) - allocation = weights if success else {t: 1.0 / len(tickers) for t in tickers} + if not prices_subset.empty: + weights, optim_method = get_optimal_weights(prices_subset) + allocation = {t: w for t, w in weights.items() if w > 1e-4} next_month = monthly_dates[i + 1] - trading_days = daily_prices.index[(daily_prices.index >= month_date) & (daily_prices.index < next_month)] + trading_days = daily_prices.index[ + (daily_prices.index >= month_date) & + (daily_prices.index < next_month) + ] - day_records, portfolio_value, benchmark_value = _simulate_daily_returns( - allocation, trading_days, daily_returns, + day_records, portfolio_value, benchmark_value, drifted_allocation = _simulate_daily_returns( + allocation, drifted_allocation, + trading_days, daily_returns, benchmark_returns, portfolio_value, benchmark_value, ) all_records.extend(day_records) - rebalance_log.append({"Date": month_date, "N_Stocks": len(allocation), "Allocation": allocation}) - logger.info(f"Backtest complete. Final value: {portfolio_value:.2f}") - return pd.DataFrame(all_records).set_index("Date"), pd.DataFrame(rebalance_log).set_index("Date") \ No newline at end of file + rebalance_log.append({ + "Date": month_date, + "N_Stocks": len(allocation), + "Optim_Method": optim_method, + "Allocation": allocation, + "Top_Ticker": max(allocation, key=allocation.get) if allocation else None, + }) + + results_df = pd.DataFrame(all_records).set_index("Date") + rebalance_df = pd.DataFrame(rebalance_log).set_index("Date") + + metrics = compute_performance_metrics(results_df, rebalance_df) + + logger.info(f" Backtest complete | Final value: {portfolio_value:.2f}") + + return results_df, rebalance_df, metrics diff --git a/src/pipeline/daily_run.py b/src/pipeline/daily_run.py new file mode 100644 index 0000000..81ba442 --- /dev/null +++ b/src/pipeline/daily_run.py @@ -0,0 +1,104 @@ +import os +import json +import warnings +from pathlib import Path + +import pandas as pd +from huggingface_hub import HfApi + +from src.utils.logger import setup_logger +from src.pipeline.etl import get_data_pipeline +from src.pipeline.backtest import backtest_strategy_with_rebalancing +from src.models.model_loader import load_champion +from const import BACKTEST_YEARS + +# ============================================================================= +# CONFIGURATION GLOBALE +# ============================================================================= +warnings.filterwarnings("ignore") +HF_TOKEN = os.getenv("HF_TOKEN") +HF_REPO_ID = os.getenv("HF_REPO_ID", "soradata/alphaedge-data") +hf_api = HfApi() + + +def upload_to_hf(local_path: Path, hf_filename: str, market_name: str): + """Upload silencieux vers le repo Hugging Face.""" + if not HF_TOKEN: + return + try: + hf_api.upload_file( + path_or_fileobj=str(local_path), + path_in_repo=f"data/{market_name}/{hf_filename}", + repo_id=HF_REPO_ID, + repo_type="dataset", + token=HF_TOKEN, + ) + except Exception as e: + print(f"Erreur Upload HF pour {market_name}: {e}") + + +# ============================================================================= +# PIPELINE PRINCIPAL +# ============================================================================= +def run_pipeline(market_config: dict): + market_name = market_config.get("market_name", "UNKNOWN") + logger = setup_logger(f"Pipeline_{market_name}") + logger.info(f" STARTING PIPELINE | MARKET: {market_name}") + + try: + # 1. Chargement du modèle champion depuis MLflow + model = load_champion(market_name) + logger.info(f"Modèle champion prêt pour {market_name}") + + # 2. ETL + df_daily, df_monthly = get_data_pipeline(market_config) + if df_daily is None or df_monthly is None: + raise ValueError("L'ETL n'a renvoyé aucune donnée.") + + cutoff = pd.Timestamp.now() - pd.DateOffset(years=BACKTEST_YEARS) + df_daily_bt = df_daily[df_daily.index.get_level_values("date") >= cutoff] + df_monthly_bt = df_monthly[df_monthly.index.get_level_values("date") >= cutoff] + + if df_daily_bt.empty or df_monthly_bt.empty: + raise ValueError(f"Pas de données sur les {BACKTEST_YEARS} dernières années.") + + logger.info( + f"Backtest window : {cutoff.date()} → aujourd'hui " + f"({len(df_monthly_bt.index.get_level_values('date').unique())} mois)" + ) + + # 4. Backtest + logger.info(f"Executing backtest for {market_name}...") + hist_df, rebal_df, metrics = backtest_strategy_with_rebalancing( + df_daily_bt, + df_monthly_bt, + model, + benchmark_ticker=market_config.get("benchmark_ticker", "^FCHI"), + ) + + # 5. Sauvegarde locale + base_dir = Path(f"data/processed/{market_name}") + base_dir.mkdir(parents=True, exist_ok=True) + hist_path = base_dir / "portfolio_history.parquet" + rebal_path = base_dir / "rebalance_history.parquet" + hist_df.to_parquet(hist_path) + rebal_df.to_parquet(rebal_path) + + # 6. Synchronisation Cloud + upload_to_hf(hist_path, "portfolio_history.parquet", market_name) + upload_to_hf(rebal_path, "rebalance_history.parquet", market_name) + logger.info(f" Pipeline terminé | Sharpe: {metrics.get('Sharpe', 'N/A')}") + + except Exception as e: + logger.critical(f"CRITICAL FAILURE {market_name}: {e}", exc_info=True) + + +# ============================================================================= +# ORCHESTRATEUR +# ============================================================================= +if __name__ == "__main__": + config_dir = Path("config/markets") + for config_file in sorted(config_dir.glob("*.json")): + with open(config_file) as f: + market_config = json.load(f) + run_pipeline(market_config) \ No newline at end of file diff --git a/src/pipeline/etl.py b/src/pipeline/etl.py index 43ef670..6e40355 100644 --- a/src/pipeline/etl.py +++ b/src/pipeline/etl.py @@ -1,137 +1,76 @@ import json -import time import pickle from datetime import datetime from typing import Optional, Tuple, Any import pandas as pd -import yfinance as yf - -from const import DATA_DIR, BASE_DIR, VARS_TO_LAG, RESAMPLE_MEAN_COLS, RESAMPLE_LAST_EXCLUDE -from src.transform.features import ( - compute_technical_indicators, - calculate_returns, - get_fama_french_betas -) -from src.transform.ticker_manager import handle_ticker_changes, validate_and_clean_tickers -from src.utils.config_loader import TICKERS + +from const import DATA_DIR, BASE_DIR +from src.extract.extractor import MarketExtractor +from src.transform.processor import MarketDataProcessor +from src.transform.ticker_manager import handle_ticker_changes from src.utils.logger import setup_logger logger = setup_logger("etl") -_DOWNLOAD_RETRIES = 3 -_DOWNLOAD_RETRY_WAIT = 5 -_HISTORY_YEARS = 10 - - -def _download_raw_prices(tickers: list[str]) -> Optional[pd.DataFrame]: - """Downloads adjusted OHLCV data from Yahoo Finance with retry logic.""" - end = (datetime.today() + pd.DateOffset(days=1)).strftime("%Y-%m-%d") - start = (pd.Timestamp.today() - pd.DateOffset(years=_HISTORY_YEARS)).strftime("%Y-%m-%d") - - logger.info(f"Downloading market data ({start} → {end}) for {len(tickers)} assets...") - - for attempt in range(1, _DOWNLOAD_RETRIES + 1): - try: - df = yf.download( - tickers, - start=start, - end=end, - progress=False, - auto_adjust=False, - threads=True, - ) - if not df.empty: - logger.info(f"Download successful (attempt {attempt})") - return df - logger.warning(f"Empty response (attempt {attempt})") - except Exception as exc: - logger.warning(f"Download error (attempt {attempt}): {exc}") - time.sleep(_DOWNLOAD_RETRY_WAIT) - return None - - -def _resample_to_monthly(df: pd.DataFrame) -> pd.DataFrame: - """Resamples daily data to business-month-end frequency.""" - last_cols = [c for c in df.columns if c not in RESAMPLE_LAST_EXCLUDE] - - monthly = pd.concat( - [ - df.unstack("ticker")[RESAMPLE_MEAN_COLS[0]] - .resample("BM").mean() - .stack("ticker") - .to_frame(RESAMPLE_MEAN_COLS[0]), - df.unstack()[last_cols].resample("BM").last().stack("ticker"), - ], - axis=1, - ).dropna() - return monthly - - -def get_data_pipeline() -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame]]: - """Full ETL pipeline orchestrator.""" + +def get_data_pipeline(market_config: dict) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame]]: + """Orchestrateur global du pipeline ETL : Extract -> Process -> Load.""" + market_name = market_config['market_name'] + tickers = market_config['tickers'] + # Résolution des tickers (changement de noms, delisting) ticker_changes, delisted = handle_ticker_changes() active_tickers = [ - ticker_changes.get(t, t) - for t in TICKERS - if t not in delisted + ticker_changes.get(t, t) for t in tickers if t not in delisted ] - raw = _download_raw_prices(active_tickers) - if raw is None: + # ========================================== + # 1. EXTRACT (Extraction) + # ========================================== + extractor = MarketExtractor(market_name=market_name, tickers=active_tickers) + raw = extractor.fetch_market_data() + if raw is None or raw.empty: + logger.error(f"Abandon du pipeline pour {market_name} : Aucune donnée extraite.") return None, None - df = raw.stack(future_stack=True) - df.index.names = ["date", "ticker"] - df.columns = df.columns.str.lower() - if "adj close" not in df.columns and "close" in df.columns: - df["adj close"] = df["close"] + # ========================================== + # 2. TRANSFORM (Processing) + # ========================================== + processor = MarketDataProcessor(active_tickers=active_tickers) + df_daily, df_monthly, alerts = processor.process(raw) - df, valid_tickers, alerts = validate_and_clean_tickers(df, active_tickers) - - # Sauvegarde des logs de validation + # ========================================== + # 3. LOAD + # ========================================== BASE_DIR.mkdir(parents=True, exist_ok=True) - with open(BASE_DIR / "ticker_validation.json", "w") as fh: + with open(BASE_DIR / f"{market_name}_ticker_validation.json", "w") as fh: json.dump( - {"date": str(datetime.now()), "alerts": alerts, "valid_tickers": len(valid_tickers)}, + { + "date": str(datetime.now()), + "alerts": alerts, + "valid_tickers": len(active_tickers) - len(alerts) + }, fh, indent=2, ) - logger.info(f"Saving raw data to {DATA_DIR}...") - DATA_DIR.mkdir(parents=True, exist_ok=True) - df.to_parquet(DATA_DIR / "daily_raw.parquet", compression="gzip") - - # Calcul des indicateurs (la fonction corrigée) - df = compute_technical_indicators(df) - - logger.info("Resampling to monthly frequency...") - df_monthly = _resample_to_monthly(df) - df_monthly = df_monthly.groupby(level=1, group_keys=False).apply(calculate_returns) - df_monthly = get_fama_french_betas(df_monthly) - - # Ajout des lags - for col in VARS_TO_LAG: - if col in df_monthly.columns: - df_monthly[f"{col}_lag1"] = df_monthly.groupby("ticker")[col].shift(1) - - logger.info("Saving monthly features...") - df_monthly.to_parquet(DATA_DIR / "monthly_features.parquet", compression="gzip") + processed_dir = DATA_DIR / "processed" / market_name + processed_dir.mkdir(parents=True, exist_ok=True) + logger.info(f"Sauvegarde des données dans {processed_dir}...") + df_daily.to_parquet(processed_dir / "daily_raw.parquet", compression="gzip") + df_monthly.to_parquet(processed_dir / "monthly_features.parquet", compression="gzip") - return df, df_monthly + return df_daily, df_monthly def load_models() -> Tuple[Optional[Any], Optional[Any]]: - """ - Charge les modèles pré-entraînés XGBoost et KMeans depuis le dossier MODEL_DIR. - """ + """Charge les modèles pré-entraînés XGBoost et KMeans depuis MODEL_DIR.""" from const import MODEL_DIR logger.info(f"Loading ML models from {MODEL_DIR}...") try: - # On s'assure que les fichiers existent avant d'ouvrir xgb_path = MODEL_DIR / 'xgboost_model.pkl' kmeans_path = MODEL_DIR / 'kmeans_model.pkl' if not xgb_path.exists() or not kmeans_path.exists(): - logger.error("Model files missing in MODEL_DIR.") + logger.error("Modèles introuvables dans MODEL_DIR.") return None, None with open(xgb_path, 'rb') as f: @@ -140,5 +79,29 @@ def load_models() -> Tuple[Optional[Any], Optional[Any]]: kmeans = pickle.load(f) return xgb, kmeans except Exception as e: - logger.error(f"Error loading models: {e}") + logger.error(f"Erreur lors du chargement des modèles : {e}") return None, None +# ========================================== +# BLOC DE TEST LOCAL +# ========================================== +if __name__ == "__main__": + import sys + import json + from pathlib import Path + + config_path = Path("config/markets/cac40.json") + if config_path.exists(): + config = json.load(open(config_path)) + else: + config = { + "market_name": "CAC40_Test", + "tickers": ["AI.PA", "AIR.PA", "OR.PA"] + } + + print(f"Lancement ETL — {config['market_name']}...") + df_daily, df_monthly = get_data_pipeline(config) + + if df_daily is not None: + print(f"✅ Succès ! Shape daily: {df_daily.shape}, Shape monthly: {df_monthly.shape}") + else: + print("❌ Échec.") \ No newline at end of file diff --git a/src/transform/.gitkeep b/src/transform/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/src/transform/features.py b/src/transform/features.py deleted file mode 100644 index 0a4551a..0000000 --- a/src/transform/features.py +++ /dev/null @@ -1,133 +0,0 @@ -import numpy as np -import pandas as pd -import statsmodels.api as sm -from statsmodels.regression.rolling import RollingOLS -from ta.momentum import RSIIndicator -from ta.volatility import BollingerBands - - -from const import ( - RSI_WINDOW, - BB_WINDOW, - BB_STD, - MIN_HISTORY_TA, - MIN_HISTORY_FF, - MOMENTUM_LAGS, - WINSOR_CUTOFF, - VARS_TO_LAG, - FAMA_FRENCH_FACTORS, -) -from src.utils.feature_utils import compute_atr, compute_macd -from src.utils.logger import setup_logger - -logger = setup_logger("features") - - -def compute_technical_indicators(df: pd.DataFrame) -> pd.DataFrame: - """ - Adds Garman-Klass volatility, RSI, Bollinger Bands, ATR, MACD, - and euro volume to a multi-index (date, ticker) DataFrame. - """ - logger.info("Computing technical indicators...") - - df["garman_klass_vol"] = ( - (np.log(df["high"]) - np.log(df["low"])) ** 2 / 2 - - (2 * np.log(2) - 1) - * (np.log(df["adj close"]) - np.log(df["open"])) ** 2 - ) - - for ticker in df.index.get_level_values(1).unique(): - idx = (slice(None), ticker) - close = df.loc[idx, "adj close"] - - if len(close) > MIN_HISTORY_TA: - df.loc[idx, "rsi"] = RSIIndicator( - close=close, window=RSI_WINDOW - ).rsi().values - - bb = BollingerBands( - close=np.log1p(close), window=BB_WINDOW, window_dev=BB_STD - ) - df.loc[idx, "bb_low"] = bb.bollinger_lband().values - df.loc[idx, "bb_mid"] = bb.bollinger_mavg().values - df.loc[idx, "bb_high"] = bb.bollinger_hband().values - - df["atr"] = df.groupby(level=1, group_keys=False).apply(compute_atr) - df["macd"] = df.groupby(level=1, group_keys=False).apply(compute_macd) - df["euro_volume"] = (df["adj close"] * df["volume"]) / 1e6 - - return df - - -def calculate_returns(df: pd.DataFrame) -> pd.DataFrame: - """ - Adds winsorized momentum return columns for each lag in MOMENTUM_LAGS. - Operates on a single-ticker slice (called via groupby). - """ - for lag in MOMENTUM_LAGS: - raw = df["adj close"].pct_change(lag) - lower = raw.expanding(min_periods=12).quantile(WINSOR_CUTOFF) - upper = raw.expanding(min_periods=12).quantile(1 - WINSOR_CUTOFF) - df[f"return_{lag}m"] = raw.clip(lower=lower, upper=upper) - return df - - -def get_fama_french_betas(data: pd.DataFrame) -> pd.DataFrame: - """ - Fetches Europe 5-Factor data from Kenneth French's library and computes - rolling 24-month OLS betas for each ticker. Fills with zeros on failure. - """ - logger.info("Retrieving Fama-French factors (Europe 5)...") - - try: - import pandas_datareader.data as web - except ImportError as exc: - logger.error(f"pandas_datareader unavailable ({exc}). Filling betas with zeros.") - return data.assign(**{f: 0.0 for f in FAMA_FRENCH_FACTORS}) - - try: - factor_data = ( - web.DataReader("Europe_5_Factors", "famafrench", start="2010")[0] - .drop("RF", axis=1) - ) - factor_data.index = factor_data.index.to_timestamp() - factor_data = factor_data.resample("BM").last().div(100) - factor_data.index.name = "date" - - if "return_1m" not in data.columns: - return data - - betas_list = [] - for ticker in data.index.get_level_values(1).unique(): - y = data.xs(ticker, level=1).get("return_1m") - if y is None or y.dropna().empty: - continue - - X = factor_data.loc[factor_data.index.intersection(y.index)] - y = y.loc[X.index] - - if len(y) <= MIN_HISTORY_FF: - continue - - params = ( - RollingOLS(y, sm.add_constant(X[FAMA_FRENCH_FACTORS]), window=MIN_HISTORY_FF) - .fit() - .params.drop("const", axis=1) - ) - params["ticker"] = ticker - betas_list.append(params) - - if not betas_list: - return data - - betas_df = pd.concat(betas_list).set_index("ticker", append=True) - data = data.join(betas_df.groupby("ticker").shift()) - data.loc[:, FAMA_FRENCH_FACTORS] = ( - data.groupby("ticker", group_keys=False)[FAMA_FRENCH_FACTORS] - .apply(lambda x: x.fillna(x.mean())) - ) - return data - - except Exception as exc: - logger.error(f"Fama-French retrieval failed ({exc}). Filling with zeros.") - return data.assign(**{f: 0.0 for f in FAMA_FRENCH_FACTORS}) diff --git a/src/transform/processor.py b/src/transform/processor.py new file mode 100644 index 0000000..cdbbea8 --- /dev/null +++ b/src/transform/processor.py @@ -0,0 +1,135 @@ +""" +MarketDataProcessor +==================== +Pipeline de transformation complète : + 1. Validation & nettoyage des tickers + 2. Indicateurs techniques (daily) + 3. Agrégation mensuelle + 4. Returns winsorisés (mensuel) + 5. Betas Fama-French rolling (mensuel) + 6. Alpha features + 7. Lag des variables (anti-leakage) +""" + +import pandas as pd +from typing import Tuple, List + +from const import VARS_TO_LAG, RESAMPLE_MEAN_COLS, RESAMPLE_LAST_EXCLUDE +# processor.py — imports corrects +from src.features.alpha_features import ( + compute_technical_indicators, + calculate_returns, + get_fama_french_betas, +) +from src.features.alpha_features import add_all_features +from src.transform.ticker_manager import validate_and_clean_tickers +from src.utils.logger import setup_logger + + +logger = setup_logger("processor") + + +class MarketDataProcessor: + """ + Encapsule toute la logique de transformation et de nettoyage des données. + + Parameters + ---------- + active_tickers : list[str] + Liste des tickers actifs sur le marché considéré. + """ + + def __init__(self, active_tickers: List[str]): + self.active_tickers = active_tickers + + # ── Agrégation mensuelle + def _resample_to_monthly(self, df: pd.DataFrame) -> pd.DataFrame: + """ + Agrège les données journalières en mensuel (Business Month End). + + - euro_volume : moyenne mensuelle (RESAMPLE_MEAN_COLS) + - autres cols : dernière valeur du mois (last) + """ + last_cols = [c for c in df.columns if c not in RESAMPLE_LAST_EXCLUDE] + + mean_part = ( + df.unstack("ticker")[RESAMPLE_MEAN_COLS[0]] + .resample("BME").mean() + .stack("ticker") + .to_frame(RESAMPLE_MEAN_COLS[0]) + ) + last_part = ( + df.unstack()[last_cols] + .resample("BME").last() + .stack("ticker", future_stack=True) + ) + + monthly = pd.concat([mean_part, last_part], axis=1).dropna(how="all") + + # S'assurer que l'index est bien ordonné (date, ticker) + monthly = monthly.sort_index() + + return monthly + + # Lag des variables macro/volume + def _apply_lags(self, df: pd.DataFrame) -> pd.DataFrame: + """ + Lag d'une période pour toutes les variables dans VARS_TO_LAG. + + Note : les lags des indicateurs TA (rsi, macd, bb_*, atr, cluster) + sont gérés directement dans alpha_features._lag_ta_indicators() + pour éviter la duplication. + """ + for col in VARS_TO_LAG: + if col in df.columns: + df[f"{col}_lag1"] = df.groupby(level="ticker")[col].shift(1) + return df + + # Pipeline principale + def process(self, raw_df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame, dict]: + """ + Exécute la pipeline de transformation complète. + + Parameters + ---------- + raw_df : pd.DataFrame + Données brutes multi-index (date, ticker) issues de l'extracteur. + + Returns + ------- + df_daily : pd.DataFrame — données journalières enrichies + df_monthly : pd.DataFrame — données mensuelles avec toutes les features + alerts : dict — alertes de validation des tickers + """ + logger.info("Début du processing des données...") + df = raw_df.copy() + if "adj close" not in df.columns and "close" in df.columns: + df["adj close"] = df["close"] + logger.warning("adj close absent — utilisation de close comme proxy.") + df, valid_tickers, alerts = validate_and_clean_tickers(df, self.active_tickers) + + df = compute_technical_indicators(df) + + logger.info("Agrégation à la fréquence mensuelle...") + df_monthly = self._resample_to_monthly(df) + + df_monthly = df_monthly.groupby( + level=1, group_keys=False + ).apply(calculate_returns) + + df_monthly = get_fama_french_betas(df_monthly) + + # alpha features + df_monthly = add_all_features(df_monthly) + + df_monthly = self._apply_lags(df_monthly) + + # Final verif + n_features = df_monthly.shape[1] + n_obs = len(df_monthly) + logger.info( + f"Processing terminé avec succès. " + f"Monthly shape : ({n_obs}, {n_features})" + ) + + return df, df_monthly, alerts diff --git a/src/utils/config_loader.py b/src/utils/config_loader.py index 16b8a7c..a2318be 100644 --- a/src/utils/config_loader.py +++ b/src/utils/config_loader.py @@ -1,33 +1,20 @@ import json +from pathlib import Path from typing import Dict, Any -from const import CONFIG_FILE from src.utils.logger import setup_logger logger = setup_logger("ConfigLoader") -def load_market_config() -> Dict[str, Any]: - """Loads market configuration from JSON file with safe fallbacks.""" - default_config = { - "market_name": "Default (CAC40)", - "benchmark_ticker": "^FCHI", - "assets": ["AI.PA", "AIR.PA", "SAN.PA", "MC.PA"] - } - if CONFIG_FILE.exists(): - logger.info(f"Loading configuration from: {CONFIG_FILE}") - try: - with open(CONFIG_FILE, 'r') as f: - return json.load(f) - except json.JSONDecodeError as e: - logger.error(f"Invalid JSON in config file: {e}. Using defaults.") - else: - logger.warning(f"Config file not found at {CONFIG_FILE}. Using defaults.") - return default_config - - -MARKET_CONFIG = load_market_config() -TICKERS = MARKET_CONFIG.get('assets', []) -BENCHMARK_TICKER = MARKET_CONFIG.get('benchmark_ticker', '^FCHI') -MARKET_NAME = MARKET_CONFIG.get('market_name', 'Unknown Market') - -logger.info(f"Market: {MARKET_NAME} | Assets: {len(TICKERS)} | Benchmark: {BENCHMARK_TICKER}") +def load_market_config(config_path: Path) -> Dict[str, Any]: + """Charge une configuration spécifique depuis un fichier JSON donné.""" + if not config_path.exists(): + logger.error(f"Fichier de config introuvable : {config_path}") + return {} + + try: + with open(config_path, 'r') as f: + return json.load(f) + except json.JSONDecodeError as e: + logger.error(f"Erreur JSON dans {config_path}: {e}") + return {} \ No newline at end of file diff --git a/src/utils/feature_utils.py b/src/utils/feature_utils.py index b9b6904..274d3ec 100644 --- a/src/utils/feature_utils.py +++ b/src/utils/feature_utils.py @@ -1,32 +1,87 @@ +""" +Feature Utils +============== +Fonctions utilitaires pour le calcul des indicateurs techniques. +Appelées depuis features.py via groupby().apply(). + +Chaque fonction reçoit un slice single-ticker et retourne une Series +alignée sur l'index du slice. + +Notes +----- +- ATR et MACD sont z-scorés pour homogénéiser l'échelle entre tickers. + XGBoost/LGB sont invariants à cette transformation, mais elle aide + Ridge et LR (utilisés dans l'ensemble). +- Garman-Klass est centralisé ici pour faciliter les tests unitaires. +""" + import numpy as np import pandas as pd from ta.volatility import AverageTrueRange -from ta.trend import MACD +from ta.trend import MACD as MACDIndicator from const import ATR_WINDOW, MACD_SLOW, MACD_FAST, MACD_SIGN, MIN_HISTORY_TA +# ══════════════════════════════════════════════════════════════════ +# HELPERS +# ══════════════════════════════════════════════════════════════════ + def _safe_normalize(series: pd.Series) -> pd.Series: - """Z-score normalization; returns NaN series when std is zero.""" + """ + Z-score normalization. + Retourne une série de NaN si std == 0 (évite la division par zéro). + """ std = series.std() - if std == 0: + if std == 0 or np.isnan(std): return pd.Series(np.nan, index=series.index) return series.sub(series.mean()).div(std) +def _get_close(stock_data: pd.DataFrame) -> pd.Series: + """ + Retourne adj close si disponible, sinon close. + Garantit la cohérence entre les indicateurs du pipeline. + """ + if "adj close" in stock_data.columns: + return stock_data["adj close"] + if "close" in stock_data.columns: + return stock_data["close"] + raise KeyError("Ni 'adj close' ni 'close' trouvés dans le DataFrame.") + + +# ══════════════════════════════════════════════════════════════════ +# INDICATEURS TECHNIQUES +# ══════════════════════════════════════════════════════════════════ + def compute_atr(stock_data: pd.DataFrame) -> pd.Series: """ - Normalized Average True Range for a single ticker slice. - Expects columns: high, low, close. - Returns a z-scored ATR series aligned to stock_data.index. + Average True Range normalisé (z-score) pour un single ticker. + + Utilise adj close pour la cohérence avec le reste du pipeline. + Nécessite les colonnes : high, low, adj close (ou close). + + Parameters + ---------- + stock_data : pd.DataFrame — slice single ticker + + Returns + ------- + pd.Series — ATR z-scoré, aligné sur stock_data.index """ + required = {"high", "low"} + if not required.issubset(stock_data.columns): + return pd.Series(np.nan, index=stock_data.index) + if len(stock_data) < ATR_WINDOW + 1: return pd.Series(np.nan, index=stock_data.index) + close = _get_close(stock_data) + atr = AverageTrueRange( high=stock_data["high"], low=stock_data["low"], - close=stock_data["close"], + close=close, window=ATR_WINDOW, ).average_true_range() @@ -35,18 +90,91 @@ def compute_atr(stock_data: pd.DataFrame) -> pd.Series: def compute_macd(stock_data: pd.DataFrame) -> pd.Series: """ - Normalized MACD line for a single ticker slice. - Expects column: adj close. - Returns a z-scored MACD series aligned to stock_data.index. + Ligne MACD normalisée (z-score) pour un single ticker. + + Utilise adj close. + Nécessite la colonne : adj close (ou close). + + Parameters + ---------- + stock_data : pd.DataFrame — slice single ticker + + Returns + ------- + pd.Series — MACD z-scoré, aligné sur stock_data.index """ if len(stock_data) < MACD_SLOW + MIN_HISTORY_TA: return pd.Series(np.nan, index=stock_data.index) - macd_val = MACD( - close=stock_data["adj close"], + close = _get_close(stock_data) + + macd_val = MACDIndicator( + close=close, window_slow=MACD_SLOW, window_fast=MACD_FAST, window_sign=MACD_SIGN, ).macd() - return _safe_normalize(macd_val) \ No newline at end of file + return _safe_normalize(macd_val) + + +def compute_garman_klass_vol(stock_data: pd.DataFrame) -> pd.Series: + """ + Volatilité Garman-Klass pour un single ticker. + + Estimateur de volatilité plus efficace que la volatilité close-to-close. + Formule : GK = 0.5 * ln(H/L)² - (2*ln(2)-1) * ln(C/O)² + + Nécessite les colonnes : high, low, adj close (ou close), open. + + Parameters + ---------- + stock_data : pd.DataFrame — slice single ticker + + Returns + ------- + pd.Series — volatilité GK, alignée sur stock_data.index + """ + required = {"high", "low", "open"} + if not required.issubset(stock_data.columns): + return pd.Series(np.nan, index=stock_data.index) + + close = _get_close(stock_data) + + log_hl = np.log(stock_data["high"]) - np.log(stock_data["low"]) + log_co = np.log(close) - np.log(stock_data["open"]) + + gk = 0.5 * log_hl ** 2 - (2 * np.log(2) - 1) * log_co ** 2 + + return gk + + +def compute_macd_histogram(stock_data: pd.DataFrame) -> pd.Series: + """ + Histogramme MACD normalisé (MACD line - Signal line). + + Plus réactif que la ligne MACD seule pour détecter les retournements. + + Parameters + ---------- + stock_data : pd.DataFrame — slice single ticker + + Returns + ------- + pd.Series — histogramme MACD z-scoré + """ + if len(stock_data) < MACD_SLOW + MIN_HISTORY_TA: + return pd.Series(np.nan, index=stock_data.index) + + close = _get_close(stock_data) + + indicator = MACDIndicator( + close=close, + window_slow=MACD_SLOW, + window_fast=MACD_FAST, + window_sign=MACD_SIGN, + ) + + histogram = indicator.macd_diff() + + return _safe_normalize(histogram) diff --git a/src/utils/market_utils.py b/src/utils/market_utils.py index 988328e..11e4fc3 100644 --- a/src/utils/market_utils.py +++ b/src/utils/market_utils.py @@ -1,11 +1,43 @@ +""" +Market Utils +============= +Utilitaires pour la récupération du benchmark et l'export des signaux. + +Fonctions : + - get_benchmark_returns() : télécharge et reindex les rendements du benchmark + - build_export_df() : formate le snapshot journalier pour l'export HF +""" + +import time import pandas as pd import yfinance as yf - from src.utils.logger import setup_logger logger = setup_logger("market_utils") +_BENCHMARK_RETRIES = 3 +_BENCHMARK_RETRY_WAIT = 5 + + +# ══════════════════════════════════════════════════════════════════ +# HELPERS +# ══════════════════════════════════════════════════════════════════ + +def _strip_timezone(index: pd.DatetimeIndex) -> pd.DatetimeIndex: + """ + Supprime le timezone d'un DatetimeIndex de manière sécurisée. + - Si timezone-aware → convertit en UTC puis supprime le tz + - Si timezone-naive → retourne tel quel (pas de crash) + """ + if index.tz is not None: + return index.tz_convert("UTC").tz_localize(None) + return index + + +# ══════════════════════════════════════════════════════════════════ +# BENCHMARK +# ══════════════════════════════════════════════════════════════════ def get_benchmark_returns( benchmark_ticker: str, @@ -14,67 +46,159 @@ def get_benchmark_returns( reindex_to: pd.DatetimeIndex, ) -> pd.Series: """ - Downloads benchmark daily returns and reindexes to the strategy calendar. - Falls back to the mean of reindex_to index returns on failure. + Télécharge les rendements journaliers du benchmark et les reindex + sur le calendrier de la stratégie. + + Fallback progressif : + 1. Retry 3 fois avec 5s d'attente + 2. Si échec total → série de zéros + warning + + Parameters + ---------- + benchmark_ticker : str — ticker yfinance (ex: '^FCHI' pour CAC40) + start : pd.Timestamp — date de début + end : pd.Timestamp — date de fin + reindex_to : pd.DatetimeIndex — calendrier de la stratégie + + Returns + ------- + pd.Series — rendements journaliers reindexés, NaN → 0 """ - try: - raw = yf.download( - benchmark_ticker, - start=start, - end=end + pd.DateOffset(days=1), - progress=False, - auto_adjust=False, - ) - - if raw.empty: - logger.error(f"No data found for benchmark {benchmark_ticker}") - return pd.Series(0.0, index=reindex_to) - - if isinstance(raw.columns, pd.MultiIndex): - prices = raw["Close"].iloc[:, 0] - else: - prices = raw["Close"] - - prices.index = prices.index.tz_localize(None) - reindex_to = reindex_to.tz_localize(None) - - bench_returns = ( - prices.pct_change() - .reindex(reindex_to, method="ffill") - .fillna(0) - ) - - return bench_returns - - except Exception as e: - logger.error(f"Benchmark processing failed: {e}") - return pd.Series(0.0, index=reindex_to) + # Normaliser le calendrier cible une seule fois + reindex_clean = _strip_timezone(reindex_to) + + for attempt in range(1, _BENCHMARK_RETRIES + 1): + try: + raw = yf.download( + benchmark_ticker, + start=start, + end=end + pd.DateOffset(days=1), + progress=False, + auto_adjust=False, + threads=False, + ) + + if raw.empty: + logger.warning( + f"Réponse vide pour {benchmark_ticker} " + f"(tentative {attempt}/{_BENCHMARK_RETRIES})" + ) + time.sleep(_BENCHMARK_RETRY_WAIT) + continue + + # Extraction du prix de clôture + if isinstance(raw.columns, pd.MultiIndex): + prices = raw["Adj Close"].iloc[:, 0] # Adj Close > Close pour benchmark + else: + prices = raw["Adj Close"] if "Adj Close" in raw.columns else raw["Close"] + + # Normalisation timezone + prices.index = _strip_timezone(prices.index) + + # Rendements journaliers reindexés sur le calendrier stratégie + bench_returns = ( + prices.pct_change() + .reindex(reindex_clean, method="ffill") + .fillna(0.0) + ) + + logger.info( + f" Benchmark {benchmark_ticker} chargé : " + f"{len(bench_returns)} jours | " + f"Rendement total : {(1 + bench_returns).prod() - 1:.2%}" + ) + return bench_returns + + except Exception as e: + logger.warning( + f"Erreur benchmark {benchmark_ticker} " + f"(tentative {attempt}/{_BENCHMARK_RETRIES}) : {e}" + ) + if attempt < _BENCHMARK_RETRIES: + time.sleep(_BENCHMARK_RETRY_WAIT) + + logger.error( + f" Impossible de charger le benchmark {benchmark_ticker} " + f"après {_BENCHMARK_RETRIES} tentatives. Série de zéros utilisée." + ) + return pd.Series(0.0, index=reindex_clean) + + +# ══════════════════════════════════════════════════════════════════ +# EXPORT DES SIGNAUX +# ══════════════════════════════════════════════════════════════════ def build_export_df( today_data: pd.DataFrame, final_alloc: dict, ) -> pd.DataFrame: """ - Formats the daily signal snapshot into a clean export-ready DataFrame. - Columns: Ticker, Cluster, Proba_Hausse (%), RSI, Return_3M, Allocation, Signal. + Formate le snapshot journalier des signaux ML pour l'export vers HF Space. + + Colonnes de sortie : + Ticker, Proba_Hausse (%), RSI, Return_3M, Allocation (%), Signal + + Compatible avec les deux versions du pipeline : + - Colonnes laggées (rsi_lag1, return_3m_lag1) — nouveau pipeline + - Colonnes directes (rsi, return_3m) — ancien pipeline + + Parameters + ---------- + today_data : pd.DataFrame — slice mensuelle du jour (index = ticker) + final_alloc : dict — {ticker: weight} issu de get_optimal_weights() + + Returns + ------- + pd.DataFrame — propre et prêt pour l'export CSV / Streamlit """ + df = today_data.copy() + + # Résolution des colonnes (laggées en priorité, directes en fallback) + rsi_col = "rsi_lag1" if "rsi_lag1" in df.columns else "rsi" + return3m_col = "return_3m" if "return_3m" in df.columns else None + cluster_col = "cluster_lag1" if "cluster_lag1" in df.columns else "cluster" + + # Colonnes à exporter + cols_to_keep = ["proba_upside"] + if rsi_col: + cols_to_keep.append(rsi_col) + if return3m_col: + cols_to_keep.append(return3m_col) + if cluster_col in df.columns: + cols_to_keep.append(cluster_col) + export = ( - today_data[["cluster", "proba_upside", "rsi", "return_3m"]] + df[cols_to_keep] .reset_index() - .rename( - columns={ - "ticker": "Ticker", - "cluster": "Cluster", - "proba_upside": "Proba_Hausse", - "rsi": "RSI", - "return_3m": "Return_3M", - } - ) + .rename(columns={"ticker": "Ticker"}) ) - export["Proba_Hausse"] = (export["Proba_Hausse"] * 100).round(2) - export["Allocation"] = export["Ticker"].map(final_alloc).fillna(0.0) - export["Signal"] = export["Allocation"].apply( - lambda w: "BUY" if w > 0 else "NEUTRAL" + + # Renommage dynamique + rename_map = {"proba_upside": "Proba_Hausse (%)"} + if rsi_col in export.columns: + rename_map[rsi_col] = "RSI" + if return3m_col and return3m_col in export.columns: + rename_map[return3m_col] = "Return_3M (%)" + if cluster_col in export.columns: + rename_map[cluster_col] = "Cluster" + + export = export.rename(columns=rename_map) + + # Formatage + export["Proba_Hausse (%)"] = (export["Proba_Hausse (%)"] * 100).round(2) + if "Return_3M (%)" in export.columns: + export["Return_3M (%)"] = (export["Return_3M (%)"] * 100).round(2) + + # Allocation & Signal + export["Allocation (%)"] = ( + export["Ticker"].map(final_alloc).fillna(0.0) * 100 + ).round(2) + export["Signal"] = export["Allocation (%)"].apply( + lambda w: "🟢 BUY" if w > 0 else "⚪ NEUTRAL" ) + + # Tri par probabilité décroissante + export = export.sort_values("Proba_Hausse (%)", ascending=False).reset_index(drop=True) + return export diff --git a/src/utils/metrics.py b/src/utils/metrics.py new file mode 100644 index 0000000..b997688 --- /dev/null +++ b/src/utils/metrics.py @@ -0,0 +1,22 @@ +import numpy as np +import pandas as pd + + +def calculate_financial_metrics(df_test: pd.DataFrame, probas: np.ndarray, threshold: float = 0.5) -> dict: + signals = (probas > threshold).astype(int) + strategy_returns = signals * df_test["future_return"] + portfolio_returns = strategy_returns.groupby(level="date").mean() + if portfolio_returns.std() == 0: + return {"sharpe": 0.0, "max_drawdown": 0.0, "total_return": 0.0} + annualization_factor = np.sqrt(12) + sharpe_ratio = (portfolio_returns.mean() / portfolio_returns.std()) * annualization_factor + cumulative_returns = (1 + portfolio_returns).cumprod() + rolling_max = cumulative_returns.cummax() + drawdown = (cumulative_returns - rolling_max) / rolling_max + max_drawdown = drawdown.min() + total_return = cumulative_returns.iloc[-1] - 1 if not cumulative_returns.empty else 0.0 + return { + "sharpe": round(sharpe_ratio, 4), + "max_drawdown": round(max_drawdown, 4), + "total_return": round(total_return, 4) + } diff --git a/tests/get_composition.py b/tests/get_composition.py new file mode 100644 index 0000000..f67a323 --- /dev/null +++ b/tests/get_composition.py @@ -0,0 +1,26 @@ +import pandas as pd +from pathlib import Path + +def show_composition(market_name="CAC40"): + # Chemin vers l'historique de rebalancement + path = Path(f"data/processed/{market_name}/rebalance_history.parquet") + + if not path.exists(): + print("❌ Fichier de rebalancement introuvable. Relance daily_run.py.") + return + + df = pd.read_parquet(path) + + # On prend la dernière ligne disponible (le dernier rebalancement) + latest_weights = df.iloc[-1] + + # On filtre les actions qui ont un poids > 0 + portfolio = latest_weights[latest_weights > 0.001].sort_values(ascending=False) + + print(f"\n--- Composition du Portefeuille {market_name} (Dernière mise à jour) ---") + print(portfolio.to_string(formatters={'Weight': '{:,.2%}'.format})) + print("\nTotal investi : {:.2%}".format(portfolio.sum())) + +if __name__ == "__main__": + # Tu peux changer pour "US_TECH" si besoin + show_composition("CAC40") \ No newline at end of file diff --git a/tests/plot_results.py b/tests/plot_results.py new file mode 100644 index 0000000..2d03c41 --- /dev/null +++ b/tests/plot_results.py @@ -0,0 +1,83 @@ +import pandas as pd +import plotly.graph_objects as go +from plotly.subplots import make_subplots +from pathlib import Path + +def plot_performance(market_name="CAC40"): + print(f"📊 Génération du graphique pour {market_name}...") + + # 1. Chargement des données + base_dir = Path(f"data/processed/{market_name}") + hist_path = base_dir / 'portfolio_history.parquet' + + if not hist_path.exists(): + print(f"❌ Erreur : Fichier introuvable ({hist_path}). Lance d'abord le daily_run.py.") + return + + df = pd.read_parquet(hist_path) + + # 2. Rebase à 100 pour une comparaison parfaite + df['Strategy'] = (df['Strategy'] / df['Strategy'].iloc[0]) * 100 + df['Benchmark'] = (df['Benchmark'] / df['Benchmark'].iloc[0]) * 100 + + # 3. Calcul des Drawdowns (pertes maximales depuis le sommet) + df['Strat_Peak'] = df['Strategy'].cummax() + df['Strat_DD'] = (df['Strategy'] - df['Strat_Peak']) / df['Strat_Peak'] * 100 + + df['Bench_Peak'] = df['Benchmark'].cummax() + df['Bench_DD'] = (df['Benchmark'] - df['Bench_Peak']) / df['Bench_Peak'] * 100 + + # 4. Création de la figure (2 sous-graphiques) + fig = make_subplots( + rows=2, cols=1, + shared_xaxes=True, + vertical_spacing=0.03, + row_heights=[0.7, 0.3], + subplot_titles=("Croissance du Capital (Base 100)", "Drawdown (%)") + ) + + # --- LIGNE 1 : PERFORMANCE --- + fig.add_trace( + go.Scatter(x=df.index, y=df['Strategy'], name='AlphaEdge (Stratégie)', + line=dict(color='#00d2ff', width=2)), + row=1, col=1 + ) + fig.add_trace( + go.Scatter(x=df.index, y=df['Benchmark'], name='Benchmark (^FCHI)', + line=dict(color='#888888', width=2)), + row=1, col=1 + ) + + # --- LIGNE 2 : DRAWDOWN --- + fig.add_trace( + go.Scatter(x=df.index, y=df['Strat_DD'], name='DD Stratégie', + fill='tozeroy', line=dict(color='#ff4b4b', width=1)), + row=2, col=1 + ) + fig.add_trace( + go.Scatter(x=df.index, y=df['Bench_DD'], name='DD Benchmark', + fill='tozeroy', line=dict(color='#888888', width=1, dash='dot')), + row=2, col=1 + ) + + # 5. Esthétique et Layout + final_strat = round(df['Strategy'].iloc[-1], 2) + final_bench = round(df['Benchmark'].iloc[-1], 2) + + fig.update_layout( + title=f"AlphaEdge vs {market_name} | Final: Stratégie {final_strat}€ vs Bench {final_bench}€", + template="plotly_dark", + hovermode="x unified", + height=800, + legend=dict(yanchor="top", y=0.99, xanchor="left", x=0.01), + margin=dict(l=40, r=40, t=60, b=40) + ) + + fig.update_yaxes(title_text="Valeur (€)", row=1, col=1) + fig.update_yaxes(title_text="Drawdown (%)", row=2, col=1) + + # AFFICHER LE GRAPHIQUE DIRECTEMENT + fig.show() + +if __name__ == "__main__": + plot_performance("CAC40") \ No newline at end of file diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..2021a7a --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,10 @@ +# test_pipeline.py +from daily_run import run_pipeline +import json + +# Charge un seul marché pour tester +with open("config/markets/CAC40.json", "r") as f: + config = json.load(f) + +# Lance le pipeline sans uploader (tu peux commenter la ligne upload_to_hf dans daily_run) +run_pipeline(config) \ No newline at end of file