Narrow bot to USDCAD H4 Grade B + paper-trade tracker + /status command

Backtest-driven config change. The previous wide-scan setup (4 pairs ×
M15+H1, alert grades A+/A) was empirically losing money in every
backtested permutation. The 2-year backtest identified one robust niche
(USDCAD H4 Grade B: train +20.8 R / test +14.0 R / PF 1.52 / max DD
-9.6 R, positive in every 6-month sub-period). This commit restricts
the live bot to that niche and runs it in paper-trade mode for
validation before any real capital is risked.

Config:
  SCAN_PAIRS         ["USDCAD=X"]   (was 4 pairs)
  SCAN_TIMEFRAMES    ["H4"]          (was M15+H1)
  ALERT_GRADES       {"B"}           (was A+/A — those are anti-edge)
  AUTO_SCAN_INTERVAL 1800            (was 600 — H4 candles only print
                                       every 4h, polling more wastes credits)
  PAPER_TRADE_MODE   True            (alerts get [PAPER] prefix and
                                       are NOT for real money)

New machinery:
- _open_paper_position() / _check_paper_positions(): on every scan,
  walks any open paper trades forward against the latest H4 bars using
  the same first-touch SL/TP1 logic as backtest.simulate_trade.
- paper_open.json + paper_trades.csv: persistent state across restarts.
  Open positions survive bot restarts; closed trades append to the CSV.
- /status: uptime, scan count, last scan time, alert count, open paper
  positions, win rate, total R, profit factor.
- /paper: full ledger — open trades + last 10 closed.
- INFO-level logging in main() so the user sees scan activity in
  stdout instead of wondering whether the bot is alive.
- Help text updated to reflect the calibrated strategy and warn that
  A+/A grades are anti-edge in backtest.
- gitignore: paper_open.json + paper_trades.csv + backtest_results_*/

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

.gitignore

@@ -15,3 +15,12 @@ venv/
 # OS
 .DS_Store
 Thumbs.db
+
+# Backtest data + outputs (regenerable from seed_history.py / backtest.py)
+history/
+backtest_results/
+backtest_results_*/
+
+# Paper-trade ledger (live state — do not version control)
+paper_open.json
+paper_trades.csv

