Audit execution and add frozen research phases

app.py

@@ -6,9 +6,10 @@ import gradio as gr
 import pandas as pd
 import plotly.graph_objects as go
 
-from config import DEFAULT_CRYPTO_SYMBOLS, DEFAULT_FOREX_SYMBOLS, MODEL_NAMES, PROVIDERS, TIMEFRAMES
+from config import DEFAULT_CRYPTO_SYMBOLS, DEFAULT_FOREX_SYMBOLS, MODEL_NAMES, PROVIDERS, RESEARCH_PHASES, TIMEFRAMES
 from data.providers import catalog
 from pipeline import RunConfig, backtest_result, export_run, live_result, model_readiness
+from research import research_run
 
 DISCLAIMER = '**Educational / research project — not financial advice.** Trading carries substantial risk of loss.'
 CSS = """
@@ -32,12 +33,16 @@ def server_port():
     return port
 
 
-def ui_config(provider, market, symbol, timeframe, mode, models, days, window, horizon, step, evaluations, features, spread_bps, fee_bps, slippage_bps, random_baseline):
+def ui_config(provider, market, symbol, timeframe, mode, models, days, window, horizon, step, evaluations, features, spread_bps, fee_bps, slippage_bps, random_baseline, inverted_signals=False, research_phase='Tune + Validate', minimum_edge_atr=0.20, minimum_risk_reward=1.20, atr_stop_multiplier=1.25):
     return RunConfig(provider=provider, market=market, symbol=symbol, timeframe=timeframe,
                      mode=mode, models=tuple(models or ()), lookback_days=days, window=window,
                      horizon=horizon, step=step, max_evaluations=evaluations, use_features=features,
                      spread_bps=spread_bps, fee_bps=fee_bps, slippage_bps=slippage_bps,
-                     random_baseline=random_baseline)
+                     random_baseline=random_baseline,
+                     inverted_signals=inverted_signals, research_phase=research_phase,
+                     minimum_edge_atr=minimum_edge_atr,
+                     minimum_risk_reward=minimum_risk_reward,
+                     atr_stop_multiplier=atr_stop_multiplier)
 
 
 def metadata_text(metadata):
@@ -129,7 +134,7 @@ def run_backtest_ui(*args):
             f"### Walk-forward evaluation completed — {failed} failed/skipped model evaluations\n\n"
             f"History window: **{config.window}** · horizon: **{config.horizon}** · origin stride: **{config.step}** · "
             f"at most **{config.max_evaluations} most-recent evaluation origins per phase/model**. "
-            "The report contains separate **TUNING (65%)** and **FINAL_CHECK (35%)** rows; final-check target candles are not used during tuning. "
+            f"The report uses the selected **{config.research_phase}** phases from a chronological **TUNE (60%) / VALIDATE (20%) / VAULT (20%)** split; later-phase target candles are not used during tuning. "
             'Orders are reselected using only each training window; no future target prices/features are supplied to models. '
             'Targets crossing gaps/market closures are explicitly skipped. Flat prices are classified as FLAT.\n\n'
             '**Accuracy denominator:** successful forecast evaluations; failures and coverage are shown separately. '
@@ -207,13 +212,21 @@ with gr.Blocks(title='MYTHOS · MAYTHOS × RINKING', theme=THEME, css=CSS) as de
                 spread_bps = gr.Number(value=1.0, minimum=0, maximum=1000, label='Assumed spread (bps)')
                 fee_bps = gr.Number(value=1.0, minimum=0, maximum=1000, label='Assumed fees (bps, round trip)')
                 slippage_bps = gr.Number(value=1.0, minimum=0, maximum=1000, label='Assumed slippage (bps)')
-            random_baseline = gr.Checkbox(value=False, label='Random-signal baseline (audit mode; default off)')
-            gr.Markdown('ARIMA-family order selection runs at every evaluation origin and may take several minutes. SL/TP uses closed-candle Wilder ATR plus prior swing structure; levels are omitted when no valid positive-risk target exists. Backtests enter at the next candle open. The backtest report splits ordered candles into 65% tuning and a final 35% check; final targets are never used during tuning. Parallel mode runs independent models concurrently (two workers). Future indicator covariates use the last observed values. Moirai/TimesFM require their real pretrained weights.')
+            with gr.Row():
+                random_baseline = gr.Checkbox(value=False, label='Matched random-signal control (200 runs; default off)')
+                inverted_signals = gr.Checkbox(value=False, label='Inverted-signal control (audit mode; default off)')
+                research_phase = gr.Dropdown(RESEARCH_PHASES, value='Tune + Validate', label='Research phase — Vault requires a frozen candidate')
+            with gr.Row():
+                minimum_edge_atr = gr.Number(value=0.20, minimum=0, maximum=100, label='Minimum forecast edge (ATR)')
+                minimum_risk_reward = gr.Number(value=1.20, minimum=0.01, maximum=100, label='Minimum reward/risk')
+                atr_stop_multiplier = gr.Number(value=1.25, minimum=0.05, maximum=100, label='Stop distance (ATR multiple)')
+            gr.Markdown('ARIMA-family order selection runs at every evaluation origin and may take several minutes. SL/TP uses closed-candle Wilder ATR plus prior swing structure; levels are omitted when no valid positive-risk target exists. Backtests enter at the next candle open and one position may be open at a time. The ordered-candle split is 60% TUNE, 20% VALIDATE, 20% VAULT; targets are horizon-trimmed at boundaries, and only the selected research phase is evaluated. Future indicator covariates use the last observed values. Moirai/TimesFM require their real pretrained weights.')
         with gr.Accordion('Model dependency status', open=False):
-            gr.Dataframe(value=model_readiness(), interactive=False)
+            dependency_table = gr.Dataframe(value=model_readiness(), interactive=False)
+            dependency_button = gr.Button('Refresh dependency status')
             gr.Markdown('“Packages detected” is not an inference or accuracy certification. See README for installing optional foundation models.')
 
-    inputs = [provider, market, symbol, timeframe, mode, models, days, window, horizon, step, evaluations, features, spread_bps, fee_bps, slippage_bps, random_baseline]
+    inputs = [provider, market, symbol, timeframe, mode, models, days, window, horizon, step, evaluations, features, spread_bps, fee_bps, slippage_bps, random_baseline, inverted_signals, research_phase, minimum_edge_atr, minimum_risk_reward, atr_stop_multiplier]
     with gr.Tab('Live Signal'):
         live_button = gr.Button('Refresh live signal', variant='primary')
         live_summary = gr.Markdown('Choose your configuration and refresh to fetch real closed candles.')
@@ -230,11 +243,20 @@ with gr.Blocks(title='MYTHOS · MAYTHOS × RINKING', theme=THEME, css=CSS) as de
         maythos_details = gr.Dataframe(label='MAYTHOS per-evaluation details', interactive=False)
         backtest_file = gr.File(label='Download backtest data, results & reproducibility manifest', interactive=False)
 
+    with gr.Tab('Frozen research'):
+        gr.Markdown('One immutable historical dataset, separate TUNE / VALIDATE / VAULT phases, fixed per-trade risk, explicit costs and matched controls. Settings are sent as JSON; only the requested phase is evaluated.')
+        research_settings = gr.JSON(value={'snapshot_only': True}, label='Research settings')
+        research_button = gr.Button('Run selected frozen research phase')
+        research_summary = gr.JSON(label='Auditable metrics')
+        research_file = gr.File(label='Download phase snapshot, trade rows and manifest', interactive=False)
+
     provider.change(default_symbols, inputs=[provider, market], outputs=[symbol, catalog_note], queue=False)
     market.change(default_symbols, inputs=[provider, market], outputs=[symbol, catalog_note], queue=False)
     search_button.click(search_symbols, inputs=[provider, market, query, symbol], outputs=[symbol, catalog_note], api_name='symbol_catalog')
     mode.change(update_mode, inputs=[mode, models], outputs=models, queue=False)
     models.change(enforce_single, inputs=[mode, models], outputs=models, queue=False)
+    dependency_button.click(model_readiness, outputs=dependency_table, api_name='dependency_status')
+    research_button.click(research_run, inputs=research_settings, outputs=[research_summary,research_file], concurrency_limit=1, concurrency_id='analysis', api_name='research')
     live_button.click(run_live_signal, inputs=inputs, outputs=[live_plot, live_summary, live_table, signal_table, live_file], concurrency_limit=1, concurrency_id='analysis', api_name='live_signal')
     backtest_button.click(run_backtest_ui, inputs=inputs, outputs=[forecast_summary, maythos_summary, forecast_details, maythos_details, backtest_status, backtest_file], concurrency_limit=1, concurrency_id='analysis', api_name='backtest')
 

backtest/backtester.py

@@ -3,7 +3,7 @@ import numpy as np
 import pandas as pd
 from features.feature_pipeline import slice_features
 from risk_management import (
-    CostModel, average_true_range, build_trade_levels, simulate_trade,
+    CostModel, TradeLevels, average_true_range, build_trade_levels, simulate_trade,
     summarize_trades,
 )
 from utils.helpers import MINUTES_PER_CANDLE, price_direction
@@ -30,6 +30,10 @@ def backtest_model(
     costs: CostModel = None,
     origin_start: int = None,
     origin_end: int = None,
+    minimum_edge_atr: float = 0.20,
+    minimum_risk_reward: float = 1.20,
+    atr_stop_multiplier: float = 1.25,
+    prevent_overlap: bool = True,
 ) -> pd.DataFrame:
     if not isinstance(df, pd.DataFrame):
         raise TypeError('Backtest candles must be provided as a pandas DataFrame.')
@@ -79,6 +83,7 @@ def backtest_model(
         if max_evaluations < 1:
             raise ValueError('max_evaluations must be positive.')
         positions = positions[-max_evaluations:]
+    next_entry_index = window
     for i in positions:
         history = close.iloc[i - window : i]
         feature_window = (
@@ -129,6 +134,7 @@ def backtest_model(
         actual_direction = price_direction(actual_next_close, current_close)
         correct = predicted_direction == actual_direction
         trade_direction = {'up': 'BUY', 'down': 'SELL'}.get(predicted_direction)
+        trade_taken = False
         try:
             levels = build_trade_levels(
                 df.iloc[:i], trade_direction, market, symbol,
@@ -142,16 +148,22 @@ def backtest_model(
             # was replaced by a distant swing target, creating timeout losses
             # unrelated to the forecast signal.
             forecast_only=True,
-            minimum_edge_atr=0.20,
-            minimum_risk_reward=1.20,
-            use_structure_stop=False,
+             minimum_edge_atr=minimum_edge_atr,
+             minimum_risk_reward=minimum_risk_reward,
+             use_structure_stop=False,
+             atr_stop_multiplier=atr_stop_multiplier,
             minimum_edge_bps=(
                 costs.spread_bps + costs.fee_bps + 2.0 * costs.slippage_bps
             ) * 1.25,
             )
+            trade_taken = levels is not None and (not prevent_overlap or i >= next_entry_index)
             trade = {'outcome': 'NO_LEVELS', 'exit_price': None, 'net_r': None, 'bars_held': 0}
-            if levels is not None:
+            if levels is not None and not trade_taken:
+                trade = {**trade, 'outcome': 'OVERLAP_SKIPPED'}
+            elif levels is not None:
                 trade = {**trade, **simulate_trade(levels, df.iloc[i:target_idx + 1], costs)}
+                if prevent_overlap:
+                    next_entry_index = i + max(1, int(trade.get('bars_held', 0) or 0))
             trade_error = None
         except Exception as error:
             levels = None
@@ -167,14 +179,14 @@ def backtest_model(
                 "actual_direction": actual_direction,
                 "correct": correct,
                 "continuous_target": continuous,
-                "trade_taken": levels is not None,
-                "entry": levels.entry if levels else np.nan,
-                "stop_loss": levels.stop_loss if levels else np.nan,
-                "take_profit": levels.take_profit if levels else np.nan,
-                "risk_reward": levels.risk_reward if levels else np.nan,
-                "atr": levels.atr if levels else np.nan,
-                "swing_support": levels.swing_support if levels else np.nan,
-                "swing_resistance": levels.swing_resistance if levels else np.nan,
+                 "trade_taken": trade_taken,
+                 "entry": levels.entry if levels and trade_taken else np.nan,
+                 "stop_loss": levels.stop_loss if levels and trade_taken else np.nan,
+                 "take_profit": levels.take_profit if levels and trade_taken else np.nan,
+                 "risk_reward": levels.risk_reward if levels and trade_taken else np.nan,
+                 "atr": levels.atr if levels and trade_taken else np.nan,
+                 "swing_support": levels.swing_support if levels and trade_taken else np.nan,
+                 "swing_resistance": levels.swing_resistance if levels and trade_taken else np.nan,
                 "outcome": trade.get('outcome'),
                 "exit_price": trade.get('exit_price'),
                 "net_r": trade.get('net_r'),
@@ -310,6 +322,183 @@ def backtest_random_signals(
     return pd.DataFrame(rows)
 
 
+MATCHED_RANDOM_RUNS = 200
+
+
+def _matched_levels(row, direction: str) -> TradeLevels | None:
+    """Reuse a model trade's causal distances for a direction control.
+
+    The control never inspects a future price.  It receives only the model's
+    already-eligible entry, stop/target distances, and the same future path
+    length used by the model trade.  Random directions are then permuted while
+    preserving the model trade count and long/short ratio.
+    """
+    try:
+        entry = float(row['entry'])
+        stop = float(row['stop_loss'])
+        target = float(row['take_profit'])
+        risk = abs(entry - stop)
+        reward = abs(target - entry)
+        atr = float(row.get('atr', 1.0))
+        support = float(row.get('swing_support', entry))
+        resistance = float(row.get('swing_resistance', entry))
+    except (TypeError, ValueError, KeyError):
+        return None
+    if any(not np.isfinite(value) or value <= 0 for value in (entry, risk, reward, atr, support, resistance)):
+        return None
+    if direction == 'BUY':
+        stop_loss = entry - risk
+        take_profit = entry + reward
+    elif direction == 'SELL':
+        stop_loss = entry + risk
+        take_profit = entry - reward
+    else:
+        return None
+    if stop_loss <= 0 or take_profit <= 0:
+        return None
+    return TradeLevels(
+        direction=direction,
+        entry=entry,
+        stop_loss=stop_loss,
+        take_profit=take_profit,
+        risk_per_unit=risk,
+        reward_per_unit=reward,
+        risk_reward=reward / risk,
+        atr=atr,
+        swing_support=support,
+        swing_resistance=resistance,
+        tick_size=1.0,
+        precision_digits=0,
+        level_basis='matched causal model-trade distances; control direction only',
+    )
+
+
+def matched_direction_control(
+    model_results: pd.DataFrame,
+    df: pd.DataFrame,
+    horizon: int,
+    costs: CostModel,
+    *,
+    inverted: bool = False,
+    runs: int = MATCHED_RANDOM_RUNS,
+    seed: int = 42,
+):
+    """Evaluate matched random or inverted directions without changing exits.
+
+    Only rows that the model actually traded are used.  Each random run is a
+    permutation of the model's BUY/SELL labels, so the trade count and the
+    long/short ratio are identical.  The original level distances and the
+    model's bars-held path length are retained; only direction is changed.
+    """
+    if not isinstance(model_results, pd.DataFrame) or model_results.empty:
+        empty = pd.DataFrame()
+        return {
+            'model': 'Inverted signals' if inverted else 'Random baseline',
+            'trades': 0,
+            'matched_trade_count': 0,
+            'control_runs': 1 if inverted else int(runs),
+            'net_profit_r': None,
+            'random_95pct_net_profit_r': None,
+            'random_5pct_net_profit_r': None,
+            'status': 'NO_MODEL_TRADES',
+        }, empty
+    eligible = model_results[
+        model_results.get('trade_taken', False).astype(bool) &
+        model_results['net_r'].notna()
+    ].copy()
+    eligible = eligible[eligible['predicted_direction'].isin(('up', 'down'))]
+    if eligible.empty:
+        empty = pd.DataFrame()
+        return {
+            'model': 'Inverted signals' if inverted else 'Random baseline',
+            'trades': 0,
+            'matched_trade_count': 0,
+            'control_runs': 1 if inverted else int(runs),
+            'net_profit_r': None,
+            'random_95pct_net_profit_r': None,
+            'random_5pct_net_profit_r': None,
+            'status': 'NO_MODEL_TRADES',
+        }, empty
+    if isinstance(runs, (bool, np.bool_)) or int(runs) < 1:
+        raise ValueError('Matched control runs must be a positive integer.')
+    runs = int(runs)
+    index_lookup = {str(timestamp): position for position, timestamp in enumerate(df.index)}
+    original = np.array(['BUY' if value == 'up' else 'SELL' for value in eligible['predicted_direction']], dtype=object)
+    control_summaries = []
+    representative_rows = []
+    actual_runs = 1 if inverted else runs
+    for run_number in range(actual_runs):
+        if inverted:
+            assigned = np.array(['SELL' if value == 'BUY' else 'BUY' for value in original], dtype=object)
+        else:
+            rng = np.random.default_rng(int(seed) + run_number)
+            assigned = rng.permutation(original)
+        trade_rows = []
+        for row_number, (_, source_row) in enumerate(eligible.iterrows()):
+            timestamp = str(source_row['datetime'])
+            target_idx = index_lookup.get(timestamp)
+            if target_idx is None:
+                try:
+                    target_idx = df.index.get_loc(pd.Timestamp(timestamp))
+                except (KeyError, TypeError, ValueError):
+                    continue
+            start_idx = int(target_idx) - int(horizon) + 1
+            bars_held = max(1, int(source_row.get('bars_held', horizon) or horizon))
+            if start_idx < 0 or start_idx >= len(df):
+                continue
+            path = df.iloc[start_idx:min(len(df), start_idx + min(bars_held, int(horizon)))]
+            levels = _matched_levels(source_row, str(assigned[row_number]))
+            if levels is None or path.empty:
+                continue
+            trade = simulate_trade(levels, path, costs)
+            trade_rows.append({
+                'trade_taken': True,
+                'net_r': trade.get('net_r'),
+                'outcome': trade.get('outcome'),
+                'predicted_direction': 'up' if levels.direction == 'BUY' else 'down',
+                'actual_direction': source_row.get('actual_direction'),
+                'datetime': timestamp,
+                'control_run': run_number + 1,
+                'entry': levels.entry,
+                'stop_loss': levels.stop_loss,
+                'take_profit': levels.take_profit,
+                'exit_price': trade.get('exit_price'),
+                'bars_held': trade.get('bars_held', 0),
+                'error': None,
+            })
+        details = pd.DataFrame(trade_rows)
+        control_summaries.append(summarize_trades(details))
+        if run_number == 0:
+            representative_rows = trade_rows
+    net_values = np.array([
+        float(summary['net_profit_r']) for summary in control_summaries
+        if summary.get('net_profit_r') is not None and np.isfinite(summary['net_profit_r'])
+    ], dtype=float)
+    if not len(net_values):
+        net_values = np.array([np.nan])
+    aggregate = {
+        'model': 'Inverted signals' if inverted else 'Random baseline',
+        'trades': int(len(eligible)),
+        'matched_trade_count': int(len(eligible)),
+        'control_runs': actual_runs,
+        'wins': round(float(np.mean([s['wins'] for s in control_summaries])), 6),
+        'losses': round(float(np.mean([s['losses'] for s in control_summaries])), 6),
+        'timeouts': round(float(np.mean([s['timeouts'] for s in control_summaries])), 6),
+        'win_rate_pct': round(float(np.mean([s['win_rate_pct'] for s in control_summaries])), 6),
+        'profit_factor': round(float(np.mean([s['profit_factor'] for s in control_summaries if s['profit_factor'] is not None and np.isfinite(s['profit_factor'])])), 6) if any(s['profit_factor'] is not None and np.isfinite(s['profit_factor']) for s in control_summaries) else None,
+        'expectancy_r': round(float(np.nanmean([s['expectancy_r'] for s in control_summaries])), 6),
+        'max_drawdown_r': round(float(np.nanmean([s['max_drawdown_r'] for s in control_summaries])), 6),
+        'average_r': round(float(np.nanmean([s['average_r'] for s in control_summaries])), 6),
+        'net_profit_r': round(float(np.nanmean(net_values)), 6),
+        'top3_removed_profit_r': round(float(np.nanmean([s['top3_removed_profit_r'] for s in control_summaries])), 6),
+        'random_mean_net_profit_r': round(float(np.nanmean(net_values)), 6),
+        'random_5pct_net_profit_r': round(float(np.nanquantile(net_values, 0.05)), 6),
+        'random_95pct_net_profit_r': round(float(np.nanquantile(net_values, 0.95)), 6),
+        'status': 'OK',
+    }
+    return aggregate, pd.DataFrame(representative_rows)
+
+
 def summarize_backtest(results: pd.DataFrame, model_name: str) -> dict:
     if results.empty:
         return {

config.py

@@ -47,6 +47,7 @@ CUSTOM_TIMEFRAMES = {
 TIMEFRAMES = ['1m', '2m', '5m', '10m', '15m', '30m', '1h', '2h']
 MODEL_NAMES = ['ARIMA', 'Auto-ARIMA', 'ARIMA-GARCH', 'Moirai', 'TimesFM']
 PROVIDERS = ['Yahoo Finance', 'Binance Spot', 'Twelve Data']
+RESEARCH_PHASES = ['Tune only', 'Tune + Validate', 'Vault only']
 MOIRAI_CHECKPOINT = 'Salesforce/moirai-1.1-R-small'
 TIMESFM_CHECKPOINT = 'google/timesfm-2.5-200m-pytorch'
 DEFAULT_SYMBOL = 'BTC-USD'

data/providers.py

@@ -146,7 +146,7 @@ def _validate_frame(frame):
     return validate_ohlcv(frame, normalize=True)
 
 
-def _binance_history(symbol, timeframe, days):
+def _binance_history(symbol, timeframe, days, end_utc=None):
     days = positive_integer(days, 'lookback_days', 10000)
     source = {'2m': '1m', '10m': '5m'}.get(timeframe, timeframe)
     if source not in MINUTES_PER_CANDLE:
@@ -159,6 +159,15 @@ def _binance_history(symbol, timeframe, days):
         end_ms = int(clock['serverTime'])
     except (KeyError, TypeError, ValueError):
         raise DataFetchError('Binance returned an invalid server clock response.') from None
+    provider_end_ms = end_ms
+    if end_utc is not None:
+        requested_end = pd.Timestamp(end_utc)
+        if requested_end.tzinfo is None:
+            raise DataFetchError('Historical end_utc must explicitly specify UTC.')
+        requested_end = requested_end.tz_convert('UTC')
+        if requested_end > pd.Timestamp(provider_end_ms, unit='ms', tz='UTC'):
+            raise DataFetchError('Historical end_utc cannot be in the future.')
+        end_ms = int(requested_end.timestamp() * 1000)
     start_ms = end_ms - int(days * 86_400_000)
     cursor = start_ms
     rows = []
@@ -194,7 +203,10 @@ def _binance_history(symbol, timeframe, days):
         ], columns=['Open', 'High', 'Low', 'Close', 'Volume'], index=pd.to_datetime([int(r[0]) for r in rows], unit='ms', utc=True)).astype(float)
     except (TypeError, ValueError, OverflowError):
         raise DataFetchError('Binance returned malformed OHLCV values.') from None
-    frame.attrs['provider_now'] = pd.Timestamp(end_ms, unit='ms', tz='UTC').isoformat()
+    frame.attrs['provider_now'] = pd.Timestamp(provider_end_ms, unit='ms', tz='UTC').isoformat()
+    frame.attrs['requested_start_utc'] = pd.Timestamp(start_ms, unit='ms', tz='UTC').isoformat()
+    frame.attrs['requested_end_utc_exclusive'] = pd.Timestamp(end_ms, unit='ms', tz='UTC').isoformat()
+    frame.attrs['effective_lookback_days'] = days
     frame.attrs['source_interval'] = source
     try:
         exchange_info = _binance_get('/exchangeInfo', {'symbol': symbol})
@@ -283,17 +295,19 @@ def _twelve_history(symbol, timeframe, days):
     return frame
 
 
-def fetch_market_history(provider, market, symbol, timeframe, lookback_days):
+def fetch_market_history(provider, market, symbol, timeframe, lookback_days, end_utc=None):
     if timeframe not in TIMEFRAMES:
         raise ValueError('Select one of: ' + ', '.join(TIMEFRAMES))
     # The UI caps user history at 730 days; feature warm-up may add a small
     # provider request padding before the requested period is trimmed.
     lookback_days = positive_integer(lookback_days, 'lookback_days', 10000)
     symbol = normalize_symbol(symbol, provider, market)
+    if end_utc is not None and provider != 'Binance Spot':
+        raise DataFetchError('Fixed historical endpoints are currently supported only by Binance Spot.')
     if provider == 'Yahoo Finance':
         frame = get_historical(symbol, timeframe, int(lookback_days))
     elif provider == 'Binance Spot':
-        frame = _binance_history(symbol, timeframe, int(lookback_days))
+        frame = _binance_history(symbol, timeframe, int(lookback_days), end_utc=end_utc)
     else:
         frame = _twelve_history(symbol, timeframe, int(lookback_days))
     frame = _validate_frame(frame)

models/timesfm_model.py

@@ -63,6 +63,33 @@ class TimesFMModel(BaseForecastModel):
         with self._lock, _MODEL_CACHE_LOCK:
             return self._predict(history, horizon, features)
 
+    def predict_batch(self, histories, horizon=1):
+        """Real pretrained inference on a bounded batch of closed histories."""
+        import torch
+        with self._lock, _MODEL_CACHE_LOCK:
+            torch.set_num_threads(2)
+            contexts=[]
+            for history in histories:
+                validate_history_and_horizon(history,horizon,self.name)
+                values=np.asarray(history,dtype=np.float32)[-self.max_context:]
+                if len(values)<32 or horizon>_MAX_HORIZON:
+                    raise ValueError('Invalid TimesFM batch context/horizon.')
+                contexts.append(values)
+            if not contexts:
+                return []
+            model=self._ensure_compiled_for(max(map(len,contexts)))
+            point,_=model.forecast(horizon=horizon,inputs=contexts)
+            output=[]
+            for item in point:
+                values=np.asarray(item,dtype=float)[-horizon:]
+                if values.shape!=(horizon,) or not np.isfinite(values).all() or (values<=0).any():
+                    raise ValueError('TimesFM batch returned invalid real-model forecasts.')
+                output.append(values.tolist())
+            self.last_diagnostics={'checkpoint':self.checkpoint,'context_candles':len(contexts[0]),
+                                   'backcast_trimmed':True,'batch_size':len(contexts),
+                                   'fallback':False,'weights_loaded':True}
+            return output
+
     def _predict(self, history, horizon, features):
         validate_history_and_horizon(history, horizon, self.name)
         values = np.asarray(history, dtype=np.float32)

pipeline.py

@@ -15,8 +15,11 @@ import zipfile
 import numpy as np
 import pandas as pd
 
-from backtest.backtester import backtest_model, backtest_random_signals, summarize_backtest
-from config import MODEL_NAMES, OUTPUT_DIR, PROVIDERS, RANDOM_SEED, TIMEFRAMES
+from backtest.backtester import (
+    MATCHED_RANDOM_RUNS, backtest_model, matched_direction_control,
+    summarize_backtest,
+)
+from config import MODEL_NAMES, OUTPUT_DIR, PROVIDERS, RANDOM_SEED, RESEARCH_PHASES, TIMEFRAMES
 from data.providers import fetch_market_history, normalize_symbol
 from features.feature_pipeline import compute_feature_frame_for_period
 from maythos_integration import backtest_maythos, latest_maythos_output
@@ -43,12 +46,20 @@ class RunConfig:
     fee_bps: float = 1.0
     slippage_bps: float = 1.0
     random_baseline: bool = False
+    inverted_signals: bool = False
+    research_phase: str = 'Tune + Validate'
+    minimum_edge_atr: float = 0.20
+    minimum_risk_reward: float = 1.20
+    atr_stop_multiplier: float = 1.25
+    end_utc: str | None = None
 
     def __post_init__(self):
         if self.provider not in PROVIDERS or self.timeframe not in TIMEFRAMES:
             raise ValueError('Choose a supported provider and timeframe.')
         if self.mode not in ('Single model', 'Parallel models'):
             raise ValueError('Choose Single model or Parallel models.')
+        if self.research_phase not in RESEARCH_PHASES:
+            raise ValueError('Choose a supported research phase.')
         if isinstance(self.models, str):
             raise ValueError('models must be a sequence of model names, not a string.')
         models = tuple(dict.fromkeys(self.models or ()))
@@ -78,11 +89,31 @@ class RunConfig:
             if isinstance(value, bool) or not finite or value < 0 or value > 1000:
                 raise ValueError(f'{name} must be a finite number between 0 and 1000 bps.')
             object.__setattr__(self, name, float(value))
+        for name, minimum in (
+            ('minimum_edge_atr', 0.0), ('minimum_risk_reward', 0.01),
+            ('atr_stop_multiplier', 0.05),
+        ):
+            value = getattr(self, name)
+            try:
+                finite = np.isfinite(value)
+            except TypeError:
+                finite = False
+            if isinstance(value, (bool, np.bool_)) or not finite or value < minimum or value > 100:
+                raise ValueError(f'{name} must be a finite number between {minimum} and 100.')
+            object.__setattr__(self, name, float(value))
         if not isinstance(self.random_baseline, (bool, np.bool_)):
             raise ValueError('random_baseline must be a boolean.')
+        if not isinstance(self.inverted_signals, (bool, np.bool_)):
+            raise ValueError('inverted_signals must be a boolean.')
         object.__setattr__(self, 'random_baseline', bool(self.random_baseline))
+        object.__setattr__(self, 'inverted_signals', bool(self.inverted_signals))
         object.__setattr__(self, 'models', models)
         object.__setattr__(self, 'symbol', normalize_symbol(self.symbol, self.provider, self.market))
+        if self.end_utc is not None:
+            ending = pd.Timestamp(self.end_utc)
+            if ending.tzinfo is None or self.provider != 'Binance Spot':
+                raise ValueError('end_utc requires an explicit UTC timestamp and Binance Spot.')
+            object.__setattr__(self, 'end_utc', ending.tz_convert('UTC').isoformat())
 
     @property
     def candle_step(self):
@@ -100,19 +131,26 @@ def model_readiness():
         'Moirai': ['uni2ts', 'gluonts', 'torch'], 'TimesFM': ['timesfm', 'torch'],
     }
     rows = []
+    timesfm_loaded = False
+    try:
+        from models.timesfm_model import _MODEL_CACHE
+        from config import TIMESFM_CHECKPOINT
+        timesfm_loaded = TIMESFM_CHECKPOINT in _MODEL_CACHE
+    except ImportError:
+        pass
     for name, packages in required.items():
         missing = [package for package in packages if importlib.util.find_spec(package) is None]
         rows.append({
             'Model': name,
             'Dependency status': 'Missing: ' + ', '.join(missing) if missing else 'Packages detected',
-            'Runtime requirement': 'Pretrained weights download on first use; inference must succeed' if name in ('Moirai', 'TimesFM') else 'Fits actual selected history',
+            'Runtime requirement': ('Real TimesFM pretrained weights loaded; no fallback' if name == 'TimesFM' and timesfm_loaded else 'Pretrained weights download on first use; inference must succeed') if name in ('Moirai', 'TimesFM') else 'Fits actual selected history',
         })
     return pd.DataFrame(rows)
 
 
 def prepare_history(config):
     padding = feature_warmup_days(config.timeframe, minimum_days=0) if config.use_features else 0
-    frame = fetch_market_history(config.provider, config.market, config.symbol, config.timeframe, config.lookback_days + padding)
+    frame = fetch_market_history(config.provider, config.market, config.symbol, config.timeframe, config.lookback_days + padding, end_utc=config.end_utc)
     attrs = dict(frame.attrs)
     features = None
     if config.use_features:
@@ -248,9 +286,10 @@ def live_result(config, frame=None, features=None):
                 frame, signal['direction'], config.market, config.symbol,
                 forecast_target=forecast_target,
                 forecast_only=True,
-                minimum_edge_atr=0.20,
-                minimum_risk_reward=1.20,
+                minimum_edge_atr=config.minimum_edge_atr,
+                minimum_risk_reward=config.minimum_risk_reward,
                 use_structure_stop=False,
+                atr_stop_multiplier=config.atr_stop_multiplier,
                 minimum_edge_bps=(
                     config.spread_bps + config.fee_bps + 2.0 * config.slippage_bps
                 ) * 1.25,
@@ -275,22 +314,29 @@ def backtest_result(config, frame=None, features=None):
     if len(frame) < config.window + config.horizon:
         raise ValueError(f'Backtesting needs at least {config.window + config.horizon} closed candles; provider returned {len(frame)}.')
 
-    # The first portion is available for configuration research.  The final
-    # portion is addressed by origin, not by slicing away its training
-    # history: a final origin may train on older candles, but no final target
-    # candle can enter tuning or model fitting before that origin.
-    split_index = int(len(frame) * 0.65)
-    tuning_end = split_index - config.horizon + 1
-    final_start = split_index
-    final_end = len(frame) - config.horizon + 1
-    if tuning_end <= config.window:
-        raise ValueError('The fetched history is too short to provide a 65% tuning segment after the selected training window and horizon.')
-    if final_end <= max(config.window, final_start):
-        raise ValueError('The fetched history is too short to provide a 35% final-check segment after the selected training window and horizon.')
-    phases = (
-        ('TUNING', 0, tuning_end),
-        ('FINAL_CHECK', final_start, final_end),
-    )
+    # Splits are chronological and expressed in origin coordinates. A phase's
+    # final origin is shortened by the horizon so no target candle from a later
+    # phase can enter an earlier phase's score.
+    tune_start = 0
+    validate_start = int(len(frame) * 0.60)
+    vault_start = int(len(frame) * 0.80)
+    phase_bounds = {
+        'TUNE': (tune_start, validate_start - config.horizon + 1),
+        'VALIDATE': (validate_start, vault_start - config.horizon + 1),
+        'VAULT': (vault_start, len(frame) - config.horizon + 1),
+    }
+    if phase_bounds['TUNE'][1] <= config.window:
+        raise ValueError('The fetched history is too short to provide a 60% TUNE segment after the selected training window and horizon.')
+    if phase_bounds['VALIDATE'][1] <= phase_bounds['VALIDATE'][0]:
+        raise ValueError('The fetched history is too short to provide a VALIDATE segment after the selected horizon.')
+    if phase_bounds['VAULT'][1] <= phase_bounds['VAULT'][0]:
+        raise ValueError('The fetched history is too short to provide a VAULT segment after the selected horizon.')
+    phase_names = {
+        'Tune only': ('TUNE',),
+        'Tune + Validate': ('TUNE', 'VALIDATE'),
+        'Vault only': ('VAULT',),
+    }
+    phases = tuple((phase, *phase_bounds[phase]) for phase in phase_names[config.research_phase])
 
     def evaluate_one(name, origin_start, origin_end, phase):
         try:
@@ -301,6 +347,10 @@ def backtest_result(config, frame=None, features=None):
                 max_evaluations=config.max_evaluations, timeframe=config.timeframe,
                 market=config.market, symbol=config.symbol, costs=config.costs,
                 origin_start=origin_start, origin_end=origin_end,
+                minimum_edge_atr=config.minimum_edge_atr,
+                minimum_risk_reward=config.minimum_risk_reward,
+                atr_stop_multiplier=config.atr_stop_multiplier,
+                prevent_overlap=True,
             )
             summary = summarize_backtest(details, name)
             diagnostics = dict(getattr(model, 'last_diagnostics', {}) or {})
@@ -341,37 +391,48 @@ def backtest_result(config, frame=None, features=None):
             config,
             lambda name: evaluate_one(name, origin_start, origin_end, phase),
         )
+        phase_model_details = {}
         for summary, details, diagnostics in evaluated:
             forecast_summaries.append(summary)
             forecast_details.append(details)
+            phase_model_details[summary['Model']] = details
             model_diagnostics[f'{phase}:{summary["Model"]}'] = diagnostics
 
+        def append_control(control_name, inverted=False):
+            for model_name, model_details in phase_model_details.items():
+                control_summary, control_details = matched_direction_control(
+                    model_details, frame, config.horizon, config.costs,
+                    inverted=inverted,
+                    runs=1 if inverted else MATCHED_RANDOM_RUNS,
+                    seed=RANDOM_SEED,
+                )
+                control_summary['Model'] = f'{control_summary.pop("model")} [{model_name}]'
+                control_summary['Phase'] = phase
+                control_summary['total_predictions'] = len(model_details)
+                control_summary['failed_predictions'] = 0
+                control_summary['evaluations_attempted'] = len(model_details)
+                control_summary['coverage_pct'] = 100.0 if len(model_details) else None
+                control_summary['cost_spread_bps'] = config.spread_bps
+                control_summary['cost_fee_bps'] = config.fee_bps
+                control_summary['cost_slippage_bps'] = config.slippage_bps
+                control_summary['control_type'] = 'inverted' if inverted else 'matched_random'
+                forecast_summaries.append(control_summary)
+                if not control_details.empty:
+                    control_details.insert(0, 'Model', control_summary['Model'])
+                    control_details.insert(1, 'Phase', phase)
+                    forecast_details.append(control_details)
+                model_diagnostics[f'{phase}:{control_summary["Model"]}'] = {
+                    'type': control_summary['control_type'],
+                    'seed': RANDOM_SEED,
+                    'runs': control_summary.get('control_runs'),
+                    'matched_trade_count': control_summary.get('matched_trade_count'),
+                    'fallback': False,
+                }
+
         if config.random_baseline:
-            random_details = backtest_random_signals(
-                frame, window=config.window, horizon=config.horizon, step=config.step,
-                max_evaluations=config.max_evaluations, timeframe=config.timeframe,
-                market=config.market, symbol=config.symbol, costs=config.costs,
-                seed=RANDOM_SEED, origin_start=origin_start, origin_end=origin_end,
-            )
-            random_summary = summarize_backtest(random_details, 'Random baseline')
-            random_summary['Model'] = random_summary.pop('model')
-            random_summary['Phase'] = phase
-            valid = random_details.dropna(subset=['correct'])
-            random_summary['evaluations_attempted'] = len(random_details)
-            random_summary['coverage_pct'] = round(100 * len(valid) / len(random_details), 2) if len(random_details) else None
-            random_summary['status'] = 'OK' if random_summary['failed_predictions'] == 0 else ('PARTIAL' if len(valid) else 'FAILED')
-            random_summary['cost_spread_bps'] = config.spread_bps
-            random_summary['cost_fee_bps'] = config.fee_bps
-            random_summary['cost_slippage_bps'] = config.slippage_bps
-            random_details.insert(0, 'Model', 'Random baseline')
-            random_details.insert(1, 'Phase', phase)
-            forecast_summaries.append(random_summary)
-            forecast_details.append(random_details)
-            model_diagnostics[f'{phase}:Random baseline'] = {
-                'type': 'deterministic random-direction control',
-                'seed': RANDOM_SEED,
-                'fallback': False,
-            }
+            append_control('Random baseline', inverted=False)
+        if config.inverted_signals:
+            append_control('Inverted signals', inverted=True)
 
         try:
             phase_signal_details, phase_signal_summary = backtest_maythos(
@@ -388,15 +449,32 @@ def backtest_result(config, frame=None, features=None):
         signal_summaries.append(phase_signal_summary)
         signal_details.append(phase_signal_details)
 
+    def split_timestamp(position):
+        if position >= len(frame):
+            return None
+        return pd.Timestamp(frame.index[position]).isoformat()
+
     split_metadata = {
-        'policy': 'first 65% of ordered closed candles for tuning; last 35% for final check',
-        'fraction_tuning': 0.65,
-        'split_index': split_index,
-        'tuning_candles': split_index,
-        'final_check_candles': len(frame) - split_index,
-        'tuning_origin_end_exclusive': tuning_end,
-        'final_origin_start_inclusive': final_start,
-        'final_origin_end_exclusive': final_end,
+        'policy': 'oldest 60% TUNE, next 20% VALIDATE, newest 20% VAULT; origins are horizon-trimmed and each run selects only its requested phase',
+        'fraction_tune': 0.60,
+        'fraction_validate': 0.20,
+        'fraction_vault': 0.20,
+        'research_phase': config.research_phase,
+        'active_phases': [phase for phase, _, _ in phases],
+        'phase_bounds': {
+            phase: {
+                'start_index': start,
+                'end_index_exclusive': end,
+                'start_candle_open_utc': split_timestamp(start),
+                'end_candle_open_utc_exclusive': split_timestamp(
+                    validate_start if phase == 'TUNE' else vault_start if phase == 'VALIDATE' else len(frame)
+                ),
+            }
+            for phase, (start, end) in phase_bounds.items()
+        },
+        'tune_candles': validate_start,
+        'validate_candles': vault_start - validate_start,
+        'vault_candles': len(frame) - vault_start,
         'final_targets_are_unseen_during_tuning': True,
     }
     return {

research.py

@@ -0,0 +1,359 @@
+"""Frozen-snapshot research API. All inference and execution run in the Space.
+
+The API accepts ordinary JSON settings. No local simulator, fake market data,
+model substitute, or pre-computed performance result is used.
+"""
+from dataclasses import asdict
+import hashlib
+import importlib.metadata
+import json
+from pathlib import Path
+import time
+import zipfile
+
+import numpy as np
+import pandas as pd
+
+from config import OUTPUT_DIR
+from data.providers import fetch_market_history
+from features.feature_pipeline import compute_feature_frame
+from pipeline import RunConfig, new_configured_model
+from risk_management import CostModel, TradeLevels, simulate_trade, summarize_trades
+from utils.helpers import MINUTES_PER_CANDLE
+
+
+_FRAMES = {}
+_FORECASTS = {}
+ROOT = Path(OUTPUT_DIR) / 'research'
+DEFAULT_DATA = {
+    'provider': 'Binance Spot', 'market': 'Crypto', 'symbol': 'BTCUSDT',
+    'timeframe': '15m', 'lookback_days': 270,
+    'end_utc': '2026-07-01T00:00:00+00:00',
+}
+
+
+def _clean(value):
+    if isinstance(value, dict):
+        return {key: _clean(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_clean(item) for item in value]
+    if isinstance(value, (float, np.floating)) and not np.isfinite(value):
+        return None
+    if isinstance(value, np.generic):
+        return value.item()
+    return value
+
+
+def _frame(settings):
+    data = {key: settings.get(key, default) for key, default in DEFAULT_DATA.items()}
+    key = json.dumps(data, sort_keys=True)
+    if key not in _FRAMES:
+        ROOT.mkdir(parents=True, exist_ok=True)
+        filename = ROOT / ('snapshot_' + hashlib.sha256(key.encode()).hexdigest()[:20] + '.csv')
+        metadata_filename = filename.with_suffix('.json')
+        if filename.exists() and metadata_filename.exists():
+            frame = pd.read_csv(filename, index_col=0, parse_dates=True, float_precision='round_trip')
+            frame.index = pd.to_datetime(frame.index, utc=True)
+            frame.attrs.update(json.loads(metadata_filename.read_text()))
+        else:
+            frame = fetch_market_history(**data)
+            frame.to_csv(filename, index_label='candle_open_utc', float_format='%.17g')
+            metadata_filename.write_text(json.dumps(frame.attrs, default=str, indent=2))
+        _FRAMES[key] = frame
+    frame = _FRAMES[key]
+    snapshot = frame.to_csv(index_label='candle_open_utc', float_format='%.17g')
+    digest = hashlib.sha256(snapshot.encode()).hexdigest()
+    if settings.get('expected_snapshot_sha256') and settings['expected_snapshot_sha256'] != digest:
+        raise ValueError('Frozen snapshot hash changed; no evaluation was run.')
+    return frame, data, digest
+
+
+def snapshot_info(settings):
+    frame, data, digest = _frame(settings)
+    n = len(frame)
+    a, b = int(n * .6), int(n * .8)
+    bounds = {'TUNE': (0, a), 'VALIDATE': (a, b), 'VAULT': (b, n)}
+    return {
+        'data_config': data, 'snapshot_sha256': digest, 'candles': n,
+        'first_utc': frame.index[0].isoformat(), 'last_utc': frame.index[-1].isoformat(),
+        'gaps': int((frame.index.to_series().diff().dropna() != pd.Timedelta(minutes=MINUTES_PER_CANDLE[data['timeframe']])).sum()),
+        'splits': {phase: {'start': start, 'end_exclusive': end,
+                           'first_utc': frame.index[start].isoformat(),
+                           'last_utc': frame.index[end-1].isoformat(),
+                           'candles': end-start}
+                   for phase, (start, end) in bounds.items()},
+        'provider_metadata': dict(frame.attrs),
+    }
+
+
+def _atr_series(frame, period=14):
+    previous = frame.Close.shift(1)
+    tr = pd.concat([frame.High-frame.Low, (frame.High-previous).abs(), (frame.Low-previous).abs()], axis=1).max(axis=1).to_numpy()
+    result = np.full(len(frame), np.nan)
+    if len(frame) >= period:
+        result[period-1] = tr[:period].mean()
+        for i in range(period, len(frame)):
+            result[i] = (result[i-1]*(period-1)+tr[i])/period
+    return result
+
+
+def _forecasts(frame, positions, settings, digest):
+    window = int(settings.get('window', 256))
+    horizon = int(settings.get('horizon', 12))
+    model_name = settings.get('model', 'TimesFM')
+    featured = bool(settings.get('use_features', False))
+    conf = RunConfig(**{key: settings.get(key, default) for key, default in DEFAULT_DATA.items()},
+                     models=(model_name,), window=window, horizon=horizon,
+                     use_features=featured)
+    model = new_configured_model(model_name, conf)
+    feature_frame = compute_feature_frame(frame) if featured else None
+    cache_key = json.dumps({'sha': digest, 'model': model_name, 'window': window,
+                            'horizon': horizon, 'features': featured}, sort_keys=True)
+    if cache_key not in _FORECASTS:
+        path = ROOT / ('forecast_' + hashlib.sha256(cache_key.encode()).hexdigest()[:20] + '.json')
+        _FORECASTS[cache_key] = json.loads(path.read_text()) if path.exists() else {}
+    cache = _FORECASTS[cache_key]
+    path = ROOT / ('forecast_' + hashlib.sha256(cache_key.encode()).hexdigest()[:20] + '.json')
+    uncached = [i for i in positions if str(i) not in cache]
+    # Real TimesFM batched inference uses the identical history windows. Small
+    # batches keep the 200M checkpoint comfortably within free CPU memory.
+    if uncached and model_name == 'TimesFM' and not featured:
+        for start in range(0, len(uncached), 8):
+            batch = uncached[start:start+8]
+            histories = [frame.Close.iloc[i-window:i] for i in batch]
+            try:
+                predictions = model.predict_batch(histories, horizon)
+                for i, prediction in zip(batch, predictions):
+                    cache[str(i)] = {'prediction': prediction, 'error': None}
+            except Exception as error:
+                for i in batch:
+                    cache[str(i)] = {'prediction': None, 'error': str(error)}
+            if start % 256 == 0:
+                path.write_text(json.dumps(cache))
+    else:
+        for i in uncached:
+            try:
+                feats = feature_frame.iloc[i-window:i].reset_index(drop=True) if featured else None
+                prediction = model.predict(frame.Close.iloc[i-window:i], horizon=horizon, features=feats)
+                cache[str(i)] = {'prediction': prediction, 'error': None}
+            except Exception as error:
+                cache[str(i)] = {'prediction': None, 'error': str(error)}
+    if uncached:
+        path.write_text(json.dumps(cache))
+    return {i: cache[str(i)] for i in positions}, dict(getattr(model, 'last_diagnostics', {}) or {}), len(uncached)
+
+
+def _levels(entry, atr, sign, stop_atr, reward_atr):
+    risk, reward = atr*stop_atr, atr*reward_atr
+    stop, target = entry-sign*risk, entry+sign*reward
+    if min(entry, risk, reward, stop, target) <= 0:
+        raise ValueError('Trade has nonpositive risk/levels.')
+    return TradeLevels('BUY' if sign > 0 else 'SELL', entry, stop, target,
+                       risk, reward, reward/risk, atr, entry, entry, .01, 2,
+                       'closed-candle Wilder ATR; fixed-risk position; real next-open fill')
+
+
+def _evaluate(frame, positions, forecast, settings, start, end, *, control=None):
+    horizon = int(settings.get('horizon', 12))
+    holding = int(settings.get('time_stop', horizon))
+    delay = int(settings.get('entry_delay', 0))
+    costs = CostModel(float(settings.get('spread_bps', 1)), float(settings.get('fee_bps', 1)), float(settings.get('slippage_bps', 1)))
+    if min(costs.spread_bps, costs.fee_bps, costs.slippage_bps) < 1:
+        raise ValueError('Research requires at least 1/1/1 bps costs.')
+    atrs = _atr_series(frame)
+    close = frame.Close.to_numpy()
+    trend_length = int(settings.get('trend_length', 48))
+    mean = frame.Close.rolling(trend_length).mean().to_numpy()
+    threshold = float(settings.get('threshold', .5))
+    stop_atr = float(settings.get('stop_atr', 1.25))
+    rr = float(settings.get('reward_risk', 1.5))
+    cost_multiple = float(settings.get('cost_multiple', 1.25))
+    strategy = settings.get('strategy', 'forecast')
+    next_available = start
+    daily = {}
+    rows = []
+    span = pd.Timedelta(minutes=MINUTES_PER_CANDLE[settings.get('timeframe', '15m')])
+    max_daily = int(settings.get('max_trades_per_day', 10000))
+    cooldown = int(settings.get('cooldown', 0))
+    for i in positions:
+        record = forecast[i]
+        row = {'origin_index': i, 'signal_utc': (frame.index[i-1]+span).isoformat(),
+               'trade_taken': False, 'net_r': np.nan, 'net_r_x2': np.nan,
+               'outcome': 'NO_TRADE', 'error': record['error'],
+               'entry': np.nan, 'stop_loss': np.nan, 'take_profit': np.nan,
+               'exit_price': np.nan, 'direction': None, 'bars_held': 0}
+        if record['error']:
+            row['outcome'] = 'FORECAST_FAILED'
+            rows.append(row)
+            continue
+        prediction = np.asarray(record['prediction'], dtype=float)
+        if len(prediction) != horizon or not np.isfinite(prediction).all() or (prediction <= 0).any():
+            raise ValueError('Real model returned an invalid forecast path.')
+        atr = atrs[i-1]
+        price = close[i-1]
+        move = float(prediction[-1]-price)
+        sign = 1 if move > 0 else -1 if move < 0 else 0
+        row.update(predicted_close=float(prediction[-1]), current_close=price,
+                   forecast_edge_atr=move/atr, trend_z=(price-mean[i-1])/atr)
+        if not np.isfinite(atr) or atr <= 0 or not sign:
+            rows.append(row); continue
+        cost_price = price*(costs.spread_bps+costs.fee_bps+2*costs.slippage_bps)/10000
+        if control != 'naive' and abs(move) < max(threshold*atr, cost_multiple*cost_price):
+            rows.append(row); continue
+        trend_sign = np.sign(price-mean[i-1])
+        if control == 'naive':
+            sign = int(np.sign(close[i-1]-close[i-2]))
+        elif strategy == 'inverted_forecast':
+            sign = -sign
+        elif strategy == 'forecast_trend' and sign != trend_sign:
+            rows.append(row); continue
+        elif strategy == 'forecast_reversion' and sign == trend_sign:
+            rows.append(row); continue
+        elif strategy == 'trend':
+            sign = int(trend_sign)
+        elif strategy == 'reversion':
+            z = (price-mean[i-1])/atr
+            if abs(z) < float(settings.get('regime_threshold', 2)):
+                rows.append(row); continue
+            sign = -int(np.sign(z))
+        elif strategy not in ('forecast','forecast_trend','forecast_reversion','inverted_forecast'):
+            raise ValueError('Unknown research strategy.')
+        if not sign:
+            rows.append(row); continue
+        if control == 'inverted':
+            sign = -sign
+        entry_index = i+delay
+        if entry_index < next_available:
+            row['outcome'] = 'OVERLAP_SKIPPED'; rows.append(row); continue
+        if entry_index+holding > end or entry_index >= len(frame):
+            row['outcome'] = 'BOUNDARY_SKIPPED'; rows.append(row); continue
+        times = frame.index[i-1:entry_index+holding]
+        if not bool((times.to_series().diff().dropna() == span).all()):
+            row['outcome'] = 'GAP_SKIPPED'; row['error'] = 'Missing candle/market-closure path'; rows.append(row); continue
+        day = str(frame.index[entry_index].date())
+        if daily.get(day, 0) >= max_daily:
+            row['outcome'] = 'DAILY_LIMIT'; rows.append(row); continue
+        hour = frame.index[i-1].hour
+        if 'session_start' in settings and not (int(settings['session_start']) <= hour < int(settings.get('session_end', 24))):
+            row['outcome'] = 'SESSION_FILTER'; rows.append(row); continue
+        if control == 'naive' and not sign:
+            rows.append(row); continue
+        entry = float(frame.Open.iloc[entry_index])
+        # Causal level distances are fixed before the entry candle's high/low.
+        # Each trade is one R of fixed cash risk; equity never rescales risk.
+        levels = _levels(entry, atr, sign, stop_atr, stop_atr*rr)
+        path = frame.iloc[entry_index:entry_index+holding]
+        trade = simulate_trade(levels, path, costs)
+        stress = simulate_trade(levels, path, CostModel(2*costs.spread_bps,2*costs.fee_bps,2*costs.slippage_bps))
+        row.update(trade_taken=True, entry=entry, stop_loss=levels.stop_loss,
+                   take_profit=levels.take_profit, risk_per_unit=levels.risk_per_unit,
+                   atr=atr, direction=levels.direction,
+                   entry_utc=frame.index[entry_index].isoformat(),
+                   net_r_x2=stress['net_r'], **trade)
+        next_available = entry_index+int(trade['bars_held'])+cooldown
+        daily[day] = daily.get(day,0)+1
+        rows.append(row)
+    return pd.DataFrame(rows)
+
+
+def _controls(frame, rows, settings, runs=200):
+    """Matched long/short permutations with the same SL/TP/time-stop rules.
+
+    A matched slot is capped by the following real trade entry, ensuring every
+    control is non-overlapping. Both original and control use this same cap.
+    """
+    traded = rows[rows.trade_taken].copy()
+    if traded.empty:
+        return {'runs':runs, 'trades':0, 'mean_r':None, 'p95_mean_r':None, 'p99_mean_r':None}, pd.DataFrame()
+    directions = np.where(traded.direction.eq('BUY'),1,-1)
+    costs = CostModel(settings.get('spread_bps',1),settings.get('fee_bps',1),settings.get('slippage_bps',1))
+    holding = int(settings.get('time_stop',settings.get('horizon',12)))
+    stop_atr = float(settings.get('stop_atr',1.25)); rr = float(settings.get('reward_risk',1.5))
+    slots = []
+    records = traded.to_dict('records')
+    for j,row in enumerate(records):
+        ix = frame.index.get_loc(pd.Timestamp(row['entry_utc']))
+        limit = ix+holding
+        if j+1 < len(records):
+            limit = min(limit, frame.index.get_loc(pd.Timestamp(records[j+1]['entry_utc'])))
+        path = frame.iloc[ix:limit]
+        values = {}
+        for sign in (1,-1):
+            levels = _levels(row['entry'],row['atr'],sign,stop_atr,stop_atr*rr)
+            values[sign] = simulate_trade(levels,path,costs)['net_r']
+        slots.append(values)
+    totals = []
+    for seed in range(42,42+int(runs)):
+        permutation = np.random.default_rng(seed).permutation(directions)
+        returns = np.array([slot[int(sign)] for slot,sign in zip(slots,permutation)])
+        totals.append({'seed':seed,'trades':len(returns),'mean_r':float(returns.mean()),'net_profit_r':float(returns.sum())})
+    random = pd.DataFrame(totals)
+    inverted = np.array([slot[-int(sign)] for slot,sign in zip(slots,directions)])
+    original = np.array([slot[int(sign)] for slot,sign in zip(slots,directions)])
+    return {'runs':runs,'trades':len(traded),'longs':int((directions>0).sum()),
+            'shorts':int((directions<0).sum()),
+            'mean_r':float(random.mean_r.mean()),
+            'p95_mean_r':float(random.mean_r.quantile(.95)),
+            'p99_mean_r':float(random.mean_r.quantile(.99)),
+            'original_matched_mean_r':float(original.mean()),
+            'inverted_mean_r':float(inverted.mean()),
+            'inverted_profit_r':float(inverted.sum())}, random
+
+
+def research_run(settings):
+    if isinstance(settings,str):
+        settings = json.loads(settings)
+    settings = dict(settings)
+    if settings.pop('snapshot_only',False):
+        return _clean(snapshot_info(settings)), None
+    frame,data,digest = _frame(settings)
+    settings.update(data)
+    n=len(frame); a=int(n*.6); b=int(n*.8)
+    phase=settings.get('phase','TUNE')
+    if phase not in ('TUNE','VALIDATE','VAULT'):
+        raise ValueError('Choose exactly one TUNE, VALIDATE, or VAULT phase.')
+    start,end={'TUNE':(0,a),'VALIDATE':(a,b),'VAULT':(b,n)}[phase]
+    window=int(settings.get('window',256)); horizon=int(settings.get('horizon',12))
+    stride=int(settings.get('step',6)); max_eval=int(settings.get('max_evaluations',20000))
+    holding=int(settings.get('time_stop',horizon)); delay=int(settings.get('entry_delay',0))
+    if not (32<=window<=2048 and 1<=horizon<=64 and stride>=1 and 1<=holding<=512 and delay in (0,1)):
+        raise ValueError('Invalid window/horizon/stride/time-stop/entry-delay.')
+    positions=list(range(max(window,start),end-max(horizon,holding+delay)+1,stride))
+    if len(positions)>max_eval:
+        raise ValueError(f'{len(positions)} evaluation origins exceed max_evaluations; increase it to cover the full frozen phase.')
+    predictions,diagnostics,new_inferences=_forecasts(frame,positions,settings,digest)
+    rows=_evaluate(frame,positions,predictions,settings,start,end)
+    summary=summarize_trades(rows)
+    traded=rows[rows.trade_taken & rows.net_r.notna()]
+    summary.update(phase=phase,evaluations=len(rows),failed_evaluations=int(rows.error.notna().sum()),
+                   failed_pct=100*float(rows.error.notna().mean()) if len(rows) else None,
+                   net_profit_r_x2=float(traded.net_r_x2.sum()) if len(traded) else None,
+                   win_rate_net_pct=100*float(traded.net_r.gt(0).mean()) if len(traded) else None,
+                   fresh_model_inferences=new_inferences)
+    naive=_evaluate(frame,positions,predictions,settings,start,end,control='naive')
+    summary['naive']=summarize_trades(naive)
+    random_rows=pd.DataFrame()
+    if settings.get('audit',False):
+        summary['controls'],random_rows=_controls(frame,rows,settings,200)
+    run_id=hashlib.sha256(json.dumps(settings,sort_keys=True).encode()).hexdigest()[:12]+'_'+str(time.time_ns())
+    directory=ROOT/run_id; directory.mkdir(parents=True,exist_ok=False)
+    # Export only active phase plus its causal history, keeping the vault sealed
+    # in TUNE/VALIDATE artifacts. The complete snapshot hash identifies it.
+    phase_frame=frame.iloc[max(0,start-window):end]
+    phase_frame.to_csv(directory/'candles.csv',index_label='candle_open_utc',float_format='%.17g')
+    rows.to_csv(directory/'trades.csv',index=False,float_format='%.17g')
+    naive.to_csv(directory/'naive_trades.csv',index=False,float_format='%.17g')
+    random_rows.to_csv(directory/'random_runs.csv',index=False,float_format='%.17g')
+    clean_summary=_clean(summary)
+    (directory/'summary.json').write_text(json.dumps(clean_summary,indent=2))
+    manifest={'config':settings,'full_snapshot_sha256':digest,'phase':phase,
+              'phase_start_index':start,'phase_end_index_exclusive':end,
+              'snapshot':snapshot_info(settings),'diagnostics':diagnostics,
+              'execution_policy':'fixed 1R cash risk, next open, no overlap, SL first, gap paths rejected, 1/1/1 bps minimum, fees round-trip, slippage both sides',
+              'source_revision':__import__('os').getenv('SPACE_COMMIT_SHA',None),
+              'packages':{name:importlib.metadata.version(name) for name in ('numpy','pandas','torch','timesfm')}}
+    (directory/'manifest.json').write_text(json.dumps(_clean(manifest),indent=2,default=str))
+    zip_path=directory.with_suffix('.zip')
+    with zipfile.ZipFile(zip_path,'w',zipfile.ZIP_DEFLATED) as archive:
+        for file in directory.iterdir(): archive.write(file,file.name)
+    return clean_summary,str(zip_path)

risk_management.py

@@ -205,6 +205,7 @@ def build_trade_levels(
     minimum_risk_reward: float = MIN_RISK_REWARD,
     use_structure_stop: bool = True,
     minimum_edge_bps: float = 0.0,
+    atr_stop_multiplier: float = ATR_STOP_MULTIPLIER,
 ) -> TradeLevels | None:
     """Build causal, volatility-aware levels for one closed-candle signal.
 
