ApexAlgo v2.2.0 — reproducible backtests

backend/core/database.py

@@ -77,6 +77,7 @@ def run_migrations():
                         close     REAL,
                         volume    REAL,
                         marketcap REAL,
+                        fetched_at DATETIME,
                         UNIQUE (exchange, symbol, timeframe, timestamp)
                     )
                 """))
@@ -96,6 +97,9 @@ def run_migrations():
                 conn.execute(text("CREATE INDEX IF NOT EXISTS ix_candles_timeframe ON candles (timeframe)"))
                 conn.execute(text("CREATE INDEX IF NOT EXISTS ix_candles_timestamp ON candles (timestamp)"))
                 logger.info("Migration: candles table rebuilt successfully.")
+            elif 'fetched_at' not in can_cols:
+                conn.execute(text("ALTER TABLE candles ADD COLUMN fetched_at DATETIME"))
+                logger.info("Migration: added 'fetched_at' to candles")
 
         # ── signals — deduplicate and add unique constraint ────────────────────
         if "signals" in existing_tables:
@@ -141,6 +145,36 @@ def run_migrations():
                 conn.execute(text("CREATE INDEX IF NOT EXISTS ix_signals_name      ON signals (name)"))
                 logger.info("Migration: signals table rebuilt with unique constraint.")
 
+        # ── bot_config_runs — from "set of configs" to a run log ───────────────
+        # The unique key grew from (bot, config) to (bot, config, slice, data),
+        # which SQLite cannot alter in place. Old rows keep counting as
+        # distinct configs (empty slice/data hash), so the total is unchanged.
+        if "bot_config_runs" in existing_tables:
+            run_cols = {c['name'] for c in inspector.get_columns('bot_config_runs')}
+            if 'slice_key' not in run_cols:
+                logger.info("Migration: rebuilding bot_config_runs table as a run log...")
+                conn.execute(text("""
+                    CREATE TABLE bot_config_runs_migration (
+                        id          INTEGER PRIMARY KEY,
+                        bot_name    VARCHAR NOT NULL,
+                        config_hash VARCHAR(32) NOT NULL,
+                        slice_key   VARCHAR(32) NOT NULL DEFAULT '',
+                        window_from DATETIME,
+                        window_to   DATETIME,
+                        data_hash   VARCHAR(64) NOT NULL DEFAULT '',
+                        run_at      DATETIME NOT NULL,
+                        CONSTRAINT uq_bot_config_runs_run UNIQUE (bot_name, config_hash, slice_key, data_hash)
+                    )
+                """))
+                conn.execute(text("""
+                    INSERT INTO bot_config_runs_migration (id, bot_name, config_hash, run_at)
+                    SELECT id, bot_name, config_hash, first_run_at FROM bot_config_runs
+                """))
+                conn.execute(text("DROP TABLE bot_config_runs"))
+                conn.execute(text("ALTER TABLE bot_config_runs_migration RENAME TO bot_config_runs"))
+                conn.execute(text("CREATE INDEX IF NOT EXISTS ix_bot_config_runs_bot_name ON bot_config_runs (bot_name)"))
+                logger.info("Migration: bot_config_runs table rebuilt.")
+
         # Composite indexes for hot query paths (idempotent)
         conn.execute(text("CREATE INDEX IF NOT EXISTS ix_positions_botname_status ON positions (bot_name, status)"))
         conn.execute(text("CREATE INDEX IF NOT EXISTS ix_orders_posid_side_status ON orders (position_id, side, status)"))

backend/engine/backtest.py

@@ -12,7 +12,10 @@
 from backend.models.signals import Signal
 from backend.models.orders import Order
 from backend.models.positions import Position
-from backend.engine.sizing import _num, _int, _naive_utc, sim_frictions, calculate_trade_amount, _record_config_run
+from backend.engine.sizing import (
+    _num, _int, _naive_utc, sim_frictions, calculate_trade_amount, _record_config_run,
+    backtest_pin, combined_fingerprint, slice_key,
+)
 from backend.core import bot_log_buffer as blb
 
 logger = logging.getLogger("apexalgo.bot_manager")
@@ -322,15 +325,34 @@ def simulate(db, bot, sym_contexts, exchange_name, run_backtest, *, check_exits,
     )
 
 
-def build_summary(db, bot, res: SimResult, sym_contexts) -> dict:
+def build_summary(db, bot, res: SimResult, sym_contexts, exchange_name=None) -> dict:
     """`last_backtest_summary`: the numbers the bot card and analytics show,
     including the drawdown the gate actually enforces (the closed-trade curve
-    in the UI understates intra-trade dips)."""
+    in the UI understates intra-trade dips), plus the run's identity — slice,
+    data hash and both variant counters — for reproducibility."""
     closed_bt = db.query(Position.profit_abs).filter(
         Position.bot_name == bot.name, Position.mode == "backtest", Position.status == "closed"
     ).all()
     pnls = [float(p[0] or 0) for p in closed_bt]
     wins = sum(1 for p in pnls if p > 0)