analyze_trades.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python3
+"""
+Phase 1 analysis: post-hoc filters on an existing trades.csv.
+
+Asks "which simple filters, applied to the trades the engine already
+produced, would have moved total R toward zero or positive?" — without
+re-running the backtest engine.
+
+Discipline: candidate filters are derived on the **first 70% of trades
+by entry_time** (train slice), then evaluated on the **last 30%** (test
+slice). This is the same time-based split the meta-model uses, so the
+numbers are comparable.
+
+Outputs (printed + saved to backtest_results_all/phase1_summary.txt):
+  - Session-of-day breakdown
+  - Day-of-week breakdown
+  - Pair × regime cross-tab
+  - Per-pair contribution
+  - Combined-filter sweep (drop-worst-hours / drop-worst-pairs / both)
+
+Usage:
+    python analyze_trades.py
+    python analyze_trades.py --trades backtest_results_all/trades.csv
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+import pandas as pd
+
+
+def fmt(df: pd.DataFrame) -> pd.DataFrame:
+    return df.round(3)
+
+
+def split(df: pd.DataFrame, train_frac: float = 0.7):
+    df = df.sort_values("entry_time").reset_index(drop=True)
+    cutoff = int(len(df) * train_frac)
+    return df.iloc[:cutoff].copy(), df.iloc[cutoff:].copy()
+
+
+def stats(df: pd.DataFrame) -> dict:
+    if len(df) == 0:
+        return {"n": 0, "wr": 0.0, "avg_r": 0.0, "total_r": 0.0, "pf": 0.0}
+    wins = (df["r_multiple"] > 0).sum()
+    losses = (df["r_multiple"] <= 0).sum()
+    gross_win = df.loc[df["r_multiple"] > 0, "r_multiple"].sum()
+    gross_loss = -df.loc[df["r_multiple"] <= 0, "r_multiple"].sum()
+    return {
+        "n":       len(df),
+        "wr":      wins / len(df) * 100,
+        "avg_r":   df["r_multiple"].mean(),
+        "total_r": df["r_multiple"].sum(),
+        "pf":      gross_win / gross_loss if gross_loss > 0 else float("inf"),
+    }
+
+
+def stats_line(label: str, df: pd.DataFrame) -> str:
+    s = stats(df)
+    return (
+        f"{label:<40s}"
+        f"  n={s['n']:>5}  WR={s['wr']:5.1f}%"
+        f"  avgR={s['avg_r']:+6.3f}  totalR={s['total_r']:+9.2f}  PF={s['pf']:.2f}"
+    )
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="Phase 1 post-hoc analysis on backtest trades.")
+    ap.add_argument("--trades", default="backtest_results_all/trades.csv")
+    ap.add_argument("--out", default="backtest_results_all/phase1_summary.txt")
+    ap.add_argument("--train-frac", type=float, default=0.7)
+    args = ap.parse_args()
+
+    p = Path(args.trades)
+    if not p.exists():
+        print(f"❌ {p} not found")
+        return 1
+
+    df = pd.read_csv(p)
+    df["entry_time"] = pd.to_datetime(df["entry_time"])
+    df["hour"] = df["entry_time"].dt.hour
+    df["dow"] = df["entry_time"].dt.dayofweek
+    df["winner"] = (df["r_multiple"] > 0).astype(int)
+
+    train, test = split(df, args.train_frac)
+    out_lines = []
+
+    def emit(line: str = ""):
+        print(line)
+        out_lines.append(line)
+
+    emit("=" * 92)
+    emit(f"PHASE 1 ANALYSIS — {len(df):,} trades total")
+    emit(f"  full:  {df['entry_time'].min().date()} → {df['entry_time'].max().date()}")
+    emit(f"  train: {train['entry_time'].min().date()} → {train['entry_time'].max().date()}  ({len(train):,} trades)")
+    emit(f"  test:  {test['entry_time'].min().date()} → {test['entry_time'].max().date()}  ({len(test):,} trades)")
+    emit("=" * 92)
+    emit()
+    emit(stats_line("BASELINE  full",  df))
+    emit(stats_line("BASELINE  train", train))
+    emit(stats_line("BASELINE  test",  test))
+    emit()
+
+    # ── Per-hour (training set, used for filter selection) ──────────
+    emit("─" * 92)
+    emit("HOUR (UTC) — train slice  (used for filter selection)")
+    emit("─" * 92)
+    hr_stats = train.groupby("hour").apply(
+        lambda x: pd.Series(stats(x)), include_groups=False,
+    ).round(3)
+    emit(hr_stats.to_string())
+    emit()
+
+    # Bottom-K hours by avg_r
+    bad_hours_4 = hr_stats.sort_values("avg_r").head(4).index.tolist()
+    bad_hours_6 = hr_stats.sort_values("avg_r").head(6).index.tolist()
+    emit(f"  Worst 4 hours by avg_r (train): {sorted(bad_hours_4)}")
+    emit(f"  Worst 6 hours by avg_r (train): {sorted(bad_hours_6)}")
+    emit()
+
+    # ── Per-DOW (training set) ──────────────────────────────────────
+    emit("─" * 92)
+    emit("DAY OF WEEK — train slice  (0=Mon, 4=Fri)")
+    emit("─" * 92)
+    dow_stats = train.groupby("dow").apply(
+        lambda x: pd.Series(stats(x)), include_groups=False,
+    ).round(3)
+    emit(dow_stats.to_string())
+    emit()
+    bad_dow = dow_stats.sort_values("avg_r").head(2).index.tolist()
+    emit(f"  Worst 2 days by avg_r (train): {sorted(bad_dow)}")
+    emit()
+
+    # ── Per-pair (training set) ─────────────────────────────────────
+    emit("─" * 92)
+    emit("PAIR — train slice")
+    emit("─" * 92)
+    pair_stats = train.groupby("pair").apply(
+        lambda x: pd.Series(stats(x)), include_groups=False,
+    ).round(3)
+    emit(pair_stats.to_string())
+    emit()
+    bad_pairs = pair_stats.sort_values("total_r").head(3).index.tolist()
+    emit(f"  Worst 3 pairs by total_r (train): {bad_pairs}")
+    emit()
+
+    # ── Per-regime × direction (training set) ───────────────────────
+    emit("─" * 92)
+    emit("REGIME × DIRECTION — train slice")
+    emit("─" * 92)
+    reg_stats = train.groupby(["regime", "direction"]).apply(
+        lambda x: pd.Series(stats(x)), include_groups=False,
+    ).round(3)
+    emit(reg_stats.to_string())
+    emit()
+
+    # ── Filter sweep ON THE TEST SET (rule selected on train) ───────
+    emit("=" * 92)
+    emit("FILTER SWEEP — applied to TEST slice (rule chosen from train, no leakage)")
+    emit("=" * 92)
+    test_filtered_dropworst4hours = test[~test["hour"].isin(bad_hours_4)]
+    test_filtered_dropworst6hours = test[~test["hour"].isin(bad_hours_6)]
+    test_filtered_droppairs       = test[~test["pair"].isin(bad_pairs)]
+    test_filtered_droprangingsell = test[~((test["regime"] == "ranging") & (test["direction"] == "SELL"))]
+    test_filtered_only_ranging    = test[test["regime"] == "ranging"]
+    test_filtered_drop_dow        = test[~test["dow"].isin(bad_dow)]
+
+    test_combo_hr_pair = test[
+        (~test["hour"].isin(bad_hours_4)) & (~test["pair"].isin(bad_pairs))
+    ]
+    test_combo_hr_pair_dow = test[
+        (~test["hour"].isin(bad_hours_4))
+        & (~test["pair"].isin(bad_pairs))
+        & (~test["dow"].isin(bad_dow))
+    ]
+
+    emit()
+    emit(stats_line("test  baseline (no filter)",         test))
+    emit(stats_line("test  drop 4 worst hours",           test_filtered_dropworst4hours))
+    emit(stats_line("test  drop 6 worst hours",           test_filtered_dropworst6hours))
+    emit(stats_line("test  drop 3 worst pairs",           test_filtered_droppairs))
+    emit(stats_line("test  drop 2 worst DOWs",            test_filtered_drop_dow))
+    emit(stats_line("test  ONLY ranging regime",          test_filtered_only_ranging))
+    emit(stats_line("test  hours+pairs (combined)",       test_combo_hr_pair))
+    emit(stats_line("test  hours+pairs+DOW (combined)",   test_combo_hr_pair_dow))
+    emit()
+
+    # ── Same filters applied to TRAIN slice for reference ───────────
+    emit("─" * 92)
+    emit("Same filters — TRAIN slice for reference (these are the in-sample numbers)")
+    emit("─" * 92)
+    train_drop_hours = train[~train["hour"].isin(bad_hours_4)]
+    train_drop_pairs = train[~train["pair"].isin(bad_pairs)]
+    train_combo      = train[(~train["hour"].isin(bad_hours_4)) & (~train["pair"].isin(bad_pairs))]
+
+    emit(stats_line("train baseline (no filter)",         train))
+    emit(stats_line("train drop 4 worst hours",           train_drop_hours))
+    emit(stats_line("train drop 3 worst pairs",           train_drop_pairs))
+    emit(stats_line("train hours+pairs (combined)",       train_combo))
+    emit()
+
+    # ── Verdict ─────────────────────────────────────────────────────
+    emit("=" * 92)
+    emit("VERDICT")
+    emit("=" * 92)
+    test_base = stats(test)
+    test_combo = stats(test_combo_hr_pair_dow)
+    delta_r = test_combo["total_r"] - test_base["total_r"]
+    delta_avgr = test_combo["avg_r"] - test_base["avg_r"]
+    emit(f"  Best combined filter (hours+pairs+DOW) on TEST set:")
+    emit(f"    {test_base['n']} → {test_combo['n']} trades  "
+         f"({test_combo['n']/test_base['n']*100:.0f}% retained)")
+    emit(f"    total R: {test_base['total_r']:+.2f} → {test_combo['total_r']:+.2f}  "
+         f"(Δ {delta_r:+.2f})")
+    emit(f"    avg R:   {test_base['avg_r']:+.3f} → {test_combo['avg_r']:+.3f}  "
+         f"(Δ {delta_avgr:+.3f})")
+    emit(f"    win rate: {test_base['wr']:.1f}% → {test_combo['wr']:.1f}%")
+    emit()
+    if test_combo["total_r"] > 0:
+        emit("  ✅ Filtered subset is profitable on the held-out test set.")
+        emit("     → Phase 2 (geometry grid) may be unnecessary; this filter alone may suffice.")
+    elif test_combo["avg_r"] > test_base["avg_r"]:
+        emit("  ⚠️  Filtered subset improves but is still negative.")
+        emit("     → Phase 2 (geometry grid) needed to find profitable structure.")
+    else:
+        emit("  ❌ Filters do not survive the train/test split — rejected as noise.")
+        emit("     → Skip Phase 1's filter list; rely on Phase 2 (geometry) instead.")
+
+    Path(args.out).parent.mkdir(parents=True, exist_ok=True)
+    Path(args.out).write_text("\n".join(out_lines))
+    emit()
+    emit(f"📁 Saved: {args.out}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

backtest.py

@@ -0,0 +1,712 @@
+#!/usr/bin/env python3
+"""
+Walk-forward backtest harness for the Forex signal engine.
+
+Replays historical OHLCV (loaded from history/<PAIR>_<TF>.csv) through
+the existing SignalAggregator with strict no-look-ahead semantics: at
+each primary-TF bar close, the engine sees only bars that have *fully
+closed by that moment* — never the bar still in formation.
+
+Trade simulation: first-touch (SL vs TP1) on the primary timeframe,
+worst-case if both touched in the same bar (assume SL first), with
+per-pip spread cost deducted from the final R-multiple.
+
+Outputs:
+    backtest_results/trades.csv     — every simulated trade
+    backtest_results/summary.txt    — per-grade win rate / avg R / PF
+    backtest_results/equity.png     — cumulative R curve (if --plot)
+    backtest_results/by_grade.png   — win rate by grade (if --plot)
+
+Usage:
+    python backtest.py --primary M15 --pairs EURUSD
+    python backtest.py --primary H1 --pairs EURUSD GBPUSD USDJPY --plot
+    python backtest.py --primary M15 --grades A+ A
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+import pandas as pd
+
+from config import PlatformConfig, SignalStrength, Timeframe, MAJOR_PAIRS
+from data_provider import DataUnavailableError, ForexDataProvider
+from signal_aggregator import SignalAggregator, SignalReport
+
+logger = logging.getLogger(__name__)
+
+HISTORY_DIR = Path("history")
+RESULTS_DIR = Path("backtest_results")
+
+_TF_MAP = {
+    "M1":  Timeframe.M1,  "M5":  Timeframe.M5,
+    "M15": Timeframe.M15, "M30": Timeframe.M30,
+    "H1":  Timeframe.H1,  "H4":  Timeframe.H4,
+    "D1":  Timeframe.D1,  "W1":  Timeframe.W1,
+}
+
+_TF_INTERVAL = {
+    Timeframe.M1:  pd.Timedelta(minutes=1),
+    Timeframe.M5:  pd.Timedelta(minutes=5),
+    Timeframe.M15: pd.Timedelta(minutes=15),
+    Timeframe.M30: pd.Timedelta(minutes=30),
+    Timeframe.H1:  pd.Timedelta(hours=1),
+    Timeframe.H4:  pd.Timedelta(hours=4),
+    Timeframe.D1:  pd.Timedelta(days=1),
+    Timeframe.W1:  pd.Timedelta(weeks=1),
+}
+
+# Confirmation TFs per primary — same shape Telegram bot uses
+_TF_CONFIRMATIONS: Dict[Timeframe, List[Timeframe]] = {
+    Timeframe.M15: [Timeframe.H1,  Timeframe.D1],
+    Timeframe.M30: [Timeframe.H1,  Timeframe.D1],
+    Timeframe.H1:  [Timeframe.M15, Timeframe.D1],
+    Timeframe.H4:  [Timeframe.H1,  Timeframe.D1],
+    Timeframe.D1:  [Timeframe.H4],
+}
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  Stub news engine (no historical news data — returns neutral)
+# ══════════════════════════════════════════════════════════════════════
+
+class _StubFundamentalContext:
+    """Placeholder — backtest doesn't replay news, so context is empty."""
+    pass
+
+
+class StubNewsEngine:
+    """
+    Drop-in for `NewsEngine` in backtest mode. Returns neutral sentiment
+    and 'clear' event risk for every call. The signal aggregator's news
+    adjustments thus become no-ops, isolating the test to pure technicals.
+    """
+
+    def get_fundamental_context(self):
+        return _StubFundamentalContext()
+
+    def get_event_risk_level(self, symbol, context):
+        return ("clear", "")
+
+    def get_pair_sentiment(self, symbol, context):
+        return (0.0, "")
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  Backtest data provider (slices history at "as of" timestamp)
+# ══════════════════════════════════════════════════════════════════════
+
+class BacktestProvider(ForexDataProvider):
+    """
+    Replaces network fetches with in-memory slices of preloaded OHLCV.
+
+    `set_time(ts)` sets the cut-off; `fetch_ohlcv` returns only bars whose
+    *close* time is ≤ ts. Bars that are still in formation (open ≤ ts but
+    close > ts) are excluded — this is the no-look-ahead guarantee.
+    """
+
+    def __init__(
+        self,
+        config: PlatformConfig,
+        history: Dict[Tuple[str, Timeframe], pd.DataFrame],
+    ):
+        super().__init__(config)
+        self._history = history
+        self._current_ts: Optional[pd.Timestamp] = None
+
+    def set_time(self, ts: pd.Timestamp) -> None:
+        if ts.tzinfo is None:
+            ts = ts.tz_localize("UTC")
+        self._current_ts = ts
+
+    def fetch_ohlcv(
+        self,
+        symbol: str,
+        timeframe: Timeframe,
+        lookback_days: Optional[int] = None,
+    ) -> pd.DataFrame:
+        df = self._history.get((symbol, timeframe))
+        if df is None:
+            raise DataUnavailableError(
+                f"No history loaded for {symbol} {timeframe.value}. "
+                f"Run seed_history.py with --tfs including this TF."
+            )
+        if self._current_ts is not None:
+            interval = _TF_INTERVAL[timeframe]
+            df = df[df.index + interval <= self._current_ts]
+        if df.empty:
+            raise DataUnavailableError(
+                f"No bars before {self._current_ts} for {symbol} {timeframe.value}"
+            )
+        return df
+
+    def prefetch_batch(self, symbols, timeframes, lookback_days=None):
+        return  # no-op: data is in memory
+
+    def get_spread_estimate(self, symbol: str) -> float:
+        # Backtest doesn't need a live spread estimate; return a fixed value
+        # so the aggregator's spread field is filled but does not affect SL/TP.
+        return 0.0
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  History loader
+# ══════════════════════════════════════════════════════════════════════
+
+def load_history(
+    pairs: List[str],
+    timeframes: List[Timeframe],
+    history_dir: Path = HISTORY_DIR,
+) -> Dict[Tuple[str, Timeframe], pd.DataFrame]:
+    """
+    Load CSVs produced by seed_history.py into a dict keyed by (symbol, tf).
+
+    Drops CSVs that are missing — caller decides whether to abort or continue
+    with reduced TF coverage.
+    """
+    history: Dict[Tuple[str, Timeframe], pd.DataFrame] = {}
+    missing: List[str] = []
+
+    for symbol in pairs:
+        for tf in timeframes:
+            csv_p = history_dir / f"{symbol.replace('=X', '')}_{tf.value}.csv"
+            if not csv_p.exists():
+                missing.append(str(csv_p))
+                continue
+            df = pd.read_csv(csv_p, index_col="Datetime", parse_dates=["Datetime"])
+            if df.index.tz is None:
+                df.index = df.index.tz_localize("UTC")
+            history[(symbol, tf)] = df.sort_index()
+            logger.info(
+                f"loaded {symbol} {tf.value}: {len(df):,} bars "
+                f"({df.index.min().date()} → {df.index.max().date()})"
+            )
+
+    if missing:
+        logger.warning(f"missing CSVs ({len(missing)}): {missing[:5]}…")
+
+    return history
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  Trade record + first-touch simulator
+# ══════════════════════════════════════════════════════════════════════
+
+@dataclass
+class Trade:
+    symbol: str
+    pair_display: str
+    primary_tf: str
+    grade: str
+    direction: str
+    entry_time: pd.Timestamp
+    entry_price: float
+    sl: float
+    tp1: float
+    tp2: float
+    risk_pips: float
+    reward_pips: float
+    score: float
+    confidence: float
+    regime: str
+    mtf_aligned: bool
+
+    # Filled by simulation
+    exit_time: Optional[pd.Timestamp] = None
+    exit_price: float = 0.0
+    outcome: str = "OPEN"     # WIN_TP1, LOSS_SL, TIMEOUT, INVALID
+    bars_held: int = 0
+    r_multiple: float = 0.0   # net R after spread cost
+
+
+def simulate_trade(
+    trade: Trade,
+    forward_bars: pd.DataFrame,
+    max_bars: int,
+    spread_pips: float,
+) -> Trade:
+    """
+    First-touch simulation on the primary timeframe.
+
+    Conservatives:
+    - If both SL and TP1 touch within the same bar, assume SL first.
+    - Spread cost (entry + exit round-trip) is deducted from final R.
+    - TP2 is recorded but not used in this baseline simulation — we cap at
+      TP1 to keep the R-multiple math clean. Extend with breakeven-trail
+      management later if needed.
+    """
+    if trade.risk_pips <= 0:
+        trade.outcome = "INVALID"
+        return trade
+
+    spread_r = spread_pips / trade.risk_pips
+    bars_examined = 0
+
+    for ts, bar in forward_bars.iterrows():
+        bars_examined += 1
+        if bars_examined > max_bars:
+            break
+
+        bar_high = bar["High"]
+        bar_low  = bar["Low"]
+
+        if trade.direction == "BUY":
+            sl_hit = bar_low  <= trade.sl
+            tp_hit = bar_high >= trade.tp1
+        else:
+            sl_hit = bar_high >= trade.sl
+            tp_hit = bar_low  <= trade.tp1
+
+        if sl_hit:
+            trade.outcome    = "LOSS_SL"
+            trade.exit_time  = ts
+            trade.exit_price = trade.sl
+            trade.bars_held  = bars_examined
+            trade.r_multiple = -1.0 - spread_r
+            return trade
+
+        if tp_hit:
+            trade.outcome    = "WIN_TP1"
+            trade.exit_time  = ts
+            trade.exit_price = trade.tp1
+            trade.bars_held  = bars_examined
+            # TP1 sits at risk × RR away from entry → R = reward_pips / risk_pips
+            trade.r_multiple = (trade.reward_pips / trade.risk_pips) - spread_r
+            return trade
+
+    # Timeout — exit at last seen close
+    if bars_examined == 0:
+        trade.outcome = "INVALID"
+        return trade
+
+    last_idx = bars_examined - 1
+    last = forward_bars.iloc[last_idx]
+    last_ts = forward_bars.index[last_idx]
+
+    if trade.direction == "BUY":
+        pnl_price = last["Close"] - trade.entry_price
+    else:
+        pnl_price = trade.entry_price - last["Close"]
+
+    sl_distance_price = abs(trade.entry_price - trade.sl)
+    pnl_r = pnl_price / sl_distance_price if sl_distance_price > 0 else 0.0
+
+    trade.outcome    = "TIMEOUT"
+    trade.exit_time  = last_ts
+    trade.exit_price = float(last["Close"])
+    trade.bars_held  = bars_examined
+    trade.r_multiple = pnl_r - spread_r
+    return trade
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  Backtester (the replay loop)
+# ══════════════════════════════════════════════════════════════════════
+
+class Backtester:
+    def __init__(
+        self,
+        primary_tf: Timeframe,
+        pairs: List[str],
+        history: Dict[Tuple[str, Timeframe], pd.DataFrame],
+        target_grades: List[str],
+        max_hold_bars: int = 100,
+        spread_pips: float = 1.0,
+        warmup_bars: int = 250,
+    ):
+        self.primary_tf      = primary_tf
+        self.pairs           = pairs
+        self.history         = history
+        self.target_grades   = set(target_grades)
+        self.max_hold_bars   = max_hold_bars
+        self.spread_pips     = spread_pips
+        self.warmup_bars     = warmup_bars
+        self.confirmation_tfs = _TF_CONFIRMATIONS.get(primary_tf, [])
+
+    def _build_config(self, symbol: str) -> PlatformConfig:
+        cfg = PlatformConfig()
+        cfg.primary_timeframe       = self.primary_tf
+        cfg.confirmation_timeframes = list(self.confirmation_tfs)
+        cfg.watch_pairs             = [symbol]
+        return cfg
+
+    def _make_aggregator(self, config: PlatformConfig, provider: BacktestProvider) -> SignalAggregator:
+        agg = SignalAggregator(config, data_provider=provider)
+        # Replace live news engine with neutral stub (no historical news data)
+        agg.news_engine = StubNewsEngine()
+        return agg
+
+    def run(self) -> List[Trade]:
+        trades: List[Trade] = []
+        provider = BacktestProvider(PlatformConfig(), self.history)
+
+        for symbol in self.pairs:
+            primary_df = self.history.get((symbol, self.primary_tf))
+            if primary_df is None or primary_df.empty:
+                logger.warning(f"{symbol}: no primary {self.primary_tf.value} data, skipping")
+                continue
+
+            config = self._build_config(symbol)
+            aggregator = self._make_aggregator(config, provider)
+
+            interval = _TF_INTERVAL[self.primary_tf]
+            timestamps = primary_df.index
+            n_total = len(timestamps)
+
+            print(f"\n📊 {symbol} {self.primary_tf.value}: "
+                  f"{n_total:,} bars, warmup={self.warmup_bars}")
+
+            open_until: Optional[pd.Timestamp] = None  # block overlapping entries
+            n_signals = 0
+
+            for i, bar_open_ts in enumerate(timestamps):
+                if i < self.warmup_bars:
+                    continue
+                if open_until is not None and bar_open_ts < open_until:
+                    continue  # previous trade still open — skip
+
+                # Run engine "as if it's now this bar's close"
+                bar_close_ts = bar_open_ts + interval
+                provider.set_time(bar_close_ts)
+
+                try:
+                    report = aggregator.analyze_pair(symbol)
+                except DataUnavailableError:
+                    continue
+                except Exception as e:
+                    logger.debug(f"engine error at {bar_open_ts}: {e}")
+                    continue
+
+                if report.signal_quality not in self.target_grades:
+                    continue
+                if report.trade_setup is None:
+                    continue
+                if report.final_signal == SignalStrength.NEUTRAL:
+                    continue
+
+                ts = report.trade_setup
+                trade = Trade(
+                    symbol       = symbol,
+                    pair_display = report.display_name,
+                    primary_tf   = self.primary_tf.value,
+                    grade        = report.signal_quality,
+                    direction    = ts.direction,
+                    entry_time   = bar_open_ts,
+                    entry_price  = ts.entry_price,
+                    sl           = ts.stop_loss,
+                    tp1          = ts.take_profit_1,
+                    tp2          = ts.take_profit_2,
+                    risk_pips    = ts.risk_pips,
+                    reward_pips  = ts.reward_pips,
+                    score        = report.final_score,
+                    confidence   = report.final_confidence,
+                    regime       = report.market_regime,
+                    mtf_aligned  = report.mtf_agreement,
+                )
+
+                # Forward window for simulation
+                forward = primary_df.iloc[i + 1 : i + 1 + self.max_hold_bars]
+                if forward.empty:
+                    continue
+
+                trade = simulate_trade(
+                    trade,
+                    forward,
+                    max_bars=self.max_hold_bars,
+                    spread_pips=self.spread_pips,
+                )
+                trades.append(trade)
+                n_signals += 1
+
+                if trade.exit_time is not None:
+                    open_until = trade.exit_time
+
+                if n_signals % 25 == 0:
+                    print(f"   [{i:>6}/{n_total}]  {n_signals} trades so far "
+                          f"(latest: {trade.grade} {trade.direction} {trade.outcome} "
+                          f"R={trade.r_multiple:+.2f})")
+
+            print(f"   ✅ {symbol}: {n_signals} trades")
+
+        return trades
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  Reporting
+# ══════════════════════════════════════════════════════════════════════
+
+def trades_to_dataframe(trades: List[Trade]) -> pd.DataFrame:
+    rows = [
+        {
+            "entry_time":   t.entry_time,
+            "exit_time":    t.exit_time,
+            "symbol":       t.symbol,
+            "pair":         t.pair_display,
+            "tf":           t.primary_tf,
+            "grade":        t.grade,
+            "direction":    t.direction,
+            "entry":        t.entry_price,
+            "sl":           t.sl,
+            "tp1":          t.tp1,
+            "exit_price":   t.exit_price,
+            "outcome":      t.outcome,
+            "bars_held":    t.bars_held,
+            "r_multiple":   t.r_multiple,
+            "score":        t.score,
+            "confidence":   t.confidence,
+            "regime":       t.regime,
+            "mtf_aligned":  t.mtf_aligned,
+            "risk_pips":    t.risk_pips,
+            "reward_pips":  t.reward_pips,
+        }
+        for t in trades
+    ]
+    return pd.DataFrame(rows)
+
+
+def summarize(df: pd.DataFrame) -> str:
+    """Per-grade win rate / average R / profit factor summary."""
+    if df.empty:
+        return "No trades recorded."
+
+    lines: List[str] = []
+    lines.append("=" * 78)
+    lines.append(f"BACKTEST SUMMARY  ({len(df)} trades)")
+    lines.append(f"  Date range: {df['entry_time'].min()} → {df['entry_time'].max()}")
+    lines.append("=" * 78)
+    lines.append("")
+
+    overall = _grade_block("ALL", df)
+    lines.append(overall)
+    lines.append("")
+
+    for grade in ["A+", "A", "B", "C", "D"]:
+        sub = df[df["grade"] == grade]
+        if sub.empty:
+            continue
+        lines.append(_grade_block(grade, sub))
+        lines.append("")
+
+    # Per-pair breakdown
+    lines.append("─" * 78)
+    lines.append("PER-PAIR SUMMARY")
+    lines.append("─" * 78)
+    by_pair = df.groupby("pair").agg(
+        trades=("r_multiple", "count"),
+        win_rate=("r_multiple", lambda s: (s > 0).mean() * 100),
+        avg_r=("r_multiple", "mean"),
+        total_r=("r_multiple", "sum"),
+    ).round(2)
+    lines.append(by_pair.to_string())
+    lines.append("")
+
+    # Per-direction
+    lines.append("─" * 78)
+    lines.append("BY DIRECTION")
+    lines.append("─" * 78)
+    by_dir = df.groupby("direction").agg(
+        trades=("r_multiple", "count"),
+        win_rate=("r_multiple", lambda s: (s > 0).mean() * 100),
+        avg_r=("r_multiple", "mean"),
+    ).round(2)
+    lines.append(by_dir.to_string())
+    lines.append("")
+
+    return "\n".join(lines)
+
+
+def _grade_block(label: str, sub: pd.DataFrame) -> str:
+    n        = len(sub)
+    wins     = (sub["r_multiple"] > 0).sum()
+    losses   = (sub["r_multiple"] <= 0).sum()
+    win_rate = wins / n * 100 if n else 0.0
+    avg_r    = sub["r_multiple"].mean()
+    sum_r    = sub["r_multiple"].sum()
+
+    gross_win  = sub.loc[sub["r_multiple"] > 0, "r_multiple"].sum()
+    gross_loss = -sub.loc[sub["r_multiple"] <= 0, "r_multiple"].sum()
+    profit_factor = (gross_win / gross_loss) if gross_loss > 0 else float("inf")
+
+    # Equity-curve drawdown (in R)
+    cum = sub["r_multiple"].cumsum()
+    running_peak = cum.cummax()
+    drawdown = (cum - running_peak).min()
+
+    return (
+        f"{label:>4}: {n:>4} trades | "
+        f"win {win_rate:5.1f}% ({wins}W/{losses}L) | "
+        f"avg R {avg_r:+.3f} | total R {sum_r:+.2f} | "
+        f"PF {profit_factor:.2f} | maxDD {drawdown:.2f}R"
+    )
+
+
+def plot_results(df: pd.DataFrame, out_dir: Path) -> None:
+    """Generate equity curve and per-grade bar chart. Requires matplotlib."""
+    try:
+        import matplotlib
+        matplotlib.use("Agg")  # headless
+        import matplotlib.pyplot as plt
+    except ImportError:
+        print("⚠️  matplotlib not installed — skipping plots. "
+              "pip install matplotlib")
+        return
+
+    if df.empty:
+        return
+
+    df = df.sort_values("entry_time").reset_index(drop=True)
+    df["cum_r"] = df["r_multiple"].cumsum()
+
+    # Equity curve
+    fig, ax = plt.subplots(figsize=(11, 5))
+    ax.plot(df["entry_time"], df["cum_r"], linewidth=1.5)
+    ax.axhline(0, color="gray", linewidth=0.5)
+    ax.fill_between(df["entry_time"], df["cum_r"], 0,
+                    where=df["cum_r"] >= 0, alpha=0.2, color="green")
+    ax.fill_between(df["entry_time"], df["cum_r"], 0,
+                    where=df["cum_r"] < 0, alpha=0.2, color="red")
+    ax.set_title(f"Cumulative R  ({len(df)} trades, total {df['cum_r'].iloc[-1]:+.2f}R)")
+    ax.set_xlabel("Trade entry time")
+    ax.set_ylabel("Cumulative R (1R = risked amount per trade)")
+    ax.grid(True, alpha=0.3)
+    fig.tight_layout()
+    fig.savefig(out_dir / "equity.png", dpi=120)
+    plt.close(fig)
+
+    # Win rate by grade
+    grades = ["A+", "A", "B", "C", "D"]
+    grade_stats = []
+    for g in grades:
+        sub = df[df["grade"] == g]
+        if sub.empty:
+            continue
+        wr = (sub["r_multiple"] > 0).mean() * 100
+        avg_r = sub["r_multiple"].mean()
+        grade_stats.append((g, len(sub), wr, avg_r))
+
+    if grade_stats:
+        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(11, 4.5))
+        labels = [f"{g}\n(n={n})" for g, n, _, _ in grade_stats]
+        wrs    = [w for _, _, w, _ in grade_stats]
+        avgs   = [r for _, _, _, r in grade_stats]
+
+        ax1.bar(labels, wrs, color=["#16a34a", "#22c55e", "#84cc16", "#eab308", "#f97316"][:len(labels)])
+        ax1.axhline(50, linestyle="--", color="gray", linewidth=0.8)
+        ax1.set_ylabel("Win rate (%)")
+        ax1.set_title("Win rate by grade")
+        for i, w in enumerate(wrs):
+            ax1.text(i, w + 1, f"{w:.0f}%", ha="center", fontsize=9)
+
+        ax2.bar(labels, avgs, color=["#16a34a", "#22c55e", "#84cc16", "#eab308", "#f97316"][:len(labels)])
+        ax2.axhline(0, color="black", linewidth=0.5)
+        ax2.set_ylabel("Average R per trade")
+        ax2.set_title("Average R by grade")
+        for i, a in enumerate(avgs):
+            ax2.text(i, a + (0.02 if a >= 0 else -0.05), f"{a:+.2f}",
+                     ha="center", fontsize=9)
+
+        fig.tight_layout()
+        fig.savefig(out_dir / "by_grade.png", dpi=120)
+        plt.close(fig)
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  CLI
+# ══════════════════════════════════════════════════════════════════════
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Walk-forward backtest of the Forex signal engine.",
+    )
+    parser.add_argument("--primary", default="M15",
+                        help="Primary timeframe (default M15)")
+    parser.add_argument("--pairs", nargs="+", default=None,
+                        help="Pairs to test (default: all majors)")
+    parser.add_argument("--grades", nargs="+", default=["A+", "A"],
+                        help='Grades to take (default "A+ A")')
+    parser.add_argument("--history-dir", default=str(HISTORY_DIR))
+    parser.add_argument("--out-dir", default=str(RESULTS_DIR))
+    parser.add_argument("--max-hold-bars", type=int, default=100,
+                        help="Max bars to hold a trade before timeout (default 100)")
+    parser.add_argument("--spread-pips", type=float, default=1.0,
+                        help="Round-trip spread cost in pips (default 1.0)")
+    parser.add_argument("--warmup-bars", type=int, default=250,
+                        help="Bars to skip at start so indicators stabilize (default 250)")
+    parser.add_argument("--plot", action="store_true",
+                        help="Generate equity / by-grade PNGs (needs matplotlib)")
+    parser.add_argument("-v", "--verbose", action="store_true")
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        level=logging.INFO if args.verbose else logging.WARNING,
+        format="%(asctime)s [%(levelname)s] %(message)s",
+        datefmt="%H:%M:%S",
+    )
+
+    primary_tf = _TF_MAP[args.primary.upper()]
+    confirmations = _TF_CONFIRMATIONS.get(primary_tf, [])
+    needed_tfs = [primary_tf] + confirmations
+
+    pairs = args.pairs if args.pairs else MAJOR_PAIRS
+    pairs = [p if p.endswith("=X") else f"{p}=X" for p in pairs]
+
+    history_dir = Path(args.history_dir)
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    print(f"🔬 Backtest: primary={primary_tf.value}  "
+          f"confirmations={[t.value for t in confirmations]}")
+    print(f"   Pairs:  {[p.replace('=X','') for p in pairs]}")
+    print(f"   Grades: {args.grades}")
+    print(f"   History dir: {history_dir}/")
+    print(f"   Results dir: {out_dir}/")
+    print()
+
+    history = load_history(pairs, needed_tfs, history_dir=history_dir)
+    if not history:
+        print(f"❌ No history loaded from {history_dir}/. "
+              f"Run: python seed_history.py --tfs {' '.join(t.value for t in needed_tfs)}")
+        return 1
+
+    bt = Backtester(
+        primary_tf     = primary_tf,
+        pairs          = pairs,
+        history        = history,
+        target_grades  = args.grades,
+        max_hold_bars  = args.max_hold_bars,
+        spread_pips    = args.spread_pips,
+        warmup_bars    = args.warmup_bars,
+    )
+
+    trades = bt.run()
+
+    if not trades:
+        print("\n❌ No trades emitted. Try lowering --grades (e.g., 'B') or "
+              "check that primary+confirmation TFs are in history/.")
+        return 0
+
+    df = trades_to_dataframe(trades)
+    df.to_csv(out_dir / "trades.csv", index=False)
+
+    summary = summarize(df)
+    (out_dir / "summary.txt").write_text(summary)
+
+    print("\n" + summary)
+    print(f"\n📁 trades.csv  → {out_dir/'trades.csv'}")
+    print(f"📁 summary.txt → {out_dir/'summary.txt'}")
+
+    if args.plot:
+        plot_results(df, out_dir)
+        print(f"📈 equity.png  → {out_dir/'equity.png'}")
+        print(f"📈 by_grade.png → {out_dir/'by_grade.png'}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

data_provider.py

@@ -456,6 +456,125 @@ class ForexDataProvider:
                 f"Failed to fetch {td_symbol}: {e}"
             ) from e
 
