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 BrowserSYMBOL = “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 BrowserWORKER = 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}% ”
f”edge {m[‘edge’]:+.2f}% ({n_sig} signals, {m[‘duration_s’]}s)”, flush=True)
except Exception as e:
out[“errors”].append(f”{params}: {e}”)
print(f” [{i:>2}/{len(combos)}] {params} FAILED: {e}”, flush=True)
traceback.print_exc()
await is_data.stop()
if not out[“grid”]:
json.dump(out, open(OUT, “w”)); raise SystemExit(“no successful runs”)
best = max(out[“grid”], key=lambda r: r[“edge”])
out[“best”] = {k: best[k] for k in keys}
print(f”n best in-sample config: {out[‘best’]} (edge {best[‘edge’]:+.2f}%)”, flush=True)
print(“n” + “=” * 78 + “n OUT-OF-SAMPLE VALIDATION (never optimised on)n” + “=” * 78,
flush=True)
oos_data, _ = await load_data(CFG[“out_of_sample”])
res, n_sig = await backtest(oos_data, out[“best”], plot=True, storage=True)
m = metrics(res); m.update(out[“best”])
m[“signals”] = n_sig; m[“edge”] = m[“profitability”] – m[“market”]
out[“oos”] = m
print(f” OOS P&L {m[‘profitability’]:+.2f}% vs market {m[‘market’]:+.2f}% ”
f”edge {m[‘edge’]:+.2f}% ({n_sig} signals)”, flush=True)
print(” ” + res.describe(), flush=True)
report_dir = os.path.join(os.getcwd(), “report”)
os.makedirs(report_dir, exist_ok=True)
try:
plot = await res.plot(report_file=os.path.join(report_dir, “report.html”), show=False)
out[“bundle”] = os.path.join(os.path.dirname(os.path.abspath(plot.report_file)),
“report.json”)
print(f” ✓ report bundle: {out[‘bundle’]}”, flush=True)
except Exception as e:
out[“errors”].append(f”report: {e}”)
print(f” ✗ report generation failed: {e}”, flush=True)
await oos_data.stop()
json.dump(out, open(OUT, “w”), indent=2, default=str)
print(“n✓ results written to”, OUT, flush=True)
asyncio.run(main())
”’
with open(WORKER, “w”) as f:
f.write(WORKER_SRC)
We build the standalone OctoBot worker that contains the strategy logic and executes inside the isolated virtual environment. We calculate RSI, fast and slow EMAs, and ATR values, generate entry signals when oversold conditions align with an upward trend, and derive volatility-adjusted stop-loss and take-profit offsets. We also define the historical data loader, backtest runner, grid-search loop, out-of-sample validation, performance metrics, and report generation process.
Copy CodeCopiedUse a different Browserimport datetime as _dt
def ts(d):
return int(_dt.datetime.strptime(d, “%Y-%m-%d”)
.replace(tzinfo=_dt.timezone.utc).timestamp())
CONFIG_PATH = os.path.join(WORK_DIR, “config.json”)
RESULTS_PATH = os.path.join(WORK_DIR, “results.json”)
json.dump({
“symbol”: SYMBOL, “time_frame”: TIME_FRAME, “exchanges”: EXCHANGES,
“in_sample”: [ts(IN_SAMPLE[0]), ts(IN_SAMPLE[1])],
“out_of_sample”: [ts(OUT_OF_SAMPLE[0]), ts(OUT_OF_SAMPLE[1])],
“grid”: GRID, “fixed”: FIXED,
}, open(CONFIG_PATH, “w”), indent=2)
env = dict(os.environ, OBS_CONFIG=CONFIG_PATH, OBS_OUT=RESULTS_PATH,
PYTHONUNBUFFERED=”1″)
t0 = time.time()
proc = subprocess.Popen([PY, WORKER], cwd=WORK_DIR, env=env, text=True,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1)
for line in proc.stdout:
print(line.rstrip())
proc.wait()
print(f”n total backtesting time: {time.time() – t0:.1f}s (exit {proc.returncode})”)
if not os.path.exists(RESULTS_PATH):
raise SystemExit(“No results produced — read the log above. ”
“Most common cause: every exchange refused the data request.”)
R = json.load(open(RESULTS_PATH))
We convert the selected in-sample and out-of-sample dates into UTC timestamps and serialize the complete experiment configuration into a JSON file. We launch the OctoBot worker as a separate subprocess so its dependency environment remains isolated from the main Colab kernel while its logs stream directly into the notebook. We then verify that the run produces a results file and load the generated JSON output for downstream analysis.
Copy CodeCopiedUse a different Browserimport pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
pd.set_option(“display.width”, 160)
grid = pd.DataFrame(R[“grid”]).sort_values(“edge”, ascending=False)
cols = [c for c in [“rsi_period”, “rsi_threshold”, “tp_atr_mult”, “signals”,
“profitability”, “market”, “edge”, “duration_s”] if c in grid.columns]
print(“n=== IN-SAMPLE GRID (ranked by edge over buy & hold) ===”)
print(grid[cols].to_string(index=False, float_format=lambda v: f”{v:,.2f}”))
if R.get(“oos”):
o = R[“oos”]
print(“n=== OUT-OF-SAMPLE ===”)
print(f” config : { {k: o[k] for k in GRID} }”)
print(f” strategy return : {o[‘profitability’]:+.2f}%”)
print(f” buy & hold : {o[‘market’]:+.2f}%”)
print(f” edge : {o[‘edge’]:+.2f}% ← the only number that matters”)
print(f” entries taken : {o[‘signals’]}”)
print(f” end portfolio : {o[‘end_portfolio’]}”)
is_edge = grid.iloc[0][“edge”]
decay = o[“edge”] – is_edge
print(f”n edge decay IS→OOS: {decay:+.2f} pts ”
f”({‘holds up’ if decay > -5 else ‘likely overfit — treat with suspicion’})”)
We move back into the Colab environment and organize the grid-search results with Pandas for easier comparison and interpretation. We rank every parameter configuration according to its excess return over the market and print the key performance metrics for both the in-sample search and out-of-sample validation. We also calculate the change in strategy edge between the two periods to obtain a simple indication of whether the optimized parameters generalize or show signs of overfitting.
Copy CodeCopiedUse a different Browserif {“rsi_period”, “rsi_threshold”} <= set(grid.columns):
pivot = grid.pivot_table(index=”rsi_threshold”, columns=”rsi_period”,
values=”edge”, aggfunc=”mean”)
fig = go.Figure(go.Heatmap(z=pivot.values, x=pivot.columns, y=pivot.index,
colorscale=”RdYlGn”, zmid=0,
colorbar=dict(title=”edge %”),
text=pivot.round(1).values, texttemplate=”%{text}”))
fig.update_layout(title=”In-sample edge vs buy & hold — a broad plateau is trustworthy, ”
“an isolated hot cell is noise”,
xaxis_title=”RSI period”, yaxis_title=”RSI buy threshold”,
height=380, template=”plotly_dark”)
fig.show()
def harvest(node, found):
“””The report bundle nests display elements arbitrarily; walk it and grab
anything that looks like a plottable series.”””
if isinstance(node, dict):
if isinstance(node.get(“x”), list) and len(node[“x”]) > 1:
if all(k in node for k in (“open”, “high”, “low”, “close”)):
found[“candles”].append(node)
elif isinstance(node.get(“y”), list) and len(node[“y”]) == len(node[“x”]):
found[“series”].append(node)
for v in node.values():
harvest(v, found)
elif isinstance(node, list):
for v in node:
harvest(v, found)
return found
bundle_path = R.get(“bundle”)
We visualize the parameter-search surface by plotting the average strategy edge across RSI periods and entry thresholds as an interactive Plotly heatmap. We use this surface to inspect whether strong performance appears across a broad parameter region or only around an isolated configuration that may represent noise. We also define a recursive report-harvesting function that searches OctoBot’s nested report structure for candle data and other plottable time-series elements.
Copy CodeCopiedUse a different Browserif bundle_path and os.path.exists(bundle_path):
bundle = json.load(open(bundle_path))
f = harvest(bundle, {“candles”: [], “series”: []})
print(f”n=== REPORT BUNDLE === {len(f[‘candles’])} candle set(s), ”
f”{len(f[‘series’])} series”)
def norm_x(xs):
xs = [float(v) for v in xs]
unit = “ms” if (xs and max(xs) > 1e11) else “s”
return pd.to_datetime(xs, unit=unit)
fig = make_subplots(rows=2, cols=1, shared_xaxes=True,
row_heights=[0.62, 0.38], vertical_spacing=0.06,
subplot_titles=(“Price & executed trades”,
“Portfolio value / indicators”))
if f[“candles”]:
c = max(f[“candles”], key=lambda d: len(d[“x”]))
fig.add_trace(go.Candlestick(x=norm_x(c[“x”]), open=c[“open”], high=c[“high”],
low=c[“low”], close=c[“close”], name=SYMBOL),
row=1, col=1)
portfolio_kw = (“portfolio”, “value”, “wallet”, “balance”)
for s in f[“series”]:
title = str(s.get(“title”) or s.get(“name”) or “series”)
n = len(s[“x”])
if n < 3:
continue
mode = s.get(“mode”) or (“markers” if n < 60 else “lines”)
row = 2 if any(k in title.lower() for k in portfolio_kw) or “rsi” in title.lower()
or “atr” in title.lower() else 1
fig.add_trace(go.Scatter(x=norm_x(s[“x”]), y=s[“y”], name=title[:38],
mode=mode, opacity=0.9), row=row, col=1)
fig.update_layout(height=760, template=”plotly_dark”, xaxis_rangeslider_visible=False,
title=f”OctoBot out-of-sample run — {SYMBOL} {TIME_FRAME} ”
f”({R.get(‘exchange’, ‘?’)}) — {OUT_OF_SAMPLE[0]} → {OUT_OF_SAMPLE[1]}”,
legend=dict(orientation=”h”, y=-0.08))
fig.show()
else:
print(“n(no report bundle — charts skipped; the numeric results above are still valid)”)
if R.get(“errors”):
print(“n non-fatal errors during the run:”)
for e in R[“errors”]:
print(” -“, e)
print(“””
──────────────────────────────────────────────────────────────────────────────
WHERE TO GO NEXT
• Edit GRID / FIXED at the top and re-run — the env is cached, only backtests rerun.
• obs.limit(ctx, “sell”, amount=”50%”, offset=”2%”) → limit orders
• obs.set_leverage(ctx, 3) + exchange_type=”future” → futures / shorts
• get_data() accepts LISTS for symbol and time_frame → multi-asset, multi-TF strategies
• Swap tulipy for pandas-ta / your own ML model: initialize() just needs to fill a
set of entry timestamps, so a trained classifier drops straight in.
• Docs: https://www.octobot.cloud/en/guides/octobot-script
• For live/paper trading use the full OctoBot app, not this scripting layer.
Reminder: past performance in a backtest tells you about the past. Slippage, fees
beyond the simulator’s model, liquidity and regime change all bite in live markets.
──────────────────────────────────────────────────────────────────────────────
“””)
We load the generated OctoBot report bundle and reconstruct the out-of-sample trading results as interactive price and indicator charts. We normalize timestamps, display candlestick data, and dynamically add available portfolio, RSI, ATR, trade, and other report series to a multi-panel Plotly visualization. We finally surface any non-fatal execution errors and outline several directions for extending the workflow, including limit orders, futures, multi-asset strategies, and machine-learning-based signals.
In conclusion, we implemented an end-to-end OctoBot quantitative research pipeline that moves beyond a simple single-run backtest and introduces a more disciplined strategy-development process. We isolated OctoBot’s dependency stack, retrieved exchange data through its native infrastructure, defined a volatility-aware RSI and EMA strategy, optimized its parameters on historical in-sample data, and evaluated the winning configuration on an untouched out-of-sample window. By comparing strategy profitability against buy-and-hold performance and examining parameter surfaces and out-of-sample edge decay, we gained a clearer view of whether our results represent a robust trading signal or merely an overfitted historical pattern. We also transformed OctoBot’s generated report bundle into interactive visualizations that make strategy behavior, market movements, indicators, and portfolio dynamics easier to inspect.
Check out the Full Codes here. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
The post Building and Validating a Quantitative Trading Strategy with OctoBot, Walk-Forward Backtesting, Parameter Optimization, and Interactive Analysis appeared first on MarkTechPost.