@@ -227,10 +228,15 @@ def build_trade_levels(
         minimum_edge_bps = float(minimum_edge_bps)
     except (TypeError, ValueError):
         raise ValueError('minimum_edge_atr/minimum_edge_bps must be numeric and minimum_risk_reward must be numeric.') from None
+    try:
+        atr_stop_multiplier = float(atr_stop_multiplier)
+    except (TypeError, ValueError):
+        raise ValueError('atr_stop_multiplier must be a finite positive number.') from None
     if (isinstance(minimum_edge_atr, (bool, np.bool_)) or not np.isfinite(minimum_edge_atr) or minimum_edge_atr < 0 or
             isinstance(minimum_risk_reward, (bool, np.bool_)) or not np.isfinite(minimum_risk_reward) or minimum_risk_reward <= 0 or
-            isinstance(minimum_edge_bps, (bool, np.bool_)) or not np.isfinite(minimum_edge_bps) or minimum_edge_bps < 0):
-        raise ValueError('minimum_edge_atr/minimum_edge_bps must be non-negative and minimum_risk_reward must be positive.')
+            isinstance(minimum_edge_bps, (bool, np.bool_)) or not np.isfinite(minimum_edge_bps) or minimum_edge_bps < 0 or
+            not np.isfinite(atr_stop_multiplier) or atr_stop_multiplier <= 0):
+        raise ValueError('minimum_edge_atr/minimum_edge_bps must be non-negative, minimum_risk_reward and atr_stop_multiplier must be positive.')
     if (isinstance(atr_period, (bool, np.bool_)) or not isinstance(atr_period, numbers.Integral) or atr_period < 1 or
             isinstance(swing_lookback, (bool, np.bool_)) or not isinstance(swing_lookback, numbers.Integral) or swing_lookback < 1):
         raise ValueError('atr_period and swing_lookback must be positive integers.')