+
+    data_from = _naive_utc(res.timeline[0][0]) if res.timeline else None
+    data_to = _naive_utc(res.timeline[-1][0]) if res.timeline else None
+    # Slice = the candles walked: the pinned window when there is one, else
+    # the realized range (which moves with every unpinned run)
+    pin_from, pin_to = backtest_pin(bot.settings)
+    pinned = pin_from is not None
+    window_from, window_to = (pin_from, pin_to) if pinned else (data_from, data_to)
+    hash_by_symbol = {c["symbol"]: c["data_hash"] for c in sym_contexts if c.get("data_hash")}
+    data_hash = combined_fingerprint(hash_by_symbol)
+    slice_hash = slice_key(exchange_name or bot.settings.get("data_exchange", "okx"),
+                           [c["symbol"] for c in sym_contexts], bot.settings.get("timeframe"), window_from, window_to)
+    # Same slice as the previous saved run but different candles underneath
+    # (re-download, gap repair, exchange restatement) — None when the slice
+    # moved, because then a differing hash says nothing
+    prev = (bot.settings or {}).get("last_backtest_summary") or {}
+    data_changed = (prev["data_hash"] != data_hash) if prev.get("data_hash") and prev.get("slice_key") == slice_hash else None
+    variants, variants_on_slice = _record_config_run(db, bot.name, bot.settings, slice_hash, data_hash, window_from, window_to)
     summary = {
         "trades": len(pnls),
         "wins": wins,
@@ -344,8 +366,8 @@ def build_summary(db, bot, res: SimResult, sym_contexts) -> dict:
         "candles": sum(len(c["df"]) for c in sym_contexts),
         # Data range the backtest walked — lets the analytics page
         # measure flat periods before the first / after the last trade
-        "data_from": res.timeline[0][0].isoformat() if res.timeline else None,
-        "data_to": res.timeline[-1][0].isoformat() if res.timeline else None,
+        "data_from": data_from.isoformat() if data_from else None,
+        "data_to": data_to.isoformat() if data_to else None,
         # Buy & hold over the same walked range, per symbol — the
         # only fair benchmark for the backtest return above
         "buy_hold": {
@@ -357,9 +379,19 @@ def build_summary(db, bot, res: SimResult, sym_contexts) -> dict:
             }
             for c in sym_contexts if len(c["df"]) > 0
         },
-        # Distinct configurations this bot has backtested — plain
-        # tweak-awareness, no judgement attached
-        "variants": _record_config_run(db, bot.name, bot.settings),
+        # Distinct configurations this bot has backtested — ever, and on
+        # exactly this slice of data — plain tweak-awareness, no judgement
+        "variants": variants,
+        "variants_on_slice": variants_on_slice,
+        # Run identity: pin + slice + raw-candle hash make the run reproducible
+        # and let the next run on the same slice detect altered data
+        "pinned": pinned,
+        "window_from": window_from.isoformat() if window_from else None,
+        "window_to": window_to.isoformat() if window_to else None,
+        "slice_key": slice_hash,
+        "data_hash": data_hash,
+        "data_hash_by_symbol": hash_by_symbol,
+        "data_changed": data_changed,
         "finished_at": datetime.now(timezone.utc).isoformat(),
     }
     return summary

