Upload 13 files

README.md

@@ -4,7 +4,7 @@ emoji: 📊
 colorFrom: green
 colorTo: gray
 sdk: gradio
-sdk_version: 4.44.0
+sdk_version: 5.9.1
 app_file: app.py
 pinned: true
 license: mit

app.py

@@ -59,7 +59,7 @@ def refresh_kb():
 #  TAB 1 — UPLOAD & EXTRACT
 # ═══════════════════════════════════════════════════
 
-def run_extraction(pdf_files, progress=gr.Progress(track_tqdm=True)):
+def run_extraction(pdf_files, progress=gr.Progress()):
     """Full extraction pipeline for uploaded PDFs."""
     if not pdf_files:
         return "⚠️ No PDFs uploaded.", ""
@@ -242,7 +242,7 @@ def run_backtests(
     strategy_filter,
     max_strategies,
     viable_only_push,
-    progress=gr.Progress(track_tqdm=True),
+    progress=gr.Progress(),
 ):
     if not cfg.HF_TICK_REPO:
         return "❌ HF_TICK_REPO secret not set.", ""

pipeline/backtester.py

@@ -0,0 +1,700 @@
+"""
+pipeline/backtester.py
+─────────────────────────────────────────────────────
+Strategy compiler, vectorized backtest engine,
+walk-forward optimizer, and output file generator.
+
+Everything that runs on HF Spaces after extraction.
+"""
+
+import json
+import time
+import hashlib
+import itertools
+import re
+from dataclasses import dataclass, field
+from typing import Optional, Callable
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FTimeout
+from datetime import datetime
+import numpy as np
+import pandas as pd
+import anthropic
+from loguru import logger
+
+import utils.config as cfg
+import pipeline.indicators as ind
+
+
+# ═══════════════════════════════════════════════════
+#  STRATEGY COMPILER
+# ═══════════════════════════════════════════════════
+
+@dataclass
+class Executable:
+    canonical_id:    str
+    name:            str
+    code:            str
+    generate_signals: Optional[Callable] = field(default=None, repr=False)
+    get_param_grid:   Optional[Callable] = field(default=None, repr=False)
+    is_valid:        bool = False
+    error:           str  = ""
+
+
+class StrategyCompiler:
+    MODEL   = "claude-sonnet-4-20250514"
+    TIMEOUT = 10  # seconds for sandboxed execution
+
+    def __init__(self):
+        self.client = anthropic.Anthropic(api_key=cfg.ANTHROPIC_API_KEY)
+
+    def compile(self, record: dict) -> Executable:
+        cid  = record.get("canonical_id", "")
+        name = record.get("name", "Unknown")
+
+        # Try cache
+        cache = cfg.TMP_COMPILED / f"{cid}.py"
+        if cache.exists():
+            code = cache.read_text(encoding="utf-8")
+            exe, _ = self._load(code, cid, name)
+            if exe.is_valid:
+                return exe
+
+        # Generate via Claude
+        code = self._generate(record)
+        if not code:
+            return Executable(cid, name, "", error="API returned empty")
+
+        exe, err = self._load(code, cid, name)
+
+        # One retry with error feedback
+        if not exe.is_valid:
+            code2 = self._generate(record, prev_error=err)
+            if code2:
+                exe, _ = self._load(code2, cid, name)
+
+        if exe.is_valid:
+            cache.write_text(code, encoding="utf-8")
+
+        return exe
+
+    def _generate(self, record: dict, prev_error: str = "") -> str:
+        compact = {k: record.get(k) for k in
+                   ("name","category","description","entry_rules","exit_rules",
+                    "filters","parameters","mathematical_basis")}
+        prompt = cfg.COMPILER_PROMPT.format(strategy_json=json.dumps(compact, indent=2))
+        if prev_error:
+            prompt += f"\n\nPREVIOUS ERROR: {prev_error}\nFix the code."
+        try:
+            resp = self.client.messages.create(
+                model=self.MODEL,
+                max_tokens=2000,
+                messages=[{"role": "user", "content": prompt}],
+            )
+            code = resp.content[0].text.strip() if resp.content else ""
+            if code.startswith("```"):
+                code = "\n".join(code.split("\n")[1:]).rstrip("`").strip()
+            return code
+        except Exception as e:
+            logger.error(f"Compile API error: {e}")
+            return ""
+
+    def _load(self, code: str, cid: str, name: str) -> tuple[Executable, str]:
+        exe = Executable(cid, name, code)
+        ns  = self._safe_ns()
+
+        def _exec():
+            exec(compile(code, "<strategy>", "exec"), ns)
+
+        try:
+            with ThreadPoolExecutor(1) as pool:
+                pool.submit(_exec).result(timeout=self.TIMEOUT)
+        except FTimeout:
+            exe.error = "Execution timed out"
+            return exe, exe.error
+        except SyntaxError as e:
+            exe.error = f"SyntaxError: {e}"
+            return exe, exe.error
+        except Exception as e:
+            exe.error = f"{type(e).__name__}: {e}"
+            return exe, exe.error
+
+        if "generate_signals" not in ns:
+            exe.error = "Missing: generate_signals(df, params)"
+            return exe, exe.error
+        if "get_parameter_grid" not in ns:
+            exe.error = "Missing: get_parameter_grid()"
+            return exe, exe.error
+
+        # Smoke test
+        err = self._smoke(ns)
+        if err:
+            exe.error = err
+            return exe, err
+
+        exe.generate_signals = ns["generate_signals"]
+        exe.get_param_grid   = ns["get_parameter_grid"]
+        exe.is_valid         = True
+        return exe, ""
+
+    @staticmethod
+    def _smoke(ns: dict) -> str:
+        try:
+            grid = ns["get_parameter_grid"]()
+            if not isinstance(grid, dict):
+                return "get_parameter_grid() must return dict"
+            params = {k: v[0] for k, v in grid.items() if v}
+            n = 200
+            rng = np.random.default_rng(42)
+            price = 100 + np.cumsum(rng.normal(0, 0.5, n))
+            df = pd.DataFrame({
+                "open":   price + rng.uniform(-0.1, 0.1, n),
+                "high":   price + rng.uniform(0, 0.5, n),
+                "low":    price - rng.uniform(0, 0.5, n),
+                "close":  price,
+                "volume": rng.uniform(1000, 10000, n),
+            }, index=pd.date_range("2020-01-01", periods=n, freq="1h", tz="UTC"))
+            sigs = ns["generate_signals"](df, params)
+            if not isinstance(sigs, pd.Series):
+                return f"generate_signals must return pd.Series, got {type(sigs).__name__}"
+            if len(sigs) != n:
+                return f"Signal length {len(sigs)} != df length {n}"
+            vals = set(int(round(v)) for v in sigs.dropna().unique())
+            if not vals.issubset({-1, 0, 1}):
+                return f"Signal values must be in {{-1,0,1}}, got {vals}"
+        except Exception as e:
+            return f"Smoke test: {type(e).__name__}: {e}"
+        return ""
+
+    @staticmethod
+    def _safe_ns() -> dict:
+        return {
+            "__builtins__": {
+                "abs":abs,"all":all,"any":any,"bool":bool,"dict":dict,
+                "enumerate":enumerate,"float":float,"int":int,
+                "isinstance":isinstance,"len":len,"list":list,
+                "max":max,"min":min,"print":lambda*a,**k:None,
+                "range":range,"round":round,"sorted":sorted,
+                "str":str,"sum":sum,"tuple":tuple,"zip":zip,
+                "None":None,"True":True,"False":False,
+            },
+            "np": np, "pd": pd, "indicators": ind,
+        }
+
+
+# ═══════════════════════════════════════════════════
+#  BACKTEST ENGINE
+# ═══════════════════════════════════════════════════
+
+@dataclass
+class BacktestResult:
+    strategy:      str
+    symbol:        str
+    timeframe:     str
+    params:        dict
+
+    total_return:  float = 0.0
+    cagr:          float = 0.0
+    sharpe:        float = 0.0
+    sortino:       float = 0.0
+    calmar:        float = 0.0
+    max_dd:        float = 0.0
+    n_trades:      int   = 0
+    win_rate:      float = 0.0
+    profit_factor: float = 0.0
+    avg_win:       float = 0.0
+    avg_loss:      float = 0.0
+    expectancy:    float = 0.0
+    final_equity:  float = 0.0
+    total_comm:    float = 0.0
+    data_start:    str   = ""
+    data_end:      str   = ""
+    n_bars:        int   = 0
+    is_valid:      bool  = False
+    error:         str   = ""
+    equity_curve:  list  = field(default_factory=list)
+
+    BARS_PER_YEAR = {"1m":525600,"5m":105120,"15m":35040,"30m":17520,
+                     "1h":8760,"4h":2190,"1d":252,"1w":52}
+
+    def to_dict(self) -> dict:
+        return {k: v for k, v in self.__dict__.items() if k != "equity_curve"}
+
+
+class BacktestEngine:
+    def __init__(self):
+        self.equity    = cfg.INITIAL_EQUITY
+        self.comm      = cfg.COMMISSION_PCT
+        self.slip      = cfg.SLIPPAGE_PCT
+        self.risk      = cfg.RISK_PER_TRADE
+        self.atr_mult  = 2.0
+        self.max_pos   = 0.20
+
+    def run(self, name: str, symbol: str, tf: str,
+            df: pd.DataFrame, signals: pd.Series, params: dict) -> BacktestResult:
+        res = BacktestResult(name, symbol, tf, params,
+                             n_bars=len(df),
+                             data_start=str(df.index[0]) if len(df) else "",
+                             data_end=str(df.index[-1]) if len(df) else "")
+        if len(df) < 50:
+            res.error = "Insufficient bars (<50)"
+            return res
+        if signals.isna().all():
+            res.error = "All signals NaN"
+            return res
+        try:
+            trades, eq = self._simulate(df, signals)
+            res = self._metrics(res, trades, eq, tf)
+            res.is_valid = True
+        except Exception as e:
+            res.error = f"{type(e).__name__}: {e}"
+        return res
+
+    def _simulate(self, df, signals):
+        n      = len(df)
+        closes = df["close"].values.astype(float)
+        highs  = df["high"].values.astype(float)
+        lows   = df["low"].values.astype(float)
+        sigs   = signals.fillna(0).round().clip(-1,1).astype(int).values
+
+        atr_s  = ind.atr(df["high"], df["low"], df["close"], 14).values.astype(float)
+        eq     = np.full(n, self.equity, dtype=float)
+        trades = []
+        pos, entry, sz, stop = 0, 0.0, 0.0, 0.0
+        cash = self.equity
+        last_trade = -2
+
+        for i in range(1, n):
+            price = closes[i]
+            sig   = sigs[i]
+
+            # Stop loss
+            if pos != 0:
+                hit = (pos == 1 and lows[i] <= stop) or (pos == -1 and highs[i] >= stop)
+                if hit:
+                    ep = stop * (1 + self.slip * pos)
+                    pnl, comm = self._pnl(pos, entry, ep, sz)
+                    trades.append((pos, entry, ep, pnl - comm)); cash += pnl - comm; pos = 0; last_trade = i
+
+            # Exit on signal flip or flat
+            if pos != 0 and (sig == 0 or sig == -pos):
+                ep = price * (1 + self.slip * pos)
+                pnl, comm = self._pnl(pos, entry, ep, sz)
+                trades.append((pos, entry, ep, pnl - comm)); cash += pnl - comm; pos = 0; last_trade = i
+
+            # Entry
+            if pos == 0 and sig != 0 and (i - last_trade) >= 1:
+                entry = price * (1 + self.slip * sig)
+                a     = atr_s[i] if not np.isnan(atr_s[i]) else price * 0.01
+                dist  = self.atr_mult * a
+                risk$ = cash * self.risk
+                sz    = min(risk$ / max(dist, 1e-8), cash * self.max_pos / entry)
+                sz    = max(sz, 1e-8)
+                pos   = sig
+                stop  = entry - sig * dist
+
+            mtm = pos * (closes[i] - entry) * sz if pos != 0 else 0
+            eq[i] = cash + mtm
+
+        if pos != 0:
+            ep = closes[-1]
+            pnl, comm = self._pnl(pos, entry, ep, sz)
+            trades.append((pos, entry, ep, pnl - comm)); cash += pnl - comm
+            eq[-1] = cash
+
+        return trades, eq
+
+    @staticmethod
+    def _pnl(direction, entry, exit_, size):
+        gross = direction * (exit_ - entry) * size
+        comm  = (entry + exit_) * size * cfg.COMMISSION_PCT
+        return gross, comm
+
+    def _metrics(self, res, trades, eq, tf):
+        init = self.equity; final = eq[-1]; n = len(eq)
+        bpy  = BacktestResult.BARS_PER_YEAR.get(tf, 252)
+
+        res.final_equity  = float(final)
+        res.total_return  = float((final-init)/init*100)
+        res.equity_curve  = eq.tolist()
+
+        years = n / bpy
+        res.cagr = float(((final/init)**(1/years)-1)*100) if years > 0 and final > 0 else 0.0
+
+        # Drawdown
+        peak = np.maximum.accumulate(eq)
+        dd   = (eq - peak) / peak
+        res.max_dd = float(-dd.min() * 100)
+
+        # Sharpe / Sortino
+        rets = np.diff(eq) / eq[:-1]
+        rets = rets[~np.isnan(rets)]
+        if len(rets) > 1:
+            mu = rets.mean(); sig = rets.std()
+            ds = rets[rets < 0].std() if (rets < 0).any() else sig
+            af = np.sqrt(bpy)
+            res.sharpe  = float(mu/sig*af) if sig > 0 else 0.0
+            res.sortino = float(mu/ds*af)  if ds  > 0 else 0.0
+            res.calmar  = float(res.cagr/res.max_dd) if res.max_dd > 0 else 0.0
+
+        # Trade stats
+        res.n_trades = len(trades)
+        if not trades:
+            return res
+
+        pnls  = np.array([t[3] for t in trades])
+        wins  = pnls > 0; losses = pnls < 0
+        res.win_rate      = float(wins.sum()/len(pnls)*100)
+        gross_w = pnls[wins].sum() if wins.any() else 0
+        gross_l = abs(pnls[losses].sum()) if losses.any() else 0
+        res.profit_factor = float(gross_w/gross_l) if gross_l > 0 else float("inf")
+        entry_vals = np.array([t[1] for t in trades])
+        pct        = pnls / (entry_vals + 1e-10)
+        res.avg_win  = float(pct[wins].mean()*100)  if wins.any()   else 0.0
+        res.avg_loss = float(pct[losses].mean()*100) if losses.any() else 0.0
+        res.expectancy = float(res.win_rate/100*res.avg_win + (1-res.win_rate/100)*res.avg_loss)
+        res.total_comm = float(sum(
+            (t[1]+t[2])*1e-4*cfg.COMMISSION_PCT for t in trades
+        ))
+        return res
+
+
+# ═══════════════════════════════════════════════════
+#  WALK-FORWARD OPTIMIZER
+# ═══════════════════════════════════════════════════
+
+@dataclass
+class OptResult:
+    strategy: str; symbol: str; timeframe: str
+    optimal_params:      dict  = field(default_factory=dict)
+    oos_sharpe_mean:     float = 0.0
+    oos_sharpe_std:      float = 0.0
+    oos_win_rate_mean:   float = 0.0
+    oos_max_dd_mean:     float = 0.0
+    oos_pf_mean:         float = 0.0
+    oos_total_trades:    int   = 0
+    wf_efficiency:       float = 0.0
+    robustness_score:    float = 0.0
+    is_viable:           bool  = False
+    reasons:             list  = field(default_factory=list)
+    oos_results:         list  = field(default_factory=list)
+
+    def to_dict(self) -> dict:
+        return {k:v for k,v in self.__dict__.items() if k != "oos_results"}
+
+
+class WalkForwardOptimizer:
+    def __init__(self):
+        self.n_win  = cfg.WF_WINDOWS
+        self.is_r   = cfg.WF_IS_RATIO
+        self.max_c  = cfg.MAX_PARAM_COMBOS
+        self.engine = BacktestEngine()
+
+    def optimize(self, exe: Executable, symbol: str, tf: str, df: pd.DataFrame) -> OptResult:
+        res = OptResult(exe.name, symbol, tf)
+        if not exe.is_valid:
+            res.reasons.append("Strategy did not compile"); return res
+        if len(df) < 200:
+            res.reasons.append(f"Only {len(df)} bars, need ≥200"); return res
+
+        try:
+            grid = exe.get_param_grid()
+        except Exception as e:
+            res.reasons.append(f"param grid error: {e}"); return res
+
+        combos = self._combos(grid)
+        windows = self._windows(len(df))
+        if not windows:
+            res.reasons.append("Cannot build walk-forward windows"); return res
+
+        oos_results, is_sharpes, win_params = [], [], []
+
+        for is_s, is_e, oos_s, oos_e in windows:
+            is_df  = df.iloc[is_s:is_e]
+            oos_df = df.iloc[oos_s:oos_e]
+
+            best_p, best_sh = None, -np.inf
+            for p in combos:
+                r = self._run(exe, symbol, tf, is_df, p)
+                if r.is_valid and r.n_trades >= cfg.MIN_TRADES and r.sharpe > best_sh:
+                    best_sh = r.sharpe; best_p = p
+
+            if best_p is None:
+                continue
+
+            win_params.append(best_p)
+            is_sharpes.append(best_sh)
+            oos_r = self._run(exe, symbol, tf, oos_df, best_p)
+            oos_results.append(oos_r)
+
+        res.oos_results = oos_results
+        if not oos_results:
+            res.reasons.append("No valid WF windows produced results"); return res
+
+        # Aggregate OOS
+        valid_oos = [r for r in oos_results if r.is_valid and r.n_trades >= cfg.MIN_TRADES]
+        if valid_oos:
+            sharpes = np.array([r.sharpe for r in valid_oos])
+            res.oos_sharpe_mean   = float(sharpes.mean())
+            res.oos_sharpe_std    = float(sharpes.std())
+            res.oos_win_rate_mean = float(np.mean([r.win_rate for r in valid_oos]))
+            res.oos_max_dd_mean   = float(np.mean([r.max_dd for r in valid_oos]))
+            res.oos_pf_mean       = float(np.mean([r.profit_factor for r in valid_oos if r.profit_factor < 100]))
+            res.oos_total_trades  = sum(r.n_trades for r in valid_oos)
+
+        # WF efficiency
+        if is_sharpes and oos_results:
+            oos_sh = [r.sharpe for r in oos_results if r.is_valid]
+            if is_sharpes and oos_sh:
+                res.wf_efficiency = float(np.mean(oos_sh) / np.mean(is_sharpes)) if np.mean(is_sharpes) > 0 else 0.0
+
+        # Optimal params
+        res.optimal_params = self._vote(win_params, oos_results)
+
+        # Robustness & viability
+        res.robustness_score = self._robustness(res)
+        res.is_viable, res.reasons = self._viability(res)
+        return res
+
+    def _run(self, exe, symbol, tf, df, params):
+        try:
+            sigs = exe.generate_signals(df.copy(), params)
+            return self.engine.run(exe.name, symbol, tf, df, sigs, params)
+        except Exception as e:
+            r = BacktestResult(exe.name, symbol, tf, params, error=str(e))
+            return r
+
+    def _combos(self, grid: dict) -> list[dict]:
+        if not grid:
+            return [{}]
+        keys   = list(grid.keys())
+        values = [grid[k] for k in keys]
+        all_c  = list(itertools.product(*values))
+        if len(all_c) > self.max_c:
+            rng   = np.random.default_rng(42)
+            idx   = rng.choice(len(all_c), self.max_c, replace=False)
+            all_c = [all_c[i] for i in idx]
+        return [dict(zip(keys, c)) for c in all_c]
+
+    def _windows(self, n: int) -> list[tuple]:
+        oos_sz = max(50, n // (self.n_win * 2))
+        wins   = []
+        for i in range(self.n_win):
+            oos_e = n - i * oos_sz
+            oos_s = oos_e - oos_sz
+            is_e  = oos_s
+            if is_e < 100 or oos_e - oos_s < 50:
+                continue
+            wins.append((0, is_e, oos_s, oos_e))
+        return list(reversed(wins))
+
+    @staticmethod
+    def _vote(params_list: list[dict], oos_results: list) -> dict:
+        if not params_list:
+            return {}
+        if len(params_list) == 1:
+            return params_list[0]
+        weights = np.array([max(0, r.sharpe) for r in oos_results[:len(params_list)]])
+        if weights.sum() == 0:
+            weights = np.ones(len(params_list))
+        weights /= weights.sum()
+        keys = set()
+        for p in params_list: keys.update(p.keys())
+        optimal = {}
+        for k in keys:
+            vals = [p[k] for p in params_list if k in p]
+            try:
+                arr = np.array(vals, dtype=float)
+                w   = weights[:len(arr)]
+                si  = np.argsort(arr)
+                cw  = np.cumsum(w[si])
+                mi  = np.searchsorted(cw, 0.5)
+                optimal[k] = float(arr[si[min(mi, len(arr)-1)]])
+            except Exception:
+                from collections import Counter
+                optimal[k] = Counter(str(v) for v in vals).most_common(1)[0][0]
+        return optimal
+
+    @staticmethod
+    def _robustness(r: OptResult) -> float:
+        s = 0.0
+        s += min(1.0, max(0.0, r.wf_efficiency)) * 40
+        if r.oos_sharpe_mean > 0:
+            cv = r.oos_sharpe_std / (r.oos_sharpe_mean + 1e-9)
+            s += max(0.0, 1.0 - cv) * 30
+        target = cfg.MIN_TRADES * cfg.WF_WINDOWS
+        s += min(1.0, r.oos_total_trades / max(1, target * 2)) * 20
+        if r.oos_pf_mean > 1:
+            s += min(1.0, (r.oos_pf_mean - 1) / 2) * 10
+        return round(s, 1)
+
+    @staticmethod
+    def _viability(r: OptResult) -> tuple[bool, list[str]]:
+        reasons = []
+        if r.oos_sharpe_mean < cfg.MIN_SHARPE:
+            reasons.append(f"OOS Sharpe {r.oos_sharpe_mean:.2f} < {cfg.MIN_SHARPE}")
+        if r.oos_total_trades < cfg.MIN_TRADES:
+            reasons.append(f"Too few OOS trades: {r.oos_total_trades}")
+        if r.oos_max_dd_mean > 30:
+            reasons.append(f"High avg DD: {r.oos_max_dd_mean:.1f}%")
+        if r.wf_efficiency < 0.3:
+            reasons.append(f"Low WF efficiency: {r.wf_efficiency:.2f}")
+        if r.oos_pf_mean < 1.1:
+            reasons.append(f"Profit factor {r.oos_pf_mean:.2f} < 1.1")
+        viable = len(reasons) == 0
+        if viable:
+            reasons = [f"✅ Sharpe={r.oos_sharpe_mean:.2f} DD={r.oos_max_dd_mean:.1f}% "
+                       f"WFE={r.wf_efficiency:.2f} Score={r.robustness_score:.0f}/100"]
+        return viable, reasons
+
+
+# ═══════════════════════════════════════════════════
+#  OUTPUT FILE GENERATORS
+# ═══════════════════════════════════════════════════
+
+def slugify(text: str) -> str:
+    text = text.lower().strip()
+    text = re.sub(r"[^\w\s-]", "", text)
+    text = re.sub(r"[\s_-]+", "-", text)
+    return text[:50]
+
+
+def build_markdown_report(opt: OptResult, record: dict) -> str:
+    now    = datetime.now().strftime("%Y-%m-%d %H:%M UTC")
+    status = "✅ VIABLE" if opt.is_viable else "❌ NOT VIABLE"
+    lines  = [
+        f"# {opt.strategy}",
+        f"## Backtest Report — {opt.symbol} {opt.timeframe}",
+        f"",
+        f"> **Status:** {status}  ",
+        f"> **Robustness Score:** {opt.robustness_score:.0f}/100  ",
+        f"> **Generated:** {now}  ",
+        f"> **Sources:** {', '.join(record.get('sources', []))[:80]}",
+        f"",
+        "---", "",
+        "## Viability", "",
+    ]
+    for r in opt.reasons:
+        lines.append(f"- {r}")
+    lines += [""]
+
+    if opt.optimal_params:
+        lines += ["## Optimal Parameters", "", "| Parameter | Value |", "|-----------|-------|"]
+        for k, v in opt.optimal_params.items():
+            lines.append(f"| `{k}` | `{v}` |")
+        lines += [""]
+
+    lines += [
+        "## OOS Performance (Walk-Forward Aggregate)", "",
+        "| Metric | Value |", "|--------|-------|",
+        f"| Sharpe (mean ± std) | `{opt.oos_sharpe_mean:.3f} ± {opt.oos_sharpe_std:.3f}` |",
+        f"| Win Rate | `{opt.oos_win_rate_mean:.1f}%` |",
+        f"| Max Drawdown (mean) | `{opt.oos_max_dd_mean:.2f}%` |",
+        f"| Profit Factor | `{opt.oos_pf_mean:.3f}` |",
+        f"| Total OOS Trades | `{opt.oos_total_trades}` |",
+        f"| WF Efficiency Ratio | `{opt.wf_efficiency:.3f}` |",
+        "", "## Window Results", "",
+        "| Window | Trades | Sharpe | Max DD | Win% | PF |",
+        "|--------|--------|--------|--------|------|----|",
+    ]
+    for i, r in enumerate(opt.oos_results, 1):
+        if r.is_valid:
+            lines.append(f"| {i} | {r.n_trades} | {r.sharpe:.2f} | "
+                         f"{r.max_dd:.1f}% | {r.win_rate:.1f}% | {r.profit_factor:.2f} |")
+        else:
+            lines.append(f"| {i} | — | — | — | — | — | *{r.error[:40]}* |")
+
+    if record.get("description"):
+        lines += ["", "## Strategy Description", "", record["description"]]
+    if record.get("entry_rules"):
+        lines += ["", "## Entry Rules", ""]
+        for i, r in enumerate(record["entry_rules"], 1):
+            lines.append(f"{i}. {r}")
+    if record.get("exit_rules"):
+        lines += ["", "## Exit Rules", ""]
+        for i, r in enumerate(record["exit_rules"], 1):
+            lines.append(f"{i}. {r}")
+
+    lines += ["", "---", "*Generated by Quant Knowledge Extractor on HuggingFace*"]
+    return "\n".join(lines)
+
+
+def build_optimal_json(opt: OptResult, record: dict) -> dict:
+    return {
+        "metadata": {
+            "strategy_name": opt.strategy,
+            "symbol":        opt.symbol,
+            "timeframe":     opt.timeframe,
+            "generated":     datetime.now().isoformat(),
+            "is_viable":     opt.is_viable,
+            "robustness":    opt.robustness_score,
+        },
+        "optimal_parameters": opt.optimal_params,
+        "performance": {
+            "oos_sharpe_mean":  round(opt.oos_sharpe_mean, 4),
+            "oos_sharpe_std":   round(opt.oos_sharpe_std, 4),
+            "oos_win_rate":     round(opt.oos_win_rate_mean, 2),
+            "oos_max_dd_pct":   round(opt.oos_max_dd_mean, 2),
+            "oos_pf_mean":      round(opt.oos_pf_mean, 4),
+            "oos_total_trades": opt.oos_total_trades,
+            "wf_efficiency":    round(opt.wf_efficiency, 4),
+        },
+        "viability": {"is_viable": opt.is_viable, "reasons": opt.reasons},
+    }
+
+
+def build_mt5_set(opt: OptResult, record: dict) -> str:
+    lines = [
+        f"; MT5 Strategy Tester Set File",
+        f"; Strategy: {opt.strategy}",
+        f"; Symbol: {opt.symbol}  Timeframe: {opt.timeframe}",
+        f"; OOS Sharpe: {opt.oos_sharpe_mean:.3f}  Score: {opt.robustness_score:.0f}/100",
+        f"; Generated: {datetime.now().strftime('%Y-%m-%d')}",
+        f"; Viable: {opt.is_viable}",
+        f";",
+        f"; HOW TO USE:",
+        f"; 1. Open MetaTrader 5 → View → Strategy Tester",
+        f"; 2. Select Expert Advisor",
+        f"; 3. Inputs tab → Open → select this .set file",
+        f"; 4. Run backtest or deploy live",
+        "",
+    ]
+    for k, v in opt.optimal_params.items():
+        mt5_name = "".join(w.capitalize() for w in k.replace("-","_").split("_"))
+        try:
+            fv   = float(v)
+            step = max(1.0, abs(fv) * 0.1)
+            lines.append(f"{mt5_name}={fv}||{max(0,fv-step*3):.4f}||{step:.4f}||{fv+step*3:.4f}||1")
+        except Exception:
+            lines.append(f"{mt5_name}={v}")
+    lines += ["", "RiskPercent=1.0||0.5||0.5||3.0||1", "LotSize=0.1||0.01||0.01||1.0||1"]
+    return "\n".join(lines)
+
+
+def build_julia_config(opt: OptResult) -> str:
+    struct = "".join(w.capitalize() for w in opt.strategy.split()[:4])
+    params = "\n    ".join(
+        f'{k.lower().replace("-","_")} = {float(v):.6g}' if isinstance(v,(int,float)) else f'{k} = "{v}"'
+        for k, v in opt.optimal_params.items()
+    ) or 'stop_loss = 0.002'
+    return f'''# Optimal Parameters — {opt.strategy}
+# Symbol: {opt.symbol}  Timeframe: {opt.timeframe}
+# OOS Sharpe: {opt.oos_sharpe_mean:.3f}  Score: {opt.robustness_score:.0f}/100
+# Viable: {opt.is_viable}
+
+function optimal_params()::{struct}Params
+    return {struct}Params(
+        {params}
+    )
+end
+
+const BACKTEST_METADATA = (
+    strategy  = "{opt.strategy}",
+    symbol    = "{opt.symbol}",
+    timeframe = "{opt.timeframe}",
+    oos_sharpe = {opt.oos_sharpe_mean:.4f},
+    max_dd_pct = {opt.oos_max_dd_mean:.2f},
+    win_rate   = {opt.oos_win_rate_mean:.1f},
+    wf_eff     = {opt.wf_efficiency:.4f},
+    viable     = {str(opt.is_viable).lower()},
+)
+'''

pipeline/exporter.py

@@ -0,0 +1,176 @@
+"""
+pipeline/exporter.py
+Generates markdown content (as strings) for extracted items.
+No file system writes — returns strings for HF upload.
+"""
+import re
+from datetime import datetime
+
+
+def slugify(text: str) -> str:
+    text = text.lower().strip()
+    text = re.sub(r"[^\w\s-]", "", text)
+    text = re.sub(r"[\s_-]+", "-", text)
+    return text[:50]
+
+
+def strategy_markdown(record: dict) -> str:
+    name     = record.get("name", "Unknown")
+    sources  = record.get("sources", [])
+    category = record.get("category", "")
+    desc     = record.get("description", "")
+    cid      = record.get("canonical_id", "")
+    layers   = record.get("layers", [])
+
+    lines = [
+        f"# {name}", "",
+        f"> **Category:** {category}  ",
+        f"> **Sources:** {', '.join(sources)[:80]}  ",
+        f"> **Extracted:** {datetime.now().strftime('%Y-%m-%d')}  ",
+        f"> **ID:** `{cid}`", "",
+        "---", "", "## Description", "", desc, "",
+    ]
+    for section, key in [("## Entry Rules", "entry_rules"), ("## Exit Rules", "exit_rules"),
+                          ("## Filters", "filters")]:
+        items = record.get(key, [])
+        if items:
+            lines += [section, ""]
+            for i, r in enumerate(items, 1):
+                lines.append(f"{i}. {r}")
+            lines.append("")
+
+    if record.get("timeframes") or record.get("instruments"):
+        lines += ["## Applicability", ""]
+        if record.get("timeframes"):
+            lines.append(f"**Timeframes:** {', '.join(record['timeframes'])}")
+        if record.get("instruments"):
+            lines.append(f"**Instruments:** {', '.join(record['instruments'])}")
+        lines.append("")
+
+    params = record.get("parameters", {})
+    if params:
+        lines += ["## Parameters", "", "| Parameter | Description |", "|-----------|-------------|"]
+        for k, v in params.items():
+            lines.append(f"| `{k}` | {v} |")
+        lines.append("")
+
+    if record.get("mathematical_basis"):
+        lines += ["## Mathematical Basis", "", record["mathematical_basis"], ""]
+
+    if layers:
+        lines += [f"## Variants Found in {len(layers)} Additional Sources", ""]
+        for i, layer in enumerate(layers, 1):
+            src  = layer.get("source_file", "")
+            data = layer.get("data", {})
+            lines.append(f"### Variant {i} — {src}")
+            if data.get("description"):
+                lines.append(data["description"])
+            lines.append("")
+
+    lines += ["---", "", "*Generated by Quant Knowledge Extractor — HuggingFace Spaces*"]
+    return "\n".join(lines)
+
+
+def formula_markdown(record: dict) -> str:
+    name    = record.get("name", "Unknown")
+    sources = record.get("sources", [])
+    latex   = record.get("latex", "")
+    plain   = record.get("plain_text", "")
+    layers  = record.get("layers", [])
+
+    lines = [
+        f"# {name}", "",
+        f"> **Category:** {record.get('category', '')}  ",
+        f"> **Sources:** {', '.join(sources)[:80]}  ",
+        f"> **ID:** `{record.get('canonical_id', '')}`",
+        "", "---", "",
+    ]
+    if latex:
+        lines += ["## Formula (LaTeX)", "", f"$$", latex, f"$$", ""]
+    if plain:
+        lines += ["## Plain Text", "", f"```", plain, f"```", ""]
+    if record.get("purpose"):
+        lines += ["## Purpose", "", record["purpose"], ""]
+    if record.get("usage_context"):
+        lines += ["## Usage in Trading", "", record["usage_context"], ""]
+    if record.get("variables"):
+        lines += ["## Variables", "", "| Symbol | Description |", "|--------|-------------|"]
+        for sym, desc in record["variables"].items():
+            lines.append(f"| `{sym}` | {desc} |")
+        lines.append("")
+    if layers:
+        lines += [f"## Variant Forms ({len(layers)})", ""]
+        for i, l in enumerate(layers, 1):
+            d = l.get("data", {})
+            lines.append(f"### Variant {i} — {l.get('source_file','')}")
+            if d.get("latex"):
+                lines.append(f"$$\n{d['latex']}\n$$")
+            elif d.get("plain_text"):
+                lines.append(f"```\n{d['plain_text']}\n```")
+            lines.append("")
+    lines += ["---", "", "*Generated by Quant Knowledge Extractor — HuggingFace Spaces*"]
+    return "\n".join(lines)
+
+
+def system_markdown(record: dict) -> str:
+    name = record.get("name", "Unknown")
+    lines = [
+        f"# {name}", "",
+        f"> **Sources:** {', '.join(record.get('sources', []))[:80]}  ",
+        f"> **ID:** `{record.get('canonical_id', '')}`",
+        "", "---", "",
+    ]
+    for label, key in [
+        ("System Components", "components"), ("Entry System", "entry_system"),
+        ("Exit System", "exit_system"), ("Risk Management", "risk_management"),
+        ("Position Sizing", "position_sizing"), ("Backtesting Notes", "backtesting_notes"),
+    ]:
+        val = record.get(key)
+        if not val:
+            continue
+        lines += [f"## {label}", ""]
+        if isinstance(val, list):
+            for v in val: lines.append(f"- {v}")
+        else:
+            lines.append(str(val))
+        lines.append("")
+    lines += ["---", "", "*Generated by Quant Knowledge Extractor — HuggingFace Spaces*"]
+    return "\n".join(lines)
+
+
+def backtest_index_markdown(all_results: list) -> str:
+    viable = [r for r in all_results if r.is_viable]
+    viable.sort(key=lambda r: r.oos_sharpe_mean, reverse=True)
+    not_viable = [r for r in all_results if not r.is_viable]
+    now = datetime.now().strftime("%Y-%m-%d %H:%M UTC")
+    lines = [
+        "# Backtest Results Index", "",
+        f"> Generated: {now}  ",
+        f"> Total Tested: {len(all_results)}  ",
+        f"> Viable: {len(viable)}",
+        "", "---", "",
+        "## ✅ Viable Strategies (by OOS Sharpe)", "",
+        "| # | Strategy | Symbol | TF | Sharpe | DD% | Win% | PF | Score |",
+        "|---|----------|--------|----|--------|-----|------|----|-------|",
+    ]
+    for i, r in enumerate(viable, 1):
+        lines.append(
+            f"| {i} | {r.strategy[:35]} | {r.symbol} | {r.timeframe} | "
+            f"{r.oos_sharpe_mean:.2f} | {r.oos_max_dd_mean:.1f} | "
+            f"{r.oos_win_rate_mean:.1f} | {r.oos_pf_mean:.2f} | {r.robustness_score:.0f} |"
+        )
+    lines += ["", "## ❌ Not Viable", "",
+              "| Strategy | Symbol | TF | Reason |",
+              "|----------|--------|----|--------|"]
+    for r in not_viable[:30]:
+        reason = r.reasons[0][:60] if r.reasons else ""
+        lines.append(f"| {r.strategy[:35]} | {r.symbol} | {r.timeframe} | {reason} |")
+    lines += ["", "---", "", "## File Locations (HuggingFace Dataset)", "",
+              "```", "optimal_sets/",
+              "├── BACKTEST_INDEX.md          ← this file",
+              "├── backtest_index.json        ← machine-readable",
+              "├── {strategy}_{sym}_{tf}.set  ← MT5 drag-and-drop",
+              "├── {strategy}_{sym}_{tf}_optimal.json",
+              "└── {strategy}_{sym}_{tf}_config.jl",
+              "```"]
+    return "\n".join(lines)

pipeline/extractor.py

@@ -0,0 +1,206 @@
+"""
+pipeline/extractor.py
+─────────────────────────────────────────────────────
+AI extraction + 3-layer deduplication.
+
+Extraction:  Chunk → Claude API → structured JSON
+Dedup:       SHA256 exact match → TF-IDF cosine similarity → layered merge
+"""
+
+import json
+import time
+import hashlib
+from typing import Optional
+import numpy as np
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
+import anthropic
+from loguru import logger
+import utils.config as cfg
+
+
+# ── Extraction ────────────────────────────────────────
+
+class AIExtractor:
+    MODEL = "claude-sonnet-4-20250514"
+
+    def __init__(self):
+        self.client = anthropic.Anthropic(api_key=cfg.ANTHROPIC_API_KEY)
+        self.tokens_used = 0
+
+    def extract(self, chunk) -> dict:
+        """Extract from one Chunk. Returns {strategies, formulas, systems}."""
+        if chunk.word_count < 20:
+            return {"strategies": [], "formulas": [], "systems": []}
+
+        prompt = cfg.EXTRACTION_PROMPT.format(
+            source_file=chunk.source_file,
+            page_start=chunk.page_start,
+            page_end=chunk.page_end,
+            text=chunk.text,
+        )
+
+        raw = self._call(prompt)
+        if not raw:
+            return {"strategies": [], "formulas": [], "systems": []}
+
+        return self._parse(raw, chunk)
+
+    def _call(self, prompt: str, retries: int = 3) -> str:
+        delay = 2.0
+        for attempt in range(retries):
+            try:
+                resp = self.client.messages.create(
+                    model=self.MODEL,
+                    max_tokens=4096,
+                    messages=[{"role": "user", "content": prompt}],
+                )
+                self.tokens_used += resp.usage.input_tokens + resp.usage.output_tokens
+                return resp.content[0].text if resp.content else ""
+            except anthropic.RateLimitError:
+                logger.warning(f"Rate limit — waiting {delay}s")
+                time.sleep(delay); delay *= 2
+            except Exception as e:
+                logger.error(f"API error: {e}")
+                if attempt == retries - 1:
+                    return ""
+                time.sleep(delay); delay *= 2
+        return ""
+
+    def _parse(self, raw: str, chunk) -> dict:
+        raw = raw.strip()
+        if raw.startswith("```"):
+            raw = "\n".join(raw.split("\n")[1:])
+            raw = raw.rstrip("`").strip()
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError:
+            try:
+                s = raw.find("{"); e = raw.rfind("}")
+                data = json.loads(raw[s:e+1]) if s != -1 else {}
+            except Exception:
+                return {"strategies": [], "formulas": [], "systems": []}
+
+        source_info = {
+            "source_file":  chunk.source_file,
+            "source_pages": f"{chunk.page_start}-{chunk.page_end}",
+        }
+        result = {}
+        for kind in ("strategies", "formulas", "systems"):
+            result[kind] = []
+            for item in data.get(kind, []):
+                if isinstance(item, dict) and item.get("name"):
+                    item.update(source_info)
+                    item["content_hash"] = _hash(
+                        item.get("description", "") +
+                        item.get("plain_text", "") +
+                        item.get("entry_system", "") +
+                        item.get("name", "")
+                    )
+                    result[kind].append(item)
+        return result
+
+
+# ── Deduplication ─────────────────────────────────────
+
+class Deduplicator:
+    """
+    3-layer dedup against a live knowledge base dict.
+    Layer 1: exact SHA256 hash → skip
+    Layer 2: TF-IDF cosine > threshold → merge as variant layer
+    Layer 3: genuinely new → add with canonical_id
+    """
+
+    def __init__(self, threshold: float = cfg.SIMILARITY_THRESHOLD):
+        self.threshold = threshold
+        self._vec = TfidfVectorizer(ngram_range=(1, 2), max_features=5000, stop_words="english")
+
+    def process(self, extracted: dict, kb: dict) -> dict:
+        """
+        Merge extracted items into kb in-place.
+        Returns stats dict.
+        """
+        stats = {k: {"added": 0, "merged": 0, "skipped": 0}
+                 for k in ("strategies", "formulas", "systems")}
+
+        for kind in ("strategies", "formulas", "systems"):
+            for item in extracted.get(kind, []):
+                action = self._process_one(item, kb[kind], kind)
+                stats[kind][action] += 1
+
+        return stats
+
+    def _process_one(self, item: dict, store: dict, kind: str) -> str:
+        h = item.get("content_hash", "")
+
+        # Layer 1: exact hash
+        for existing in store.values():
+            if existing.get("content_hash") == h:
+                self._add_source(item, existing)
+                return "skipped"
+
+        # Layer 2: semantic similarity
+        similar_id = self._find_similar(item, store, kind)
+        if similar_id:
+            self._merge_layer(item, store[similar_id])
+            return "merged"
+
+        # Layer 3: new
+        cid = _canonical_id(item["name"], h, kind)
+        item["canonical_id"] = cid
+        item["sources"]  = [item.get("source_file", "")]
+        item["layers"]   = []
+        store[cid]       = item
+        return "added"
+
+    def _find_similar(self, item: dict, store: dict, kind: str) -> Optional[str]:
+        if not store:
+            return None
+        item_text   = _item_text(item, kind)
+        store_items = list(store.items())
+        texts       = [_item_text(v, kind) for _, v in store_items] + [item_text]
+        try:
+            mat  = self._vec.fit_transform(texts)
+            sims = cosine_similarity(mat[-1], mat[:-1])[0]
+            idx  = int(np.argmax(sims))
+            if sims[idx] >= self.threshold:
+                return store_items[idx][0]
+        except Exception:
+            pass
+        return None
+
+    @staticmethod
+    def _add_source(item: dict, existing: dict):
+        src = item.get("source_file", "")
+        if src and src not in existing.get("sources", []):
+            existing.setdefault("sources", []).append(src)
+
+    @staticmethod
+    def _merge_layer(item: dict, existing: dict):
+        Deduplicator._add_source(item, existing)
+        layers = existing.setdefault("layers", [])
+        hashes = {l.get("content_hash") for l in layers}
+        if item.get("content_hash") not in hashes:
+            layers.append({
+                "source_file":  item.get("source_file"),
+                "source_pages": item.get("source_pages"),
+                "content_hash": item.get("content_hash"),
+                "data":         {k: v for k, v in item.items()
+                                 if k not in ("sources","layers","canonical_id")},
+            })
+
+
+# ── Helpers ───────────────────────────────────────────
+
+def _hash(text: str) -> str:
+    return hashlib.sha256(" ".join(text.lower().split()).encode()).hexdigest()[:16]
+
+def _canonical_id(name: str, h: str, kind: str) -> str:
+    return hashlib.md5(f"{kind}_{name}_{h}".encode()).hexdigest()[:12]
+
+def _item_text(item: dict, kind: str) -> str:
+    if kind == "strategies":
+        return f"{item.get('name','')} {item.get('description','')} {' '.join(item.get('entry_rules',[]))}"
+    if kind == "formulas":
+        return f"{item.get('name','')} {item.get('plain_text','')} {item.get('purpose','')}"
+    return f"{item.get('name','')} {item.get('entry_system','')} {item.get('exit_system','')}"

pipeline/indicators.py

@@ -0,0 +1,96 @@
+"""
+pipeline/indicators.py
+Vectorized technical indicator library used by compiled strategies.
+All functions return pd.Series aligned to input index.
+"""
+import numpy as np
+import pandas as pd
+
+
+def sma(s, n):   return s.rolling(n, min_periods=n).mean()
+def ema(s, n):   return s.ewm(span=n, min_periods=n, adjust=False).mean()
+def std_dev(s, n): return s.rolling(n, min_periods=n).std()
+def highest(s, n): return s.rolling(n, min_periods=n).max()
+def lowest(s, n):  return s.rolling(n, min_periods=n).min()
+def momentum(s, n): return s - s.shift(n)
+def roc(s, n):   return 100 * (s - s.shift(n)) / s.shift(n).replace(0, np.nan)
+
+def wma(s, n):
+    w = np.arange(1, n + 1, dtype=float)
+    return s.rolling(n, min_periods=n).apply(lambda x: np.dot(x, w) / w.sum(), raw=True)
+
+def tema(s, n):
+    e1 = ema(s, n); e2 = ema(e1, n); e3 = ema(e2, n)
+    return 3*e1 - 3*e2 + e3
+
+def rsi(s, n=14):
+    d = s.diff()
+    g = d.clip(lower=0); l = (-d).clip(lower=0)
+    ag = g.ewm(com=n-1, min_periods=n, adjust=False).mean()
+    al = l.ewm(com=n-1, min_periods=n, adjust=False).mean()
+    return 100 - 100/(1 + ag/al.replace(0, np.nan))
+
+def macd(s, fast=12, slow=26, sig=9):
+    ml = ema(s, fast) - ema(s, slow)
+    sl = ema(ml, sig)
+    return ml, sl, ml - sl
+
+def stoch(h, l, c, k=14, d=3):
+    ll = l.rolling(k, min_periods=k).min()
+    hh = h.rolling(k, min_periods=k).max()
+    K  = 100 * (c - ll) / (hh - ll).replace(0, np.nan)
+    return K, K.rolling(d, min_periods=d).mean()
+
+def atr(h, l, c, n=14):
+    tr = pd.concat([h-l, (h-c.shift()).abs(), (l-c.shift()).abs()], axis=1).max(axis=1)
+    return tr.ewm(com=n-1, min_periods=n, adjust=False).mean()
+
+def bbands(s, n=20, k=2.0):
+    mid = sma(s, n); sd = std_dev(s, n)
+    return mid + k*sd, mid, mid - k*sd
+
+def keltner(h, l, c, n=20, k=2.0):
+    mid = ema(c, n); a = atr(h, l, c, n)
+    return mid + k*a, mid, mid - k*a
+
+def donchian(h, l, n=20):
+    u = h.rolling(n, min_periods=n).max()
+    lo = l.rolling(n, min_periods=n).min()
+    return u, (u+lo)/2, lo
+
+def adx(h, l, c, n=14):
+    tr = atr(h, l, c, 1)
+    up = h - h.shift(); dn = l.shift() - l
+    pdm = up.where((up>dn)&(up>0), 0.0)
+    ndm = dn.where((dn>up)&(dn>0), 0.0)
+    sm  = tr.ewm(com=n-1, min_periods=n, adjust=False).mean()
+    pdi = 100*pdm.ewm(com=n-1, min_periods=n, adjust=False).mean()/sm.replace(0,np.nan)
+    ndi = 100*ndm.ewm(com=n-1, min_periods=n, adjust=False).mean()/sm.replace(0,np.nan)
+    dx  = 100*(pdi-ndi).abs()/(pdi+ndi).replace(0,np.nan)
+    return dx.ewm(com=n-1, min_periods=n, adjust=False).mean(), pdi, ndi
+
+def cci(h, l, c, n=20):
+    tp = (h+l+c)/3; s = sma(tp, n)
+    md = tp.rolling(n).apply(lambda x: np.abs(x-x.mean()).mean(), raw=True)
+    return (tp-s)/(0.015*md.replace(0,np.nan))
+
+def williams_r(h, l, c, n=14):
+    hh = h.rolling(n, min_periods=n).max()
+    ll = l.rolling(n, min_periods=n).min()
+    return -100*(hh-c)/(hh-ll).replace(0,np.nan)
+
+def vwap(h, l, c, v):
+    tp = (h+l+c)/3
+    return (tp*v).cumsum()/v.cumsum().replace(0,np.nan)
+
+def obv(c, v):
+    return (np.sign(c.diff())*v).cumsum()
+
+def zscore(s, n=20):
+    return (s - sma(s, n))/std_dev(s, n).replace(0,np.nan)
+
+def crossover(a, b):
+    return (a > b) & (a.shift() <= b.shift())
+
+def crossunder(a, b):
+    return (a < b) & (a.shift() >= b.shift())

pipeline/pdf_processor.py

@@ -0,0 +1,132 @@
+"""
+pipeline/pdf_processor.py
+Extract text from PDFs. OCR fallback for scanned books.
+Runs entirely in HF Spaces /tmp — no local disk needed.
+"""
+
+import re
+import io
+from pathlib import Path
+from typing import Generator
+from dataclasses import dataclass
+import fitz          # PyMuPDF
+import pdfplumber
+import pytesseract
+from pdf2image import convert_from_path
+from loguru import logger
+import utils.config as cfg
+
+
+@dataclass
+class Chunk:
+    source_file: str
+    page_start:  int
+    page_end:    int
+    text:        str
+    was_ocr:     bool = False
+
+    @property
+    def word_count(self) -> int:
+        return len(self.text.split())
+
+
+class PDFProcessor:
+    MIN_CHARS   = 80     # chars per page to skip OCR
+    CHUNK_WORDS = cfg.MAX_TOKENS_PER_CHUNK // 2
+
+    def process(self, pdf_path: Path) -> Generator[Chunk, None, None]:
+        """Yield text chunks from one PDF."""
+        pages = self._extract_pages(pdf_path)
+        yield from self._chunk(pages, pdf_path.name)
+
+    # ── Extraction ────────────────────────────────────
+
+    def _extract_pages(self, path: Path) -> list[tuple[int, str, bool]]:
+        """Returns [(page_num, text, was_ocr), ...]"""
+        results = []
+
+        # Pass 1: PyMuPDF (fastest)
+        mupdf_text = {}
+        try:
+            doc = fitz.open(str(path))
+            for i, page in enumerate(doc):
+                mupdf_text[i + 1] = self._clean(page.get_text("text"))
+            doc.close()
+        except Exception as e:
+            logger.warning(f"PyMuPDF failed {path.name}: {e}")
+
+        # Pass 2: pdfplumber (better table/layout handling)
+        plumber_text = {}
+        try:
+            with pdfplumber.open(str(path)) as pdf:
+                for i, page in enumerate(pdf.pages):
+                    try:
+                        plumber_text[i + 1] = self._clean(page.extract_text() or "")
+                    except Exception:
+                        plumber_text[i + 1] = ""
+        except Exception as e:
+            logger.warning(f"pdfplumber failed {path.name}: {e}")
+
+        total = max(len(mupdf_text), len(plumber_text), 1)
+        ocr_needed = []
+
+        for pnum in range(1, total + 1):
+            a = mupdf_text.get(pnum, "")
+            b = plumber_text.get(pnum, "")
+            best = a if len(a) > len(b) else b
+            if len(best) >= self.MIN_CHARS:
+                results.append((pnum, best, False))
+            else:
+                results.append((pnum, best, False))  # placeholder
+                ocr_needed.append(pnum)
+
+        # Pass 3: OCR for scanned pages
+        if ocr_needed:
+            logger.info(f"  OCR: {len(ocr_needed)} pages in {path.name}")
+            ocr = self._ocr_pages(path, ocr_needed)
+            for i, (pnum, text, _) in enumerate(results):
+                if pnum in ocr:
+                    results[i] = (pnum, ocr[pnum], True)
+
+        return results
+
+    def _ocr_pages(self, path: Path, pages: list[int]) -> dict[int, str]:
+        out = {}
+        try:
+            images = convert_from_path(
+                str(path), dpi=cfg.OCR_DPI,
+                first_page=min(pages), last_page=max(pages),
+            )
+            for i, pnum in enumerate(range(min(pages), max(pages) + 1)):
+                if pnum in pages and i < len(images):
+                    text = pytesseract.image_to_string(images[i], lang="eng", config="--psm 6")
+                    out[pnum] = self._clean(text)
+        except Exception as e:
+            logger.warning(f"OCR error: {e}")
+        return out
+
+    def _chunk(self, pages: list[tuple[int, str, bool]], source: str) -> Generator[Chunk, None, None]:
+        buf, words, p_start, any_ocr = [], 0, 1, False
+        for pnum, text, ocr in pages:
+            if not text:
+                continue
+            buf.append(text)
+            words += len(text.split())
+            if ocr:
+                any_ocr = True
+            if words >= self.CHUNK_WORDS:
+                yield Chunk(source, p_start, pnum, "\n\n".join(buf), any_ocr)
+                buf, words, p_start, any_ocr = [text], len(text.split()), pnum, ocr
+        if buf:
+            last = pages[-1][0] if pages else p_start
+            yield Chunk(source, p_start, last, "\n\n".join(buf), any_ocr)
+
+    @staticmethod
+    def _clean(text: str) -> str:
+        if not text:
+            return ""
+        text = re.sub(r"(\w)-\n(\w)", r"\1\2", text)
+        text = re.sub(r"\n{3,}", "\n\n", text)
+        text = re.sub(r"[ \t]+", " ", text)
+        text = re.sub(r"\x00", "", text)
+        return text.strip()

requirements.txt

@@ -1,6 +1,7 @@
-gradio==4.44.0
-pyaudioop; python_version>="3.13"
-huggingface_hub>=0.24.0
+# Gradio 5.x + huggingface_hub 0.26.x are compatible
+# gradio 4.x broke because HfFolder was removed from huggingface_hub>=0.24
+gradio==5.9.1
+huggingface_hub==0.26.5
 datasets>=2.20.0
 anthropic>=0.30.0
 

utils/config.py

@@ -0,0 +1,150 @@
+"""
+utils/config.py
+All config is read from HuggingFace Spaces Secrets (environment variables).
+Set these in: Space Settings → Variables and Secrets
+No .env file needed — Spaces injects secrets automatically.
+"""
+
+import os
+from pathlib import Path
+
+# ── HF Spaces uses /tmp for ephemeral working storage ─
+# All persistent data is written to HF Dataset via API.
+TMP = Path("/tmp/quant")
+TMP_PDFS       = TMP / "pdfs"
+TMP_CHUNKS     = TMP / "chunks"
+TMP_COMPILED   = TMP / "compiled"
+TMP_BACKTESTS  = TMP / "backtests"
+TMP_EXPORTS    = TMP / "exports"
+TMP_TICK_CACHE = TMP / "tick_cache"
+TMP_KB         = TMP / "knowledge_base.jsonl"
+
+for d in [TMP_PDFS, TMP_CHUNKS, TMP_COMPILED,
+          TMP_BACKTESTS, TMP_EXPORTS, TMP_TICK_CACHE]:
+    d.mkdir(parents=True, exist_ok=True)
+
+
+def get(key: str, default: str = "") -> str:
+    return os.environ.get(key, default)
+
+
+# ── API Keys (set in HF Space Secrets) ────────────────
+ANTHROPIC_API_KEY = get("ANTHROPIC_API_KEY")
+HF_TOKEN          = get("HF_TOKEN")
+
+# ── HuggingFace Dataset repos ─────────────────────────
+HF_DATASET_REPO = get("HF_DATASET_REPO")   # knowledge base + results
+HF_TICK_REPO    = get("HF_TICK_REPO")       # your tick data
+
+# ── Extraction settings ───────────────────────────────
+SIMILARITY_THRESHOLD  = float(get("SIMILARITY_THRESHOLD", "0.85"))
+MAX_TOKENS_PER_CHUNK  = int(get("MAX_TOKENS_PER_CHUNK", "3000"))
+OCR_DPI               = int(get("OCR_DPI", "300"))
+
+# ── Backtest settings ─────────────────────────────────
+INITIAL_EQUITY        = float(get("INITIAL_EQUITY", "10000"))
+COMMISSION_PCT        = float(get("COMMISSION_PCT", "0.0002"))
+SLIPPAGE_PCT          = float(get("SLIPPAGE_PCT", "0.0001"))
+RISK_PER_TRADE        = float(get("RISK_PER_TRADE", "0.01"))
+WF_WINDOWS            = int(get("WF_WINDOWS", "5"))
+WF_IS_RATIO           = float(get("WF_IS_RATIO", "0.70"))
+MAX_PARAM_COMBOS      = int(get("MAX_PARAM_COMBOS", "300"))
+MIN_TRADES            = int(get("MIN_TRADES", "30"))
+MIN_SHARPE            = float(get("MIN_SHARPE", "0.5"))
+BACKTEST_TIMEFRAMES   = get("BACKTEST_TIMEFRAMES", "1h,4h,1d").split(",")
+
+# ── Strategy categories ───────────────────────────────
+CATEGORIES = [
+    "Trend Following", "Mean Reversion", "Statistical Arbitrage",
+    "Momentum", "Breakout", "Volatility Trading", "Market Making",
+    "Pattern Recognition", "Machine Learning", "Options Strategy",
+    "High Frequency", "Pairs Trading", "Carry Trade",
+    "Seasonal / Calendar", "Risk Management", "Position Sizing",
+    "Portfolio Construction", "Market Microstructure", "Other",
+]
+
+# ── Extraction prompt ─────────────────────────────────
+EXTRACTION_PROMPT = """
+You are a quantitative finance knowledge extraction engine.
+
+Extract ALL trading strategies, mathematical formulas, and complete trading systems
+from the text chunk below, taken from an algorithmic trading book.
+
+Output ONLY valid JSON — no markdown fences, no preamble:
+{
+  "strategies": [{
+    "name": "string",
+    "category": "string",
+    "description": "string",
+    "entry_rules": ["string"],
+    "exit_rules": ["string"],
+    "filters": ["string"],
+    "timeframes": ["string"],
+    "instruments": ["string"],
+    "parameters": {"name": "description with typical value"},
+    "mathematical_basis": "string",
+    "source_context": "brief quote from text"
+  }],
+  "formulas": [{
+    "name": "string",
+    "category": "string",
+    "latex": "LaTeX string",
+    "plain_text": "string",
+    "variables": {"symbol": "description"},
+    "purpose": "string",
+    "usage_context": "string",
+    "source_context": "string"
+  }],
+  "systems": [{
+    "name": "string",
+    "components": ["string"],
+    "entry_system": "string",
+    "exit_system": "string",
+    "risk_management": "string",
+    "position_sizing": "string",
+    "backtesting_notes": "string",
+    "source_context": "string"
+  }]
+}
+
+Rules:
+- Return empty arrays [] if nothing found for a category
+- Never paraphrase formulas — preserve exact mathematical form
+- Include LaTeX for every formula that has a standard form
+- Be exhaustive — extract even briefly mentioned concepts
+
+Source: {source_file} | Pages: {page_start}–{page_end}
+--- TEXT ---
+{text}
+--- END ---
+""".strip()
+
+# ── Strategy compiler prompt ──────────────────────────
+COMPILER_PROMPT = """
+You are a Python algorithmic trading code generator.
+
+Convert the strategy JSON below into a Python module.
+Output ONLY valid Python — no markdown, no explanation.
+
+Required functions:
+
+def get_parameter_grid() -> dict:
+    # Returns {param_name: [list_of_values_to_try]}
+    # Provide 3-5 values per parameter covering a reasonable range
+    ...
+
+def generate_signals(df: pd.DataFrame, params: dict) -> pd.Series:
+    # df columns: open, high, low, close, volume (DatetimeIndex, UTC)
+    # Returns pd.Series of int: 1=long, -1=short, 0=flat, same index as df
+    # Use ONLY: numpy as np, pandas as pd, indicators module
+    # Available in indicators: sma, ema, rsi, atr, bbands, macd,
+    #   stoch, adx, donchian, keltner, vwap, obv, cci, williams_r,
+    #   momentum, roc, std_dev, zscore, crossover, crossunder
+    ...
+
+Allowed imports: numpy as np, pandas as pd, indicators (pre-imported)
+No file I/O. No network. No print statements.
+
+Strategy:
+{strategy_json}
+""".strip()

utils/hf_io.py

@@ -0,0 +1,440 @@
+"""
+utils/hf_io.py
+─────────────────────────────────────────────────────
+Single module for all HuggingFace Hub I/O.
+
+The HF dataset repo is the persistent store for everything:
+  ├── pdfs/                  uploaded source ebooks
+  ├── knowledge_base.jsonl   deduplicated strategies/formulas
+  ├── extracted/             individual markdown files
+  ├── backtests/             per-strategy backtest reports
+  └── optimal_sets/          .set, .json, .jl files
+
+The HF tick repo is read-only from this Space:
+  └── {SYMBOL}/ticks.parquet  or  {SYMBOL}/{tf}.parquet
+"""
+
+import json
+import io
+from pathlib import Path
+from typing import Optional
+import pandas as pd
+from huggingface_hub import (
+    HfApi, hf_hub_download, list_repo_files, CommitOperationAdd
+)
+from loguru import logger
+import utils.config as cfg
+
+
+def _api() -> HfApi:
+    return HfApi(token=cfg.HF_TOKEN)
+
+
+# ═══════════════════════════════════════════════════
+#  KNOWLEDGE BASE — read / write
+# ═══════════════════════════════════════════════════
+
+def kb_load() -> dict:
+    """
+    Download knowledge_base.jsonl from HF dataset.
+    Returns {"strategies": [...], "formulas": [...], "systems": [...]}.
+    """
+    empty = {"strategies": {}, "formulas": {}, "systems": {}}
+    if not cfg.HF_DATASET_REPO:
+        return empty
+    try:
+        path = hf_hub_download(
+            repo_id=cfg.HF_DATASET_REPO,
+            filename="knowledge_base.jsonl",
+            repo_type="dataset",
+            token=cfg.HF_TOKEN,
+            local_dir=str(cfg.TMP),
+            force_download=True,
+        )
+        result = {"strategies": {}, "formulas": {}, "systems": {}}
+        with open(path, encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if not line:
+                    continue
+                rec = json.loads(line)
+                kind = rec.get("_type", "")
+                cid  = rec.get("canonical_id", "")
+                if kind in result and cid:
+                    result[kind][cid] = rec
+        logger.info(
+            f"KB loaded: {len(result['strategies'])} strategies, "
+            f"{len(result['formulas'])} formulas, {len(result['systems'])} systems"
+        )
+        return result
+    except Exception as e:
+        logger.warning(f"KB load (may not exist yet): {e}")
+        return empty
+
+
+def kb_save(kb: dict) -> bool:
+    """Push knowledge base JSONL to HF dataset."""
+    if not cfg.HF_DATASET_REPO:
+        return False
+    try:
+        lines = []
+        for kind in ("strategies", "formulas", "systems"):
+            for rec in kb[kind].values():
+                lines.append(json.dumps({**rec, "_type": kind}))
+        content = "\n".join(lines).encode("utf-8")
+        _api().upload_file(
+            path_or_fileobj=io.BytesIO(content),
+            path_in_repo="knowledge_base.jsonl",
+            repo_id=cfg.HF_DATASET_REPO,
+            repo_type="dataset",
+            commit_message="Update knowledge base",
+        )
+        logger.info(f"KB saved: {len(lines)} records")
+        return True
+    except Exception as e:
+        logger.error(f"KB save failed: {e}")
+        return False
+
+
+# ═══════════════════════════════════════════════════
+#  PDF — upload
+# ═══════════════════════════════════════════════════
+
+def pdf_upload(local_path: Path) -> str:
+    """Upload a PDF to the HF dataset. Returns URL or ''."""
+    if not cfg.HF_DATASET_REPO:
+        return ""
+    try:
+        url = _api().upload_file(
+            path_or_fileobj=str(local_path),
+            path_in_repo=f"pdfs/{local_path.name}",
+            repo_id=cfg.HF_DATASET_REPO,
+            repo_type="dataset",
+            commit_message=f"Add PDF: {local_path.name}",
+        )
+        return str(url)
+    except Exception as e:
+        logger.warning(f"PDF upload failed {local_path.name}: {e}")
+        return ""
+
+
+def pdf_list() -> list[str]:
+    """List all PDFs stored in HF dataset."""
+    if not cfg.HF_DATASET_REPO:
+        return []
+    try:
+        return [
+            f for f in list_repo_files(
+                repo_id=cfg.HF_DATASET_REPO,
+                repo_type="dataset",
+                token=cfg.HF_TOKEN,
+            )
+            if f.startswith("pdfs/") and f.endswith(".pdf")
+        ]
+    except Exception as e:
+        logger.warning(f"PDF list failed: {e}")
+        return []
+
+
+# ═════════════════════════���═════════════════════════
+#  EXTRACTED MARKDOWN — push
+# ═══════════════════════════════════════════════════
+
+def push_markdown(content: str, remote_path: str) -> bool:
+    """Push a single markdown file to HF dataset."""
+    if not cfg.HF_DATASET_REPO:
+        return False
+    try:
+        _api().upload_file(
+            path_or_fileobj=io.BytesIO(content.encode("utf-8")),
+            path_in_repo=remote_path,
+            repo_id=cfg.HF_DATASET_REPO,
+            repo_type="dataset",
+            commit_message=f"Add: {remote_path}",
+        )
+        return True
+    except Exception as e:
+        logger.debug(f"MD push failed {remote_path}: {e}")
+        return False
+
+
+def push_batch(files: list[tuple[str, bytes]], commit_msg: str = "Batch update") -> int:
+    """
+    Push multiple files in a single commit.
+    files: list of (remote_path, bytes_content)
+    Returns number of files successfully pushed.
+    """
+    if not cfg.HF_DATASET_REPO or not files:
+        return 0
+    ops = [
+        CommitOperationAdd(
+            path_in_repo=remote_path,
+            path_or_fileobj=io.BytesIO(content),
+        )
+        for remote_path, content in files
+    ]
+    # HF allows max 100 ops per commit
+    pushed = 0
+    for i in range(0, len(ops), 100):
+        batch = ops[i:i+100]
+        try:
+            _api().create_commit(
+                repo_id=cfg.HF_DATASET_REPO,
+                repo_type="dataset",
+                operations=batch,
+                commit_message=f"{commit_msg} [{i+1}–{i+len(batch)}]",
+            )
+            pushed += len(batch)
+        except Exception as e:
+            logger.error(f"Batch push failed: {e}")
+    return pushed
+
+
+# ═══════════════════════════════════════════════════
+#  TICK DATA — read from HF tick repo
+# ═══════════════════════════════════════════════════
+
+def tick_list_symbols() -> list[str]:
+    """List all symbols in the tick dataset."""
+    if not cfg.HF_TICK_REPO:
+        return []
+    try:
+        files = list(list_repo_files(
+            repo_id=cfg.HF_TICK_REPO,
+            repo_type="dataset",
+            token=cfg.HF_TOKEN,
+        ))
+        seen = set()
+        symbols = []
+        for f in files:
+            parts = f.split("/")
+            if len(parts) >= 2 and parts[0] not in seen:
+                seen.add(parts[0])
+                symbols.append(parts[0])
+        return sorted(symbols)
+    except Exception as e:
+        logger.warning(f"Tick symbol list failed: {e}")
+        return []
+
+
+def tick_load(symbol: str, timeframe: str = "1h") -> Optional[pd.DataFrame]:
+    """
+    Load OHLCV data for a symbol at given timeframe.
+    Downloads from HF tick repo, caches locally in /tmp.
+    
+    Looks for (in order):
+      {symbol}/{timeframe}.parquet
+      {symbol}/ticks.parquet  → then resamples to timeframe
+      {symbol}/data.parquet
+    """
+    cache_path = cfg.TMP_TICK_CACHE / f"{symbol}_{timeframe}.parquet"
+    if cache_path.exists():
+        return pd.read_parquet(cache_path)
+
+    if not cfg.HF_TICK_REPO:
+        return None
+
+    # Try pre-built timeframe file
+    for fname in [f"{timeframe}.parquet", f"{timeframe}.csv"]:
+        df = _try_download_tick(symbol, fname)
+        if df is not None:
+            df = _normalize_ohlcv(df)
+            if df is not None:
+                df.to_parquet(cache_path)
+                return df
+
+    # Fall back to ticks + resample
+    for fname in ["ticks.parquet", "data.parquet", "ticks.csv"]:
+        ticks = _try_download_tick(symbol, fname)
+        if ticks is not None:
+            df = _resample(ticks, timeframe)
+            if df is not None and not df.empty:
+                df.to_parquet(cache_path)
+                return df
+
+    logger.warning(f"No data found for {symbol} in tick repo")
+    return None
+
+
+def _try_download_tick(symbol: str, filename: str) -> Optional[pd.DataFrame]:
+    try:
+        local = cfg.TMP_TICK_CACHE / symbol
+        local.mkdir(parents=True, exist_ok=True)
+        path = hf_hub_download(
+            repo_id=cfg.HF_TICK_REPO,
+            filename=f"{symbol}/{filename}",
+            repo_type="dataset",
+            token=cfg.HF_TOKEN,
+            local_dir=str(local),
+            force_download=False,
+        )
+        if filename.endswith(".parquet"):
+            return pd.read_parquet(path)
+        return pd.read_csv(path)
+    except Exception:
+        return None
+
+
+_RESAMPLE_MAP = {
+    "1m":"1min","3m":"3min","5m":"5min","15m":"15min","30m":"30min",
+    "1h":"1h","2h":"2h","4h":"4h","6h":"6h","12h":"12h",
+    "1d":"1D","1w":"1W",
+}
+
+
+def _resample(ticks: pd.DataFrame, timeframe: str) -> Optional[pd.DataFrame]:
+    """Resample raw tick data to OHLCV."""
+    import numpy as np
+    ticks = ticks.copy()
+
+    # Parse timestamp
+    ts_col = next((c for c in ticks.columns if "time" in c.lower() or "date" in c.lower()), None)
+    if ts_col:
+        ticks.index = pd.to_datetime(ticks[ts_col], utc=True)
+        ticks = ticks.drop(columns=[ts_col])
+    else:
+        try:
+            ticks.index = pd.to_datetime(ticks.index, utc=True)
+        except Exception:
+            return None
+
+    ticks.index = ticks.index.tz_convert("UTC") if ticks.index.tz else ticks.index.tz_localize("UTC")
+
+    # Derive price
+    if "bid" in ticks.columns and "ask" in ticks.columns:
+        ticks["_price"] = (ticks["bid"] + ticks["ask"]) / 2
+    elif "price" in ticks.columns:
+        ticks["_price"] = ticks["price"]
+    elif "close" in ticks.columns:
+        ticks["_price"] = ticks["close"]
+    else:
+        return None
+
+    rule = _RESAMPLE_MAP.get(timeframe, "1h")
+    ohlcv = ticks["_price"].resample(rule).ohlc()
+    ohlcv.columns = ["open", "high", "low", "close"]
+
+    vol_col = next((c for c in ticks.columns if "vol" in c.lower() or "size" in c.lower()), None)
+    ohlcv["volume"] = ticks[vol_col].resample(rule).sum() if vol_col else ticks["_price"].resample(rule).count()
+    return ohlcv.dropna()
+
+
+def _normalize_ohlcv(df: pd.DataFrame) -> Optional[pd.DataFrame]:
+    """Normalize column names and index for any OHLCV DataFrame."""
+    df = df.copy()
+
+    # Parse index
+    ts_col = next((c for c in df.columns if "time" in c.lower() or "date" in c.lower()), None)
+    if ts_col:
+        df.index = pd.to_datetime(df[ts_col], utc=True)
+        df = df.drop(columns=[ts_col])
+    else:
+        try:
+            df.index = pd.to_datetime(df.index, utc=True)
+        except Exception:
+            return None
+
+    df.index = df.index.tz_convert("UTC") if df.index.tz else df.index.tz_localize("UTC")
+    df.index.name = "timestamp"
+    df = df.sort_index()
+
+    # Normalize column names
+    renames = {}
+    for c in df.columns:
+        lc = c.lower().strip()
+        if lc in ("o", "open"):           renames[c] = "open"
+        elif lc in ("h", "high"):         renames[c] = "high"
+        elif lc in ("l", "low"):          renames[c] = "low"
+        elif lc in ("c", "close"):        renames[c] = "close"
+        elif lc in ("v","vol","volume","tick_volume"): renames[c] = "volume"
+    df = df.rename(columns=renames)
+
+    for col in ["open","high","low","close"]:
+        if col not in df.columns:
+            return None
+    if "volume" not in df.columns:
+        df["volume"] = 0.0
+
+    df = df[["open","high","low","close","volume"]].astype(float)
+    df = df.dropna(subset=["open","high","low","close"])
+
+    # Fix inverted high/low
+    bad = df["high"] < df["low"]
+    if bad.any():
+        df.loc[bad, ["high","low"]] = df.loc[bad, ["low","high"]].values
+
+    return df
+
+
+# ═══════════════════════════════════════════════════
+#  BACKTEST RESULTS — push
+# ═══════════════════════════════════════════════════
+
+def push_backtest_result(
+    strategy_name: str,
+    symbol: str,
+    timeframe: str,
+    report_md: str,
+    optimal_json: dict,
+    mt5_set: str,
+    julia_cfg: str,
+) -> bool:
+    """Push all output files for one backtest result to HF dataset."""
+    from pipeline.exporter import slugify
+    slug = slugify(strategy_name)
+    prefix = f"{slug}_{symbol}_{timeframe}"
+
+    files = [
+        (f"backtests/{slug}/{prefix}_report.md",   report_md.encode()),
+        (f"optimal_sets/{prefix}_optimal.json",     json.dumps(optimal_json, indent=2).encode()),
+        (f"optimal_sets/{prefix}.set",              mt5_set.encode()),
+        (f"optimal_sets/{prefix}_config.jl",        julia_cfg.encode()),
+    ]
+    pushed = push_batch(files, commit_msg=f"Backtest: {strategy_name} {symbol} {timeframe}")
+    return pushed == len(files)
+
+
+def push_backtest_index(index_md: str, index_json: dict) -> bool:
+    """Push master index files."""
+    files = [
+        ("optimal_sets/BACKTEST_INDEX.md",       index_md.encode()),
+        ("optimal_sets/backtest_index.json",     json.dumps(index_json, indent=2).encode()),
+    ]
+    return push_batch(files, "Update backtest index") == 2
+
+
+def fetch_backtest_index() -> dict:
+    """Download backtest index JSON from HF dataset."""
+    if not cfg.HF_DATASET_REPO:
+        return {}
+    try:
+        path = hf_hub_download(
+            repo_id=cfg.HF_DATASET_REPO,
+            filename="optimal_sets/backtest_index.json",
+            repo_type="dataset",
+            token=cfg.HF_TOKEN,
+            local_dir=str(cfg.TMP),
+            force_download=True,
+        )
+        return json.loads(Path(path).read_text())
+    except Exception:
+        return {}
+
+
+def fetch_file(remote_path: str) -> Optional[bytes]:
+    """Download any file from HF dataset as bytes."""
+    if not cfg.HF_DATASET_REPO:
+        return None
+    try:
+        path = hf_hub_download(
+            repo_id=cfg.HF_DATASET_REPO,
+            filename=remote_path,
+            repo_type="dataset",
+            token=cfg.HF_TOKEN,
+            local_dir=str(cfg.TMP / "downloads"),
+            force_download=True,
+        )
+        return Path(path).read_bytes()
+    except Exception as e:
+        logger.debug(f"Fetch {remote_path}: {e}")
+        return None