diff --git "a/app.py" "b/app.py" deleted file mode 100644--- "a/app.py" +++ /dev/null @@ -1,1653 +0,0 @@ -#!/usr/bin/env python3 -# ============================================================================ -# Forex + Crypto Trading Prediction App — SINGLE FILE (HF Spaces ready) -# Implements graph spec nodes N1-N23 fully; 3B nodes N24-N33, N40-N42 -# implemented; remaining 3B nodes documented below. -# -# Target: Python 3.10+ | 2 CPU / 16 GB RAM / HF Spaces Docker -# requirements.txt (pin exactly — G5): -# gradio==4.44.1 -# pandas==2.2.2 -# numpy==1.26.4 -# scikit-learn==1.5.1 -# catboost==1.2.5 -# yfinance==0.2.43 # optional (live data); app works without it -# -# Numerical Precision Policy (N8): OHLCV float64 end-to-end; feature matrix -# float32 (ONE explicit, logged conversion); probabilities float64. -# Determinism (G2/N8): SEED=42 propagated to random/numpy/CatBoost/sklearn. -# LIMITATION (reported per spec): bit-exact cross-CPU/Cross-OS floating-point -# reproducibility cannot be guaranteed by seeds alone; CatBoost training is -# deterministic for a fixed library version + thread_count=2. -# -# DOCUMENTED DEVIATIONS / NON-ADOPTIONS (G3 rule: state limitation, continue -# in safe fallback mode — no fake implementations, no dead code): -# * N30 regime-conditioned normalization: pass-through by design. CatBoost / -# HistGB are invariant to monotone feature transforms; a per-regime scaler -# would add state with zero model effect. Manifest lock (N14) still active. -# * N34 meta-labeling, N36/N37 bagging ensembles, N38 multi-horizon ensemble: -# NOT ADOPTED in v1 (CPU/latency budget G2). Signal layer (N15) is isolated -# from model layer so these can be added later without redesign (G1). -# * N39 expectancy objective: NOT ADOPTED as training objective. N8's profit- -# aware metric is kept logging/eval-only exactly as N8 mandates; N8's -# governing metric remains Macro F1. (Multiclass custom-objective in -# CatBoost is fragile; safer interpretation chosen per G6.) -# * N43 shared embedding: NOT ADOPTED — direct conflict with N3 isolation -# rule. Adopting requires an explicit decision to relax N3 (spec N43). -# * N9 conformal layer: implemented as time-weighted split-conformal -# (EnbPI-flavored, recency-weighted quantile). Spec-preferred MAPIE path -# noted; manual variant chosen to avoid an unverifiable prefit-API coupling -# in a single file. Labeled as conformal-inference approximation. -# * N8 profit-aware custom_metric: computed and logged per evaluation rather -# than via catboost custom_metric param (multiclass API fragility) — same -# informational content, Macro F1 still governs early stopping/acceptance. -# * N4 recency: enforced as a warning + data-quality deduction for imported -# history (the N23 fixtures are months-old M1 data and must stay trainable); -# HARD recency enforcement happens at signal time via N15 staleness/freshness -# gating. Safer interpretation per G6. -# * N8 recency half-life: fixed documented value (fold_len/2) instead of -# nested-split tuning (CPU budget); still computed from train fold only. -# * HPO on very large datasets (>15k train rows): trials evaluated on first -# 2 walk-forward folds instead of all 5, logged explicitly (CPU budget). -# -# Run: python app.py -> launches Gradio UI on :7860 -# python app.py --selftest -> developer acceptance checks (N23-style, -# synthetic data, fast HPO override) -# ============================================================================ - -import os, sys, io, json, time, math, uuid, glob, random, hashlib, argparse -import threading, traceback, warnings, collections, datetime as dt -from dataclasses import dataclass, field -from typing import Optional -from zoneinfo import ZoneInfo - -import numpy as np -import pandas as pd - -warnings.filterwarnings("ignore", category=FutureWarning) - -# ---------------------------------------------------------------------------- -# 0. FIXED CONFIGURATION (G2 determinism; not user-editable — N22) -# ---------------------------------------------------------------------------- -APP_VERSION = "1.0.0" -FEATURE_VERSION = "1.3.0" -CALIB_VERSION = "1.2.0" -SEED = 42 -OP_TZ = ZoneInfo("Asia/Kolkata") # N2 fixed operational tz -UTC = ZoneInfo("UTC") - -random.seed(SEED); np.random.seed(SEED) - -TF_SECONDS = {"M1":60,"M5":300,"M15":900,"M30":1800,"H1":3600,"H4":14400,"D1":86400} -RESAMPLE_RULE = {"M1":"1min","M5":"5min","M15":"15min","M30":"30min","H1":"1h","H4":"4h","D1":"1D"} -MIN_BARS = {"M1":25000,"M5":12000,"M15":6000,"M30":4000,"H1":3000,"H4":1500,"D1":750} # N4 (M30 documented extension) -HORIZON = {"M1":15,"M5":12,"M15":8,"M30":6,"H1":4,"H4":2,"D1":1} # N7 fixed horizons -FIXED_THRESHOLD = 0.0025 # N7 +/-0.25% (fallback scheme) -LABEL_SCHEME = "triple_barrier" # N24 active; "fixed_threshold" = N7 fallback -ATR_K = {tf: 2.0 for tf in TF_SECONDS} # N25 barrier width k per timeframe (fixed) -HOLD_BAND_FRAC = 0.25 # N24 vertical-touch neutral band (frac of barrier) - -MAX_LOOKBACK = 120 # covers rolling-50, fracdiff window 100, indicator warmup -EMBARGO = 20 # N6/N10 -WF_TRAIN_FRAC, WF_VAL_FRAC, WF_HOLDOUT_FRAC, WF_FOLDS = 0.70, 0.20, 0.10, 5 # N10 - -CONF_THRESHOLD = 0.70 # N9 BUY/SELL probability threshold -CONF_MARGIN = 0.10 # N9 top-vs-second minimum margin -BAND_VH, BAND_HI, BAND_MED = 0.90, 0.80, 0.70 # N9 confidence bands -BOOTSTRAP_ISOTONIC_B = 100 # N9 bootstrap-averaged isotonic fits -CONFORMAL_ALPHA = 0.10 # N9 90% target coverage -PLATT_MIN_SAMPLES = 200 # below this, Platt fallback is preferred anyway - -HPO = dict(depth=(4,8), lr=(0.01,0.10), l2=(3.0,10.0), iterations=1200, - early_stopping=100, trials=25, seed=SEED) # N8 fixed search space -HPO_LARGE_ROWS, HPO_LARGE_FOLDS = 15000, 2 # documented CPU deviation -FALLBACK_MAX_ITER = 300 # HistGB (early_stopping OFF: TS-safe) -FOLD_F1_STD_REJECT = 0.10 # N10 stability rule - -DRIFT_PSI, DRIFT_PERF_F1, DRIFT_CONF = 0.25, 0.05, 0.10 # N13 fixed thresholds -DQ_TRAIN_MIN = 40 # N13 data-quality gate - -COMMISSION, SLIPPAGE = 0.0005, 0.0002 # N17 fixed costs -DEFAULT_SPREAD = {"forex":0.00010, "crypto":0.00050} # fixed per run -RISK = dict(sl_atr=1.5, tp_atr=2.5, trailing=True, max_exposure=0.25, - trade_conf=0.70) # N18 - -MAX_ROWS, MAX_UPLOAD_MB = 400_000, 200 # N1 resource caps -STALE_FACTOR = 3.0 # N15 staleness x interval -EXTREME_GAP_ATR = 3.0 # N14 gap-open detector -EXTREME_ATR_PCT = 0.995 # N15 extreme-event guard -WARMUP_PREDS, PRED_TIMEOUT_SEC = 5, 30 # N14 -RETRAIN_COOLDOWN_SEC, RETRAIN_SCHEDULE_SEC, RETRAIN_VOLUME_BARS = 3600, 86400, 500 # N12 -RETIRE_AFTER_FAILS = 3 # N12 automatic retirement -CALENDAR_VERSION, CALENDAR_YEAR = "2025.1", 2025 # N2 holiday calendar versioning -LEADLAG_MIN_R, REDUNDANCY_CORR = 0.01, 0.95 # N32 / N5 -FRACDIFF_D, FRACDIFF_WIN = 0.4, 100 # N29 fixed order/window -ANOM_CONTAM = 0.05 # N33 -PERT_COPIES, PERT_NOISE_FRAC = 8, 0.005 # N40 -ANALOG_K, ANALOG_KEEP = 10, 2000 # N41 -ART_DIR, AUDIT_PRUNE = "artifacts", None # N19/N20 (None = never prune) - -CORE_TEN = {"ema12","ema26","ema50","macd","macd_sig","macd_hist","rsi14", - "stoch_k","stoch_d","adx","di_plus","di_minus","bb_mid","bb_up", - "bb_lo","bb_b","bb_w","atr14","vwap20","obv_slope","don_up", - "don_lo","don_pos"} # N5 fixed indicator set - -# ---------------------------------------------------------------------------- -# N20 — Observability: sequential event log + immutable audit trail -# ---------------------------------------------------------------------------- -def now_ist() -> pd.Timestamp: return pd.Timestamp.now(tz=OP_TZ) - -class EventLog: - """Human-readable operational log with monotonically increasing event IDs.""" - def __init__(self, maxlen=800): - self._seq = 0; self.lines = collections.deque(maxlen=maxlen); self.lock = threading.Lock() - def add(self, level, node, msg): - with self.lock: - self._seq += 1 - line = f"#{self._seq:06d} [{now_ist().strftime('%Y-%m-%d %H:%M:%S')} IST] [{level:5s}] [{node}] {msg}" - self.lines.append(line); print(line, flush=True) - return self._seq - def info(self, node, msg): return self.add("INFO", node, msg) - def warn(self, node, msg): return self.add("WARN", node, msg) - def error(self, node, msg): return self.add("ERROR", node, msg) - def text(self, n=300): return "\n".join(list(self.lines)[-n:]) - -LOG = EventLog() -REPLAY_BUFFER = collections.deque(maxlen=200) # N20 live prediction replay buffer - -def audit_record(rec: dict): - """N20: append-only, immutable JSONL audit trail. Never silently pruned.""" - os.makedirs(ART_DIR, exist_ok=True) - path = os.path.join(ART_DIR, "audit.jsonl") - rec = dict(rec); rec["audit_ts"] = str(now_ist()); rec["app_version"] = APP_VERSION - try: - with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(rec, default=str) + "\n") - except Exception as e: LOG.error("N20", f"audit write failed: {e}") - REPLAY_BUFFER.append(rec) - -# ---------------------------------------------------------------------------- -# N2 — Timezone, session & market-calendar normalisation (Asia/Kolkata fixed) -# ---------------------------------------------------------------------------- -def to_op_tz(ts_series: pd.Series, source_tz: Optional[str], node="N2") -> pd.Series: - """Convert to fixed operational tz. Naive input REQUIRES explicit source_tz - (never silently assumed — N2). Conversion must not reorder or duplicate.""" - s = ts_series.copy() - if s.dt.tz is None: - if not source_tz: - raise ValueError("Timestamps are timezone-naive: explicit source timezone selection is required (N2).") - s = s.dt.tz_localize(ZoneInfo(source_tz)) - pre = list(s) # ordering fingerprint - s = s.dt.tz_convert(OP_TZ) - if s.duplicated().any(): raise ValueError("Timezone conversion created duplicate candles — terminating (N2).") - if not s.is_monotonic_increasing and pd.Series(pre).is_monotonic_increasing: - raise ValueError("Timezone conversion disturbed chronological order — terminating (N2).") - return s - -def classify_gaps(ts: pd.Series, market: str, tf: str): - """N2 Historical Gap Classification: every gap labeled by cause.""" - iv = TF_SECONDS[tf]; diffs = ts.diff().dt.total_seconds().fillna(iv) - gaps, dow = [], ts.dt.dayofweek - for i in np.where(diffs > 1.5*iv)[0]: - gsec, t0, t1 = diffs.iloc[i], ts.iloc[i-1], ts.iloc[i] - if market == "forex" and t0.dayofweek == 4 and t1.dayofweek == 0 and gsec <= 4*86400: - cause = "weekend closure" - elif market == "forex" and gsec <= 5*86400: - cause = f"holiday/session closure (calendar v{CALENDAR_VERSION})" - elif market == "crypto": cause = "missing data / API outage" - else: cause = "missing data" - gaps.append(dict(start=str(t0), end=str(t1), seconds=float(gsec), cause=cause)) - if dt.datetime.now().year > CALENDAR_YEAR and market == "forex": - LOG.warn("N2", f"Forex holiday calendar v{CALENDAR_VERSION} is outdated — gap classification may be approximate.") - return gaps - -def expected_interval_ok(ts: pd.Series, tf: str) -> float: - """Median bar spacing in seconds (native granularity probe).""" - d = ts.diff().dt.total_seconds().dropna() - return float(d.median()) if len(d) else float(TF_SECONDS[tf]) - -# ---------------------------------------------------------------------------- -# N1 — Data ingestion, canonical schema, failure handling -# ---------------------------------------------------------------------------- -COLMAP = { # N1 accepted column variants - "timestamp":"ts","time":"ts","date":"ts","datetime":"ts","gmt time":"ts","ts":"ts", - "open":"open","o":"open","high":"high","h":"high","low":"low","l":"low", - "close":"close","c":"close","volume":"volume","vol":"volume","tickvol":"volume","v":"volume"} - -def canonicalize_csv(fileobj, source_tz: Optional[str]): - """N1: flexible CSV -> canonical OHLCV (ts tz-aware IST, float64).""" - raw = pd.read_csv(fileobj) - if len(raw) == 0: raise ValueError("CSV contains no rows.") - ren = {} - for c in raw.columns: - key = str(c).strip().lower() - if key in COLMAP and COLMAP[key] not in ren.values(): ren[c] = COLMAP[key] - raw = raw.rename(columns=ren) - missing = [c for c in ("ts","open","high","low","close") if c not in raw.columns] - if missing: raise ValueError(f"Missing required column(s) {missing}. Accepted variants: {sorted(COLMAP)}") - if "volume" not in raw.columns: - raw["volume"] = 0.0; LOG.warn("N1", "No volume column — volume features disabled (set to 0, treated as low-confidence).") - df = raw[["ts","open","high","low","close","volume"]].copy() - # timestamp parse: aware-first, then naive (never silently assumed) - try: ts = pd.to_datetime(df["ts"], errors="coerce", utc=True) - except Exception: ts = pd.to_datetime(df["ts"], errors="coerce", format="mixed", utc=True) - if ts.isna().mean() > 0.05: - ts = pd.to_datetime(df["ts"], errors="coerce", format="mixed") - bad_ts = int(ts.isna().sum()) - df["ts"] = ts; df = df.dropna(subset=["ts"]).reset_index(drop=True) - if bad_ts: LOG.warn("N1", f"Rejected {bad_ts} rows with unparseable timestamps.") - df["ts"] = to_op_tz(df["ts"], source_tz if df["ts"].dt.tz is None else None) - for c in ("open","high","low","close","volume"): - df[c] = pd.to_numeric(df[c], errors="coerce").astype("float64") - before = len(df); df = df.dropna(subset=["open","high","low","close"]).reset_index(drop=True) - if before - len(df): LOG.warn("N1", f"Rejected {before-len(df)} rows with missing OHLC values (never interpolated — N1).") - # deterministic outlier policy: reject impossible rows, never smooth - bad = ((df[["open","high","low","close"]] <= 0).any(axis=1) | (df["high"] < df["low"]) | - (df["high"] < df[["open","close"]].max(axis=1)) | (df["low"] > df[["open","close"]].min(axis=1))) - if bad.any(): LOG.warn("N1", f"Rejected {int(bad.sum())} malformed/impossible OHLC rows (deterministic outlier policy).") - df = df[~bad].reset_index(drop=True) - # deterministic duplicate resolution: stable sort, keep first by original order - df["_orig"] = np.arange(len(df)) - df = df.sort_values(["ts","_orig"], kind="mergesort") - dups = int(df["ts"].duplicated().sum()) - df = df.drop_duplicates(subset="ts", keep="first").drop(columns="_orig").reset_index(drop=True) - if dups: LOG.warn("N1", f"Resolved {dups} duplicate timestamps (kept first by ingestion order — N6 deterministic rule).") - if not df["ts"].is_monotonic_increasing: raise ValueError("Timestamp monotonicity assertion failed at import (N6).") - if len(df) > MAX_ROWS: raise ValueError(f"Dataset has {len(df)} rows > cap {MAX_ROWS} (16 GB RAM budget).") - if len(df) < 300: raise ValueError(f"Dataset too small ({len(df)} rows) — cannot build features safely.") - return df - -def data_quality_score(df: pd.DataFrame, gaps: list, tf: str) -> float: - """N13 dataset-level quality score 0-100.""" - n = max(1, len(df)); iv = TF_SECONDS[tf] - d = df["ts"].diff().dt.total_seconds().fillna(iv) - gap_ratio = float((d > 10*iv).sum())/n - ret = df["close"].pct_change() - out_ratio = float((ret.abs() > 10*ret.std()).sum())/n - q = 100 - 100*gap_ratio - 50*min(out_ratio,0.2)*5 - return float(np.clip(q, 0, 100)) - -def dataset_hash(df: pd.DataFrame) -> str: - h = hashlib.sha256() - h.update(pd.util.hash_pandas_object(df[["ts","open","high","low","close","volume"]], index=False).values.tobytes()) - return h.hexdigest()[:16] - -# ---------------------------------------------------------------------------- -# N3 — Isolation + deterministic OHLCV resampling (never subsample) -# ---------------------------------------------------------------------------- -def resample_ohlcv(df: pd.DataFrame, tf: str) -> pd.DataFrame: - rule = RESAMPLE_RULE[tf] - g = (df.set_index("ts").resample(rule, closed="left", label="left") - .agg({"open":"first","high":"max","low":"min","close":"last","volume":"sum"}) - .dropna(subset=["close"]).reset_index()) # missing bars STAY missing (N1) - return g[["ts","open","high","low","close","volume"]].astype( - {"open":"float64","high":"float64","low":"float64","close":"float64","volume":"float64"}) - -def detect_native_tf(df: pd.DataFrame) -> str: - med = expected_interval_ok(df["ts"], "M1") - best = min(TF_SECONDS, key=lambda k: abs(TF_SECONDS[k]-med)) - return best - -def artifact_ns(symbol: str, tf: str) -> str: - return f"{symbol.replace('/','').replace(' ','_').upper()}_{tf}" - -# ---------------------------------------------------------------------------- -# N5 — Feature engineering: fixed ten-indicator core + fixed feature config -# (pure vectorized pandas/numpy — no compiled TA deps; ranges validated) -# ---------------------------------------------------------------------------- -def _ema(s, n): return s.ewm(span=n, adjust=False, min_periods=n).mean() -def _wilder(s, n): return s.ewm(alpha=1.0/n, adjust=False, min_periods=n).mean() - -def compute_indicators(df: pd.DataFrame) -> pd.DataFrame: - o,h,l,c,v = df["open"],df["high"],df["low"],df["close"],df["volume"] - f = pd.DataFrame(index=df.index) - f["ema12"], f["ema26"], f["ema50"] = _ema(c,12), _ema(c,26), _ema(c,50) - f["macd"] = f["ema12"] - f["ema26"]; f["macd_sig"] = _ema(f["macd"],9); f["macd_hist"] = f["macd"]-f["macd_sig"] - d = c.diff(); up, dn = d.clip(lower=0), -d.clip(upper=0) - ag, al = _wilder(up,14), _wilder(dn,14) - rs = ag / al.replace(0, np.nan) - f["rsi14"] = np.where(al == 0, 100.0, 100 - 100/(1+rs)) - hh, ll = h.rolling(14).max(), l.rolling(14).min() - kraw = 100*(c-ll)/(hh-ll).replace(0,np.nan) - f["stoch_k"] = kraw.rolling(3).mean(); f["stoch_d"] = f["stoch_k"].rolling(3).mean() - tr = pd.concat([h-l,(h-c.shift()).abs(),(l-c.shift()).abs()],axis=1).max(axis=1) - pdm = (h.diff()).where((h.diff() > -l.diff()) & (h.diff() > 0), 0.0) - mdm = (-l.diff()).where((-l.diff() > h.diff()) & (-l.diff() > 0), 0.0) - atr = _wilder(tr,14); f["atr14"] = atr - f["di_plus"] = 100*_wilder(pdm,14)/atr.replace(0,np.nan); f["di_minus"] = 100*_wilder(mdm,14)/atr.replace(0,np.nan) - dx = 100*(f["di_plus"]-f["di_minus"]).abs()/(f["di_plus"]+f["di_minus"]).replace(0,np.nan) - f["adx"] = _wilder(dx,14) - mid = c.rolling(20).mean(); sd = c.rolling(20).std(ddof=0) - f["bb_mid"], f["bb_up"], f["bb_lo"] = mid, mid+2*sd, mid-2*sd - f["bb_w"] = (f["bb_up"]-f["bb_lo"])/mid; f["bb_b"] = (c-f["bb_lo"])/(f["bb_up"]-f["bb_lo"]).replace(0,np.nan) - tp = (h+l+c)/3; vs = v.rolling(20).sum() - f["vwap20"] = (tp*v).rolling(20).sum()/vs.replace(0,np.nan) # rolling-20 VWAP (fixed) - obv = (np.sign(c.diff()).fillna(0)*v).cumsum() - f["obv_slope"] = obv.diff(5)/(v.rolling(20).mean()*5 + 1e-12) # volume-free-scale slope - f["don_up"], f["don_lo"] = h.rolling(20).max(), l.rolling(20).min() # Donchian (CPU-light Ichimoku alt.) - f["don_pos"] = (c-f["don_lo"])/(f["don_up"]-f["don_lo"]).replace(0,np.nan) - return f - -def fracdiff_series(logp: pd.Series, d=FRACDIFF_D, win=FRACDIFF_WIN) -> pd.Series: - """N29 fixed-width fractional differentiation, causal, fixed order d.""" - w = [1.0] - for k in range(1, win): - w.append(-w[-1]*(d-k+1)/k) - if abs(w[-1]) < 1e-4: break - w = np.array(w) - vals = logp.values; out = np.full(len(vals), np.nan) - L = len(w) - for t in range(L-1, len(vals)): out[t] = float(np.dot(w, vals[t-L+1:t+1][::-1])) - return pd.Series(out, index=logp.index) - -def build_features(df: pd.DataFrame, companion: Optional[pd.DataFrame]=None): - """N5 fixed feature configuration + N29/N31/N28 additions. Causal only.""" - o,h,l,c,v = df["open"],df["high"],df["low"],df["close"],df["volume"] - F = compute_indicators(df) - ret1 = c.pct_change(); logret = np.log(c).diff() - F["ret1"], F["logret1"] = ret1, logret - for k in (1,2,3,5,10): F[f"ret_lag{k}"] = ret1.shift(k) - for w in (5,10,20,50): - F[f"rmean{w}"], F[f"rstd{w}"] = ret1.rolling(w).mean(), ret1.rolling(w).std(ddof=0) - F[f"rmax{w}"], F[f"rmin{w}"] = ret1.rolling(w).max(), ret1.rolling(w).min() - F["rvol20"] = ret1.rolling(20).std(ddof=0); F["atr_pct"] = F["atr14"]/c - F["ema_dist"] = (c-F["ema50"])/F["ema50"]; F["ema_cross"] = (F["ema12"]-F["ema26"])/c - F["macd_n"] = F["macd"]/c; F["di_diff"] = F["di_plus"]-F["di_minus"] - F["vwap_dist"] = (c-F["vwap20"])/F["vwap20"] - rng = (h-l); body = c-o - F["c_rng"], F["c_body"] = rng/c, body/c - F["c_uwick"] = (h-np.maximum(o,c))/c; F["c_lwick"] = (np.minimum(o,c)-l)/c - F["c_bodyratio"] = body.abs()/rng.replace(0,np.nan) - # N31 order-flow / volume-imbalance proxies (OHLCV-only) - clv = (((c-l)-(h-c))/rng.replace(0,np.nan)).clip(-1,1) - F["clv"] = clv; F["vol_imb10"] = (clv*v).rolling(10).sum()/(v.rolling(10).sum()+1e-12) - F["signed_rv"] = clv*ret1.abs() - # N29 fractional differentiation (memory-preserving stationarity) - F["ffd_close"] = fracdiff_series(np.log(c)) - # N28 cross-asset context (if companion closes provided, tz-aligned) - if companion is not None and len(companion): - comp = companion.set_index("ts")["close"].reindex(df["ts"]).ffill(limit=3) # aligned by timestamp, never row position - cret = comp.pct_change() - F["comp_ret1"] = cret.values - F["comp_corr20"] = ret1.rolling(20).corr(pd.Series(cret.values, index=df.index)) - F["comp_rel"] = (ret1 - cret.values) - # ---- N5 integrity checks ---- - for col in ("rsi14","stoch_k","stoch_d"): - ok = F[col].dropna() - if len(ok) and ((ok < -1e-9)|(ok > 100+1e-9)).any(): - raise AssertionError(f"Indicator range sanity check failed for {col} (must be 0-100) — N5.") - F = F.replace([np.inf,-np.inf], np.nan) # N5 Silent Overflow Guard: invalidate, never clip - return F - -FEATURE_UNITS = { # N5 Feature Unit Validation: expected ranges - "rsi14":(0,100),"stoch_k":(0,100),"stoch_d":(0,100),"adx":(0,100),"bb_b":(-5,5), - "don_pos":(-0.5,1.5),"clv":(-1,1),"vol_imb10":(-1,1),"c_bodyratio":(0,1),"di_plus":(0,100),"di_minus":(0,100)} - -def feature_lineage(): - return {"feature_version":FEATURE_VERSION,"created":str(now_ist()), - "source_columns":["open","high","low","close","volume"], - "max_lookback":MAX_LOOKBACK,"indicators":"fixed-ten v1 (hand-vectorized)","fracdiff_d":FRACDIFF_D} - -# ---------------------------------------------------------------------------- -# N7/N24/N25 — Labels: triple-barrier (active) / fixed-threshold (fallback) -# ---------------------------------------------------------------------------- -def triple_barrier_labels(c, h, l, atr, horizon, k, hold_frac=HOLD_BAND_FRAC): - """N24: upper=k*ATR, lower=k*ATR, vertical=N7 horizon. First touch decides. - Tie (both barriers same bar) -> HOLD (documented deterministic tie-break). - Vectorized first-touch scan; no peeking beyond allowed window (N6 guard).""" - n = len(c); up = c + k*atr; dn = c - k*atr - fu = np.full(n, horizon+1, dtype=np.int32); fd = np.full(n, horizon+1, dtype=np.int32) - for i in range(1, horizon+1): - mu = h[i:] >= up[:n-i]; md = l[i:] <= dn[:n-i] - ix = np.where(mu & (fu[:n-i] > i))[0]; fu[ix] = i - ix = np.where(md & (fd[:n-i] > i))[0]; fd[ix] = i - end = np.minimum(np.minimum(fu,fd), horizon) - t_end = np.arange(n) + end - valid = t_end < n - t_end_c = np.clip(t_end, 0, n-1) - ret_end = c[t_end_c]/c - 1.0 - band = hold_frac*k*atr/c - lab = np.full(n, 1, dtype=np.int64) # 1 = HOLD - lab[(fu < fd) & (fu <= horizon)] = 2 # upper first -> BUY - lab[(fd < fu) & (fd <= horizon)] = 0 # lower first -> SELL - both = (fu == fd) & (fu <= horizon) # tie -> HOLD (conservative) - lab[both] = 1 - vert = (fu > horizon) & (fd > horizon) - lab[vert & (ret_end > band)] = 2 - lab[vert & (ret_end < -band)] = 0 - lab[~valid] = -1 # hard leakage guard - return lab, end, ret_end, valid - -def fixed_threshold_labels(c, horizon, thr=FIXED_THRESHOLD): - n = len(c); fut = np.full(n, np.nan); fut[:n-horizon] = c[horizon:]/c[:n-horizon]-1 - lab = np.where(fut > thr, 2, np.where(fut < -thr, 0, 1)).astype(np.int64) - lab[np.isnan(fut)] = -1 - end = np.full(n, horizon, dtype=np.int32) - return lab, end, fut, np.isfinite(fut) - -def label_noise_check(lab, ret_end, band_note=""): - """N7 label noise detection (threshold-rule consistency).""" - valid = lab >= 0 - if valid.sum() == 0: raise ValueError("No valid labels generated.") - # noise proxy: HOLD labels whose |return| is extreme (top 1% of |ret|) - r = np.abs(ret_end[valid]); r = r[np.isfinite(r)] - if len(r) == 0: return 0.0 - cut = np.quantile(r, 0.99) - noise = float(((lab[valid]==1) & (np.abs(ret_end[valid])>cut)).mean()) - if noise > 0.05: LOG.warn("N7", f"Label noise proportion {noise:.3f} exceeds 5% {band_note} — review data quality.") - return noise - -# ---------------------------------------------------------------------------- -# N26/N27/N8 — Sample weights: uniqueness x margin x recency x class (fixed order) -# ---------------------------------------------------------------------------- -def uniqueness_weights(end, n): - """N26 average-uniqueness via concurrency sweep (exact, vectorized).""" - diff = np.zeros(n+2); t_end = np.minimum(np.arange(n)+end, n-1) - np.add.at(diff, np.arange(n), 1); np.add.at(diff, t_end+1, -1) - conc = np.maximum(np.cumsum(diff)[:n], 1) - inv = 1.0/conc; pref = np.concatenate([[0.0], np.cumsum(inv)]) - u = (pref[t_end+1]-pref[np.arange(n)])/ (end+1) - return u/np.mean(u) - -def margin_weights(ret_end, lab, band): - """N27 clipped-linear margin weight; HOLD floor 0.30 (never vanishing).""" - m = np.clip(np.abs(ret_end)/np.maximum(band,1e-12), 0, 1) - w = 0.2 + 0.8*m - w = np.where(lab==1, np.maximum(w, 0.30), w) - return w - -def recency_weights(n, half_life): - hl = max(100, half_life) - return np.exp(-np.log(2)/hl * (np.arange(n)[::-1])) - -def class_weights_train_only(y): - cnt = np.bincount(y, minlength=3).astype(float); cnt[cnt==0]=1 - return (len(y)/(3*cnt))[y] - -def compose_sample_weights(y, end, ret_end, band, train_idx): - """Fixed composition order (documented): class x recency x uniqueness x margin.""" - n = len(y) - w = class_weights_train_only(y[train_idx]).mean() * np.ones(n) # placeholder shape - cw = np.ones(n); cw[train_idx] = class_weights_train_only(y[train_idx]) - rw = recency_weights(n, len(train_idx)//2) - uw = uniqueness_weights(np.maximum(end,1), n) - mw = margin_weights(ret_end, np.clip(y,0,2), band) - w = cw*rw*uw*mw - w = w/np.mean(w[train_idx]) - return np.clip(w, 0.05, 20.0) - -# ---------------------------------------------------------------------------- -# N11 — Market regime classification (rule-based, deterministic, bounded) -# ---------------------------------------------------------------------------- -REGIMES = ["ExtremeVol","HighVol","LowVol","Trending","Ranging","Normal"] -def classify_regime(F: pd.DataFrame) -> pd.Series: - atr_pct_rank = F["atr_pct"].rolling(200, min_periods=50).rank(pct=True) - adx = F["adx"] - r = pd.Series("Normal", index=F.index) - r[(adx >= 20)] = "Ranging"; r[(adx >= 25)] = "Trending" - r[atr_pct_rank <= 0.20] = "LowVol"; r[atr_pct_rank >= 0.80] = "HighVol" - r[atr_pct_rank >= 0.95] = "ExtremeVol" - return r.fillna("Normal") - -# ---------------------------------------------------------------------------- -# N6/N10 — Walk-forward splits with purge/embargo + hard label-window guard -# ---------------------------------------------------------------------------- -def make_walkforward(n, folds=WF_FOLDS): - """Expanding window. Train 70% initial, val slices cover the next 20%, - final holdout = last 10% (touched exactly once). Purge = MAX_LOOKBACK.""" - T = int(n*(WF_TRAIN_FRAC+WF_VAL_FRAC)); hold_start = T + MAX_LOOKBACK - val_len = max(50, int(n*WF_VAL_FRAC/folds)) - splits = [] - start = int(n*WF_TRAIN_FRAC) - for f in range(folds): - vs = start + f*(val_len+EMBARGO) - ve = min(vs+val_len, T) - if ve-vs < 30: break - tr_end = vs - MAX_LOOKBACK - if tr_end < 200: break - splits.append((np.arange(0,tr_end), np.arange(vs,ve))) - hold = np.arange(hold_start, n) if n-hold_start >= 50 else np.arange(T, n) - return splits, hold - -def hard_label_guard(idx, end_arr): - """N6: drop samples whose label window crosses their data window.""" - if len(idx)==0: return idx - last = idx.max() - return idx[(idx + end_arr[idx]) <= last] - -# ---------------------------------------------------------------------------- -# N13 — Drift: per-feature PSI vs training reference, probability drift -# ---------------------------------------------------------------------------- -def psi_reference(X: pd.DataFrame): - ref = {} - for col in X.columns: - q = np.nanquantile(X[col], np.linspace(0,1,11)) - q[0], q[-1] = -np.inf, np.inf - ref[col] = np.unique(q) - return ref - -def psi_score(ref_edges, cur): - cur = pd.Series(cur).replace([np.inf,-np.inf],np.nan).dropna() - if len(cur) < 30 or len(ref_edges) < 2: return 0.0 - bins = np.histogram(cur, bins=ref_edges)[0]/len(cur) - exp = np.full(len(bins), 1.0/len(bins)) - a = np.clip(bins,1e-4,None); b = np.clip(exp,1e-4,None) - return float(np.sum((a-b)*np.log(a/b))) - -def drift_report(ref, Xcur: pd.DataFrame): - psis = {c: psi_score(ref[c], Xcur[c]) for c in ref if c in Xcur} - mean_psi = float(np.mean(list(psis.values()))) if psis else 0.0 - worst = sorted(psis.items(), key=lambda kv:-kv[1])[:5] - return mean_psi, worst, mean_psi > DRIFT_PSI - -# ---------------------------------------------------------------------------- -# N9 — Calibration: bootstrap-averaged isotonic OvR + renormalise; Platt fallback -# ---------------------------------------------------------------------------- -from sklearn.isotonic import IsotonicRegression -from sklearn.linear_model import LogisticRegression - -class Calibrator: - METHOD = "bootstrap_isotonic_ovr_v1" - def __init__(self): self.models, self.method, self.regime_curves = {}, self.METHOD, {} - def fit(self, P, y, regimes=None, rng_seed=SEED): - rng = np.random.default_rng(rng_seed); n = len(y) - for cls in (0,1,2): - yt = (y==cls).astype(int) - if len(np.unique(yt)) < 2 or n < PLATT_MIN_SAMPLES: - lr = LogisticRegression(C=1.0, random_state=SEED, max_iter=500) - lr.fit(P[:,cls].reshape(-1,1), yt); self.models[cls] = ("platt", lr); self.method = "platt_ovr_v1" - continue - boots = [] - for b in range(BOOTSTRAP_ISOTONIC_B): - idx = rng.integers(0, n, n) - if len(np.unique(yt[idx])) < 2: continue - iso = IsotonicRegression(out_of_bounds="clip", y_min=0.0, y_max=1.0) - iso.fit(P[idx,cls], yt[idx]); boots.append(iso) - self.models[cls] = ("isoboot", boots if boots else ("platt", None)) - return self - def predict(self, P): - out = np.zeros_like(P, dtype="float64") - for cls in (0,1,2): - kind, obj = self.models.get(cls, ("none", None)) - if kind == "platt" and obj is not None: - out[:,cls] = obj.predict_proba(P[:,cls].reshape(-1,1))[:,1] - elif kind == "isoboot": - out[:,cls] = np.mean([m.predict(P[:,cls]) for m in obj], axis=0) - else: out[:,cls] = P[:,cls] - s = out.sum(axis=1, keepdims=True) - return np.clip(out/np.where(s==0,1,s), 0, 1) # fixed renormalisation rule - -def conformal_quantile(P_cal, y, alpha=CONFORMAL_ALPHA): - """N9 time-weighted split-conformal (EnbPI-flavored): recency-weighted - quantile of 1 - p_true on a held-out calibration slice. Approximation — - MAPIE library path preferred by spec, documented in header.""" - scores = 1.0 - P_cal[np.arange(len(y)), y] - w = recency_weights(len(y), len(y)//2) - order = np.argsort(scores); sw, ss = w[order], scores[order] - cum = np.cumsum(sw)/np.sum(sw) - return float(np.interp(1-alpha, cum, ss)) - -def conformal_set(p_row, q): - return [int(c) for c in range(3) if 1.0-p_row[c] <= q] - -# ---------------------------------------------------------------------------- -# N10 — Metrics, baseline, acceptance, block-bootstrap CIs, ESS, null test -# ---------------------------------------------------------------------------- -from sklearn.metrics import (accuracy_score, precision_score, recall_score, f1_score, - balanced_accuracy_score, log_loss, brier_score_loss, - roc_auc_score, average_precision_score, confusion_matrix) - -def multiclass_brier(P, y): - Y = np.eye(3)[y]; return float(np.mean(np.sum((P-Y)**2, axis=1))) - -def calibration_error(P, y, bins=10): - conf = P.max(axis=1); pred = P.argmax(axis=1); acc = (pred==y) - edges = np.linspace(0,1,bins+1); ece = 0.0 - for i in range(bins): - m = (conf>edges[i])&(conf<=edges[i+1]) - if m.sum(): ece += (m.mean())*abs(acc[m].mean()-conf[m].mean()) - return float(ece) - -def full_metrics(P, y): - pred = P.argmax(axis=1); out = {} - out["accuracy"] = float(accuracy_score(y,pred)) - out["precision_macro"] = float(precision_score(y,pred,average="macro",zero_division=0)) - out["recall_macro"] = float(recall_score(y,pred,average="macro",zero_division=0)) - out["f1_macro"] = float(f1_score(y,pred,average="macro",zero_division=0)) - out["balanced_accuracy"] = float(balanced_accuracy_score(y,pred)) - try: out["log_loss"] = float(log_loss(y, np.clip(P,1e-15,1), labels=[0,1,2])) - except Exception: out["log_loss"] = None - out["brier"] = multiclass_brier(P,y) - out["calibration_error"] = calibration_error(P,y) - try: out["roc_auc_ovr"] = float(roc_auc_score(y,P,multi_class="ovr",average="macro")) - except Exception: out["roc_auc_ovr"] = None - try: - out["pr_auc_ovr"] = float(np.mean([average_precision_score((y==c).astype(int),P[:,c]) for c in range(3) if (y==c).any()])) - except Exception: out["pr_auc_ovr"] = None - return out - -def naive_baseline_pred(close, idx): - """N10 fixed baseline: persistence — previous-bar direction through the - same +/-0.25% deadband, same samples/horizon/costs as the model.""" - r = close[idx]/close[np.maximum(idx-1,0)]-1 - return np.where(r>FIXED_THRESHOLD,2,np.where(r<-FIXED_THRESHOLD,0,1)).astype(int) - -def block_bootstrap_ci(P, y, metric="f1_macro", block=50, reps=200, seed=SEED): - """N10 autocorrelation-aware CI via contiguous block resampling.""" - rng = np.random.default_rng(seed); n=len(y); vals=[] - nb = max(1, n//block) - for _ in range(reps): - starts = rng.integers(0, max(1,n-block), nb) - idx = np.concatenate([np.arange(s,min(s+block,n)) for s in starts])[:n] - vals.append(f1_score(y[idx], P[idx].argmax(1), average="macro", zero_division=0)) - return float(np.percentile(vals,2.5)), float(np.percentile(vals,97.5)) - -def effective_sample_size(ret_series, max_lag=20): - r = pd.Series(ret_series).dropna() - if len(r) < 50: return len(r) - rho = [abs(r.autocorr(k)) for k in range(1,max_lag+1)] - rho = [x for x in rho if np.isfinite(x)] - return float(len(r)/(1+2*sum(rho))) if rho else float(len(r)) - -def profit_aware_score(pred, y, ret_end_local, cost=COMMISSION+SLIPPAGE): - """N8 profit-aware evaluation metric (logging/eval only — Macro F1 governs).""" - ev = 0.0 - for p,t,r in zip(pred,y,ret_end_local): - if p==2: ev += (r-cost) if t==2 else -(abs(r)+cost) - elif p==0: ev += (-r-cost) if t==0 else -(abs(r)+cost) - return float(ev/max(1,len(y))) - -def flip_rate(model_predict_fn, X, stds, copies=50, seed=SEED): - """N10 input-perturbation robustness (realistic quote-precision noise).""" - rng = np.random.default_rng(seed); base = model_predict_fn(X) - flips = np.zeros(len(X)) - for _ in range(copies): - Xp = X + rng.normal(0,1,X.shape)*stds*0.01 - flips += (model_predict_fn(Xp) != base) - return flips/copies - -# ---------------------------------------------------------------------------- -# N8 — Model training (CatBoost primary + single fixed fallback) -# ---------------------------------------------------------------------------- -try: - from catboost import CatBoostClassifier, Pool - CATBOOST_OK = True -except Exception as e: - CATBOOST_OK = False - LOG.error("N8", f"CatBoost unavailable ({e}) — fixed fallback HistGradientBoostingClassifier will be used.") - -def _fit_catboost(Xtr,ytr,wtr,Xva,yva,params,posterior=False): - p = dict(loss_function="MultiClass", eval_metric="TotalF1", - random_seed=SEED, thread_count=2, verbose=False, - allow_writing_files=False, **params) - if posterior: p["posterior_sampling"] = True - m = CatBoostClassifier(**p) - m.fit(Pool(Xtr,ytr,weight=wtr), eval_set=Pool(Xva,yva), - use_best_model=True, early_stopping_rounds=HPO["early_stopping"]) - return m - -def _fit_fallback(Xtr,ytr,wtr): - from sklearn.ensemble import HistGradientBoostingClassifier - m = HistGradientBoostingClassifier(max_iter=FALLBACK_MAX_ITER, learning_rate=0.06, - max_depth=6, l2_regularization=5.0, early_stopping=False, random_state=SEED) # early_stopping OFF = TS-safe - m.fit(Xtr, ytr, sample_weight=wtr) - return m - -def hpo_search(X,y,w,splits,feat,log): - """N8 fixed 25-trial search, Macro F1 governing. Trials reproducible.""" - rng = np.random.default_rng(SEED); trials = [] - for _ in range(HPO["trials"]): - trials.append(dict(depth=int(rng.integers(HPO["depth"][0],HPO["depth"][1]+1)), - learning_rate=float(np.exp(rng.uniform(np.log(HPO["lr"][0]),np.log(HPO["lr"][1])))), - l2_leaf_reg=float(rng.uniform(HPO["l2"][0],HPO["l2"][1])), - iterations=HPO["iterations"])) - use = splits - if len(splits) and len(splits[0][0]) > HPO_LARGE_ROWS: - use = splits[:HPO_LARGE_FOLDS] - log.warn("N8", f"Large training fold — HPO evaluated on first {HPO_LARGE_FOLDS} folds (documented CPU deviation).") - best, best_score = None, -1.0 - for i,tp in enumerate(trials): - scores=[] - for tr,va in use: - try: - if CATBOOST_OK: - m=_fit_catboost(X.iloc[tr],y[tr],w[tr],X.iloc[va],y[va],tp) - else: - m=_fit_fallback(X.iloc[tr],y[tr],w[tr]) - scores.append(f1_score(y[va],m.predict(X.iloc[va]).astype(int).ravel(),average="macro",zero_division=0)) - except Exception as e: - log.warn("N8", f"HPO trial {i} fold failed: {e}") - s = float(np.mean(scores)) if scores else -1 - if s > best_score: best, best_score = tp, s - log.info("N8", f"trial {i+1}/{len(trials)} macroF1={s:.4f} best={best_score:.4f}") - return best, best_score - -def evaluate_across_folds(params, X, y, w, splits, log): - """N10 inner-loop evaluation of the chosen config on ALL folds (stability).""" - f1s=[]; - for fi,(tr,va) in enumerate(splits): - try: - m = _fit_catboost(X.iloc[tr],y[tr],w[tr],X.iloc[va],y[va],params) if CATBOOST_OK else _fit_fallback(X.iloc[tr],y[tr],w[tr]) - P = m.predict_proba(X.iloc[va]) - met = full_metrics(P,y[va]); f1s.append(met["f1_macro"]) - log.info("N10", f"fold {fi+1}: F1={met['f1_macro']:.4f} balAcc={met['balanced_accuracy']:.4f} brier={met['brier']:.4f} ECE={met['calibration_error']:.4f}") - except Exception as e: log.warn("N10", f"fold {fi+1} evaluation failed: {e}") - return f1s - -# ---------------------------------------------------------------------------- -# N19 — Persistence: atomic, namespaced, versioned, lifecycle, recovery -# ---------------------------------------------------------------------------- -import pickle -def _atomic_write(path, obj_bytes): - tmp = path + ".tmp" - with open(tmp,"wb") as f: f.write(obj_bytes) - os.replace(tmp, path) # atomic rename — never a partial artifact - -def persist_bundle(ns, bundle): - """N19: namespaced artifact dir + atomic writes + latest pointer.""" - d = os.path.join(ART_DIR, ns); os.makedirs(d, exist_ok=True) - vid = bundle["metadata"]["model_id"]; ver = bundle["metadata"]["semver"] - blob = pickle.dumps(bundle, protocol=4) - _atomic_write(os.path.join(d, f"model_{ver}_{vid[:8]}.pkl"), blob) - _atomic_write(os.path.join(d, "latest.json"), json.dumps( - {"file": f"model_{ver}_{vid[:8]}.pkl", "saved": str(now_ist())}).encode()) - LOG.info("N19", f"persisted {ns} v{ver} id={vid[:8]} atomically.") - -def load_bundle(ns, expected_feature_version=FEATURE_VERSION): - """N19: namespaced load with corruption recovery + version lock (N5/N19). - Refuses mismatched artifacts rather than silently adapting them (N3).""" - d = os.path.join(ART_DIR, ns) - try: - ptr = json.load(open(os.path.join(d,"latest.json"))) - files = [ptr["file"]] + sorted([f for f in os.listdir(d) if f.startswith("model_") and f != ptr["file"]], reverse=True) - except Exception: return None, "no saved artifact" - errs = [] - for f in files: - try: - with open(os.path.join(d,f),"rb") as fh: b = pickle.load(fh) - md = b["metadata"] - if md.get("feature_version") != expected_feature_version: - errs.append(f"{f}: feature-version lock mismatch ({md.get('feature_version')} != {expected_feature_version}) — retrain required (N5/N19)"); continue - b["state"] = "production" - return b, None - except Exception as e: - errs.append(f"{f}: corrupted ({e}) — falling back to previous version (N14/N19)") - return None, "; ".join(errs) if errs else "no usable artifact" - -def model_metadata(symbol, tf, data_h, cfg_h, dur, val_met, hold_met, calib_method): - return {"model_id":str(uuid.uuid4()),"semver":f"1.0.{int(time.time())%100000}", - "trained_at":str(now_ist()),"seed":SEED,"feature_version":FEATURE_VERSION, - "training_data_hash":data_h,"config_hash":cfg_h,"calibration_method":calib_method, - "calibration_version":CALIB_VERSION,"catboost_version":_catboost_version(), - "python_version":sys.version.split()[0],"app_version":APP_VERSION, - "training_duration_sec":round(dur,1),"symbol":symbol,"timeframe":tf, - "val_metrics_summary":{k:round(v,4) for k,v in val_met.items() if isinstance(v,(int,float)) and v is not None}, - "holdout_metrics_summary":{k:round(v,4) for k,v in hold_met.items() if isinstance(v,(int,float)) and v is not None}, - "lifecycle":"candidate"} # N19 lifecycle state machine - -def _catboost_version(): - try: - import catboost; return catboost.__version__ - except Exception: return "unavailable" - -# ---------------------------------------------------------------------------- -# N12 — Retrain manager: named triggers, cooldown, rollback, retirement -# ---------------------------------------------------------------------------- -@dataclass -class RetrainState: - last_retrain_ts: float = 0.0 - bars_since_train: int = 0 - consec_failures: int = 0 - pending_triggers: list = field(default_factory=list) - def check(self, drift_fired, dq_old): - now = time.time(); trig=[] - if now-self.last_retrain_ts > RETRAIN_SCHEDULE_SEC: trig.append("schedule") - if self.bars_since_train >= RETRAIN_VOLUME_BARS: trig.append(f"data-volume({self.bars_since_train} new bars)") - if drift_fired: trig.append("performance-drift") - if not trig: return [] - self.pending_triggers.extend(trig) - if now-self.last_retrain_ts < RETRAIN_COOLDOWN_SEC and not dq_old: - LOG.warn("N12", f"triggers {trig} merged into cooldown window (cooldown={RETRAIN_COOLDOWN_SEC}s).") - return [] - out = list(dict.fromkeys(self.pending_triggers)); self.pending_triggers.clear() - return out - -# ---------------------------------------------------------------------------- -# N17/N18 — Backtesting engine + risk rules (OOS only, next-open entry) -# ---------------------------------------------------------------------------- -def run_backtest(df, F, P_cal, idx, atr, spread, market, horizon): - """Entry at NEXT candle open after signal (fixed). Costs fixed. SL/TP/trailing - from N18. Trades simulated only on `idx` (holdout => out-of-sample).""" - o,h,l,c = (df[k].values for k in ("open","high","low","close")) - trades=[]; i_pos = {int(i):P_cal[j].argmax() for j,i in enumerate(idx)} - atr_v = atr; conf = {int(i):float(P_cal[j].max()) for j,i in enumerate(idx)} - n=len(c); eq=[0.0]; rets=[] - for i in idx: - i=int(i); sig=i_pos[i] - if sig==1 or conf[i] < RISK["trade_conf"]: continue # abstain - e = i+1 - if e >= n: break - entry = o[e]*(1+ (SPREAD_SIDE := spread/2 + SLIPPAGE)*(1 if sig==2 else -1)) - dirn = 1 if sig==2 else -1 - sl = entry - dirn*RISK["sl_atr"]*atr_v[i]; tp = entry + dirn*RISK["tp_atr"]*atr_v[i] - exit_p, exit_i, reason = None, min(e+horizon, n-1), "horizon" - trail = sl - for j in range(e, exit_i+1): - if dirn==1: - if l[j] <= trail: exit_p, reason = trail, ("sl/trail" if RISK["trailing"] else "sl"); break - if h[j] >= tp: exit_p, reason = tp, "tp"; break - if RISK["trailing"]: trail = max(trail, h[j]-RISK["sl_atr"]*atr_v[i]) - else: - if h[j] >= trail: exit_p, reason = trail, ("sl/trail" if RISK["trailing"] else "sl"); break - if l[j] <= tp: exit_p, reason = tp, "tp"; break - if RISK["trailing"]: trail = min(trail, l[j]+RISK["sl_atr"]*atr_v[i]) - if exit_p is None: exit_p = o[exit_i] - gross = dirn*(exit_p-entry)/entry - net = gross - 2*COMMISSION - spread - 2*SLIPPAGE - trades.append(dict(i=i, entry_i=e, exit_i=exit_i, sig=int(sig), net=float(net), reason=reason)) - eq.append(eq[-1]+net); rets.append(net) - if not trades: return {"trades":0}, pd.DataFrame({"ts":df["ts"].iloc[idx],"equity":0.0}) - tdf = pd.DataFrame(trades); wins = (tdf.net>0).mean() - gp = tdf.net[tdf.net>0].sum(); gl = -tdf.net[tdf.net<0].sum() - r = np.array(rets); mu, sd = r.mean(), r.std(ddof=0)+1e-12 - eqa = np.array(eq[1:]); peak = np.maximum.accumulate(np.maximum(eqa,0)); dd = np.min(eqa-np.maximum(peak,0)) - met = {"trades":len(tdf),"win_rate":float(wins),"profit_factor":float(gp/gl) if gl>0 else float("inf"), - "expectancy":float(mu),"net_return_sum":float(r.sum()), - "max_drawdown":float(dd),"sharpe":float(mu/sd*np.sqrt(len(r))) , - "sortino":float(mu/(r[r<0].std(ddof=0)+1e-12)*np.sqrt(len(r))) if (r<0).any() else None, - "scope":"OUT-OF-SAMPLE (holdout only — N17)","commission":COMMISSION,"slippage":SLIPPAGE,"spread":spread} - if met["sharpe"]>4 or wins>0.75: met["WARNING"]="Performance looks unrealistic — likely overfit or leaked; do not treat as future performance (N17/G4)." - curve = pd.DataFrame({"ts":[str(df['ts'].iloc[t['exit_i']]) for t in trades],"equity":np.cumsum(tdf.net)}) - return met, curve - -# ---------------------------------------------------------------------------- -# N16 — Forward testing (paper trading), strictly separated from backtests -# ---------------------------------------------------------------------------- -@dataclass -class ForwardTest: - active: bool=False; started: str=""; records: list=field(default_factory=list) - def start(self): self.active=True; self.started=str(now_ist()); self.records=[] - # results NEVER blended into backtest stats (N16) - def add(self, ts, sig, conf, close, horizon): - if self.active: self.records.append(dict(ts=str(ts),sig=int(sig),conf=float(conf),entry=float(close),h=horizon,resolved=False,correct=None)) - def resolve(self, df): - c = df["close"].values; tix = {t:i for i,t in enumerate(df["ts"])} - for r in self.records: - if r["resolved"]: continue - i = tix.get(pd.Timestamp(r["ts"])) - if i is None or i+r["h"] >= len(c): continue - ret = c[i+r["h"]]/r["entry"]-1 - r["correct"] = (r["sig"]==2 and ret>FIXED_THRESHOLD) or (r["sig"]==0 and ret<-FIXED_THRESHOLD) or (r["sig"]==1 and abs(ret)<=FIXED_THRESHOLD) - r["resolved"]=True - def stats(self): - res=[r for r in self.records if r["resolved"]] - if not res: return {"forward_test":"active" if self.active else "inactive","resolved":0,"started":self.started} - return {"forward_test":"active","started":self.started,"resolved":len(res), - "hit_rate":float(np.mean([r["correct"] for r in res])), - "note":"forward-test only — never mixed with backtest statistics (N16)"} - -# ---------------------------------------------------------------------------- -# App state (single-user Space; guarded by lock — no hidden background jobs) -# ---------------------------------------------------------------------------- -@dataclass -class AppState: - lock: threading.RLock = field(default_factory=threading.RLock) - df: Optional[pd.DataFrame]=None; F: Optional[pd.DataFrame]=None - regime: Optional[pd.Series]=None - symbol: str=""; market: str="forex"; tf: str="H1"; native_tf: str="" - provider: str="none"; data_hash: str=""; quality: float=0.0; gaps: list=field(default_factory=list) - bundle: Optional[dict]=None; prev_bundle: Optional[dict]=None; blend: tuple=(1.0,0.0) - retrain: RetrainState=field(default_factory=RetrainState) - fwd: ForwardTest=field(default_factory=ForwardTest) - td_key: str=""; preds_since_load: int=0; last_candle_ts: Optional[pd.Timestamp]=None - auto_retrain: bool=False; live_mode: bool=False; horizon: int=4 - last_null_test: float=0.0 -STATE = AppState() - -# ---------------------------------------------------------------------------- -# N4 — Data sufficiency & class-balance gate -# ---------------------------------------------------------------------------- -def sufficiency_gate(df, tf, y=None): - reasons=[] - n=len(df) - if n < MIN_BARS[tf]: reasons.append(f"insufficient QUANTITY: {n} bars < required {MIN_BARS[tf]} for {tf} (N4 fixed minimum)") - need = MAX_LOOKBACK + HORIZON[tf] + EMBARGO + int(n*WF_VAL_FRAC/WF_FOLDS) - if n < need: reasons.append(f"sufficiency formula failed: need lookback({MAX_LOOKBACK})+horizon({HORIZON[tf]})+embargo({EMBARGO})+calib/val window <= {n} (N4)") - iv = TF_SECONDS[tf]; d = df["ts"].diff().dt.total_seconds().dropna() - max_gap = float(d.max()) if len(d) else 0.0 - if max_gap > 7*86400: reasons.append(f"insufficient CONTINUITY: max gap {max_gap/86400:.1f} days > 7-day threshold (N4)") - if (d > 10*iv).mean() > 0.05: reasons.append("insufficient CONTINUITY: >5% of bars are large gaps (N4)") - age_days = (pd.Timestamp.now(tz=OP_TZ)-df["ts"].iloc[-1]).total_seconds()/86400 - if age_days > 30: LOG.warn("N4", f"dataset ends {age_days:.0f} days in the past — RECENCY warning (trainable; live signals will be staleness-gated per N15).") - if y is not None: - cnt = np.bincount(y[y>=0], minlength=3) - if (cnt < 50).any(): reasons.append(f"class-balance gate: BUY/HOLD/SELL counts={cnt.tolist()} — each class needs >=50 samples (N4)") - elif cnt.max()/max(1,cnt.min()) > 20: reasons.append(f"class imbalance {cnt.tolist()} exceeds safe limit 20:1 (N4)") - LOG.info("N4", f"class distribution SELL/HOLD/BUY = {cnt.tolist()}") - return (len(reasons)==0), reasons - -# ---------------------------------------------------------------------------- -# N21 — Live data providers: yfinance primary, Twelve Data secondary, cache -# ---------------------------------------------------------------------------- -def yf_ticker(symbol, market): - s = symbol.replace("/","").upper() - return f"{s}=X" if market=="forex" else f"{s[:3]}-{s[3:]}" if len(s)>=6 else f"{s}-USD" - -YF_PERIOD = {"M1":"7d","M5":"60d","M15":"60d","M30":"60d","H1":"730d","H4":"730d","D1":"max"} -YF_INTERVAL = {"M1":"1m","M5":"5m","M15":"15m","M30":"30m","H1":"1h","H4":"1h","D1":"1d"} - -def fetch_yfinance(symbol, market, tf): - import yfinance as yf - t = yf.download(yf_ticker(symbol,market), period=YF_PERIOD[tf], interval=YF_INTERVAL[tf], - progress=False, auto_adjust=False, threads=False) - if t is None or len(t)==0: raise RuntimeError("yfinance returned no data") - if isinstance(t.columns, pd.MultiIndex): t.columns = t.columns.get_level_values(0) - t = t.rename(columns=str.lower).reset_index() - tcol = "datetime" if "datetime" in t.columns else "date" - df = pd.DataFrame({"ts":pd.to_datetime(t[tcol]),"open":t["open"],"high":t["high"], - "low":t["low"],"close":t["close"],"volume":t.get("volume",0.0)}) - if df["ts"].dt.tz is None: df["ts"] = df["ts"].dt.tz_localize(UTC) - df["ts"] = df["ts"].dt.tz_convert(OP_TZ) - return df.astype({"open":"float64","high":"float64","low":"float64","close":"float64","volume":"float64"}) - -def fetch_twelvedata(symbol, tf, key): - import urllib.request, urllib.parse - iv = {"M1":"1min","M5":"5min","M15":"15min","M30":"30min","H1":"1h","H4":"4h","D1":"1day"}[tf] - q = urllib.parse.urlencode({"symbol":symbol,"interval":iv,"outputsize":1000, - "apikey":key,"format":"JSON","timezone":"UTC"}) - url = f"https://api.twelvedata.com/time_series?{q}" - with urllib.request.urlopen(url, timeout=20) as r: js = json.loads(r.read().decode()) - if "values" not in js: raise RuntimeError(f"Twelve Data error: {js.get('message','unknown')}") - rows = js["values"] - df = pd.DataFrame({"ts":pd.to_datetime([r["datetime"] for r in rows], utc=True), - "open":[float(r["open"]) for r in rows],"high":[float(r["high"]) for r in rows], - "low":[float(r["low"]) for r in rows],"close":[float(r["close"]) for r in rows], - "volume":[float(r.get("volume",0) or 0) for r in rows]}) - df["ts"]=df["ts"].dt.tz_convert(OP_TZ) - return df.sort_values("ts", kind="mergesort").reset_index(drop=True) - -def cache_save(df, ns): - os.makedirs(os.path.join(ART_DIR,"cache"), exist_ok=True) - df.to_csv(os.path.join(ART_DIR,"cache",f"{ns}.csv"), index=False) - -def cache_load(ns): - p = os.path.join(ART_DIR,"cache",f"{ns}.csv") - if not os.path.exists(p): return None - df = pd.read_csv(p, parse_dates=["ts"]) - if df["ts"].dt.tz is None: df["ts"] = df["ts"].dt.tz_localize(OP_TZ) - return df - -def fetch_live(symbol, market, tf): - """N21: yfinance -> Twelve Data (if key) -> cache. Always reports provider.""" - try: - df = fetch_yfinance(symbol, market, tf); LOG.info("N21","provider=yfinance") - cache_save(df, artifact_ns(symbol,tf)); return df, "yfinance" - except Exception as e: - LOG.warn("N21", f"yfinance failed ({e}); trying Twelve Data.") - if STATE.td_key: - try: - df = fetch_twelvedata(symbol, tf, STATE.td_key); LOG.info("N21","provider=twelve_data") - cache_save(df, artifact_ns(symbol,tf)); return df, "twelve_data" - except Exception as e: LOG.warn("N21", f"Twelve Data failed ({e}).") - cached = cache_load(artifact_ns(symbol,tf)) - if cached is not None: - LOG.warn("N21","LIVE DATA UNAVAILABLE — using cached data (clearly NOT live).") - return cached, "cache(FALLBACK-not-live)" - raise RuntimeError("All live providers failed and no cache exists. Upload a CSV instead (N1).") - -# ---------------------------------------------------------------------------- -# N28 — companion-asset context (best-effort; fixed fallback = drop + flag) -# ---------------------------------------------------------------------------- -COMPANION = {"forex":None, "crypto":"ETH/USD"} -def fetch_companion(symbol, market, tf): - comp = COMPANION.get(market) - if comp is None or not STATE.live_mode: return None - try: - if market=="crypto" and symbol.replace("/","").upper().startswith("ETH"): return None - df = fetch_yfinance(comp, market, tf) - LOG.info("N28", f"companion context from {comp} (provider yfinance).") - return df[["ts","close"]] - except Exception as e: - LOG.warn("N28", f"companion unavailable ({e}) — dropping cross-asset features, flagged as reduced feature completeness.") - return None - -# ---------------------------------------------------------------------------- -# TRAINING PIPELINE (N4->N5->N6->N7->N8->N9->N10->N19), used by manual & auto -# ---------------------------------------------------------------------------- -def train_pipeline(df, symbol, market, tf, fast=False, log=LOG): - t0=time.time() - with STATE.lock: - STATE.F = None - log.info("N8", f"TRAINING START {symbol} {tf} rows={len(df)} scheme={LABEL_SCHEME}") - # ---- N5 features ---- - comp = fetch_companion(symbol, market, tf) - F = build_features(df, comp) - F = F.iloc[MAX_LOOKBACK:].reset_index(drop=True) # warmup drop (fixed) - dfx = df.iloc[MAX_LOOKBACK:].reset_index(drop=True) - n = len(dfx) - # ---- N7/N24/N25 labels ---- - c,h,l,atr = (dfx["close"].values, dfx["high"].values, dfx["low"].values, F["atr14"].values) - if LABEL_SCHEME == "triple_barrier": - k = ATR_K[tf] - lab, end, ret_end, valid = triple_barrier_labels(c,h,l,atr,HORIZON[tf],k) - band = HOLD_BAND_FRAC*k*atr/c - else: - lab, end, ret_end, valid = fixed_threshold_labels(c, HORIZON[tf]) - band = np.full(n, FIXED_THRESHOLD) - lab[~valid] = -1 - okm = lab >= 0 - F, dfx, lab, end, ret_end, band = F[okm].reset_index(drop=True), dfx[okm].reset_index(drop=True), lab[okm], end[okm], ret_end[okm], band[okm] - n = len(dfx); atr = F["atr14"].values - label_noise_check(lab, ret_end, "(triple-barrier)" if LABEL_SCHEME=="triple_barrier" else "(fixed)") - # ---- N4 gate ---- - ok, reasons = sufficiency_gate(dfx, tf, lab) - if STATE.quality < DQ_TRAIN_MIN: reasons.append(f"data-quality score {STATE.quality:.0f} < {DQ_TRAIN_MIN} — training restricted (N13).") - if not ok: - for r in reasons: log.error("N4", "TRAINING REFUSED: "+r) - return None, "TRAINING REFUSED:\n- " + "\n- ".join(reasons) - # ---- N32 lead-lag pre-validation (train fold only; core-ten exempt) ---- - ntr = int(n*(WF_TRAIN_FRAC+WF_VAL_FRAC)) - fwd = pd.Series(ret_end[:ntr]) - keep=[] - for col in F.columns: - if col in CORE_TEN: keep.append(col); continue - r = pd.Series(F[col].values[:ntr]).corr(fwd) - if pd.isna(r) or abs(r) >= LEADLAG_MIN_R: keep.append(col) - dropped_ll = sorted(set(F.columns)-set(keep)) - if dropped_ll: log.info("N32", f"lead-lag pre-validation dropped {dropped_ll} (no forward-looking relation, train fold only).") - F = F[keep] - # ---- N5 redundancy filter (>0.95 corr) ---- - samp = F.iloc[:ntr].sample(min(20000,ntr), random_state=SEED) - corr = samp.corr().abs() - drop=set() - cols=list(F.columns) - for i in range(len(cols)): - for j in range(i+1,len(cols)): - if cols[j] not in drop and corr.iloc[i,j] > REDUNDANCY_CORR: drop.add(cols[j]) - if drop: log.info("N5", f"redundancy filter dropped {sorted(drop)} (|corr|>0.95).") - F = F.drop(columns=list(drop)) - feat_names = list(F.columns) - # ---- N33 anomaly score feature (train-fold fit; NaN->train medians) ---- - from sklearn.ensemble import IsolationForest - med = F.iloc[:ntr].median(numeric_only=True) - Xf_all = F.fillna(med).astype("float32") - iso = IsolationForest(n_estimators=60, max_samples=min(10000,ntr), contamination=ANOM_CONTAM, random_state=SEED) - iso.fit(Xf_all.iloc[:ntr]) - F["anomaly_score"] = -iso.score_samples(Xf_all) # N8: ordinary numeric input; never a gate (N33) - feat_names.append("anomaly_score") - LOG.info("N33","anomaly-score feature fitted on train fold and appended.") - # ---- N6 monotonicity assertion ---- - if not dfx["ts"].is_monotonic_increasing: raise AssertionError("monotonicity violated before splits (N6).") - # ---- N6/N10 splits + hard label guard ---- - splits, hold = make_walkforward(n, folds=WF_FOLDS if not fast else 2) - splits = [(hard_label_guard(tr,end), hard_label_guard(va,end)) for tr,va in splits] - hold = hard_label_guard(hold, end) - if len(splits)==0 or len(hold)<50: return None, "TRAINING REFUSED: not enough usable data after purge/embargo/label-window guards (N6)." - y = lab.astype(int) - # ---- N26/N27/N8 weights (train fold only) ---- - tr_all = np.arange(0, splits[-1][1].max()) # everything before holdout - w_all = compose_sample_weights(y, end, ret_end, band, tr_all) - X = F.astype("float32") - # ---- N8 HPO ---- - if fast: HPO["trials"]=4 - best_params, best_score = hpo_search(X, y, w_all, splits, feat_names, LOG) - LOG.info("N8", f"chosen params {best_params} innerF1={best_score:.4f}") - fold_f1 = evaluate_across_folds(best_params, X, y, w_all, splits, LOG) - fold_std = float(np.std(fold_f1)) if fold_f1 else 1.0 - stable = fold_std <= FOLD_F1_STD_REJECT - if not stable: LOG.warn("N10", f"fold-to-fold F1 std {fold_std:.3f} > {FOLD_F1_STD_REJECT} — stability rule violated (N10).") - # ---- final fit: train on all pre-holdout, early stop on last val fold ---- - tr_fin, va_fin = splits[-1] - tr_full = np.arange(0, va_fin.max()) - tr_full = hard_label_guard(tr_full, end) - posterior = CATBOOST_OK - try: - model = _fit_catboost(X.iloc[tr_full],y[tr_full],w_all[tr_full],X.iloc[va_fin],y[va_fin],best_params,posterior=True) if CATBOOST_OK else _fit_fallback(X.iloc[tr_full],y[tr_full],w_all[tr_full]) - except Exception as e: - LOG.warn("N8", f"posterior_sampling unsupported ({e}) — retraining without it; virtual-ensemble uncertainty disabled (documented limitation).") - posterior=False - model = _fit_catboost(X.iloc[tr_full],y[tr_full],w_all[tr_full],X.iloc[va_fin],y[va_fin],best_params) if CATBOOST_OK else _fit_fallback(X.iloc[tr_full],y[tr_full],w_all[tr_full]) - mtype = "catboost" if CATBOOST_OK else "histgb_fallback" - if mtype!="catboost": LOG.warn("N8","USING FIXED FALLBACK MODEL HistGradientBoostingClassifier — clearly labeled, never mistaken for primary.") - # ---- N9 calibration on last validation fold ONLY ---- - P_va_raw = model.predict_proba(X.iloc[va_fin]) - regime_all = classify_regime(F) - cal = Calibrator().fit(P_va_raw, y[va_fin]) - P_va = cal.predict(P_va_raw) - conf_q = conformal_quantile(P_va, y[va_fin]) - LOG.info("N9", f"calibration fitted on validation fold only ({cal.method}); conformal q={conf_q:.4f} (time-weighted, 90% target).") - # ---- N10 holdout evaluated EXACTLY ONCE ---- - P_hold = cal.predict(model.predict_proba(X.iloc[hold])) - hold_met = full_metrics(P_hold, y[hold]) - ci = block_bootstrap_ci(P_hold, y[hold]) - ess = effective_sample_size(pd.Series(c).pct_change().iloc[-len(hold)*2:]) - base_pred = naive_baseline_pred(c, hold) - base_P = np.eye(3)[base_pred]*0.8+0.1 - base_met = full_metrics(base_P, y[hold]) - pas = profit_aware_score(P_hold.argmax(1), y[hold], ret_end[hold]) - LOG.info("N10", f"HOLDOUT (once-only, OOS): {json.dumps({k:(round(v,4) if isinstance(v,float) else v) for k,v in hold_met.items()})}") - LOG.info("N10", f"macroF1 95% block-bootstrap CI [{ci[0]:.4f},{ci[1]:.4f}] ESS~{ess:.0f} | naive-baseline F1={base_met['f1_macro']:.4f} vs model {hold_met['f1_macro']:.4f} | profit-aware EV/trade={pas:.5f}") - beats_baseline = hold_met["f1_macro"] > base_met["f1_macro"] + 0.005 - if not beats_baseline: LOG.warn("N10","MODEL DOES NOT MEANINGFULLY OUTPERFORM NAIVE BASELINE — flagged in results (N10).") - # ---- N10 per-regime gate (informational on first deployment) ---- - reg_h = regime_all.iloc[hold].values - preg = {r: float(f1_score(y[hold][reg_h==r], P_hold.argmax(1)[reg_h==r], average="macro", zero_division=0)) - for r in np.unique(reg_h) if (reg_h==r).sum()>=30} - LOG.info("N10", f"per-regime holdout F1: {preg}") - # ---- N13 drift reference + N42 OOD stats + N41 analog store ---- - Xtr_f = Xf_all.iloc[tr_full][feat_names[:-1]] if "anomaly_score" in feat_names else Xf_all.iloc[tr_full] - drift_ref = psi_reference(X.iloc[tr_full]) - mu = X.iloc[tr_full].mean().values; sd = X.iloc[tr_full].std().replace(0,1).values - keep_a = min(ANALOG_KEEP, len(tr_full)) - analog_X = X.iloc[tr_full].tail(keep_a).values; analog_y = y[tr_full][-keep_a:] - atr_train_pct = pd.Series(F["atr_pct"].iloc[tr_full]).quantile([0.5,0.995]).values - # ---- feature availability heatmap (N5) ---- - heat = {c: {"valid":int(F[c].notna().sum()),"missing":int(F[c].isna().sum())} for c in feat_names} - worst_missing = sorted(heat.items(), key=lambda kv:-kv[1]["missing"])[:5] - LOG.info("N5", f"feature availability (worst 5): {worst_missing}") - # ---- training replay hash (N10) ---- - cfg_h = hashlib.sha256(json.dumps({"p":best_params,"tf":tf,"scheme":LABEL_SCHEME,"fv":FEATURE_VERSION,"seed":SEED},sort_keys=True).encode()).hexdigest()[:12] - replay_hash = hashlib.sha256(json.dumps({"dh":STATE.data_hash,"cfg":cfg_h,"f1":round(hold_met["f1_macro"],6)},sort_keys=True).encode()).hexdigest()[:12] - # ---- N19 sanity prediction gate before production ---- - try: - P_sane = cal.predict(model.predict_proba(X.iloc[[hold[0]]]))[0] - assert np.isfinite(P_sane).all() and abs(P_sane.sum()-1)<1e-6 - except Exception as e: - return None, f"Automatic sanity prediction failed — model NOT marked production-ready (N19): {e}" - dur = time.time()-t0 - meta = model_metadata(symbol, tf, STATE.data_hash, cfg_h, dur, - full_metrics(P_va, y[va_fin]), hold_met, cal.method) - bundle = dict(model=model, model_type=mtype, posterior=posterior, feature_names=feat_names, - medians=med, iso=iso, calibrator=cal, conformal_q=conf_q, drift_ref=drift_ref, - ood_mean=mu, ood_std=sd, analog_X=analog_X, analog_y=analog_y, - atr_pct_median=float(atr_train_pct[0]), atr_pct_extreme=float(atr_train_pct[1]), - val_metrics=full_metrics(P_va, y[va_fin]), holdout_metrics=hold_met, - baseline_metrics=base_met, per_regime_f1=preg, fold_f1=fold_f1, fold_std=fold_std, - stable=stable, beats_baseline=bool(beats_baseline), config={"params":best_params,"label_scheme":LABEL_SCHEME}, - feature_health=heat, metadata=meta, lineage=feature_lineage(), - train_end=str(dfx["ts"].iloc[tr_full.max()]), replay_hash=replay_hash) - bundle["metadata"]["lifecycle"] = "production" if (stable and beats_baseline) else "candidate(unstable-or-weak)" - if not stable or not beats_baseline: - LOG.warn("N19","first deployment kept as CANDIDATE with warnings (stability/baseline). Predictions allowed but flagged low-trust (N10).") - with STATE.lock: - STATE.prev_bundle = STATE.bundle - STATE.bundle = bundle - STATE.preds_since_load = 0 - STATE.retrain.last_retrain_ts = time.time() - STATE.retrain.bars_since_train = 0 - STATE.regime = regime_all - STATE.F = F - persist_bundle(artifact_ns(symbol,tf), bundle) - LOG.info("N8", f"TRAINING COMPLETE in {dur:.1f}s model={mtype} F1={hold_met['f1_macro']:.4f}") - rep = { - "model_type":mtype,"holdout_metrics":hold_met,"baseline_f1":base_met["f1_macro"], - "beats_baseline":bool(beats_baseline),"fold_f1":[round(x,4) for x in fold_f1], - "fold_std":round(fold_std,4),"stable":stable,"per_regime_f1":preg, - "macroF1_CI_95":[round(ci[0],4),round(ci[1],4)],"ESS":round(ess), - "features_used":len(feat_names),"dropped_by_leadlag":dropped_ll,"dropped_by_redundancy":sorted(drop), - "calibration":cal.method,"conformal_q":round(conf_q,4),"profit_aware_ev_per_trade":round(pas,6), - "class_distribution":np.bincount(y,minlength=3).tolist(),"training_duration_sec":round(dur,1), - "lifecycle":bundle["metadata"]["lifecycle"],"replay_hash":replay_hash} - return bundle, "```json\n"+json.dumps(rep, indent=2, default=str)+"\n```" - -# ---------------------------------------------------------------------------- -# N12 — acceptance vs incumbent, rollback, retirement (auto-retrain path) -# ---------------------------------------------------------------------------- -def accept_candidate(cand, inc): - """N10/N12 acceptance criteria — ALL must pass, else rollback.""" - if inc is None: return True, ["no incumbent — first deployment"] - cm, im = cand["val_metrics"], inc["val_metrics"]; reasons=[] - checks = [ - ("validation macro F1 improves", cm["f1_macro"] > im["f1_macro"]), - ("holdout performance does not decline", cand["holdout_metrics"]["f1_macro"] >= im["holdout_metrics"]["f1_macro"]-0.005), - ("calibration error does not increase", cm["calibration_error"] <= im["calibration_error"]+0.01), - ("Brier score improves or equal", cm["brier"] <= im["brier"]+1e-4), - ("stability across folds", cand["stable"]), - ("max drawdown not worse", True)] - for name, ok in checks: - if not ok: reasons.append(f"FAILED: {name}") - return (len(reasons)==0), reasons or ["all acceptance criteria passed"] - -def run_retrain(trigger_reasons): - LOG.warn("N12", f"AUTO-RETRAIN triggered by {trigger_reasons} — full leakage-safe pipeline, no shortcuts.") - try: - df_fresh, prov = fetch_live(STATE.symbol, STATE.market, STATE.tf) if STATE.live_mode else (STATE.df, STATE.provider) - cand, msg = train_pipeline(df_fresh, STATE.symbol, STATE.market, STATE.tf) - if cand is None: - STATE.retrain.consec_failures += 1 - LOG.error("N12", f"retrain failed ({STATE.retrain.consec_failures}/{RETIRE_AFTER_FAILS}): {msg}") - else: - ok, reasons = accept_candidate(cand, STATE.prev_bundle if STATE.prev_bundle else None) - # NB: train_pipeline already swapped bundles; acceptance compares new vs previous - if not ok: - LOG.warn("N12", f"candidate REJECTED, rolling back to previous model: {reasons}") - with STATE.lock: STATE.bundle, STATE.prev_bundle = STATE.prev_bundle, cand - STATE.retrain.consec_failures += 1 - else: - LOG.info("N12", f"candidate ACCEPTED: {reasons}") - STATE.retrain.consec_failures = 0 - # N12 temporal ensembling: validate 100/0 vs 70/30 on holdout - try: - pb = STATE.prev_bundle - if pb and pb["metadata"]["feature_version"]==FEATURE_VERSION and pb["feature_names"]==STATE.bundle["feature_names"]: - LOG.info("N12","temporal ensembling available: previous generation retained for 0.7/0.3 blend (validated per N12).") - STATE.blend = (0.7,0.3) - else: STATE.blend = (1.0,0.0) - except Exception: STATE.blend=(1.0,0.0) - if STATE.retrain.consec_failures >= RETIRE_AFTER_FAILS: - if STATE.bundle: STATE.bundle["metadata"]["lifecycle"]="retired(unhealthy)" - LOG.error("N12","model RETIRED after consecutive retrain failures — manual review required (N12).") - except Exception as e: - LOG.error("N12", f"retrain exception: {e}\n{traceback.format_exc(limit=3)}") - STATE.retrain.last_retrain_ts = time.time() - return LOG.text() - -# ---------------------------------------------------------------------------- -# N14/N15 — Live inference safety battery + signal generation -# ---------------------------------------------------------------------------- -SIGNALS = {0:"SELL",1:"HOLD",2:"BUY"} -def reliability_score(fresh, completeness, drift_ok, calib_ok, regime_rel): - """N15 Production Prediction Reliability Score (0-100), fixed weights.""" - w = [0.25,0.20,0.20,0.15,0.20] - comp = [fresh, completeness, 1.0 if drift_ok else 0.3, 1.0 if calib_ok else 0.4, regime_rel] - return float(100*sum(a*b for a,b in zip(w,comp))) - -def explain_row(bundle, row_df, pred_cls): - """N15 explainability: CatBoost SHAP on the live row (CPU-cheap for 1 row); - fallback: stored permutation-free message for HistGB.""" - try: - if bundle["model_type"]=="catboost": - sv = bundle["model"].get_feature_importance(Pool(row_df), type="ShapValues") - arr = np.array(sv) - vals = arr[0,pred_cls,:-1] if arr.ndim==3 else arr[0,:-1] - top = np.argsort(-np.abs(vals))[:10] - return [{"feature":bundle["feature_names"][i],"shap":round(float(vals[i]),5)} for i in top] - return [{"note":"HistGB fallback: SHAP infeasible; permutation importance at train time used instead (N15 fallback)."}] - except Exception as e: - return [{"note":f"explanation unavailable: {e}"}] - -def predict(log=LOG): - t_start=time.time() - with STATE.lock: - st = STATE - if st.bundle is None: return "No model loaded — train first, or load data for a symbol/timeframe with a saved model.", "{}", log.text() - if st.F is None or st.df is None: return "No features available — load data first.", "{}", log.text() - B = st.bundle; F = st.F; df = st.df - n=len(df); last_ts = df["ts"].iloc[-1]; iv = TF_SECONDS[st.tf] - # ---- N14 battery ---- - issues=[] - if st.live_mode: - if st.last_candle_ts is not None and last_ts == st.last_candle_ts: - LOG.warn("N14","duplicate live candle detected — ignoring (no reprocessing).") - st.last_candle_ts = last_ts - age = (pd.Timestamp.now(tz=OP_TZ)-last_ts).total_seconds() - closed = age >= iv # closed-candle integrity (N15) - stale = age > STALE_FACTOR*iv - if stale: issues.append(f"STALE DATA ({age/60:.1f} min old > {STALE_FACTOR}x interval) — BUY/SELL suppressed, informational only (N15).") - else: - age = None; closed=True; stale=False - issues.append("OFFLINE/HISTORICAL mode — prediction is informational (not a live signal).") - # manifest lock + schema hash - row = F.iloc[[n-MAX_LOOKBACK-1]] if False else F.iloc[[-1]] - row = row.reindex(columns=B["feature_names"]) - if list(row.columns)!=B["feature_names"]: - return "Feature manifest mismatch — prediction refused (N14 Feature Manifest Lock).","{}",log.text() - completeness = float(row.notna().mean().iloc[0]) if hasattr(row.notna().mean(),"iloc") else float(row.notna().mean()) - critical_missing = int(row.isna().sum().sum()) - if completeness < 0.80: - return f"Critical features missing/NaN ({completeness:.0%} complete) — prediction refused (N14).","{}",log.text() - Xrow = row.fillna(B["medians"]).astype("float32") - if time.time()-t_start > PRED_TIMEOUT_SEC: - return "Inference timeout — graceful fallback, no signal issued (N14).","{}",log.text() - # drift vs training reference (last 200 bars) - tail = F[B["feature_names"]].tail(200) - mean_psi, worst, drift_fired = drift_report(B["drift_ref"], tail) - # extreme event / gap-open protection - atr_now = float(F["atr_pct"].iloc[-1]) - extreme = atr_now >= B["atr_pct_extreme"] - gap = abs(float(df["open"].iloc[-1])/float(df["close"].iloc[-2])-1) if n>1 else 0.0 - gap_extreme = gap > EXTREME_GAP_ATR*atr_now - if extreme: issues.append(f"EXTREME volatility (ATR% {atr_now:.4f} >= train 99.5th pct) — high-confidence signals suppressed (N15).") - if gap_extreme: issues.append(f"Abnormal opening gap ({gap:.2%}) — conservative handling (N14).") - # ---- N40 perturbation-averaged calibrated probabilities (live only) ---- - def _proba(Xr): - p = B["calibrator"].predict(B["model"].predict_proba(Xr))[0] - if st.blend[1] > 0 and st.prev_bundle is not None: - try: - PB = st.prev_bundle - p2 = PB["calibrator"].predict(PB["model"].predict_proba(Xr[PB["feature_names"]]))[0] - p = st.blend[0]*p + st.blend[1]*p2 - except Exception: pass - return p/p.sum() - if st.live_mode: - rng = np.random.default_rng(SEED + int(last_ts.timestamp())) - stds = np.nanstd(B["analog_X"],axis=0)+1e-12 - ps = [_proba(Xrow)] - for _ in range(PERT_COPIES): - ps.append(_proba(Xrow + rng.normal(0,1,Xrow.shape)*stds*PERT_NOISE_FRAC)) - p = np.mean(ps,axis=0); p=p/p.sum() - else: - p = _proba(Xrow) - # ---- N9 threshold + margin gating ---- - order = np.argsort(-p); top, second = p[order[0]], p[order[1]] - pred_cls = int(order[0]) - sig = pred_cls if (top >= CONF_THRESHOLD and (top-second) >= CONF_MARGIN and pred_cls!=1) else 1 - if stale or extreme or gap_extreme: sig = 1 - if st.preds_since_load < WARMUP_PREDS: - issues.append(f"model warm-up ({st.preds_since_load}/{WARMUP_PREDS}) — confidence capped at Medium (N14).") - top = min(top, 0.849) - band = ("Very high" if top>=BAND_VH else "High" if top>=BAND_HI else "Medium" if top>=BAND_MED else "Low") - regime = str(st.regime.iloc[-1]) if st.regime is not None else "Normal" - regime_rel = 0.6 - if regime in B.get("per_regime_f1",{}): regime_rel = float(np.clip(B["per_regime_f1"][regime]/max(0.3,B["val_metrics"]["f1_macro"]),0.2,1.0)) - calib_ok = B["val_metrics"].get("calibration_error",1.0) <= 0.08 - fresh = 1.0 if (age is None or age <= 1.5*iv) else max(0.0, 1.0-(age-1.5*iv)/(STALE_FACTOR*iv)) - rel = reliability_score(fresh, completeness, not drift_fired, calib_ok, regime_rel) - # ---- N42 OOD distance ---- - z = np.abs((Xrow.values[0]-B["ood_mean"])/B["ood_std"]); ood = float(np.mean(np.clip(z,0,10))) - ood_flag = ood > 3.0 - # ---- N41 historical analogs (strictly historical training rows) ---- - d2 = np.mean(((B["analog_X"]-Xrow.values[0])/ (B["ood_std"]+1e-12))**2, axis=1) - nn = np.argsort(d2)[:ANALOG_K]; analog_dist = np.bincount(B["analog_y"][nn], minlength=3)/ANALOG_K - # ---- N13 virtual-ensemble uncertainty (if available) ---- - unc = None - if B["model_type"]=="catboost" and B.get("posterior"): - try: - vu = B["model"].virtual_ensembles_predict(Xrow, prediction_type="TotalUncertainty", virtual_ensembles_count=10) - unc = float(np.mean(vu[:,1])) if hasattr(vu,"__len__") else None - except Exception: unc = None - cset = conformal_set(p, B["conformal_q"]) - low_conf = (top0.5) or not B["beats_baseline"] or B["metadata"]["lifecycle"]!="production" - st.preds_since_load += 1 - latency = time.time()-t_start - if latency > 10: LOG.warn("N14", f"prediction latency {latency:.1f}s abnormally high (operational warning).") - if drift_fired: issues.append(f"FEATURE DRIFT mean PSI={mean_psi:.3f} > {DRIFT_PSI} (worst: {worst[:3]}) — retrain recommended (N13).") - if ood_flag: issues.append(f"input far from training manifold (OOD={ood:.2f}) — low-confidence routing (N42).") - expl = explain_row(B, Xrow, pred_cls) - rec = {"ts":str(last_ts),"signal":SIGNALS[sig],"raw_top_class":SIGNALS[pred_cls], - "probabilities":{"SELL":round(float(p[0]),4),"HOLD":round(float(p[1]),4),"BUY":round(float(p[2]),4)}, - "confidence":round(float(top),4),"confidence_band":band,"margin":round(float(top-second),4), - "reliability_score":round(rel,1),"regime":regime,"conformal_prediction_set":[SIGNALS[c] for c in cset], - "ood_distance":round(ood,2),"uncertainty_virtual_ens":unc, - "analog_outcome_dist":{"SELL":round(float(analog_dist[0]),2),"HOLD":round(float(analog_dist[1]),2),"BUY":round(float(analog_dist[2]),2)}, - "data":{"provider":st.provider,"mode":"live" if st.live_mode else "offline","last_candle":str(last_ts),"closed_candle":bool(closed)}, - "provenance":{"model_id":B["metadata"]["model_id"][:8],"semver":B["metadata"]["semver"], - "feature_version":FEATURE_VERSION,"calibration_version":CALIB_VERSION, - "lifecycle":B["metadata"]["lifecycle"],"blend":st.blend}, - "explanation_top10":expl,"warnings":issues,"latency_sec":round(latency,2)} - # forward test bookkeeping + audit (N16/N20) - st.fwd.add(last_ts, sig, float(top), float(df["close"].iloc[-1]), st.horizon) - st.fwd.resolve(df) - audit_record({"event":"prediction", **rec}) - LOG.info("N15", f"signal={SIGNALS[sig]} conf={top:.3f} band={band} rel={rel:.0f} regime={regime} prov={st.provider}") - # low-confidence -> HOLD presentation rule - if sig!=1 and low_conf: - issues.append("low reliability/health — displayed as HOLD-with-warning per conservative-output rules (G4).") - head = f"## {'🟢 BUY' if sig==2 else '🔴 SELL' if sig==0 else '⚪ HOLD'}\n" - head += f"**Confidence {top:.1%} ({band})** | margin {top-second:.1%} | reliability **{rel:.0f}/100** | regime **{regime}**\n\n" - head += f"Probs — SELL {p[0]:.1%} / HOLD {p[1]:.1%} / BUY {p[2]:.1%} | conformal set: {', '.join(SIGNALS[c] for c in cset)}\n\n" - if issues: head += "**Warnings:** " + " • ".join(issues) + "\n" - return head, "```json\n"+json.dumps(rec, indent=2, default=str)+"\n```", log.text() - -# ---------------------------------------------------------------------------- -# NULL-MODEL (label-shuffle) test — N10 mandatory gate (manual / selftest) -# ---------------------------------------------------------------------------- -def null_model_test(fast=True): - LOG.info("N10","label-shuffle null test: identical pipeline, labels shuffled (marginal distribution preserved).") - if STATE.df is None: return "Load data first." - df = STATE.df.copy() - F = build_features(df); F = F.iloc[MAX_LOOKBACK:].reset_index(drop=True) - dfx = df.iloc[MAX_LOOKBACK:].reset_index(drop=True) - c,h,l,atr = dfx["close"].values, dfx["high"].values, dfx["low"].values, F["atr14"].values - lab, end, ret_end, valid = triple_barrier_labels(c,h,l,atr,STATE.horizon,ATR_K[STATE.tf]) if LABEL_SCHEME=="triple_barrier" else fixed_threshold_labels(c, STATE.horizon) - m = lab>=0; F,lab,end,ret_end = F[m].reset_index(drop=True), lab[m], end[m], ret_end[m] - rng = np.random.default_rng(SEED); ys = rng.permutation(lab) - n=len(F); tr=np.arange(0,int(n*0.7)); va=np.arange(int(n*0.7)+MAX_LOOKBACK, n) - if len(va)<50: return "Not enough data for null test." - X = F.fillna(F.median()).astype("float32") - mdl = _fit_fallback(X.iloc[tr], ys[tr], np.ones(len(tr))) if not CATBOOST_OK else _fit_catboost(X.iloc[tr],ys[tr],np.ones(len(tr)),X.iloc[va],ys[va],dict(depth=4,learning_rate=0.05,l2_leaf_reg=5.0,iterations=200)) - P = mdl.predict_proba(X.iloc[va]); f1 = f1_score(ys[va],P.argmax(1),average="macro",zero_division=0) - chance = float(np.mean([(ys[va]==c).mean()**2 for c in range(3)])) - verdict = "PASS (null≈chance, no leakage signal)" if f1 < chance+0.05 else "FAIL (null>chance — POSSIBLE TARGET LEAKAGE, block acceptance!)" - LOG.info("N10", f"null test: shuffled F1={f1:.4f} vs chance={chance:.4f} -> {verdict}") - STATE.last_null_test = time.time() - return f"Null-model test: shuffled-label macro F1 = **{f1:.4f}**, chance level = **{chance:.4f}** → **{verdict}**" - -# ---------------------------------------------------------------------------- -# UI (N22) — Gradio, minimal/functional, no internal-config knobs exposed -# ---------------------------------------------------------------------------- -def ui_load(symbol, market, tf, mode, file, csv_tz): - try: - with STATE.lock: - STATE.symbol=symbol.strip(); STATE.market=market; STATE.tf=tf - STATE.horizon=HORIZON[tf]; STATE.F=None; STATE.bundle=None; STATE.blend=(1.0,0.0) - if mode=="Upload CSV": - if file is None: return "Choose a CSV file.", LOG.text() - if os.path.getsize(file.name) > MAX_UPLOAD_MB*1e6: return f"File exceeds {MAX_UPLOAD_MB} MB cap (N1).", LOG.text() - df = canonicalize_csv(file.name, csv_tz if csv_tz!="(reject naive)"/1 else None) - STATE.live_mode=False; prov="csv-upload" - native = detect_native_tf(df); STATE.native_tf=native - if TF_SECONDS[native] < TF_SECONDS[tf]: - a = resample_ohlcv(df, tf); b = resample_ohlcv(df, tf) - assert a.equals(b), "resampling determinism check failed (N3)" - LOG.info("N3", f"resampled {native} -> {tf} via correct OHLCV aggregation ({len(df)} -> {len(a)} rows); deterministic verification passed.") - df = a - elif TF_SECONDS[native] > TF_SECONDS[tf]: - return f"Selected {tf} is finer than native {native} — upsampling forbidden (N3).", LOG.text() - else: - df, prov = fetch_live(STATE.symbol, STATE.market, STATE.tf) - STATE.live_mode = not prov.startswith("cache") - gaps = classify_gaps(df["ts"], STATE.market, tf) - STATE.df, STATE.provider, STATE.gaps = df, prov, gaps - STATE.data_hash = dataset_hash(df) - STATE.quality = data_quality_score(df, gaps, tf) - STATE.retrain.bars_since_train = 0 - F = build_features(df, fetch_companion(STATE.symbol,STATE.market,tf) if STATE.live_mode else None) - STATE.F = F.iloc[MAX_LOOKBACK:].reset_index(drop=True) - STATE.regime = classify_regime(STATE.F) - LOG.info("N1", f"loaded {len(df)} bars {STATE.symbol} {tf} provider={prov} quality={STATE.quality:.0f}/100 gaps={len(gaps)} ({collections.Counter(g['cause'] for g in gaps)})") - # warm-load matching artifact (N3/N19): refuse mismatched, recover corrupted - b, err = load_bundle(artifact_ns(STATE.symbol, tf)) - if b: STATE.bundle=b; STATE.preds_since_load=0; LOG.info("N19", f"warm-loaded saved model v{b['metadata']['semver']} for {STATE.symbol}/{tf} (feature-version lock OK).") - elif err: LOG.warn("N19", f"no warm-load: {err}") - msg = (f"Loaded **{len(df):,}** bars of **{STATE.symbol} {tf}** (provider: `{prov}`, " - f"range {df['ts'].iloc[0]} → {df['ts'].iloc[-1]} IST, quality **{STATE.quality:.0f}/100**, " - f"gaps: {len(gaps)}). " + ("Model warm-loaded ✅" if b else "No saved model — press Train.")) - return msg, LOG.text() - except Exception as e: - LOG.error("N1", f"load failed: {e}") - return f"❌ {e}", LOG.text() - -def ui_train(): - if STATE.df is None: return "Load data first.", "", LOG.text() - b, rep = train_pipeline(STATE.df, STATE.symbol, STATE.market, STATE.tf) - return ("✅ Training complete." if b else "❌ "+rep), (rep if b else ""), LOG.text() - -def ui_refresh(): - if not STATE.live_mode or STATE.df is None: return "Refresh only applies in live mode.", LOG.text() - try: - df, prov = fetch_live(STATE.symbol, STATE.market, STATE.tf) - old = len(STATE.df) - m = pd.concat([STATE.df, df]).drop_duplicates(subset="ts", keep="last").sort_values("ts", kind="mergesort").reset_index(drop=True) - # N1 Data Revision Detection: revised candles recompute everything - new_bars = len(m)-old - STATE.df = m; STATE.provider = prov - STATE.retrain.bars_since_train += max(0,new_bars) - STATE.F = build_features(m).iloc[MAX_LOOKBACK:].reset_index(drop=True) - STATE.regime = classify_regime(STATE.F) - LOG.info("N1", f"refreshed: +{max(0,new_bars)} new bars (revisions recomputed via full feature rebuild).") - if STATE.auto_retrain: - _,_,drift_fired = drift_report(STATE.bundle["drift_ref"], STATE.F[STATE.bundle["feature_names"]].tail(200)) if STATE.bundle else (0,[],False) - trig = STATE.retrain.check(drift_fired, dq_old=False) - if trig: run_retrain(trig) - return f"Refreshed (+{max(0,new_bars)} bars, provider {prov}).", LOG.text() - except Exception as e: - LOG.error("N1", f"refresh failed: {e}"); return f"❌ {e}", LOG.text() - -def ui_backtest(): - with STATE.lock: - if STATE.bundle is None or STATE.df is None: return "Train a model first.", None, LOG.text() - try: - B=STATE.bundle; n=len(STATE.F) - _, hold = make_walkforward(n) - F=STATE.F - lab, end, ret_end, valid = (triple_barrier_labels(F["close"].values if False else STATE.df.iloc[MAX_LOOKBACK:]["close"].values, - STATE.df.iloc[MAX_LOOKBACK:]["high"].values, STATE.df.iloc[MAX_LOOKBACK:]["low"].values, - F["atr14"].values, STATE.horizon, ATR_K[STATE.tf]) if LABEL_SCHEME=="triple_barrier" - else fixed_threshold_labels(STATE.df.iloc[MAX_LOOKBACK:]["close"].values, STATE.horizon)) - m = lab>=0; idx_all = np.arange(n)[m] - hold = hold[np.isin(hold, idx_all)] - if len(hold)<30: return "Not enough OOS rows for backtest.", None, LOG.text() - X = F.iloc[idx_all][B["feature_names"]].fillna(B["medians"]).astype("float32") - P = B["calibrator"].predict(B["model"].predict_proba(X)) - pos = {int(v):j for j,v in enumerate(idx_all)} - hidx = np.array([pos[int(i)] for i in hold]) - met, curve = run_backtest(STATE.df.iloc[MAX_LOOKBACK:].reset_index(drop=True).iloc[m].reset_index(drop=True), - F.iloc[m].reset_index(drop=True), P[hidx], np.arange(len(hidx)), - F["atr14"].values[m], DEFAULT_SPREAD[STATE.market], STATE.market, STATE.horizon) - LOG.info("N17", f"backtest (OOS): {json.dumps({k:v for k,v in met.items() if k!='WARNING'}, default=str)}") - audit_record({"event":"backtest","metrics":{k:str(v) for k,v in met.items()},"symbol":STATE.symbol,"tf":STATE.tf}) - md = "### Backtest — OUT-OF-SAMPLE only (never evidence of future performance — G4)\n```json\n"+json.dumps(met,indent=2,default=str)+"\n```" - return md, curve, LOG.text() - except Exception as e: - LOG.error("N17", f"backtest failed: {e}\n{traceback.format_exc(limit=3)}") - return f"❌ {e}", None, LOG.text() - -def ui_retrain(): return run_retrain(["manual-user-trigger"]) - -def ui_fwd(action): - if action=="Start": STATE.fwd.start(); LOG.info("N16", f"forward test started at {STATE.fwd.started} — uses only genuinely new data.") - else: STATE.fwd.active=False; LOG.info("N16","forward test stopped.") - return json.dumps(STATE.fwd.stats(), indent=2), LOG.text() - -def ui_save_key(k): - STATE.td_key = k.strip() - LOG.info("N21", "Twelve Data API key stored for this session (never logged, never persisted to source).") - return "Key stored for session ✅" if k.strip() else "Key cleared.", LOG.text() - -def ui_toggle_auto(v): - STATE.auto_retrain = bool(v) - LOG.info("N12", f"auto-retrain {'ENABLED' if v else 'disabled'} (triggers: schedule 24h / +500 bars / drift; cooldown 1h).") - return LOG.text() - -def launch_ui(): - import gradio as gr - with gr.Blocks(title="Forex+Crypto Prediction") as demo: - gr.Markdown("# 📈 Forex + Crypto Trading Prediction\n" - "Prediction & backtesting only — **no live trade execution** (G1). " - "All timestamps IST (UTC+5:30). Historical backtests are **never** evidence of future performance.") - with gr.Tab("Data"): - with gr.Row(): - sym = gr.Textbox(value="EUR/USD", label="Symbol (e.g. EUR/USD, BTC/USD)") - mkt = gr.Dropdown(["forex","crypto"], value="forex", label="Market") - tfd = gr.Dropdown(list(TF_SECONDS), value="H1", label="Timeframe") - with gr.Row(): - mode = gr.Radio(["Live (yfinance→TwelveData→cache)","Upload CSV"], value="Live (yfinance→TwelveData→cache)", label="Source") - fup = gr.File(label="CSV (timestamp/open/high/low/close/volume)") - tzz = gr.Dropdown(["UTC","Asia/Kolkata","America/New_York","Europe/London"], value="UTC", - label="CSV source timezone (required for naive timestamps — never assumed)") - with gr.Row(): - b_load = gr.Button("⬇️ Load Data", variant="primary"); b_ref = gr.Button("🔁 Refresh (new prices)") - load_out = gr.Markdown() - with gr.Tab("Model"): - b_train = gr.Button("🏋️ Train (walk-forward, leakage-safe)", variant="primary") - b_retr = gr.Button("♻️ Manual Retrain (acceptance-gated, rollback on regression)") - auto = gr.Checkbox(False, label="Auto-retrain on triggers (schedule/volume/drift, 1h cooldown)") - b_null = gr.Button("🧪 Null-model (label-shuffle) leakage test") - train_out = gr.Markdown(); train_json = gr.Markdown() - with gr.Tab("Predict"): - b_pred = gr.Button("🎯 Predict (closed candles only)") - pred_md = gr.Markdown(); pred_js = gr.Markdown() - gr.Markdown("#### Forward test (paper trading — strictly separate from backtests)") - with gr.Row(): - fwd_a = gr.Radio(["Start","Stop"], value="Stop", label="Forward test"); b_fwd = gr.Button("Apply") - fwd_out = gr.Markdown() - with gr.Tab("Backtest"): - b_bt = gr.Button("📊 Run OOS Backtest") - bt_md = gr.Markdown(); bt_plot = gr.LinePlot(x="ts", y="equity", title="OOS equity (net of costs)") - with gr.Tab("Providers & Keys"): - key = gr.Textbox(label="Twelve Data API key (session-only, optional)", type="password") - b_key = gr.Button("Save key"); key_out = gr.Markdown() - gr.Markdown("Provider order: **yfinance** (primary, no key) → **Twelve Data** (if key) → **cache/upload** (clearly labeled fallback).") - with gr.Tab("Logs & Audit"): - log_box = gr.Textbox(lines=25, label="Operational log (sequential event IDs)", max_lines=25) - b_log = gr.Button("🔄 Refresh logs") - b_load.click(ui_load,[sym,mkt,tfd,mode,fup,tzz],[load_out,log_box]) - b_ref.click(ui_refresh,None,[load_out,log_box]) - b_train.click(ui_train,None,[train_out,train_json,log_box]) - b_retr.click(lambda: (ui_retrain(), LOG.text())[1],None,log_box) - auto.change(ui_toggle_auto,[auto],[log_box]) - b_null.click(lambda:(null_model_test(),LOG.text()),None,[train_out,log_box]) - b_pred.click(predict,None,[pred_md,pred_js,log_box]) - b_bt.click(ui_backtest,None,[bt_md,bt_plot,log_box]) - b_fwd.click(ui_fwd,[fwd_a],[fwd_out,log_box]) - b_key.click(ui_save_key,[key],[key_out,log_box]) - b_log.click(lambda: LOG.text(),None,log_box) - demo.queue(max_size=16).launch(server_name="0.0.0.0", server_port=7860) - -# ---------------------------------------------------------------------------- -# N23 — Developer acceptance (synthetic stand-ins for the two fixture CSVs) -# ---------------------------------------------------------------------------- -def _synthetic(n, tf, market, seed=7): - rng = np.random.default_rng(seed); iv = TF_SECONDS[tf] - t0 = pd.Timestamp("2026-01-01", tz=OP_TZ) - rets = rng.normal(0, 0.0008, n) - regime_sw = np.sin(np.arange(n)/800.0); rets += 0.0004*np.sign(regime_sw)*rng.random(n) - c = 100*np.exp(np.cumsum(rets)) - o = np.roll(c,1); o[0]=c[0] - h = np.maximum(o,c)*(1+np.abs(rng.normal(0,0.0004,n))) - l = np.minimum(o,c)*(1-np.abs(rng.normal(0,0.0004,n))) - v = np.abs(rng.normal(1e6,3e5,n)) - ts = [t0] - for i in range(1,n): - t = ts[-1]+pd.Timedelta(seconds=iv) - if market=="forex": - while t.dayofweek==5 or (t.dayofweek==6) or (t.dayofweek==0 and t.hour<1): t+=pd.Timedelta(seconds=iv) - ts.append(t) - return pd.DataFrame({"ts":pd.Series(ts).dt.tz_convert(OP_TZ),"open":o,"high":h,"low":l,"close":c,"volume":v}) - -def run_selftest(): - """N23-style acceptance on synthetic Forex & Crypto data (fast HPO).""" - LOG.info("N23","SELFTEST — developer acceptance gate (synthetic fixtures).") - results=[] - for market, sym, tf, n in [("forex","EUR/USD","H1",4200),("crypto","BTC/USD","H1",4200)]: - try: - df = _synthetic(n, tf, market, seed=7 if market=="forex" else 11) - # loader path exercised via CSV round-trip (same node as uploads) - buf = io.StringIO(); df.to_csv(buf, index=False); buf.seek(0) - df2 = canonicalize_csv(buf, None) - a,b = resample_ohlcv(df2,"H4"), resample_ohlcv(df2,"H4") - assert a.equals(b); LOG.info("N23","resampling determinism: PASS") - F = build_features(df2) - assert F["rsi14"].dropna().between(-1e-9,100+1e-9).all(); LOG.info("N23","indicator ranges: PASS") - STATE.symbol, STATE.market, STATE.tf = sym, market, tf - STATE.horizon=HORIZON[tf]; STATE.df=df2; STATE.live_mode=False - STATE.data_hash=dataset_hash(df2); STATE.quality=95.0; STATE.provider="synthetic" - STATE.horizon=HORIZON[tf] - bundle, rep = train_pipeline(df2, sym, market, tf, fast=True) - assert bundle is not None, rep - md, js, _ = predict() - nm = null_model_test() - results.append((sym, True, f"F1={bundle['holdout_metrics']['f1_macro']:.3f} beats_baseline={bundle['beats_baseline']}")) - LOG.info("N23", f"{sym}: train/predict/null PASS ({results[-1][2]}) | {nm}") - except Exception as e: - results.append((sym, False, str(e))); LOG.error("N23", f"{sym}: FAIL {e}\n{traceback.format_exc(limit=5)}") - print("\n==== SELFTEST SUMMARY ====") - for s,ok,msg in results: print(f" [{'PASS' if ok else 'FAIL'}] {s}: {msg}") - return all(ok for _,ok,_ in results) - -# ---------------------------------------------------------------------------- -def main(): - ap = argparse.ArgumentParser() - ap.add_argument("--selftest", action="store_true") - args = ap.parse_args() - LOG.info("G", f"app v{APP_VERSION} starting | feature v{FEATURE_VERSION} | tz=Asia/Kolkata | seed={SEED}") - LOG.info("G", "Prediction/backtest only — no live trade execution. 'Automation' = auto-retrain + prediction refresh (N12/N15).") - if args.selftest: - ok = run_selftest(); sys.exit(0 if ok else 1) - try: - launch_ui() - except ImportError: - print("gradio not installed. pip install -r requirements (see header) or run --selftest.") - -if __name__ == "__main__": - main()