In this tutorial, we build a complete quantitative backtesting workflow with OctoBot and OctoBot-Script while keeping the environment isolated from Colab’s preinstalled dependencies. We configure a rule-based trading strategy that combines RSI-based oversold signals, EMA trend confirmation, and ATR-driven adaptive stop-loss and take-profit levels, and we execute it through OctoBot’s native market-order and backtesting APIs. We also retrieve historical OHLCV data through OctoBot’s data layer with automatic exchange fallback, perform a multi-parameter grid search over an in-sample period, and select the strongest configuration based on its excess return relative to buy-and-hold. We then validate the selected parameters on a completely separate out-of-sample period to assess generalization and identify potential overfitting. Finally, we extract OctoBot’s backtest report data and use Pandas and Plotly to analyze parameter sensitivity, portfolio performance, price action, indicators, and execution results in an interactive Colab environment. Copy CodeCopiedUse a different Browser SYMBOL = “BTC/USDT” TIME_FRAME = “1d” EXCHANGES = [“binance”, “kucoin”, “okx”, “bybit”, “mexc”, “kraken”] IN_SAMPLE = (“2019-01-01”, “2023-01-01”) OUT_OF_SAMPLE = (“2023-01-01”, “2025-06-01”) GRID = { “rsi_period”: [7, 14, 21], “rsi_threshold”: [25, 30, 35], “tp_atr_mult”: [3.0, 5.0], } FIXED = { “ema_fast”: 50, “ema_slow”: 200, “atr_period”: 14, “sl_atr_mult”: 2.0, “position_size”: “20%”, “min_offset_pct”: 1.0, “max_offset_pct”: 40.0, } VENV_DIR = “/content/octobot_env” WORK_DIR = “/content/octobot_lab” OCTOBOT_V = “2.1.1” PY_VERSION = “3.12” import json, os, subprocess, sys, textwrap, time, itertools, shutil os.makedirs(WORK_DIR, exist_ok=True) PY = os.path.join(VENV_DIR, “bin”, “python”) MARKER = os.path.join(VENV_DIR, “.octobot_ready”) def sh(cmd, **kw): “””Run a command, streaming its output live into the Colab cell.””” print(f”$ {‘ ‘.join(cmd)}”) p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, **kw) for line in p.stdout: print(” ” + line.rstrip()) p.wait() if p.returncode != 0: raise RuntimeError(f”command failed ({p.returncode}): {‘ ‘.join(cmd)}”) if not os.path.exists(MARKER): print(“=” * 90, “n BUILDING OCTOBOT ENVIRONMENT (one-off, ~2 min)n”, “=” * 90) subprocess.run([sys.executable, “-m”, “pip”, “install”, “-q”, “uv”], check=True) UV = [sys.executable, “-m”, “uv”] sh(UV + [“venv”, “–python”, PY_VERSION, VENV_DIR]) sh(UV + [“pip”, “install”, “–python”, PY, “-q”, f”OctoBot=={OCTOBOT_V}”, “wheel”, “setuptools”, “appdirs==1.4.4”]) sh(UV + [“pip”, “install”, “–python”, PY, “-q”, “–no-build-isolation”, “octobot-script”]) sh([PY, “-m”, “octobot_script.cli”, “install_tentacles”, “–quite”]) sh([PY, “-c”, textwrap.dedent(“”” import os, shutil, octobot_script.resources as r base = r.get_report_resource_path(“”) src, dst_dir = os.path.join(base, “index.html”), os.path.join(base, “dist”) os.makedirs(dst_dir, exist_ok=True) dst = os.path.join(dst_dir, “index.html”) if os.path.exists(src) and not os.path.exists(dst): shutil.copy2(src, dst); print(“patched report template ->”, dst) else: print(“report template already fine”) “””)]) open(MARKER, “w”).write(“ok”) print(“n environment readyn”) else: print(” environment already built (delete”, VENV_DIR, “to rebuild)n”) We define the core trading configuration, including the symbol, timeframe, exchange fallback list, backtesting windows, parameter grid, and fixed strategy settings. We then create an isolated Python environment with uv and install the pinned OctoBot and OctoBot-Script dependencies required for the workflow. We also install the OctoBot tentacles package and patch the report-template path so later backtest reporting works correctly inside the Colab environment. Copy CodeCopiedUse a different Browser WORKER = os.path.join(WORK_DIR, “octobot_worker.py”) WORKER_SRC = r”’ import asyncio, itertools, json, os, sys, time, traceback import numpy as np import tulipy import octobot_script as obs CFG = json.load(open(os.environ[“OBS_CONFIG”])) OUT = os.environ[“OBS_OUT”] FIX = CFG[“fixed”] for kw in (“Close”, “High”, “Low”, “Time”, “market”, “current_live_time”, “plot_indicator”): if not hasattr(obs, kw): raise RuntimeError( f”octobot_script.{kw} missing -> tentacles are not installed. ” “Run: python -m octobot_script.cli install_tentacles” ) def tail(*arrays): “””tulipy indicators return different lengths; right-align them all.””” n = min(len(a) for a in arrays) return [np.asarray(a)[-n:] for a in arrays] def clamp(v): return float(min(max(v, FIX[“min_offset_pct”]), FIX[“max_offset_pct”])) def build_callbacks(params, run_data): “”” OctoBot-Script splits a strategy into: initialize(ctx) -> runs once on the first candle. Do vectorised work here. strategy(ctx) -> runs on EVERY closed candle. Keep it cheap. “”” async def initialize(ctx): closes = await obs.Close(ctx, max_history=True) highs = await obs.High(ctx, max_history=True) lows = await obs.Low(ctx, max_history=True) times = await obs.Time(ctx, max_history=True, use_close_time=True) rsi = tulipy.rsi(closes, period=params[“rsi_period”]) ema_f = tulipy.ema(closes, period=FIX[“ema_fast”]) ema_s = tulipy.ema(closes, period=FIX[“ema_slow”]) atr = tulipy.atr(highs, lows, closes, period=FIX[“atr_period”]) t, c, rsi, ema_f, ema_s, atr = tail(times, closes, rsi, ema_f, ema_s, atr) atr_pct = np.where(c > 0, atr / c * 100.0, 0.0) entries, offsets = set(), {} for i in range(len(t)): oversold = rsi[i] < params[“rsi_threshold”] uptrend = ema_f[i] > ema_s[i] if oversold and uptrend and atr_pct[i] > 0: ts = float(t[i]) entries.add(ts) offsets[ts] = ( clamp(FIX[“sl_atr_mult”] * atr_pct[i]), clamp(params[“tp_atr_mult”] * atr_pct[i]), ) run_data[“entries”] = entries run_data[“offsets”] = offsets if run_data.get(“plot”): await obs.plot_indicator(ctx, f”RSI({params[‘rsi_period’]})”, t, rsi, entries) await obs.plot_indicator(ctx, f”EMA{FIX[’ema_fast’]}”, t, ema_f) await obs.plot_indicator(ctx, f”EMA{FIX[’ema_slow’]}”, t, ema_s) await obs.plot_indicator(ctx, “ATR %”, t, atr_pct) async def strategy(ctx): now = obs.current_live_time(ctx) if now not in run_data[“entries”]: return sl, tp = run_data[“offsets”][now] await obs.market( ctx, “buy”, amount=FIX[“position_size”], stop_loss_offset=f”-{sl:.2f}%”, take_profit_offset=f”{tp:.2f}%”, ) return initialize, strategy def metrics(res): br = res.report.get(“bot_report”, {}) first = lambda d: float(list(d.values())[0]) if isinstance(d, dict) and d else float(“nan”) return { “profitability”: first(br.get(“profitability”, {})), “market”: first(br.get(“market_average_profitability”, {})), “reference”: br.get(“reference_market”), “start_portfolio”: str(br.get(“starting_portfolio”)), “end_portfolio”: str(br.get(“end_portfolio”)), “candles”: res.candles_count, “duration_s”: round(res.duration or 0, 2), “errors”: res.report.get(“errors_count”), } async def load_data(window): “””Try each exchange until one serves data (Binance blocks many datacenter IPs).””” start, end = window last = None for ex in CFG[“exchanges”]: try: print(f” ↓ fetching {CFG[‘symbol’]} {CFG[‘time_frame’]} from {ex} ” f”[{time.strftime(‘%Y-%m-%d’, time.gmtime(start))} → ” f”{time.strftime(‘%Y-%m-%d’, time.gmtime(end))}]”, flush=True) data = await obs.get_data( CFG[“symbol”], CFG[“time_frame”], exchange=ex, exchange_type=”spot”, start_timestamp=start, end_timestamp=end, social_services=[], ) print(f” ✓ {ex} ok -> {data.data_files}”, flush=True) return data, ex except Exception as e: last = e print(f” ✗ {ex}: {type(e).__name__}: {e}”, flush=True) raise RuntimeError(f”no exchange served data; last error: {last}”) async def backtest(data, params, plot=False, storage=False): run_data = {“entries”: None, “offsets”: {}, “plot”: plot} init_f, strat_f = build_callbacks(params, run_data) res = await obs.run( data, params, strategy_func=strat_f, initialize_func=init_f, enable_logs=False, enable_storage=storage, ) return res, len(run_data[“entries”] or ()) async def main(): out = {“grid”: [], “best”: None, “oos”: None, “errors”: []} print(“n” + “=” * 78 + “n IN-SAMPLE GRID SEARCHn” + “=” * 78, flush=True) is_data, ex_used = await load_data(CFG[“in_sample”]) out[“exchange”] = ex_used keys = list(CFG[“grid”].keys()) combos = [dict(zip(keys, v)) for v in itertools.product(*CFG[“grid”].values())] print(f” {len(combos)} configurations to evaluaten”, flush=True) for i, params in enumerate(combos, 1): try: res, n_sig = await backtest(is_data, params) m = metrics(res) m.update(params); m[“signals”] = n_sig m[“edge”] = m[“profitability”] – m[“market”] out[“grid”].append(m) print(f” [{i:>2}/{len(combos)}] {params} ” f”P&L {m[‘profitability’]:+.2f}% vs market {m[‘market’]:+.2f}%