+    def fetch_historical(
+        self,
+        symbol: str,
+        timeframe: Timeframe,
+        start: datetime,
+        end: datetime,
+    ) -> pd.DataFrame:
+        """
+        Paginated historical OHLCV fetch for backtesting seed.
+
+        Walks backward from `end` to `start` in 5000-bar chunks (the free-tier
+        per-call max), 1 credit per chunk. Honours the shared rate limiter so
+        it cooperates with any concurrent live scans.
+
+        Returns a single concatenated DataFrame indexed by UTC timestamp.
+        """
+        if not self._api_key:
+            raise DataUnavailableError("No Twelve Data API key set.")
+
+        if start.tzinfo is None:
+            start = start.replace(tzinfo=timezone.utc)
+        if end.tzinfo is None:
+            end = end.replace(tzinfo=timezone.utc)
+
+        bars_per_day = self.BARS_PER_DAY[timeframe]
+        # Stay well clear of the 5000-bar cap — leave a small margin for
+        # weekend gaps that don't decrement bars on calendar-day arithmetic.
+        chunk_days = max(1, int(4500 / bars_per_day))
+
+        td_symbol = self._to_td_symbol(symbol)
+        interval = self.TD_INTERVALS[timeframe]
+
+        chunks: List[pd.DataFrame] = []
+        cursor_end = end
+
+        while cursor_end > start:
+            cursor_start = max(start, cursor_end - timedelta(days=chunk_days))
+
+            self._throttle(credits=1)
+
+            resp = self._session.get(
+                f"{self.TD_BASE_URL}/time_series",
+                params={
+                    "symbol":     td_symbol,
+                    "interval":   interval,
+                    "start_date": cursor_start.strftime("%Y-%m-%d %H:%M:%S"),
+                    "end_date":   cursor_end.strftime("%Y-%m-%d %H:%M:%S"),
+                    "outputsize": 5000,
+                    "timezone":   "UTC",
+                    "format":     "json",
+                },
+                timeout=30,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+
+            if isinstance(data, dict) and data.get("status") == "error":
+                msg = data.get("message", "unknown")
+                # If we already have chunks, an error on a deeper-history call
+                # almost always means "you've walked past the available start."
+                # Treat as graceful end-of-history instead of fatal.
+                if chunks:
+                    logger.info(
+                        f"  {td_symbol} {interval}: end of available history "
+                        f"reached ({msg})"
+                    )
+                    break
+                raise DataUnavailableError(
+                    f"Twelve Data error for {td_symbol}: {msg}"
+                )
+
+            values = data.get("values") or []
+            if not values:
+                logger.info(
+                    f"  {td_symbol} {interval}: empty range "
+                    f"{cursor_start.date()}..{cursor_end.date()}, stopping"
+                )
+                break
+
+            records = [
+                {
+                    "Datetime": pd.Timestamp(v["datetime"], tz="UTC"),
+                    "Open":   float(v["open"]),
+                    "High":   float(v["high"]),
+                    "Low":    float(v["low"]),
+                    "Close":  float(v["close"]),
+                    "Volume": float(v.get("volume") or 0),
+                }
+                for v in values if v.get("datetime")
+            ]
+            df_chunk = pd.DataFrame(records).set_index("Datetime").sort_index()
+            df_chunk.index.name = "Datetime"
+            chunks.append(df_chunk)
+
+            earliest = df_chunk.index.min()
+            logger.info(
+                f"  {td_symbol} {interval}: chunk "
+                f"{earliest.date()} → {df_chunk.index.max().date()} "
+                f"({len(df_chunk):,} bars)"
+            )
+
+            # Stop when:
+            #   1. The current chunk's start window already hit the requested
+            #      `start` boundary — nothing earlier was asked for.
+            #   2. The chunk is unusually thin (< 100 bars), suggesting we've
+            #      reached the edge of the symbol's available history.
+            if cursor_start <= start or len(df_chunk) < 100:
+                break
+
+            cursor_end = earliest - timedelta(seconds=1)
+
+        if not chunks:
+            raise DataUnavailableError(f"No historical data returned for {td_symbol}")
+
+        df = pd.concat(chunks).sort_index()
+        df = df[~df.index.duplicated(keep="last")]
+        df = df[(df.index >= start) & (df.index <= end)]
+        return df
+
     def _validate_and_clean(
         self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
     ) -> pd.DataFrame:

geometry_grid.py

@@ -0,0 +1,361 @@
+#!/usr/bin/env python3
+"""
+Phase 2: re-simulate the existing trade pool under different SL/TP geometries
+without re-running the engine.
+
+The engine already gave us, per trade: entry_time, entry_price, direction,
+and the original SL distance (= 1.5 × ATR by default). That distance lets
+us recover ATR and re-derive new SL/TP for any (sl_mult, rr) combo. We
+then walk forward through the saved OHLCV bars, applying first-touch
+simulation with worst-case (SL first if both touch same bar).
+
+This is faster than re-running the engine because the signal output (when
+to trade, in which direction, at what price) is independent of SL/TP
+geometry — only the exits change.
+
+Discipline: we report TRAIN and TEST stats for every cell, time-split at
+70/30 by entry_time. A cell wins only if it's positive on the held-out
+TEST slice.
+
+Usage:
+    python geometry_grid.py
+    python geometry_grid.py --apply-phase1-filters
+    python geometry_grid.py --pairs USDJPY  # single-pair grid
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+from typing import Dict, Tuple
+
+import numpy as np
+import pandas as pd
+
+# Reuse history loader so we don't re-implement
+from backtest import load_history
+from config import Timeframe
+
+# trades.csv stores TF as Twelve-Data interval string ("15min", "1h", "1d", …)
+_TFSTR_TO_ENUM = {
+    "1min":  Timeframe.M1,  "5min":  Timeframe.M5,
+    "15min": Timeframe.M15, "30min": Timeframe.M30,
+    "1h":    Timeframe.H1,  "4h":    Timeframe.H4,
+    "1d":    Timeframe.D1,  "1wk":   Timeframe.W1,
+}
+
+# Filters carried over from Phase 1
+PHASE1_BAD_HOURS = {10, 17, 19, 21}
+PHASE1_BAD_PAIRS = {"NZD/USD", "AUD/USD", "GBP/USD"}
+PHASE1_BAD_DOWS  = {5, 6}  # Sat / Sun
+
+
+def simulate_one(
+    direction: str,
+    entry: float,
+    new_sl: float,
+    new_tp: float,
+    forward_bars: pd.DataFrame,
+    max_bars: int,
+    spread_pips: float,
+    pip_value: float,
+) -> Tuple[float, str]:
+    """First-touch simulation. Returns (r_multiple, outcome)."""
+    sl_distance = abs(entry - new_sl)
+    if sl_distance == 0:
+        return 0.0, "INVALID"
+    risk_pips = sl_distance / pip_value
+    spread_r = spread_pips / risk_pips if risk_pips > 0 else 0.0
+    rr_target = abs(new_tp - entry) / sl_distance
+
+    bars_examined = 0
+    for ts, bar in forward_bars.iterrows():
+        bars_examined += 1
+        if bars_examined > max_bars:
+            break
+        bar_high = bar["High"]
+        bar_low  = bar["Low"]
+        if direction == "BUY":
+            sl_hit = bar_low <= new_sl
+            tp_hit = bar_high >= new_tp
+        else:
+            sl_hit = bar_high >= new_sl
+            tp_hit = bar_low <= new_tp
+        if sl_hit:
+            return -1.0 - spread_r, "LOSS_SL"
+        if tp_hit:
+            return rr_target - spread_r, "WIN_TP1"
+
+    if bars_examined == 0:
+        return 0.0, "INVALID"
+    last = forward_bars.iloc[bars_examined - 1]
+    pnl = (last["Close"] - entry) if direction == "BUY" else (entry - last["Close"])
+    return (pnl / sl_distance) - spread_r, "TIMEOUT"
+
+
+def pip_value_for(symbol: str) -> float:
+    """Return pip value: 0.01 for JPY pairs, 0.0001 elsewhere."""
+    return 0.01 if "JPY" in symbol else 0.0001
+
+
+def grid_run(
+    df: pd.DataFrame,
+    history: Dict[Tuple[str, str], pd.DataFrame],
+    sl_mults: list,
+    rr_ratios: list,
+    max_hold: int = 100,
+    spread_pips: float = 1.0,
+) -> pd.DataFrame:
+    """For each (sl_mult, rr) cell, re-simulate every trade and report stats."""
+    # Pre-index history by (symbol, tf_str) for fast lookup
+    rows = []
+    n_total = len(df) * len(sl_mults) * len(rr_ratios)
+    print(f"   evaluating {n_total:,} simulations "
+          f"({len(df):,} trades × {len(sl_mults)} SL × {len(rr_ratios)} RR)…")
+
+    for sl_mult in sl_mults:
+        for rr in rr_ratios:
+            r_multiples = []
+            outcomes = []
+            entry_times = []
+            pairs = []
+            tfs = []
+            for trade in df.itertuples(index=False):
+                # Recover ATR from original SL distance (engine used 1.5×ATR)
+                atr = abs(trade.entry - trade.sl) / 1.5
+                new_sl_dist = sl_mult * atr
+                if trade.direction == "BUY":
+                    new_sl = trade.entry - new_sl_dist
+                    new_tp = trade.entry + new_sl_dist * rr
+                else:
+                    new_sl = trade.entry + new_sl_dist
+                    new_tp = trade.entry - new_sl_dist * rr
+
+                hist_df = history.get((trade.symbol, trade.tf))
+                if hist_df is None:
+                    continue
+                ts = pd.Timestamp(trade.entry_time)
+                # Forward bars (strictly after entry)
+                forward = hist_df[hist_df.index > ts].head(max_hold)
+                if forward.empty:
+                    continue
+
+                pip = pip_value_for(trade.symbol)
+                r, outcome = simulate_one(
+                    trade.direction, trade.entry,
+                    new_sl, new_tp, forward,
+                    max_bars=max_hold, spread_pips=spread_pips,
+                    pip_value=pip,
+                )
+                r_multiples.append(r)
+                outcomes.append(outcome)
+                entry_times.append(ts)
+                pairs.append(trade.pair)
+                tfs.append(trade.tf)
+
+            arr = np.array(r_multiples) if r_multiples else np.array([0.0])
+            wins = (arr > 0).sum()
+            losses = (arr <= 0).sum()
+            gross_win = arr[arr > 0].sum()
+            gross_loss = -arr[arr <= 0].sum()
+            pf = gross_win / gross_loss if gross_loss > 0 else float("inf")
+            cum = np.cumsum(arr)
+            running_peak = np.maximum.accumulate(cum)
+            max_dd = (cum - running_peak).min()
+
+            rows.append({
+                "sl_mult":  sl_mult,
+                "rr":       rr,
+                "n":        len(arr),
+                "wins":     int(wins),
+                "wr":       wins / len(arr) * 100 if len(arr) else 0,
+                "avg_r":    arr.mean() if len(arr) else 0,
+                "total_r":  arr.sum(),
+                "pf":       pf,
+                "max_dd":   max_dd,
+                # Carry the per-trade results so we can split train/test downstream
+                "_arr":         arr,
+                "_entry_times": entry_times,
+                "_pairs":       pairs,
+                "_outcomes":    outcomes,
+            })
+    return pd.DataFrame(rows)
+
+
+def split_summary(cell: pd.Series, train_frac: float = 0.7) -> dict:
+    """Split a cell's per-trade results into train/test by entry_time."""
+    arr = cell["_arr"]
+    times = cell["_entry_times"]
+    if len(arr) == 0:
+        return {"train_total": 0, "train_wr": 0, "test_total": 0, "test_wr": 0}
+    order = np.argsort(times)
+    cutoff = int(len(arr) * train_frac)
+    train_idx = order[:cutoff]
+    test_idx  = order[cutoff:]
+    train_arr = arr[train_idx]
+    test_arr  = arr[test_idx]
+    return {
+        "train_n":     len(train_arr),
+        "train_wr":    (train_arr > 0).mean() * 100 if len(train_arr) else 0,
+        "train_avg":   train_arr.mean() if len(train_arr) else 0,
+        "train_total": train_arr.sum(),
+        "test_n":      len(test_arr),
+        "test_wr":     (test_arr > 0).mean() * 100 if len(test_arr) else 0,
+        "test_avg":    test_arr.mean() if len(test_arr) else 0,
+        "test_total":  test_arr.sum(),
+    }
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--trades", default="backtest_results_all/trades.csv")
+    ap.add_argument("--out", default="backtest_results_all/geometry_grid.txt")
+    ap.add_argument("--max-hold", type=int, default=100)
+    ap.add_argument("--spread-pips", type=float, default=1.0)
+    ap.add_argument("--apply-phase1-filters", action="store_true",
+                    help="Drop weekend trades, worst 4 hours, worst 3 pairs before grid")
+    ap.add_argument("--sl-mults", nargs="+", type=float,
+                    default=[1.0, 1.5, 2.0, 2.5, 3.0])
+    ap.add_argument("--rr-ratios", nargs="+", type=float,
+                    default=[1.0, 1.5, 2.0, 2.5])
+    args = ap.parse_args()
+
+    p = Path(args.trades)
+    if not p.exists():
+        print(f"❌ {p} not found")
+        return 1
+
+    df = pd.read_csv(p)
+    df["entry_time"] = pd.to_datetime(df["entry_time"])
+    df["hour"] = df["entry_time"].dt.hour
+    df["dow"] = df["entry_time"].dt.dayofweek
+
+    print(f"📊 Loaded {len(df):,} trades from {p}")
+
+    if args.apply_phase1_filters:
+        before = len(df)
+        df = df[~df["hour"].isin(PHASE1_BAD_HOURS)]
+        df = df[~df["pair"].isin(PHASE1_BAD_PAIRS)]
+        df = df[~df["dow"].isin(PHASE1_BAD_DOWS)]
+        df = df.reset_index(drop=True)
+        print(f"   Phase 1 filters applied: {before:,} → {len(df):,} trades "
+              f"({len(df)/before*100:.0f}% retained)")
+
+    # Load history for forward simulation
+    print(f"📁 Loading history…")
+    pairs = sorted(df["symbol"].unique().tolist())
+    tfs   = sorted({_TFSTR_TO_ENUM[t] for t in df["tf"].unique() if t in _TFSTR_TO_ENUM},
+                   key=lambda x: x.value)
+    history_raw = load_history(pairs, tfs)
+    # Rekey by (symbol, tf_string) to match trades.csv
+    history = {}
+    for (sym, tf_enum), hdf in history_raw.items():
+        history[(sym, tf_enum.value)] = hdf
+
+    print(f"\n🧮 Grid search:")
+    print(f"   SL multipliers (× ATR): {args.sl_mults}")
+    print(f"   RR ratios:               {args.rr_ratios}\n")
+
+    grid = grid_run(
+        df, history, args.sl_mults, args.rr_ratios,
+        max_hold=args.max_hold, spread_pips=args.spread_pips,
+    )
+
+    # ── Heatmap (full-data total R) ─────────────────────────────────
+    pivot_total = grid.pivot(index="sl_mult", columns="rr", values="total_r").round(1)
+    pivot_wr    = grid.pivot(index="sl_mult", columns="rr", values="wr").round(1)
+    pivot_avg   = grid.pivot(index="sl_mult", columns="rr", values="avg_r").round(3)
+    pivot_pf    = grid.pivot(index="sl_mult", columns="rr", values="pf").round(2)
+
+    out_lines = []
+    def emit(line: str = ""):
+        print(line)
+        out_lines.append(line)
+
+    emit("=" * 80)
+    emit(f"GEOMETRY GRID — {len(df):,} trades")
+    emit(f"  filters applied: {args.apply_phase1_filters}")
+    emit(f"  spread cost:     {args.spread_pips} pip")
+    emit(f"  max hold:        {args.max_hold} bars")
+    emit("=" * 80)
+    emit()
+    emit("─── Total R (rows = SL multiplier × ATR; cols = RR ratio) ───")
+    emit(pivot_total.to_string())
+    emit()
+    emit("─── Win rate % ───")
+    emit(pivot_wr.to_string())
+    emit()
+    emit("─── Avg R per trade ───")
+    emit(pivot_avg.to_string())
+    emit()
+    emit("─── Profit factor ───")
+    emit(pivot_pf.to_string())
+    emit()
+
+    # ── Train/Test discipline on top cells ──────────────────────────
+    grid_sorted = grid.sort_values("total_r", ascending=False)
+    emit("=" * 80)
+    emit("TOP 5 CELLS (by full-period total R) — with TRAIN / TEST split discipline")
+    emit("=" * 80)
+    for _, cell in grid_sorted.head(5).iterrows():
+        s = split_summary(cell)
+        emit(
+            f"  SL={cell['sl_mult']:.1f}×ATR  RR=1:{cell['rr']:.1f}  "
+            f"|  full: {cell['n']} trades  WR {cell['wr']:.1f}%  "
+            f"avgR {cell['avg_r']:+.3f}  totalR {cell['total_r']:+.1f}  "
+            f"PF {cell['pf']:.2f}"
+        )
+        emit(
+            f"      train ({s['train_n']}):  "
+            f"WR {s['train_wr']:.1f}%  avgR {s['train_avg']:+.3f}  totalR {s['train_total']:+.1f}"
+        )
+        emit(
+            f"      test  ({s['test_n']}):   "
+            f"WR {s['test_wr']:.1f}%  avgR {s['test_avg']:+.3f}  totalR {s['test_total']:+.1f}"
+        )
+        emit()
+
+    # ── Bottom 3 for contrast ───────────────────────────────────────
+    emit("─── BOTTOM 3 cells (worst total R) ───")
+    for _, cell in grid_sorted.tail(3).iterrows():
+        emit(
+            f"  SL={cell['sl_mult']:.1f}×ATR  RR=1:{cell['rr']:.1f}  "
+            f"totalR {cell['total_r']:+.1f}  WR {cell['wr']:.1f}%"
+        )
+    emit()
+
+    # ── Verdict ─────────────────────────────────────────────────────
+    emit("=" * 80)
+    emit("VERDICT")
+    emit("=" * 80)
+    best = grid_sorted.iloc[0]
+    s = split_summary(best)
+    emit(
+        f"  Best cell: SL={best['sl_mult']:.1f}×ATR, RR=1:{best['rr']:.1f}  "
+        f"(full totalR {best['total_r']:+.1f})"
+    )
+    emit(
+        f"  Test slice: {s['test_n']} trades, WR {s['test_wr']:.1f}%, "
+        f"totalR {s['test_total']:+.1f}, avgR {s['test_avg']:+.3f}"
+    )
+    if s["test_total"] > 0:
+        emit("  ✅ Best geometry is profitable on held-out test set.")
+        emit("     → Phase 3A: hardcode in _generate_trade_setup, re-run end-to-end.")
+    elif s["test_avg"] > -0.05:
+        emit("  ⚠️  Best geometry is near-break-even on test set.")
+        emit("     → Phase 3A worth attempting; combine with Phase 1 filters for margin.")
+    else:
+        emit("  ❌ No geometry is profitable on held-out test set.")
+        emit("     → Phase 3B: M15 confluence is structurally not edge-bearing.")
+        emit("        Consider H4 primary, mean-reversion-only, or indicator-level meta-model.")
+
+    Path(args.out).parent.mkdir(parents=True, exist_ok=True)
+    Path(args.out).write_text("\n".join(out_lines))
+    emit()
+    emit(f"📁 Saved: {args.out}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

meta_model.py

@@ -0,0 +1,368 @@
+#!/usr/bin/env python3
+"""
+Meta-model: learn an empirical "will this trade win?" probability from
+backtest output, replacing the engine's heuristic A+/A/B grades with a
+calibrated win-probability score.
+
+Inputs:
+    backtest_results/trades.csv  (or --trades <path>)
+
+Outputs (under backtest_results/meta/):
+    feature_importances.txt
+    classification_report.txt
+    calibration.png              — predicted prob vs actual win rate
+    decile_returns.png           — avg R per predicted-prob decile
+    equity_compare.png           — engine grades vs meta-top-N
+    summary.txt                  — headline numbers + recommended threshold
+
+Train/test split is **time-based** (first 70% train, last 30% test) — random
+splits leak future regime information into training and overstate test
+performance on financial time series.
+
+Usage:
+    python meta_model.py
+    python meta_model.py --trades backtest_results_h1/trades.csv --out meta_h1
+    python meta_model.py --top-decile 0.2   # take top 20% predictions
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from pathlib import Path
+from typing import Tuple
+
+import numpy as np
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+
+def engineer_features(df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Series]:
+    """
+    Build features available at trade entry time. Strict no-leakage:
+    nothing from exit_time, exit_price, outcome, bars_held, r_multiple.
+    """
+    df = df.copy()
+    df["entry_time"] = pd.to_datetime(df["entry_time"])
+
+    # Time-of-day / day-of-week
+    df["hour"] = df["entry_time"].dt.hour
+    df["dow"]  = df["entry_time"].dt.dayofweek
+    df["is_buy"] = (df["direction"] == "BUY").astype(int)
+
+    # MTF flag
+    df["mtf"] = df["mtf_aligned"].astype(int)
+
+    # Score / confidence — keep raw and signed/abs versions; model picks
+    df["abs_score"] = df["score"].abs()
+    df["score_sign"] = np.sign(df["score"])
+
+    # Risk geometry
+    df["rr"] = df["reward_pips"] / df["risk_pips"].replace(0, np.nan)
+    df["rr"] = df["rr"].fillna(2.0)
+
+    # Regime one-hot
+    regime_dummies = pd.get_dummies(df["regime"], prefix="regime").astype(int)
+    df = pd.concat([df, regime_dummies], axis=1)
+
+    # Pair one-hot (model can learn pair-specific behavior)
+    pair_dummies = pd.get_dummies(df["pair"], prefix="pair").astype(int)
+    df = pd.concat([df, pair_dummies], axis=1)
+
+    # Grade one-hot — included so the meta-model can re-weight rather
+    # than blindly disregard the engine's view
+    grade_dummies = pd.get_dummies(df["grade"], prefix="grade").astype(int)
+    df = pd.concat([df, grade_dummies], axis=1)
+
+    feature_cols = (
+        ["score", "abs_score", "score_sign", "confidence", "rr",
+         "is_buy", "mtf", "hour", "dow", "risk_pips", "reward_pips"]
+        + list(regime_dummies.columns)
+        + list(pair_dummies.columns)
+        + list(grade_dummies.columns)
+    )
+
+    X = df[feature_cols].astype(float)
+    y = (df["r_multiple"] > 0).astype(int)
+    return X, y
+
+
+def time_based_split(
+    X: pd.DataFrame, y: pd.Series, df: pd.DataFrame, train_frac: float = 0.7,
+):
+    """First 70% by entry_time = train, last 30% = test. Standard for time series."""
+    order = df["entry_time"].argsort().values
+    cutoff = int(len(order) * train_frac)
+    train_idx = order[:cutoff]
+    test_idx  = order[cutoff:]
+    return (
+        X.iloc[train_idx], X.iloc[test_idx],
+        y.iloc[train_idx], y.iloc[test_idx],
+        df.iloc[train_idx], df.iloc[test_idx],
+    )
+
+
+def evaluate_decile_strategy(
+    df_test: pd.DataFrame, p_win: np.ndarray, top_frac: float = 0.2,
+) -> dict:
+    """
+    Take only the top X% by predicted win probability. Compare to taking
+    every trade in the test set (the 'engine baseline').
+    """
+    n = len(df_test)
+    cutoff_n = max(1, int(n * top_frac))
+
+    df_eval = df_test.copy()
+    df_eval["p_win"] = p_win
+    df_eval = df_eval.sort_values("p_win", ascending=False)
+
+    top = df_eval.head(cutoff_n)
+    base = df_eval
+
+    return {
+        "top_frac":    top_frac,
+        "top_n":       len(top),
+        "top_wr":      (top["r_multiple"] > 0).mean() * 100,
+        "top_avg_r":   top["r_multiple"].mean(),
+        "top_total_r": top["r_multiple"].sum(),
+        "base_n":      len(base),
+        "base_wr":     (base["r_multiple"] > 0).mean() * 100,
+        "base_avg_r":  base["r_multiple"].mean(),
+        "base_total_r": base["r_multiple"].sum(),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Meta-model on backtest trades.")
+    parser.add_argument("--trades", default="backtest_results/trades.csv",
+                        help="Path to trades.csv (default backtest_results/trades.csv)")
+    parser.add_argument("--out", default="backtest_results/meta",
+                        help="Output directory")
+    parser.add_argument("--train-frac", type=float, default=0.7)
+    parser.add_argument("--top-decile", type=float, default=0.2,
+                        help="Top fraction of predictions to 'trade' (default 0.2)")
+    parser.add_argument("--min-trades", type=int, default=300,
+                        help="Refuse to train below this many trades")
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+
+    try:
+        from sklearn.linear_model import LogisticRegression
+        from sklearn.ensemble import GradientBoostingClassifier
+        from sklearn.preprocessing import StandardScaler
+        from sklearn.pipeline import Pipeline
+        from sklearn.metrics import (
+            classification_report, roc_auc_score, brier_score_loss,
+        )
+    except ImportError:
+        print("❌ scikit-learn not installed.  pip install scikit-learn")
+        return 1
+
+    trades_path = Path(args.trades)
+    if not trades_path.exists():
+        print(f"❌ {trades_path} not found. Run backtest.py first.")
+        return 1
+
+    out_dir = Path(args.out)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    df = pd.read_csv(trades_path)
+    df["entry_time"] = pd.to_datetime(df["entry_time"])
+    if len(df) < args.min_trades:
+        print(f"❌ Only {len(df)} trades — need ≥{args.min_trades} for training.")
+        print("   Seed more pairs / longer history, or lower --min-trades.")
+        return 1
+
+    print(f"📊 Loaded {len(df):,} trades from {trades_path}")
+    print(f"   Pairs: {sorted(df['pair'].unique())}")
+    print(f"   Date range: {df['entry_time'].min()} → {df['entry_time'].max()}")
+    print(f"   Baseline win rate: {(df['r_multiple'] > 0).mean()*100:.1f}%")
+    print(f"   Baseline avg R:   {df['r_multiple'].mean():+.3f}")
+    print()
+
+    X, y = engineer_features(df)
+    print(f"🔧 Features: {list(X.columns)}")
+    print(f"   Shape: {X.shape}, target balance: {y.mean()*100:.1f}% wins\n")
+
+    X_tr, X_te, y_tr, y_te, df_tr, df_te = time_based_split(X, y, df, args.train_frac)
+    print(f"📚 Train: {len(X_tr):,} trades  ({df_tr['entry_time'].min().date()} → {df_tr['entry_time'].max().date()})")
+    print(f"   Test:  {len(X_te):,} trades  ({df_te['entry_time'].min().date()} → {df_te['entry_time'].max().date()})\n")
+
+    # ── Model 1: Logistic regression (interpretable baseline) ─────────
+    lr = Pipeline([
+        ("scale", StandardScaler()),
+        ("lr", LogisticRegression(max_iter=2000, C=1.0)),
+    ])
+    lr.fit(X_tr, y_tr)
+    p_lr = lr.predict_proba(X_te)[:, 1]
+
+    # ── Model 2: Gradient boosting (non-linear, captures interactions) ─
+    gb = GradientBoostingClassifier(
+        n_estimators=200, max_depth=3, learning_rate=0.05,
+        subsample=0.8, random_state=42,
+    )
+    gb.fit(X_tr, y_tr)
+    p_gb = gb.predict_proba(X_te)[:, 1]
+
+    # ── Metrics ───────────────────────────────────────────────────────
+    print("══ Logistic Regression ══")
+    print(f"  AUC:    {roc_auc_score(y_te, p_lr):.3f}")
+    print(f"  Brier:  {brier_score_loss(y_te, p_lr):.3f}  (lower=better, 0.25=random for 50% prior)")
+    print()
+    print("══ Gradient Boosting ══")
+    print(f"  AUC:    {roc_auc_score(y_te, p_gb):.3f}")
+    print(f"  Brier:  {brier_score_loss(y_te, p_gb):.3f}")
+    print()
+
+    # Pick the better-AUC model for downstream evaluation
+    p_best = p_gb if roc_auc_score(y_te, p_gb) >= roc_auc_score(y_te, p_lr) else p_lr
+    best_name = "GradientBoosting" if p_best is p_gb else "LogisticRegression"
+    print(f"➡️  Selecting {best_name} for top-decile evaluation\n")
+
+    # ── Decile-strategy comparison ─────────────────────────────────────
+    res = evaluate_decile_strategy(df_te, p_best, args.top_decile)
+    print(f"══ {int(args.top_decile*100)}% top-pick strategy vs trade-everything baseline (test set) ══")
+    print(f"  Trade everything:  {res['base_n']:>5} trades  WR {res['base_wr']:5.1f}%  avgR {res['base_avg_r']:+.3f}  totalR {res['base_total_r']:+8.2f}")
+    print(f"  Top {int(args.top_decile*100)}% by p_win:    {res['top_n']:>5} trades  WR {res['top_wr']:5.1f}%  avgR {res['top_avg_r']:+.3f}  totalR {res['top_total_r']:+8.2f}")
+    print()
+
+    # ── Per-decile breakdown ──────────────────────────────────────────
+    df_eval = df_te.copy()
+    df_eval["p_win"] = p_best
+    df_eval["decile"] = pd.qcut(df_eval["p_win"], 10, labels=False, duplicates="drop")
+    decile_summary = df_eval.groupby("decile").agg(
+        n=("r_multiple", "count"),
+        avg_p_win=("p_win", "mean"),
+        actual_wr=("r_multiple", lambda s: (s > 0).mean()),
+        avg_r=("r_multiple", "mean"),
+        total_r=("r_multiple", "sum"),
+    ).round(3)
+    decile_summary.index = decile_summary.index.astype(int) + 1  # 1..10
+    print("══ Per-decile breakdown (1=lowest p_win, 10=highest) ══")
+    print(decile_summary.to_string())
+    print()
+
+    # ── Feature importances ────────────────────────────────────────────
+    print("══ Top features by importance (from gradient boosting) ══")
+    importances = pd.Series(gb.feature_importances_, index=X.columns)
+    top_features = importances.sort_values(ascending=False).head(15)
+    for name, imp in top_features.items():
+        print(f"  {name:30s}  {imp:.4f}")
+    print()
+
+    # ── Save full reports ──────────────────────────────────────────────
+    (out_dir / "classification_report.txt").write_text(
+        f"Logistic Regression\n{'-'*40}\n"
+        f"AUC:   {roc_auc_score(y_te, p_lr):.3f}\n"
+        f"Brier: {brier_score_loss(y_te, p_lr):.3f}\n\n"
+        f"{classification_report(y_te, (p_lr > 0.5).astype(int))}\n"
+        f"\nGradient Boosting\n{'-'*40}\n"
+        f"AUC:   {roc_auc_score(y_te, p_gb):.3f}\n"
+        f"Brier: {brier_score_loss(y_te, p_gb):.3f}\n\n"
+        f"{classification_report(y_te, (p_gb > 0.5).astype(int))}\n"
+    )
+    (out_dir / "feature_importances.txt").write_text(
+        importances.sort_values(ascending=False).to_string()
+    )
+    (out_dir / "decile_breakdown.csv").write_text(decile_summary.to_csv())
+    (out_dir / "summary.txt").write_text(
+        f"Meta-model on {len(df):,} trades  "
+        f"({df['entry_time'].min()} → {df['entry_time'].max()})\n\n"
+        f"Baseline (trade everything):\n"
+        f"  win rate: {(df['r_multiple'] > 0).mean()*100:.1f}%\n"
+        f"  avg R:    {df['r_multiple'].mean():+.3f}\n"
+        f"  total R:  {df['r_multiple'].sum():+.2f}\n\n"
+        f"Test set ({len(X_te):,} trades, last 30% by date):\n"
+        f"  Trade everything:  WR {res['base_wr']:.1f}%  avgR {res['base_avg_r']:+.3f}  totalR {res['base_total_r']:+.2f}\n"
+        f"  Top {int(args.top_decile*100)}% by p_win:  WR {res['top_wr']:.1f}%  avgR {res['top_avg_r']:+.3f}  totalR {res['top_total_r']:+.2f}\n\n"
+        f"Best model: {best_name}\n"
+        f"  AUC:   {roc_auc_score(y_te, p_best):.3f}  (0.5 = random, >0.55 = real signal)\n"
+        f"  Brier: {brier_score_loss(y_te, p_best):.3f}\n"
+    )
+
+    # ── Plots ─────────────────────────────────────────────────────────
+    import matplotlib
+    matplotlib.use("Agg")
+    import matplotlib.pyplot as plt
+
+    # Calibration curve
+    bins = pd.cut(p_best, bins=np.linspace(0, 1, 11))
+    cal = pd.DataFrame({"p_win": p_best, "y": y_te.values, "bin": bins})
+    grouped = cal.groupby("bin", observed=True).agg(
+        mean_pred=("p_win", "mean"),
+        actual=("y", "mean"),
+        n=("y", "count"),
+    )
+    fig, ax = plt.subplots(figsize=(6, 5))
+    ax.plot([0, 1], [0, 1], "--", color="gray", label="Perfect calibration")
+    ax.plot(grouped["mean_pred"], grouped["actual"], "o-", linewidth=2, markersize=8)
+    for _, row in grouped.iterrows():
+        ax.annotate(f"n={int(row['n'])}",
+                    (row["mean_pred"], row["actual"]),
+                    textcoords="offset points", xytext=(8, -3), fontsize=8)
+    ax.set_xlabel("Mean predicted win probability")
+    ax.set_ylabel("Actual win rate")
+    ax.set_title(f"Calibration ({best_name}, test set)")
+    ax.set_xlim(0, 1); ax.set_ylim(0, 1)
+    ax.legend(); ax.grid(alpha=0.3)
+    fig.tight_layout()
+    fig.savefig(out_dir / "calibration.png", dpi=120)
+    plt.close(fig)
+
+    # Decile returns
+    fig, ax = plt.subplots(figsize=(8, 5))
+    colors = ["#dc2626" if r < 0 else "#16a34a" for r in decile_summary["avg_r"]]
+    ax.bar(decile_summary.index.astype(str), decile_summary["avg_r"], color=colors)
+    ax.axhline(0, color="black", linewidth=0.5)
+    ax.set_xlabel("Predicted-probability decile (1=lowest, 10=highest)")
+    ax.set_ylabel("Average R per trade")
+    ax.set_title(f"Trade outcome by predicted-probability decile ({best_name})")
+    for i, (idx, row) in enumerate(decile_summary.iterrows()):
+        ax.text(i, row["avg_r"] + (0.02 if row["avg_r"] >= 0 else -0.05),
+                f"n={int(row['n'])}\nWR={row['actual_wr']*100:.0f}%",
+                ha="center", fontsize=8)
+    ax.grid(alpha=0.3, axis="y")
+    fig.tight_layout()
+    fig.savefig(out_dir / "decile_returns.png", dpi=120)
+    plt.close(fig)
+
+    # Equity comparison: trade-everything vs top-decile (test set, time-ordered)
+    df_compare = df_te.copy().sort_values("entry_time").reset_index(drop=True)
+    df_compare["p_win"] = p_best
+    threshold = np.quantile(p_best, 1 - args.top_decile)
+    df_compare["take_top"] = (df_compare["p_win"] >= threshold).astype(int)
+    df_compare["base_cum"] = df_compare["r_multiple"].cumsum()
+    df_compare["top_cum"] = (df_compare["r_multiple"] * df_compare["take_top"]).cumsum()
+
+    fig, ax = plt.subplots(figsize=(11, 5))
+    ax.plot(df_compare["entry_time"], df_compare["base_cum"],
+            label="Trade everything (engine baseline)", linewidth=1.5)
+    ax.plot(df_compare["entry_time"], df_compare["top_cum"],
+            label=f"Top {int(args.top_decile*100)}% by p_win (meta-model)",
+            linewidth=1.8)
+    ax.axhline(0, color="gray", linewidth=0.5)
+    ax.set_xlabel("Trade entry time")
+    ax.set_ylabel("Cumulative R")
+    ax.set_title(f"Cumulative R: engine baseline vs meta-model top-{int(args.top_decile*100)}%  (test set)")
+    ax.legend(); ax.grid(alpha=0.3)
+    fig.tight_layout()
+    fig.savefig(out_dir / "equity_compare.png", dpi=120)
+    plt.close(fig)
+
+    print(f"📁 Reports → {out_dir}/")
+    print(f"   • classification_report.txt")
+    print(f"   • feature_importances.txt")
+    print(f"   • decile_breakdown.csv")
+    print(f"   • summary.txt")
+    print(f"📈 Plots:")
+    print(f"   • calibration.png")
+    print(f"   • decile_returns.png")
+    print(f"   • equity_compare.png")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

requirements.txt

@@ -11,3 +11,6 @@ python-dotenv>=1.0     # loads GEMINI_API_KEY and TELEGRAM_BOT_TOKEN from .env
 
 # Telegram bot
 python-telegram-bot[job-queue]>=20.0
+
+# Backtest analysis (only needed for `python backtest.py --plot`)
+matplotlib>=3.7

seed_history.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""
+One-shot historical OHLCV seeder for backtesting.
+
+Downloads bars from Twelve Data for the configured pairs/timeframes,
+paginated to stay within the 5000-bar per-call limit, and writes each
+(symbol, timeframe) to history/<PAIR>_<TF>.csv.
+
+Usage:
+    python seed_history.py --years 1
+    python seed_history.py --years 2 --pairs EURUSD GBPUSD USDJPY
+    python seed_history.py --tfs M15 H1 D1 --years 1 --force
+
+Cost (free tier, 800 credits/day):
+    1 year M15 + H1 + D1 × 7 majors  ≈ 56 credits
+    5 years same                     ≈ 231 credits
+"""
+
+import argparse
+import logging
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+try:
+    from dotenv import load_dotenv
+    load_dotenv()
+except ImportError:
+    pass
+
+from config import MAJOR_PAIRS, PlatformConfig, Timeframe
+from data_provider import DataUnavailableError, ForexDataProvider
+
+DEFAULT_TFS = ["M15", "H1", "D1"]
+HISTORY_DIR = Path("history")
+
+_TF_MAP = {
+    "M1":  Timeframe.M1,  "M5":  Timeframe.M5,
+    "M15": Timeframe.M15, "M30": Timeframe.M30,
+    "H1":  Timeframe.H1,  "H4":  Timeframe.H4,
+    "D1":  Timeframe.D1,  "W1":  Timeframe.W1,
+}
+
+
+def csv_path(out_dir: Path, symbol: str, tf: Timeframe) -> Path:
+    return out_dir / f"{symbol.replace('=X', '')}_{tf.value}.csv"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Seed historical OHLCV for backtesting (Twelve Data, free tier).",
+    )
+    parser.add_argument("--years", type=float, default=1.0,
+                        help="Years of history to download (default 1.0)")
+    parser.add_argument("--pairs", nargs="+", default=None,
+                        help="Pair symbols, e.g. EURUSD GBPUSD (default all 7 majors)")
+    parser.add_argument("--tfs", nargs="+", default=DEFAULT_TFS,
+                        help=f"Timeframes (default {' '.join(DEFAULT_TFS)})")
+    parser.add_argument("--out", default=str(HISTORY_DIR),
+                        help="Output directory (default history/)")
+    parser.add_argument("--force", action="store_true",
+                        help="Re-download even if a CSV already exists")
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(message)s",
+        datefmt="%H:%M:%S",
+    )
+
+    out_dir = Path(args.out)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    pairs = args.pairs if args.pairs else MAJOR_PAIRS
+    pairs = [p if p.endswith("=X") else f"{p}=X" for p in pairs]
+
+    try:
+        tfs = [_TF_MAP[t.upper()] for t in args.tfs]
+    except KeyError as e:
+        print(f"❌ Unknown timeframe: {e}. Valid: {list(_TF_MAP.keys())}")
+        return 2
+
+    end = datetime.now(timezone.utc)
+    start = end - timedelta(days=int(args.years * 365))
+
+    provider = ForexDataProvider(PlatformConfig())
+
+    total = len(pairs) * len(tfs)
+    print(f"📥 Seeding {total} pair × tf combinations from "
+          f"{start.date()} to {end.date()}  ({args.years} year(s))")
+    print(f"   Pairs: {[p.replace('=X', '') for p in pairs]}")
+    print(f"   TFs:   {[tf.value for tf in tfs]}")
+    print(f"   Output: {out_dir}/\n")
+
+    done = 0
+    failed: list = []
+    for symbol in pairs:
+        for tf in tfs:
+            done += 1
+            out_path = csv_path(out_dir, symbol, tf)
+
+            if out_path.exists() and not args.force:
+                print(f"[{done}/{total}] ⏭  {symbol} {tf.value} — exists, "
+                      f"skipping (--force to overwrite)")
+                continue
+
+            print(f"[{done}/{total}] 📥 {symbol} {tf.value}…", flush=True)
+            try:
+                df = provider.fetch_historical(symbol, tf, start, end)
+                df.to_csv(out_path)
+                print(f"   ✅ {len(df):,} bars  "
+                      f"({df.index.min().date()} → {df.index.max().date()})  "
+                      f"→ {out_path}")
+            except DataUnavailableError as e:
+                print(f"   ❌ {e}")
+                failed.append((symbol, tf.value, str(e)))
+            except Exception as e:
+                print(f"   ❌ unexpected: {e}")
+                failed.append((symbol, tf.value, str(e)))
+
+    print()
+    if failed:
+        print(f"⚠️  {len(failed)} fetch(es) failed:")
+        for sym, tfn, msg in failed:
+            print(f"     {sym} {tfn}: {msg}")
+        return 1
+
+    print("✅ Seeding complete. Run: python backtest.py --primary M15")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

