Spaces:
Runtime error
Runtime error
| #!/usr/bin/env python3 | |
| """ | |
| 100optimization.py — safe Hugging Face CRT strategy optimizer. | |
| This script is designed for a Hugging Face Space. It: | |
| 1. Calls load_dataset("Mikecode123/volatility_100_index") with remote dataset | |
| code disabled. | |
| 2. Locates volatility_100_index.zip in the dataset snapshot. | |
| 3. Inspects the archive and extracts CSV files only. Scripts, executables, | |
| symlinks, traversal paths, and oversized archives are rejected. | |
| 4. Loads M1.csv, M5.csv, M15.csv, M30.csv, H1.csv, and H4.csv case-insensitively. | |
| 5. Implements a clearly documented Candle Range Theory (CRT) rule: | |
| - use a completed H1 or H4 reference candle; | |
| - wait for a sweep of its high/low; | |
| - require a close back inside the reference range; | |
| - require the next-bar confirmation through the sweep candle extreme; | |
| - enter at confirmation close; | |
| - target the opposite edge of the reference range and evaluate staged | |
| ATR-based TP/SL outcomes. | |
| 6. Uses a bounded deterministic candidate search. The supervisor model can | |
| select only from supplied candidate IDs; it cannot generate code or change | |
| the evaluation rules. | |
| 7. Saves every new best result and milestone result (20%, 30%, 80%) to the | |
| Space checkpoint directory, then creates a downloadable ZIP. | |
| There is no guarantee that 80% accuracy is achievable. The target is a stopping | |
| criterion, not a promise. Validation is used for optimization; the final test | |
| period remains untouched until the end. | |
| Supervisor model (pinned and approved): | |
| google/flan-t5-small | |
| revision=0fc9ddf78a1e988dac52e2dac162b0ede4fd74ab | |
| trust_remote_code=False | |
| Expected Space requirements: | |
| datasets | |
| huggingface_hub | |
| transformers | |
| torch | |
| pandas | |
| numpy | |
| Research sources for the CRT specification: | |
| https://innercircletrader.net/tutorials/candle-range-theory-crt/ | |
| https://tradingwyckoff.com/en/crt/ | |
| The public CRT descriptions are educational retail sources, not peer-reviewed | |
| proof of profitability. The implementation therefore reports validation and | |
| test results separately and records coverage beside accuracy. | |
| """ | |
| from __future__ import annotations | |
| import gc | |
| import itertools | |
| import json | |
| import math | |
| import os | |
| import random | |
| import re | |
| import shutil | |
| import stat | |
| import time | |
| import zipfile | |
| from dataclasses import asdict, dataclass | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from typing import Any, Optional | |
| import numpy as np | |
| import pandas as pd | |
| try: | |
| from datasets import load_dataset | |
| except ImportError as exc: # pragma: no cover - dependency supplied by the Space | |
| raise RuntimeError( | |
| "Install the Space requirements first: datasets, huggingface_hub, " | |
| "transformers, torch, pandas, numpy" | |
| ) from exc | |
| # --------------------------------------------------------------------------- | |
| # Configuration | |
| # --------------------------------------------------------------------------- | |
| DATASET_ID = os.environ.get("HF_DATASET_ID", "Mikecode123/volatility_100_index") | |
| SUPERVISOR_MODEL_ID = "google/flan-t5-small" | |
| SUPERVISOR_REVISION = "0fc9ddf78a1e988dac52e2dac162b0ede4fd74ab" | |
| TF_MINUTES = { | |
| "M1": 1, | |
| "M5": 5, | |
| "M15": 15, | |
| "M30": 30, | |
| "H1": 60, | |
| "H4": 240, | |
| } | |
| CLASS_NAMES = { | |
| 0: "NO_TRADE", | |
| 1: "SL", | |
| 2: "BE", | |
| 3: "TP1", | |
| 4: "TP2", | |
| 5: "TP3", | |
| } | |
| LOCAL_R_VALUE = np.array([0.0, -1.0, 0.0, 2.0, 4.0, 6.0], dtype=np.float32) | |
| LOCAL_RANK = np.array([1, 0, 2, 3, 4, 5], dtype=np.int8) | |
| class Config: | |
| dataset_id: str = DATASET_ID | |
| seed: int = int(os.environ.get("OPTIMIZER_SEED", "42")) | |
| target_accuracy: float = float(os.environ.get("TARGET_ACCURACY", "0.80")) | |
| max_trials: int = int(os.environ.get("MAX_TRIALS", "30")) | |
| min_trades: int = int(os.environ.get("MIN_TRADES", "100")) | |
| min_coverage: float = float(os.environ.get("MIN_COVERAGE", "0.001")) | |
| max_rows: int = int(os.environ.get("MAX_ROWS", "0")) | |
| reference_timeframe: str = os.environ.get("CRT_REFERENCE_TF", "H1").upper() | |
| confirm_bars: int = int(os.environ.get("CRT_CONFIRM_BARS", "1")) | |
| atr_period: int = int(os.environ.get("ATR_PERIOD", "14")) | |
| default_sl_atr_mult: float = float(os.environ.get("SL_ATR_MULT", "1.0")) | |
| default_horizon: int = int(os.environ.get("MAX_HORIZON", "60")) | |
| checkpoint_dir: Path = Path( | |
| os.environ.get("CHECKPOINT_DIR", "/data/100optimization_checkpoints") | |
| ) | |
| data_dir: Path = Path( | |
| os.environ.get("DATA_EXTRACT_DIR", "/tmp/volatility_100_index_data") | |
| ) | |
| CFG = Config() | |
| # --------------------------------------------------------------------------- | |
| # Logging and filesystem helpers | |
| # --------------------------------------------------------------------------- | |
| def log(message: str, *args: Any, level: str = "INFO") -> None: | |
| if args: | |
| message = message.format(*args) | |
| stamp = datetime.now().strftime("%H:%M:%S") | |
| print(f"[{stamp}] [{level}] {message}", flush=True) | |
| def utc_now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def ensure_checkpoint_dir() -> Path: | |
| try: | |
| CFG.checkpoint_dir.mkdir(parents=True, exist_ok=True) | |
| return CFG.checkpoint_dir | |
| except PermissionError: | |
| fallback = Path("./100optimization_checkpoints") | |
| fallback.mkdir(parents=True, exist_ok=True) | |
| log( | |
| "Cannot write to {}; using {} instead", | |
| CFG.checkpoint_dir, | |
| fallback, | |
| level="WARN", | |
| ) | |
| CFG.checkpoint_dir = fallback | |
| return fallback | |
| # --------------------------------------------------------------------------- | |
| # Hugging Face dataset discovery and safe archive handling | |
| # --------------------------------------------------------------------------- | |
| def _snapshot_dataset_files() -> Path: | |
| """Download only dataset files, never executable model or code files.""" | |
| from huggingface_hub import snapshot_download | |
| return Path( | |
| snapshot_download( | |
| repo_id=CFG.dataset_id, | |
| repo_type="dataset", | |
| allow_patterns=["*.zip", "*.ZIP", "*.csv", "*.CSV"], | |
| ) | |
| ) | |
| def locate_zip(snapshot_dir: Path) -> Path: | |
| candidates = sorted( | |
| p for p in snapshot_dir.rglob("*") | |
| if p.is_file() and p.suffix.lower() == ".zip" | |
| ) | |
| if not candidates: | |
| raise FileNotFoundError( | |
| f"No ZIP file was found in the dataset snapshot {snapshot_dir}." | |
| ) | |
| exact = [p for p in candidates if p.name.lower() == "volatility_100_index.zip"] | |
| if exact: | |
| return exact[0] | |
| if len(candidates) == 1: | |
| log("Using the only ZIP in the dataset snapshot: {}", candidates[0], level="WARN") | |
| return candidates[0] | |
| raise FileNotFoundError( | |
| "Multiple ZIP files were found and none was named " | |
| f"volatility_100_index.zip: {candidates}" | |
| ) | |
| def inspect_zip(zip_path: Path) -> list[zipfile.ZipInfo]: | |
| """Inspect the archive before extraction; allow data CSVs only.""" | |
| max_entries = 2_000 | |
| max_file_size = 25 * 1024 * 1024 | |
| max_total_size = 250 * 1024 * 1024 | |
| with zipfile.ZipFile(zip_path, "r") as archive: | |
| infos = archive.infolist() | |
| if len(infos) > max_entries: | |
| raise ValueError(f"Archive contains too many entries: {len(infos)}") | |
| total_size = 0 | |
| files: list[zipfile.ZipInfo] = [] | |
| log("Inspecting archive {}", zip_path) | |
| for info in infos: | |
| name = Path(info.filename.replace("\\", "/")) | |
| if name.is_absolute() or ".." in name.parts: | |
| raise ValueError(f"Unsafe archive path rejected: {info.filename}") | |
| file_mode = (info.external_attr >> 16) & 0o170000 | |
| if file_mode == stat.S_IFLNK: | |
| raise ValueError(f"Symlink rejected: {info.filename}") | |
| if info.is_dir(): | |
| log(" [directory] {}", info.filename) | |
| continue | |
| if name.suffix.lower() != ".csv": | |
| raise ValueError( | |
| f"Non-CSV/script/configuration member rejected: {info.filename}. " | |
| "This optimizer accepts a data-only archive." | |
| ) | |
| if info.file_size > max_file_size: | |
| raise ValueError(f"Archive member is too large: {info.filename}") | |
| total_size += info.file_size | |
| if total_size > max_total_size: | |
| raise ValueError("Archive expanded size exceeds safety limit") | |
| log(" [csv] {} ({:,} bytes)", info.filename, info.file_size) | |
| files.append(info) | |
| if not files: | |
| raise ValueError("Archive contains no CSV files") | |
| return files | |
| def safe_extract_zip(zip_path: Path, destination: Path) -> Path: | |
| infos = inspect_zip(zip_path) | |
| if destination.exists(): | |
| shutil.rmtree(destination) | |
| destination.mkdir(parents=True, exist_ok=True) | |
| root = destination.resolve() | |
| with zipfile.ZipFile(zip_path, "r") as archive: | |
| for info in infos: | |
| target = (destination / info.filename).resolve() | |
| if target != root and root not in target.parents: | |
| raise ValueError(f"Extraction escaped destination: {info.filename}") | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| with archive.open(info, "r") as source, target.open("wb") as sink: | |
| shutil.copyfileobj(source, sink, length=1024 * 1024) | |
| return destination | |
| def find_timeframe_file(directory: Path, timeframe: str) -> Optional[Path]: | |
| wanted = f"{timeframe}.csv".lower() | |
| for candidate in directory.iterdir(): | |
| if candidate.is_file() and candidate.name.lower() == wanted: | |
| return candidate | |
| return None | |
| def find_data_directory(root: Path) -> Path: | |
| for m1 in sorted(root.rglob("*.csv")): | |
| if m1.is_file() and m1.stem.lower() == "m1": | |
| required = ("M1", "M5", "M15", "M30", "H1", "H4") | |
| missing = [tf for tf in required if find_timeframe_file(m1.parent, tf) is None] | |
| if not missing: | |
| return m1.parent | |
| log("Ignoring {} because it is missing {}", m1.parent, missing, level="WARN") | |
| raise FileNotFoundError( | |
| f"Could not find a directory containing M1/M5/M15/M30/H1/H4 CSVs below {root}" | |
| ) | |
| def load_dataset_archive() -> Path: | |
| """Call load_dataset as requested, then use the ZIP snapshot safely.""" | |
| try: | |
| loaded = load_dataset(CFG.dataset_id, trust_remote_code=False) | |
| if hasattr(loaded, "keys"): | |
| log("load_dataset opened splits: {}", list(loaded.keys())) | |
| else: | |
| log("load_dataset opened {} rows", len(loaded)) | |
| except Exception as exc: | |
| # A repository containing only a ZIP may not have a Datasets builder. | |
| # This fallback downloads data files only and still inspects the ZIP | |
| # before extraction. | |
| log( | |
| "load_dataset could not parse the ZIP-only repository: {}. " | |
| "Using the data-file snapshot fallback.", | |
| exc, | |
| level="WARN", | |
| ) | |
| snapshot = _snapshot_dataset_files() | |
| archive = locate_zip(snapshot) | |
| log("Dataset archive: {}", archive) | |
| return archive | |
| # --------------------------------------------------------------------------- | |
| # OHLC loading and CRT feature construction | |
| # --------------------------------------------------------------------------- | |
| COLUMN_ALIASES = { | |
| "timestamp": ["timestamp", "time", "date", "datetime", "open_time"], | |
| "open": ["open", "o"], | |
| "high": ["high", "h"], | |
| "low": ["low", "l"], | |
| "close": ["close", "c", "adj_close"], | |
| "volume": ["volume", "vol", "v", "tick_volume"], | |
| } | |
| def resolve_column(columns: list[str], aliases: list[str]) -> Optional[str]: | |
| mapping = {column.lower(): column for column in columns} | |
| for alias in aliases: | |
| if alias in mapping: | |
| return mapping[alias] | |
| return None | |
| def load_ohlcv(path: Path) -> pd.DataFrame: | |
| header = pd.read_csv(path, nrows=0) | |
| columns = list(header.columns) | |
| resolved = { | |
| key: resolve_column(columns, aliases) | |
| for key, aliases in COLUMN_ALIASES.items() | |
| } | |
| required = ("timestamp", "open", "high", "low", "close") | |
| missing = [key for key in required if resolved[key] is None] | |
| if missing: | |
| raise ValueError(f"{path} is missing required columns: {missing}") | |
| usecols = [resolved[key] for key in required] | |
| if resolved["volume"] is not None: | |
| usecols.append(resolved["volume"]) | |
| dtype = { | |
| resolved[key]: np.float32 | |
| for key in ("open", "high", "low", "close") | |
| } | |
| if resolved["volume"] is not None: | |
| dtype[resolved["volume"]] = np.float32 | |
| raw = pd.read_csv( | |
| path, | |
| usecols=list(dict.fromkeys(usecols)), | |
| dtype=dtype, | |
| parse_dates=[resolved["timestamp"]], | |
| ) | |
| out = pd.DataFrame({ | |
| "timestamp": raw[resolved["timestamp"]], | |
| "open": raw[resolved["open"]], | |
| "high": raw[resolved["high"]], | |
| "low": raw[resolved["low"]], | |
| "close": raw[resolved["close"]], | |
| }) | |
| out["volume"] = ( | |
| raw[resolved["volume"]] | |
| if resolved["volume"] is not None | |
| else np.float32(0.0) | |
| ) | |
| out = ( | |
| out.dropna(subset=["timestamp", "open", "high", "low", "close"]) | |
| .sort_values("timestamp") | |
| .drop_duplicates("timestamp", keep="last") | |
| .reset_index(drop=True) | |
| ) | |
| invalid = ( | |
| (out["high"] < out["low"]) | |
| | (out[["open", "high", "low", "close"]] <= 0).any(axis=1) | |
| ) | |
| if invalid.any(): | |
| log("{}: dropping {} invalid rows", path.name, int(invalid.sum()), level="WARN") | |
| out = out.loc[~invalid].reset_index(drop=True) | |
| return out | |
| def compute_atr(df: pd.DataFrame, period: int) -> pd.Series: | |
| previous_close = df["close"].shift(1) | |
| true_range = pd.concat( | |
| [ | |
| df["high"] - df["low"], | |
| (df["high"] - previous_close).abs(), | |
| (df["low"] - previous_close).abs(), | |
| ], | |
| axis=1, | |
| ).max(axis=1) | |
| return true_range.rolling(period, min_periods=period).mean() | |
| def closed_reference_features( | |
| base: pd.DataFrame, | |
| reference: pd.DataFrame, | |
| timeframe: str, | |
| ) -> pd.DataFrame: | |
| """Attach only fully closed reference-candle levels to M1 rows.""" | |
| ref = reference[["timestamp", "open", "high", "low", "close"]].copy() | |
| ref["timestamp"] = ref["timestamp"] + pd.Timedelta( | |
| minutes=TF_MINUTES[timeframe] | |
| ) | |
| ref = ref.rename(columns={ | |
| "open": "crt_open", | |
| "high": "crt_high", | |
| "low": "crt_low", | |
| "close": "crt_close", | |
| }) | |
| return pd.merge_asof( | |
| base.sort_values("timestamp"), | |
| ref.sort_values("timestamp"), | |
| on="timestamp", | |
| direction="backward", | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Staged outcomes used for CRT backtesting | |
| # --------------------------------------------------------------------------- | |
| def simulate_direction( | |
| close: np.ndarray, | |
| high: np.ndarray, | |
| low: np.ndarray, | |
| risk: np.ndarray, | |
| direction: int, | |
| horizon: int, | |
| ) -> tuple[np.ndarray, np.ndarray]: | |
| """Simulate SL -> BE -> TP1-lock -> TP3 for one direction.""" | |
| n = len(close) | |
| sign = 1.0 if direction > 0 else -1.0 | |
| entry = close | |
| stop = entry - sign * risk | |
| tp1 = entry + sign * risk * 2.0 | |
| tp2 = entry + sign * risk * 4.0 | |
| tp3 = entry + sign * risk * 6.0 | |
| lock = tp1 | |
| stage = np.zeros(n, dtype=np.int8) | |
| resolved = np.zeros(n, dtype=bool) | |
| outcome = np.zeros(n, dtype=np.int8) | |
| valid = np.isfinite(risk) | |
| for step in range(1, horizon + 1): | |
| owners = np.arange(n) | |
| future = owners + step | |
| active = (~resolved) & valid & (future < n) | |
| if not active.any(): | |
| continue | |
| owners = owners[active] | |
| future = future[active] | |
| snapshot = stage[owners].copy() | |
| for stage_id, favorable, adverse in ( | |
| (0, tp1, stop), | |
| (1, tp2, entry), | |
| (2, tp3, lock), | |
| ): | |
| mask = snapshot == stage_id | |
| if not mask.any(): | |
| continue | |
| rows = owners[mask] | |
| hi = high[future[mask]] | |
| lo = low[future[mask]] | |
| if direction > 0: | |
| touched_favorable = hi >= favorable[rows] | |
| touched_adverse = lo <= adverse[rows] | |
| else: | |
| touched_favorable = lo <= favorable[rows] | |
| touched_adverse = hi >= adverse[rows] | |
| adverse_rows = rows[touched_adverse] | |
| favorable_rows = rows[touched_favorable & ~touched_adverse] | |
| if stage_id == 0: | |
| outcome[adverse_rows] = 1 | |
| resolved[adverse_rows] = True | |
| stage[favorable_rows] = 1 | |
| elif stage_id == 1: | |
| outcome[adverse_rows] = 2 | |
| resolved[adverse_rows] = True | |
| stage[favorable_rows] = 2 | |
| else: | |
| outcome[adverse_rows] = 3 | |
| resolved[adverse_rows] = True | |
| final_rows = rows[touched_favorable & ~touched_adverse] | |
| outcome[final_rows] = 5 | |
| resolved[final_rows] = True | |
| open_rows = (~resolved) & valid | |
| outcome[open_rows & (stage == 0)] = 0 | |
| outcome[open_rows & (stage == 1)] = 3 | |
| outcome[open_rows & (stage == 2)] = 4 | |
| return outcome, LOCAL_R_VALUE[outcome] | |
| def build_outcomes( | |
| m1: pd.DataFrame, | |
| atr_period: int, | |
| sl_atr_mult: float, | |
| horizon: int, | |
| ) -> dict[str, np.ndarray]: | |
| atr = compute_atr(m1, atr_period).to_numpy(dtype=np.float32) | |
| risk = np.float32(sl_atr_mult) * atr | |
| close = m1["close"].to_numpy(dtype=np.float32) | |
| high = m1["high"].to_numpy(dtype=np.float32) | |
| low = m1["low"].to_numpy(dtype=np.float32) | |
| long_outcome, long_r = simulate_direction(close, high, low, risk, 1, horizon) | |
| short_outcome, short_r = simulate_direction(close, high, low, risk, -1, horizon) | |
| long_rank = LOCAL_RANK[long_outcome] | |
| short_rank = LOCAL_RANK[short_outcome] | |
| choose_long = (long_rank > short_rank) | ( | |
| (long_rank == short_rank) & (long_r >= short_r) | |
| ) | |
| chosen_outcome = np.where(choose_long, long_outcome, short_outcome).astype(np.int8) | |
| chosen_direction = np.where( | |
| chosen_outcome == 0, | |
| 0, | |
| np.where(choose_long, 1, -1), | |
| ).astype(np.int8) | |
| return { | |
| "long_outcome": long_outcome, | |
| "short_outcome": short_outcome, | |
| "long_r": long_r, | |
| "short_r": short_r, | |
| "chosen_outcome": chosen_outcome, | |
| "chosen_direction": chosen_direction, | |
| } | |
| # --------------------------------------------------------------------------- | |
| # CRT signals and metrics | |
| # --------------------------------------------------------------------------- | |
| class Candidate: | |
| candidate_id: int | |
| reference_timeframe: str | |
| min_sweep_atr: float | |
| use_reference_bias: bool | |
| confirmation_window: int | |
| cooldown_bars: int | |
| sl_atr_mult: float | |
| horizon: int | |
| def make_candidates(seed: int) -> list[Candidate]: | |
| rng = random.Random(seed) | |
| candidates: list[Candidate] = [] | |
| candidate_id = 0 | |
| for values in itertools.product( | |
| ("H1", "H4"), | |
| (0.00, 0.05, 0.10, 0.25), | |
| (False, True), | |
| (1, 2), | |
| (0, 5, 15), | |
| (0.75, 1.00, 1.50), | |
| (30, 60), | |
| ): | |
| candidates.append(Candidate(candidate_id=candidate_id, **dict(zip( | |
| ( | |
| "reference_timeframe", | |
| "min_sweep_atr", | |
| "use_reference_bias", | |
| "confirmation_window", | |
| "cooldown_bars", | |
| "sl_atr_mult", | |
| "horizon", | |
| ), | |
| values, | |
| )))) | |
| candidate_id += 1 | |
| rng.shuffle(candidates) | |
| return candidates | |
| def generate_crt_signals( | |
| m1: pd.DataFrame, | |
| reference: pd.DataFrame, | |
| atr: pd.Series, | |
| candidate: Candidate, | |
| ) -> np.ndarray: | |
| base = m1[["timestamp", "open", "high", "low", "close"]].copy() | |
| merged = closed_reference_features(base, reference, candidate.reference_timeframe) | |
| atr_values = atr.to_numpy(dtype=np.float32) | |
| bullish_sweep = ( | |
| (merged["low"].to_numpy() < merged["crt_low"].to_numpy()) | |
| & (merged["close"].to_numpy() > merged["crt_low"].to_numpy()) | |
| & ( | |
| (merged["crt_low"].to_numpy() - merged["low"].to_numpy()) | |
| >= np.float32(candidate.min_sweep_atr) * atr_values | |
| ) | |
| ) | |
| bearish_sweep = ( | |
| (merged["high"].to_numpy() > merged["crt_high"].to_numpy()) | |
| & (merged["close"].to_numpy() < merged["crt_high"].to_numpy()) | |
| & ( | |
| (merged["high"].to_numpy() - merged["crt_high"].to_numpy()) | |
| >= np.float32(candidate.min_sweep_atr) * atr_values | |
| ) | |
| ) | |
| ref_open = merged["crt_open"].to_numpy() | |
| ref_close = merged["crt_close"].to_numpy() | |
| if candidate.use_reference_bias: | |
| bullish_sweep &= ref_close >= ref_open | |
| bearish_sweep &= ref_close <= ref_open | |
| highs = merged["high"].to_numpy() | |
| lows = merged["low"].to_numpy() | |
| closes = merged["close"].to_numpy() | |
| signals = np.zeros(len(merged), dtype=np.int8) | |
| # Confirmation is a close through the sweep candle's opposite extreme. | |
| for delay in range(1, candidate.confirmation_window + 1): | |
| prior_bull = np.zeros(len(merged), dtype=bool) | |
| prior_bear = np.zeros(len(merged), dtype=bool) | |
| if delay < len(merged): | |
| prior_bull[delay:] = bullish_sweep[:-delay] | |
| prior_bear[delay:] = bearish_sweep[:-delay] | |
| signals[delay:][prior_bull[delay:] & (closes[delay:] > highs[:-delay])] = 1 | |
| signals[delay:][prior_bear[delay:] & (closes[delay:] < lows[:-delay])] = -1 | |
| if candidate.cooldown_bars > 0: | |
| last_signal = -candidate.cooldown_bars - 1 | |
| for i in range(len(signals)): | |
| if signals[i] != 0: | |
| if i - last_signal <= candidate.cooldown_bars: | |
| signals[i] = 0 | |
| else: | |
| last_signal = i | |
| return signals | |
| def metrics_for_slice( | |
| signals: np.ndarray, | |
| outcomes: dict[str, np.ndarray], | |
| start: int, | |
| end: int, | |
| min_trades: int, | |
| min_coverage: float, | |
| ) -> dict[str, Any]: | |
| signal = signals[start:end] | |
| chosen_direction = outcomes["chosen_direction"][start:end] | |
| long_r = outcomes["long_r"][start:end] | |
| short_r = outcomes["short_r"][start:end] | |
| long_outcome = outcomes["long_outcome"][start:end] | |
| short_outcome = outcomes["short_outcome"][start:end] | |
| trades = signal != 0 | |
| n_rows = len(signal) | |
| n_trades = int(trades.sum()) | |
| coverage = float(n_trades / n_rows) if n_rows else 0.0 | |
| if n_trades: | |
| realized_r = np.where(signal > 0, long_r, short_r) | |
| realized_outcome = np.where(signal > 0, long_outcome, short_outcome) | |
| trade_accuracy = float((realized_r[trades] > 0).mean()) | |
| direction_accuracy = float( | |
| (signal[trades] == chosen_direction[trades]).mean() | |
| ) | |
| tp1_rate = float(np.isin(realized_outcome[trades], [3, 4, 5]).mean()) | |
| tp2_rate = float(np.isin(realized_outcome[trades], [4, 5]).mean()) | |
| tp3_rate = float((realized_outcome[trades] == 5).mean()) | |
| sl_rate = float((realized_outcome[trades] == 1).mean()) | |
| expectancy = float(realized_r[trades].mean()) | |
| total_r = float(realized_r[trades].sum()) | |
| positive = float(realized_r[trades][realized_r[trades] > 0].sum()) | |
| negative = float(-realized_r[trades][realized_r[trades] < 0].sum()) | |
| profit_factor = positive / negative if negative > 0 else math.inf | |
| else: | |
| trade_accuracy = direction_accuracy = 0.0 | |
| tp1_rate = tp2_rate = tp3_rate = sl_rate = 0.0 | |
| expectancy = total_r = 0.0 | |
| profit_factor = 0.0 | |
| eligible = n_trades >= min_trades and coverage >= min_coverage | |
| return { | |
| "rows": n_rows, | |
| "trades": n_trades, | |
| "coverage": coverage, | |
| "eligible": eligible, | |
| "trade_accuracy": trade_accuracy, | |
| "direction_accuracy": direction_accuracy, | |
| "tp1_or_better_rate": tp1_rate, | |
| "tp2_or_better_rate": tp2_rate, | |
| "tp3_rate": tp3_rate, | |
| "sl_rate": sl_rate, | |
| "expectancy_R": expectancy, | |
| "profit_factor": profit_factor, | |
| "total_R": total_r, | |
| } | |
| def objective(metrics: dict[str, Any]) -> tuple[float, float, float]: | |
| if not metrics["eligible"]: | |
| return (-1.0, metrics["coverage"], metrics["expectancy_R"]) | |
| return ( | |
| metrics["trade_accuracy"], | |
| metrics["coverage"], | |
| metrics["expectancy_R"], | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Pinned FLAN-T5 supervisor | |
| # --------------------------------------------------------------------------- | |
| class Supervisor: | |
| def __init__(self) -> None: | |
| from transformers import AutoModelForSeq2SeqLM, AutoTokenizer | |
| import torch | |
| self.torch = torch | |
| log( | |
| "Loading pinned supervisor {} at revision {}", | |
| SUPERVISOR_MODEL_ID, | |
| SUPERVISOR_REVISION, | |
| ) | |
| self.tokenizer = AutoTokenizer.from_pretrained( | |
| SUPERVISOR_MODEL_ID, | |
| revision=SUPERVISOR_REVISION, | |
| trust_remote_code=False, | |
| ) | |
| self.model = AutoModelForSeq2SeqLM.from_pretrained( | |
| SUPERVISOR_MODEL_ID, | |
| revision=SUPERVISOR_REVISION, | |
| trust_remote_code=False, | |
| ) | |
| self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") | |
| self.model.to(self.device) | |
| self.model.eval() | |
| def choose( | |
| self, | |
| candidates: list[Candidate], | |
| history: list[dict[str, Any]], | |
| ) -> tuple[Optional[int], str]: | |
| if not candidates: | |
| return None, "no candidates" | |
| candidate_text = "\n".join( | |
| f"ID {c.candidate_id}: ref={c.reference_timeframe}, " | |
| f"sweep_atr={c.min_sweep_atr}, bias={c.use_reference_bias}, " | |
| f"confirm={c.confirmation_window}, cooldown={c.cooldown_bars}, " | |
| f"sl={c.sl_atr_mult}, horizon={c.horizon}" | |
| for c in candidates[:24] | |
| ) | |
| history_text = "\n".join( | |
| f"trial {row['trial']}: candidate={row['candidate_id']}, " | |
| f"accuracy={row.get('validation', {}).get('trade_accuracy', 0):.4f}, " | |
| f"coverage={row.get('validation', {}).get('coverage', 0):.4f}" | |
| for row in history[-8:] | |
| ) or "No previous trials." | |
| prompt = ( | |
| "You are a constrained trading-strategy supervisor. Choose one " | |
| "candidate ID from the list. Do not invent an ID. Prefer enough " | |
| "coverage and realistic validation accuracy. Reply exactly as " | |
| "CANDIDATE_ID=<integer> followed by one short reason.\n\n" | |
| f"Candidates:\n{candidate_text}\n\nHistory:\n{history_text}" | |
| ) | |
| inputs = self.tokenizer( | |
| prompt, | |
| return_tensors="pt", | |
| truncation=True, | |
| max_length=768, | |
| ).to(self.device) | |
| with self.torch.no_grad(): | |
| output = self.model.generate(**inputs, max_new_tokens=48) | |
| reply = self.tokenizer.decode(output[0], skip_special_tokens=True) | |
| match = re.search(r"CANDIDATE_ID\s*=\s*(\d+)", reply) | |
| selected = int(match.group(1)) if match else None | |
| valid_ids = {candidate.candidate_id for candidate in candidates} | |
| if selected not in valid_ids: | |
| selected = None | |
| return selected, reply | |
| # --------------------------------------------------------------------------- | |
| # Checkpointing and optimization loop | |
| # --------------------------------------------------------------------------- | |
| def save_json(path: Path, value: Any) -> None: | |
| path.write_text(json.dumps(value, indent=2, default=str), encoding="utf-8") | |
| def save_checkpoint( | |
| checkpoint_dir: Path, | |
| candidate: Candidate, | |
| result: dict[str, Any], | |
| milestone: Optional[int] = None, | |
| ) -> None: | |
| prefix = "best" if milestone is None else f"milestone_{milestone:02d}pct" | |
| payload = { | |
| "saved_at": utc_now(), | |
| "candidate": asdict(candidate), | |
| "result": result, | |
| "supervisor_model": { | |
| "id": SUPERVISOR_MODEL_ID, | |
| "revision": SUPERVISOR_REVISION, | |
| "trust_remote_code": False, | |
| }, | |
| "crt_sources": [ | |
| "https://innercircletrader.net/tutorials/candle-range-theory-crt/", | |
| "https://tradingwyckoff.com/en/crt/", | |
| ], | |
| } | |
| save_json(checkpoint_dir / f"{prefix}_checkpoint.json", payload) | |
| log("Saved {} checkpoint at validation accuracy {:.2%}", prefix, result["validation"]["trade_accuracy"]) | |
| def package_checkpoints(checkpoint_dir: Path) -> Path: | |
| archive_path = checkpoint_dir.parent / "100optimization_checkpoints.zip" | |
| if archive_path.exists(): | |
| archive_path.unlink() | |
| with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as archive: | |
| for path in sorted(checkpoint_dir.rglob("*")): | |
| if path.is_file(): | |
| archive.write(path, arcname=f"{checkpoint_dir.name}/{path.relative_to(checkpoint_dir)}") | |
| log("Checkpoint archive: {}", archive_path) | |
| return archive_path | |
| def load_market_data() -> tuple[dict[str, pd.DataFrame], Path]: | |
| archive = load_dataset_archive() | |
| extracted = safe_extract_zip(archive, CFG.data_dir) | |
| data_dir = find_data_directory(extracted) | |
| raw: dict[str, pd.DataFrame] = {} | |
| for timeframe in ("M1", "M5", "M15", "M30", "H1", "H4"): | |
| path = find_timeframe_file(data_dir, timeframe) | |
| if path is None: | |
| raise FileNotFoundError(f"Missing {timeframe}.csv in {data_dir}") | |
| raw[timeframe] = load_ohlcv(path) | |
| log("Loaded {}: {:,} rows", timeframe, len(raw[timeframe])) | |
| return raw, data_dir | |
| def run() -> dict[str, Any]: | |
| checkpoint_dir = ensure_checkpoint_dir() | |
| random.seed(CFG.seed) | |
| np.random.seed(CFG.seed) | |
| raw, data_dir = load_market_data() | |
| m1 = raw["M1"] | |
| if CFG.max_rows and len(m1) > CFG.max_rows: | |
| m1 = m1.tail(CFG.max_rows).reset_index(drop=True) | |
| log("Using the last {:,} M1 rows because MAX_ROWS is set", len(m1), level="WARN") | |
| atr = compute_atr(m1, CFG.atr_period) | |
| candidates = make_candidates(CFG.seed) | |
| outcome_cache: dict[tuple[float, int], dict[str, np.ndarray]] = {} | |
| signal_cache: dict[int, np.ndarray] = {} | |
| history: list[dict[str, Any]] = [] | |
| evaluated: set[int] = set() | |
| best_result: Optional[dict[str, Any]] = None | |
| best_candidate: Optional[Candidate] = None | |
| milestones_saved: set[int] = set() | |
| n = len(m1) | |
| train_end = int(n * 0.60) | |
| validation_end = int(n * 0.80) | |
| purge = max(CFG.default_horizon, 60) | |
| validation_start = min(n, train_end + purge) | |
| test_start = min(n, validation_end + purge) | |
| supervisor: Optional[Supervisor] | |
| try: | |
| supervisor = Supervisor() | |
| except Exception as exc: | |
| log("Supervisor unavailable: {}. Continuing deterministically.", exc, level="WARN") | |
| supervisor = None | |
| log( | |
| "CRT optimization rows={} | train={} | validation={} | test={}", | |
| n, | |
| train_end, | |
| validation_end - validation_start, | |
| n - test_start, | |
| ) | |
| for trial in range(CFG.max_trials): | |
| remaining = [candidate for candidate in candidates if candidate.candidate_id not in evaluated] | |
| if not remaining: | |
| break | |
| selected_id: Optional[int] = None | |
| supervisor_reply = "" | |
| if supervisor is not None and history: | |
| selected_id, supervisor_reply = supervisor.choose(remaining, history) | |
| if selected_id is None: | |
| # Deterministic fallback: evaluate candidates in the seeded order. | |
| selected_id = remaining[0].candidate_id | |
| candidate = next(c for c in candidates if c.candidate_id == selected_id) | |
| evaluated.add(candidate.candidate_id) | |
| key = (candidate.sl_atr_mult, candidate.horizon) | |
| if key not in outcome_cache: | |
| outcome_cache[key] = build_outcomes( | |
| m1, | |
| CFG.atr_period, | |
| candidate.sl_atr_mult, | |
| candidate.horizon, | |
| ) | |
| outcomes = outcome_cache[key] | |
| if candidate.candidate_id not in signal_cache: | |
| signal_cache[candidate.candidate_id] = generate_crt_signals( | |
| m1, | |
| raw[candidate.reference_timeframe], | |
| atr, | |
| candidate, | |
| ) | |
| signals = signal_cache[candidate.candidate_id] | |
| validation = metrics_for_slice( | |
| signals, | |
| outcomes, | |
| validation_start, | |
| validation_end, | |
| CFG.min_trades, | |
| CFG.min_coverage, | |
| ) | |
| test = metrics_for_slice( | |
| signals, | |
| outcomes, | |
| test_start, | |
| n, | |
| CFG.min_trades, | |
| CFG.min_coverage, | |
| ) | |
| result = { | |
| "trial": trial + 1, | |
| "candidate_id": candidate.candidate_id, | |
| "candidate": asdict(candidate), | |
| "validation": validation, | |
| "test_preview": test, | |
| "supervisor_reply": supervisor_reply, | |
| } | |
| history.append(result) | |
| save_json(checkpoint_dir / "trials.json", history) | |
| log( | |
| "Trial {} candidate={} validation accuracy={:.2%} coverage={:.2%} " | |
| "trades={} expectancy={:.3f}R test accuracy={:.2%}", | |
| trial + 1, | |
| candidate.candidate_id, | |
| validation["trade_accuracy"], | |
| validation["coverage"], | |
| validation["trades"], | |
| validation["expectancy_R"], | |
| test["trade_accuracy"], | |
| ) | |
| if best_result is None or objective(validation) > objective(best_result["validation"]): | |
| best_result = result | |
| best_candidate = candidate | |
| save_checkpoint(checkpoint_dir, candidate, result) | |
| achieved = validation["eligible"] and validation["trade_accuracy"] >= CFG.target_accuracy | |
| for milestone in (20, 30, 80): | |
| if ( | |
| milestone not in milestones_saved | |
| and validation["eligible"] | |
| and validation["trade_accuracy"] >= milestone / 100.0 | |
| ): | |
| save_checkpoint(checkpoint_dir, candidate, result, milestone=milestone) | |
| milestones_saved.add(milestone) | |
| if achieved: | |
| log( | |
| "Target validation accuracy reached: {:.2%}. " | |
| "No further optimization trials will run.", | |
| validation["trade_accuracy"], | |
| ) | |
| break | |