Delete pipeline

pipeline/backtester.py

@@ -1,700 +0,0 @@
-"""
-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_amt = cash * self.risk
-                sz    = min(risk_amt / 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

@@ -1,176 +0,0 @@
-"""
-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

@@ -1,206 +0,0 @@
-"""
-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

@@ -1,96 +0,0 @@
-"""
-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

@@ -1,132 +0,0 @@
-"""
-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()