telegram_bot.py

@@ -17,14 +17,19 @@ Commands:
 """
 
 import asyncio
+import csv
 import html
+import json
 import logging
 import os
 from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime
+from datetime import datetime, timezone
-from typing import Dict, List, Optional, Tuple
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
 from zoneinfo import ZoneInfo
 
+import pandas as pd
+
 # Load .env so TELEGRAM_BOT_TOKEN / GEMINI_API_KEY / TWELVE_DATA_API_KEY
 # don't need to be exported manually.
 try:
@@ -48,17 +53,32 @@ logger = logging.getLogger(__name__)
 
 # ── Config ──────────────────────────────────────────────────────────
 
-SCAN_TIMEFRAMES = ["M15", "H1"]  # two cards (M15 and H1) per /scan; cache shared
+# ─────────────────────────────────────────────────────────────────────
+#  Strategy config — calibrated by 2-year backtest (May 2024 → May 2026)
+# ─────────────────────────────────────────────────────────────────────
+# Backtest verdict: M15 has zero edge across 9,659 trades. Only USDCAD
+# on H4 at Grade B survived 2-year regime stability — train +20.8 R,
+# test +14.0 R, PF 1.52, max DD -9.6 R, positive in every 6-month
+# sub-period. Grades A+/A are anti-edge (highest grade ≠ best trades).
+# See backtest_results_h4_2y/ for the data.
 
-# Pairs analyzed by the /scan command. Limited to the most liquid majors
+SCAN_TIMEFRAMES = ["H4"]               # H4 only — M15 has no edge
-# to keep credits low and signal quality high.
+SCAN_PAIRS      = ["USDCAD=X"]          # USDCAD only — only pair with stable edge
-SCAN_PAIRS = ["EURUSD=X", "USDJPY=X", "GBPUSD=X", "AUDUSD=X"]
+ALERT_GRADES    = {"B"}                 # B only — A+/A are anti-edge
 
-# Auto-scan job: fires every AUTO_SCAN_INTERVAL seconds and pushes any A/A+
+# 30-min cadence: H4 candles update only every 4 hours, so polling more
-# signal to ALERT_CHAT_ID (env var). 10 min cadence ≈ 595 credits/day on
+# often wastes API credits. 48 scans/day × ~5 credits each ≈ 240/day,
-# cold-cache average — fits inside the 800/day Twelve Data free-tier budget.
+# well inside the 800/day free-tier budget.
-AUTO_SCAN_INTERVAL = 600
+AUTO_SCAN_INTERVAL = 1800
-ALERT_GRADES = {"A+", "A"}
+
+# Paper-trade mode: when True, alerts get a [PAPER] prefix and are
+# logged to paper_trades.csv so we can validate the strategy against
+# the backtest distribution before risking real money.
+PAPER_TRADE_MODE = True
+
+# Where paper-trade state lives (gitignored)
+_PAPER_OPEN_FILE   = Path("paper_open.json")
+_PAPER_TRADES_FILE = Path("paper_trades.csv")
 
 _TF_MAP = {
     "M5":  Timeframe.M5,  "M15": Timeframe.M15,
@@ -66,11 +86,11 @@ _TF_MAP = {
     "H4":  Timeframe.H4,  "D1":  Timeframe.D1,
 }
 
-# D1 kept as confirmation for higher-timeframe trend context.
+# Confirmation TFs per primary — mirrors backtest.py's mapping
-# W1 dropped — only the D1 scan used it, and we no longer scan D1.
 _TF_CONFIRMATIONS = {
     "M15": [Timeframe.H1,  Timeframe.D1],
     "H1":  [Timeframe.M15, Timeframe.D1],
+    "H4":  [Timeframe.H1,  Timeframe.D1],
 }
 
 _GRADE_RANK      = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5}
@@ -281,11 +301,17 @@ def _build_alert_card(tf: str, r: SignalReport, now_str: str) -> str:
     sym    = _grade_symbol(r.signal_quality)
     action = _signal_action(r.final_signal)
 
+    paper_tag = "📝 [PAPER] " if PAPER_TRADE_MODE else ""
     lines.append(
-        f"{sym} <b>{_h(r.signal_quality)} ALERT · {_h(r.display_name)}"
+        f"{paper_tag}{sym} <b>{_h(r.signal_quality)} ALERT · {_h(r.display_name)}"
         f" · {action} · {tf}</b>"
     )
     lines.append(f"🕐 {now_str}")
+    if PAPER_TRADE_MODE:
+        lines.append(
+            "<i>Paper-trade mode — this is logged to paper_trades.csv. "
+            "Do NOT trade with real money until validation period ends.</i>"
+        )
     lines.append("")
 
     # ── Trade setup ──────────────────────────────────────────────
@@ -374,6 +400,52 @@ def _build_alert_card(tf: str, r: SignalReport, now_str: str) -> str:
     return "\n".join(lines)
 
 
+def _walk_forward(pos: dict, forward_bars: pd.DataFrame, spread_pips: float) -> tuple:
+    """
+    First-touch SL/TP1 simulation on the primary timeframe — same logic as
+    backtest.simulate_trade. Returns (outcome, exit_price, exit_ts, bars_held, r_multiple).
+    Outcome is OPEN, WIN_TP1, LOSS_SL, or TIMEOUT.
+    """
+    if pos["risk_pips"] <= 0:
+        return "INVALID", 0.0, None, 0, 0.0
+
+    spread_r = spread_pips / pos["risk_pips"]
+    bars = 0
+    for ts, bar in forward_bars.iterrows():
+        bars += 1
+        if bars > pos["max_hold"]:
+            break
+        if pos["direction"] == "BUY":
+            sl_hit = bar["Low"]  <= pos["sl"]
+            tp_hit = bar["High"] >= pos["tp1"]
+        else:
+            sl_hit = bar["High"] >= pos["sl"]
+            tp_hit = bar["Low"]  <= pos["tp1"]
+
+        if sl_hit:
+            return "LOSS_SL", float(pos["sl"]), ts, bars, -1.0 - spread_r
+        if tp_hit:
+            r = (pos["reward_pips"] / pos["risk_pips"]) - spread_r
+            return "WIN_TP1", float(pos["tp1"]), ts, bars, r
+
+    if bars == 0:
+        return "OPEN", 0.0, None, 0, 0.0
+    if bars >= pos["max_hold"]:
+        last_idx = bars - 1
+        last = forward_bars.iloc[last_idx]
+        last_ts = forward_bars.index[last_idx]
+        if pos["direction"] == "BUY":
+            pnl_price = last["Close"] - pos["entry_price"]
+        else:
+            pnl_price = pos["entry_price"] - last["Close"]
+        sl_dist = abs(pos["entry_price"] - pos["sl"])
+        r = (pnl_price / sl_dist) - spread_r if sl_dist > 0 else 0.0
+        return "TIMEOUT", float(last["Close"]), last_ts, bars, r
+
+    # Reached end of available bars without SL/TP/timeout — still open
+    return "OPEN", 0.0, None, bars, 0.0
+
+
 # ══════════════════════════════════════════════════════════════════════
 #  Bot
 # ══════════════════════════════════════════════════════════════════════
@@ -381,7 +453,7 @@ def _build_alert_card(tf: str, r: SignalReport, now_str: str) -> str:
 class ForexTelegramBot:
     def __init__(self):
         self._executor = ThreadPoolExecutor(
-            max_workers=len(SCAN_TIMEFRAMES), thread_name_prefix="forex"
+            max_workers=max(1, len(SCAN_TIMEFRAMES)), thread_name_prefix="forex"
         )
         # Persistent ForexDataProvider so the OHLCV cache survives across
         # /scan and /signal calls. Without this each call burns API credits
@@ -393,6 +465,17 @@ class ForexTelegramBot:
         # setup persists. Bounded to avoid unbounded growth on long uptime.
         self._alerted_keys: set = set()
 
+        # Auto-scan visibility — exposed by /status
+        self._scan_count    = 0
+        self._alert_count   = 0
+        self._last_scan_at: Optional[datetime] = None
+        self._last_alert_at: Optional[datetime] = None
+        self._started_at = datetime.now(timezone.utc)
+
+        # Paper-trade tracker: open positions persist across restarts via
+        # paper_open.json; closed trades append to paper_trades.csv.
+        self._paper_open: List[dict] = self._load_paper_open()
+
     # ── Engine helpers ────────────────────────────────────────────────
 
     def _build_config(self, tf: str, pairs: List[str]) -> PlatformConfig:
@@ -461,6 +544,133 @@ class ForexTelegramBot:
     async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
         await self.cmd_help(update, context)
 
+    async def cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
+        """
+        /status — proves the auto-scan is alive and shows a summary.
+        Reports: uptime, scan count, last scan time, alert count, open
+        paper positions, paper-trade P&L (R-units).
+        """
+        now = datetime.now(timezone.utc)
+        uptime = now - self._started_at
+        h, rem = divmod(int(uptime.total_seconds()), 3600)
+        m, _ = divmod(rem, 60)
+
+        last_scan = (
+            f"{(now - self._last_scan_at).total_seconds()/60:.1f} min ago "
+            f"({self._last_scan_at.strftime('%H:%M:%S UTC')})"
+            if self._last_scan_at else "never (waiting for first scan)"
+        )
+        last_alert = (
+            f"{(now - self._last_alert_at).total_seconds()/60:.1f} min ago"
+            if self._last_alert_at else "no alerts yet"
+        )
+
+        # Paper P&L summary
+        paper_summary = self._paper_summary_lines()
+
+        text = (
+            f"🤖 <b>Bot Status</b>\n\n"
+            f"<b>Strategy:</b> {SCAN_PAIRS} on {SCAN_TIMEFRAMES} · "
+            f"grades {sorted(ALERT_GRADES)}"
+            f"{' · 📝 PAPER MODE' if PAPER_TRADE_MODE else ''}\n"
+            f"<b>Cadence:</b> every {AUTO_SCAN_INTERVAL//60} min\n\n"
+            f"<b>Uptime:</b> {h}h {m}m\n"
+            f"<b>Scans run:</b> {self._scan_count}\n"
+            f"<b>Last scan:</b> {last_scan}\n"
+            f"<b>Total alerts:</b> {self._alert_count}\n"
+            f"<b>Last alert:</b> {last_alert}\n\n"
+            f"{paper_summary}"
+        )
+        await update.message.reply_text(text, parse_mode=ParseMode.HTML)
+
+    async def cmd_paper(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
+        """
+        /paper — full paper-trade ledger summary: open positions and last 10 closed.
+        """
+        text = self._paper_full_report()
+        await update.message.reply_text(text, parse_mode=ParseMode.HTML)
+
+    def _paper_summary_lines(self) -> str:
+        """Short paper-trade stats block for /status."""
+        n_open = len(self._paper_open)
+        if not _PAPER_TRADES_FILE.exists():
+            return (
+                f"<b>Paper trades:</b>\n"
+                f"  Open: {n_open}\n"
+                f"  Closed: 0  (no trades closed yet)"
+            )
+        try:
+            df = pd.read_csv(_PAPER_TRADES_FILE)
+        except Exception:
+            return f"<b>Paper trades:</b>\n  Open: {n_open}  (csv unreadable)"
+        if df.empty:
+            return f"<b>Paper trades:</b>\n  Open: {n_open}\n  Closed: 0"
+        n = len(df)
+        wr = (df["r_multiple"] > 0).mean() * 100
+        total_r = df["r_multiple"].sum()
+        avg_r = df["r_multiple"].mean()
+        return (
+            f"<b>Paper trades:</b>\n"
+            f"  Open: {n_open}\n"
+            f"  Closed: {n}  ·  WR {wr:.1f}%  ·  total R <b>{total_r:+.2f}</b>  ·  avg R {avg_r:+.3f}\n"
+            f"<i>(backtest expectation: ~+0.30 R/trade for USDCAD H4 B)</i>"
+        )
+
+    def _paper_full_report(self) -> str:
+        lines = ["📝 <b>Paper-trade ledger</b>", ""]
+
+        # Open
+        if self._paper_open:
+            lines.append(f"<b>Open ({len(self._paper_open)}):</b>")
+            for p in self._paper_open:
+                lines.append(
+                    f"  • {_h(p['pair'])} {p['tf']} {p['direction']}  "
+                    f"@ {p['entry_price']:.5f}  "
+                    f"SL {p['sl']:.5f} / TP {p['tp1']:.5f}  "
+                    f"({p.get('bars_held', 0)}/{p['max_hold']} bars)"
+                )
+        else:
+            lines.append("<b>Open:</b> none")
+        lines.append("")
+
+        # Closed (last 10)
+        if not _PAPER_TRADES_FILE.exists():
+            lines.append("<b>Closed:</b> none yet")
+            return "\n".join(lines)
+
+        try:
+            df = pd.read_csv(_PAPER_TRADES_FILE)
+        except Exception as e:
+            lines.append(f"<b>Closed:</b> error reading {_PAPER_TRADES_FILE}: {e}")
+            return "\n".join(lines)
+
+        if df.empty:
+            lines.append("<b>Closed:</b> none yet")
+            return "\n".join(lines)
+
+        n = len(df)
+        wr = (df["r_multiple"] > 0).mean() * 100
+        total_r = df["r_multiple"].sum()
+        avg_r = df["r_multiple"].mean()
+        gross_win = df.loc[df["r_multiple"] > 0, "r_multiple"].sum()
+        gross_loss = -df.loc[df["r_multiple"] <= 0, "r_multiple"].sum()
+        pf = gross_win / gross_loss if gross_loss > 0 else float("inf")
+
+        lines.append(
+            f"<b>Closed ({n}):</b>  WR {wr:.1f}%  ·  totalR <b>{total_r:+.2f}</b>  ·  "
+            f"avgR {avg_r:+.3f}  ·  PF {pf:.2f}"
+        )
+        lines.append("")
+        lines.append("<b>Last 10 closed:</b>")
+        recent = df.tail(10)
+        for _, row in recent.iterrows():
+            emoji = "✅" if row["r_multiple"] > 0 else "❌"
+            lines.append(
+                f"  {emoji} {_h(row['pair'])} {row['tf']} {row['direction']}  "
+                f"{row['outcome']}  R={row['r_multiple']:+.2f}"
+            )
+        return "\n".join(lines)
+
     async def cmd_myid(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
         """
         Returns the current chat's ID so the operator can set ALERT_CHAT_ID
