| 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
|