backend/engine/data_verify.py

@@ -0,0 +1,110 @@
+"""Verify stored candles against the exchange.
+
+Stored candles are never overwritten by the poller or a download (only
+missing rows are added), so a backtest on them is reproducible — but an
+exchange can silently restate history (Binance most often the volume). This
+module re-fetches a window and diffs it against the local rows; with
+``accept=True`` the restated rows are overwritten (and stamped `fetched_at`),
+which is the one deliberate path that changes a slice's data hash.
+"""
+import logging
+import math
+import time
+from datetime import datetime, timezone
+
+from backend.models.candles import Candle
+from backend.engine.sizing import _naive_utc
+
+logger = logging.getLogger("apexalgo.data_verify")
+
+FIELDS = ("open", "high", "low", "close", "volume")
+REL_TOL = 1e-9
+MAX_DETAIL = 50
+
+
+def _differs(a, b) -> bool:
+    if a is None or b is None:
+        return (a is None) != (b is None)
+    try:
+        return not math.isclose(float(a), float(b), rel_tol=REL_TOL, abs_tol=0.0)
+    except (TypeError, ValueError):
+        return True
+
+
+def fetch_window(exch, symbol: str, timeframe: str, start: datetime, end: datetime, limit: int = 500, pause: float = 0.35):
+    """Paginated OHLCV between two naive-UTC datetimes (inclusive), as
+    ``{naive_utc_ts: (o, h, l, c, v)}``."""
+    start_ms = int(start.replace(tzinfo=timezone.utc).timestamp() * 1000)
+    end_ms = int(end.replace(tzinfo=timezone.utc).timestamp() * 1000)
+    out = {}
+    since = start_ms
+    while since <= end_ms:
+        batch = exch.fetch_ohlcv(symbol, timeframe, since=since, limit=limit)
+        if not batch:
+            break
+        for row in batch:
+            ts_ms = int(row[0])
+            if start_ms <= ts_ms <= end_ms:
+                ts = datetime.fromtimestamp(ts_ms / 1000.0, tz=timezone.utc).replace(tzinfo=None)
+                out[ts] = tuple(float(x) if x is not None else None for x in row[1:6])
+        last = int(batch[-1][0])
+        if last >= end_ms or len(batch) < 2 or last < since:
+            break
+        since = last + 1
+        time.sleep(pause)
+    return out
+
+
+def verify_window(db, exch, exchange_id: str, symbol: str, timeframe: str, start, end, accept: bool = False) -> dict:
+    """Diff the exchange's view of ``[start, end]`` against the stored rows.
+
+    Returns counts plus up to ``MAX_DETAIL`` restated rows with per-field
+    (local, exchange) pairs. ``accept`` overwrites restated local rows in
+    place (same ids, so signals keep their candle reference) and commits.
+    """
+    start = _naive_utc(start)
+    end = _naive_utc(end)
+    remote = fetch_window(exch, symbol, timeframe, start, end)
+    local_rows = db.query(Candle).filter(
+        Candle.exchange == exchange_id, Candle.symbol == symbol, Candle.timeframe == timeframe,
+        Candle.timestamp >= start, Candle.timestamp <= end,
+    ).all()
+    local = {_naive_utc(r.timestamp): r for r in local_rows}
+
+    restated, missing_local = [], 0
+    restated_total = accepted = 0
+    now = datetime.now(timezone.utc).replace(tzinfo=None)
+    for ts, values in sorted(remote.items()):
+        row = local.get(ts)
+        if row is None:
+            missing_local += 1
+            continue
+        diffs = {}
+        for field, new in zip(FIELDS, values):
+            old = getattr(row, field)
+            if _differs(old, new):
+                diffs[field] = {"local": old, "exchange": new}
+        if not diffs:
+            continue
+        restated_total += 1
+        if len(restated) < MAX_DETAIL:
+            restated.append({"timestamp": ts.isoformat(), "fields": diffs})
+        if accept:
+            for field, d in diffs.items():
+                setattr(row, field, d["exchange"])
+            row.fetched_at = now
+            accepted += 1
+    missing_exchange = sum(1 for ts in local if ts not in remote)
+    if accept and accepted:
+        db.commit()
+        logger.info("Accepted %d restated candle(s) from %s for %s/%s", accepted, exchange_id, symbol, timeframe)
+
+    return {
+        "exchange": exchange_id, "symbol": symbol, "timeframe": timeframe,
+        "from": start.isoformat(), "to": end.isoformat(),
+        "checked": len(remote), "local_rows": len(local),
+        "restated_count": restated_total, "restated": restated,
+        "missing_local": missing_local, "missing_exchange": missing_exchange,
+        "accepted": accepted,
+        "verified_at": now.isoformat(),
+    }