@@ -475,29 +685,185 @@ class ForexTelegramBot:
         )
         await update.message.reply_text(text, parse_mode=ParseMode.HTML)
 
+    # ── Paper-trade tracker ──────────────────────────────────────────
+
+    @staticmethod
+    def _load_paper_open() -> List[dict]:
+        if not _PAPER_OPEN_FILE.exists():
+            return []
+        try:
+            return json.loads(_PAPER_OPEN_FILE.read_text())
+        except Exception as e:
+            logger.warning(f"Could not load {_PAPER_OPEN_FILE}: {e}")
+            return []
+
+    def _save_paper_open(self) -> None:
+        try:
+            _PAPER_OPEN_FILE.write_text(json.dumps(self._paper_open, indent=2))
+        except Exception as e:
+            logger.error(f"Could not save {_PAPER_OPEN_FILE}: {e}")
+
+    def _append_closed_trade(self, row: dict) -> None:
+        """Append one closed-trade row to paper_trades.csv (header on first write)."""
+        write_header = not _PAPER_TRADES_FILE.exists()
+        try:
+            with _PAPER_TRADES_FILE.open("a", newline="") as f:
+                w = csv.DictWriter(f, fieldnames=list(row.keys()))
+                if write_header:
+                    w.writeheader()
+                w.writerow(row)
+        except Exception as e:
+            logger.error(f"Could not append to {_PAPER_TRADES_FILE}: {e}")
+
+    def _open_paper_position(self, tf: str, r: SignalReport) -> None:
+        """Record a fresh alert as an open paper position."""
+        if r.trade_setup is None:
+            return
+        ts = r.trade_setup
+        bar_ts = r.primary_analysis.timestamp if r.primary_analysis else ""
+        pos = {
+            "alert_id":    f"{r.symbol}|{tf}|{ts.direction}|{bar_ts}",
+            "symbol":      r.symbol,
+            "pair":        r.display_name,
+            "tf":          tf,
+            "grade":       r.signal_quality,
+            "direction":   ts.direction,
+            "entry_time":  bar_ts,                 # bar's close timestamp
+            "entry_price": ts.entry_price,
+            "sl":          ts.stop_loss,
+            "tp1":         ts.take_profit_1,
+            "tp2":         ts.take_profit_2,
+            "risk_pips":   ts.risk_pips,
+            "reward_pips": ts.reward_pips,
+            "score":       r.final_score,
+            "confidence":  r.final_confidence,
+            "regime":      r.market_regime,
+            "mtf_aligned": bool(r.mtf_agreement),
+            "max_hold":    100,
+            "status":      "OPEN",
+        }
+        # Don't double-open if alert_id already in flight
+        if any(p["alert_id"] == pos["alert_id"] for p in self._paper_open):
+            return
+        self._paper_open.append(pos)
+        self._save_paper_open()
+        logger.info(f"📝 paper_open: {pos['pair']} {tf} {ts.direction} @ {ts.entry_price:.5f}")
+
+    async def _check_paper_positions(self, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """
+        Walk forward each open paper position using the warm OHLCV cache.
+        First-touch SL/TP1 — same logic as backtest.simulate_trade. Closed
+        positions are appended to paper_trades.csv and a brief notification
+        is sent to the alert chat.
+        """
+        if not self._paper_open:
+            return
+
+        chat_id = context.job.data["chat_id"]
+        loop = asyncio.get_running_loop()
+        still_open: List[dict] = []
+        spread_pips = 1.0
+
+        for pos in self._paper_open:
+            symbol = pos["symbol"]
+            tf_enum = _TF_MAP[pos["tf"]]
+            try:
+                df = await loop.run_in_executor(
+                    None, self._provider.fetch_ohlcv, symbol, tf_enum,
+                )
+            except Exception as e:
+                logger.warning(f"check_paper: {symbol} fetch failed: {e}")
+                still_open.append(pos)
+                continue
+
+            entry_ts = pd.Timestamp(pos["entry_time"])
+            if entry_ts.tz is None:
+                entry_ts = entry_ts.tz_localize("UTC")
+            forward = df[df.index > entry_ts].head(pos["max_hold"])
+            if forward.empty:
+                still_open.append(pos)
+                continue
+
+            outcome, exit_price, exit_ts, bars_held, r_mult = _walk_forward(
+                pos, forward, spread_pips,
+            )
+
+            if outcome == "OPEN":
+                pos["bars_held"] = bars_held
+                still_open.append(pos)
+                continue
+
+            # Position closed — log to CSV and notify
+            closed = dict(pos)
+            closed["status"]      = "CLOSED"
+            closed["outcome"]     = outcome
+            closed["exit_time"]   = str(exit_ts)
+            closed["exit_price"]  = exit_price
+            closed["bars_held"]   = bars_held
+            closed["r_multiple"]  = round(r_mult, 4)
+            self._append_closed_trade(closed)
+            logger.info(
+                f"📊 paper_close: {pos['pair']} {pos['tf']} {pos['direction']} "
+                f"{outcome} R={r_mult:+.2f}"
+            )
+
+            # Notify chat
+            emoji = "✅" if r_mult > 0 else "❌"
+            msg = (
+                f"{emoji} <b>[PAPER] Position closed</b>\n"
+                f"{_h(pos['pair'])} {pos['tf']} {pos['direction']}\n"
+                f"Entry {pos['entry_price']:.5f} → Exit {exit_price:.5f}\n"
+                f"Outcome: <b>{outcome}</b> · R = <b>{r_mult:+.2f}</b>\n"
+                f"Held {bars_held} {pos['tf']} bars"
+            )
+            try:
+                await context.bot.send_message(
+                    chat_id=chat_id, text=msg, parse_mode=ParseMode.HTML,
+                )
+            except Exception as e:
+                logger.error(f"close-notify failed: {e}")
+
+        self._paper_open = still_open
+        self._save_paper_open()
+
     # ── Auto-scan job (every AUTO_SCAN_INTERVAL seconds) ─────────────
 
     async def _auto_scan_job(self, context: ContextTypes.DEFAULT_TYPE):
         """
-        Periodic scan that pushes A/A+ alerts to the configured chat.
+        Periodic scan: (1) checks any open paper positions for SL/TP exits,
-        Dedup: one alert per (symbol, tf, direction, candle_open) — so a
+        (2) emits new alerts for setups in ALERT_GRADES, (3) opens fresh
-        persistent setup re-alerts once per new bar, not every 10 min.
+        paper positions for every alert.
+
+        Dedup: one alert per (symbol, tf, direction, candle_open) — a
+        persistent setup re-alerts once per new bar, not every scan tick.
         """
         chat_id = context.job.data["chat_id"]
 
+        # 1. Walk open paper positions forward against the latest bars
+        await self._check_paper_positions(context)
+
+        # 2. Run the engine
         try:
             results_by_tf = await self._analyze_all_tfs(list(SCAN_PAIRS))
         except Exception as e:
             logger.error(f"Auto-scan failed: {e}", exc_info=True)
             return
 
+        # 3. Update visibility counters
+        self._scan_count += 1
+        self._last_scan_at = datetime.now(timezone.utc)
+        logger.info(
+            f"auto_scan #{self._scan_count}: scanned {SCAN_PAIRS} on "
+            f"{SCAN_TIMEFRAMES}; open paper positions: {len(self._paper_open)}"
+        )
+
+        # 4. Emit alerts for graded setups
         now_str = _now_est()
         sent = 0
         for tf, reports in results_by_tf.items():
             for r in reports:
                 if r.signal_quality not in ALERT_GRADES:
                     continue
-                # Skip if engine blocked it via event danger
                 if r.event_risk == "danger":
                     continue
                 ts = r.primary_analysis.timestamp if r.primary_analysis else ""
@@ -514,35 +880,55 @@ class ForexTelegramBot:
                         parse_mode=ParseMode.HTML,
                     )
                     sent += 1
