dualea/Experts/Advisors/DualEA/ML/features.py

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