""" Mad Turtle ML Pipeline - Generates/loads OHLCV features for XAUUSD H1 - Trains ensemble models (BUY/SELL sub-models) - Exports to ONNX for MT5 inference via REST bridge """ import os import numpy as np import pandas as pd from pathlib import Path from datetime import datetime, timedelta, timezone import json import onnx import onnxruntime as ort try: from skl2onnx import convert_sklearn from skl2onnx.common.data_types import FloatTensorType from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier, VotingClassifier from sklearn.preprocessing import StandardScaler from sklearn.pipeline import Pipeline from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report, accuracy_score import joblib HAS_SKLEARN = True except Exception as e: HAS_SKLEARN = False SKLEARN_IMPORT_ERROR = str(e) ROOT = Path(__file__).resolve().parent.parent DATA_DIR = ROOT / "data" MODELS_DIR = ROOT / "models" DATA_DIR.mkdir(exist_ok=True) MODELS_DIR.mkdir(exist_ok=True) def generate_synthetic_gold_data(days: int = 2000, seed: int = 42) -> pd.DataFrame: """Generate realistic synthetic XAUUSD H1 data when no real feed is available.""" rng = np.random.default_rng(seed) n = days * 24 base = 1800.0 returns = rng.normal(loc=0.00002, scale=0.0008, size=n) prices = base * np.exp(np.cumsum(returns)) df = pd.DataFrame({"close": prices}) df["open"] = df["close"].shift(1).fillna(base) df["high"] = df[["open", "close"]].max(axis=1) * (1 + np.abs(rng.normal(0, 0.0003, n))) df["low"] = df[["open", "close"]].min(axis=1) * (1 - np.abs(rng.normal(0, 0.0003, n))) df["volume"] = rng.lognormal(mean=10, sigma=1.0, size=n) df.index = pd.date_range(end=datetime.now(timezone.utc), periods=n, freq="h") return df def load_real_data(csv_path: Path) -> pd.DataFrame: """Load OHLCV from CSV (datetime,open,high,low,close,volume).""" if not csv_path.exists(): raise FileNotFoundError(f"Real data CSV not found: {csv_path}") df = pd.read_csv(csv_path, parse_dates=["datetime"]) df.sort_values("datetime", inplace=True) df.reset_index(drop=True, inplace=True) df.set_index("datetime", inplace=True) df.dropna(subset=["open", "high", "low", "close"], inplace=True) if "volume" not in df.columns: df["volume"] = 0.0 df["volume"] = df["volume"].fillna(0) return df def engineer_features(df: pd.DataFrame) -> pd.DataFrame: """Feature set inspired by price-action + momentum + volatility.""" out = df.copy() out["returns_1"] = np.log(out["close"] / out["close"].shift(1)) out["returns_3"] = np.log(out["close"] / out["close"].shift(3)) out["returns_6"] = np.log(out["close"] / out["close"].shift(6)) out["sma_10"] = out["close"].rolling(10).mean() out["sma_20"] = out["close"].rolling(20).mean() out["sma_50"] = out["close"].rolling(50).mean() out["ema_12"] = out["close"].ewm(span=12, adjust=False).mean() out["ema_26"] = out["close"].ewm(span=26, adjust=False).mean() out["macd"] = out["ema_12"] - out["ema_26"] out["macd_signal"] = out["macd"].ewm(span=9, adjust=False).mean() delta = out["close"].diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).rolling(14).mean() rs = gain / (loss + 1e-9) out["rsi_14"] = 100.0 - (100.0 / (1.0 + rs)) tr1 = out["high"] - out["low"] tr2 = (out["high"] - out["close"].shift(1)).abs() tr3 = (out["low"] - out["close"].shift(1)).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) out["atr_14"] = tr.rolling(14).mean() out["atr_pct"] = out["atr_14"] / (out["close"] + 1e-9) out["vol_ratio"] = out["volume"] / (out["volume"].rolling(20).mean() + 1e-9) out["high_low_range"] = (out["high"] - out["low"]) / (out["close"] + 1e-9) out["dist_sma20"] = (out["close"] - out["sma_20"]) / (out["close"] + 1e-9) out.dropna(inplace=True) return out def make_target(df: pd.DataFrame, horizon: int = 3) -> pd.DataFrame: """Multi-class target: 0 = SELL (return < -threshold) 1 = HOLD (return within threshold) 2 = BUY (return > +threshold) """ fwd = np.log(df["close"].shift(-horizon) / df["close"]) thr = fwd.std() * 0.3 target = pd.cut(fwd, bins=[-np.inf, -thr, thr, np.inf], labels=[0, 1, 2]) df["target"] = target df.dropna(subset=["target"], inplace=True) df["target"] = df["target"].astype(int) return df FEATURES = [ "returns_1", "returns_3", "returns_6", "sma_10", "sma_20", "sma_50", "macd", "macd_signal", "rsi_14", "atr_14", "atr_pct", "vol_ratio", "high_low_range", "dist_sma20", ] def build_pipeline(): return Pipeline([ ("scaler", StandardScaler()), ("clf", RandomForestClassifier(n_estimators=300, max_depth=12, random_state=42, n_jobs=-1, class_weight="balanced")), ]) def train_models(df: pd.DataFrame): buy = df[df["target"] == 2].copy() hold = df[df["target"] == 1].copy() sell = df[df["target"] == 0].copy() min_n = min(len(buy), len(hold), len(sell)) if min_n < 200: raise ValueError(f"Not enough samples per class (min={min_n}). Provide more data or reduce horizon.") buy = buy.sample(len(buy), random_state=42) if len(buy) > min_n else buy hold = hold.sample(len(hold), random_state=42) if len(hold) > min_n else hold sell = sell.sample(len(sell), random_state=42) if len(sell) > min_n else sell balanced = pd.concat([buy, hold, sell]).sample(frac=1, random_state=42).reset_index(drop=True) X = balanced[FEATURES].values y = balanced["target"].values X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, stratify=y, random_state=42) pipe = build_pipeline() pipe.fit(X_train, y_train) preds = pipe.predict(X_test) print("Accuracy:", accuracy_score(y_test, preds)) print(classification_report(y_test, preds, target_names=["SELL", "HOLD", "BUY"])) return pipe, FEATURES def export_onnx(model: Pipeline, features: list, path: Path): initial_types = [("float_input", FloatTensorType([None, len(features)]))] onnx_model = convert_sklearn(model, initial_types=initial_types, target_opset=15) onnx.save(onnx_model, str(path)) print(f"Saved ONNX model -> {path}") def save_metadata(meta: dict, path: Path): with open(path, "w") as f: json.dump(meta, f, indent=2) def main(): meta = { "symbol": "XAUUSD", "timeframe": "H1", "features": FEATURES, "target_horizon": 3, "built_at": datetime.now(timezone.utc).isoformat(), "models": {}, } real_csv = DATA_DIR / "xauusd_h1.csv" if HAS_SKLEARN: if real_csv.exists(): print(f"Loading real data from {real_csv}") df = load_real_data(real_csv) else: print("Real data CSV not found, generating synthetic data") df = generate_synthetic_gold_data(days=1500) df = engineer_features(df) df = make_target(df, horizon=3) buy_pipe, feats = train_models(pd.concat([df[df["target"] == 2], df[df["target"] != 2]])) export_onnx(buy_pipe, feats, MODELS_DIR / "xauusd_h1_ensemble.onnx") meta["models"]["ensemble"] = {"features": feats, "path": "models/xauusd_h1_ensemble.onnx"} else: print("sklearn/skl2onnx not available (", SKLEARN_IMPORT_ERROR, ")") print("Falling back to demo ONNX model via build_onnx_raw.py") import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from build_onnx_raw import build_model, save_metadata as save_meta build_model() save_meta() with open(MODELS_DIR / "metadata.json") as f: meta = json.load(f) save_metadata(meta, MODELS_DIR / "metadata.json") print("Done.") if __name__ == "__main__": main()