fx-crypto-predictor.py

#!/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 = (top<CONF_THRESHOLD) or ood_flag or (unc is not None and unc>0.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()