+                    self._alert_count += 1
+                    self._last_alert_at = datetime.now(timezone.utc)
                     logger.info(
                         f"Alert sent: {r.symbol} {tf} {r.final_signal.name} "
                         f"({r.signal_quality})"
                     )
+                    # Open a paper position
+                    if PAPER_TRADE_MODE:
+                        self._open_paper_position(tf, r)
                 except Exception as e:
                     logger.error(f"Failed to send alert for {r.symbol} {tf}: {e}")
 
-        # Bound dedup memory: keep most recent 256 keys
         if len(self._alerted_keys) > 256:
             self._alerted_keys = set(list(self._alerted_keys)[-256:])
 
         if sent == 0:
-            logger.debug("Auto-scan complete: no new A-grade signals")
+            logger.debug(f"auto_scan #{self._scan_count}: no new alerts")
 
     async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
         tfs = " · ".join(SCAN_TIMEFRAMES)
+        pairs = " · ".join(p.replace("=X", "") for p in SCAN_PAIRS)
+        grades = " · ".join(sorted(ALERT_GRADES))
+        paper_note = (
+            "📝 <b>Paper-trade mode is ACTIVE.</b> "
+            "Alerts are tagged [PAPER] and logged to paper_trades.csv. "
+            "Do NOT use real capital until validation period ends.\n\n"
+            if PAPER_TRADE_MODE else ""
+        )
         text = (
             "🌐 <b>Forex Signal Intelligence Platform</b>\n\n"
+            f"{paper_note}"
+            "<b>Strategy (calibrated by 2-yr backtest):</b>\n"
+            f"  Pair: <code>{pairs}</code>\n"
+            f"  Timeframe: <code>{tfs}</code>\n"
+            f"  Alert grades: <code>{grades}</code>\n"
+            f"  Auto-scan: every {AUTO_SCAN_INTERVAL//60} min\n\n"
             "<b>Commands:</b>\n"
-            "  /signal EURUSD — Full signal analysis across all timeframes\n"
+            "  /status — auto-scan health + paper P&amp;L\n"
-            "  /scan          — Scan all pairs, compact Grade B+ summary\n"
+            "  /paper  — full paper-trade ledger\n"
-            "  /help          — This message\n\n"
+            "  /signal EURUSD — full analysis on one pair (any TF)\n"
-            f"<b>Timeframes analyzed:</b> <code>{tfs}</code>\n\n"
+            "  /scan   — manual scan of configured pairs\n"
+            "  /myid   — chat ID (for ALERT_CHAT_ID setup)\n"
+            "  /help   — this message\n\n"
             "<b>Signal Grades:</b>\n"
-            "  🔥 A+ — Strongest signal, MTF aligned, high confidence\n"
+            "  🔥 A+ — Highest engine score (⚠️ backtest: anti-edge)\n"
-            "  ✅ A  — Strong signal, MTF aligned\n"
+            "  ✅ A  — High engine score (⚠️ backtest: anti-edge)\n"
-            "  👍 B  — Decent setup, wait for confirmation\n"
+            "  👍 B  — Mid-confluence (✅ backtest: where the edge lives)\n"
-            "  ⚠️ C  — Weak signal, watch only\n"
+            "  ⚠️ C  — Weak signal\n"
-            "  📉 D  — Very weak, no action\n"
+            "  📉 D  — Very weak\n"
             "  ❌ F  — No clear signal\n\n"
             "<i>⚠️ Educational use only. Not financial advice.</i>"
         )
