import os from datetime import datetime, timedelta from typing import Tuple, List, Optional import numpy as np import pandas as pd try: import yfinance as yf except Exception: yf = None from sklearn.preprocessing import StandardScaler FX_SUFFIX = "=X" SYMBOL_MAP = { # Indices "US30": "^DJI", "DJI": "^DJI", "US500": "^GSPC", "SPX500": "^GSPC", "SPX": "^GSPC", "NAS100": "^IXIC", "NDX": "^NDX", "GER40": "^GDAXI", "DE40": "^GDAXI", "UK100": "^FTSE", "FR40": "^FCHI", "JP225": "^N225", # Crypto "BTCUSD": "BTC-USD", "ETHUSD": "ETH-USD", # Metals FX-style still use =X on Yahoo "XAUUSD": "XAUUSD=X", "XAGUSD": "XAGUSD=X", } def map_symbol_to_yahoo(symbol: str) -> str: # Map common indices/crypto, then fallback to FX mapping (EURUSD -> EURUSD=X) sym = symbol.replace("/", "").upper() if sym in SYMBOL_MAP: return SYMBOL_MAP[sym] if sym.endswith(FX_SUFFIX): return sym return f"{sym}{FX_SUFFIX}" def _tf_to_interval_and_freq(tf_minutes: Optional[int]) -> Tuple[str, str]: """Map MT5 timeframe (minutes) to yfinance interval and pandas freq.""" if tf_minutes is None or tf_minutes <= 0: return "1d", "1D" if tf_minutes <= 1: return "1m", "1min" if tf_minutes <= 5: return "5m", "5min" if tf_minutes <= 15: return "15m", "15min" if tf_minutes <= 30: return "30m", "30min" if tf_minutes <= 60: return "60m", "60min" if tf_minutes <= 240: return "60m", f"{tf_minutes}min" return "1d", "1D" def enrich_with_yahoo(df: pd.DataFrame, symbol_col: str = "symbol", time_col: str = "close_time", target_timeframe_minutes: Optional[int] = None, tz: str = "UTC", fill_holidays: bool = True) -> pd.DataFrame: if symbol_col not in df.columns: # fallback: try 'entry_symbol' or assume single symbol return df if time_col not in df.columns: # try alternate column names for alt in ("time", "timestamp", "entry_time"): if alt in df.columns: time_col = alt break if time_col not in df.columns: return df # If yfinance is unavailable, skip enrichment gracefully if yf is None: return df df = df.copy() df[time_col] = pd.to_datetime(df[time_col], utc=True, errors="coerce") if tz: # Normalize to UTC for stable asof joins try: df[time_col] = df[time_col].dt.tz_convert("UTC") except Exception: # If tz-naive after parsing, localize then convert try: df[time_col] = df[time_col].dt.tz_localize("UTC") except Exception: pass # For each symbol, fetch yahoo OHLCV over covered date range enriched_frames: List[pd.DataFrame] = [] for sym, g in df.groupby(symbol_col): start = g[time_col].min() - timedelta(days=5) end = g[time_col].max() + timedelta(days=2) ysym = map_symbol_to_yahoo(sym) interval, freq = _tf_to_interval_and_freq(target_timeframe_minutes) try: y = yf.download(ysym, start=start.tz_convert("UTC").to_pydatetime(), end=end.tz_convert("UTC").to_pydatetime(), interval=interval) except Exception: y = pd.DataFrame() if y is None or y.empty: enriched_frames.append(g) continue # Flatten potential MultiIndex columns from yfinance if isinstance(y.columns, pd.MultiIndex): y.columns = ["_".join([str(c) for c in tup if c is not None and str(c) != ""]).strip("_") for tup in y.columns] # Standardize column names (case-insensitive) rename_map = {} for c in list(y.columns): cl = str(c).strip().lower() if cl == "open": rename_map[c] = "y_open" elif cl == "high": rename_map[c] = "y_high" elif cl == "low": rename_map[c] = "y_low" elif cl == "close": rename_map[c] = "y_close" elif cl in ("adj close", "adjclose", "adj_close"): rename_map[c] = "y_adj_close" elif cl == "volume": rename_map[c] = "y_volume" if rename_map: y = y.rename(columns=rename_map) # Ensure datetime index and then expose as a normal column in UTC if not isinstance(y.index, pd.DatetimeIndex): try: y.index = pd.to_datetime(y.index, utc=True) except Exception: y.index = pd.to_datetime(y.index, utc=True, errors="coerce") else: if y.index.tz is None: y.index = y.index.tz_localize("UTC") else: y.index = y.index.tz_convert("UTC") if fill_holidays and freq: # Reindex to continuous timeline at target frequency and ffill to bridge holidays/weekends try: full_idx = pd.date_range(start=y.index.min(), end=y.index.max(), freq=freq, tz="UTC") y = y.reindex(full_idx).ffill() except Exception: pass y = y.reset_index() # The reset index column is commonly named 'Date' or 'Datetime' for yfinance if "Date" in y.columns: y = y.rename(columns={"Date": time_col}) elif "Datetime" in y.columns: y = y.rename(columns={"Datetime": time_col}) else: # If the datetime column got a different name, coerce to target name y.columns = [time_col if str(c).lower() in ("date", "datetime", "index") else c for c in y.columns] # Ensure y_close exists (robust, case-insensitive) if "y_close" not in y.columns: lower_map = {str(c).lower(): c for c in y.columns} for cand in ("y_close", "close", "adj close", "adj_close", "y_adj_close"): if cand in lower_map: y["y_close"] = y[lower_map[cand]] break if "y_close" not in y.columns: # If still missing, create NaNs to keep pipeline running y["y_close"] = np.nan # Derived features y["y_ret_1d"] = y["y_close"].pct_change(1) y["y_vol_5"] = y["y_ret_1d"].rolling(5).std().fillna(0.0) y["y_vol_20"] = y["y_ret_1d"].rolling(20).std().fillna(0.0) # Ensure left frame doesn't carry a time index level gg = g.copy() if isinstance(gg.index, pd.MultiIndex) or (gg.index.name == time_col): gg = gg.reset_index() # Make sure no MultiIndex columns remain if isinstance(gg.columns, pd.MultiIndex): gg.columns = ["_".join([str(c) for c in tup if c is not None and str(c) != ""]).strip("_") for tup in gg.columns] if isinstance(y.columns, pd.MultiIndex): y.columns = ["_".join([str(c) for c in tup if c is not None and str(c) != ""]).strip("_") for tup in y.columns] merged = pd.merge_asof( gg.sort_values(time_col), y.sort_values(time_col), on=time_col, direction="backward", tolerance=pd.Timedelta(minutes=target_timeframe_minutes) if target_timeframe_minutes else None, ) enriched_frames.append(merged) return pd.concat(enriched_frames, ignore_index=True) # Exact order shared with Include/CMarketModelFeatures.mqh. Python diagnostics # may consume this contract, but must not infer inputs from every numeric CSV # column (which previously admitted close prices and realized profit). MARKET_BASE_FEATURES = ( "mm_signed_logret1", "mm_signed_logret3", "mm_signed_logret12", "mm_signed_zscore12", "mm_return_std12", "mm_range", "mm_signed_body", "mm_spread", "mm_hour_sin", "mm_hour_cos", "mm_direction", "mm_atr14", ) def build_feature_matrix(df: pd.DataFrame, drop_cols: Tuple[str, ...] = (), *, feature_names: Optional[List[str]] = None) -> Tuple[np.ndarray, List[str]]: """Read only the explicit shared causal contract; never guess predictors. The supported trainer remains Build-MarketModel.ps1. This helper is for inspecting dense parity data, not rebuilding post-trade snapshot features. """ if feature_names is None: raise ValueError("Explicit shared market feature_names metadata is required; use Build-MarketModel.ps1 for training") names = list(feature_names) if tuple(names[:12]) != MARKET_BASE_FEATURES or (len(names) - 12) % 4: raise ValueError("Feature names do not match the shared market feature contract") if len(names) != len(set(names)): raise ValueError("Duplicate model feature names") for offset in range(12, len(names), 4): first = names[offset] if not first.startswith("sym_") or not first.endswith("_flag"): raise ValueError("Invalid symbol-conditioned feature group") prefix = first[:-5] if names[offset:offset + 4] != [prefix + suffix for suffix in ("_flag", "_ret1", "_ret3", "_zscore")]: raise ValueError("Symbol-conditioned features have the wrong order") absent = set(names) - set(df.columns) if absent: raise ValueError(f"Missing observed causal features: {sorted(absent)}") matrix = df.loc[:, names].apply(pd.to_numeric, errors="coerce").to_numpy(dtype=np.float64) if matrix.ndim != 2 or not np.isfinite(matrix).all(): raise ValueError("Unknown or nonfinite causal features cannot be imputed silently") return matrix, names def scale_features(x: np.ndarray, *, scaler: Optional[StandardScaler] = None, fit_training_partition: bool = False) -> Tuple[np.ndarray, StandardScaler]: """Fit only when a caller explicitly supplies the training partition. Validation/test calls must provide the already fitted scaler and leave fit_training_partition false. No full-dataset default fitting remains. """ values = np.asarray(x, dtype=np.float64) if values.ndim != 2 or values.shape[0] == 0 or not np.isfinite(values).all(): raise ValueError("Scaling requires a nonempty finite observed feature matrix") if scaler is None: if not fit_training_partition: raise ValueError("Supply a training-fitted scaler, or explicitly fit_training_partition=True on training rows only") scaler = StandardScaler() return scaler.fit_transform(values), scaler if fit_training_partition: raise ValueError("Refitting an existing scaler is prohibited; validation/test must only transform") if not hasattr(scaler, "mean_") or not hasattr(scaler, "scale_"): raise ValueError("Scaler must already be fitted on training rows") return scaler.transform(values), scaler