@@ -273,7 +279,7 @@ def build_trade_levels(
     resistance = float(pd.to_numeric(structure['High'], errors='coerce').max())
     buffer = max(ATR_STRUCTURE_BUFFER * atr, 2.0 * tick)
     if direction == 'BUY':
-        atr_stop = entry - ATR_STOP_MULTIPLIER * atr
+        atr_stop = entry - atr_stop_multiplier * atr
         stop_raw = min(atr_stop, support - buffer) if use_structure_stop else atr_stop
         forecast_valid = (forecast_value is not None and
                           np.isfinite(forecast_value) and forecast_value > 0 and forecast_value > entry)
@@ -289,7 +295,7 @@ def build_trade_levels(
         risk = entry - stop
         reward = target - entry
     else:
-        atr_stop = entry + ATR_STOP_MULTIPLIER * atr
+        atr_stop = entry + atr_stop_multiplier * atr
         stop_raw = max(atr_stop, resistance + buffer) if use_structure_stop else atr_stop
         forecast_valid = (forecast_value is not None and
                           np.isfinite(forecast_value) and forecast_value > 0 and forecast_value < entry)
@@ -386,6 +392,8 @@ def simulate_trade(levels: TradeLevels, path: pd.DataFrame, costs: CostModel) ->
         if direction == 'BUY':
             if opening is not None and opening <= levels.stop_loss:
                 exit_price, outcome = opening, 'SL'
+            elif low <= levels.stop_loss and high >= levels.take_profit:
+                exit_price, outcome = levels.stop_loss, 'SL'
             elif opening is not None and opening >= levels.take_profit:
                 exit_price, outcome = opening, 'TP'
             elif low <= levels.stop_loss:
@@ -397,6 +405,8 @@ def simulate_trade(levels: TradeLevels, path: pd.DataFrame, costs: CostModel) ->
         else:
             if opening is not None and opening >= levels.stop_loss:
                 exit_price, outcome = opening, 'SL'
+            elif high >= levels.stop_loss and low <= levels.take_profit:
+                exit_price, outcome = levels.stop_loss, 'SL'
             elif opening is not None and opening <= levels.take_profit:
                 exit_price, outcome = opening, 'TP'
             elif high >= levels.stop_loss:
@@ -439,8 +449,8 @@ def summarize_trades(details: pd.DataFrame) -> dict:
             'net_profit_r': None, 'top3_removed_profit_r': None,
         }
     returns = trades['net_r'].astype(float).to_numpy()
-    wins = int((trades['outcome'] == 'TP').sum())
-    losses = int((trades['outcome'] == 'SL').sum())
+    wins = int((returns > 0).sum())
+    losses = int((returns < 0).sum())
     timeouts = int((trades['outcome'] == 'TIMEOUT').sum())
     gross_profit = float(returns[returns > 0].sum())
     gross_loss = float(-returns[returns < 0].sum())