@@ -658,8 +1044,10 @@ class ForexTelegramBot:
 
         async def _register_commands(app: Application) -> None:
             await app.bot.set_my_commands([
+                BotCommand("status", "Auto-scan status, scan count, paper P&L"),
+                BotCommand("paper",  "Paper-trade ledger — open positions + last 10 closed"),
                 BotCommand("signal", "Signal analysis for one pair — /signal EURUSD"),
-                BotCommand("scan",   "Scan all pairs, best Grade B+ per pair"),
+                BotCommand("scan",   "Manual scan of configured pairs"),
                 BotCommand("myid",   "Show this chat's ID (for ALERT_CHAT_ID setup)"),
                 BotCommand("help",   "Commands and grade guide"),
             ])
@@ -675,6 +1063,8 @@ class ForexTelegramBot:
         app.add_handler(CommandHandler("help",   self.cmd_help))
         app.add_handler(CommandHandler("signal", self.cmd_signal))
         app.add_handler(CommandHandler("scan",   self.cmd_scan))
+        app.add_handler(CommandHandler("status", self.cmd_status))
+        app.add_handler(CommandHandler("paper",  self.cmd_paper))
         app.add_handler(CommandHandler("myid",   self.cmd_myid))
 
         # ── Auto-scan job: only enabled when ALERT_CHAT_ID is set ─────
