| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| 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) |
|
|
| |
| |
| |
| APP_VERSION = "1.0.0" |
| FEATURE_VERSION = "1.3.0" |
| CALIB_VERSION = "1.2.0" |
| SEED = 42 |
| OP_TZ = ZoneInfo("Asia/Kolkata") |
| 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} |
| HORIZON = {"M1":15,"M5":12,"M15":8,"M30":6,"H1":4,"H4":2,"D1":1} |
| FIXED_THRESHOLD = 0.0025 |
| LABEL_SCHEME = "triple_barrier" |
| ATR_K = {tf: 2.0 for tf in TF_SECONDS} |
| HOLD_BAND_FRAC = 0.25 |
|
|
| MAX_LOOKBACK = 120 |
| EMBARGO = 20 |
| WF_TRAIN_FRAC, WF_VAL_FRAC, WF_HOLDOUT_FRAC, WF_FOLDS = 0.70, 0.20, 0.10, 5 |
|
|
| CONF_THRESHOLD = 0.70 |
| CONF_MARGIN = 0.10 |
| BAND_VH, BAND_HI, BAND_MED = 0.90, 0.80, 0.70 |
| BOOTSTRAP_ISOTONIC_B = 100 |
| CONFORMAL_ALPHA = 0.10 |
| PLATT_MIN_SAMPLES = 200 |
|
|
| HPO = dict(depth=(4,8), lr=(0.01,0.10), l2=(3.0,10.0), iterations=1200, |
| early_stopping=100, trials=25, seed=SEED) |
| HPO_LARGE_ROWS, HPO_LARGE_FOLDS = 15000, 2 |
| FALLBACK_MAX_ITER = 300 |
| FOLD_F1_STD_REJECT = 0.10 |
|
|
| DRIFT_PSI, DRIFT_PERF_F1, DRIFT_CONF = 0.25, 0.05, 0.10 |
| DQ_TRAIN_MIN = 40 |
|
|
| COMMISSION, SLIPPAGE = 0.0005, 0.0002 |
| DEFAULT_SPREAD = {"forex":0.00010, "crypto":0.00050} |
| RISK = dict(sl_atr=1.5, tp_atr=2.5, trailing=True, max_exposure=0.25, |
| trade_conf=0.70) |
|
|
| MAX_ROWS, MAX_UPLOAD_MB = 400_000, 200 |
| STALE_FACTOR = 3.0 |
| EXTREME_GAP_ATR = 3.0 |
| EXTREME_ATR_PCT = 0.995 |
| WARMUP_PREDS, PRED_TIMEOUT_SEC = 5, 30 |
| RETRAIN_COOLDOWN_SEC, RETRAIN_SCHEDULE_SEC, RETRAIN_VOLUME_BARS = 3600, 86400, 500 |
| RETIRE_AFTER_FAILS = 3 |
| CALENDAR_VERSION, CALENDAR_YEAR = "2025.1", 2025 |
| LEADLAG_MIN_R, REDUNDANCY_CORR = 0.01, 0.95 |
| FRACDIFF_D, FRACDIFF_WIN = 0.4, 100 |
| ANOM_CONTAM = 0.05 |
| PERT_COPIES, PERT_NOISE_FRAC = 8, 0.005 |
| ANALOG_K, ANALOG_KEEP = 10, 2000 |
| ART_DIR, AUDIT_PRUNE = "artifacts", None |
|
|
| 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"} |
|
|
| |
| |
| |
| 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) |
|
|
| 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) |
|
|
| |
| |
| |
| 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) |
| 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]) |
|
|
| |
| |
| |
| COLMAP = { |
| "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() |
| |
| 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).") |
| |
| 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) |
| |
| 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] |
|
|
| |
| |
| |
| 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()) |
| 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}" |
|
|
| |
| |
| |
| |
| 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) |
| obv = (np.sign(c.diff()).fillna(0)*v).cumsum() |
| f["obv_slope"] = obv.diff(5)/(v.rolling(20).mean()*5 + 1e-12) |
| f["don_up"], f["don_lo"] = h.rolling(20).max(), l.rolling(20).min() |
| 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) |
| |
| 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() |
| |
| F["ffd_close"] = fracdiff_series(np.log(c)) |
| |
| if companion is not None and len(companion): |
| comp = companion.set_index("ts")["close"].reindex(df["ts"]).ffill(limit=3) |
| 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) |
| |
| 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) |
| return F |
|
|
| FEATURE_UNITS = { |
| "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} |
|
|
| |
| |
| |
| 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) |
| lab[(fu < fd) & (fu <= horizon)] = 2 |
| lab[(fd < fu) & (fd <= horizon)] = 0 |
| both = (fu == fd) & (fu <= horizon) |
| lab[both] = 1 |
| vert = (fu > horizon) & (fd > horizon) |
| lab[vert & (ret_end > band)] = 2 |
| lab[vert & (ret_end < -band)] = 0 |
| lab[~valid] = -1 |
| 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.") |
| |
| 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 |
|
|
| |
| |
| |
| 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) |
| 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) |
|
|
| |
| |
| |
| 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") |
|
|
| |
| |
| |
| 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] |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| 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) |
|
|
| 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] |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| 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) |
| 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 |
|
|
| |
| |
| |
| 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) |
|
|
| 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"} |
|
|
| def _catboost_version(): |
| try: |
| import catboost; return catboost.__version__ |
| except Exception: return "unavailable" |
|
|
| |
| |
| |
| @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 |
|
|
| |
| |
| |
| 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 |
| 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 |
|
|
| |
| |
| |
| @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=[] |
| |
| 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)"} |
|
|
| |
| |
| |
| @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() |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| 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).") |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| 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}") |
| |
| comp = fetch_companion(symbol, market, tf) |
| F = build_features(df, comp) |
| F = F.iloc[MAX_LOOKBACK:].reset_index(drop=True) |
| dfx = df.iloc[MAX_LOOKBACK:].reset_index(drop=True) |
| n = len(dfx) |
| |
| 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)") |
| |
| 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) |
| |
| 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] |
| |
| 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) |
| |
| 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) |
| feat_names.append("anomaly_score") |
| LOG.info("N33","anomaly-score feature fitted on train fold and appended.") |
| |
| if not dfx["ts"].is_monotonic_increasing: raise AssertionError("monotonicity violated before splits (N6).") |
| |
| 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) |
| |
| tr_all = np.arange(0, splits[-1][1].max()) |
| w_all = compose_sample_weights(y, end, ret_end, band, tr_all) |
| X = F.astype("float32") |
| |
| 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).") |
| |
| 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.") |
| |
| 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).") |
| |
| 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).") |
| |
| 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}") |
| |
| 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 |
| |
| 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}") |
| |
| 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] |
| |
| 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```" |
|
|
| |
| |
| |
| 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) |
| |
| 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 |
| |
| 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() |
|
|
| |
| |
| |
| 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] |
| |
| 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 |
| 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).") |
| |
| 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() |
| |
| tail = F[B["feature_names"]].tail(200) |
| mean_psi, worst, drift_fired = drift_report(B["drift_ref"], tail) |
| |
| 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).") |
| |
| 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) |
| |
| 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) |
| |
| 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 |
| |
| 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 |
| |
| 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)} |
| |
| 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}") |
| |
| 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() |
|
|
| |
| |
| |
| 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}**" |
|
|
| |
| |
| |
| 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)})") |
| |
| 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) |
| |
| 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) |
|
|
| |
| |
| |
| 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) |
| |
| 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() |
|
|