backend/engine/settings_validator.py

@@ -1,5 +1,6 @@
 import logging
 import re
+from datetime import datetime
 
 from backend.core.exchange_registry import get_exchange_timeframes
 from backend.engine.indicator_registry import get_spec
@@ -197,6 +198,20 @@ def validate_bot_settings(settings: dict, exchange_id: str | None = None) -> dic
         warnings.append(f"backtest_lookback ({lookback}) is short for the longest indicator window ({longest}): the first "
                         f"~{longest} candles are warm-up, leaving little to trade on. Use at least {longest + LOOKBACK_MARGIN}.")
 
+    # Pinned backtest window: both ends or neither, valid ISO, from < to
+    pin_raw = {k: settings.get(k) for k in ("backtest_from", "backtest_to")}
+    if any(pin_raw.values()):
+        if not all(pin_raw.values()):
+            errors.append("backtest_from and backtest_to must be set together (or both cleared to rerun against the latest data).")
+        else:
+            try:
+                _pf = datetime.fromisoformat(str(pin_raw["backtest_from"]).replace("Z", "+00:00"))
+                _pt = datetime.fromisoformat(str(pin_raw["backtest_to"]).replace("Z", "+00:00"))
+                if _pf >= _pt:
+                    errors.append("backtest_from must be before backtest_to.")
+            except (ValueError, TypeError):
+                errors.append("backtest_from / backtest_to must be ISO 8601 timestamps.")
+
     # Trade settings
     trade_settings = settings.get("trade_settings", {})
     entry_ts = trade_settings.get("entry", {})

backend/engine/sizing.py