@@ -713,14 +1103,25 @@ class ForexTelegramBot:
 # ══════════════════════════════════════════════════════════════════════
 
 def main():
+    # INFO level so the user can see auto-scan activity in stdout
+    # ("auto_scan #5: scanned ['USDCAD=X'] on ['H4'] …").
+    # Drop external libraries to WARNING to keep the log readable.
     logging.basicConfig(
-        level=logging.WARNING,
+        level=logging.INFO,
-        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        format="%(asctime)s [%(levelname)s] %(message)s",
         datefmt="%H:%M:%S",
     )
+    for noisy in ("httpx", "httpcore", "telegram", "apscheduler", "asyncio"):
+        logging.getLogger(noisy).setLevel(logging.WARNING)
+
     bot = ForexTelegramBot()
     app = bot.build_application()
-    print("🤖 Forex Signal Bot started. Press Ctrl+C to stop.")
+    print("🤖 Forex Signal Bot started.")
+    print(f"   Strategy: {SCAN_PAIRS} on {SCAN_TIMEFRAMES} · grades {sorted(ALERT_GRADES)}")
+    print(f"   Auto-scan: every {AUTO_SCAN_INTERVAL//60} min")
+    if PAPER_TRADE_MODE:
+        print(f"   📝 Paper-trade mode ON — logging to {_PAPER_TRADES_FILE}")
+    print("   Ctrl+C to stop.\n")
     app.run_polling(drop_pending_updates=True)