Building and Validating a Quantitative Trading Strategy with OctoBot, Walk-Forward Backtesting, Parameter Optimization, and Interactive Analysis

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 material/octobot_env"
WORK_DIR   = "/content material/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.be part of(VENV_DIR, "bin", "python")
MARKER = os.path.be part of(VENV_DIR, ".octobot_ready")
def sh(cmd, **kw):
   """Run a command, streaming its output reside into the Colab cell."""
   print(f"$ {' '.be part of(cmd)}")
   p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                        textual content=True, bufsize=1, **kw)
   for line in p.stdout:
       print("   " + line.rstrip())
   p.wait()
   if p.returncode != 0:
       elevate RuntimeError(f"command failed ({p.returncode}): {' '.be part of(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"], examine=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("okay")
   print("n surroundings readyn")
else:
   print(" surroundings already constructed (delete", VENV_DIR, "to rebuild)n")

WORKER = os.path.be part of(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):
       elevate RuntimeError(
           f"octobot_script.{kw} lacking -> tentacles aren't put in. "
           "Run: python -m octobot_script.cli install_tentacles"
       )
def tail(*arrays):
   """tulipy indicators return totally different lengths; right-align all of them."""
   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 technique into:
     initialize(ctx) -> runs as soon as on the primary candle. Do vectorised work right here.
     technique(ctx)   -> runs on EVERY closed candle. Keep it low cost.
   """
   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)
       instances  = await obs.Time(ctx,  max_history=True, use_close_time=True)
       rsi  = tulipy.rsi(closes, interval=params["rsi_period"])
       ema_f = tulipy.ema(closes, interval=FIX["ema_fast"])
       ema_s = tulipy.ema(closes, interval=FIX["ema_slow"])
       atr   = tulipy.atr(highs, lows, closes, interval=FIX["atr_period"])
       t, c, rsi, ema_f, ema_s, atr = tail(instances, closes, rsi, ema_f, ema_s, atr)
       atr_pct = np.the place(c > 0, atr / c * 100.0, 0.0)
       entries, offsets = set(), {}
       for i in vary(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 technique(ctx):
       now = obs.current_live_time(ctx)
       if not in run_data["entries"]:
           return
       sl, tp = run_data["offsets"][now]
       await obs.market(
           ctx, "purchase",
           quantity=FIX["position_size"],
           stop_loss_offset=f"-{sl:.2f}%",
           take_profit_offset=f"{tp:.2f}%",
       )
   return initialize, technique
def metrics(res):
   br = res.report.get("bot_report", {})
   first = lambda d: float(record(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":     spherical(res.length or 0, 2),
       "errors":         res.report.get("errors_count"),
   }
async def load_data(window):
   """Try every change till one serves information (Binance blocks many datacenter IPs)."""
   begin, finish = window
   final = None
   for ex in CFG["exchanges"]:
       attempt:
           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)
           information = await obs.get_data(
               CFG["symbol"], CFG["time_frame"],
               change=ex, exchange_type="spot",
               start_timestamp=begin, end_timestamp=finish,
               social_services=[],
           )
           print(f"    ✓ {ex} okay -> {information.data_files}", flush=True)
           return information, ex
       besides Exception as e:
           final = e
           print(f"    ✗ {ex}: {kind(e).__name__}: {e}", flush=True)
   elevate RuntimeError(f"no change served information; final error: {final}")
async def backtest(information, 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(
       information, params,
       strategy_func=strat_f,
       initialize_func=init_f,
       enable_logs=False,
       enable_storage=storage,
   )
   return res, len(run_data["entries"] or ())
async def major():
   out = {"grid": [], "finest": 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  = record(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):
       attempt:
           res, n_sig = await backtest(is_data, params)
           m = metrics(res)
           m.replace(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} indicators, {m['duration_s']}s)", flush=True)
       besides 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.cease()
   if not out["grid"]:
       json.dump(out, open(OUT, "w")); elevate SystemExit("no profitable runs")
   finest = max(out["grid"], key=lambda r: r["edge"])
   out["best"] = {ok: finest[k] for ok in keys}
   print(f"n   finest in-sample config: {out['best']}  (edge {finest['edge']:+.2f}%)", flush=True)
   print("n" + "=" * 78 + "n  OUT-OF-SAMPLE VALIDATION (by no means 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.replace(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} indicators)", flush=True)
   print("  " + res.describe(), flush=True)
   report_dir = os.path.be part of(os.getcwd(), "report")
   os.makedirs(report_dir, exist_ok=True)
   attempt:
       plot = await res.plot(report_file=os.path.be part of(report_dir, "report.html"), present=False)
       out["bundle"] = os.path.be part of(os.path.dirname(os.path.abspath(plot.report_file)),
                                    "report.json")
       print(f"  ✓ report bundle: {out['bundle']}", flush=True)
   besides Exception as e:
       out["errors"].append(f"report: {e}")
       print(f"  ✗ report technology failed: {e}", flush=True)
   await oos_data.cease()
   json.dump(out, open(OUT, "w"), indent=2, default=str)
   print("n✓ outcomes written to", OUT, flush=True)
asyncio.run(major())
'''
with open(WORKER, "w") as f:
   f.write(WORKER_SRC)

import datetime as _dt
def ts(d):
   return int(_dt.datetime.strptime(d, "%Y-%m-%d")
              .exchange(tzinfo=_dt.timezone.utc).timestamp())
CONFIG_PATH  = os.path.be part of(WORK_DIR, "config.json")
RESULTS_PATH = os.path.be part of(WORK_DIR, "outcomes.json")
json.dump({
   "image": 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, "fastened": 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, textual content=True,
                       stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1)
for line in proc.stdout:
   print(line.rstrip())
proc.wait()
print(f"n  complete backtesting time: {time.time() - t0:.1f}s  (exit {proc.returncode})")
if not os.path.exists(RESULTS_PATH):
   elevate SystemExit("No outcomes produced — learn the log above. "
                    "Most frequent trigger: each change refused the information request.")
R = json.load(open(RESULTS_PATH))

import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
pd.set_option("show.width", 160)
grid = pd.DataBody(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 purchase & maintain) ===")
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          : { {ok: o[k] for ok in GRID} }")
   print(f"  technique return : {o['profitability']:+.2f}%")
   print(f"  purchase & maintain      : {o['market']:+.2f}%")
   print(f"  edge            : {o['edge']:+.2f}%   ← the one quantity that issues")
   print(f"  entries taken   : {o['signals']}")
   print(f"  finish 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 'possible overfit — deal with with suspicion'})")

if {"rsi_period", "rsi_threshold"} <= set(grid.columns):
   pivot = grid.pivot_table(index="rsi_threshold", columns="rsi_period",
                            values="edge", aggfunc="imply")
   fig = go.Figure(go.Heatmap(z=pivot.values, x=pivot.columns, y=pivot.index,
                              colorscale="RdYlGn", zmid=0,
                              colorbar=dict(title="edge %"),
                              textual content=pivot.spherical(1).values, texttemplate="%{textual content}"))
   fig.update_layout(title="In-sample edge vs purchase & maintain — a broad plateau is reliable, "
                           "an remoted sizzling cell is noise",
                     xaxis_title="RSI interval", yaxis_title="RSI purchase threshold",
                     peak=380, template="plotly_dark")
   fig.present()
def harvest(node, discovered):
   """The report bundle nests show parts arbitrarily; stroll it and seize
   something that appears like a plottable sequence."""
   if isinstance(node, dict):
       if isinstance(node.get("x"), record) and len(node["x"]) > 1:
           if all(ok in node for ok in ("open", "excessive", "low", "shut")):
               discovered["candles"].append(node)
           elif isinstance(node.get("y"), record) and len(node["y"]) == len(node["x"]):
               discovered["series"].append(node)
       for v in node.values():
           harvest(v, discovered)
   elif isinstance(node, record):
       for v in node:
           harvest(v, discovered)
   return discovered
bundle_path = R.get("bundle")

if bundle_path and os.path.exists(bundle_path):
   bundle = json.load(open(bundle_path))
   f = harvest(bundle, {"candles": [], "sequence": []})
   print(f"n=== REPORT BUNDLE === {len(f['candles'])} candle set(s), "
         f"{len(f['series'])} sequence")
   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 worth / 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"], excessive=c["high"],
                                    low=c["low"], shut=c["close"], identify=SYMBOL),
                     row=1, col=1)
   portfolio_kw = ("portfolio", "worth", "pockets", "stability")
   for s in f["series"]:
       title = str(s.get("title") or s.get("identify") or "sequence")
       n = len(s["x"])
       if n < 3:
           proceed
       mode = s.get("mode") or ("markers" if n < 60 else "strains")
       row = 2 if any(ok in title.decrease() for ok in portfolio_kw) or "rsi" in title.decrease() 
               or "atr" in title.decrease() else 1
       fig.add_trace(go.Scatter(x=norm_x(s["x"]), y=s["y"], identify=title[:38],
                                mode=mode, opacity=0.9), row=row, col=1)
   fig.update_layout(peak=760, template="plotly_dark", xaxis_rangeslider_visible=False,
                     title=f"OctoBot out-of-sample run — {SYMBOL} {TIME_FRAME} "
                           f"({R.get('change', '?')}) — {OUT_OF_SAMPLE[0]} → {OUT_OF_SAMPLE[1]}",
                     legend=dict(orientation="h", y=-0.08))
   fig.present()
else:
   print("n(no report bundle — charts skipped; the numeric outcomes above are nonetheless legitimate)")
if R.get("errors"):
   print("n non-fatal errors in the course of the run:")
   for e in R["errors"]:
       print("  -", e)
print("""
──────────────────────────────────────────────────────────────────────────────
WHERE TO GO NEXT
 • Edit GRID / FIXED on the prime and re-run — the env is cached, solely backtests rerun.
 • obs.restrict(ctx, "promote", quantity="50%", offset="2%")  → restrict orders
 • obs.set_leverage(ctx, 3) + exchange_type="future"  → futures / shorts
 • get_data() accepts LISTS for image and time_frame → multi-asset, multi-TF methods
 • Swap tulipy for pandas-ta / your individual ML mannequin: initialize() simply must fill a
   set of entry timestamps, so a educated classifier drops straight in.
 • Docs: https://www.octobot.cloud/en/guides/octobot-script
 • For reside/paper buying and selling use the complete OctoBot app, not this scripting layer.
Reminder: previous efficiency in a backtest tells you concerning the previous. Slippage, charges
past the simulator's mannequin, liquidity and regime change all chew in reside markets.
──────────────────────────────────────────────────────────────────────────────
""")