@@ -2,7 +2,7 @@
 and the live tick. Pure functions over settings + DB — no engine state."""
 import json
 import logging
-from hashlib import md5
+from hashlib import md5, sha256
 from datetime import datetime, timezone
 from sqlalchemy import text, func
 from backend.models.bots import BotConfig
@@ -59,10 +59,12 @@ def _naive_utc(ts):
 
 
 # Settings that do not change what the strategy does on the data: layout,
-# order routing, live sizing and the engine's own runtime bookkeeping.
+# order routing, live sizing, the pinned backtest window (which data, not
+# what the strategy does on it) and the engine's own runtime bookkeeping.
 _NON_STRATEGY_KEYS = frozenset({
     "ui_layout", "api_execution", "api_key_name", "live_allocation_pct", "max_order_value",
-    "backtest_on_start", "last_backtest_summary", "last_backtest_max_drawdown",
+    "backtest_on_start", "backtest_from", "backtest_to",
+    "last_backtest_summary", "last_backtest_max_drawdown",
     "last_stop_reason", "drawdown_peak_reset_at", "live_starting_capital",
 })
 
@@ -74,14 +76,72 @@ def _config_fingerprint(settings: dict) -> str:
     return md5(json.dumps(relevant, sort_keys=True, default=str).encode()).hexdigest()
 
 
-def _record_config_run(db, bot_name: str, settings: dict) -> int:
-    """Register this configuration as backtested and return the number of
-    distinct configurations the bot has run so far (this one included)."""
+def data_fingerprint(df) -> str:
+    """sha256 of the raw OHLCV rows a backtest walked (sorted by timestamp).
+    `repr` on the floats makes it byte-stable across runs, so two runs on the
+    same slice hash equal unless a candle was actually altered — the check
+    behind the "historical data changed" warning."""
+    h = sha256()
+    if df is None or len(df) == 0:
+        return h.hexdigest()
+    cols = ["timestamp", "open", "high", "low", "close", "volume"]
+    for ts, o, hi, lo, c, v in df.sort_values("timestamp")[cols].itertuples(index=False, name=None):
+        ts = _naive_utc(ts)
+        h.update(f"{ts.isoformat() if ts is not None else ''},{o!r},{hi!r},{lo!r},{c!r},{v!r}\n".encode())
+    return h.hexdigest()
+
+
+def backtest_pin(settings) -> tuple:
+    """``(window_from, window_to)`` as naive UTC datetimes when the bot's
+    backtest window is pinned (``backtest_from``/``backtest_to`` ISO strings),
+    else ``(None, None)``. A half or malformed pin counts as no pin."""
+    out = []
+    for key in ("backtest_from", "backtest_to"):
+        raw = (settings or {}).get(key)
+        if not raw:
+            return None, None
+        try:
+            out.append(_naive_utc(datetime.fromisoformat(str(raw).replace("Z", "+00:00"))))
+        except (TypeError, ValueError):
+            return None, None
+    if out[0] >= out[1]:
+        return None, None
+    return out[0], out[1]
+
+
+def combined_fingerprint(by_symbol: dict) -> str:
+    """One hash over the per-symbol data hashes (sorted by symbol)."""
+    return sha256("|".join(f"{s}:{by_symbol[s]}" for s in sorted(by_symbol)).encode()).hexdigest()
+
+
+def slice_key(exchange: str, symbols, timeframe: str, window_from, window_to) -> str:
+    """Identity of a backtest slice: which candles were walked. Same pairs,
+    timeframe and window on the same exchange = same slice, whatever the
+    strategy settings are."""
+    f = _naive_utc(window_from)
+    t = _naive_utc(window_to)
+    parts = [str(exchange or ""), ",".join(sorted(str(s) for s in (symbols or []))), str(timeframe or ""),
+             f.isoformat() if f else "", t.isoformat() if t else ""]
+    return md5("|".join(parts).encode()).hexdigest()
+
+
+def _record_config_run(db, bot_name: str, settings: dict, slice_hash: str = "", data_hash: str = "",
+                       window_from=None, window_to=None) -> tuple:
+    """Log this backtest run and return ``(variants_total, variants_on_slice)``:
+    distinct configurations the bot has ever run, and distinct configurations
+    run on exactly this slice of data (this one included). One row per
+    (config, slice, data) — a rerun on unchanged data is a no-op."""
     db.execute(
-        text("INSERT OR IGNORE INTO bot_config_runs (bot_name, config_hash, first_run_at) VALUES (:bn, :h, :ts)"),
-        {"bn": bot_name, "h": _config_fingerprint(settings), "ts": datetime.now(timezone.utc).replace(tzinfo=None)},
+        text("INSERT OR IGNORE INTO bot_config_runs (bot_name, config_hash, slice_key, window_from, window_to, data_hash, run_at) "
+             "VALUES (:bn, :h, :sk, :wf, :wt, :dh, :ts)"),
+        {"bn": bot_name, "h": _config_fingerprint(settings), "sk": slice_hash or "", "wf": _naive_utc(window_from),
+         "wt": _naive_utc(window_to), "dh": data_hash or "", "ts": datetime.now(timezone.utc).replace(tzinfo=None)},
     )
-    return int(db.execute(text("SELECT COUNT(*) FROM bot_config_runs WHERE bot_name = :bn"), {"bn": bot_name}).scalar() or 0)
+    total = int(db.execute(text("SELECT COUNT(DISTINCT config_hash) FROM bot_config_runs WHERE bot_name = :bn"),
+                           {"bn": bot_name}).scalar() or 0)
+    on_slice = int(db.execute(text("SELECT COUNT(DISTINCT config_hash) FROM bot_config_runs WHERE bot_name = :bn AND slice_key = :sk"),
+                              {"bn": bot_name, "sk": slice_hash or ""}).scalar() or 0)
+    return total, on_slice
 
 
 def sim_frictions(settings):